Centralize sensitive log redaction across services

backend/ai-service/Dockerfile

@@ -32,6 +32,7 @@ RUN uv export \
     && pip install --no-cache-dir -r /tmp/requirements-ai.txt \
     && pip install --no-cache-dir --force-reinstall opencv-contrib-python-headless==4.10.0.84
 
+COPY backend/shared /app/backend/shared
 COPY backend/ai-service /app/backend/ai-service
 
 WORKDIR /app/backend/ai-service

backend/ai-service/app/core/safe_logging.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+_BACKEND_ROOT = Path(__file__).resolve().parents[3]
+if str(_BACKEND_ROOT) not in sys.path:
+    sys.path.insert(0, str(_BACKEND_ROOT))
+
+from shared.logging_redaction import (  # noqa: E402
+    LOG_FORMAT_ERROR,
+    REDACTED,
+    REDACTED_BINARY,
+    install_safe_logging,
+    is_sensitive_key,
+    redact_text,
+    redact_value,
+    sanitize_log_record,
+)
+
+__all__ = [
+    "LOG_FORMAT_ERROR",
+    "REDACTED",
+    "REDACTED_BINARY",
+    "install_safe_logging",
+    "is_sensitive_key",
+    "redact_text",
+    "redact_value",
+    "sanitize_log_record",
+]

backend/ai-service/app/main.py

@@ -1,17 +1,23 @@
-from fastapi import FastAPI
+from app.core.safe_logging import install_safe_logging
 
-from app.core.auth import enforce_internal_token as _enforce_internal_token
-from app.routers.health import healthcheck, readiness, router as health_router
-from app.routers.processing import (
+# Install before importing processing modules so initialization failures and
+# exception paths cannot emit raw evidence or credentials.
+install_safe_logging()
+
+from fastapi import FastAPI  # noqa: E402
+
+from app.core.auth import enforce_internal_token as _enforce_internal_token  # noqa: E402
+from app.routers.health import healthcheck, readiness, router as health_router  # noqa: E402
+from app.routers.processing import (  # noqa: E402
     document_classify,
     document_ocr,
     document_quality,
     face_compare,
     liveness_check,
     router as processing_router,
 )
-from app.runtime import configure_runtime_environment
-from app.schemas.processing import (
+from app.runtime import configure_runtime_environment  # noqa: E402
+from app.schemas.processing import (  # noqa: E402
     AIResultResponse,
     DocumentClassificationRequest,
     DocumentOCRRequest,
@@ -21,7 +27,7 @@
     LivenessCheckRequest,
     ReadinessResponse,
 )
-from app.settings import get_settings
+from app.settings import get_settings  # noqa: E402
 
 
 settings = get_settings()

backend/ai-service/app/pipeline.py

@@ -31,9 +31,7 @@ class ProcessingError(RuntimeError):
 
 class MediaAssetNotFoundError(RuntimeError):
     def __init__(self, storage_key: str, bucket_name: str):
-        message = (
-            f"Media asset '{storage_key}' was not found in bucket '{bucket_name}'."
-        )
+        message = "Media asset was not found in configured storage (media_asset_not_found)."
         super().__init__(message)
         self.storage_key = storage_key
         self.bucket_name = bucket_name

backend/ai-service/tests/test_safe_logging.py

@@ -0,0 +1,125 @@
+import io
+import logging
+
+from app.core.safe_logging import REDACTED, install_safe_logging, redact_value
+
+
+def _capture(logger: logging.Logger, formatter: str = "%(message)s"):
+    stream = io.StringIO()
+    handler = logging.StreamHandler(stream)
+    handler.setFormatter(logging.Formatter(formatter))
+    logger.handlers = [handler]
+    logger.propagate = False
+    logger.setLevel(logging.INFO)
+    return stream
+
+
+def test_managed_ai_uses_shared_nested_redaction_corpus():
+    install_safe_logging()
+    payload = {
+        "authorization": "Bearer ai-secret",
+        "request": {
+            "email": "subject@example.test",
+            "document_number": "P1234567",
+            "safe_operation": "face_compare",
+        },
+        "biometrics": {
+            "face_embedding": [0.12, 0.34],
+            "selfie_image": "base64-selfie",
+        },
+    }
+
+    redacted = redact_value(payload)
+
+    assert redacted["authorization"] == REDACTED
+    assert redacted["request"]["email"] == REDACTED
+    assert redacted["request"]["document_number"] == REDACTED
+    assert redacted["request"]["safe_operation"] == "face_compare"
+    # A container explicitly named "biometrics" is sensitive as a whole. The
+    # redactor intentionally fails closed instead of retaining its structure.
+    assert redacted["biometrics"] == REDACTED
+
+
+def test_managed_ai_logger_redacts_structured_context_and_exception_text():
+    install_safe_logging()
+    logger = logging.getLogger("identitycore.ai.redaction-test")
+    stream = _capture(logger, "%(message)s context=%(context)s")
+
+    try:
+        raise ValueError(
+            "api_key=provider-secret email=subject@example.test "
+            "image_base64=raw-biometric"
+        )
+    except ValueError:
+        logger.exception(
+            "AI processing failed Authorization: Bearer %s",
+            "internal-shared-token",
+            extra={
+                "context": {
+                    "document_storage_key": "tenant/evidence/document.jpg",
+                    "ocr_text": "raw OCR contents",
+                    "operation": "document_ocr",
+                }
+            },
+        )
+
+    output = stream.getvalue()
+    assert "provider-secret" not in output
+    assert "subject@example.test" not in output
+    assert "raw-biometric" not in output
+    assert "internal-shared-token" not in output
+    assert "tenant/evidence/document.jpg" not in output
+    assert "raw OCR contents" not in output
+    assert "document_ocr" in output
+    assert "Traceback" in output
+
+
+def test_uvicorn_access_formatter_preserves_protocol_without_private_data():
+    from uvicorn.logging import AccessFormatter
+
+    install_safe_logging()
+    logger = logging.getLogger("uvicorn.access")
+    previous = (logger.handlers[:], logger.propagate, logger.level)
+    stream = _capture(logger)
+    logger.handlers[0].setFormatter(
+        AccessFormatter(
+            '%(client_addr)s - "%(request_line)s" %(status_code)s', use_colors=False
+        )
+    )
+    try:
+        logger.info(
+            '%s - "%s %s HTTP/%s" %d',
+            "192.0.2.1:1234",
+            "GET",
+            "/check?access_token=private-token",
+            "1.1",
+            200,
+        )
+        output = stream.getvalue()
+        assert "GET" in output
+        assert "200 OK" in output
+        assert "private-token" not in output
+        assert "192.0.2.1" not in output
+        assert "HTTP/1.1" in output
+    finally:
+        logger.handlers, logger.propagate, logger.level = previous
+
+
+def test_numpy_buffers_do_not_render_pixels_or_embeddings():
+    import numpy as np
+    from app.core.safe_logging import REDACTED_BINARY
+
+    for value in (
+        np.array([17, 18], dtype=np.uint8),
+        np.array([0.125, 0.25], dtype=np.float32),
+    ):
+        assert redact_value({"frame": value}) == {"frame": REDACTED_BINARY}
+
+
+def test_missing_media_error_does_not_expose_evidence_location():
+    from app.pipeline import MediaAssetNotFoundError
+
+    error = MediaAssetNotFoundError("tenant/evidence/private.jpg", "private-bucket")
+    assert "tenant/evidence/private.jpg" not in str(error)
+    assert "private-bucket" not in str(error)
+    assert "media_asset_not_found" in str(error)

backend/django/Dockerfile

@@ -22,6 +22,7 @@ RUN uv export \
     --output-file /tmp/requirements-django.txt \
     && pip install --no-cache-dir -r /tmp/requirements-django.txt
 
+COPY backend/shared /app/backend/shared
 COPY backend/django /app/backend/django
 COPY docs/openapi/identitycore-public-api.yaml /app/docs/openapi/identitycore-public-api.yaml
 

backend/django/apps/core/apps.py

@@ -5,3 +5,10 @@ class CoreConfig(AppConfig):
     default_auto_field = "django.db.models.BigAutoField"
     name = "apps.core"
     label = "core"
+
+    def ready(self) -> None:
+        # Install the redaction boundary after Django configures logging but before
+        # request/worker code can emit application records.
+        from common.safe_logging import install_safe_logging
+
+        install_safe_logging()

backend/django/apps/identity_documents/tasks.py

@@ -563,10 +563,10 @@ def process_identity_document_task(identity_document_id: str) -> str:
                 promote_upload_to_media_by_storage_key(capture.storage_key)
             except Exception as exc:
                 logger.warning(
-                    "Failed to promote document upload %s for verification %s: %s",
-                    capture.storage_key,
+                    "Failed to promote document upload for verification %s (%s)",
                     verification.public_id,
-                    exc,
+                    type(exc).__name__,
+                    extra={"storage_key": capture.storage_key},
                 )
                 record_audit_event(
                     tenant=verification.tenant,

backend/django/apps/identity_documents/tests.py

@@ -437,7 +437,10 @@ def test_process_identity_document_task_keeps_processing_when_promotion_fails(
         }
         mock_promote.side_effect = RuntimeError("storage unavailable")
 
-        result = process_identity_document_task(self.identity_document.public_id)
+        with self.assertLogs("apps.identity_documents.tasks", level="WARNING") as captured:
+            result = process_identity_document_task(self.identity_document.public_id)
+        self.assertNotIn(self.upload.storage_key, " ".join(captured.output))
+        self.assertEqual(captured.records[0].storage_key, "[REDACTED]")
 
         self.assertEqual(result, IdentityDocumentStatus.PROCESSED)
         self.identity_document.refresh_from_db()

backend/django/common/safe_logging.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+_BACKEND_ROOT = Path(__file__).resolve().parents[2]
+if str(_BACKEND_ROOT) not in sys.path:
+    sys.path.insert(0, str(_BACKEND_ROOT))
+
+from shared.logging_redaction import (  # noqa: E402
+    LOG_FORMAT_ERROR,
+    REDACTED,
+    REDACTED_BINARY,
+    install_safe_logging,
+    is_sensitive_key,
+    redact_text,
+    redact_value,
+    sanitize_log_record,
+)
+
+__all__ = [
+    "LOG_FORMAT_ERROR",
+    "REDACTED",
+    "REDACTED_BINARY",
+    "install_safe_logging",
+    "is_sensitive_key",
+    "redact_text",
+    "redact_value",
+    "sanitize_log_record",
+]