fix: restrict CORS and guard scan endpoints by origin (#334)

agentic_security/app.py

@@ -2,6 +2,7 @@
 from .core.logging import setup_logging
 from .middleware.cors import setup_cors
 from .middleware.logging import LogNon200ResponsesMiddleware
+from .middleware.origin_guard import OriginGuardMiddleware
 from .routes import (
     probe_router,
     proxy_router,
@@ -15,6 +16,7 @@
 app = create_app()
 
 # Setup middleware
+app.add_middleware(OriginGuardMiddleware)
 setup_cors(app)
 app.add_middleware(LogNon200ResponsesMiddleware)
 

agentic_security/config.py

@@ -161,6 +161,11 @@ def generate_default_settings(self, host: str = "127.0.0.1", port: int = 8718):
 timeout_connect = 30
 timeout_response = 90
 
+[server]
+# Browser origins allowed to call the local scanner API (empty = same-origin only).
+# Override at runtime with AGENTIC_SECURITY_CORS_ORIGINS=http://localhost:3000,http://127.0.0.1:3000
+cors_allow_origins = []
+
 [fuzzer]
 max_prompt_lenght = 2048
 budget_multiplier = 100000000

agentic_security/middleware/cors.py

@@ -1,13 +1,38 @@
+import logging
+import os
+
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
+from agentic_security.config import settings_var
+
+logger = logging.getLogger(__name__)
+
+
+def get_cors_allow_origins() -> list[str]:
+    """Return configured browser origins allowed to call the local scanner API."""
+    env_val = os.getenv("AGENTIC_SECURITY_CORS_ORIGINS")
+    if env_val is not None:
+        origins = [origin.strip() for origin in env_val.split(",") if origin.strip()]
+    else:
+        configured = settings_var("server.cors_allow_origins", None)
+        origins = list(configured) if configured else []
+
+    if "*" in origins:
+        logger.warning(
+            "server.cors_allow_origins includes '*' — any web origin can drive "
+            "unauthenticated /scan and /verify against the local scanner"
+        )
+    return origins
+
 
-def setup_cors(app: FastAPI):
-    origins = ["*"]
+def setup_cors(app: FastAPI) -> None:
+    origins = get_cors_allow_origins()
 
     app.add_middleware(
         CORSMiddleware,
         allow_origins=origins,
-        allow_methods=["*"],  # Allows all methods
-        allow_headers=["*"],  # Allows all headers
+        allow_credentials=False,
+        allow_methods=["*"],
+        allow_headers=["*"],
     )

agentic_security/middleware/origin_guard.py

@@ -0,0 +1,58 @@
+"""Block cross-origin browser calls to state-changing scan endpoints.
+
+CORS alone does not stop simple POST requests (/stop, /scan-csv). This middleware
+allows same-origin and explicitly allowlisted origins; non-browser clients without
+an Origin header continue to work unchanged.
+"""
+
+from collections.abc import Callable
+
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+
+from agentic_security.middleware.cors import get_cors_allow_origins
+
+GUARDED_ROUTES: frozenset[tuple[str, str]] = frozenset(
+    {
+        ("/stop", "POST"),
+        ("/scan-csv", "POST"),
+        ("/verify", "POST"),
+        ("/scan", "POST"),
+    }
+)
+
+
+def _request_origin(request: Request) -> str | None:
+    origin = request.headers.get("origin")
+    if origin:
+        return origin.rstrip("/")
+    return None
+
+
+def _same_origin(request: Request, origin: str) -> bool:
+    host = request.headers.get("host")
+    if not host:
+        return False
+    scheme = request.url.scheme
+    return origin.rstrip("/") == f"{scheme}://{host}".rstrip("/")
+
+
+class OriginGuardMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request: Request, call_next: Callable) -> Response:
+        route = (request.url.path, request.method.upper())
+        if route not in GUARDED_ROUTES:
+            return await call_next(request)
+
+        origin = _request_origin(request)
+        if origin is None:
+            return await call_next(request)
+
+        if _same_origin(request, origin):
+            return await call_next(request)
+
+        allowed_origins = {allowed.rstrip("/") for allowed in get_cors_allow_origins()}
+        if origin.rstrip("/") in allowed_origins:
+            return await call_next(request)
+
+        return JSONResponse(status_code=403, content={"detail": "Origin not allowed"})

tests/unit/test_cors_middleware.py

@@ -1,66 +1,69 @@
-"""Unit tests for CORS middleware configuration.
+"""Tests for CORS allowlist configuration (agentic_security #334)."""
 
-Verifies that the wildcard-origins + allow_credentials=True spec violation
-(CORS spec §3.2, Fetch §4.7) has been removed.  Browsers silently strip
-credentials when the response carries Access-Control-Allow-Origin: * paired
-with Access-Control-Allow-Credentials: true, so the old config was both
-broken and misleading.
-"""
+import os
+from unittest.mock import patch
 
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.testclient import TestClient
 
-from agentic_security.middleware.cors import setup_cors
+from agentic_security.middleware.cors import get_cors_allow_origins, setup_cors
 
 
 def _get_cors_options(app: FastAPI) -> dict:
-    """Extract CORS middleware options from the app's middleware stack."""
     for middleware in app.user_middleware:
         if middleware.cls is CORSMiddleware:
             return middleware.kwargs
     return {}
 
 
 class TestCorsSetup:
-    """CORS middleware is configured correctly."""
-
-    def test_cors_middleware_is_registered(self):
-        """setup_cors adds CORSMiddleware to the app."""
-        app = FastAPI()
-        setup_cors(app)
-        cls_names = [m.cls.__name__ for m in app.user_middleware]
-        assert "CORSMiddleware" in cls_names
-
-    def test_wildcard_origins_without_credentials(self):
-        """allow_origins=['*'] must not be paired with allow_credentials=True.
-
-        The combination is forbidden by the CORS spec and causes browsers to
-        silently drop credentials on every cross-origin request.
-        """
+    def test_default_allowlist_is_empty(self):
+        with patch.dict(os.environ, {}, clear=True):
+            with patch(
+                "agentic_security.middleware.cors.settings_var",
+                return_value=[],
+            ):
+                assert get_cors_allow_origins() == []
+
+    def test_env_override_parses_comma_separated_origins(self):
+        with patch.dict(
+            os.environ,
+            {
+                "AGENTIC_SECURITY_CORS_ORIGINS": "http://localhost:3000, http://127.0.0.1:3000"
+            },
+            clear=True,
+        ):
+            assert get_cors_allow_origins() == [
+                "http://localhost:3000",
+                "http://127.0.0.1:3000",
+            ]
+
+    def test_cors_middleware_uses_allowlist_without_credentials(self):
         app = FastAPI()
-        setup_cors(app)
+        with patch(
+            "agentic_security.middleware.cors.get_cors_allow_origins",
+            return_value=["http://localhost:3000"],
+        ):
+            setup_cors(app)
         opts = _get_cors_options(app)
-        allow_origins = opts.get("allow_origins", [])
-        allow_credentials = opts.get("allow_credentials", False)
+        assert opts["allow_origins"] == ["http://localhost:3000"]
+        assert opts["allow_credentials"] is False
 
-        if "*" in allow_origins or allow_origins == ["*"]:
-            assert not allow_credentials, (
-                "allow_origins=['*'] with allow_credentials=True is invalid per "
-                "the CORS spec — browsers reject it and credentials are silently dropped"
-            )
-
-    def test_cors_allows_cross_origin_requests(self):
-        """Cross-origin preflight requests return a 200 with CORS headers."""
+    def test_preflight_from_allowlisted_origin_succeeds(self):
         app = FastAPI()
 
         @app.get("/probe")
         async def probe():
             return {"ok": True}
 
-        setup_cors(app)
-        client = TestClient(app, raise_server_exceptions=True)
+        with patch(
+            "agentic_security.middleware.cors.get_cors_allow_origins",
+            return_value=["http://localhost:3000"],
+        ):
+            setup_cors(app)
 
+        client = TestClient(app, raise_server_exceptions=True)
         response = client.options(
             "/probe",
             headers={
@@ -69,26 +72,30 @@ async def probe():
             },
         )
         assert response.status_code == 200
-        assert "access-control-allow-origin" in response.headers
+        assert (
+            response.headers.get("access-control-allow-origin")
+            == "http://localhost:3000"
+        )
 
-    def test_cors_no_credentials_header_with_wildcard(self):
-        """With wildcard origins, the response must not include
-        Access-Control-Allow-Credentials: true."""
+    def test_preflight_from_foreign_origin_is_rejected_with_empty_allowlist(self):
         app = FastAPI()
 
         @app.get("/probe")
         async def probe():
             return {"ok": True}
 
-        setup_cors(app)
-        client = TestClient(app)
-        response = client.get("/probe", headers={"Origin": "http://evil.example.com"})
+        with patch(
+            "agentic_security.middleware.cors.get_cors_allow_origins",
+            return_value=[],
+        ):
+            setup_cors(app)
 
-        acao = response.headers.get("access-control-allow-origin", "")
-        acac = response.headers.get("access-control-allow-credentials", "false")
-
-        if acao == "*":
-            assert acac.lower() != "true", (
-                "Wildcard ACAO + ACAC:true is a spec violation (RFC 6454 §7.2, "
-                "Fetch §4.7) and silently breaks credentialed cross-origin requests"
-            )
+        client = TestClient(app, raise_server_exceptions=True)
+        response = client.options(
+            "/probe",
+            headers={
+                "Origin": "http://evil.example.com",
+                "Access-Control-Request-Method": "GET",
+            },
+        )
+        assert response.status_code == 400

tests/unit/test_origin_guard.py

@@ -0,0 +1,70 @@
+"""Tests for OriginGuardMiddleware (agentic_security #334)."""
+
+from unittest.mock import patch
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from agentic_security.middleware.origin_guard import OriginGuardMiddleware
+
+
+def _app_with_guard() -> FastAPI:
+    app = FastAPI()
+    app.add_middleware(OriginGuardMiddleware)
+
+    @app.post("/stop")
+    async def stop_scan():
+        return {"status": "Scan stopped"}
+
+    @app.post("/scan-csv")
+    async def scan_csv():
+        return {"status": "ok"}
+
+    @app.get("/health")
+    async def health():
+        return {"ok": True}
+
+    return app
+
+
+class TestOriginGuardMiddleware:
+    def test_allows_requests_without_origin_header(self):
+        client = TestClient(_app_with_guard())
+        assert client.post("/stop").status_code == 200
+
+    def test_blocks_foreign_origin_on_simple_post(self):
+        client = TestClient(_app_with_guard())
+        response = client.post(
+            "/stop",
+            headers={"Origin": "http://evil.example.com"},
+        )
+        assert response.status_code == 403
+
+    def test_allows_same_origin_post(self):
+        client = TestClient(_app_with_guard(), base_url="http://testserver")
+        response = client.post(
+            "/stop",
+            headers={"Origin": "http://testserver"},
+        )
+        assert response.status_code == 200
+
+    def test_allows_explicitly_allowlisted_origin(self):
+        app = _app_with_guard()
+        with patch(
+            "agentic_security.middleware.origin_guard.get_cors_allow_origins",
+            return_value=["http://localhost:3000"],
+        ):
+            client = TestClient(app)
+            response = client.post(
+                "/scan-csv",
+                headers={"Origin": "http://localhost:3000"},
+            )
+        assert response.status_code == 200
+
+    def test_does_not_guard_unlisted_routes(self):
+        client = TestClient(_app_with_guard())
+        response = client.get(
+            "/health",
+            headers={"Origin": "http://evil.example.com"},
+        )
+        assert response.status_code == 200