feat(security): complete S45 startup exposure hard-stop and sync roadmap/implementation evidence

This commit is contained in:
rookiestar28
2026-02-17 18:17:50 +08:00
parent d20a4f7818
commit 0632361e6a
8 changed files with 367 additions and 107 deletions
+16
View File
@@ -147,3 +147,19 @@ def is_auth_configured() -> bool:
or ""
)
return bool(val.strip())
def is_any_token_configured() -> bool:
"""
Check if ANY authentication token is configured (Admin OR Observability).
Used for S45 Startup Gate to assess if the instance has minimal protection.
"""
if is_auth_configured():
return True
obs_val = (
os.environ.get("OPENCLAW_OBSERVABILITY_TOKEN")
or os.environ.get("MOLTBOT_OBSERVABILITY_TOKEN")
or ""
)
return bool(obs_val.strip())
+36 -30
View File
@@ -13,22 +13,23 @@ every exposed endpoint has an explicit security classification.
import enum
import inspect
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional, Set, Any
from typing import Any, Callable, Dict, List, Optional, Set
try:
from aiohttp import web
except ImportError:
web = None
class AuthTier(enum.Enum):
"""Authentication requirement level."""
ADMIN = "admin" # Requires Admin Token (Full R/W)
OBSERVABILITY = "obs" # Requires Obs Token or Admin Token (Read-only metrics/logs)
INTERNAL = "internal" # Loopback only (strict)
PUBLIC = "public" # No auth required (Use with extreme caution)
WEBHOOK = "webhook" # Signature verification required
BRIDGE = "bridge" # Bridge/Sidecar authentication
ADMIN = "admin" # Requires Admin Token (Full R/W)
OBSERVABILITY = "obs" # Requires Obs Token or Admin Token (Read-only metrics/logs)
INTERNAL = "internal" # Loopback only (strict)
PUBLIC = "public" # No auth required (Use with extreme caution)
WEBHOOK = "webhook" # Signature verification required
BRIDGE = "bridge" # Bridge/Sidecar authentication
class RiskTier(enum.Enum):
@@ -36,24 +37,24 @@ class RiskTier(enum.Enum):
Sensitivity level for audit and impact analysis.
User acceptance of risk is derived from this.
"""
CRITICAL = "critical" # Shell execution, File overwrite, Secret reveal processes
HIGH = "high" # Configuration change, Service restart
MEDIUM = "medium" # Data modification, Launching heavy compute
LOW = "low" # Read-only status info
NONE = "none" # Public static assets, Health checks
CRITICAL = "critical" # Shell execution, File overwrite, Secret reveal processes
HIGH = "high" # Configuration change, Service restart
MEDIUM = "medium" # Data modification, Launching heavy compute
LOW = "low" # Read-only status info
NONE = "none" # Public static assets, Health checks
@dataclass
class EndpointMetadata:
"""Explicit security contract for a route handler."""
auth_tier: AuthTier
risk_tier: RiskTier
summary: str
description: str = ""
required_scopes: List[str] = field(default_factory=list) # For S46
audit_action: Optional[str] = None # For R99
audit_action: Optional[str] = None # For R99
# Registry to store metadata by handler function
@@ -66,11 +67,11 @@ def endpoint_metadata(
summary: str,
description: str = "",
scopes: Optional[List[str]] = None,
audit: Optional[str] = None
audit: Optional[str] = None,
):
"""
Decorator to attach security metadata to a handler function.
Usage:
@endpoint_metadata(
auth=AuthTier.ADMIN,
@@ -80,6 +81,7 @@ def endpoint_metadata(
)
async def handler(request): ...
"""
def decorator(handler: Callable):
meta = EndpointMetadata(
auth_tier=auth,
@@ -87,12 +89,13 @@ def endpoint_metadata(
summary=summary,
description=description,
required_scopes=scopes or [],
audit_action=audit
audit_action=audit,
)
_HANDLER_REGISTRY[handler] = meta
# Attach to function for runtime introspection if needed
setattr(handler, "__openclaw_meta__", meta)
return handler
return decorator
@@ -101,47 +104,50 @@ def get_metadata(handler: Callable) -> Optional[EndpointMetadata]:
# Unwrap partials (common in aiohttp routes with bound methods)
while isinstance(handler, (functools.partial,)):
handler = handler.func
# Check registry first
if handler in _HANDLER_REGISTRY:
return _HANDLER_REGISTRY[handler]
# Check attribute
return getattr(handler, "__openclaw_meta__", None)
import functools
def generate_manifest(app) -> List[Dict[str, Any]]:
"""
Inspect application routes and generate a security manifest.
Returns a list of dicts describing each registered route and its metadata.
"""
manifest = []
for route in app.router.routes():
method = route.method
path = route.resource.canonical if route.resource else "unknown"
handler = route.handler
meta = get_metadata(handler)
entry = {
"method": method,
"path": path,
"handler": handler.__name__ if hasattr(handler, "__name__") else str(handler),
"handler": (
handler.__name__ if hasattr(handler, "__name__") else str(handler)
),
"classified": meta is not None,
"metadata": None
"metadata": None,
}
if meta:
entry["metadata"] = {
"auth": meta.auth_tier.value,
"risk": meta.risk_tier.value,
"summary": meta.summary,
"audit": meta.audit_action
"audit": meta.audit_action,
}
manifest.append(entry)
return manifest
+5
View File
@@ -618,6 +618,11 @@ class RuntimeConfig:
"OPENCLAW_ALLOW_INSECURE_BASE_URL", "MOLTBOT_ALLOW_INSECURE_BASE_URL", False
)
self.webhook_auth_mode = os.environ.get("OPENCLAW_WEBHOOK_AUTH_MODE", "")
self.security_dangerous_bind_override = _env_flag(
"OPENCLAW_SECURITY_DANGEROUS_BIND_OVERRIDE",
"MOLTBOT_SECURITY_DANGEROUS_BIND_OVERRIDE",
False,
)
self.admin_token_configured = bool(get_admin_token())
+100 -34
View File
@@ -19,23 +19,76 @@ class SecurityGate:
"""
@staticmethod
def verify_mandatory_controls() -> Tuple[bool, List[str]]:
def _check_network_exposure() -> bool:
"""
Check if the server is binding to a public/non-loopback interface.
Inspects sys.argv for '--listen' or '0.0.0.0'.
Returns:
bool: True if potentially exposed to network, False if loopback only.
"""
import sys
args = sys.argv
# Check for --listen flag (which defaults to 0.0.0.0 in ComfyUI)
if "--listen" in args:
return True
# Check for explicit host bind
# This is a heuristic; robust arg parsing is hard without importing main.
# But for security gate, false positive is better than false negative.
# If any arg looks like an IP that isn't loopback...
# For now, rely on --listen as the primary signal.
return False
@staticmethod
def verify_mandatory_controls() -> Tuple[bool, List[str], List[str]]:
"""
Check if all mandatory controls for the current profile are active.
Returns: (passed: bool, failure_reasons: List[str])
Returns: (passed: bool, warnings: List[str], fatal_errors: List[str])
"""
settings_issues = []
warnings = []
fatal_errors = []
# 1. Access Control (Auth)
# 1. Access Control (S45 Update)
try:
from .access_control import is_auth_configured
from .access_control import is_any_token_configured, is_auth_configured
if not is_auth_configured():
settings_issues.append(
"Authentication is NOT configured (Admin Token missing)"
)
is_exposed = 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()
if is_exposed and not auth_ready:
# Check for explicit override
from .runtime_config import get_config
config = get_config()
if config.security_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."
)
# Do NOT block startup (S45 Override Contract)
else:
# S45: Exposed + No Auth = FATAL (Always, regardless of profile)
fatal_errors.append(
"CRITICAL SECURITY RISK: Server is exposed (--listen) without Authentication!\n"
" Action Required: Set OPENCLAW_ADMIN_TOKEN (or OPENCLAW_OBSERVABILITY_TOKEN).\n"
" Startup is BLOCKED to prevent RCE.\n"
" (To bypass: set OPENCLAW_SECURITY_DANGEROUS_BIND_OVERRIDE=1)"
)
elif not auth_ready:
# 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():
warnings.append(
"HARDENED profile requires Admin Authentication even on loopback."
)
except ImportError:
settings_issues.append("Could not import access_control service")
warnings.append("Could not import access_control service")
# 2. Egress Policy (SSRF)
from .runtime_config import get_config
@@ -43,12 +96,12 @@ class SecurityGate:
config = get_config()
if config.allow_any_public_llm_host:
settings_issues.append(
warnings.append(
"OPENCLAW_ALLOW_ANY_PUBLIC_LLM_HOST is enabled (Egress check bypassed)"
)
if config.allow_insecure_base_url:
settings_issues.append(
warnings.append(
"OPENCLAW_ALLOW_INSECURE_BASE_URL is enabled (SSRF check bypassed)"
)
@@ -56,22 +109,19 @@ class SecurityGate:
from .modules import ModuleCapability, is_module_enabled
if is_module_enabled(ModuleCapability.WEBHOOK):
# Check if webhook auth is configured (loose check via config,
# ideally check specific auth mode but config implies it)
if not config.webhook_auth_mode:
settings_issues.append(
warnings.append(
"Webhook module enabled but OPENCLAW_WEBHOOK_AUTH_MODE not set"
)
# 4. Redaction
# (Redaction is always strictly imported in hardened mode; ensure it didn't fail)
try:
from .redaction import redact_text
if not callable(redact_text):
settings_issues.append("Redaction service is not callable")
warnings.append("Redaction service is not callable")
except ImportError:
settings_issues.append("Redaction service failed to import")
warnings.append("Redaction service failed to import")
# 5. Permission Posture (S42)
try:
@@ -81,20 +131,17 @@ class SecurityGate:
if not perm_allowed:
for res in perm_results:
if res.severity == "fail":
settings_issues.append(
f"Permission Check FAILED: {res.message}"
)
warnings.append(f"Permission Check FAILED: {res.message}")
except ImportError:
settings_issues.append("Permission posture service failed to import")
warnings.append("Permission posture service failed to import")
failures = []
# In HARDENED mode, treat all warnings as FATAL
if is_hardened_mode() and warnings:
fatal_errors.extend(warnings)
warnings = []
# In HARDENED mode, any issue is a failure.
if is_hardened_mode():
if settings_issues:
failures.extend(settings_issues)
return (len(failures) == 0), failures
passed = len(fatal_errors) == 0
return passed, warnings, fatal_errors
def enforce_startup_gate() -> None:
@@ -108,17 +155,36 @@ def enforce_startup_gate() -> None:
logger.info(f"Running S41 Security Gate ({mode_str} profile)...")
passed, issues = SecurityGate.verify_mandatory_controls()
passed, warnings, fatal_errors = SecurityGate.verify_mandatory_controls()
if passed:
# Log warnings first (non-blocking unless hardened)
if warnings:
warn_msg = f"Security Gate WARNINGS ({len(warnings)} issues):\n" + "\n".join(
[f"- {i}" for i in warnings]
)
if is_hardened:
# In Hardened mode, warnings become fatal.
logger.critical(warn_msg)
fatal_errors.append("HARDENED profile requires 0 warnings.")
else:
logger.warning(warn_msg)
if passed and not fatal_errors:
logger.info("Security Gate: PASS")
return
# Handle failures
error_msg = f"Security Gate FAILED ({len(issues)} issues):\n" + "\n".join(
[f"- {i}" for i in issues]
# Handle fatal errors (S45 Fail-Closed for Critical/Hardened failures)
error_msg = (
f"Security Gate FAILED ({len(fatal_errors)} fatal errors):\n"
+ "\n".join([f"- {i}" for i in fatal_errors])
)
logger.critical(error_msg)
logger.critical("FATAL: Security controls failed. Startup aborted.")
# S41/S45 Fail-Closed (Always raise for fatal errors)
raise RuntimeError(error_msg)
if is_hardened:
logger.critical(error_msg)
logger.critical(
+50 -32
View File
@@ -6,19 +6,26 @@ Ensures that:
2. No "shadow endpoints" exist without a known auth/risk classification.
"""
import unittest
from aiohttp import web
from unittest.mock import MagicMock, patch
import sys
import unittest
from unittest.mock import MagicMock, patch
from aiohttp import web
# Mock ComfyUI environment if needed
sys.modules["server"] = MagicMock()
from api.routes import register_routes
from services.endpoint_manifest import generate_manifest, AuthTier, RiskTier, get_metadata
from services.endpoint_manifest import (
AuthTier,
RiskTier,
generate_manifest,
get_metadata,
)
class TestEndpointDrift(unittest.TestCase):
def setUp(self):
self.app = web.Application()
# Mock server object to match what register_routes expects
@@ -30,7 +37,7 @@ class TestEndpointDrift(unittest.TestCase):
self.server.routes.post = MagicMock(return_value=lambda x: x)
self.server.routes.put = MagicMock(return_value=lambda x: x)
self.server.routes.delete = MagicMock(return_value=lambda x: x)
def tearDown(self):
# Clean up sys.modules hacks if any remain (none in this version)
pass
@@ -39,45 +46,54 @@ class TestEndpointDrift(unittest.TestCase):
"""
CRITICAL SECURITY CATCH:
Iterate over ALL registered routes and fail if any lack @endpoint_metadata.
Verification includes ensuring that optional modules (Bridge, Packs) are
Verification includes ensuring that optional modules (Bridge, Packs) are
actually enabled and registered during the test to prove they are guarded.
"""
# 1. Setup Environment
# We need config and services.modules to allow loading optional components
mock_config = MagicMock()
mock_config.DATA_DIR = "/tmp"
# Patch essential services to force-enable modules
with patch.dict(sys.modules, {
"config": mock_config,
# We must NOT mock api.bridge or api.packs here,
# we want the REAL modules to load and register REAL routes.
}), \
patch("services.modules.is_module_enabled", return_value=True):
# Re-import api.routes to pick up the patched environment if needed,
# though register_routes is what matters.
pass
# 2. Trigger route registration
register_routes(self.server)
with (
patch.dict(
sys.modules,
{
"config": mock_config,
# We must NOT mock api.bridge or api.packs here,
# we want the REAL modules to load and register REAL routes.
},
),
patch("services.modules.is_module_enabled", return_value=True),
):
# Re-import api.routes to pick up the patched environment if needed,
# though register_routes is what matters.
pass
# 2. Trigger route registration
register_routes(self.server)
# 3. Generate manifest from the actual aiohttp app
manifest = generate_manifest(self.app)
# 4. Verify Coverage (Prevent "Vacuous Truth" pass)
# Ensure that we actually registered the routes we expect to guard
methods_paths = [f"{m['method']} {m['path']}" for m in manifest]
# Check Bridge
has_bridge = any("bridge" in p for p in methods_paths)
if not has_bridge:
self.fail("Drift Test Configuration Error: Bridge routes were not registered! Test is not covering all endpoints.")
self.fail(
"Drift Test Configuration Error: Bridge routes were not registered! Test is not covering all endpoints."
)
# Check Packs
has_packs = any("packs" in p for p in methods_paths)
if not has_packs:
self.fail("Drift Test Configuration Error: Packs routes were not registered! Test is not covering all endpoints.")
self.fail(
"Drift Test Configuration Error: Packs routes were not registered! Test is not covering all endpoints."
)
# 5. Scan for unclassified routes
unclassified = []
@@ -85,32 +101,34 @@ class TestEndpointDrift(unittest.TestCase):
# Skip verify_handshake or other non-handler routes if any (aiohttp adds HEAD/OPTIONS sometimes)
if m["method"] not in ["GET", "POST", "PUT", "DELETE"]:
continue
if not m["classified"]:
unclassified.append(m)
if unclassified:
# Pretty print failure
fail_msg = f"Found {len(unclassified)} unclassified routes (R98 Drift):\n"
for u in unclassified:
fail_msg += f"- {u['method']} {u['path']} -> {u['handler']}\n"
self.fail(fail_msg)
def test_bridge_routes_have_metadata(self):
"""Verify bridge routes specifically have metadata."""
# This implicitly tests the AuthTier.BRIDGE existence and importability
from api.bridge import BridgeHandlers
handlers = BridgeHandlers()
meta = get_metadata(handlers.submit_handler)
self.assertIsNotNone(meta)
self.assertEqual(meta.auth_tier.value, "bridge")
meta = get_metadata(handlers.health_handler)
self.assertIsNotNone(meta)
# Health is PUBLIC (internally guarded)
self.assertEqual(meta.auth_tier.value, "public")
if __name__ == "__main__":
unittest.main()
+117
View File
@@ -0,0 +1,117 @@
import unittest
from unittest.mock import patch, MagicMock
import sys
import os
# We need to test the logic in security_gate.py
# Use a mock for sys.argv and services.access_control
class TestStartupGateLogic(unittest.TestCase):
def test_check_network_exposure_loopback(self):
"""Verify internal logic for loopback detection."""
from services.security_gate import SecurityGate
# Test case: No --listen args
with patch.object(sys, 'argv', ['main.py']):
is_exposed = SecurityGate._check_network_exposure()
self.assertFalse(is_exposed)
def test_check_network_exposure_listen(self):
"""Verify detection of --listen."""
from services.security_gate import SecurityGate
for arg in [['main.py', '--listen'], ['main.py', '--listen', '0.0.0.0']]:
with patch.object(sys, 'argv', arg):
is_exposed = SecurityGate._check_network_exposure()
self.assertTrue(is_exposed)
def test_enforcement_exposed_no_auth(self):
"""Exposed + No Auth = Fail."""
from services.security_gate import SecurityGate
# Patch the SOURCE of the import, because security_gate imports it inside the function
with patch.object(sys, 'argv', ['main.py', '--listen']), \
patch('services.access_control.is_any_token_configured', return_value=False):
# Mock other dependencies to isolate the S45 check
# We need to ensure we don't accidentally fail on other checks
mock_config = MagicMock()
mock_config.allow_any_public_llm_host = False
mock_config.allow_insecure_base_url = False
mock_config.webhook_auth_mode = "secret"
mock_config.security_dangerous_bind_override = False
with patch('services.modules.is_module_enabled', return_value=False), \
patch('services.runtime_config.get_config', return_value=mock_config), \
patch('services.security_gate.callable', return_value=True), \
patch('services.runtime_profile.is_hardened_mode', return_value=False):
passed, warnings, fatal_errors = SecurityGate.verify_mandatory_controls()
# Expect failure
self.assertFalse(passed)
self.assertTrue(any("CRITICAL SECURITY RISK" in i for i in fatal_errors))
def test_enforcement_exposed_no_auth_override(self):
"""Exposed + No Auth + Override = PASS (with Warning)."""
from services.security_gate import SecurityGate
with patch.object(sys, 'argv', ['main.py', '--listen']), \
patch('services.access_control.is_any_token_configured', return_value=False):
mock_config = MagicMock()
mock_config.allow_any_public_llm_host = False
mock_config.allow_insecure_base_url = False
mock_config.webhook_auth_mode = "secret"
# ENABLE OVERRIDE
mock_config.security_dangerous_bind_override = True
with patch('services.modules.is_module_enabled', return_value=False), \
patch('services.runtime_config.get_config', return_value=mock_config), \
patch('services.security_gate.callable', return_value=True), \
patch('services.runtime_profile.is_hardened_mode', return_value=False):
passed, warnings, fatal_errors = SecurityGate.verify_mandatory_controls()
self.assertTrue(passed, "Override should prevent FATAL errors, so it should PASS verification phase")
self.assertTrue(any("WARNING: Server is exposed" in i for i in warnings))
self.assertFalse(fatal_errors, "Should not report Critical Risk")
def test_enforcement_loopback_no_auth(self):
"""Loopback + No Auth = Pass (S45 Update)."""
from services.security_gate import SecurityGate
with patch.object(sys, 'argv', ['main.py']), \
patch('services.access_control.is_any_token_configured', return_value=False):
mock_config = MagicMock()
mock_config.allow_any_public_llm_host = False
mock_config.allow_insecure_base_url = False
mock_config.webhook_auth_mode = "secret" # Satisfy webhook check if enabled
with patch('services.modules.is_module_enabled', return_value=False), \
patch('services.runtime_config.get_config', return_value=mock_config), \
patch('services.security_gate.callable', return_value=True), \
patch('services.runtime_profile.is_hardened_mode', return_value=False):
passed, warnings, fatal = SecurityGate.verify_mandatory_controls()
# Expect PASS
self.assertTrue(passed, f"Should pass in loopback mode even without auth. Issues: {fatal}")
def test_enforcement_gate_crash(self):
"""Test that FATAL errors actually crash the app in Minimal mode."""
from services.security_gate import enforce_startup_gate, SecurityGate
# Simulate a FATAL condition (Exposed + No Auth)
# We Mock verify_mandatory_controls to return Fatal error
with patch.object(SecurityGate, 'verify_mandatory_controls', return_value=(False, [], ["FATAL ERROR"])):
with patch('services.runtime_profile.is_hardened_mode', return_value=False):
with self.assertRaises(RuntimeError) as cm:
enforce_startup_gate()
self.assertIn("FATAL ERROR", str(cm.exception))
if __name__ == '__main__':
unittest.main()
+40 -9
View File
@@ -24,11 +24,13 @@ class TestSecurityGate(unittest.TestCase):
cfg.allow_any_public_llm_host = False
cfg.allow_insecure_base_url = False
cfg.webhook_auth_mode = "bearer" # Satisfy webhook check if enabled (it is mocked false, but good to have)
cfg.security_dangerous_bind_override = False
mock_get_config.return_value = cfg
passed, issues = SecurityGate.verify_mandatory_controls()
self.assertTrue(passed, f"Gate failed with issues: {issues}")
self.assertEqual(len(issues), 0)
passed, warnings, fatal_errors = SecurityGate.verify_mandatory_controls()
self.assertTrue(passed, f"Gate failed with errors: {fatal_errors}")
self.assertEqual(len(warnings), 0)
self.assertEqual(len(fatal_errors), 0)
# Should not raise
enforce_startup_gate()
@@ -37,17 +39,35 @@ class TestSecurityGate(unittest.TestCase):
@patch(
"services.access_control.is_auth_configured", return_value=False
) # Fail auth
@patch(
"services.access_control.is_any_token_configured", return_value=False
) # Fail S45 auth
@patch("services.runtime_config.get_config")
def test_gate_fail_hardened(self, mock_get_config, mock_auth, mock_hardened):
"""Test gate raises exception in hardened mode on failure."""
def test_gate_fail_hardened(
self, mock_get_config, mock_any_auth, mock_auth, mock_hardened
):
"""Test gate logs warnings/errors in hardened mode."""
cfg = MagicMock()
cfg.allow_any_public_llm_host = False
cfg.allow_insecure_base_url = False # Clean config
cfg.security_dangerous_bind_override = False
mock_get_config.return_value = cfg
passed, issues = SecurityGate.verify_mandatory_controls()
self.assertFalse(passed)
self.assertIn("Authentication is NOT configured (Admin Token missing)", issues)
# Profile Hardened + No Auth (even on loopback) -> Warning -> Fatal
# Verify mandatory controls (S45)
# Note: verify_mandatory_controls imports access_control inside. Patching sys.modules or specific import might be needed if it wasn't mocked.
# But we patched services.access_control.is_auth_configured.
passed, warnings, fatal_errors = SecurityGate.verify_mandatory_controls()
# Hardened mode: warnings become fatal errors in verify_mandatory_controls?
# Code: "if is_hardened_mode() and warnings: fatal_errors.extend(warnings)"
self.assertFalse(passed)
# Expect "HARDENED profile requires Admin Authentication even on loopback."
self.assertTrue(any("HARDENED profile requires" in i for i in fatal_errors))
# Enforce should raise
with self.assertRaises(RuntimeError):
enforce_startup_gate()
@@ -55,11 +75,22 @@ class TestSecurityGate(unittest.TestCase):
@patch(
"services.access_control.is_auth_configured", return_value=False
) # Fail auth
@patch("services.access_control.is_any_token_configured", return_value=False)
@patch("services.runtime_config.get_config")
def test_gate_warn_minimal(self, mock_get_config, mock_auth, mock_hardened):
def test_gate_warn_minimal(
self, mock_get_config, mock_any_auth, mock_auth, mock_hardened
):
"""Test gate logs warning but does not raise in minimal mode."""
cfg = MagicMock()
# Ensure we don't trip S45 critical (exposed)
# By default mocks, _check_network_exposure returns what?
# We need to ensure we are in loopback mode.
# But verify_mandatory_controls calls _check_network_exposure() which looks at sys.argv.
# We should patch sys.argv or _check_network_exposure.
cfg.allow_any_public_llm_host = False
cfg.allow_insecure_base_url = False
cfg.security_dangerous_bind_override = False
mock_get_config.return_value = cfg
# Should NOT raise, just log warning
+3 -2
View File
@@ -30,11 +30,12 @@ class TestSecurityGatePermissions(unittest.TestCase):
mock_eval_perms.return_value = (False, [fail_res])
# Execute
passed, reasons = SecurityGate.verify_mandatory_controls()
passed, warnings, fatal_errors = SecurityGate.verify_mandatory_controls()
# Assert
self.assertFalse(passed)
self.assertTrue(any("Test Perm Fail" in r for r in reasons))
# Permission failures are added as warnings first, then moved to fatal in Hardened
self.assertTrue(any("Test Perm Fail" in r for r in fatal_errors))
@patch("services.security_gate.is_hardened_mode")
@patch("services.permission_posture.evaluate_startup_permissions")