fix(security): freeze CMCP_DEV_MODE at process startup (TEE-002)

src/cmcp_gateway/config.py

@@ -12,6 +12,11 @@
 
 from cmcp_gateway.errors import ConfigError
 
+# TEE-002: read exactly once at import time so the value is immutable for the
+# lifetime of the process. No code may call os.environ.get("CMCP_DEV_MODE")
+# after this point.
+DEV_MODE: bool = os.environ.get("CMCP_DEV_MODE", "0") == "1"
+
 
 class TEEProvider(StrEnum):
     TPM = "tpm"
@@ -136,7 +141,7 @@ def load_config(path: str) -> Config:
     if not isinstance(policy_reload_interval, int) or policy_reload_interval < 0:
         raise ConfigError("policy_reload_interval_seconds must be a non-negative integer")
 
-    dev_mode = os.environ.get("CMCP_DEV_MODE", "0") == "1"
+    dev_mode = DEV_MODE  # TEE-002: use the frozen constant, never re-read from env
     bearer_token = os.environ.get("CMCP_BEARER_TOKEN") or None
 
     policy_bundle_path = raw.get("policy_bundle_path", "policy/")

src/cmcp_gateway/session/state.py

@@ -104,3 +104,19 @@ def reset(self, *, reason: str, authorized_by: str) -> tuple[str, str]:
         self.catalog_drift = False
         # reason and authorized_by are logged by the caller in the audit chain
         return previous_session_id, self.session_id
+
+    def upgrade_attestation(self) -> tuple[str, str]:
+        """
+        Rotate the session token when attestation upgrades (e.g. software-only → hardware TEE).
+
+        Unlike reset(), session sensitivity state is preserved — the ongoing session
+        continues at its current sensitivity level. Only the session_id is rotated so
+        that any trust assertions cached against the old ID are invalidated.
+
+        Returns (previous_session_id, new_session_id). The caller is responsible for
+        writing an attestation_refresh audit entry.
+        """
+        previous_session_id = self.session_id
+        self.session_id = str(uuid4())
+        self.attestation_stale = False
+        return previous_session_id, self.session_id

tests/unit/test_session.py

@@ -139,3 +139,42 @@ async def _reset():
     tasks = [_update() for _ in range(5)] + [_reset() for _ in range(5)]
     await asyncio.gather(*tasks)
     assert state.max_sensitivity in SENSITIVITY_ORDER
+
+
+# ── AUTH-001: attestation upgrade rotates session token ───────────────────────
+
+def test_upgrade_attestation_rotates_session_id():
+    """AUTH-001: session token (session_id) must be rotated on attestation upgrade."""
+    state = SessionState(session_id="s1")
+    old_id, new_id = state.upgrade_attestation()
+    assert old_id == "s1"
+    assert new_id != "s1"
+    assert state.session_id == new_id
+
+
+def test_upgrade_attestation_clears_stale_flag():
+    state = SessionState(session_id="s1", attestation_stale=True)
+    state.upgrade_attestation()
+    assert state.attestation_stale is False
+
+
+def test_upgrade_attestation_preserves_sensitivity():
+    """Unlike reset(), upgrade_attestation() preserves accumulated session sensitivity."""
+    state = SessionState(session_id="s1")
+    state.update_from_inspection("c1", ["hipaa_phi"], False, True)
+    assert state.max_sensitivity == "hipaa_phi"
+    state.upgrade_attestation()
+    assert state.max_sensitivity == "hipaa_phi"
+
+
+def test_upgrade_attestation_preserves_injection_events():
+    state = SessionState(session_id="s1")
+    state.update_from_inspection("c1", [], injection_detected=True, response_allowed=False)
+    state.upgrade_attestation()
+    assert len(state.injection_events) == 1
+
+
+def test_upgrade_attestation_does_not_increment_reset_count():
+    state = SessionState(session_id="s1")
+    state.upgrade_attestation()
+    assert state.reset_count == 0

tests/unit/test_startup.py

@@ -43,6 +43,10 @@
 def complete_setup(tmp_path: Path, monkeypatch):
     """Set up a complete valid config + policy + catalog for startup tests."""
     monkeypatch.setenv("CMCP_DEV_MODE", "1")
+    # TEE-002: DEV_MODE is frozen at import; patch the constant directly so
+    # load_config() sees True even though the module was already imported.
+    import cmcp_gateway.config as _cfg
+    monkeypatch.setattr(_cfg, "DEV_MODE", True)
 
     # Config
     config_path = tmp_path / "cmcp-config.yaml"

tests/unit/test_tee_dev_mode_freeze.py

@@ -0,0 +1,85 @@
+"""TEE-002: CMCP_DEV_MODE must be frozen at process startup, not re-read at runtime."""
+
+from __future__ import annotations
+
+import importlib
+import sys
+
+
+def _reload_config_with_env(monkeypatch, value: str | None) -> object:
+    """
+    Reload cmcp_gateway.config with CMCP_DEV_MODE set to *value* (or absent if None),
+    and return the fresh module so we can inspect its DEV_MODE constant.
+    """
+    if value is None:
+        monkeypatch.delenv("CMCP_DEV_MODE", raising=False)
+    else:
+        monkeypatch.setenv("CMCP_DEV_MODE", value)
+
+    # Force a clean reimport so the module-level constant is re-evaluated.
+    sys.modules.pop("cmcp_gateway.config", None)
+    mod = importlib.import_module("cmcp_gateway.config")
+    return mod
+
+
+def test_dev_mode_constant_true_when_env_set_before_import(monkeypatch):
+    """TEE-002: DEV_MODE is True when CMCP_DEV_MODE=1 is present at import time."""
+    mod = _reload_config_with_env(monkeypatch, "1")
+    assert mod.DEV_MODE is True
+
+
+def test_dev_mode_constant_false_when_env_absent_at_import(monkeypatch):
+    """TEE-002: DEV_MODE is False when CMCP_DEV_MODE is absent at import time."""
+    mod = _reload_config_with_env(monkeypatch, None)
+    assert mod.DEV_MODE is False
+
+
+def test_dev_mode_constant_false_when_env_zero_at_import(monkeypatch):
+    """TEE-002: DEV_MODE is False when CMCP_DEV_MODE=0 at import time."""
+    mod = _reload_config_with_env(monkeypatch, "0")
+    assert mod.DEV_MODE is False
+
+
+def test_dev_mode_constant_not_changed_by_later_env_mutation(monkeypatch):
+    """TEE-002: mutating os.environ AFTER import must not change DEV_MODE."""
+    # Import with dev mode off.
+    mod = _reload_config_with_env(monkeypatch, "0")
+    assert mod.DEV_MODE is False
+
+    # Now set the env var — simulates an attacker injecting it at runtime.
+    monkeypatch.setenv("CMCP_DEV_MODE", "1")
+
+    # The constant on the already-imported module must remain False.
+    assert mod.DEV_MODE is False
+
+
+def test_dev_mode_constant_not_cleared_by_later_env_removal(monkeypatch):
+    """TEE-002: removing CMCP_DEV_MODE from os.environ after import must not clear DEV_MODE."""
+    # Import with dev mode on.
+    mod = _reload_config_with_env(monkeypatch, "1")
+    assert mod.DEV_MODE is True
+
+    # Remove the env var — the constant must stay True.
+    monkeypatch.delenv("CMCP_DEV_MODE", raising=False)
+    assert mod.DEV_MODE is True
+
+
+def test_load_config_uses_frozen_constant(monkeypatch, tmp_path):
+    """TEE-002: load_config reflects DEV_MODE constant, not a live env read."""
+    import textwrap
+
+    cfg_file = tmp_path / "cmcp-config.yaml"
+    cfg_file.write_text(textwrap.dedent("""
+        attestation:
+          provider: auto
+    """))
+
+    # Import with dev mode OFF, then enable env var afterward.
+    mod = _reload_config_with_env(monkeypatch, "0")
+    assert mod.DEV_MODE is False
+
+    monkeypatch.setenv("CMCP_DEV_MODE", "1")
+
+    cfg = mod.load_config(str(cfg_file))
+    # Config must reflect the frozen False, not the live env var.
+    assert cfg.dev_mode is False