🔐 Jules02: Security hardening — Mask secrets in standard logs and audit details (#1179)

* 🔐 Jules02: Security hardening — Mask secrets in standard logs and audit details

- Harden SecretMaskingProcessor to support standard Python logging via logging.Filter.
- Implement narrative detail redaction in AuditLogger.
- Apply global masking filter to root logger in main.py.
- Add comprehensive verification tests for standard logging and audit log masking.

Co-authored-by: xnessom <255288032+xnessom@users.noreply.github.com>

* feat: 🔐 Jules02: security hardening — mask secrets in standard logs and audit details

- Implement logging.Filter in SecretMaskingProcessor for standard library log redaction.
- Hardened AuditLogger.log to redact secrets from narrative details.
- Applied global masking filter to root logger in main.py.
- Updated SECURITY_FRAMEWORK.md with new masking documentation.
- Added comprehensive tests and fixed linting violations.

Co-authored-by: xnessom <255288032+xnessom@users.noreply.github.com>

---------

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
Co-authored-by: xnessom <255288032+xnessom@users.noreply.github.com>

SECURITY_FRAMEWORK.md

@@ -54,6 +54,7 @@ Comprehensive security framework addressing data protection, API security, deplo
 - **No Hardcoding**: All credentials stored in secure vault
 - **Environment Variables**: Use .env with local .gitignore.
 - **Type-safe Secrets**: Use Pydantic `SecretStr` for all credentials (MT5, Redis, MetaAPI) to ensure automatic masking in logs and serialization.
+- **Unified Masking Filter**: A global `logging.Filter` (`SecretMaskingProcessor`) intercepts and redacts secrets from standard library logs and narrative audit trail details.
 - **Secrets Rotation**: Automatic rotation every 30 days
 - **Audit Trail**: Track all credential access attempts
 - **Emergency Access**: Break-glass procedure for critical incidents

main.py

@@ -92,6 +92,9 @@ def configure_logging(level: str = "INFO") -> None:
         level=getattr(logging, level.upper(), logging.INFO),
     )
 
+    # Apply secret masking to standard logging calls
+    logging.getLogger().addFilter(get_masking_processor())
+
 
 # -- Trading loop ----------------------------------------------------------
 

src/core/audit_log.py

@@ -92,7 +92,11 @@ class AuditLogger:
         """
         Record a new audit entry.
         """
-        # Redact metadata before persisting to database
+        # Redact narrative details and metadata before persisting to database
+        redacted_details = details
+        if details:
+            redacted_details = get_masking_processor().redact_any(details)
+
         redacted_metadata = None
         if metadata:
             redacted_metadata = get_masking_processor().redact_any(metadata)
@@ -107,7 +111,7 @@ class AuditLogger:
             entry = AuditEntry(
                 actor=actor,
                 action=action,
-                details=details,
+                details=redacted_details,
                 trace_id=trace_id,
                 metadata_json=redacted_metadata,
             )

src/core/log_config.py

@@ -6,19 +6,23 @@ src/core/log_config.py
 from __future__ import annotations
 
 import contextlib
+import logging
 from typing import Any
 
 from src.core.config import TradingConfig
 
 
-class SecretMaskingProcessor:
+class SecretMaskingProcessor(logging.Filter):
     """
-    Structlog processor that masks sensitive values in log events.
+    Unified security processor that masks sensitive values in both
+    structlog events and standard logging records.
+
     Dynamically retrieves secrets from TradingConfig to ensure any
     SecretStr field is never leaked to logs.
     """
 
     def __init__(self, config: TradingConfig | None = None, mask: str = "[MASKED]") -> None:
+        super().__init__()
         self.mask = mask
         self.secrets: set[str] = set()
         self.sensitive_patterns = [
@@ -116,6 +120,22 @@ class SecretMaskingProcessor:
         """Structlog-compatible processor interface."""
         return self.redact_any(event_dict)
 
+    def filter(self, record: logging.LogRecord) -> bool:
+        """
+        Standard logging Filter interface.
+        Redacts secrets from the log message and its arguments.
+        """
+        if isinstance(record.msg, str):
+            record.msg = self.redact_any(record.msg)
+
+        if record.args:
+            if isinstance(record.args, dict):
+                record.args = self.redact_any(record.args)
+            elif isinstance(record.args, (list, tuple)):
+                record.args = tuple(self.redact_any(arg) for arg in record.args)
+
+        return True
+
 
 _masking_processor = SecretMaskingProcessor()
 

tests/test_security_hardening.py

@@ -69,6 +69,84 @@ def test_audit_logger_redaction(tmp_path):
         assert metadata["nested"]["password"] == "[MASKED]"
 
 
+def test_audit_logger_details_redaction(tmp_path):
+    db_file = tmp_path / "test_audit_details.db"
+    db_url = f"sqlite:///{db_file}"
+
+    # Reset singleton for testing
+    AuditLogger._instance = None
+    AuditLogger._initialized = False
+
+    from src.core.log_config import get_masking_processor
+
+    processor = get_masking_processor()
+    processor.secrets.add("SENSITIVE_DETAIL_123")
+
+    logger = AuditLogger(db_url=db_url)
+
+    # Log something sensitive in the narrative details
+    logger.log(
+        actor="test",
+        action="detail_action",
+        details="User accessed SENSITIVE_DETAIL_123 in a narrative",
+    )
+
+    # Check the database
+    from sqlalchemy import create_engine, select
+
+    from src.core.audit_log import AuditEntry
+
+    engine = create_engine(db_url)
+    with engine.connect() as conn:
+        result = conn.execute(select(AuditEntry.details)).fetchone()
+        details_masked = result[0]
+        assert "SENSITIVE_DETAIL_123" not in details_masked
+        assert "[MASKED]" in details_masked
+
+
+def test_standard_logging_masking():
+    import logging
+
+    from src.core.log_config import get_masking_processor
+
+    processor = get_masking_processor()
+    processor.secrets.add("LOG_SECRET_456")
+
+    root_logger = logging.getLogger()
+    # Remove existing filters to avoid duplicates if test runs multiple times
+    for f in root_logger.filters[:]:
+        root_logger.removeFilter(f)
+
+    root_logger.addFilter(processor)
+
+    # Standard logging call
+    logging.warning("Standard log with secret: %s", "LOG_SECRET_456")
+
+    # Capture output if necessary, but here we check the record itself if we used a mock handler
+    # Or just check if the message was modified in a custom handler
+    class TestHandler(logging.Handler):
+        def __init__(self):
+            super().__init__()
+            self.records = []
+
+        def emit(self, record):
+            self.records.append(record)
+
+    handler = TestHandler()
+    root_logger.addHandler(handler)
+
+    logging.error("Another secret: LOG_SECRET_456")
+
+    found = False
+    for rec in handler.records:
+        if "Another secret" in str(rec.msg):
+            assert "LOG_SECRET_456" not in str(rec.msg)
+            assert "[MASKED]" in str(rec.msg)
+            found = True
+
+    assert found
+
+
 @pytest.mark.skipif(sys.platform == "win32", reason="Permission check only on Linux/Mac")
 def test_config_validator_file_permissions(tmp_path, monkeypatch):
     # Create insecure file