Merge pull request #39 from SquidSec/feature/b06-json-logging

feat(ops): optional JSON structured logging

.env.example

@@ -25,3 +25,5 @@ SQUIDC5_SHELL_AUTO_STABILIZE=true
 # SQUIDC5_PLUGIN_SIGNING_SECRET=
 # At-rest encryption master key for LLM API keys (or data/secrets.key)
 # SQUIDC5_SECRETS_KEY=
+# JSON structured logs to stderr
+# SQUIDC5_LOG_JSON=true

src/squidc5/config.py

@@ -56,6 +56,7 @@ class Settings(BaseSettings):
     # Hardened defaults
     expose_health_details: bool = False
     security_headers: bool = True
+    log_json: bool = False
     # TLS: unique self-signed cert under data_dir/tls/ (ops UI, API, MCP)
     tls_enabled: bool = True
     tls_cert_file: Path | None = None  # override paths if set (both must be set)

src/squidc5/logging_setup.py

@@ -0,0 +1,42 @@
+"""Optional structured JSON logging."""
+
+from __future__ import annotations
+
+import json
+import logging
+import sys
+from datetime import UTC, datetime
+from typing import Any
+
+
+class JsonFormatter(logging.Formatter):
+    def format(self, record: logging.LogRecord) -> str:
+        payload: dict[str, Any] = {
+            "ts": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
+            "level": record.levelname,
+            "logger": record.name,
+            "msg": record.getMessage(),
+        }
+        if record.exc_info:
+            payload["exc_info"] = self.formatException(record.exc_info)
+        # Never attach common secret-like extras if present
+        for key in ("token", "api_key", "password", "authorization"):
+            if hasattr(record, key):
+                payload[key] = "[redacted]"
+        return json.dumps(payload, ensure_ascii=False)
+
+
+def configure_logging(*, json_logs: bool = False, debug: bool = False) -> None:
+    root = logging.getLogger()
+    level = logging.DEBUG if debug else logging.INFO
+    # Reset handlers for idempotent test/app start
+    root.handlers.clear()
+    handler = logging.StreamHandler(sys.stderr)
+    if json_logs:
+        handler.setFormatter(JsonFormatter())
+    else:
+        handler.setFormatter(
+            logging.Formatter("%(asctime)s %(levelname)s [%(name)s] %(message)s")
+        )
+    root.addHandler(handler)
+    root.setLevel(level)

src/squidc5/main.py

@@ -24,6 +24,7 @@ from squidc5.db.store import Database
 from squidc5.features import FeatureFlags
 from squidc5.implants.registry import ImplantRegistry
 from squidc5.listeners.manager import ListenerManager
+from squidc5.logging_setup import configure_logging
 from squidc5.mcp.server import build_mcp_router
 from squidc5.metrics.collector import MetricsCollector
 from squidc5.oast.store import OastService
@@ -36,10 +37,7 @@ from squidc5.profiles.engine import ProfileEngine
 from squidc5.sessions.manager import SessionManager
 from squidc5.tasking.manager import TaskManager
 
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
-)
+configure_logging(json_logs=False, debug=False)
 log = logging.getLogger("squidc5")
 
 
@@ -153,6 +151,7 @@ async def build_state(settings: Settings) -> AppState:
 
 def create_app(settings: Settings | None = None) -> FastAPI:
     settings = settings or get_settings()
+    configure_logging(json_logs=settings.log_json, debug=settings.debug)
     for w in settings.validate_runtime():
         log.warning("config: %s", w)
 
@@ -411,6 +410,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
 
 def cli() -> None:
     settings = get_settings()
+    configure_logging(json_logs=settings.log_json, debug=settings.debug)
     try:
         for w in settings.validate_runtime():
             log.warning("config: %s", w)

tests/test_logging_format.py

@@ -0,0 +1,50 @@
+"""JSON structured logging (B06)."""
+
+from __future__ import annotations
+
+import json
+import logging
+
+from squidc5.logging_setup import JsonFormatter, configure_logging
+
+
+def test_json_formatter_parses():
+    fmt = JsonFormatter()
+    record = logging.LogRecord(
+        name="squidc5.test",
+        level=logging.INFO,
+        pathname=__file__,
+        lineno=1,
+        msg="hello %s",
+        args=("world",),
+        exc_info=None,
+    )
+    line = fmt.format(record)
+    data = json.loads(line)
+    assert data["msg"] == "hello world"
+    assert data["level"] == "INFO"
+    assert data["logger"] == "squidc5.test"
+    assert "ts" in data
+
+
+def test_json_formatter_redacts_token_attr():
+    fmt = JsonFormatter()
+    record = logging.LogRecord(
+        name="squidc5.test",
+        level=logging.INFO,
+        pathname=__file__,
+        lineno=1,
+        msg="auth event",
+        args=(),
+        exc_info=None,
+    )
+    record.token = "sc5_should_not_leak"  # type: ignore[attr-defined]
+    data = json.loads(fmt.format(record))
+    assert data["token"] == "[redacted]"
+
+
+def test_configure_logging_json_mode():
+    configure_logging(json_logs=True, debug=False)
+    root = logging.getLogger()
+    assert root.handlers
+    assert isinstance(root.handlers[0].formatter, JsonFormatter)