diff --git a/api/events.py b/api/events.py index 4fbeeb0..26d4429 100644 --- a/api/events.py +++ b/api/events.py @@ -60,9 +60,9 @@ async def events_stream_handler(request: web.Request) -> web.StreamResponse: ) # Access control (same as logs/tail) - denied = require_observability_access(request) - if denied: - return denied + allowed, error = require_observability_access(request) + if not allowed: + return web.json_response({"ok": False, "error": error}, status=403) store = get_job_event_store() @@ -154,9 +154,9 @@ async def events_poll_handler(request: web.Request) -> web.Response: ) # Access control - denied = require_observability_access(request) - if denied: - return denied + allowed, error = require_observability_access(request) + if not allowed: + return web.json_response({"ok": False, "error": error}, status=403) store = get_job_event_store() diff --git a/services/access_control.py b/services/access_control.py index 580fb62..0beed5e 100644 --- a/services/access_control.py +++ b/services/access_control.py @@ -126,7 +126,7 @@ def require_admin_token(request) -> Tuple[bool, Optional[str]]: if not is_same_origin_request(request): return ( False, - "Cross-origin request denied in convenience mode (S27). Set OPENCLAW_ADMIN_TOKEN to use token-based auth.", + "Cross-origin request denied (S33). Set OPENCLAW_ADMIN_TOKEN to use token-based auth.", ) return True, None diff --git a/services/csrf_protection.py b/services/csrf_protection.py index 1adfb74..0047e8e 100644 --- a/services/csrf_protection.py +++ b/services/csrf_protection.py @@ -70,12 +70,22 @@ def is_same_origin_request(request: web.Request) -> bool: logger.debug(f"S26+: Disallowed origin: {origin}") return False - # No Origin/Sec-Fetch-Site header (old browser or direct tool) - # Allow for backwards compat, but log - logger.debug( - "S26+: No Origin or Sec-Fetch-Site header; allowing (backwards compat)" + # No Origin/Sec-Fetch-Site header (old browser or direct tool like curl) + # S33: Strict default. explicit fallback required. + allow_no_origin = ( + os.environ.get("OPENCLAW_LOCALHOST_ALLOW_NO_ORIGIN", "").lower() == "true" ) - return True + + if allow_no_origin: + logger.debug( + "S26+: No Origin or Sec-Fetch-Site header; allowing (OPENCLAW_LOCALHOST_ALLOW_NO_ORIGIN=true)" + ) + return True + + logger.debug( + "S33: No Origin or Sec-Fetch-Site header; denying (strict localhost mode)." + ) + return False def require_same_origin_if_no_token( diff --git a/tests/test_access_control.py b/tests/test_access_control.py index 50450a7..c856f2b 100644 --- a/tests/test_access_control.py +++ b/tests/test_access_control.py @@ -26,6 +26,7 @@ class TestAccessControl(unittest.TestCase): req.remote = "127.0.0.1" req.headers = {} + # S33 (Relaxed): Observability allows simple loopback for monitoring apps allowed, error = require_observability_access(req) self.assertTrue(allowed) self.assertIsNone(error) diff --git a/tests/test_s28s29_chat_csrf_redaction.py b/tests/test_s28s29_chat_csrf_redaction.py index 0814804..4f387cb 100644 --- a/tests/test_s28s29_chat_csrf_redaction.py +++ b/tests/test_s28s29_chat_csrf_redaction.py @@ -90,15 +90,27 @@ class TestS28ChatCSRFGuard(unittest.IsolatedAsyncioTestCase): @patch("api.config.get_admin_token", return_value="") @patch("api.config.check_rate_limit", return_value=True) @patch("api.config.require_admin_token", return_value=(True, None)) - async def test_no_origin_header_allowed_backwards_compat( + async def test_no_origin_header_denied_by_default_s33( self, _admin, _rate, _get_token ): - """No Origin/Sec-Fetch-Site header: allowed for backwards compat.""" + """S33: No Origin/Sec-Fetch-Site header: Denied by default (strict).""" request = _make_request( body={"user_message": "hello"}, ) resp = await llm_chat_handler(request) - self.assertNotEqual(resp.status, 403, "No-header should be allowed") + self.assertEqual(resp.status, 403, "No-header should be denined by S33 default") + + @patch("api.config.get_admin_token", return_value="") + @patch("api.config.check_rate_limit", return_value=True) + @patch("api.config.require_admin_token", return_value=(True, None)) + @patch.dict("os.environ", {"OPENCLAW_LOCALHOST_ALLOW_NO_ORIGIN": "true"}) + async def test_no_origin_allowed_with_legacy_flag(self, _admin, _rate, _get_token): + """S33: No Origin allowed if OPENCLAW_LOCALHOST_ALLOW_NO_ORIGIN=true.""" + request = _make_request( + body={"user_message": "hello"}, + ) + resp = await llm_chat_handler(request) + self.assertNotEqual(resp.status, 403, "Legacy flag should allow no-origin") @unittest.skipIf(web is None, "aiohttp not installed") diff --git a/tests/test_s33r76r78_auth_hardening.py b/tests/test_s33r76r78_auth_hardening.py new file mode 100644 index 0000000..0555da2 --- /dev/null +++ b/tests/test_s33r76r78_auth_hardening.py @@ -0,0 +1,213 @@ +""" +S33/R76/R78 auth hardening contract tests. + +Covers: +- S33: strict localhost same-origin behavior for state-changing requests. +- R76: observability auth denial parity for events poll/stream handlers. +- R78: route auth-class manifest coverage + handler contract checks. +""" + +import inspect +import os +import unittest +from unittest.mock import MagicMock, patch + +# CI guardrail: keep test importable when aiohttp is absent. +try: + from aiohttp import web +except ImportError: # pragma: no cover + web = None + +from services.csrf_protection import is_same_origin_request + +# R78: Explicit auth class contract per method+route suffix. +# Any non-optional /openclaw|/moltbot route added in register_routes must be +# classified here, otherwise tests fail. +AUTH_CLASS_BY_ROUTE = { + ("GET", "/health"): "public-safe", + ("GET", "/logs/tail"): "observability", + ("GET", "/jobs"): "public-safe", + ("GET", "/trace/{prompt_id}"): "observability", + ("POST", "/webhook"): "webhook-auth", + ("POST", "/webhook/submit"): "webhook-auth", + ("POST", "/webhook/validate"): "webhook-auth", + ("GET", "/capabilities"): "public-safe", + ("GET", "/config"): "observability", + ("PUT", "/config"): "admin", + ("POST", "/llm/test"): "admin", + ("POST", "/llm/chat"): "admin", + ("GET", "/llm/models"): "admin", + ("GET", "/templates"): "observability", + ("POST", "/preflight"): "admin", + ("GET", "/preflight/inventory"): "admin", + ("GET", "/checkpoints"): "admin", + ("POST", "/checkpoints"): "admin", + ("GET", "/checkpoints/{id}"): "admin", + ("DELETE", "/checkpoints/{id}"): "admin", + ("GET", "/secrets/status"): "admin", + ("PUT", "/secrets"): "admin", + ("DELETE", "/secrets/{provider}"): "admin", + ("GET", "/events/stream"): "observability", + ("GET", "/events"): "observability", + ("GET", "/security/doctor"): "admin", +} + +OPTIONAL_SUFFIX_PREFIXES = ( + "/assist/", + "/packs", +) + + +def _strip_prefix(path: str): + for prefix in ("/openclaw", "/moltbot"): + if path.startswith(prefix): + return prefix, path[len(prefix) :] + return None, None + + +@unittest.skipIf(web is None, "aiohttp not installed") +class TestS33LocalhostHardening(unittest.TestCase): + def setUp(self): + self.patcher = patch.dict(os.environ, {}, clear=True) + self.patcher.start() + self.addCleanup(self.patcher.stop) + + @staticmethod + def _make_req(headers): + req = MagicMock() + req.headers = headers + return req + + def test_strict_origin_defaults(self): + req = self._make_req({"Sec-Fetch-Site": "same-origin"}) + self.assertTrue(is_same_origin_request(req)) + + req = self._make_req({"Sec-Fetch-Site": "cross-site"}) + self.assertFalse(is_same_origin_request(req)) + + req = self._make_req({}) + self.assertFalse(is_same_origin_request(req)) + + def test_legacy_origin_flag(self): + os.environ["OPENCLAW_LOCALHOST_ALLOW_NO_ORIGIN"] = "true" + req = self._make_req({}) + self.assertTrue(is_same_origin_request(req)) + + +@unittest.skipIf(web is None, "aiohttp not installed") +class TestR76ObservabilityAuth(unittest.IsolatedAsyncioTestCase): + async def test_events_handlers_return_403_on_denial(self): + with ( + patch( + "api.events.require_observability_access", + return_value=(False, "denied"), + ), + patch("api.events.check_rate_limit", return_value=True), + ): + from api.events import events_poll_handler, events_stream_handler + + req = MagicMock() + req.headers = {} + req.query = {} + + poll_resp = await events_poll_handler(req) + self.assertEqual(poll_resp.status, 403) + + stream_resp = await events_stream_handler(req) + self.assertEqual(stream_resp.status, 403) + + +@unittest.skipIf(web is None, "aiohttp not installed") +class TestR78AuthMatrix(unittest.TestCase): + @staticmethod + def _register_and_collect_routes(): + from api.routes import register_routes + + server = MagicMock() + server.routes.get = MagicMock() + server.routes.post = MagicMock() + server.routes.put = MagicMock() + server.routes.delete = MagicMock() + server.app.router.add_route = MagicMock() + + register_routes(server) + + rows = [] + for call in server.app.router.add_route.call_args_list: + method, path, handler = call.args + rows.append((method, path, handler)) + return rows + + def test_auth_manifest_covers_registered_core_routes(self): + rows = self._register_and_collect_routes() + by_method_path = {(method, path) for method, path, _handler in rows} + + # 1) Explicit expectations for all mandatory core routes (both namespaces + /api). + for method, suffix in AUTH_CLASS_BY_ROUTE: + for prefix in ("/openclaw", "/moltbot"): + base = f"{prefix}{suffix}" + api = f"/api{base}" + self.assertIn((method, base), by_method_path) + self.assertIn((method, api), by_method_path) + + # 2) Drift guard: any newly-registered non-optional core route must be classified. + unclassified = [] + for method, path, _handler in rows: + if path.startswith("/api"): + continue + prefix, suffix = _strip_prefix(path) + if prefix is None: + continue + if any(suffix.startswith(p) for p in OPTIONAL_SUFFIX_PREFIXES): + continue + if (method, suffix) not in AUTH_CLASS_BY_ROUTE: + unclassified.append((method, path)) + + self.assertEqual( + [], + unclassified, + msg=f"Unclassified core routes detected: {unclassified}", + ) + + def test_handler_auth_contract_matches_declared_class(self): + rows = self._register_and_collect_routes() + + # Pick /openclaw base routes only to avoid duplicate checks across aliases. + handler_by_route = {} + for method, path, handler in rows: + if not path.startswith("/openclaw"): + continue + _prefix, suffix = _strip_prefix(path) + if suffix is None: + continue + if any(suffix.startswith(p) for p in OPTIONAL_SUFFIX_PREFIXES): + continue + handler_by_route[(method, suffix)] = handler + + missing = [k for k in AUTH_CLASS_BY_ROUTE if k not in handler_by_route] + self.assertEqual([], missing, msg=f"Missing handlers for routes: {missing}") + + for route_key, auth_class in AUTH_CLASS_BY_ROUTE.items(): + handler = handler_by_route[route_key] + source = inspect.getsource(handler) + + if auth_class == "observability": + self.assertIn("require_observability_access", source, msg=route_key) + elif auth_class == "admin": + self.assertTrue( + ("require_admin_token(" in source or "_require_admin(" in source), + msg=f"{route_key} expected admin guard", + ) + elif auth_class == "webhook-auth": + self.assertIn("require_auth(", source, msg=route_key) + elif auth_class == "public-safe": + self.assertNotIn("require_admin_token(", source, msg=route_key) + self.assertNotIn("_require_admin(", source, msg=route_key) + self.assertNotIn("require_observability_access", source, msg=route_key) + self.assertNotIn("require_auth(", source, msg=route_key) + else: + self.fail(f"Unknown auth class for {route_key}: {auth_class}") + + +if __name__ == "__main__": # pragma: no cover + unittest.main()