feat(logging): Implement log sanitization to protect sensitive data

- Introduced log sanitization utilities in `log_sanitizer.py` to mask sensitive information in logs, including URLs, parameters, and tokens.
- Updated logging configuration to include a `SensitiveDataFilter` for automatic redaction of sensitive patterns.
- Enhanced logging statements across the application to utilize sanitization functions, ensuring sensitive data is not exposed in logs.
- Added comprehensive tests for log sanitization utilities to verify functionality and edge cases.

py-src/data_formulator/app.py

@@ -86,10 +86,15 @@ def configure_logging():
     log_level_str = os.getenv("LOG_LEVEL", "INFO").strip().upper()
     app_log_level = getattr(logging, log_level_str, logging.INFO)
 
+    from data_formulator.security.log_sanitizer import SensitiveDataFilter
+
+    handler = logging.StreamHandler(sys.stdout)
+    handler.addFilter(SensitiveDataFilter())
+
     logging.basicConfig(
         level=logging.WARNING,
         format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
-        handlers=[logging.StreamHandler(sys.stdout)]
+        handlers=[handler],
     )
 
     logging.getLogger('data_formulator').setLevel(app_log_level)
@@ -99,10 +104,13 @@ def configure_logging():
     logging.getLogger('openai').setLevel(logging.WARNING)
 
     app.logger.handlers = []
-    for handler in logging.getLogger().handlers:
-        app.logger.addHandler(handler)
+    for h in logging.getLogger().handlers:
+        app.logger.addHandler(h)
 
-    logging.getLogger('data_formulator').info(f"Log level: {log_level_str}")
+    logging.getLogger('data_formulator').info(
+        "Log level: %s (sanitize=%s)", log_level_str,
+        os.getenv("LOG_SANITIZE", "true"),
+    )
 
 
 _blueprints_registered = False

py-src/data_formulator/auth/gateways/github_gateway.py

@@ -88,7 +88,8 @@ def github_callback():
         "raw_token": access_token,
         "provider": "github",
     }
-    logger.info("GitHub login successful: user=%s (%s)", login, user_id)
+    logger.info("GitHub login successful: user=%s (id=%s...)",
+                 login, user_id[:8] if len(user_id) > 8 else user_id)
 
     return redirect("/")
 

py-src/data_formulator/auth/providers/oidc.py

@@ -45,6 +45,7 @@
 from flask import Flask, Request
 
 from .base import AuthProvider, AuthResult, AuthenticationError
+from data_formulator.security.log_sanitizer import sanitize_url, redact_token
 
 logger = logging.getLogger(__name__)
 
@@ -115,13 +116,14 @@ def _try_discovery(self, discovery_url: str) -> dict[str, Any] | None:
                 logger.warning(
                     "Discovery response from %s missing authorization_endpoint "
                     "(likely not a valid discovery document), ignoring",
-                    discovery_url,
+                    sanitize_url(discovery_url),
                 )
                 return None
 
             return doc
         except Exception as exc:
-            logger.warning("Discovery request failed for %s: %s", discovery_url, exc)
+            logger.warning("Discovery request failed for %s: %s",
+                           sanitize_url(discovery_url), type(exc).__name__)
             return None
 
     def on_configure(self, app: Flask) -> None:
@@ -160,7 +162,7 @@ def on_configure(self, app: Flask) -> None:
             if "id_token_signing_alg_values_supported" in discovery:
                 self._algorithms = discovery["id_token_signing_alg_values_supported"]
             self._discovery_ok = True
-            logger.info("Discovery succeeded for issuer %s", self._issuer)
+            logger.info("Discovery succeeded for issuer %s", sanitize_url(self._issuer))
         else:
             logger.warning(
                 "All discovery attempts failed — using manual endpoints"
@@ -170,18 +172,20 @@ def on_configure(self, app: Flask) -> None:
         if self._jwks_url:
             try:
                 self._jwks_client = PyJWKClient(self._jwks_url, cache_keys=True)
-                logger.info("JWKS client initialised: %s", self._jwks_url)
+                logger.info("JWKS client initialised: %s", sanitize_url(self._jwks_url))
             except Exception as exc:
-                logger.warning("Failed to initialise JWKS client: %s", exc)
+                logger.warning("Failed to initialise JWKS client: %s", type(exc).__name__)
 
         # 3) Summarise
         mode = "JWKS" if self._jwks_client else (
             "userinfo" if self._userinfo_url else "NONE"
         )
         logger.info(
-            "OIDC provider ready: issuer=%s, client_id=%s, "
+            "OIDC provider ready: issuer=%s, client_id=%s..., "
             "protocol=%s, discovery=%s, token_validation=%s",
-            self._issuer, self._client_id, self._protocol,
+            sanitize_url(self._issuer),
+            self._client_id[:4] + "..." if len(self._client_id) > 4 else self._client_id,
+            self._protocol,
             "ok" if self._discovery_ok else "manual",
             mode,
         )

py-src/data_formulator/data_loader/mssql_data_loader.py

@@ -117,7 +117,8 @@ def auth_instructions() -> str:
 **Troubleshooting:** Ensure SQL Server service is running. Verify TCP/IP is enabled in SQL Server Configuration Manager. Test with `sqlcmd -S <server> -d <database> -U <user> -P <password>`."""
 
     def __init__(self, params: dict[str, Any]):
-        log.info(f"Initializing MSSQL DataLoader with parameters: {params}")
+        from data_formulator.security.log_sanitizer import sanitize_params
+        log.info("Initializing MSSQL DataLoader with parameters: %s", sanitize_params(params))
 
         self.params = params
 

py-src/data_formulator/routes/agents.py

@@ -170,8 +170,10 @@ def check_available_models():
     all_public = model_registry.list_public()
     logger.info("=" * 60)
     logger.info(f"[check-available-models] Checking {len(all_public)} global models")
+    from data_formulator.security.log_sanitizer import sanitize_url
     for p in all_public:
-        logger.info(f"  -> {p['id']}  (endpoint={p['endpoint']}, model={p['model']}, api_base={p['api_base']})")
+        logger.info("  -> %s  (endpoint=%s, model=%s, api_base=%s)",
+                     p['id'], p['endpoint'], p['model'], sanitize_url(p.get('api_base', '')))
     overall_start = time.time()
 
     def _check_one(public_info: dict) -> dict:

py-src/data_formulator/security/log_sanitizer.py

@@ -0,0 +1,202 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Log sanitization utilities for preventing sensitive data leakage.
+
+Provides two layers of defense:
+
+1. **Explicit utilities** — ``sanitize_url``, ``sanitize_params``,
+   ``redact_token`` — called at logging call-sites for precise control.
+
+2. **SensitiveDataFilter** — a ``logging.Filter`` registered on handlers
+   as a safety net, automatically redacting patterns that slip through.
+
+Usage::
+
+    from data_formulator.security.log_sanitizer import (
+        sanitize_url, sanitize_params, redact_token,
+    )
+
+    logger.info("Connected to %s", sanitize_url(url))
+    logger.info("Params: %s", sanitize_params(params))
+    logger.debug("Token: %s", redact_token(token))
+
+The filter is registered in ``app.py:configure_logging()`` and can be
+disabled via ``LOG_SANITIZE=false`` for local debugging.
+"""
+
+from __future__ import annotations
+
+import copy
+import logging
+import os
+import re
+from typing import Any
+
+# ---------------------------------------------------------------------------
+# Sensitive key names (case-insensitive matching)
+# ---------------------------------------------------------------------------
+
+SENSITIVE_KEYS: frozenset[str] = frozenset({
+    "password", "passwd", "pwd",
+    "secret", "client_secret",
+    "token", "access_token", "refresh_token",
+    "api_key", "apikey", "api-key",
+    "access_key", "secret_key", "secret_access_key",
+    "credential", "credentials",
+    "private_key", "private-key",
+    "authorization",
+    "connection_string", "conn_str",
+})
+
+_REDACTED = "***"
+_REDACTED_TOKEN = "[REDACTED]"
+
+# ---------------------------------------------------------------------------
+# Pre-compiled regex patterns for the safety-net filter
+# ---------------------------------------------------------------------------
+
+# URL with embedded credentials: ://user:password@host
+_RE_URL_CREDS = re.compile(
+    r"(://[^/:@\s]+:)[^/@\s]+(@)",
+)
+
+# key=value patterns (password=xxx, api_key=xxx, secret=xxx, etc.)
+_SENSITIVE_KEY_NAMES = (
+    r"password|passwd|pwd|secret|client_secret"
+    r"|token|access_token|refresh_token"
+    r"|api_key|apikey|api[-_]key"
+    r"|access_key|secret_key|secret_access_key"
+    r"|credential|credentials"
+    r"|private_key|private[-_]key"
+    r"|authorization"
+    r"|connection_string|conn_str"
+)
+_RE_KEY_VALUE = re.compile(
+    rf"({_SENSITIVE_KEY_NAMES})"     # group 1: key name
+    r"(\s*[=:]\s*)"                  # group 2: separator (= or :)
+    r"(\S+)",                        # group 3: value
+    re.IGNORECASE,
+)
+
+# Bearer token: Bearer eyJxxx... (long opaque strings)
+_RE_BEARER = re.compile(
+    r"(Bearer\s+)\S{12,}",
+    re.IGNORECASE,
+)
+
+# Bare JWT-like strings: eyJ followed by 28+ base64url chars
+_RE_JWT_LIKE = re.compile(
+    r"\beyJ[A-Za-z0-9_-]{28,}",
+)
+
+# Python dict repr with sensitive keys: 'password': 'value' or "password": "value"
+_RE_DICT_SENSITIVE = re.compile(
+    rf"""(['"])({_SENSITIVE_KEY_NAMES})\1"""   # quoted key
+    r"""(\s*:\s*)"""                            # colon separator
+    r"""(['"])(.+?)\4""",                       # quoted value
+    re.IGNORECASE,
+)
+
+
+# ---------------------------------------------------------------------------
+# Explicit utility functions (Layer 1)
+# ---------------------------------------------------------------------------
+
+def sanitize_url(url: str) -> str:
+    """Mask credentials embedded in a URL.
+
+    ``https://user:s3cret@host/path`` → ``https://user:***@host/path``
+
+    Safe to call on URLs without credentials (returns unchanged).
+    """
+    if not url or "://" not in url:
+        return url
+    return _RE_URL_CREDS.sub(rf"\g<1>{_REDACTED}\2", url)
+
+
+def sanitize_params(params: dict[str, Any],
+                    extra_keys: frozenset[str] | set[str] | None = None,
+                    ) -> dict[str, Any]:
+    """Return a shallow copy of *params* with sensitive values masked.
+
+    Keys are matched case-insensitively against ``SENSITIVE_KEYS``
+    (plus *extra_keys* when provided).
+    """
+    keys_to_mask = SENSITIVE_KEYS | (extra_keys or frozenset())
+    sanitized = {}
+    for k, v in params.items():
+        if k.lower() in keys_to_mask:
+            sanitized[k] = _REDACTED
+        elif isinstance(v, dict):
+            sanitized[k] = sanitize_params(v, extra_keys)
+        else:
+            sanitized[k] = v
+    return sanitized
+
+
+def redact_token(token: str, visible: int = 4) -> str:
+    """Abbreviate a token to first/last *visible* characters.
+
+    Short tokens (≤ 2×visible) are fully masked.
+
+    ``"eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.xxx"``
+    → ``"eyJh...eCJ9"``  (visible=4)
+    """
+    if not token:
+        return ""
+    if len(token) <= visible * 2:
+        return _REDACTED_TOKEN
+    return f"{token[:visible]}...{token[-visible:]}"
+
+
+# ---------------------------------------------------------------------------
+# SensitiveDataFilter — logging.Filter safety net (Layer 2)
+# ---------------------------------------------------------------------------
+
+def _apply_patterns(text: str) -> str:
+    """Run all redaction patterns on *text* and return the sanitized version."""
+    # Order matters: most specific patterns first to avoid partial matches.
+    text = _RE_URL_CREDS.sub(rf"\g<1>{_REDACTED}\2", text)
+    text = _RE_BEARER.sub(rf"\g<1>{_REDACTED_TOKEN}", text)
+    text = _RE_KEY_VALUE.sub(rf"\g<1>\g<2>{_REDACTED}", text)
+    text = _RE_DICT_SENSITIVE.sub(
+        rf"\g<1>\g<2>\g<1>\g<3>\g<4>{_REDACTED}\g<4>",
+        text,
+    )
+    text = _RE_JWT_LIKE.sub(_REDACTED_TOKEN, text)
+    return text
+
+
+class SensitiveDataFilter(logging.Filter):
+    """Safety-net filter that auto-redacts sensitive patterns in log messages.
+
+    Attach to handlers in ``configure_logging()``::
+
+        handler.addFilter(SensitiveDataFilter())
+
+    Disable at runtime via ``LOG_SANITIZE=false``.
+    """
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        if os.environ.get("LOG_SANITIZE", "true").strip().lower() in (
+            "false", "0", "no",
+        ):
+            return True
+
+        # Format %-style args into the message so we can scan the full string.
+        # After formatting we clear args to prevent double-formatting by the
+        # handler's Formatter.
+        if record.args:
+            try:
+                record.msg = record.msg % record.args
+                record.args = None
+            except (TypeError, ValueError):
+                pass
+
+        record.msg = _apply_patterns(str(record.msg))
+
+        if record.exc_text:
+            record.exc_text = _apply_patterns(record.exc_text)
+
+        return True