diff --git a/bot_bottle/gateway/egress/addon.py b/bot_bottle/gateway/egress/addon.py index d235ba9f..aa6f6271 100644 --- a/bot_bottle/gateway/egress/addon.py +++ b/bot_bottle/gateway/egress/addon.py @@ -33,11 +33,12 @@ from bot_bottle.gateway.egress.dlp import ( ) from bot_bottle.gateway.egress.matching import ( decide, - decide_git_fetch, - is_git_fetch_request, - is_git_push_request, match_route, ) +from bot_bottle.gateway.egress.request_pipeline import ( + evaluate_route_policy, + git_block_reason, +) from bot_bottle.gateway.egress.schema import route_to_yaml_dict from bot_bottle.gateway.egress.types import ( LOG_BLOCKS, @@ -435,21 +436,12 @@ class EgressAddon: request_path: str, query: str, ) -> bool: """Apply the HTTPS Git push/fetch boundary before general routing.""" - if is_git_push_request(request_path, query): - self._block( - flow, - "egress: git push over HTTPS is not supported; " - "use the bottle.git SSH path (gitleaks-scanned by " - "git-gate's pre-receive hook).", - ctx=self._req_ctx(flow), - ) - return False - if not is_git_fetch_request(request_path, query): + reason = git_block_reason( + config.routes, flow.request.pretty_host, request_path, query, + ) + if not reason: return True - git_decision = decide_git_fetch(config.routes, flow.request.pretty_host) - if git_decision.action != "block": - return True - self._block(flow, git_decision.reason, ctx=self._req_ctx(flow)) + self._block(flow, reason, ctx=self._req_ctx(flow)) return False def _apply_route_policy( @@ -461,30 +453,26 @@ class EgressAddon: # are caught above; the route may inject gateway-owned auth below. # Routes with preserve_auth=True pass the header through as-is so the # agent's own credentials (e.g. registry bearer tokens) reach the upstream. - if route is None or not route.preserve_auth: + result = evaluate_route_policy( + config, + route, + host=flow.request.pretty_host, + request_path=request_path, + method=flow.request.method, + headers=dict(flow.request.headers), + env=env, + ) + if result.strip_authorization: flow.request.headers.pop("authorization", None) - # Build headers mapping for match evaluation - req_headers = {k.lower(): v for k, v in flow.request.headers.items()} - - decision = decide( - config.routes, - flow.request.pretty_host, - request_path, - env, - request_method=flow.request.method, - request_headers=req_headers, - deny_reason=config.deny_reason, - ) - - if decision.action == "block": - self._block(flow, decision.reason, ctx=self._req_ctx(flow)) + if result.block_reason: + self._block(flow, result.block_reason, ctx=self._req_ctx(flow)) return - if decision.inject_authorization is not None: - flow.request.headers["authorization"] = decision.inject_authorization + if result.inject_authorization is not None: + flow.request.headers["authorization"] = result.inject_authorization - if config.log >= LOG_FULL: + if result.log_request: self._log_request(flow, env) def _block_dlp(self, flow: http.HTTPFlow, result: ScanResult) -> None: diff --git a/bot_bottle/gateway/egress/request_pipeline.py b/bot_bottle/gateway/egress/request_pipeline.py new file mode 100644 index 00000000..e9962248 --- /dev/null +++ b/bot_bottle/gateway/egress/request_pipeline.py @@ -0,0 +1,92 @@ +"""Framework-neutral request policy stages for the egress adapter. + +The mitmproxy addon owns flow mutation and response construction. This module +owns the ordered Git and route-policy decisions so those rules remain directly +testable without a live proxy flow. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Mapping, Sequence + +from .matching import ( + decide, + decide_git_fetch, + is_git_fetch_request, + is_git_push_request, +) +from .types import LOG_FULL, Config, Route + +GIT_PUSH_BLOCK_REASON = ( + "egress: git push over HTTPS is not supported; " + "use the bottle.git SSH path (gitleaks-scanned by " + "git-gate's pre-receive hook)." +) + + +@dataclass(frozen=True) +class RoutePolicyResult: + """The flow mutations and outcome produced by general route policy.""" + + block_reason: str = "" + strip_authorization: bool = False + inject_authorization: str | None = None + log_request: bool = False + + +def git_block_reason( + routes: Sequence[Route], + host: str, + request_path: str, + query: str, +) -> str: + """Return the HTTPS Git policy denial, or ``""`` when allowed.""" + if is_git_push_request(request_path, query): + return GIT_PUSH_BLOCK_REASON + if not is_git_fetch_request(request_path, query): + return "" + decision = decide_git_fetch(routes, host) + return decision.reason if decision.action == "block" else "" + + +def evaluate_route_policy( + config: Config, + route: Route | None, + *, + host: str, + request_path: str, + method: str, + headers: Mapping[str, str], + env: Mapping[str, str], +) -> RoutePolicyResult: + """Evaluate authorization stripping, matching, injection, and logging.""" + strip_authorization = route is None or not route.preserve_auth + effective_headers = { + name.lower(): value + for name, value in headers.items() + if not (strip_authorization and name.lower() == "authorization") + } + decision = decide( + config.routes, + host, + request_path, + env, + request_method=method, + request_headers=effective_headers, + deny_reason=config.deny_reason, + ) + return RoutePolicyResult( + block_reason=decision.reason if decision.action == "block" else "", + strip_authorization=strip_authorization, + inject_authorization=decision.inject_authorization, + log_request=config.log >= LOG_FULL, + ) + + +__all__ = [ + "GIT_PUSH_BLOCK_REASON", + "RoutePolicyResult", + "evaluate_route_policy", + "git_block_reason", +] diff --git a/tests/unit/test_egress_request_pipeline.py b/tests/unit/test_egress_request_pipeline.py new file mode 100644 index 00000000..291101de --- /dev/null +++ b/tests/unit/test_egress_request_pipeline.py @@ -0,0 +1,89 @@ +"""Unit tests for framework-neutral egress request policy stages.""" + +from __future__ import annotations + +import unittest + +from bot_bottle.gateway.egress.request_pipeline import ( + GIT_PUSH_BLOCK_REASON, + evaluate_route_policy, + git_block_reason, +) +from bot_bottle.gateway.egress.types import Config, LOG_FULL, Route + + +class TestGitPolicy(unittest.TestCase): + def test_push_is_always_blocked(self) -> None: + reason = git_block_reason( + (), "git.example.com", "/repo.git/git-receive-pack", "", + ) + self.assertEqual(GIT_PUSH_BLOCK_REASON, reason) + + def test_fetch_requires_route_opt_in(self) -> None: + path = "/repo.git/git-upload-pack" + blocked = git_block_reason((), "git.example.com", path, "") + allowed = git_block_reason( + (Route(host="git.example.com", git_fetch=True),), + "git.example.com", + path, + "", + ) + self.assertTrue(blocked) + self.assertEqual("", allowed) + + def test_non_git_request_is_not_decided_here(self) -> None: + self.assertEqual( + "", + git_block_reason((), "api.example.com", "/v1/messages", ""), + ) + + +class TestRoutePolicy(unittest.TestCase): + def test_strips_agent_auth_and_injects_gateway_auth(self) -> None: + route = Route( + host="api.example.com", + auth_scheme="Bearer", + token_env="API_TOKEN", + ) + result = evaluate_route_policy( + Config(routes=(route,)), + route, + host="api.example.com", + request_path="/v1/messages", + method="POST", + headers={"Authorization": "agent-secret"}, + env={"API_TOKEN": "gateway-secret"}, + ) + self.assertTrue(result.strip_authorization) + self.assertEqual("Bearer gateway-secret", result.inject_authorization) + self.assertFalse(result.block_reason) + + def test_preserved_auth_participates_in_matching(self) -> None: + route = Route(host="registry.example.com", preserve_auth=True) + result = evaluate_route_policy( + Config(routes=(route,), log=LOG_FULL), + route, + host="registry.example.com", + request_path="/v2/", + method="GET", + headers={"Authorization": "Bearer agent-token"}, + env={}, + ) + self.assertFalse(result.strip_authorization) + self.assertTrue(result.log_request) + + def test_missing_route_fails_closed(self) -> None: + result = evaluate_route_policy( + Config(routes=(), deny_reason="not allowed"), + None, + host="blocked.example.com", + request_path="/", + method="GET", + headers={}, + env={}, + ) + self.assertEqual("not allowed", result.block_reason) + + +if __name__ == "__main__": + unittest.main()