diff --git a/src/cmcp_gateway/config.py b/src/cmcp_gateway/config.py index f8748d73..67daee75 100644 --- a/src/cmcp_gateway/config.py +++ b/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/") diff --git a/src/cmcp_gateway/session/state.py b/src/cmcp_gateway/session/state.py index 1b23cbc9..3eba5bf2 100644 --- a/src/cmcp_gateway/session/state.py +++ b/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 diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index 298b44a2..f2810521 100644 --- a/tests/unit/test_session.py +++ b/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 diff --git a/tests/unit/test_startup.py b/tests/unit/test_startup.py index 140cbab7..0e706af0 100644 --- a/tests/unit/test_startup.py +++ b/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" diff --git a/tests/unit/test_tee_dev_mode_freeze.py b/tests/unit/test_tee_dev_mode_freeze.py new file mode 100644 index 00000000..cfb91dd1 --- /dev/null +++ b/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