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/tests/test_safe_logging.py

@@ -0,0 +1,74 @@
+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

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/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",
+]

backend/django/common/test_safe_logging.py

@@ -0,0 +1,122 @@
+import io
+import logging
+
+from celery.utils.log import get_task_logger
+from django.test import SimpleTestCase
+
+from common.safe_logging import REDACTED, REDACTED_BINARY, install_safe_logging, redact_value
+
+
+class SafeLoggingTests(SimpleTestCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        install_safe_logging()
+
+    def _capture(self, logger: logging.Logger, formatter: str = "%(levelname)s %(message)s"):
+        stream = io.StringIO()
+        handler = logging.StreamHandler(stream)
+        handler.setFormatter(logging.Formatter(formatter))
+        logger.handlers = [handler]
+        logger.propagate = False
+        logger.setLevel(logging.INFO)
+        self.addCleanup(logger.handlers.clear)
+        return stream
+
+    def test_recursive_redaction_covers_secret_pii_and_biometric_fields(self):
+        payload = {
+            "authorization": "Bearer top-secret",
+            "profile": {
+                "email": "ada@example.test",
+                "phone_number": "+233241234567",
+                "document_number": "GHA-123456789",
+                "safe_status": "pending_review",
+            },
+            "evidence": {
+                "selfie_image": "base64-sensitive-selfie",
+                "face_embedding": [0.1, 0.2, 0.3],
+                "document_storage_key": "tenant/evidence/front.jpg",
+            },
+            "binary": b"document-bytes",
+        }
+
+        redacted = redact_value(payload)
+
+        self.assertEqual(redacted["authorization"], REDACTED)
+        self.assertEqual(redacted["profile"]["email"], REDACTED)
+        self.assertEqual(redacted["profile"]["phone_number"], REDACTED)
+        self.assertEqual(redacted["profile"]["document_number"], REDACTED)
+        self.assertEqual(redacted["profile"]["safe_status"], "pending_review")
+        self.assertEqual(redacted["evidence"]["selfie_image"], REDACTED)
+        self.assertEqual(redacted["evidence"]["face_embedding"], REDACTED)
+        self.assertEqual(redacted["evidence"]["document_storage_key"], REDACTED)
+        self.assertEqual(redacted["binary"], REDACTED_BINARY)
+
+    def test_django_logger_redacts_message_arguments_and_structured_extra(self):
+        logger = logging.getLogger("django.identitycore.redaction-test")
+        stream = self._capture(logger, "%(message)s payload=%(payload)s")
+
+        logger.info(
+            "request rejected email=%s Authorization: Bearer %s",
+            "ada@example.test",
+            "secret-access-token",
+            extra={
+                "payload": {
+                    "password": "never-log-this",
+                    "selfie_image": "raw-biometric-data",
+                    "safe_reason": "credentials_missing",
+                }
+            },
+        )
+
+        output = stream.getvalue()
+        self.assertNotIn("ada@example.test", output)
+        self.assertNotIn("secret-access-token", output)
+        self.assertNotIn("never-log-this", output)
+        self.assertNotIn("raw-biometric-data", output)
+        self.assertIn("credentials_missing", output)
+        self.assertIn(REDACTED, output)
+
+    def test_celery_task_logger_uses_the_same_redaction_boundary(self):
+        logger = get_task_logger("identitycore.redaction-test")
+        stream = self._capture(logger, "%(message)s context=%(context)s")
+
+        logger.warning(
+            "worker retry token=%s",
+            "celery-secret-token",
+            extra={
+                "context": {
+                    "api_key": "provider-key",
+                    "ocr_text": "raw document text",
+                    "verification_id": "ver_safe_public_id",
+                }
+            },
+        )
+
+        output = stream.getvalue()
+        self.assertNotIn("celery-secret-token", output)
+        self.assertNotIn("provider-key", output)
+        self.assertNotIn("raw document text", output)
+        self.assertIn("ver_safe_public_id", output)
+
+    def test_exception_traceback_is_preserved_without_sensitive_values(self):
+        logger = logging.getLogger("identitycore.exception-redaction-test")
+        stream = self._capture(logger)
+
+        try:
+            raise RuntimeError(
+                "token=runtime-secret email=ada@example.test "
+                "phone=+233241234567 document_number=GHA-123456789"
+            )
+        except RuntimeError:
+            logger.exception("provider failed first_name=Ada")
+
+        output = stream.getvalue()
+        self.assertIn("Traceback", output)
+        self.assertIn("RuntimeError", output)
+        self.assertNotIn("runtime-secret", output)
+        self.assertNotIn("ada@example.test", output)
+        self.assertNotIn("+233241234567", output)
+        self.assertNotIn("GHA-123456789", output)
+        self.assertNotIn("first_name=Ada", output)
+        self.assertIn("first_name=[REDACTED]", output)