audit-trail-test-author

{
  "event_id": "audit-2026-05-06T12:00:00Z-abc123",
  "timestamp": "2026-05-06T12:00:00.123Z",
  "event_type": "phi_access",
  "actor": {
    "user_id": "alice@example.com",
    "session_id": "sess-xyz",
    "ip_address": "203.0.113.42",
    "user_agent": "Mozilla/5.0..."
  },
  "subject": {
    "type": "patient_record",
    "id": "patient:123",
    "phi_type": "clinical_notes"
  },
  "action": "read",
  "outcome": "success",
  "context": {
    "request_id": "req-abc",
    "service": "ehr-api",
    "version": "1.2.3"
  },
  "previous_event_hash": "sha256:abc...",
  "this_event_hash": "sha256:def..."
}

def test_audit_log_structure_valid():
    user.access_phi(patient_id=123)
    audit = AuditLog.objects.latest('timestamp')
    # Required fields
    for field in ['event_id', 'timestamp', 'event_type', 'actor', 'subject', 'action', 'outcome']:
        assert field in audit.to_dict()
    # Actor structure
    for sub_field in ['user_id', 'session_id', 'ip_address', 'user_agent']:
        assert sub_field in audit.actor
    # Schema-validated
    assert validate_against_schema(audit, 'audit-event-v1.json-schema')

def test_audit_log_hash_chain_integrity():
    """Each event references the previous event's hash; tampering breaks the chain."""
    events = AuditLog.objects.all().order_by('timestamp')
    for i in range(1, len(events)):
        previous = events[i - 1]
        current = events[i]
        # current event's previous_event_hash must equal previous event's hash
        assert current.previous_event_hash == previous.this_event_hash, \
            f"Hash chain broken at event {current.event_id}"
        # Each event's hash is reproducible from its content
        recomputed = sha256(current.canonical_serialization())
        assert current.this_event_hash == recomputed, \
            f"Event {current.event_id} hash mismatch (tampered?)"

def test_audit_batch_signed_by_kms():
    """Daily batches signed by KMS; tampering invalidates signature."""
    today_batch = AuditBatch.objects.get(date=today())
    # Verify signature against KMS public key
    assert kms.verify_signature(today_batch.content, today_batch.signature)

def test_audit_log_append_only():
    """Old events must not be modifiable / deletable."""
    old = AuditLog.objects.first()
    # Database row-level: UPDATE on audit_log table fails for non-admin
    with pytest.raises(PermissionDenied):
        connection.cursor().execute(f"UPDATE audit_log SET action = 'X' WHERE id = {old.id}")
    with pytest.raises(PermissionDenied):
        connection.cursor().execute(f"DELETE FROM audit_log WHERE id = {old.id}")

def test_audit_log_retention_policy():
    """Logs older than retention period are archived, not deleted (PCI: 1 year online + 1 year archive minimum)."""
    expired = AuditLog.objects.filter(timestamp__lt=timezone.now() - timedelta(days=365)).first()
    if expired:
        # Should be in archive, not main store
        assert AuditArchive.objects.filter(event_id=expired.event_id).exists()

def test_pan_not_in_audit_logs():
    """PCI Req 3.4: PAN must not appear in audit logs."""
    user.access_card(card_id=123)
    audit_text = AuditLog.objects.latest('timestamp').full_event_text
    # No 13-19 digit sequences (typical PAN length)
    assert not re.search(r'\d{13,19}', audit_text)

def test_phi_redacted_in_audit():
    """HIPAA minimum-necessary: log access event but not PHI content."""
    user.access_patient_record(patient_id=123)
    audit = AuditLog.objects.latest('timestamp')
    # Should log: who, what (event type), when, on which subject
    assert 'patient:123' in audit.subject_id
    # Should NOT log: the actual clinical notes content
    assert 'diagnosis' not in audit.full_event_text

def test_can_reconstruct_user_activity_in_window():
    """SOC 2 CC7.3: detect security events; allows post-incident reconstruction."""
    user_id = 'alice@example.com'
    start = timezone.now() - timedelta(days=7)
    end = timezone.now()

    activity = AuditLog.objects.filter(
        actor__user_id=user_id,
        timestamp__range=[start, end],
    )
    # Activity is queryable + complete (no gaps)
    assert activity.count() > 0
    timestamps = sorted([e.timestamp for e in activity])
    # No suspicious gaps (e.g., gaps > 1 hour during active session)
    for i in range(1, len(timestamps)):
        gap = (timestamps[i] - timestamps[i - 1]).total_seconds()
        # Annotate: this would be domain-specific; just ensure queryable

def test_audit_events_aggregated_centrally():
    """Distributed audit events MUST land in central store within SLA."""
    # Trigger an event in microservice A
    response = service_a.access_phi(patient_id=123)
    event_id = response.headers['X-Event-ID']

    # Within 30 seconds, central store should have it
    deadline = time.time() + 30
    while time.time() < deadline:
        if CentralAuditStore.objects.filter(event_id=event_id).exists():
            return
        time.sleep(1)
    pytest.fail(f"Event {event_id} not aggregated to central store within 30s")