From 212bd64e3bdc41d6b17afa9a8a9acc208c4c4c7c Mon Sep 17 00:00:00 2001 From: Renn F Date: Thu, 15 Oct 2026 12:01:42 +0200 Subject: [PATCH] fix(extension): route passthrough, bypass and response pass through the engine like fastapi-guard Use the engine BypassHandler for passthrough (fail_secure 403 without a client address) and for the route all-checks bypass, which now calls the real view. Run the response pass only for responses produced by the view, so blocked, preflight and bypass responses emit no request metrics. Pass global behavior return rules to the engine response pass. --- flaskapi_guard/extension.py | 91 ++++----- tests/test_engine_parity.py | 186 ++++++++++++++++++ .../test_extension/test_security_extension.py | 6 +- 3 files changed, 232 insertions(+), 51 deletions(-) create mode 100644 tests/test_engine_parity.py diff --git a/flaskapi_guard/extension.py b/flaskapi_guard/extension.py index cedf803..386ce6c 100644 --- a/flaskapi_guard/extension.py +++ b/flaskapi_guard/extension.py @@ -426,71 +426,49 @@ def _handle_preflight( preflight = self._cors_handler.build_preflight_response(request_headers) return self._build_flask_preflight_response(preflight) - def _check_passthrough( - self, - guard_request: FlaskGuardRequest, - ) -> None: - assert self.validator is not None - if not guard_request.client_host: - g.guard_passthrough = True - return - if self.validator.is_path_excluded(guard_request): - guard_request.state.guard_exclusion_scoped = True - return + def _call_view(self, _guard_request: Any) -> FlaskGuardResponse: + from flask import current_app, request - def _check_security_bypass( - self, - guard_request: FlaskGuardRequest, - route_config: RouteConfig | None, - ) -> None: - assert self.bypass_handler is not None - assert self.route_resolver is not None - assert self.event_bus is not None - assert self.config is not None - if not route_config: - return - if not self.route_resolver.should_bypass_check("all", route_config): - return - self.event_bus.send_middleware_event( - event_type="security_bypass", - request=guard_request, - action_taken="all_checks_bypassed", - reason="Route configured to bypass all security checks", - bypassed_checks=list(route_config.bypassed_checks), - endpoint=str(guard_request.url_path), - ) - if not self.config.passive_mode: - g.guard_passthrough = True - return + assert request.endpoint is not None + view = current_app.ensure_sync(current_app.view_functions[request.endpoint]) + rv = view(**(request.view_args or {})) + return FlaskGuardResponse(current_app.make_response(rv)) - def _before_request(self) -> Response | None: + def _early_response( + self, early: GuardResponse | None, request_headers: dict[str, str] + ) -> Response | None: + if early is None: + return None + return self._attach_cors_to_blocked(unwrap_response(early), request_headers) + + def _guard_request_phase(self) -> Response | None: from flask import request assert self.config is not None assert self.bypass_handler is not None assert self.route_resolver is not None - assert self.behavioral_processor is not None init_failure = self._ensure_initialized() + request_headers = dict(request.headers) if init_failure is not None: - return self._attach_cors_to_blocked(init_failure, dict(request.headers)) + return self._attach_cors_to_blocked(init_failure, request_headers) g.request_start_time = time.time() guard_request = FlaskGuardRequest(request) self._populate_guard_state(guard_request) - request_headers = dict(request.headers) - preflight_response = self._handle_preflight( guard_request, request.method, request_headers ) if preflight_response is not None: return preflight_response - self._check_passthrough(guard_request) - - if getattr(g, "guard_passthrough", False): - return None + early = self._early_response( + self.bypass_handler.handle_passthrough(guard_request, self._call_view), + request_headers, + ) + if early is not None: + return early client_ip = extract_client_ip(guard_request, self.config, self.agent_handler) route_config = self.route_resolver.get_route_config(guard_request) @@ -498,22 +476,34 @@ def _before_request(self) -> Response | None: g.client_ip = client_ip g.route_config = route_config - self._check_security_bypass(guard_request, route_config) - - if getattr(g, "guard_passthrough", False): - return None + early = self._early_response( + self.bypass_handler.handle_security_bypass( + guard_request, self._call_view, route_config + ), + request_headers, + ) + if early is not None: + return early blocking = self._execute_security_pipeline(guard_request) if blocking is not None: return self._attach_cors_to_blocked(blocking, request_headers) self._process_behavioral_usage(guard_request, client_ip, route_config) - return None + def _before_request(self) -> Response | None: + early = self._guard_request_phase() + if early is None: + g.guard_response_pass = True + return early + def _after_request(self, response: Response) -> Response: from flask import request + if not getattr(g, "guard_response_pass", False): + return response + assert self.response_factory is not None assert self.behavioral_processor is not None @@ -529,6 +519,9 @@ def _after_request(self, response: Response) -> Response: response_time, route_config, process_behavioral_rules=self.behavioral_processor.process_return_rules, + process_global_behavioral_rules=( + self.behavioral_processor.process_global_return_rules + ), ) flask_response = unwrap_response(result) diff --git a/tests/test_engine_parity.py b/tests/test_engine_parity.py new file mode 100644 index 0000000..ba7551f --- /dev/null +++ b/tests/test_engine_parity.py @@ -0,0 +1,186 @@ +"""Engine parity: the adapter routes through the engine exactly like fastapi-guard. + +Real Flask test client, real engine bypass handler and response factory. The +metrics collector is spied on (it still runs) to prove which responses reach +the engine response pass. +""" + +from typing import Any + +from flask import Flask +from guard_core.protocols.response_protocol import GuardResponse +from guard_core.sync.handlers.ipban_handler import reset_global_state + +from flaskapi_guard import SecurityConfig, SecurityDecorator +from flaskapi_guard.extension import FlaskAPIGuard + +NO_CLIENT: dict[str, Any] = {"REMOTE_ADDR": None} + + +def _build( + **overrides: Any, +) -> tuple[Flask, FlaskAPIGuard, list[str]]: + base: dict[str, Any] = { + "enable_redis": False, + "enable_penetration_detection": False, + "whitelist": None, + "blacklist": [], + "exclude_paths": ["/health"], + "security_headers": {"enabled": True}, + } + base.update(overrides) + config = SecurityConfig(**base) + app = Flask(__name__) + app.config["TESTING"] = True + guard = FlaskAPIGuard(app, config=config) + sec = SecurityDecorator(config) + guard.set_decorator_handler(sec) + metrics: list[str] = [] + assert guard.metrics_collector is not None + real = guard.metrics_collector.collect_request_metrics + + def spy(request: Any, response_time: float, status_code: int) -> None: + metrics.append(request.url_path) + real(request, response_time, status_code) + + guard.metrics_collector.collect_request_metrics = spy # type: ignore[method-assign] + + @app.route("/health") + def health() -> dict[str, str]: + return {"message": "ok"} + + @app.route("/open") + def open_() -> dict[str, str]: + return {"message": "Echo response"} + + @app.route("/bypassed") + @sec.bypass(["all"]) + def bypassed() -> dict[str, str]: + return {"message": "real view"} + + return app, guard, metrics + + +def test_excluded_path_reaches_view_with_scoped_checks_and_response_pass() -> None: + app, _guard, metrics = _build(blacklist=["10.9.9.9"]) + client = app.test_client() + + ok = client.get("/health") + assert ok.status_code == 200 + assert ok.get_json() == {"message": "ok"} + assert ok.headers.get("X-Content-Type-Options") == "nosniff" + assert metrics == ["/health"] + + blocked = client.get("/health", environ_overrides={"REMOTE_ADDR": "10.9.9.9"}) + assert blocked.status_code == 403 + assert metrics == ["/health"] + + +def test_no_client_address_gets_fail_secure_403_by_default() -> None: + app, _guard, metrics = _build() + client = app.test_client() + + resp = client.get("/open", environ_overrides=NO_CLIENT) + + assert resp.status_code == 403 + assert "Client address could not be determined" in resp.get_data(as_text=True) + assert metrics == [] + + +def test_no_client_address_on_excluded_path_is_scoped_not_bypassed() -> None: + app, _guard, metrics = _build() + client = app.test_client() + + resp = client.get("/health", environ_overrides=NO_CLIENT) + + assert resp.status_code == 200 + assert metrics == ["/health"] + + +def test_no_client_address_fail_secure_false_runs_the_pipeline() -> None: + app, _guard, metrics = _build(fail_secure=False) + client = app.test_client() + + resp = client.get("/open", environ_overrides=NO_CLIENT) + + assert resp.status_code == 200 + assert metrics == ["/open"] + + +def test_route_bypass_reaches_real_view_and_skips_response_pass() -> None: + seen: list[int] = [] + + def modifier(response: GuardResponse) -> GuardResponse: + seen.append(response.status_code) + response.headers["X-Modified"] = "1" + return response + + app, _guard, metrics = _build(custom_response_modifier=modifier) + client = app.test_client() + + resp = client.get("/bypassed") + + assert resp.status_code == 200 + assert resp.get_json() == {"message": "real view"} + assert resp.headers["X-Modified"] == "1" + assert seen == [200] + assert metrics == [] + assert "X-Content-Type-Options" not in resp.headers + + +def test_blocked_request_emits_no_request_metrics() -> None: + app, _guard, metrics = _build(blacklist=["10.9.9.9"]) + client = app.test_client() + + blocked = client.get("/open", environ_overrides={"REMOTE_ADDR": "10.9.9.9"}) + assert blocked.status_code == 403 + assert metrics == [] + + allowed = client.get("/open") + assert allowed.status_code == 200 + assert metrics == ["/open"] + + +def test_preflight_emits_no_request_metrics() -> None: + app, _guard, metrics = _build( + enable_cors=True, cors_allow_origins=["https://example.com"] + ) + client = app.test_client() + + resp = client.options( + "/open", + headers={ + "Origin": "https://example.com", + "Access-Control-Request-Method": "GET", + }, + ) + + assert resp.status_code == 200 + assert resp.headers["Access-Control-Allow-Origin"] == "https://example.com" + assert metrics == [] + + +def test_global_behavior_return_rule_fires() -> None: + reset_global_state() + app, _guard, _metrics = _build( + behavior_scan_response_body=True, + global_behavior_rules=[ + { + "rule_type": "return_pattern", + "threshold": 2, + "window": 60, + "pattern": "json:message==Echo response", + "action": "ban", + "ban_duration": 30, + } + ], + ) + client = app.test_client() + + remote = {"REMOTE_ADDR": "203.0.113.9"} + codes: list[int] = [ + client.get("/open", environ_overrides=remote).status_code for _ in range(5) + ] + + assert codes[0] == 200 + assert 403 in codes diff --git a/tests/test_extension/test_security_extension.py b/tests/test_extension/test_security_extension.py index 629d936..f69b6c1 100644 --- a/tests/test_extension/test_security_extension.py +++ b/tests/test_extension/test_security_extension.py @@ -1695,8 +1695,10 @@ def root() -> dict[str, str]: ): with app.test_client() as client: response = client.get("/") - assert response.status_code == 200 - assert response.get_json() == {"msg": "ok"} + assert response.status_code == 403 + assert "Client address could not be determined" in response.get_data( + as_text=True + ) with app.test_client() as client: baseline = client.get("/")