Authentication & Privacy¶

{
  "sub": "user_2abc123",
  "org_id": "org_tenant_apollo_001",
  "org_role": "patient",
  "org_permissions": ["read:own_data", "write:own_data"],
  "iat": 1711900800,
  "exp": 1711904400,
  "iss": "https://clerk.curaway.com"
}

from clerk_backend_api import Clerk

clerk = Clerk(bearer_auth=CLERK_SECRET_KEY)

async def verify_token(authorization: str) -> TokenClaims:
    """Verify Clerk JWT and extract claims."""
    token = authorization.replace("Bearer ", "")
    claims = clerk.verify_token(token)
    return TokenClaims(
        user_id=claims["sub"],
        tenant_id=claims["org_id"],
        role=claims["org_role"],
    )

from functools import wraps

def require_role(*allowed_roles: str):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, claims: TokenClaims, **kwargs):
            if claims.role not in allowed_roles:
                raise HTTPException(403, "Insufficient permissions")
            return await func(*args, claims=claims, **kwargs)
        return wrapper
    return decorator

# Usage
@router.get("/admin/audit-logs")
@require_role("curaway_admin", "super_admin")
async def get_audit_logs(claims: TokenClaims = Depends(get_claims)):
    ...

async def tenant_middleware(request: Request, call_next):
    claims = request.state.claims
    header_tenant = request.headers.get("X-Tenant-ID")

    if header_tenant and header_tenant != claims.tenant_id:
        raise HTTPException(403, "Tenant ID mismatch")

    request.state.tenant_id = claims.tenant_id
    response = await call_next(request)
    return response

ALTER TABLE patients ENABLE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON patients
    USING (tenant_id = current_setting('app.current_tenant_id'));

-- Set tenant context per request
SET app.current_tenant_id = 'tenant-apollo-001';

CREATE TABLE consent_records (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    patient_id UUID NOT NULL REFERENCES patients(id),
    tenant_id VARCHAR(100) NOT NULL,
    purpose VARCHAR(50) NOT NULL,
    granted BOOLEAN NOT NULL,
    version INTEGER NOT NULL,
    granted_at TIMESTAMPTZ,
    revoked_at TIMESTAMPTZ,
    ip_address INET,
    user_agent TEXT,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Consent records are append-only: no UPDATE or DELETE
REVOKE UPDATE, DELETE ON consent_records FROM app_user;

async def verify_consent(patient_id: str, purpose: str, tenant_id: str) -> bool:
    """Check if patient has active consent for a specific purpose."""
    record = await db.fetch_one(
        """
        SELECT granted FROM consent_records
        WHERE patient_id = :pid AND purpose = :purpose AND tenant_id = :tid
        ORDER BY created_at DESC
        LIMIT 1
        """,
        {"pid": patient_id, "purpose": purpose, "tid": tenant_id},
    )
    return record and record["granted"]

from cryptography.fernet import Fernet

fernet = Fernet(ENCRYPTION_KEY.encode())

def encrypt_field(value: str) -> str:
    """Encrypt a sensitive field for storage."""
    return fernet.encrypt(value.encode()).decode()

def decrypt_field(encrypted: str) -> str:
    """Decrypt a sensitive field for display."""
    return fernet.decrypt(encrypted.encode()).decode()

@router.get("/patients/{patient_id}")
@require_role("patient", "curaway_admin", "super_admin")
async def get_patient(patient_id: str, claims: TokenClaims = Depends(get_claims)):
    patient = await db.get_patient(patient_id, claims.tenant_id)
    return {
        "id": patient.id,
        "full_name": decrypt_field(patient.full_name),
        "email": decrypt_field(patient.email),
        # Operational fields returned as-is
        "preferred_language": patient.preferred_language,
    }

async def execute_data_subject_deletion(patient_id: str, tenant_id: str) -> DeletionCertificate:
    """GDPR Article 17: Right to Erasure across all data stores."""
    job_id = str(uuid4())
    results = {}

    # 1. PostgreSQL — cascade delete via foreign keys
    results["postgresql"] = await delete_patient_postgres(patient_id, tenant_id)

    # 2. Neo4j — remove patient node and all relationships
    results["neo4j"] = await delete_patient_neo4j(patient_id)

    # 3. Qdrant — delete any patient-associated vectors
    results["qdrant"] = await delete_patient_vectors(patient_id)

    # 4. R2 — delete all uploaded documents
    results["r2"] = await delete_patient_documents(patient_id, tenant_id)

    # 5. Redis — flush cached data
    results["redis"] = await flush_patient_cache(patient_id)

    # Generate immutable deletion certificate
    certificate = DeletionCertificate(
        job_id=job_id,
        patient_id=patient_id,  # Retained only in certificate
        tenant_id=tenant_id,
        deleted_at=datetime.utcnow(),
        stores=results,
        all_successful=all(r["success"] for r in results.values()),
    )

    # Store certificate in audit log (retained for compliance)
    await store_deletion_certificate(certificate)

    return certificate

CREATE TABLE audit_logs (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id VARCHAR(100) NOT NULL,
    actor_id VARCHAR(100) NOT NULL,       -- who
    action VARCHAR(100) NOT NULL,          -- what
    resource_type VARCHAR(50) NOT NULL,    -- what type
    resource_id VARCHAR(100),              -- which resource
    details JSONB,                         -- additional context
    ip_address INET,                       -- where
    created_at TIMESTAMPTZ DEFAULT NOW()   -- when
);

-- Append-only: no UPDATE or DELETE allowed
REVOKE UPDATE, DELETE ON audit_logs FROM app_user;
REVOKE DELETE ON audit_logs FROM super_admin_user;

-- Index for efficient querying
CREATE INDEX idx_audit_tenant_created ON audit_logs(tenant_id, created_at DESC);
CREATE INDEX idx_audit_actor ON audit_logs(actor_id, created_at DESC);

async def audit_log(
    tenant_id: str,
    actor_id: str,
    action: str,
    resource_type: str,
    resource_id: str = None,
    details: dict = None,
    ip_address: str = None,
):
    await db.execute(
        """
        INSERT INTO audit_logs (tenant_id, actor_id, action, resource_type, resource_id, details, ip_address)
        VALUES (:tenant_id, :actor_id, :action, :resource_type, :resource_id, :details, :ip_address)
        """,
        {
            "tenant_id": tenant_id,
            "actor_id": actor_id,
            "action": action,
            "resource_type": resource_type,
            "resource_id": resource_id,
            "details": json.dumps(details) if details else None,
            "ip_address": ip_address,
        },
    )

CREATE TABLE legal_agreements (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id VARCHAR(100) NOT NULL,
    type VARCHAR(50) NOT NULL,        -- 'terms_of_service' or 'privacy_policy'
    version INTEGER NOT NULL,
    content TEXT NOT NULL,
    effective_date TIMESTAMPTZ NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    UNIQUE(tenant_id, type, version)
);

CREATE TABLE agreement_acceptances (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    patient_id UUID NOT NULL REFERENCES patients(id),
    agreement_id UUID NOT NULL REFERENCES legal_agreements(id),
    accepted_at TIMESTAMPTZ DEFAULT NOW(),
    ip_address INET,
    user_agent TEXT
);

@router.post("/patients")
async def create_patient(
    request: CreatePatientRequest,
    idempotency_key: str = Header(alias="X-Idempotency-Key"),
    claims: TokenClaims = Depends(get_claims),
):
    # Check for existing result with this key
    existing = await redis.get(f"idempotency:{idempotency_key}")
    if existing:
        return json.loads(existing)  # Return cached result

    # Process the request
    patient = await db.create_patient(request, claims.tenant_id)
    result = patient.dict()

    # Cache result for 24 hours
    await redis.set(f"idempotency:{idempotency_key}", json.dumps(result), ex=86400)
    return result

import uuid

async def correlation_id_middleware(request: Request, call_next):
    correlation_id = request.headers.get("X-Correlation-ID", str(uuid.uuid4()))
    request.state.correlation_id = correlation_id

    response = await call_next(request)
    response.headers["X-Correlation-ID"] = correlation_id
    return response

logger.info(
    "Patient created",
    extra={
        "correlation_id": request.state.correlation_id,
        "tenant_id": request.state.tenant_id,
        "patient_id": patient.id,
    },
)

async def security_headers_middleware(request: Request, call_next):
    response = await call_next(request)
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
    return response