Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 8827c64a83 |
+52
-29
@@ -32,7 +32,7 @@ from egress_addon_core import ( # type: ignore[import-not-found] # pylint: dis
|
|||||||
is_git_push_request,
|
is_git_push_request,
|
||||||
load_config,
|
load_config,
|
||||||
match_route,
|
match_route,
|
||||||
resolve_client_config,
|
resolve_client_context,
|
||||||
outbound_scan_headers,
|
outbound_scan_headers,
|
||||||
route_to_yaml_dict,
|
route_to_yaml_dict,
|
||||||
scan_inbound,
|
scan_inbound,
|
||||||
@@ -99,17 +99,29 @@ class EgressAddon:
|
|||||||
# Absent → single-tenant (static routes file); behaviour unchanged.
|
# Absent → single-tenant (static routes file); behaviour unchanged.
|
||||||
orch_url = os.environ.get(ORCHESTRATOR_URL_ENV, "").strip()
|
orch_url = os.environ.get(ORCHESTRATOR_URL_ENV, "").strip()
|
||||||
self._resolver = PolicyResolver(orch_url) if orch_url else None
|
self._resolver = PolicyResolver(orch_url) if orch_url else None
|
||||||
# Tokens the operator has approved this session (PRD 0062). In-memory
|
# Tokens the operator has approved this session (PRD 0062), keyed by
|
||||||
# only — a restart re-prompts. Mutated only from the asyncio loop that
|
# bottle so the shared gateway keeps each bottle's safelist separate —
|
||||||
# runs the addon hooks, so no lock is needed.
|
# a global set would let bottle A's approved secret pass bottle B's DLP
|
||||||
self.safe_tokens: set[str] = set()
|
# scan. In-memory only (a restart re-prompts); mutated only from the
|
||||||
|
# asyncio loop that runs the addon hooks, so no lock is needed.
|
||||||
|
self._safe_tokens: dict[str, set[str]] = {}
|
||||||
self._supervise_slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "").strip()
|
self._supervise_slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "").strip()
|
||||||
self._token_allow_timeout = _token_allow_timeout_from_env(os.environ)
|
self._token_allow_timeout = _token_allow_timeout_from_env(os.environ)
|
||||||
self._reload(initial=True)
|
self._reload(initial=True)
|
||||||
self._install_sighup()
|
self._install_sighup()
|
||||||
|
|
||||||
def _supervise_available(self) -> bool:
|
@staticmethod
|
||||||
return bool(self._supervise_slug)
|
def _supervise_available(slug: str) -> bool:
|
||||||
|
"""Supervise is reachable for this request iff we resolved a bottle to
|
||||||
|
attribute its proposals to (single-tenant env slug, or a source-IP
|
||||||
|
-attributed bottle id). Empty → fail closed (no queue to write to)."""
|
||||||
|
return bool(slug)
|
||||||
|
|
||||||
|
def _safe_tokens_for(self, slug: str) -> set[str]:
|
||||||
|
"""This bottle's operator-approved DLP safelist (PRD 0062), created on
|
||||||
|
first use. Keyed by bottle so the shared gateway never leaks one
|
||||||
|
bottle's approved token into another's scan."""
|
||||||
|
return self._safe_tokens.setdefault(slug, set())
|
||||||
|
|
||||||
def _reload(self, *, initial: bool = False) -> None:
|
def _reload(self, *, initial: bool = False) -> None:
|
||||||
try:
|
try:
|
||||||
@@ -218,19 +230,21 @@ class EgressAddon:
|
|||||||
+ "\n"
|
+ "\n"
|
||||||
)
|
)
|
||||||
|
|
||||||
def _active_config(self, flow: http.HTTPFlow) -> Config:
|
def _resolve_flow(self, flow: http.HTTPFlow) -> "tuple[Config, str]":
|
||||||
"""The Config to apply to this request. Single-tenant → the static
|
"""The `(Config, supervise slug)` to apply to this request. Single-tenant
|
||||||
`self.config`. Consolidated → the calling bottle's Config, resolved
|
→ the static `self.config` and the env slug. Consolidated → the calling
|
||||||
by source IP (fail-closed to deny-all if unattributed). The identity
|
bottle's Config and its bottle id, resolved by source IP in one
|
||||||
token, if the agent injected one, is read then stripped so it never
|
round-trip (fail-closed to deny-all + empty slug if unattributed). The
|
||||||
leaks upstream."""
|
identity token, if the agent injected one, is read then stripped so it
|
||||||
|
never leaks upstream. The slug keys both the proposal queue and the
|
||||||
|
per-bottle safelist, so an approval only ever affects its own bottle."""
|
||||||
if self._resolver is None:
|
if self._resolver is None:
|
||||||
return self.config
|
return self.config, self._supervise_slug
|
||||||
conn = flow.client_conn
|
conn = flow.client_conn
|
||||||
client_ip = conn.peername[0] if conn and conn.peername else ""
|
client_ip = conn.peername[0] if conn and conn.peername else ""
|
||||||
token = flow.request.headers.get(IDENTITY_HEADER, "")
|
token = flow.request.headers.get(IDENTITY_HEADER, "")
|
||||||
flow.request.headers.pop(IDENTITY_HEADER, None)
|
flow.request.headers.pop(IDENTITY_HEADER, None)
|
||||||
return resolve_client_config(self._resolver, client_ip, token)
|
return resolve_client_context(self._resolver, client_ip, token)
|
||||||
|
|
||||||
async def request(self, flow: http.HTTPFlow) -> None:
|
async def request(self, flow: http.HTTPFlow) -> None:
|
||||||
request_path, _, query = flow.request.path.partition("?")
|
request_path, _, query = flow.request.path.partition("?")
|
||||||
@@ -239,14 +253,14 @@ class EgressAddon:
|
|||||||
self._serve_introspection(flow, request_path)
|
self._serve_introspection(flow, request_path)
|
||||||
return
|
return
|
||||||
|
|
||||||
config = self._active_config(flow)
|
config, slug = self._resolve_flow(flow)
|
||||||
|
|
||||||
# DLP outbound scan BEFORE stripping auth — catches tokens the
|
# DLP outbound scan BEFORE stripping auth — catches tokens the
|
||||||
# agent tried to smuggle in any header, path, query param, or body.
|
# agent tried to smuggle in any header, path, query param, or body.
|
||||||
# Hostname is included to catch DNS-tunnelling exfiltration attempts.
|
# Hostname is included to catch DNS-tunnelling exfiltration attempts.
|
||||||
route = match_route(config.routes, flow.request.pretty_host)
|
route = match_route(config.routes, flow.request.pretty_host)
|
||||||
if route is not None:
|
if route is not None:
|
||||||
if not await self._handle_outbound_dlp(flow, route):
|
if not await self._handle_outbound_dlp(flow, route, slug):
|
||||||
return
|
return
|
||||||
# The redact policy may have rewritten the request line; recompute
|
# The redact policy may have rewritten the request line; recompute
|
||||||
# the path/query the git checks below rely on.
|
# the path/query the git checks below rely on.
|
||||||
@@ -310,6 +324,7 @@ class EgressAddon:
|
|||||||
self,
|
self,
|
||||||
flow: http.HTTPFlow,
|
flow: http.HTTPFlow,
|
||||||
route: Route,
|
route: Route,
|
||||||
|
slug: str,
|
||||||
) -> bool:
|
) -> bool:
|
||||||
"""Scan the outbound request and apply the route's on-match policy
|
"""Scan the outbound request and apply the route's on-match policy
|
||||||
(PRD 0062). Returns True if the request may be forwarded, False if a
|
(PRD 0062). Returns True if the request may be forwarded, False if a
|
||||||
@@ -331,7 +346,7 @@ class EgressAddon:
|
|||||||
)
|
)
|
||||||
result = scan_outbound(
|
result = scan_outbound(
|
||||||
route, scan_text, os.environ,
|
route, scan_text, os.environ,
|
||||||
safe_tokens=self.safe_tokens, crlf_text=crlf_text,
|
safe_tokens=self._safe_tokens_for(slug), crlf_text=crlf_text,
|
||||||
)
|
)
|
||||||
if result is None or result.severity != "block":
|
if result is None or result.severity != "block":
|
||||||
return True
|
return True
|
||||||
@@ -366,10 +381,10 @@ class EgressAddon:
|
|||||||
|
|
||||||
# supervise (default): hold the request for operator approval.
|
# supervise (default): hold the request for operator approval.
|
||||||
# Fall back to a hard 403 when supervise isn't wired for the bottle.
|
# Fall back to a hard 403 when supervise isn't wired for the bottle.
|
||||||
if not self._supervise_available():
|
if not self._supervise_available(slug):
|
||||||
self._block_dlp(flow, result)
|
self._block_dlp(flow, result)
|
||||||
return False
|
return False
|
||||||
approved = await self._supervise_token_block(flow, request_path, result)
|
approved = await self._supervise_token_block(flow, request_path, result, slug)
|
||||||
if not approved:
|
if not approved:
|
||||||
return False # _supervise_token_block wrote the 403 response
|
return False # _supervise_token_block wrote the 403 response
|
||||||
# loop: the approved value is now in safe_tokens; re-scan.
|
# loop: the approved value is now in safe_tokens; re-scan.
|
||||||
@@ -412,12 +427,16 @@ class EgressAddon:
|
|||||||
flow: http.HTTPFlow,
|
flow: http.HTTPFlow,
|
||||||
request_path: str,
|
request_path: str,
|
||||||
result: ScanResult,
|
result: ScanResult,
|
||||||
|
slug: str,
|
||||||
) -> bool:
|
) -> bool:
|
||||||
"""Route a token DLP block to the operator's supervisor queue and wait.
|
"""Route a token DLP block to the operator's supervisor queue and wait.
|
||||||
|
|
||||||
Returns True if the operator approved (the matched value is added to
|
`slug` attributes the proposal to the calling bottle (its own queue +
|
||||||
`self.safe_tokens` and the caller re-scans); False if the request must
|
safelist) — in the shared gateway this is what keeps one bottle's
|
||||||
be blocked (a 403 response has been written to `flow`)."""
|
approval from unblocking another's request. Returns True if the operator
|
||||||
|
approved (the matched value is added to that bottle's safelist and the
|
||||||
|
caller re-scans); False if the request must be blocked (a 403 response
|
||||||
|
has been written to `flow`)."""
|
||||||
host = flow.request.pretty_host
|
host = flow.request.pretty_host
|
||||||
payload = build_token_allow_payload(
|
payload = build_token_allow_payload(
|
||||||
redact_tokens(host, env=os.environ),
|
redact_tokens(host, env=os.environ),
|
||||||
@@ -426,7 +445,7 @@ class EgressAddon:
|
|||||||
result,
|
result,
|
||||||
)
|
)
|
||||||
proposal = _sv.Proposal.new(
|
proposal = _sv.Proposal.new(
|
||||||
bottle_slug=self._supervise_slug,
|
bottle_slug=slug,
|
||||||
tool=_sv.TOOL_EGRESS_TOKEN_ALLOW,
|
tool=_sv.TOOL_EGRESS_TOKEN_ALLOW,
|
||||||
proposed_file=payload,
|
proposed_file=payload,
|
||||||
justification=_TOKEN_ALLOW_JUSTIFICATION,
|
justification=_TOKEN_ALLOW_JUSTIFICATION,
|
||||||
@@ -449,13 +468,13 @@ class EgressAddon:
|
|||||||
**self._req_ctx(flow),
|
**self._req_ctx(flow),
|
||||||
}) + "\n")
|
}) + "\n")
|
||||||
|
|
||||||
response = await self._await_token_response(proposal.id)
|
response = await self._await_token_response(proposal.id, slug)
|
||||||
_sv.archive_proposal(self._supervise_slug, proposal.id)
|
_sv.archive_proposal(slug, proposal.id)
|
||||||
|
|
||||||
if response is not None and response.status in (
|
if response is not None and response.status in (
|
||||||
_sv.STATUS_APPROVED, _sv.STATUS_MODIFIED,
|
_sv.STATUS_APPROVED, _sv.STATUS_MODIFIED,
|
||||||
):
|
):
|
||||||
self.safe_tokens.add(result.matched)
|
self._safe_tokens_for(slug).add(result.matched)
|
||||||
if self.config.log >= LOG_BLOCKS:
|
if self.config.log >= LOG_BLOCKS:
|
||||||
sys.stderr.write(json.dumps({
|
sys.stderr.write(json.dumps({
|
||||||
"event": "egress_token_allowed",
|
"event": "egress_token_allowed",
|
||||||
@@ -478,6 +497,7 @@ class EgressAddon:
|
|||||||
async def _await_token_response(
|
async def _await_token_response(
|
||||||
self,
|
self,
|
||||||
proposal_id: str,
|
proposal_id: str,
|
||||||
|
slug: str,
|
||||||
) -> "_sv.Response | None":
|
) -> "_sv.Response | None":
|
||||||
"""Poll the DB for the operator's response without blocking the
|
"""Poll the DB for the operator's response without blocking the
|
||||||
proxy event loop. Returns the Response, or None on timeout."""
|
proxy event loop. Returns the Response, or None on timeout."""
|
||||||
@@ -485,7 +505,7 @@ class EgressAddon:
|
|||||||
deadline = loop.time() + self._token_allow_timeout
|
deadline = loop.time() + self._token_allow_timeout
|
||||||
while True:
|
while True:
|
||||||
try:
|
try:
|
||||||
return _sv.read_response(self._supervise_slug, proposal_id)
|
return _sv.read_response(slug, proposal_id)
|
||||||
except (OSError, ValueError, KeyError):
|
except (OSError, ValueError, KeyError):
|
||||||
# Not written yet, or a partial/malformed write — retry until
|
# Not written yet, or a partial/malformed write — retry until
|
||||||
# the deadline, then fail closed.
|
# the deadline, then fail closed.
|
||||||
@@ -539,6 +559,9 @@ class EgressAddon:
|
|||||||
"""
|
"""
|
||||||
if flow.websocket is None: # type: ignore[union-attr]
|
if flow.websocket is None: # type: ignore[union-attr]
|
||||||
return
|
return
|
||||||
|
# WebSocket DLP runs against the static config only (single-tenant); in
|
||||||
|
# the consolidated gateway self.config has no routes, so this is inert
|
||||||
|
# until websocket routing is made source-IP-aware (a separate slice).
|
||||||
route = match_route(self.config.routes, flow.request.pretty_host)
|
route = match_route(self.config.routes, flow.request.pretty_host)
|
||||||
if route is None:
|
if route is None:
|
||||||
return
|
return
|
||||||
@@ -549,7 +572,7 @@ class EgressAddon:
|
|||||||
# not an injection vector here — scan only for credential leakage.
|
# not an injection vector here — scan only for credential leakage.
|
||||||
result = scan_outbound(
|
result = scan_outbound(
|
||||||
route, content, os.environ,
|
route, content, os.environ,
|
||||||
safe_tokens=self.safe_tokens, crlf_text="",
|
safe_tokens=self._safe_tokens_for(self._supervise_slug), crlf_text="",
|
||||||
)
|
)
|
||||||
if result is not None and result.severity == "block":
|
if result is not None and result.severity == "block":
|
||||||
sys.stderr.write(f"egress DLP: {result.reason}\n")
|
sys.stderr.write(f"egress DLP: {result.reason}\n")
|
||||||
|
|||||||
@@ -421,6 +421,18 @@ class PolicyResolverLike(typing.Protocol):
|
|||||||
...
|
...
|
||||||
|
|
||||||
|
|
||||||
|
def _config_from_policy(policy: "str | None") -> "Config":
|
||||||
|
"""Parse a resolved policy blob into a Config, fail-closed: None / empty /
|
||||||
|
unparseable all become a deny-all Config (no routes → every request
|
||||||
|
blocked)."""
|
||||||
|
if not policy:
|
||||||
|
return Config(routes=()) # unattributed or empty → deny-all
|
||||||
|
try:
|
||||||
|
return load_config(policy)
|
||||||
|
except ValueError:
|
||||||
|
return Config(routes=()) # unparseable policy → deny
|
||||||
|
|
||||||
|
|
||||||
def resolve_client_config(
|
def resolve_client_config(
|
||||||
resolver: PolicyResolverLike, client_ip: str, identity_token: str = ""
|
resolver: PolicyResolverLike, client_ip: str, identity_token: str = ""
|
||||||
) -> "Config":
|
) -> "Config":
|
||||||
@@ -433,12 +445,33 @@ def resolve_client_config(
|
|||||||
policy = resolver.resolve(client_ip, identity_token)
|
policy = resolver.resolve(client_ip, identity_token)
|
||||||
except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught
|
except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught
|
||||||
return Config(routes=()) # orchestrator unreachable/errored → deny
|
return Config(routes=()) # orchestrator unreachable/errored → deny
|
||||||
if not policy:
|
return _config_from_policy(policy)
|
||||||
return Config(routes=()) # unattributed or empty → deny-all
|
|
||||||
|
|
||||||
|
class ContextResolverLike(typing.Protocol):
|
||||||
|
"""The bit of `policy_resolver.PolicyResolver` `resolve_client_context`
|
||||||
|
needs — one round-trip returning both policy and bottle id."""
|
||||||
|
|
||||||
|
def resolve_policy_and_bottle_id(
|
||||||
|
self, source_ip: str, identity_token: str = ...,
|
||||||
|
) -> "tuple[str | None, str | None]":
|
||||||
|
...
|
||||||
|
|
||||||
|
|
||||||
|
def resolve_client_context(
|
||||||
|
resolver: ContextResolverLike, client_ip: str, identity_token: str = "",
|
||||||
|
) -> "tuple[Config, str]":
|
||||||
|
"""The calling client's `(Config, bottle_id)` in one round-trip —
|
||||||
|
**fail-closed**. The Config follows `resolve_client_config`'s deny-all
|
||||||
|
rules; the bottle id is `""` whenever unattributed or the orchestrator
|
||||||
|
errored, which the caller treats as "supervise unavailable for this
|
||||||
|
bottle" (never another bottle's queue). One `/resolve` keys both the
|
||||||
|
per-request egress policy and the per-bottle supervise queue + safelist."""
|
||||||
try:
|
try:
|
||||||
return load_config(policy)
|
policy, bottle_id = resolver.resolve_policy_and_bottle_id(client_ip, identity_token)
|
||||||
except ValueError:
|
except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught
|
||||||
return Config(routes=()) # unparseable policy → deny
|
return Config(routes=()), "" # orchestrator unreachable/errored → deny
|
||||||
|
return _config_from_policy(policy), (bottle_id or "")
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
@@ -833,7 +866,9 @@ __all__ = [
|
|||||||
"is_git_fetch_request",
|
"is_git_fetch_request",
|
||||||
"load_config",
|
"load_config",
|
||||||
"resolve_client_config",
|
"resolve_client_config",
|
||||||
|
"resolve_client_context",
|
||||||
"PolicyResolverLike",
|
"PolicyResolverLike",
|
||||||
|
"ContextResolverLike",
|
||||||
"match_route",
|
"match_route",
|
||||||
"outbound_scan_headers",
|
"outbound_scan_headers",
|
||||||
"parse_config",
|
"parse_config",
|
||||||
|
|||||||
@@ -100,5 +100,24 @@ class PolicyResolver:
|
|||||||
bottle_id = payload.get("bottle_id")
|
bottle_id = payload.get("bottle_id")
|
||||||
return bottle_id if isinstance(bottle_id, str) and bottle_id else None
|
return bottle_id if isinstance(bottle_id, str) and bottle_id else None
|
||||||
|
|
||||||
|
def resolve_policy_and_bottle_id(
|
||||||
|
self, source_ip: str, identity_token: str = "",
|
||||||
|
) -> tuple[str | None, str | None]:
|
||||||
|
"""Both the policy blob and the bottle id in a single `/resolve` — so
|
||||||
|
a caller that needs each (the egress addon: policy for routing, bottle
|
||||||
|
id to key the calling bottle's supervise queue + safelist) makes one
|
||||||
|
round-trip, not two. Returns `(None, None)` when unattributed (a clean
|
||||||
|
`403`). Raises `PolicyResolveError` if the orchestrator can't be
|
||||||
|
reached, so the caller still fails closed."""
|
||||||
|
payload = self._post_resolve(source_ip, identity_token)
|
||||||
|
if payload is None:
|
||||||
|
return None, None
|
||||||
|
policy = payload.get("policy")
|
||||||
|
bottle_id = payload.get("bottle_id")
|
||||||
|
return (
|
||||||
|
policy if isinstance(policy, str) else "",
|
||||||
|
bottle_id if isinstance(bottle_id, str) and bottle_id else None,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
__all__ = ["PolicyResolver", "PolicyResolveError"]
|
__all__ = ["PolicyResolver", "PolicyResolveError"]
|
||||||
|
|||||||
@@ -46,7 +46,7 @@ def _addon() -> EgressAddon:
|
|||||||
"""Return a bare EgressAddon with LOG_FULL config and no routes file."""
|
"""Return a bare EgressAddon with LOG_FULL config and no routes file."""
|
||||||
a: EgressAddon = EgressAddon.__new__(EgressAddon)
|
a: EgressAddon = EgressAddon.__new__(EgressAddon)
|
||||||
a.config = Config(routes=(), log=LOG_FULL)
|
a.config = Config(routes=(), log=LOG_FULL)
|
||||||
a.safe_tokens = set()
|
a._safe_tokens = {}
|
||||||
a._supervise_slug = ""
|
a._supervise_slug = ""
|
||||||
a._token_allow_timeout = 300.0
|
a._token_allow_timeout = 300.0
|
||||||
return a
|
return a
|
||||||
|
|||||||
@@ -211,7 +211,7 @@ def _addon(config: Config) -> EgressAddon:
|
|||||||
"""Bare EgressAddon with a supplied config and no supervise wiring."""
|
"""Bare EgressAddon with a supplied config and no supervise wiring."""
|
||||||
a: EgressAddon = EgressAddon.__new__(EgressAddon)
|
a: EgressAddon = EgressAddon.__new__(EgressAddon)
|
||||||
a.config = config
|
a.config = config
|
||||||
a.safe_tokens = set()
|
a._safe_tokens = {}
|
||||||
a._supervise_slug = ""
|
a._supervise_slug = ""
|
||||||
a._token_allow_timeout = 300.0
|
a._token_allow_timeout = 300.0
|
||||||
a.routes_path = "/nonexistent/routes.yaml"
|
a.routes_path = "/nonexistent/routes.yaml"
|
||||||
@@ -222,6 +222,29 @@ def _run_request(addon: EgressAddon, flow: _Flow) -> None:
|
|||||||
asyncio.run(addon.request(flow)) # type: ignore[arg-type]
|
asyncio.run(addon.request(flow)) # type: ignore[arg-type]
|
||||||
|
|
||||||
|
|
||||||
|
def _with_client_ip(flow: _Flow, ip: str) -> _Flow:
|
||||||
|
"""Attach a mitmproxy-style client_conn so consolidated resolution can read
|
||||||
|
the source IP (`flow.client_conn.peername[0]`)."""
|
||||||
|
flow.client_conn = types.SimpleNamespace(peername=(ip, 54321)) # type: ignore[attr-defined]
|
||||||
|
return flow
|
||||||
|
|
||||||
|
|
||||||
|
class _CtxResolver:
|
||||||
|
"""Fake orchestrator resolver: maps source IP -> bottle id, and grants the
|
||||||
|
same allow-list to any attributed bottle (unattributed -> deny)."""
|
||||||
|
|
||||||
|
def __init__(self, ip_to_bottle: dict[str, str]) -> None:
|
||||||
|
self._map = ip_to_bottle
|
||||||
|
|
||||||
|
def resolve_policy_and_bottle_id(
|
||||||
|
self, source_ip: str, identity_token: str = "",
|
||||||
|
) -> tuple[str | None, str | None]:
|
||||||
|
del identity_token
|
||||||
|
bottle_id = self._map.get(source_ip)
|
||||||
|
policy = "routes:\n - host: api.example.com\n" if bottle_id else None
|
||||||
|
return policy, bottle_id
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# Introspection endpoint
|
# Introspection endpoint
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
@@ -418,7 +441,8 @@ class TestSuperviseBranch(unittest.TestCase):
|
|||||||
with patch.object(_ea_mod, "_sv", _fake_sv("approved")):
|
with patch.object(_ea_mod, "_sv", _fake_sv("approved")):
|
||||||
_run_request(addon, flow)
|
_run_request(addon, flow)
|
||||||
self.assertIsNone(flow.response) # forwarded after approval
|
self.assertIsNone(flow.response) # forwarded after approval
|
||||||
self.assertIn(_OPENAI_KEY, addon.safe_tokens)
|
# Approval lands in the calling bottle's safelist (keyed by slug).
|
||||||
|
self.assertIn(_OPENAI_KEY, addon._safe_tokens_for("test-bottle"))
|
||||||
|
|
||||||
def test_operator_rejection_blocks(self) -> None:
|
def test_operator_rejection_blocks(self) -> None:
|
||||||
addon = self._supervised_addon()
|
addon = self._supervised_addon()
|
||||||
@@ -735,5 +759,62 @@ class TestLogFullRequest(unittest.TestCase):
|
|||||||
self.assertTrue(any(e.get("event") == "egress_request" for e in logged))
|
self.assertTrue(any(e.get("event") == "egress_request" for e in logged))
|
||||||
|
|
||||||
|
|
||||||
|
class TestSuperviseMultiTenant(unittest.TestCase):
|
||||||
|
"""Consolidated gateway: supervise proposals + the DLP safelist are keyed
|
||||||
|
per bottle, resolved by source IP (PRD 0070)."""
|
||||||
|
|
||||||
|
def _consolidated_addon(self) -> EgressAddon:
|
||||||
|
# Static config is empty; the resolver supplies each bottle's config.
|
||||||
|
addon = _addon(Config(routes=()))
|
||||||
|
addon._resolver = cast(Any, _CtxResolver({"10.0.0.1": "bottle-a", "10.0.0.2": "bottle-b"}))
|
||||||
|
addon._token_allow_timeout = 0.05
|
||||||
|
return addon
|
||||||
|
|
||||||
|
def test_approval_is_scoped_to_the_calling_bottle(self) -> None:
|
||||||
|
addon = self._consolidated_addon()
|
||||||
|
# bottle-a (10.0.0.1) sends the token; the operator approves.
|
||||||
|
flow = _with_client_ip(
|
||||||
|
_Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")),
|
||||||
|
"10.0.0.1",
|
||||||
|
)
|
||||||
|
with patch.object(_ea_mod, "_sv", _fake_sv("approved")):
|
||||||
|
_run_request(addon, flow)
|
||||||
|
self.assertIsNone(flow.response) # forwarded after approval
|
||||||
|
# The approval lands ONLY in bottle-a's safelist — never bottle-b's.
|
||||||
|
# A global set here would be the cross-tenant leak this slice closes.
|
||||||
|
self.assertIn(_OPENAI_KEY, addon._safe_tokens_for("bottle-a"))
|
||||||
|
self.assertNotIn(_OPENAI_KEY, addon._safe_tokens_for("bottle-b"))
|
||||||
|
|
||||||
|
def test_proposal_is_attributed_to_the_source_ip_bottle(self) -> None:
|
||||||
|
addon = self._consolidated_addon()
|
||||||
|
seen: list[str] = []
|
||||||
|
fake = _fake_sv("approved")
|
||||||
|
|
||||||
|
def _capture(**kw: Any) -> Any:
|
||||||
|
seen.append(kw["bottle_slug"])
|
||||||
|
return types.SimpleNamespace(id="p")
|
||||||
|
|
||||||
|
fake.Proposal = types.SimpleNamespace(new=_capture)
|
||||||
|
flow = _with_client_ip(
|
||||||
|
_Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")),
|
||||||
|
"10.0.0.2",
|
||||||
|
)
|
||||||
|
with patch.object(_ea_mod, "_sv", fake):
|
||||||
|
_run_request(addon, flow)
|
||||||
|
self.assertEqual(["bottle-b"], seen) # proposal keyed by the resolved bottle
|
||||||
|
|
||||||
|
def test_unattributed_source_ip_cannot_supervise(self) -> None:
|
||||||
|
addon = self._consolidated_addon()
|
||||||
|
# 10.9.9.9 is not in the resolver map -> deny-all config, empty slug.
|
||||||
|
flow = _with_client_ip(
|
||||||
|
_Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")),
|
||||||
|
"10.9.9.9",
|
||||||
|
)
|
||||||
|
with patch.object(_ea_mod, "_sv", _fake_sv("approved")):
|
||||||
|
_run_request(addon, flow)
|
||||||
|
self.assertIsNotNone(flow.response) # blocked (no route, no supervise)
|
||||||
|
self.assertNotIn(_OPENAI_KEY, addon._safe_tokens_for(""))
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
@@ -1,10 +1,10 @@
|
|||||||
"""Unit: resolve_client_config — fail-closed per-client egress config (PRD 0070)."""
|
"""Unit: fail-closed per-client egress resolution — config + context (PRD 0070)."""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
from bot_bottle.egress_addon_core import resolve_client_config
|
from bot_bottle.egress_addon_core import resolve_client_config, resolve_client_context
|
||||||
from bot_bottle.policy_resolver import PolicyResolveError
|
from bot_bottle.policy_resolver import PolicyResolveError
|
||||||
|
|
||||||
|
|
||||||
@@ -49,5 +49,52 @@ class TestResolveClientConfig(unittest.TestCase):
|
|||||||
self.assertEqual(("10.243.0.1", "tok"), r.calls[0])
|
self.assertEqual(("10.243.0.1", "tok"), r.calls[0])
|
||||||
|
|
||||||
|
|
||||||
|
class _FakeContextResolver:
|
||||||
|
def __init__(
|
||||||
|
self, policy: str | None = None, bottle_id: str | None = None, raises: bool = False,
|
||||||
|
) -> None:
|
||||||
|
self._policy = policy
|
||||||
|
self._bottle_id = bottle_id
|
||||||
|
self._raises = raises
|
||||||
|
|
||||||
|
def resolve_policy_and_bottle_id(
|
||||||
|
self, source_ip: str, identity_token: str = "",
|
||||||
|
) -> tuple[str | None, str | None]:
|
||||||
|
if self._raises:
|
||||||
|
raise PolicyResolveError("orchestrator down")
|
||||||
|
return self._policy, self._bottle_id
|
||||||
|
|
||||||
|
|
||||||
|
class TestResolveClientContext(unittest.TestCase):
|
||||||
|
def test_returns_config_and_bottle_id(self) -> None:
|
||||||
|
cfg, slug = resolve_client_context(
|
||||||
|
_FakeContextResolver(policy="routes:\n - host: example.com\n", bottle_id="b1"),
|
||||||
|
"10.243.0.1",
|
||||||
|
)
|
||||||
|
self.assertEqual(("example.com",), tuple(r.host for r in cfg.routes))
|
||||||
|
self.assertEqual("b1", slug)
|
||||||
|
|
||||||
|
def test_unattributed_denies_and_empty_slug(self) -> None:
|
||||||
|
cfg, slug = resolve_client_context(
|
||||||
|
_FakeContextResolver(policy=None, bottle_id=None), "10.243.0.9",
|
||||||
|
)
|
||||||
|
self.assertEqual((), cfg.routes)
|
||||||
|
self.assertEqual("", slug) # no bottle → supervise unavailable
|
||||||
|
|
||||||
|
def test_resolver_error_denies_and_empty_slug(self) -> None:
|
||||||
|
cfg, slug = resolve_client_context(_FakeContextResolver(raises=True), "10.243.0.1")
|
||||||
|
self.assertEqual((), cfg.routes)
|
||||||
|
self.assertEqual("", slug)
|
||||||
|
|
||||||
|
def test_unparseable_policy_denies_but_keeps_slug(self) -> None:
|
||||||
|
# A bad policy denies egress, but the bottle is still attributed (its
|
||||||
|
# supervise queue is keyed by the id, independent of route parsing).
|
||||||
|
cfg, slug = resolve_client_context(
|
||||||
|
_FakeContextResolver(policy="routes: notalist\n", bottle_id="b2"), "10.243.0.1",
|
||||||
|
)
|
||||||
|
self.assertEqual((), cfg.routes)
|
||||||
|
self.assertEqual("b2", slug)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
@@ -88,6 +88,20 @@ class TestPolicyResolver(unittest.TestCase):
|
|||||||
with self.assertRaises(PolicyResolveError):
|
with self.assertRaises(PolicyResolveError):
|
||||||
self.r.resolve_bottle_id("10.243.0.1", "tok")
|
self.r.resolve_bottle_id("10.243.0.1", "tok")
|
||||||
|
|
||||||
|
def test_resolve_policy_and_bottle_id_one_call(self) -> None:
|
||||||
|
with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1", "policy": "P"})) as m:
|
||||||
|
self.assertEqual(("P", "b1"), self.r.resolve_policy_and_bottle_id("10.243.0.1", "t"))
|
||||||
|
self.assertEqual(1, m.call_count) # both from a single /resolve
|
||||||
|
|
||||||
|
def test_resolve_policy_and_bottle_id_403_is_none_none(self) -> None:
|
||||||
|
with patch(_URLOPEN, side_effect=_http_error(403)):
|
||||||
|
self.assertEqual((None, None), self.r.resolve_policy_and_bottle_id("10.243.0.9"))
|
||||||
|
|
||||||
|
def test_resolve_policy_and_bottle_id_error_raises(self) -> None:
|
||||||
|
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
|
||||||
|
with self.assertRaises(PolicyResolveError):
|
||||||
|
self.r.resolve_policy_and_bottle_id("10.243.0.1")
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
Reference in New Issue
Block a user