Security hardening: logging, cost protection, SSRF, model validation (#248)

* Security hardening: logging, cost protection

- Wire up SecureFormatter in app startup (#65): call
  configure_secure_logging() before any logging occurs
- Add cost manipulation protection (#67): block models
  above $15/1M input tokens on platform/community keys,
  warn above $5/1M; BYOK users unrestricted
- Verified SSRF protection (#66) and model validation (#68)
  already have comprehensive test coverage

Closes #65, closes #66, closes #67, closes #68

* Address PR review findings

- Fix misleading "fallback rate" comment in _check_model_cost
- Add logging for unknown models (operator visibility)
- Extract _models_by_cost() test helper to reduce duplication
- Add boundary test at exact block threshold
- Add BYOK + unknown model test
- Assert BYOK guidance in error message
- Fix module docstring wording

* Fix SecureJSONFormatter broad exception catch

Split the catch-all Exception handler into specific expected
errors (ValueError, TypeError, KeyError) that include context
for debugging, and unexpected errors that re-raise after
printing to stderr. Matches the pattern already used in
SecureFormatter.format().

Author: neuromechanist

src/api/main.py

@@ -25,9 +25,14 @@
 from src.api.routers.widget_test import router as widget_test_router
 from src.api.scheduler import start_scheduler, stop_scheduler
 from src.assistants import discover_assistants, registry
+from src.core.logging import configure_secure_logging
 from src.metrics.db import init_metrics_db
 from src.metrics.middleware import MetricsMiddleware
 
+# Configure secure logging before any other logging occurs.
+# This ensures all log output uses SecureFormatter which redacts API keys.
+configure_secure_logging()
+
 logger = logging.getLogger(__name__)
 
 # Discover assistants at module load time to populate registry

src/api/routers/community.py

@@ -34,7 +34,7 @@
 from src.assistants.registry import AssistantInfo
 from src.core.config.community import WidgetConfig
 from src.core.services.litellm_llm import create_openrouter_llm
-from src.metrics.cost import estimate_cost
+from src.metrics.cost import COST_BLOCK_THRESHOLD, COST_WARN_THRESHOLD, MODEL_PRICING, estimate_cost
 from src.metrics.db import (
     RequestLogEntry,
     extract_token_usage,
@@ -602,6 +602,48 @@ def _select_model(
     return (default_model, default_provider)
 
 
+def _check_model_cost(model: str, key_source: str) -> None:
+    """Check if a model's cost exceeds platform thresholds.
+
+    Only enforced when using platform or community API keys (not BYOK).
+    Logs a warning for moderately expensive models and blocks very expensive ones.
+
+    Args:
+        model: Model identifier (e.g., "openai/gpt-4o").
+        key_source: One of "byok", "community", or "platform".
+
+    Raises:
+        HTTPException(403): If model cost exceeds the block threshold.
+    """
+    if key_source == "byok":
+        return
+
+    pricing = MODEL_PRICING.get(model)
+    if pricing is None:
+        logger.info("Model %s not in pricing table; allowing without cost check", model)
+        return
+    input_rate = pricing[0]
+
+    if input_rate >= COST_BLOCK_THRESHOLD:
+        raise HTTPException(
+            status_code=403,
+            detail=(
+                f"Model '{model}' costs ${input_rate:.2f}/1M input tokens, "
+                f"which exceeds the platform limit of ${COST_BLOCK_THRESHOLD:.2f}/1M. "
+                "To use expensive models, provide your own API key via the "
+                "X-OpenRouter-Key header. Get a key at: https://openrouter.ai/keys"
+            ),
+        )
+
+    if input_rate >= COST_WARN_THRESHOLD:
+        logger.warning(
+            "Model %s costs $%.2f/1M input tokens (warn threshold: $%.2f)",
+            model,
+            input_rate,
+            COST_WARN_THRESHOLD,
+        )
+
+
 def _derive_user_id(token: str) -> str:
     """Derive a stable user ID from API token for cache optimization.
 
@@ -717,6 +759,10 @@ def create_community_assistant(
     selected_model, selected_provider = _select_model(
         community_info, requested_model, has_byok=bool(byok)
     )
+
+    # Block expensive models on platform/community keys
+    _check_model_cost(selected_model, key_source)
+
     logger.debug(
         "Using model %s",
         selected_model,

src/core/logging.py

@@ -141,15 +141,23 @@ def format(self, record: logging.LogRecord) -> str:
 
             return json_str
 
-        except Exception as e:
-            # Fallback to safe error message
+        except (ValueError, TypeError, KeyError) as e:
+            # Expected serialization errors - include context for debugging
+            safe_msg = str(getattr(record, "msg", "<no message>"))[:200]
+            safe_name = getattr(record, "name", "<unknown>")
             error_entry = {
                 "timestamp": datetime.now(UTC).isoformat(),
                 "level": "ERROR",
                 "logger": "logging",
-                "message": f"[LOGGING ERROR: {type(e).__name__}]",
+                "message": f"[LOGGING ERROR: {type(e).__name__}: {e}]",
+                "original_logger": safe_name,
+                "original_message": safe_msg,
             }
             return json.dumps(error_entry)
+        except Exception as e:
+            # Unexpected errors - surface to stderr and re-raise
+            print(f"CRITICAL: Unexpected error in SecureJSONFormatter: {e}", file=sys.stderr)
+            raise
 
 
 def configure_secure_logging(

src/metrics/cost.py

@@ -40,6 +40,11 @@
 _FALLBACK_INPUT_RATE = 1.00  # USD per 1M tokens
 _FALLBACK_OUTPUT_RATE = 3.00  # USD per 1M tokens
 
+# Cost protection thresholds (USD per 1M input tokens)
+# Applied only when using platform/community keys (not BYOK)
+COST_WARN_THRESHOLD = 5.0  # Log warning for models above this
+COST_BLOCK_THRESHOLD = 15.0  # Block requests for models above this
+
 
 def estimate_cost(
     model: str | None,

tests/test_api/test_cost_protection.py

@@ -0,0 +1,89 @@
+"""Tests for model cost protection.
+
+Verifies that expensive models are blocked when using platform/community keys,
+but allowed when users provide their own API key (BYOK).
+"""
+
+import pytest
+from fastapi import HTTPException
+
+from src.api.routers.community import _check_model_cost
+from src.metrics.cost import COST_BLOCK_THRESHOLD, COST_WARN_THRESHOLD, MODEL_PRICING
+
+
+def _models_by_cost(min_rate: float = 0.0, max_rate: float = float("inf")) -> list[str]:
+    """Return model names with input rates in [min_rate, max_rate)."""
+    return [m for m, (inp, _) in MODEL_PRICING.items() if min_rate <= inp < max_rate]
+
+
+class TestCheckModelCost:
+    """Tests for _check_model_cost() pre-invocation cost guard."""
+
+    def test_cheap_model_on_platform_key_allowed(self) -> None:
+        """Cheap models should be allowed on platform keys without error."""
+        cheap_models = _models_by_cost(max_rate=COST_WARN_THRESHOLD)
+        assert cheap_models, "Test requires at least one cheap model in MODEL_PRICING"
+
+        _check_model_cost(cheap_models[0], "platform")
+        _check_model_cost(cheap_models[0], "community")
+
+    def test_expensive_model_blocked_on_platform_key(self) -> None:
+        """Models above block threshold should be rejected with 403 on platform keys."""
+        expensive_models = _models_by_cost(min_rate=COST_BLOCK_THRESHOLD)
+        assert expensive_models, "Test requires at least one expensive model in MODEL_PRICING"
+
+        with pytest.raises(HTTPException) as exc_info:
+            _check_model_cost(expensive_models[0], "platform")
+        assert exc_info.value.status_code == 403
+        assert "exceeds the platform limit" in exc_info.value.detail
+        assert "openrouter.ai/keys" in exc_info.value.detail
+
+    def test_expensive_model_blocked_on_community_key(self) -> None:
+        """Models above block threshold should also be rejected on community keys."""
+        expensive_models = _models_by_cost(min_rate=COST_BLOCK_THRESHOLD)
+        assert expensive_models, "Test requires at least one expensive model in MODEL_PRICING"
+
+        with pytest.raises(HTTPException) as exc_info:
+            _check_model_cost(expensive_models[0], "community")
+        assert exc_info.value.status_code == 403
+
+    def test_expensive_model_allowed_with_byok(self) -> None:
+        """BYOK users should be able to use any model, even expensive ones."""
+        expensive_models = _models_by_cost(min_rate=COST_BLOCK_THRESHOLD)
+        assert expensive_models, "Test requires at least one expensive model in MODEL_PRICING"
+
+        _check_model_cost(expensive_models[0], "byok")
+
+    def test_unknown_model_allowed_on_platform_key(self) -> None:
+        """Unknown models (not in pricing table) should be allowed."""
+        _check_model_cost("unknown/made-up-model-xyz", "platform")
+
+    def test_unknown_model_allowed_with_byok(self) -> None:
+        """BYOK users with unknown models should also be allowed."""
+        _check_model_cost("unknown/made-up-model-xyz", "byok")
+
+    def test_warn_threshold_model_not_blocked(self) -> None:
+        """Models between warn and block thresholds should be allowed (just warned)."""
+        warn_only_models = _models_by_cost(
+            min_rate=COST_WARN_THRESHOLD, max_rate=COST_BLOCK_THRESHOLD
+        )
+        if not warn_only_models:
+            pytest.skip("No models between warn and block thresholds in current pricing")
+
+        _check_model_cost(warn_only_models[0], "platform")
+
+    def test_model_at_exact_block_threshold_is_blocked(self) -> None:
+        """A model priced exactly at the block threshold should be blocked."""
+        exact_models = [m for m, (inp, _) in MODEL_PRICING.items() if inp == COST_BLOCK_THRESHOLD]
+        if not exact_models:
+            pytest.skip("No model priced exactly at block threshold")
+
+        with pytest.raises(HTTPException) as exc_info:
+            _check_model_cost(exact_models[0], "platform")
+        assert exc_info.value.status_code == 403
+
+    def test_thresholds_are_sane(self) -> None:
+        """Sanity check: warn threshold should be lower than block threshold."""
+        assert COST_WARN_THRESHOLD < COST_BLOCK_THRESHOLD
+        assert COST_WARN_THRESHOLD > 0
+        assert COST_BLOCK_THRESHOLD > 0