diff --git a/api/routes.py b/api/routes.py index 6716db2..da4fd08 100644 --- a/api/routes.py +++ b/api/routes.py @@ -594,6 +594,22 @@ def register_dual_route(server, method: str, path: str, handler) -> None: def _resolve_mae_profile() -> str: + try: + if __package__ and "." in __package__: + from ..services.effective_security_posture import ( + get_effective_security_posture, + ) + else: + from services.effective_security_posture import ( + get_effective_security_posture, + ) + posture = get_effective_security_posture(required=False) + if posture is not None: + return str(posture.mae_profile) + except ImportError: + # IMPORTANT: dependency-light import mode retains the accepted resolver below. + pass + profile = os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local").strip().lower() if profile in {"public", "hardened"}: return profile @@ -627,11 +643,23 @@ def register_routes(server) -> None: # Must run BEFORE any route or worker registration. try: try: + from ..services.effective_security_posture import ( + get_effective_security_posture, + resolve_effective_security_posture, + ) from ..services.startup_profile_gate import enforce_startup_gate except (ImportError, ValueError): + from services.effective_security_posture import ( + get_effective_security_posture, + resolve_effective_security_posture, + ) from services.startup_profile_gate import enforce_startup_gate - enforce_startup_gate() + posture = get_effective_security_posture(required=False) + if posture is None: + # Compatibility/direct-test invocation is not the process owner. + posture = resolve_effective_security_posture() + enforce_startup_gate(posture=posture) except RuntimeError: # CRITICAL: fail-closed. Never continue route registration after S56 # startup gate failure. diff --git a/services/capabilities.py b/services/capabilities.py index d056e57..3394a93 100644 --- a/services/capabilities.py +++ b/services/capabilities.py @@ -26,6 +26,15 @@ from .runtime_profile import get_runtime_profile API_VERSION = 1 +def _get_installed_posture(): + try: + from .effective_security_posture import get_effective_security_posture + + return get_effective_security_posture(required=False) + except ImportError: + return None + + def _get_control_plane_info() -> dict: """Build control-plane status for capabilities response.""" try: @@ -33,9 +42,14 @@ def _get_control_plane_info() -> dict: from .control_plane import get_blocked_surfaces, resolve_control_plane_mode - profile = os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local") - mode = resolve_control_plane_mode(profile) - blocked = get_blocked_surfaces(profile, mode) + posture = _get_installed_posture() + profile = ( + posture.deployment_profile + if posture is not None + else os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local") + ) + mode = resolve_control_plane_mode(profile, posture=posture) + blocked = get_blocked_surfaces(profile, mode, posture=posture) info = { "mode": mode.value, "blocked_surfaces": [sid for sid, _ in blocked], @@ -59,9 +73,14 @@ def get_capabilities() -> dict: Return capability surface for frontend probing. """ cp_info = _get_control_plane_info() + posture = _get_installed_posture() result = { "api_version": API_VERSION, - "runtime_profile": get_runtime_profile().value, + "runtime_profile": ( + posture.runtime_profile + if posture is not None + else get_runtime_profile().value + ), "control_plane": cp_info, "pack": { "name": PACK_NAME, diff --git a/services/connector_allowlist_posture.py b/services/connector_allowlist_posture.py index 7f022a3..fc3aee0 100644 --- a/services/connector_allowlist_posture.py +++ b/services/connector_allowlist_posture.py @@ -100,7 +100,9 @@ def _dedupe_keep_order(values: list[str]) -> list[str]: def evaluate_connector_allowlist_posture( environ: Optional[Mapping[str, str]] = None, ) -> Dict[str, Any]: - env: Mapping[str, str] = environ or os.environ + # IMPORTANT: an explicit empty mapping is an empty posture fixture, not a request + # to fall back to ambient process state. + env: Mapping[str, str] = os.environ if environ is None else environ active_platforms: list[str] = [] unguarded_platforms: list[str] = [] @@ -154,7 +156,7 @@ def evaluate_connector_allowlist_posture( def is_strict_connector_allowlist_profile( environ: Optional[Mapping[str, str]] = None, ) -> bool: - env: Mapping[str, str] = environ or os.environ + env: Mapping[str, str] = os.environ if environ is None else environ deployment_profile = (env.get("OPENCLAW_DEPLOYMENT_PROFILE") or "").strip().lower() runtime_profile = (env.get("OPENCLAW_RUNTIME_PROFILE") or "").strip().lower() return deployment_profile == "public" or runtime_profile == "hardened" diff --git a/services/control_plane.py b/services/control_plane.py index 8285477..330ba42 100644 --- a/services/control_plane.py +++ b/services/control_plane.py @@ -15,11 +15,16 @@ Enforcement rule: - profile=public + mode=EMBEDDED -> requires explicit override + warning. """ +from __future__ import annotations + import enum import logging import os from dataclasses import dataclass, field -from typing import Dict, FrozenSet, List, Optional, Tuple +from typing import TYPE_CHECKING, Dict, FrozenSet, List, Optional, Tuple + +if TYPE_CHECKING: + from .effective_security_posture import EffectiveSecurityPosture logger = logging.getLogger(__name__) @@ -64,7 +69,24 @@ HIGH_RISK_SURFACES: FrozenSet[Tuple[str, str]] = frozenset( # --------------------------------------------------------------------------- -def resolve_control_plane_mode(deployment_profile: str = "") -> ControlPlaneMode: +def _effective_posture( + posture: EffectiveSecurityPosture | None = None, +) -> EffectiveSecurityPosture | None: + if posture is not None: + return posture + try: + from .effective_security_posture import get_effective_security_posture + + return get_effective_security_posture(required=False) + except ImportError: + return None + + +def resolve_control_plane_mode( + deployment_profile: str = "", + *, + posture: EffectiveSecurityPosture | None = None, +) -> ControlPlaneMode: """ Determine the active control-plane mode. @@ -73,6 +95,10 @@ def resolve_control_plane_mode(deployment_profile: str = "") -> ControlPlaneMode 2. profile=public defaults to SPLIT. 3. Everything else defaults to EMBEDDED. """ + effective = _effective_posture(posture) + if effective is not None: + return ControlPlaneMode(effective.control_plane_mode) + explicit = os.environ.get(ENV_CONTROL_PLANE_MODE, "").lower().strip() if explicit == "split": return ControlPlaneMode.SPLIT @@ -86,12 +112,20 @@ def resolve_control_plane_mode(deployment_profile: str = "") -> ControlPlaneMode return ControlPlaneMode.EMBEDDED -def is_split_mode() -> bool: +def is_split_mode(*, posture: EffectiveSecurityPosture | None = None) -> bool: """Convenience check for split mode.""" + from .deployment_profile import evaluate_deployment_profile - profile = os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local") - return resolve_control_plane_mode(profile) == ControlPlaneMode.SPLIT + effective = _effective_posture(posture) + profile = ( + effective.deployment_profile + if effective is not None + else os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local") + ) + return ( + resolve_control_plane_mode(profile, posture=effective) == ControlPlaneMode.SPLIT + ) # --------------------------------------------------------------------------- @@ -102,6 +136,8 @@ def is_split_mode() -> bool: def get_blocked_surfaces( deployment_profile: str, mode: Optional[ControlPlaneMode] = None, + *, + posture: EffectiveSecurityPosture | None = None, ) -> List[Tuple[str, str]]: """ Return list of (surface_id, reason) blocked in current configuration. @@ -109,7 +145,11 @@ def get_blocked_surfaces( In public + split: all HIGH_RISK_SURFACES are blocked. In embedded or non-public: nothing blocked. """ - if mode is None: + effective = _effective_posture(posture) + if effective is not None: + deployment_profile = effective.deployment_profile + mode = ControlPlaneMode(effective.control_plane_mode) + elif mode is None: mode = resolve_control_plane_mode(deployment_profile) if deployment_profile == "public" and mode == ControlPlaneMode.SPLIT: @@ -118,10 +158,19 @@ def get_blocked_surfaces( return [] -def is_surface_blocked(surface_id: str) -> bool: +def is_surface_blocked( + surface_id: str, + *, + posture: EffectiveSecurityPosture | None = None, +) -> bool: """Check if a specific surface is blocked in current config.""" - profile = os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local") - blocked = get_blocked_surfaces(profile) + effective = _effective_posture(posture) + profile = ( + effective.deployment_profile + if effective is not None + else os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local") + ) + blocked = get_blocked_surfaces(profile, posture=effective) return any(sid == surface_id for sid, _ in blocked) @@ -146,7 +195,10 @@ class SplitPrereqReport: } -def validate_split_prerequisites() -> SplitPrereqReport: +def validate_split_prerequisites( + *, + posture: EffectiveSecurityPosture | None = None, +) -> SplitPrereqReport: """ Validate that all prerequisites for split mode are met. @@ -158,24 +210,38 @@ def validate_split_prerequisites() -> SplitPrereqReport: """ report = SplitPrereqReport() - url = os.environ.get(ENV_CONTROL_PLANE_URL, "").strip() - token = os.environ.get(ENV_CONTROL_PLANE_TOKEN, "").strip() + effective = _effective_posture(posture) + url_configured = ( + effective.control_plane_url_configured + if effective is not None + else bool(os.environ.get(ENV_CONTROL_PLANE_URL, "").strip()) + ) + token_configured = ( + effective.control_plane_token_configured + if effective is not None + else bool(os.environ.get(ENV_CONTROL_PLANE_TOKEN, "").strip()) + ) - if not url: + if not url_configured: report.passed = False report.errors.append( f"S62: Split mode requires {ENV_CONTROL_PLANE_URL} but it is not set." ) - if not token: + if not token_configured: report.passed = False report.errors.append( f"S62: Split mode requires {ENV_CONTROL_PLANE_TOKEN} but it is not set." ) # Check for compat override (dev-only, auditable) - compat = os.environ.get(ENV_SPLIT_COMPAT_OVERRIDE, "").lower().strip() - if compat in ("1", "true", "yes"): + compat_override = ( + effective.control_plane_compat_override + if effective is not None + else os.environ.get(ENV_SPLIT_COMPAT_OVERRIDE, "").lower().strip() + in ("1", "true", "yes") + ) + if compat_override: report.warnings.append( "S62: OPENCLAW_SPLIT_COMPAT_OVERRIDE is active. " "This bypasses split enforcement and is for dev-only use." @@ -184,7 +250,10 @@ def validate_split_prerequisites() -> SplitPrereqReport: return report -def enforce_control_plane_startup() -> Dict: +def enforce_control_plane_startup( + *, + posture: EffectiveSecurityPosture | None = None, +) -> Dict: """ Run control-plane startup validation. @@ -196,12 +265,18 @@ def enforce_control_plane_startup() -> Dict: Returns diagnostic dict for startup report. """ - profile = os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local") - mode = resolve_control_plane_mode(profile) - compat_override = os.environ.get(ENV_SPLIT_COMPAT_OVERRIDE, "").lower().strip() in ( - "1", - "true", - "yes", + effective = _effective_posture(posture) + profile = ( + effective.deployment_profile + if effective is not None + else os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local") + ) + mode = resolve_control_plane_mode(profile, posture=effective) + compat_override = ( + effective.control_plane_compat_override + if effective is not None + else os.environ.get(ENV_SPLIT_COMPAT_OVERRIDE, "").lower().strip() + in ("1", "true", "yes") ) result: Dict = { @@ -209,7 +284,7 @@ def enforce_control_plane_startup() -> Dict: "control_plane_mode": mode.value, "blocked_surfaces": [ {"id": sid, "reason": desc} - for sid, desc in get_blocked_surfaces(profile, mode) + for sid, desc in get_blocked_surfaces(profile, mode, posture=effective) ], "startup_passed": True, "errors": [], @@ -217,7 +292,7 @@ def enforce_control_plane_startup() -> Dict: } if profile == "public" and mode == ControlPlaneMode.SPLIT: - prereq = validate_split_prerequisites() + prereq = validate_split_prerequisites(posture=effective) if not prereq.passed: result["startup_passed"] = False result["errors"] = prereq.errors diff --git a/services/effective_security_posture.py b/services/effective_security_posture.py new file mode 100644 index 0000000..76fa072 --- /dev/null +++ b/services/effective_security_posture.py @@ -0,0 +1,503 @@ +"""Immutable process-static security posture contract (R232).""" + +from __future__ import annotations + +import os +import sys +import threading +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +SCHEMA_VERSION = 1 +_TRUTHY = frozenset({"1", "true", "yes", "on"}) +_FALSY = frozenset({"0", "false", "no", "off"}) +_CONTROL_PLANE_TRUTHY = frozenset({"1", "true", "yes"}) +_VALID_DEPLOYMENT_PROFILES = frozenset({"local", "lan", "public"}) +_VALID_WEBHOOK_MODES = frozenset({"bearer", "hmac", "bearer_or_hmac"}) +_installed_posture: EffectiveSecurityPosture | None = None +_posture_lock = threading.RLock() + + +@dataclass(frozen=True, slots=True, kw_only=True) +class PostureFinding: + severity: str + code: str + message: str + remediation: str = "" + + +@dataclass(frozen=True, slots=True, kw_only=True) +class EffectiveSecurityPosture: + schema_version: int + runtime_profile: str + deployment_profile: str + mae_profile: str + network_exposed: bool + admin_token_configured: bool + observability_token_configured: bool + dangerous_profile_override: bool + dangerous_bind_override: bool + localhost_no_origin_override: bool + allow_any_public_llm_host: bool + allow_insecure_base_url: bool + webhook_auth_mode: str + webhook_bearer_configured: bool + webhook_hmac_configured: bool + webhook_replay_protection_required: bool + remote_admin_enabled: bool + trust_x_forwarded_for: bool + trusted_proxies_configured: bool + callback_allow_hosts_configured: bool + external_tools_enabled: bool + registry_sync_enabled: bool + transforms_enabled: bool + bridge_enabled: bool + bridge_device_token_configured: bool + bridge_mtls_enabled: bool + bridge_device_cert_map_configured: bool + bridge_allowed_device_ids_configured: bool + public_shared_surface_acknowledged: bool + control_plane_mode: str + control_plane_url_configured: bool + control_plane_token_configured: bool + control_plane_prerequisites_satisfied: bool + control_plane_compat_override: bool + connector_active_platforms: tuple[str, ...] + connector_unguarded_platforms: tuple[str, ...] + connector_recommended_allowlist_vars: tuple[str, ...] + deployment_checks: tuple[PostureFinding, ...] + deployment_pass_codes: tuple[str, ...] + deployment_warn_codes: tuple[str, ...] + deployment_fail_codes: tuple[str, ...] + startup_profile_passed: bool + startup_profile_overridden: bool + startup_profile_violation_codes: tuple[str, ...] + blocked_surface_ids: tuple[str, ...] + decision_codes: tuple[str, ...] + reason_codes: tuple[str, ...] + + +def _read( + environ: Mapping[str, str], + primary: str, + legacy: str | None = None, + default: str = "", +) -> str: + try: + if primary in environ: + value = environ.get(primary, default) + elif legacy and legacy in environ: + value = environ.get(legacy, default) + else: + value = default + except Exception: + # CRITICAL: malformed environment providers must fail closed without echoing + # exception content or the attempted value into diagnostics. + raise ValueError("security posture input unavailable") from None + if value is None: + return "" + try: + return str(value) + except Exception: + raise ValueError("security posture input is not scalar") from None + + +def _normalized( + environ: Mapping[str, str], + primary: str, + legacy: str | None = None, + default: str = "", +) -> str: + return _read(environ, primary, legacy, default).strip().lower() + + +def _enabled( + environ: Mapping[str, str], + primary: str, + legacy: str | None = None, +) -> bool: + return _normalized(environ, primary, legacy) in _TRUTHY + + +def _configured( + environ: Mapping[str, str], + primary: str, + legacy: str | None = None, +) -> bool: + return bool(_read(environ, primary, legacy).strip()) + + +def _network_exposed_from_argv() -> bool: + # Preserve the accepted S41 heuristic exactly: only the explicit --listen flag + # changes this process-static decision. + return "--listen" in sys.argv + + +def _deployment_report(profile: str, environ: Mapping[str, str]): + try: + from .deployment_profile import evaluate_deployment_profile + except ImportError: # pragma: no cover - top-level compatibility mode + from services.deployment_profile import evaluate_deployment_profile + + return evaluate_deployment_profile(profile, environ) + + +def _connector_posture(environ: Mapping[str, str]) -> Mapping[str, Any]: + try: + from .connector_allowlist_posture import evaluate_connector_allowlist_posture + except ImportError: # pragma: no cover - top-level compatibility mode + from services.connector_allowlist_posture import ( + evaluate_connector_allowlist_posture, + ) + + return evaluate_connector_allowlist_posture(environ) + + +def _blocked_surface_ids(profile: str, mode: str) -> tuple[str, ...]: + if profile != "public" or mode != "split": + return () + # IMPORTANT: these are the stable scalar IDs from the S62 registry. Importing + # control_plane here would create a dependency cycle before R233 packages the domain. + return ( + "callback_egress", + "registry_sync", + "secrets_write", + "tool_execution", + "transforms_exec", + "webhook_execute", + ) + + +def _safe_finding(check: Any) -> PostureFinding: + message = str(check.message) + if str(check.code) == "DP-WEBHOOK-005": + # IMPORTANT: the legacy evaluator includes the raw invalid environment value. + # The immutable boundary retains the stable code but never the untrusted value. + message = "Unsupported webhook auth mode." + return PostureFinding( + severity=str(check.severity), + code=str(check.code), + message=message, + remediation=str(check.remediation), + ) + + +def resolve_effective_security_posture( + environ: Mapping[str, str] | None = None, + *, + network_exposed: bool | None = None, +) -> EffectiveSecurityPosture: + # IMPORTANT: an explicitly supplied empty mapping means empty input. Do not use + # `environ or os.environ`; doing so makes tests and lifecycle injection ambient. + env = os.environ if environ is None else environ + + resolved_network_exposed = ( + _network_exposed_from_argv() + if network_exposed is None + else bool(network_exposed) + ) + deployment_profile = _normalized( + env, "OPENCLAW_DEPLOYMENT_PROFILE", default="local" + ) + if deployment_profile not in _VALID_DEPLOYMENT_PROFILES: + raise ValueError("unsupported deployment profile") + + raw_runtime_profile = _normalized( + env, "OPENCLAW_RUNTIME_PROFILE", default="minimal" + ) + runtime_profile = "hardened" if raw_runtime_profile == "hardened" else "minimal" + mae_profile = ( + "hardened" + if runtime_profile == "hardened" and deployment_profile != "public" + else deployment_profile + ) + + report = _deployment_report(deployment_profile, env) + findings = tuple(_safe_finding(check) for check in report.checks) + pass_codes = tuple(item.code for item in findings if item.severity == "pass") + warn_codes = tuple(item.code for item in findings if item.severity == "warn") + fail_codes = tuple(item.code for item in findings if item.severity == "fail") + + dangerous_profile_override = _enabled( + env, "OPENCLAW_SECURITY_DANGEROUS_PROFILE_OVERRIDE" + ) + startup_violations = () if deployment_profile == "local" else fail_codes + startup_overridden = bool(startup_violations and dangerous_profile_override) + startup_passed = ( + deployment_profile == "local" or not startup_violations or startup_overridden + ) + + explicit_control_mode = _normalized(env, "OPENCLAW_CONTROL_PLANE_MODE") + if explicit_control_mode in {"embedded", "split"}: + control_plane_mode = explicit_control_mode + elif deployment_profile == "public": + control_plane_mode = "split" + else: + control_plane_mode = "embedded" + + control_plane_url_configured = _configured(env, "OPENCLAW_CONTROL_PLANE_URL") + control_plane_token_configured = _configured(env, "OPENCLAW_CONTROL_PLANE_TOKEN") + control_plane_prerequisites_satisfied = ( + control_plane_url_configured and control_plane_token_configured + ) + control_plane_compat_override = ( + _normalized(env, "OPENCLAW_SPLIT_COMPAT_OVERRIDE") in _CONTROL_PLANE_TRUTHY + ) + + connector = _connector_posture(env) + active_platforms = tuple( + sorted({str(item) for item in connector["active_platforms"]}) + ) + unguarded_platforms = tuple( + sorted({str(item) for item in connector["unguarded_platforms"]}) + ) + recommended_allowlist_vars = tuple( + sorted({str(item) for item in connector["recommended_allowlist_vars"]}) + ) + + reason_codes = list(startup_violations) + if deployment_profile == "public" and control_plane_mode == "split": + if not control_plane_url_configured: + reason_codes.append("CP-URL-MISSING") + if not control_plane_token_configured: + reason_codes.append("CP-TOKEN-MISSING") + elif deployment_profile == "public" and control_plane_mode == "embedded": + if not control_plane_compat_override: + reason_codes.append("CP-PUBLIC-EMBEDDED") + reason_codes.extend( + f"CONNECTOR-ALLOWLIST-{platform.upper()}" for platform in unguarded_platforms + ) + if raw_runtime_profile not in {"", "minimal", "hardened"}: + reason_codes.append("RUNTIME-PROFILE-DEFAULTED") + + decision_codes = [ + ( + "STARTUP-OVERRIDDEN" + if startup_overridden + else "STARTUP-PASS" if startup_passed else "STARTUP-DENY" + ), + ( + "CONTROL-PLANE-PASS" + if ( + deployment_profile != "public" + or ( + control_plane_mode == "split" + and control_plane_prerequisites_satisfied + ) + or (control_plane_mode == "embedded" and control_plane_compat_override) + ) + else "CONTROL-PLANE-DENY" + ), + ( + "CONNECTORS-NONE" + if not active_platforms + else "CONNECTORS-UNGUARDED" if unguarded_platforms else "CONNECTORS-GUARDED" + ), + "NETWORK-EXPOSED" if resolved_network_exposed else "NETWORK-LOOPBACK", + ] + + raw_webhook_mode = _normalized( + env, + "OPENCLAW_WEBHOOK_AUTH_MODE", + "MOLTBOT_WEBHOOK_AUTH_MODE", + ) + webhook_auth_mode = ( + raw_webhook_mode + if raw_webhook_mode in _VALID_WEBHOOK_MODES + else "unset" if not raw_webhook_mode else "invalid" + ) + replay_value = _normalized( + env, + "OPENCLAW_WEBHOOK_REQUIRE_REPLAY_PROTECTION", + "MOLTBOT_WEBHOOK_REQUIRE_REPLAY_PROTECTION", + ) + + return EffectiveSecurityPosture( + schema_version=SCHEMA_VERSION, + runtime_profile=runtime_profile, + deployment_profile=deployment_profile, + mae_profile=mae_profile, + network_exposed=resolved_network_exposed, + admin_token_configured=_configured( + env, "OPENCLAW_ADMIN_TOKEN", "MOLTBOT_ADMIN_TOKEN" + ), + observability_token_configured=_configured( + env, "OPENCLAW_OBSERVABILITY_TOKEN", "MOLTBOT_OBSERVABILITY_TOKEN" + ), + dangerous_profile_override=dangerous_profile_override, + dangerous_bind_override=_enabled( + env, + "OPENCLAW_SECURITY_DANGEROUS_BIND_OVERRIDE", + "MOLTBOT_SECURITY_DANGEROUS_BIND_OVERRIDE", + ), + localhost_no_origin_override=( + _normalized(env, "OPENCLAW_LOCALHOST_ALLOW_NO_ORIGIN") == "true" + ), + allow_any_public_llm_host=_enabled( + env, + "OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST", + "MOLTBOT_ALLOW_ANY_PUBLIC_LLM_HOST", + ), + allow_insecure_base_url=_enabled( + env, + "OPENCLAW_ALLOW_INSECURE_BASE_URL", + "MOLTBOT_ALLOW_INSECURE_BASE_URL", + ), + webhook_auth_mode=webhook_auth_mode, + webhook_bearer_configured=_configured( + env, + "OPENCLAW_WEBHOOK_BEARER_TOKEN", + "MOLTBOT_WEBHOOK_BEARER_TOKEN", + ), + webhook_hmac_configured=_configured( + env, + "OPENCLAW_WEBHOOK_HMAC_SECRET", + "MOLTBOT_WEBHOOK_HMAC_SECRET", + ), + webhook_replay_protection_required=replay_value not in _FALSY, + remote_admin_enabled=_enabled( + env, "OPENCLAW_ALLOW_REMOTE_ADMIN", "MOLTBOT_ALLOW_REMOTE_ADMIN" + ), + trust_x_forwarded_for=_enabled( + env, + "OPENCLAW_TRUST_X_FORWARDED_FOR", + "MOLTBOT_TRUST_X_FORWARDED_FOR", + ), + trusted_proxies_configured=_configured( + env, "OPENCLAW_TRUSTED_PROXIES", "MOLTBOT_TRUSTED_PROXIES" + ), + callback_allow_hosts_configured=_configured( + env, + "OPENCLAW_CALLBACK_ALLOW_HOSTS", + "MOLTBOT_CALLBACK_ALLOW_HOSTS", + ), + external_tools_enabled=_enabled(env, "OPENCLAW_ENABLE_EXTERNAL_TOOLS"), + registry_sync_enabled=_enabled(env, "OPENCLAW_ENABLE_REGISTRY_SYNC"), + transforms_enabled=_enabled(env, "OPENCLAW_ENABLE_TRANSFORMS"), + bridge_enabled=_enabled( + env, "OPENCLAW_BRIDGE_ENABLED", "MOLTBOT_BRIDGE_ENABLED" + ), + bridge_device_token_configured=_configured( + env, + "OPENCLAW_BRIDGE_DEVICE_TOKEN", + "MOLTBOT_BRIDGE_DEVICE_TOKEN", + ), + bridge_mtls_enabled=_enabled(env, "OPENCLAW_BRIDGE_MTLS_ENABLED"), + bridge_device_cert_map_configured=_configured( + env, "OPENCLAW_BRIDGE_DEVICE_CERT_MAP" + ), + bridge_allowed_device_ids_configured=_configured( + env, + "OPENCLAW_BRIDGE_ALLOWED_DEVICE_IDS", + "MOLTBOT_BRIDGE_ALLOWED_DEVICE_IDS", + ), + public_shared_surface_acknowledged=_enabled( + env, + "OPENCLAW_PUBLIC_SHARED_SURFACE_BOUNDARY_ACK", + "MOLTBOT_PUBLIC_SHARED_SURFACE_BOUNDARY_ACK", + ), + control_plane_mode=control_plane_mode, + control_plane_url_configured=control_plane_url_configured, + control_plane_token_configured=control_plane_token_configured, + control_plane_prerequisites_satisfied=control_plane_prerequisites_satisfied, + control_plane_compat_override=control_plane_compat_override, + connector_active_platforms=active_platforms, + connector_unguarded_platforms=unguarded_platforms, + connector_recommended_allowlist_vars=recommended_allowlist_vars, + deployment_checks=findings, + deployment_pass_codes=pass_codes, + deployment_warn_codes=warn_codes, + deployment_fail_codes=fail_codes, + startup_profile_passed=startup_passed, + startup_profile_overridden=startup_overridden, + startup_profile_violation_codes=startup_violations, + blocked_surface_ids=_blocked_surface_ids( + deployment_profile, control_plane_mode + ), + decision_codes=tuple(decision_codes), + reason_codes=tuple(dict.fromkeys(reason_codes)), + ) + + +def install_effective_security_posture( + posture: EffectiveSecurityPosture, +) -> EffectiveSecurityPosture: + if not isinstance(posture, EffectiveSecurityPosture): + raise TypeError("posture must be EffectiveSecurityPosture") + global _installed_posture + with _posture_lock: + if _installed_posture is None: + _installed_posture = posture + elif _installed_posture is not posture: + # CRITICAL: silently replacing process posture creates contradictory + # authorization decisions. Reset is an explicit lifecycle/test operation. + raise RuntimeError("effective security posture is already installed") + return _installed_posture + + +def get_effective_security_posture( + *, required: bool = True +) -> EffectiveSecurityPosture | None: + with _posture_lock: + posture = _installed_posture + if posture is None and required: + raise RuntimeError("effective security posture is not installed") + return posture + + +def get_or_create_effective_security_posture( + environ: Mapping[str, str] | None = None, + *, + network_exposed: bool | None = None, +) -> EffectiveSecurityPosture: + with _posture_lock: + if _installed_posture is not None: + return _installed_posture + posture = resolve_effective_security_posture( + environ, + network_exposed=network_exposed, + ) + # The RLock makes this identity-stable even under concurrent startup. + return install_effective_security_posture(posture) + + +def reset_effective_security_posture_for_tests() -> None: + global _installed_posture + with _posture_lock: + _installed_posture = None + + +def effective_security_posture_diagnostics( + posture: EffectiveSecurityPosture | None = None, +) -> dict[str, Any]: + resolved = posture or get_effective_security_posture() + assert resolved is not None + return { + "schema_version": resolved.schema_version, + "runtime_profile": resolved.runtime_profile, + "deployment_profile": resolved.deployment_profile, + "mae_profile": resolved.mae_profile, + "network_exposed": resolved.network_exposed, + "authentication": { + "admin_configured": resolved.admin_token_configured, + "observability_configured": resolved.observability_token_configured, + }, + "startup_gate": { + "passed": resolved.startup_profile_passed, + "overridden": resolved.startup_profile_overridden, + "violation_codes": list(resolved.startup_profile_violation_codes), + }, + "control_plane": { + "mode": resolved.control_plane_mode, + "prerequisites_satisfied": (resolved.control_plane_prerequisites_satisfied), + "compat_override": resolved.control_plane_compat_override, + "blocked_surface_count": len(resolved.blocked_surface_ids), + }, + "connectors": { + "active_count": len(resolved.connector_active_platforms), + "unguarded_count": len(resolved.connector_unguarded_platforms), + }, + "decision_codes": list(resolved.decision_codes), + "reason_codes": list(resolved.reason_codes), + } diff --git a/services/route_bootstrap.py b/services/route_bootstrap.py index 8d89d06..7a1d3c8 100644 --- a/services/route_bootstrap.py +++ b/services/route_bootstrap.py @@ -153,6 +153,16 @@ def _initialize_registries_and_security_gate() -> None: config = get_config() ServiceRegistry.register(SVC_RUNTIME_CONFIG, config) + from .effective_security_posture import ( + get_effective_security_posture, + resolve_effective_security_posture, + ) + + posture = get_effective_security_posture(required=False) + if posture is None: + # Direct compatibility/test invocation does not own process installation. + posture = resolve_effective_security_posture() + # Always-on modules enable_module(ModuleCapability.CORE) enable_module(ModuleCapability.SECURITY) @@ -184,7 +194,7 @@ def _initialize_registries_and_security_gate() -> None: from .security_gate import enforce_startup_gate - enforce_startup_gate() + enforce_startup_gate(posture=posture) except Exception as exc: logging.getLogger("ComfyUI-OpenClaw").error( "Required registry initialization failed (error_type=%s)", @@ -347,6 +357,15 @@ def reset_route_bootstrap_for_tests() -> None: _registration_error = None _registration_retry_thread = None _registration_condition.notify_all() + try: + from .effective_security_posture import ( + reset_effective_security_posture_for_tests, + ) + + reset_effective_security_posture_for_tests() + except ImportError: + # Dependency-light test/import mode may omit the posture module. + pass def _store_registration_success(*, generation: int | None = None) -> bool: @@ -512,6 +531,11 @@ def register_routes_once() -> None: generation = _registration_generation try: + from .effective_security_posture import get_or_create_effective_security_posture + + # CRITICAL: this required startup owner installs process-static posture once. + # Direct helper/API invocations resolve ephemeral snapshots instead. + get_or_create_effective_security_posture() _mark_required_initialization_started() try: _register_plugins_and_shutdown_hooks() diff --git a/services/scheduler/runner.py b/services/scheduler/runner.py index 47b3983..39e34a4 100644 --- a/services/scheduler/runner.py +++ b/services/scheduler/runner.py @@ -39,13 +39,21 @@ def resolve_scheduler_execution_mode(config: Optional[dict] = None) -> str: if explicit in {SCHEDULER_EXECUTION_EMBEDDED, SCHEDULER_EXECUTION_DELEGATED}: return explicit - profile = os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local").strip().lower() try: from ..control_plane import ControlPlaneMode, resolve_control_plane_mode + from ..effective_security_posture import get_effective_security_posture + + posture = get_effective_security_posture(required=False) + profile = ( + posture.deployment_profile + if posture is not None + else os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local").strip().lower() + ) if ( profile == "public" - and resolve_control_plane_mode(profile) == ControlPlaneMode.SPLIT + and resolve_control_plane_mode(profile, posture=posture) + == ControlPlaneMode.SPLIT ): return SCHEDULER_EXECUTION_DELEGATED except Exception: diff --git a/services/secrets_encryption.py b/services/secrets_encryption.py index c908954..60cb415 100644 --- a/services/secrets_encryption.py +++ b/services/secrets_encryption.py @@ -336,15 +336,22 @@ def is_secret_write_blocked() -> bool: """ try: from .control_plane import is_split_mode + from .effective_security_posture import get_effective_security_posture except ImportError: return False - if not is_split_mode(): + posture = get_effective_security_posture(required=False) + if not is_split_mode(posture=posture): return False # Check override - compat = os.environ.get(ENV_SPLIT_COMPAT_OVERRIDE, "").lower().strip() - if compat in ("1", "true", "yes"): + compat_override = ( + posture.control_plane_compat_override + if posture is not None + else os.environ.get(ENV_SPLIT_COMPAT_OVERRIDE, "").lower().strip() + in ("1", "true", "yes") + ) + if compat_override: logger.warning("S57: Secret write override active in split mode (DEV ONLY)") return False diff --git a/services/security_gate.py b/services/security_gate.py index 3e3ab73..68c4ac0 100644 --- a/services/security_gate.py +++ b/services/security_gate.py @@ -5,12 +5,17 @@ Enforces mandatory security controls when running in HARDENED profile. Fails startup if critical controls are missing or misconfigured. """ +from __future__ import annotations + import logging import os -from typing import List, Tuple +from typing import TYPE_CHECKING, List, Tuple from .runtime_profile import get_runtime_profile, is_hardened_mode +if TYPE_CHECKING: + from .effective_security_posture import EffectiveSecurityPosture + try: from .connector_allowlist_posture import evaluate_connector_allowlist_posture except Exception: @@ -50,13 +55,25 @@ class SecurityGate: return False @staticmethod - def verify_mandatory_controls() -> Tuple[bool, List[str], List[str]]: + def verify_mandatory_controls( + posture: EffectiveSecurityPosture | None = None, + ) -> Tuple[bool, List[str], List[str]]: """ Check if all mandatory controls for the current profile are active. Returns: (passed: bool, warnings: List[str], fatal_errors: List[str]) """ warnings = [] fatal_errors = [] + hardened = ( + posture.runtime_profile == "hardened" + if posture is not None + else is_hardened_mode() + ) + runtime_profile_value = ( + posture.runtime_profile + if posture is not None + else get_runtime_profile().value + ) def _emit_startup_audit(action: str, outcome: str, details: dict) -> None: try: @@ -78,8 +95,12 @@ class SecurityGate: # OPENCLAW_LOCALHOST_ALLOW_NO_ORIGIN is an explicit operator override for # localhost tooling; surfacing it early avoids silent CSRF-boundary drift. allow_no_origin = ( - os.environ.get("OPENCLAW_LOCALHOST_ALLOW_NO_ORIGIN", "").strip().lower() - == "true" + posture.localhost_no_origin_override + if posture is not None + else ( + os.environ.get("OPENCLAW_LOCALHOST_ALLOW_NO_ORIGIN", "").strip().lower() + == "true" + ) ) if allow_no_origin: logger.warning( @@ -92,7 +113,7 @@ class SecurityGate: details={ "env": "OPENCLAW_LOCALHOST_ALLOW_NO_ORIGIN", "value": "true", - "profile": get_runtime_profile().value, + "profile": runtime_profile_value, }, ) @@ -100,18 +121,35 @@ class SecurityGate: try: from .access_control import is_any_token_configured, is_auth_configured - is_exposed = SecurityGate._check_network_exposure() + is_exposed = ( + posture.network_exposed + if posture is not None + else SecurityGate._check_network_exposure() + ) # S45 Policy: If exposed, ANY token is sufficient to say "we are not wide open". # (Though Admin token is preferred for full protection, basic auth presence satisfies "not accidentally open") - auth_ready = is_any_token_configured() + auth_ready = ( + ( + posture.admin_token_configured + or posture.observability_token_configured + ) + if posture is not None + else is_any_token_configured() + ) if is_exposed and not auth_ready: # Check for explicit override - from .runtime_config import get_config + dangerous_bind_override = ( + posture.dangerous_bind_override if posture is not None else None + ) + if dangerous_bind_override is None: + from .runtime_config import get_config - config = get_config() + dangerous_bind_override = ( + get_config().security_dangerous_bind_override + ) - if config.security_dangerous_bind_override: + if dangerous_bind_override: warnings.append( "WARNING: Server is exposed (--listen) without Authentication, but override is active.\n" " This is a DANGEROUS configuration. Remote Code Execution is possible if port is accessible." @@ -122,7 +160,7 @@ class SecurityGate: details={ "reason": "exposed_without_auth", "override": True, - "profile": get_runtime_profile().value, + "profile": runtime_profile_value, }, ) # Do NOT block startup (S45 Override Contract) @@ -138,7 +176,12 @@ class SecurityGate: # Loopback + No Auth # Use strict is_auth_configured (Admin) for Hardened profile loopback check? # "HARDENED profile requires Authentication even on loopback." - if is_hardened_mode() and not is_auth_configured(): + admin_ready = ( + posture.admin_token_configured + if posture is not None + else is_auth_configured() + ) + if hardened and not admin_ready: warnings.append( "HARDENED profile requires Admin Authentication even on loopback." ) @@ -146,16 +189,24 @@ class SecurityGate: warnings.append("Could not import access_control service") # 2. Egress Policy (SSRF) - from .runtime_config import get_config + if posture is None: + from .runtime_config import get_config - config = get_config() + config = get_config() + allow_any_public_llm_host = config.allow_any_public_llm_host + allow_insecure_base_url = config.allow_insecure_base_url + webhook_auth_mode = config.webhook_auth_mode + else: + allow_any_public_llm_host = posture.allow_any_public_llm_host + allow_insecure_base_url = posture.allow_insecure_base_url + webhook_auth_mode = posture.webhook_auth_mode - if config.allow_any_public_llm_host: + if allow_any_public_llm_host: warnings.append( "OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST is enabled (Egress check bypassed)" ) - if config.allow_insecure_base_url: + if allow_insecure_base_url: warnings.append( "OPENCLAW_ALLOW_INSECURE_BASE_URL is enabled (SSRF check bypassed)" ) @@ -164,7 +215,7 @@ class SecurityGate: from .modules import ModuleCapability, is_module_enabled if is_module_enabled(ModuleCapability.WEBHOOK): - if not config.webhook_auth_mode: + if webhook_auth_mode == "unset": warnings.append( "Webhook module enabled but OPENCLAW_WEBHOOK_AUTH_MODE not set" ) @@ -206,7 +257,7 @@ class SecurityGate: try: from .control_plane import enforce_control_plane_startup - cp_result = enforce_control_plane_startup() + cp_result = enforce_control_plane_startup(posture=posture) if not cp_result.get("startup_passed", True): for err in cp_result.get("errors", []): fatal_errors.append(f"S62 Control-Plane: {err}") @@ -216,27 +267,41 @@ class SecurityGate: warnings.append("S62 control_plane module failed to import") # 7. Connector allowlist fail-closed posture (S71) - connector_posture = evaluate_connector_allowlist_posture(os.environ) - if connector_posture["has_unguarded_connectors"]: - deployment_profile = ( - os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "").strip().lower() + if posture is not None: + connector_unguarded = list(posture.connector_unguarded_platforms) + connector_allowlist_vars = list( + posture.connector_recommended_allowlist_vars ) - platforms = ", ".join(connector_posture["unguarded_platforms"]) - allowlist_vars = ", ".join(connector_posture["recommended_allowlist_vars"]) + else: + connector_posture = evaluate_connector_allowlist_posture(os.environ) + connector_unguarded = [ + str(item) for item in connector_posture["unguarded_platforms"] + ] + connector_allowlist_vars = [ + str(item) for item in connector_posture["recommended_allowlist_vars"] + ] + if connector_unguarded: + deployment_profile = ( + posture.deployment_profile + if posture is not None + else os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "").strip().lower() + ) + platforms = ", ".join(connector_unguarded) + allowlist_vars = ", ".join(connector_allowlist_vars) msg = ( "Connector allowlist coverage missing for active platform(s): " f"{platforms}. Configure allowlists ({allowlist_vars}) before enabling ingress." ) # CRITICAL: hardened/public must fail closed for unallowlisted connector ingress. - if is_hardened_mode(): + if hardened: warnings.append(f"S71 (hardened fail-closed): {msg}") _emit_startup_audit( action="startup.connector_allowlist_posture", outcome="error", details={ "mode": "hardened", - "unguarded_platforms": connector_posture["unguarded_platforms"], + "unguarded_platforms": connector_unguarded, "deployment_profile": deployment_profile or "unset", }, ) @@ -247,7 +312,7 @@ class SecurityGate: outcome="error", details={ "mode": "public", - "unguarded_platforms": connector_posture["unguarded_platforms"], + "unguarded_platforms": connector_unguarded, }, ) else: @@ -257,13 +322,13 @@ class SecurityGate: outcome="warn", details={ "mode": "warn_only", - "unguarded_platforms": connector_posture["unguarded_platforms"], + "unguarded_platforms": connector_unguarded, "deployment_profile": deployment_profile or "unset", }, ) # In HARDENED mode, treat all warnings as FATAL - if is_hardened_mode() and warnings: + if hardened and warnings: fatal_errors.extend(warnings) warnings = [] @@ -271,18 +336,26 @@ class SecurityGate: return passed, warnings, fatal_errors -def enforce_startup_gate() -> None: +def enforce_startup_gate( + posture: EffectiveSecurityPosture | None = None, +) -> None: """ Run the security gate. If in HARDENED mode and checks fail -> Raise SystemExit. If in MINIMAL mode and checks fail -> Log warnings. """ - is_hardened = is_hardened_mode() + is_hardened = ( + posture.runtime_profile == "hardened" + if posture is not None + else is_hardened_mode() + ) mode_str = "HARDENED" if is_hardened else "MINIMAL" logger.info(f"Running S41 Security Gate ({mode_str} profile)...") - passed, warnings, fatal_errors = SecurityGate.verify_mandatory_controls() + passed, warnings, fatal_errors = SecurityGate.verify_mandatory_controls( + posture=posture + ) # Log warnings first (non-blocking unless hardened) if warnings: diff --git a/services/startup_profile_gate.py b/services/startup_profile_gate.py index c194431..7497e74 100644 --- a/services/startup_profile_gate.py +++ b/services/startup_profile_gate.py @@ -21,7 +21,10 @@ import logging import os import time from dataclasses import asdict, dataclass, field -from typing import Any, Dict, List, Mapping, Optional, Tuple +from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Tuple + +if TYPE_CHECKING: + from .effective_security_posture import EffectiveSecurityPosture logger = logging.getLogger("ComfyUI-OpenClaw.services.startup_profile_gate") @@ -80,12 +83,14 @@ _PROFILE_ENV = "OPENCLAW_DEPLOYMENT_PROFILE" def _resolve_profile(environ: Optional[Mapping[str, str]] = None) -> str: """Resolve the deployment profile from environment.""" - env = environ or os.environ + env = os.environ if environ is None else environ return env.get(_PROFILE_ENV, "local").strip().lower() def evaluate_startup_gate( environ: Optional[Mapping[str, str]] = None, + *, + posture: EffectiveSecurityPosture | None = None, ) -> StartupGateResult: """ Evaluate the startup profile gate. @@ -94,7 +99,37 @@ def evaluate_startup_gate( override status. """ global _last_gate_result # noqa: PLW0603 - env: Mapping[str, str] = environ or os.environ + if posture is not None: + violation_codes = set(posture.startup_profile_violation_codes) + violations = [ + { + "code": check.code, + "severity": check.severity, + "message": check.message, + "remediation": check.remediation, + } + for check in posture.deployment_checks + if check.code in violation_codes + ] + override_reason = "" + if posture.startup_profile_overridden: + override_reason = ( + f"S56: Startup gate bypassed via {_OVERRIDE_ENV}=1. " + f"Profile '{posture.deployment_profile}' has " + f"{len(violations)} violation(s). " + "This override is intended for emergency use only." + ) + result = StartupGateResult( + profile=posture.deployment_profile, + passed=posture.startup_profile_passed, + overridden=posture.startup_profile_overridden, + override_reason=override_reason, + violations=violations, + ) + _last_gate_result = result + return result + + env: Mapping[str, str] = os.environ if environ is None else environ profile = _resolve_profile(env) # Local profile: no enforcement @@ -149,6 +184,8 @@ def evaluate_startup_gate( def enforce_startup_gate( environ: Optional[Mapping[str, str]] = None, + *, + posture: EffectiveSecurityPosture | None = None, ) -> StartupGateResult: """ Evaluate and enforce the startup profile gate. @@ -158,7 +195,7 @@ def enforce_startup_gate( Returns the gate result on success (pass or overridden). """ - result = evaluate_startup_gate(environ) + result = evaluate_startup_gate(environ, posture=posture) if result.passed and not result.overridden: logger.info(f"S56: Startup profile gate PASSED for profile '{result.profile}'.") diff --git a/services/surface_guard.py b/services/surface_guard.py index 4c1b3e7..00c6472 100644 --- a/services/surface_guard.py +++ b/services/surface_guard.py @@ -32,8 +32,23 @@ web = import_aiohttp_web() logger = logging.getLogger(__name__) +def _get_installed_posture(): + try: + from .effective_security_posture import get_effective_security_posture + + return get_effective_security_posture(required=False) + except ImportError: + return None + + def _is_fail_closed_profile() -> bool: """Return True if errors should fail-closed (block) rather than fail-open.""" + posture = _get_installed_posture() + if posture is not None: + return bool( + posture.deployment_profile == "public" + or posture.runtime_profile == "hardened" + ) profile = os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local").lower() if profile == "public": return True @@ -62,9 +77,14 @@ def check_surface(surface_id: str, request: web.Request = None) -> web.Response try: from .control_plane import get_blocked_surfaces, resolve_control_plane_mode - profile = os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local") - mode = resolve_control_plane_mode(profile) - blocked = get_blocked_surfaces(profile, mode) + posture = _get_installed_posture() + profile = ( + posture.deployment_profile + if posture is not None + else os.environ.get("OPENCLAW_DEPLOYMENT_PROFILE", "local") + ) + mode = resolve_control_plane_mode(profile, posture=posture) + blocked = get_blocked_surfaces(profile, mode, posture=posture) blocked_ids = {sid: desc for sid, desc in blocked} if surface_id not in blocked_ids: diff --git a/tests/architecture_dependency_policy.json b/tests/architecture_dependency_policy.json index be17ef4..5ca4140 100644 --- a/tests/architecture_dependency_policy.json +++ b/tests/architecture_dependency_policy.json @@ -198,6 +198,7 @@ "services/deployment_profile.py", "services/diagnostics_flags.py", "services/effective_config.py", + "services/effective_security_posture.py", "services/endpoint_manifest.py", "services/execution_budgets.py", "services/failover.py", diff --git a/tests/test_architecture_dependency_policy.py b/tests/test_architecture_dependency_policy.py index a3ab121..cd37bcb 100644 --- a/tests/test_architecture_dependency_policy.py +++ b/tests/test_architecture_dependency_policy.py @@ -289,7 +289,7 @@ class RepositoryArchitecturePolicyTests(unittest.TestCase): analysis = dependency_policy.analyze_repository(self.repo_root, policy) self.assertEqual(analysis.findings, ()) - self.assertEqual(len(analysis.owned_paths), 297) + self.assertEqual(len(analysis.owned_paths), 298) self.assertEqual(len(policy["accepted_cycles"]), 2) self.assertEqual(len(policy["dynamic_imports"]), 8) self.assertEqual(len(policy["compatibility_exceptions"]), 9) diff --git a/tests/test_bootstrap_lifecycle_outcome.py b/tests/test_bootstrap_lifecycle_outcome.py index c9d7a54..aa1f8ab 100644 --- a/tests/test_bootstrap_lifecycle_outcome.py +++ b/tests/test_bootstrap_lifecycle_outcome.py @@ -12,9 +12,9 @@ from unittest.mock import MagicMock, patch from services import route_bootstrap from services.startup_lifecycle import ( - STARTUP_DIAGNOSTIC_KEYS, MAX_DIAGNOSTIC_MS, MAX_WARMUPS, + STARTUP_DIAGNOSTIC_KEYS, StartupLifecycle, StartupPhase, StartupReason, @@ -78,9 +78,7 @@ class TestStartupOutcomeContract(unittest.TestCase): diagnostics = lifecycle.snapshot().to_diagnostics() self.assertEqual(diagnostics["phase"], "complete") self.assertEqual(diagnostics["state"], "ready") - self.assertEqual( - diagnostics["reason_code"], "route_registration_succeeded" - ) + self.assertEqual(diagnostics["reason_code"], "route_registration_succeeded") self.assertTrue(diagnostics["ready"]) self.assertFalse(diagnostics["fatal"]) self.assertEqual(diagnostics["attempt"], 2) @@ -114,7 +112,10 @@ class TestStartupOutcomeContract(unittest.TestCase): lifecycle.mark_required_initialization_started() lifecycle.mark_host_waiting(attempt=0, max_attempts=2) - for attempt, code in ((0, "ATTEMPT_NOT_INCREASING"), (3, "ATTEMPT_OUT_OF_RANGE")): + for attempt, code in ( + (0, "ATTEMPT_NOT_INCREASING"), + (3, "ATTEMPT_OUT_OF_RANGE"), + ): with self.subTest(attempt=attempt): before = lifecycle.snapshot() with self.assertRaises(StartupTransitionError) as ctx: @@ -185,11 +186,7 @@ class TestStartupWarmupProjection(unittest.TestCase): while time.monotonic() < deadline: diagnostics = get_startup_diagnostics() warmup = next( - ( - item - for item in diagnostics["warmups"] - if item["name"] == name - ), + (item for item in diagnostics["warmups"] if item["name"] == name), None, ) if warmup and warmup["state"] == state: @@ -292,14 +289,10 @@ class TestStartupWarmupProjection(unittest.TestCase): marker = "PRIVATE_MONITOR_START C:/private/monitor" failure = RuntimeError(marker) - with patch( - "services.startup_lifecycle.threading.Thread" - ) as thread_factory: + with patch("services.startup_lifecycle.threading.Thread") as thread_factory: thread_factory.return_value.start.side_effect = failure with self.assertRaises(RuntimeError) as ctx: - start_optional_warmups( - [("monitor_provider", lambda: None, 0.01)] - ) + start_optional_warmups([("monitor_provider", lambda: None, 0.01)]) self.assertIs(ctx.exception, failure) diagnostics = get_startup_diagnostics() @@ -323,9 +316,7 @@ class TestStartupWarmupProjection(unittest.TestCase): ) with ( - patch( - "services.startup_lifecycle.threading.Thread" - ) as thread_factory, + patch("services.startup_lifecycle.threading.Thread") as thread_factory, self.assertLogs("ComfyUI-OpenClaw", level="WARNING") as captured, ): thread_factory.return_value.start.side_effect = failure @@ -334,9 +325,7 @@ class TestStartupWarmupProjection(unittest.TestCase): diagnostics = get_startup_diagnostics() self.assertTrue(diagnostics["degraded"]) warmup = next( - item - for item in diagnostics["warmups"] - if item["name"] == "worker_provider" + item for item in diagnostics["warmups"] if item["name"] == "worker_provider" ) self.assertEqual(warmup["state"], "failed") self.assertNotIn(marker, json.dumps(diagnostics)) @@ -366,7 +355,9 @@ class TestRouteBootstrapOutcomeIntegration(unittest.TestCase): release.wait(timeout=1) with ( - patch.object(route_bootstrap, "_register_plugins_and_shutdown_hooks") as optional, + patch.object( + route_bootstrap, "_register_plugins_and_shutdown_hooks" + ) as optional, patch.object( route_bootstrap, "_initialize_registries_and_security_gate", @@ -432,9 +423,7 @@ class TestRouteBootstrapOutcomeIntegration(unittest.TestCase): self.assertIs(observed[1], failure) diagnostics = get_startup_diagnostics() self.assertEqual(diagnostics["state"], "fatal") - self.assertEqual( - diagnostics["reason_code"], "required_initialization_failed" - ) + self.assertEqual(diagnostics["reason_code"], "required_initialization_failed") self.assertNotIn(marker, json.dumps(diagnostics)) self.assertNotIn(marker, "\n".join(captured.output)) @@ -551,9 +540,7 @@ class TestRouteBootstrapOutcomeIntegration(unittest.TestCase): register.assert_called_once_with(server) diagnostics = get_startup_diagnostics() self.assertTrue(diagnostics["ready"]) - self.assertEqual( - diagnostics["reason_code"], "route_registration_succeeded" - ) + self.assertEqual(diagnostics["reason_code"], "route_registration_succeeded") reset_startup_lifecycle_for_tests() route_bootstrap.reset_route_bootstrap_for_tests() @@ -665,9 +652,7 @@ class TestPublicHealthLifecycleProjection(unittest.TestCase): ) marker = "PRIVATE_HEALTH_FAILURE C:/private/health" - fallback = self._health_payload( - diagnostics_side_effect=RuntimeError(marker) - ) + fallback = self._health_payload(diagnostics_side_effect=RuntimeError(marker)) self.assertEqual( tuple(fallback["startup"]), STARTUP_DIAGNOSTIC_KEYS, diff --git a/tests/test_effective_security_posture.py b/tests/test_effective_security_posture.py new file mode 100644 index 0000000..649bc8d --- /dev/null +++ b/tests/test_effective_security_posture.py @@ -0,0 +1,456 @@ +"""R232 immutable effective security-posture contract tests.""" + +from __future__ import annotations + +import dataclasses +import inspect +import json +import os +import sys +import threading +import unittest +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace +from unittest.mock import patch + +from services.effective_security_posture import ( + EffectiveSecurityPosture, + effective_security_posture_diagnostics, + get_effective_security_posture, + get_or_create_effective_security_posture, + install_effective_security_posture, + reset_effective_security_posture_for_tests, + resolve_effective_security_posture, +) + + +def _valid_public_env() -> dict[str, str]: + return { + "OPENCLAW_DEPLOYMENT_PROFILE": "public", + "OPENCLAW_RUNTIME_PROFILE": "hardened", + "OPENCLAW_ADMIN_TOKEN": "PRIVATE_ADMIN_CANARY", + "OPENCLAW_OBSERVABILITY_TOKEN": "PRIVATE_OBS_CANARY", + "OPENCLAW_ALLOW_REMOTE_ADMIN": "0", + "OPENCLAW_PUBLIC_SHARED_SURFACE_BOUNDARY_ACK": "1", + "OPENCLAW_TRUST_X_FORWARDED_FOR": "1", + "OPENCLAW_TRUSTED_PROXIES": "10.0.0.0/8", + "OPENCLAW_WEBHOOK_AUTH_MODE": "hmac", + "OPENCLAW_WEBHOOK_HMAC_SECRET": "PRIVATE_HMAC_CANARY", + "OPENCLAW_WEBHOOK_REQUIRE_REPLAY_PROTECTION": "1", + "OPENCLAW_ENABLE_EXTERNAL_TOOLS": "0", + "OPENCLAW_ENABLE_REGISTRY_SYNC": "0", + "OPENCLAW_ENABLE_TRANSFORMS": "0", + "OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST": "0", + "OPENCLAW_ALLOW_INSECURE_BASE_URL": "0", + "OPENCLAW_SECURITY_DANGEROUS_BIND_OVERRIDE": "0", + "OPENCLAW_CONTROL_PLANE_MODE": "split", + "OPENCLAW_CONTROL_PLANE_URL": "https://private-control.invalid", + "OPENCLAW_CONTROL_PLANE_TOKEN": "PRIVATE_CP_CANARY", + "OPENCLAW_CONNECTOR_TELEGRAM_TOKEN": "PRIVATE_CONNECTOR_CANARY", + "OPENCLAW_CONNECTOR_TELEGRAM_ALLOWED_USERS": "private-user", + } + + +class EffectiveSecurityPostureTestCase(unittest.TestCase): + def setUp(self) -> None: + reset_effective_security_posture_for_tests() + + def tearDown(self) -> None: + reset_effective_security_posture_for_tests() + + def test_schema_is_frozen_recursively_immutable_and_secret_free(self): + env = _valid_public_env() + snapshot = resolve_effective_security_posture(env, network_exposed=True) + + self.assertIsInstance(snapshot, EffectiveSecurityPosture) + self.assertEqual(snapshot.schema_version, 1) + self.assertEqual(snapshot.deployment_profile, "public") + self.assertEqual(snapshot.runtime_profile, "hardened") + self.assertEqual(snapshot.mae_profile, "public") + self.assertTrue(snapshot.admin_token_configured) + self.assertTrue(snapshot.control_plane_prerequisites_satisfied) + self.assertEqual(snapshot.connector_active_platforms, ("telegram",)) + self.assertEqual(snapshot.connector_unguarded_platforms, ()) + + with self.assertRaises(dataclasses.FrozenInstanceError): + snapshot.deployment_profile = "local" # type: ignore[misc] + + def assert_immutable(value): + self.assertNotIsInstance(value, (dict, list, set)) + if dataclasses.is_dataclass(value): + for field in dataclasses.fields(value): + assert_immutable(getattr(value, field.name)) + elif isinstance(value, tuple): + for item in value: + assert_immutable(item) + + assert_immutable(snapshot) + rendered = json.dumps( + effective_security_posture_diagnostics(snapshot), sort_keys=True + ) + for private_value in env.values(): + if private_value.startswith("PRIVATE_") or private_value.startswith( + "https://" + ): + self.assertNotIn(private_value, rendered) + self.assertNotIn("10.0.0.0/8", rendered) + self.assertNotIn("private-user", rendered) + + def test_diagnostic_projection_is_allowlisted_bounded_and_code_only(self): + env = { + "OPENCLAW_DEPLOYMENT_PROFILE": "public", + "OPENCLAW_WEBHOOK_AUTH_MODE": "PRIVATE_INVALID_MODE", + "OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN": "PRIVATE_SLACK_CANARY", + } + projection = effective_security_posture_diagnostics( + resolve_effective_security_posture(env, network_exposed=False) + ) + + self.assertEqual( + set(projection), + { + "schema_version", + "runtime_profile", + "deployment_profile", + "mae_profile", + "network_exposed", + "authentication", + "startup_gate", + "control_plane", + "connectors", + "decision_codes", + "reason_codes", + }, + ) + rendered = json.dumps(projection, sort_keys=True) + self.assertNotIn("PRIVATE_INVALID_MODE", rendered) + self.assertNotIn("PRIVATE_SLACK_CANARY", rendered) + self.assertLessEqual(len(rendered), 4096) + + def test_profile_matrix_matches_existing_deployment_evaluator(self): + from services.deployment_profile import evaluate_deployment_profile + from tests.test_r105_profile_matrix import PROFILE_MATRIX_FIXTURES + + for fixture in PROFILE_MATRIX_FIXTURES: + with self.subTest(fixture=fixture["id"]): + env = dict(fixture["env"]) + env["OPENCLAW_DEPLOYMENT_PROFILE"] = fixture["profile"] + report = evaluate_deployment_profile(fixture["profile"], env) + snapshot = resolve_effective_security_posture( + env, network_exposed=False + ) + expected_fail_codes = tuple( + check.code for check in report.checks if check.severity == "fail" + ) + self.assertEqual(snapshot.deployment_fail_codes, expected_fail_codes) + self.assertEqual( + snapshot.startup_profile_passed, + fixture["profile"] == "local" or not expected_fail_codes, + ) + + def test_startup_gate_snapshot_matches_explicit_environment_path(self): + from services.startup_profile_gate import evaluate_startup_gate + + rows = [ + {"OPENCLAW_DEPLOYMENT_PROFILE": "local"}, + {"OPENCLAW_DEPLOYMENT_PROFILE": "lan"}, + _valid_public_env(), + { + "OPENCLAW_DEPLOYMENT_PROFILE": "public", + "OPENCLAW_SECURITY_DANGEROUS_PROFILE_OVERRIDE": "1", + }, + ] + for env in rows: + with self.subTest(env=tuple(sorted(env))): + expected = evaluate_startup_gate(env) + snapshot = resolve_effective_security_posture( + env, network_exposed=False + ) + actual = evaluate_startup_gate(posture=snapshot) + self.assertEqual(actual.profile, expected.profile) + self.assertEqual(actual.passed, expected.passed) + self.assertEqual(actual.overridden, expected.overridden) + self.assertEqual( + [item["code"] for item in actual.violations], + [item["code"] for item in expected.violations], + ) + + def test_control_plane_snapshot_matches_existing_matrix(self): + from services.control_plane import ( + HIGH_RISK_SURFACES, + ControlPlaneMode, + enforce_control_plane_startup, + get_blocked_surfaces, + resolve_control_plane_mode, + validate_split_prerequisites, + ) + + rows = [ + {"OPENCLAW_DEPLOYMENT_PROFILE": "local"}, + {"OPENCLAW_DEPLOYMENT_PROFILE": "public"}, + _valid_public_env(), + { + "OPENCLAW_DEPLOYMENT_PROFILE": "public", + "OPENCLAW_CONTROL_PLANE_MODE": "embedded", + }, + { + "OPENCLAW_DEPLOYMENT_PROFILE": "public", + "OPENCLAW_CONTROL_PLANE_MODE": "embedded", + "OPENCLAW_SPLIT_COMPAT_OVERRIDE": "1", + }, + ] + for env in rows: + with self.subTest(env=tuple(sorted(env))): + telemetry_stub = SimpleNamespace( + get_security_telemetry=lambda: SimpleNamespace( + record_dangerous_override=lambda *_args, **_kwargs: None + ) + ) + with ( + patch.dict(os.environ, env, clear=True), + patch.dict( + sys.modules, + {"services.security_telemetry": telemetry_stub}, + ), + ): + expected_mode = resolve_control_plane_mode( + env["OPENCLAW_DEPLOYMENT_PROFILE"] + ) + expected_prereq = validate_split_prerequisites().to_dict() + expected_startup = enforce_control_plane_startup() + expected_blocked = get_blocked_surfaces( + env["OPENCLAW_DEPLOYMENT_PROFILE"], expected_mode + ) + snapshot = resolve_effective_security_posture( + env, network_exposed=False + ) + if ( + snapshot.deployment_profile == "public" + and snapshot.control_plane_mode == "split" + ): + self.assertEqual( + set(snapshot.blocked_surface_ids), + {surface_id for surface_id, _ in HIGH_RISK_SURFACES}, + ) + self.assertEqual( + resolve_control_plane_mode(posture=snapshot), + expected_mode, + ) + self.assertEqual( + validate_split_prerequisites(posture=snapshot).to_dict(), + expected_prereq, + ) + self.assertEqual( + enforce_control_plane_startup(posture=snapshot), + expected_startup, + ) + self.assertEqual( + get_blocked_surfaces( + snapshot.deployment_profile, + ControlPlaneMode(snapshot.control_plane_mode), + posture=snapshot, + ), + expected_blocked, + ) + + def test_install_is_identity_stable_thread_safe_and_rejects_replacement(self): + first = resolve_effective_security_posture( + {"OPENCLAW_DEPLOYMENT_PROFILE": "local"}, network_exposed=False + ) + second = resolve_effective_security_posture( + _valid_public_env(), network_exposed=True + ) + + self.assertIs(install_effective_security_posture(first), first) + self.assertIs(get_effective_security_posture(), first) + self.assertIs(install_effective_security_posture(first), first) + with self.assertRaisesRegex(RuntimeError, "already installed"): + install_effective_security_posture(second) + + reset_effective_security_posture_for_tests() + barrier = threading.Barrier(16) + + def resolve_once(_index): + barrier.wait() + return get_or_create_effective_security_posture( + {"OPENCLAW_DEPLOYMENT_PROFILE": "local"}, + network_exposed=False, + ) + + with ThreadPoolExecutor(max_workers=16) as pool: + results = list(pool.map(resolve_once, range(16))) + + self.assertEqual(len({id(item) for item in results}), 1) + + def test_installed_snapshot_prevents_ambient_drift_in_migrated_consumers(self): + from api import routes + from services.control_plane import ( + ControlPlaneMode, + is_surface_blocked, + resolve_control_plane_mode, + ) + from services.startup_profile_gate import evaluate_startup_gate + + snapshot = get_or_create_effective_security_posture( + _valid_public_env(), network_exposed=True + ) + with patch.dict( + os.environ, + { + "OPENCLAW_DEPLOYMENT_PROFILE": "local", + "OPENCLAW_RUNTIME_PROFILE": "minimal", + }, + clear=True, + ): + self.assertIs(get_or_create_effective_security_posture(), snapshot) + self.assertEqual( + resolve_control_plane_mode("local"), ControlPlaneMode.SPLIT + ) + self.assertTrue(is_surface_blocked("webhook_execute")) + self.assertEqual(routes._resolve_mae_profile(), "public") + gate = evaluate_startup_gate(posture=snapshot) + self.assertTrue(gate.passed) + self.assertEqual(gate.profile, "public") + + def test_security_gate_uses_explicit_snapshot_for_process_static_branches(self): + from services.security_gate import SecurityGate + + snapshot = resolve_effective_security_posture( + {"OPENCLAW_DEPLOYMENT_PROFILE": "local"}, + network_exposed=True, + ) + drifted_env = { + "OPENCLAW_DEPLOYMENT_PROFILE": "public", + "OPENCLAW_RUNTIME_PROFILE": "hardened", + "OPENCLAW_ADMIN_TOKEN": "DRIFTED_PRIVATE_TOKEN", + "OPENCLAW_SECURITY_DANGEROUS_BIND_OVERRIDE": "1", + } + with ( + patch.dict(os.environ, drifted_env, clear=True), + patch.object( + SecurityGate, + "_check_network_exposure", + side_effect=AssertionError("ambient network exposure read"), + ), + patch( + "services.security_gate.get_runtime_profile", + side_effect=AssertionError("ambient runtime profile read"), + ), + patch( + "services.security_gate.is_hardened_mode", + side_effect=AssertionError("ambient hardened profile read"), + ), + patch( + "services.security_gate.evaluate_connector_allowlist_posture", + side_effect=AssertionError("ambient connector posture read"), + ), + patch("services.modules.is_module_enabled", return_value=False), + patch("services.tool_runner.is_tools_enabled", return_value=False), + patch( + "services.permission_posture.evaluate_startup_permissions", + return_value=(True, []), + ), + ): + passed, _warnings, fatal_errors = SecurityGate.verify_mandatory_controls( + posture=snapshot + ) + + self.assertFalse(passed) + self.assertTrue( + any( + "exposed (--listen) without Authentication" in item + for item in fatal_errors + ) + ) + self.assertFalse(any("DRIFTED_PRIVATE_TOKEN" in item for item in fatal_errors)) + + def test_malformed_or_unknown_inputs_fail_with_content_free_errors(self): + class ExplodingMapping(dict): + def __contains__(self, _key): + raise RuntimeError("PRIVATE_MAPPING_CANARY") + + with self.assertRaisesRegex( + ValueError, "^security posture input unavailable$" + ) as mapping_error: + resolve_effective_security_posture( + ExplodingMapping(), network_exposed=False + ) + self.assertNotIn("PRIVATE_MAPPING_CANARY", str(mapping_error.exception)) + + with self.assertRaisesRegex( + ValueError, "^unsupported deployment profile$" + ) as profile_error: + resolve_effective_security_posture( + {"OPENCLAW_DEPLOYMENT_PROFILE": ("PRIVATE_PROFILE_CANARY")}, + network_exposed=False, + ) + self.assertNotIn("PRIVATE_PROFILE_CANARY", str(profile_error.exception)) + + def test_request_dynamic_security_state_is_not_in_snapshot_schema(self): + forbidden_names = { + "presented", + "header", + "cookie", + "registry", + "tenant", + "scope", + "client", + "origin", + "replay_state", + "session", + "credential_value", + "token_value", + "presented_token", + "request_headers", + "tenant_id", + "client_address", + "request_origin", + "live_connector_session", + } + field_names = { + field.name for field in dataclasses.fields(EffectiveSecurityPosture) + } + for field_name in forbidden_names: + self.assertNotIn( + field_name, + field_names, + f"request-dynamic field leaked into snapshot: {field_name}", + ) + + def test_consumer_contracts_accept_explicit_posture(self): + from services.control_plane import ( + enforce_control_plane_startup, + get_blocked_surfaces, + resolve_control_plane_mode, + validate_split_prerequisites, + ) + from services.security_gate import SecurityGate, enforce_startup_gate + from services.startup_profile_gate import evaluate_startup_gate + + for function in ( + evaluate_startup_gate, + SecurityGate.verify_mandatory_controls, + enforce_startup_gate, + resolve_control_plane_mode, + get_blocked_surfaces, + validate_split_prerequisites, + enforce_control_plane_startup, + ): + with self.subTest(function=function.__qualname__): + self.assertIn("posture", inspect.signature(function).parameters) + + def test_route_bootstrap_reset_clears_process_snapshot(self): + from services import route_bootstrap + + get_or_create_effective_security_posture( + {"OPENCLAW_DEPLOYMENT_PROFILE": "local"}, network_exposed=False + ) + self.assertIsNotNone(get_effective_security_posture(required=False)) + + route_bootstrap.reset_route_bootstrap_for_tests() + + self.assertIsNone(get_effective_security_posture(required=False)) + + +if __name__ == "__main__": + unittest.main()