diff --git a/agentic_security/app.py b/agentic_security/app.py index 568bae0b..4d55d528 100644 --- a/agentic_security/app.py +++ b/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) diff --git a/agentic_security/config.py b/agentic_security/config.py index a26f952c..def95ff1 100644 --- a/agentic_security/config.py +++ b/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 diff --git a/agentic_security/middleware/cors.py b/agentic_security/middleware/cors.py index fb65051d..c1a6fff8 100644 --- a/agentic_security/middleware/cors.py +++ b/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=["*"], ) diff --git a/agentic_security/middleware/origin_guard.py b/agentic_security/middleware/origin_guard.py new file mode 100644 index 00000000..69e46178 --- /dev/null +++ b/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"}) diff --git a/tests/unit/test_cors_middleware.py b/tests/unit/test_cors_middleware.py index e8d0696f..a044faab 100644 --- a/tests/unit/test_cors_middleware.py +++ b/tests/unit/test_cors_middleware.py @@ -1,21 +1,16 @@ -"""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 @@ -23,44 +18,52 @@ def _get_cors_options(app: FastAPI) -> dict: 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 diff --git a/tests/unit/test_origin_guard.py b/tests/unit/test_origin_guard.py new file mode 100644 index 00000000..3574697a --- /dev/null +++ b/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