Audit & Provenance Pipeline

Complete audit trail architecture, event capture, tamper-proof storage, compliance reporting, and data provenance.

Overview

Healthcare AI evaluation requires comprehensive audit trails for compliance, debugging, and accountability. Akhara captures every action with cryptographic integrity guarantees.

Audit Architecture

Core Principles

Complete Capture

Every action, access, and change is logged

Immutability

Audit records cannot be modified or deleted

Cryptographic Integrity

Hash chains prevent tampering

Queryable

Fast search across billions of events

Event Flow

Event Categories

Event Types

Category	Event Types	Retention
Data Events	sample.created, sample.updated, sample.deleted	7 years
Evaluation Events	evaluation.started, evaluation.completed, score.created	7 years
Review Events	task.assigned, review.submitted, consensus.reached	7 years
Access Events	phi.accessed, export.requested, api.called	7 years
System Events	config.changed, user.invited, role.updated	7 years
Export Events	export.started, export.completed, export.downloaded	7 years

Event Schema

{
  "event_id": "evt_abc123def456",
  "timestamp": "2025-01-15T10:30:00.123Z",
  "event_type": "review.submitted",
  
  "actor": {
    "type": "user",
    "id": "user_xyz789",
    "email": "[e-mail]",
    "role": "reviewer",
    "ip_address": "192.0.2.1",
    "user_agent": "Mozilla/5.0...",
    "session_id": "sess_abc123"
  },
  
  "resource": {
    "type": "review",
    "id": "rev_def456",
    "project": "proj_ghi789",
    "organization": "org_jkl012"
  },
  
  "action": {
    "type": "create",
    "details": {
      "task_id": "task_mno345",
      "sample_id": "smp_pqr678",
      "scores": {
        "triage_correct": "under_triaged",
        "red_flags_caught": "all"
      }
    }
  },
  
  "context": {
    "request_id": "req_stu901",
    "correlation_id": "corr_vwx234",
    "source": "review_ui"
  },
  
  "integrity": {
    "content_hash": "sha256:a1b2c3d4...",
    "previous_hash": "sha256:e5f6g7h8...",
    "chain_position": 1847293,
    "signature": "sig_yza567..."
  }
}

Cryptographic Integrity

Hash Chain

Each event includes:

• Content Hash: SHA-256 of event payload
• Previous Hash: Hash of the preceding event
• Chain Position: Sequential position in the chain
• Signature: KMS-signed hash for non-repudiation

Verification Process

Verification API

# Verify audit trail integrity
verification = client.audit.verify(
    start_event="evt_abc123",
    end_event="evt_xyz789"
)

print(f"Events Verified: {verification.events_checked}")
print(f"Chain Intact: {verification.chain_intact}")
print(f"All Signatures Valid: {verification.signatures_valid}")

if not verification.valid:
    for issue in verification.issues:
        print(f"Issue at {issue.event_id}: {issue.description}")

Storage Architecture

Storage Tiers

Tier	Storage	Retention	Query Latency
Hot	Amazon Timestream	0-90 days	< 100ms
Warm	Amazon OpenSearch	90 days - 1 year	< 1s
Cold	S3 Glacier	1-7 years	Minutes to hours

Immutability Controls

• S3 Object Lock: WORM (Write Once Read Many) compliance mode
• No Delete API: Audit events have no delete endpoint
• Append Only: Events can only be added, never modified
• Multi-Region: Replicated to prevent single-point tampering

Query API

Basic Queries

# Query audit events
events = client.audit.query(
    project="proj_abc123",
    
    filters={
        "event_type": {"$in": ["review.submitted", "score.created"]},
        "timestamp": {
            "$gte": "2025-01-01T00:00:00Z",
            "$lt": "2025-02-01T00:00:00Z"
        },
        "actor.id": "user_xyz789"
    },
    
    order_by="timestamp",
    order="desc",
    limit=100
)

for event in events:
    print(f"{event.timestamp}: {event.event_type} by {event.actor.email}")

PHI Access Tracking

# Track all PHI access for a sample
phi_access = client.audit.query(
    filters={
        "event_type": "phi.accessed",
        "resource.id": "smp_abc123"
    }
)

print(f"Total PHI accesses: {len(phi_access)}")
for access in phi_access:
    print(f"  {access.timestamp}: {access.actor.email} - {access.action.details.phi_elements}")

Full-Text Search

# Search across all audit events
results = client.audit.search(
    query="chest pain emergency",
    project="proj_abc123",
    event_types=["review.submitted"],
    date_range=("2025-01-01", "2025-01-31")
)

Compliance Reporting

HIPAA Audit Requirements

Requirement	Implementation
Access Logging	All PHI access logged with user, time, purpose
Modification Tracking	All changes to PHI logged with before/after
Disclosure Logging	All exports and shares logged
User Activity	Login, logout, session activity tracked
System Events	Configuration changes, security events logged

Generating Reports

# Generate HIPAA compliance report
report = client.compliance.generate_report(
    report_type="hipaa_audit",
    organization="org_abc123",
    period={
        "start": "2025-01-01",
        "end": "2025-03-31"
    },
    include_sections=[
        "phi_access_summary",
        "user_activity",
        "security_events",
        "data_exports",
        "anomaly_detection"
    ]
)

# Download report
report.download("hipaa_audit_q1_2025.pdf")

PHI Access Report

# Generate PHI access report for a patient
phi_report = client.compliance.phi_access_report(
    sample_id="smp_abc123",  # or patient_id if tracked
    period={
        "start": "2025-01-01",
        "end": "2025-12-31"
    }
)

print(f"Total Accesses: {phi_report.total_accesses}")
print(f"Unique Users: {phi_report.unique_users}")
print(f"Access Purposes: {phi_report.purposes}")

Real-Time Alerting

Alert Configuration

# Configure audit alerts
client.audit.configure_alerts(
    organization="org_abc123",
    
    alerts=[
        {
            "name": "Unusual PHI Access",
            "condition": {
                "event_type": "phi.accessed",
                "aggregation": "count",
                "threshold": 100,
                "window": "1h",
                "group_by": "actor.id"
            },
            "channels": ["slack:#security", "pagerduty"]
        },
        {
            "name": "Failed Auth Spike",
            "condition": {
                "event_type": "auth.failed",
                "aggregation": "count",
                "threshold": 10,
                "window": "5m"
            },
            "channels": ["pagerduty"]
        },
        {
            "name": "Bulk Export",
            "condition": {
                "event_type": "export.requested",
                "filter": {"action.details.record_count": {"$gt": 1000}}
            },
            "channels": ["email:[e-mail]"]
        }
    ]
)

Anomaly Detection

Detected patterns:

• Unusual access times (off-hours activity)
• Abnormal data volumes
• New access patterns for users
• Geographic anomalies
• Privilege escalation attempts

SIEM Integration

Supported Platforms

Platform	Integration Method
Splunk	HTTP Event Collector
Datadog	Log forwarding
Sumo Logic	HTTP source
Elastic	Logstash/Filebeat
AWS Security Hub	Native integration
Azure Sentinel	Webhook
Chronicle	API ingestion

Configuration

# Configure SIEM forwarding
client.audit.configure_siem(
    organization="org_abc123",
    
    siem={
        "type": "splunk",
        "endpoint": "https://splunk.company.com:8088",
        "token": "splunk_hec_token",
        
        "filters": {
            "event_types": ["auth.*", "phi.*", "export.*", "security.*"],
            "min_severity": "warning"
        },
        
        "format": "json",
        "batch_size": 100,
        "flush_interval_seconds": 30
    }
)

Data Provenance

Lineage Tracking

Track the complete history of any data point:

# Get full provenance for a score
provenance = client.audit.get_provenance(
    resource_type="score",
    resource_id="scr_abc123"
)

print("Data Lineage:")
for step in provenance.lineage:
    print(f"  {step.timestamp}: {step.action} by {step.actor}")
    
# Output:
# Data Lineage:
#   2025-01-15T10:00:00Z: sample.created by api_key:gr_live_xxx
#   2025-01-15T10:00:05Z: evaluation.queued by system
#   2025-01-15T10:00:30Z: evaluator.executed by system
#   2025-01-15T10:00:35Z: score.created by evaluator:triage_accuracy
#   2025-01-15T10:01:00Z: task.created by system
#   2025-01-15T11:30:00Z: review.submitted by user:dr_smith
#   2025-01-15T11:30:05Z: score.updated by user:dr_smith

Retention & Archival

Retention Policies

Data Type	Minimum Retention	Maximum	Configurable
Audit Events	7 years	Indefinite	No
PHI Access Logs	7 years	Indefinite	No
System Events	3 years	7 years	Yes
Debug Logs	90 days	1 year	Yes

Archival Process

Best Practices

Enable Comprehensive Logging

Log all access, not just writes:

client.projects.update(
    project="patient-triage",
    audit_config={
        "log_reads": True,
        "log_searches": True,
        "include_query_params": True
    }
)

Regular Integrity Checks

Schedule periodic verification:

# Weekly integrity check
client.audit.schedule_verification(
    schedule="0 2 * * SUN",  # Sunday 2 AM
    scope="full_chain",
    alert_on_failure=True
)

Monitor for Anomalies

Set up baseline and detect deviations:

client.audit.enable_anomaly_detection(
    baseline_period_days=30,
    sensitivity="medium"
)

Next Steps

PHI Handling

How sensitive data is protected

System Overview

High-level platform architecture