M24: implement privacy governance

backend/alembic/versions/b913a72e8c14_customer_privacy_state.py

@@ -0,0 +1,24 @@
+"""add explicit customer anonymisation state
+
+Revision ID: b913a72e8c14
+Revises: a81d0ce9f662
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "b913a72e8c14"
+down_revision = "a81d0ce9f662"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("customers", sa.Column("anonymized_at", sa.DateTime(timezone=True)))
+    op.create_index("ix_customers_anonymized_at", "customers", ["anonymized_at"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_customers_anonymized_at", table_name="customers")
+    op.drop_column("customers", "anonymized_at")

backend/app/api/routers/audit.py

@@ -1,23 +1,30 @@
 from __future__ import annotations
 
+import csv
+import io
+import json
 import uuid
 from collections.abc import Sequence
-from datetime import datetime
+from datetime import UTC, datetime, timedelta
 from typing import Any
 
-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi.responses import Response
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
 from app.api.deps import get_db, require_operations_manager
+from app.core.config import get_settings
 from app.models.audit import AuditEvent
 from app.models.booking import Booking
 from app.models.customer import Customer
 from app.models.data_quality import DataQualityIssue
 from app.models.vehicle import Vehicle
 from app.schemas import AuditEventOut, AuditEventPageOut, CurrentUser
+from app.services.audit import record_audit_event
 
 router = APIRouter(prefix="/api/v1/audit", tags=["audit"])
+settings = get_settings()
 
 # Only entity types with a stable public reference and (optionally) a real frontend route
 # are resolved here. Types like "system", "knowledge" or "mcp_tool" carry no linkable
@@ -37,6 +44,74 @@ _ROUTE_TEMPLATES: dict[str, str] = {
 }
 
 
+@router.get("/export.csv")
+def export_audit_csv(
+    occurred_from: datetime | None = Query(default=None),
+    occurred_to: datetime | None = Query(default=None),
+    db: Session = Depends(get_db),
+    actor: CurrentUser = Depends(require_operations_manager),
+) -> Response:
+    end = occurred_to or datetime.now(UTC)
+    start = occurred_from or end - timedelta(days=30)
+    if end <= start or end - start > timedelta(days=90):
+        raise HTTPException(status_code=422, detail="Audit export range must be 1 to 90 days")
+    events = db.scalars(
+        select(AuditEvent)
+        .where(AuditEvent.occurred_at >= start, AuditEvent.occurred_at <= end)
+        .order_by(AuditEvent.occurred_at)
+        .limit(settings.privacy_audit_export_max_rows + 1)
+    ).all()
+    if len(events) > settings.privacy_audit_export_max_rows:
+        raise HTTPException(status_code=413, detail="Audit export exceeds configured row limit")
+    output = io.StringIO(newline="")
+    writer = csv.writer(output)
+    writer.writerow(
+        (
+            "id",
+            "occurred_at",
+            "actor_type",
+            "actor_label",
+            "action",
+            "entity_type",
+            "entity_id",
+            "correlation_id",
+            "before",
+            "after",
+            "metadata",
+        )
+    )
+    for event in events:
+        writer.writerow(
+            (
+                event.id,
+                event.occurred_at.isoformat(),
+                event.actor_type,
+                event.actor_label,
+                event.action,
+                event.entity_type,
+                event.entity_id or "",
+                event.correlation_id,
+                json.dumps(event.before_json, separators=(",", ":"), default=str),
+                json.dumps(event.after_json, separators=(",", ":"), default=str),
+                json.dumps(event.metadata_json, separators=(",", ":"), default=str),
+            )
+        )
+    record_audit_event(
+        db,
+        actor_type="user",
+        actor_label=actor.display_name,
+        action="audit_exported",
+        entity_type="audit",
+        metadata={"from": start.isoformat(), "to": end.isoformat(), "rows": len(events)},
+    )
+    db.commit()
+    return Response(
+        output.getvalue(),
+        media_type="text/csv; charset=utf-8",
+        headers={"Content-Disposition": 'attachment; filename="mobilityops-audit.csv"'},
+    )
+
+
 def _resolve_entity_refs(db: Session, events: Sequence[AuditEvent]) -> dict[uuid.UUID, str]:
     ids_by_type: dict[str, set[uuid.UUID]] = {}
     for event in events:

backend/app/api/routers/privacy.py

@@ -0,0 +1,166 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import JSONResponse
+from sqlalchemy import func, or_, select
+from sqlalchemy.orm import Session
+
+from app.api.deps import get_db, require_operations_manager
+from app.core.config import get_settings
+from app.models.booking import Booking
+from app.models.customer import Customer
+from app.schemas import (
+    CurrentUser,
+    CustomerAnonymizeRequest,
+    CustomerAnonymizeResult,
+    PrivacyRetentionOut,
+)
+from app.services.audit import record_audit_event
+
+router = APIRouter(prefix="/api/v1/privacy", tags=["privacy"])
+settings = get_settings()
+
+
+def _retention_cutoff() -> datetime:
+    return datetime.now(UTC) - timedelta(days=settings.privacy_minimum_booking_retention_days)
+
+
+def _customer_is_eligible(db: Session, customer_id) -> bool:
+    blocking = db.scalar(
+        select(func.count())
+        .select_from(Booking)
+        .where(
+            Booking.customer_id == customer_id,
+            or_(
+                Booking.status.in_(("reserved", "active")),
+                Booking.ends_at > _retention_cutoff(),
+            ),
+        )
+    )
+    return not blocking
+
+
+@router.get("/retention", response_model=PrivacyRetentionOut)
+def retention_status(
+    db: Session = Depends(get_db),
+    _user: CurrentUser = Depends(require_operations_manager),
+) -> PrivacyRetentionOut:
+    customers = db.scalars(select(Customer)).all()
+    return PrivacyRetentionOut(
+        minimum_booking_retention_days=settings.privacy_minimum_booking_retention_days,
+        audit_retention_days=settings.privacy_audit_retention_days,
+        customers_total=len(customers),
+        customers_anonymized=sum(customer.anonymized_at is not None for customer in customers),
+        customers_eligible=sum(
+            customer.anonymized_at is None and _customer_is_eligible(db, customer.id)
+            for customer in customers
+        ),
+    )
+
+
+@router.get("/customers/{public_ref}/export")
+def export_customer_data(
+    public_ref: str,
+    db: Session = Depends(get_db),
+    actor: CurrentUser = Depends(require_operations_manager),
+) -> JSONResponse:
+    customer = db.scalar(select(Customer).where(Customer.public_ref == public_ref))
+    if customer is None:
+        raise HTTPException(status_code=404, detail="Customer not found")
+    bookings = db.scalars(
+        select(Booking).where(Booking.customer_id == customer.id).order_by(Booking.starts_at)
+    ).all()
+    payload = {
+        "generated_at": datetime.now(UTC).isoformat(),
+        "customer": {
+            "public_ref": customer.public_ref,
+            "first_name": customer.first_name,
+            "last_name": customer.last_name,
+            "email": customer.email,
+            "phone": customer.phone,
+            "postal_code": customer.postal_code,
+            "city": customer.city,
+            "date_of_birth": customer.date_of_birth.isoformat() if customer.date_of_birth else None,
+            "anonymized_at": customer.anonymized_at.isoformat() if customer.anonymized_at else None,
+        },
+        "bookings": [
+            {
+                "public_ref": booking.public_ref,
+                "starts_at": booking.starts_at.isoformat(),
+                "ends_at": booking.ends_at.isoformat(),
+                "status": booking.status,
+            }
+            for booking in bookings
+        ],
+    }
+    record_audit_event(
+        db,
+        actor_type="user",
+        actor_label=actor.display_name,
+        action="privacy_customer_exported",
+        entity_type="customer",
+        entity_id=customer.id,
+        metadata={"customer_ref": customer.public_ref, "booking_count": len(bookings)},
+    )
+    db.commit()
+    return JSONResponse(
+        payload,
+        headers={"Content-Disposition": f'attachment; filename="{public_ref}-privacy.json"'},
+    )
+
+
+@router.post("/customers/{public_ref}/anonymize", response_model=CustomerAnonymizeResult)
+def anonymize_customer(
+    public_ref: str,
+    body: CustomerAnonymizeRequest,
+    db: Session = Depends(get_db),
+    actor: CurrentUser = Depends(require_operations_manager),
+) -> CustomerAnonymizeResult:
+    customer = db.scalar(
+        select(Customer).where(Customer.public_ref == public_ref).with_for_update()
+    )
+    if customer is None:
+        raise HTTPException(status_code=404, detail="Customer not found")
+    if body.confirmation != public_ref:
+        raise HTTPException(
+            status_code=422, detail="Customer reference confirmation does not match"
+        )
+    if customer.anonymized_at is not None:
+        return CustomerAnonymizeResult(
+            public_ref=public_ref,
+            anonymized_at=customer.anonymized_at,
+            status="already_anonymized",
+        )
+    if not _customer_is_eligible(db, customer.id):
+        raise HTTPException(
+            status_code=409,
+            detail="Customer has an active/recent booking within the minimum retention period",
+        )
+    anonymized_at = datetime.now(UTC)
+    customer.first_name = "Anoniem"
+    customer.last_name = public_ref
+    customer.email = None
+    customer.phone = None
+    customer.postal_code = None
+    customer.city = None
+    customer.date_of_birth = None
+    customer.anonymized_at = anonymized_at
+    record_audit_event(
+        db,
+        actor_type="user",
+        actor_label=actor.display_name,
+        action="privacy_customer_anonymized",
+        entity_type="customer",
+        entity_id=customer.id,
+        before={"anonymized": False},
+        after={"anonymized": True},
+        metadata={"reason": body.reason, "customer_ref": public_ref},
+    )
+    db.commit()
+    return CustomerAnonymizeResult(
+        public_ref=public_ref,
+        anonymized_at=anonymized_at,
+        status="anonymized",
+    )

backend/app/api/routers/search.py

@@ -89,6 +89,12 @@ _SECTIONS: list[dict] = [
         "terms": ["audit", "history", "geschiedenis", "historique"],
         "role": "operations_manager",
     },
+    {
+        "id": "privacy",
+        "link": "/privacy",
+        "terms": ["privacy", "retention", "anonymise", "anonimiseren", "confidentialité"],
+        "role": "operations_manager",
+    },
 ]
 
 

backend/app/core/config.py

@@ -63,6 +63,9 @@ class Settings(BaseSettings):
     oidc_default_role: str = "rental_employee"
     log_level: str = "INFO"
     metrics_bearer_token: str = ""
+    privacy_minimum_booking_retention_days: int = 30
+    privacy_audit_retention_days: int = 2555
+    privacy_audit_export_max_rows: int = 10000
 
 
 @lru_cache

backend/app/main.py

@@ -21,6 +21,7 @@ from app.api.routers import (
     knowledge,
     mcp_integrations,
     observability,
+    privacy,
     search,
     users,
     vehicles,
@@ -183,3 +184,4 @@ app.include_router(search.router)
 app.include_router(integration_status.router)
 app.include_router(users.router)
 app.include_router(observability.router)
+app.include_router(privacy.router)

backend/app/models/customer.py

@@ -1,7 +1,7 @@
 import uuid
-from datetime import date
+from datetime import date, datetime
 
-from sqlalchemy import Date, ForeignKey, String
+from sqlalchemy import Date, DateTime, ForeignKey, String
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -23,3 +23,4 @@ class Customer(UUIDPrimaryKeyMixin, TimestampMixin, Base):
     merged_into_customer_id: Mapped[uuid.UUID | None] = mapped_column(
         UUID(as_uuid=True), ForeignKey("customers.id")
     )
+    anonymized_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

backend/app/schemas.py

@@ -384,6 +384,25 @@ class OidcStatusOut(BaseModel):
     provider_name: str | None = None
 
 
+class PrivacyRetentionOut(BaseModel):
+    minimum_booking_retention_days: int
+    audit_retention_days: int
+    customers_total: int
+    customers_anonymized: int
+    customers_eligible: int
+
+
+class CustomerAnonymizeRequest(BaseModel):
+    confirmation: str = Field(min_length=1, max_length=20)
+    reason: str = Field(min_length=8, max_length=500)
+
+
+class CustomerAnonymizeResult(BaseModel):
+    public_ref: str
+    anonymized_at: datetime
+    status: Literal["anonymized", "already_anonymized"]
+
+
 class N8nWorkflowEvidence(BaseModel):
     name: str
     built: bool

backend/tests/test_privacy.py

@@ -0,0 +1,96 @@
+from sqlalchemy import delete, select
+
+from app.core.db import SessionLocal
+from app.models.audit import AuditEvent
+from app.models.customer import Customer
+
+
+def test_privacy_retention_uses_persisted_counts(ops_client):
+    response = ops_client.get("/api/v1/privacy/retention")
+    assert response.status_code == 200
+    body = response.json()
+    assert body["customers_total"] >= 180
+    assert body["minimum_booking_retention_days"] == 30
+    assert body["audit_retention_days"] == 2555
+
+
+def test_privacy_retention_requires_manager(employee_client):
+    assert employee_client.get("/api/v1/privacy/retention").status_code == 403
+
+
+def test_customer_export_is_downloadable_and_audited_without_mutation(ops_client):
+    response = ops_client.get("/api/v1/privacy/customers/CUS-0001/export")
+    assert response.status_code == 200
+    assert response.headers["content-disposition"] == 'attachment; filename="CUS-0001-privacy.json"'
+    assert response.json()["customer"]["public_ref"] == "CUS-0001"
+    assert isinstance(response.json()["bookings"], list)
+    events = ops_client.get("/api/v1/audit", params={"action": "privacy_customer_exported"}).json()
+    assert any(event["metadata"]["customer_ref"] == "CUS-0001" for event in events)
+
+
+def test_active_customer_cannot_be_anonymized(ops_client):
+    response = ops_client.post(
+        "/api/v1/privacy/customers/CUS-0042/anonymize",
+        json={"confirmation": "CUS-0042", "reason": "Verified erasure request"},
+    )
+    assert response.status_code == 409
+
+
+def test_eligible_customer_is_irreversibly_anonymized_without_pii_in_audit(ops_client):
+    with SessionLocal() as db:
+        customer = Customer(
+            public_ref="CUS-PRIVACY",
+            first_name="Private",
+            last_name="Person",
+            email="private.person@example.test",
+            phone="+32000000000",
+            postal_code="1000",
+            city="Brussels",
+            date_of_birth=None,
+        )
+        db.add(customer)
+        db.commit()
+        customer_id = customer.id
+    try:
+        mismatch = ops_client.post(
+            "/api/v1/privacy/customers/CUS-PRIVACY/anonymize",
+            json={"confirmation": "CUS-WRONG", "reason": "Verified erasure request"},
+        )
+        assert mismatch.status_code == 422
+        response = ops_client.post(
+            "/api/v1/privacy/customers/CUS-PRIVACY/anonymize",
+            json={"confirmation": "CUS-PRIVACY", "reason": "Verified erasure request"},
+        )
+        assert response.status_code == 200
+        assert response.json()["status"] == "anonymized"
+        with SessionLocal() as db:
+            customer = db.scalar(select(Customer).where(Customer.id == customer_id))
+            assert customer is not None
+            assert customer.email is None
+            assert customer.phone is None
+            assert customer.first_name == "Anoniem"
+            event = db.scalar(
+                select(AuditEvent)
+                .where(AuditEvent.action == "privacy_customer_anonymized")
+                .order_by(AuditEvent.occurred_at.desc())
+            )
+            assert event is not None
+            serialized = f"{event.before_json}{event.after_json}{event.metadata_json}"
+            assert "private.person@example.test" not in serialized
+    finally:
+        with SessionLocal() as db:
+            db.execute(delete(AuditEvent).where(AuditEvent.entity_id == customer_id))
+            db.execute(delete(Customer).where(Customer.id == customer_id))
+            db.commit()
+
+
+def test_audit_csv_export_is_bounded_and_audited(ops_client):
+    response = ops_client.get("/api/v1/audit/export.csv")
+    assert response.status_code == 200
+    assert response.headers["content-type"].startswith("text/csv")
+    assert response.text.startswith("id,occurred_at,actor_type")
+    invalid = ops_client.get(
+        "/api/v1/audit/export.csv",
+        params={"occurred_from": "2025-01-01T00:00:00Z", "occurred_to": "2026-01-01T00:00:00Z"},
+    )
+    assert invalid.status_code == 422