didericis
f26579c552
All three backends (docker, firecracker, macos-container) now launch through the consolidated orchestrator, and every production gateway sets BOT_BOTTLE_ORCHESTRATOR_URL — so the legacy single-tenant (`resolver is None`) branches in the shared gateway's data plane were unreachable dead code, a second security-relevant path to keep correct in parallel with the live one. Make the orchestrator resolver mandatory and delete the single-tenant paths from the three data-plane modules. egress_addon.py: drop the static routes file entirely — EGRESS_ROUTES, _reload, the SIGHUP handler, self.config, and the SUPERVISE_BOTTLE_SLUG env slug. The per-request /resolve is the only policy source; __init__ fail-closes if BOT_BOTTLE_ORCHESTRATOR_URL is unset. Introspection (`_egress.local/allowlist`) now reports the calling bottle's *resolved* routes. The block/redact log gates and _req_ctx redaction now read the per-flow config/env from the request-time stash, so they use each bottle's log level and token set (they silently used the empty static config before). Nothing sends `docker kill --signal HUP` to the gateway in the consolidated model (the egress applicators fail closed), so removing the SIGHUP reload is safe. git_http_backend.py: resolver mandatory; no flat-root fallback. main() refuses to start without an orchestrator URL; a request whose source resolves to no bottle 404s. supervise_server.py: resolver mandatory; every proposal is attributed to the source-IP-resolved bottle. Remove handle_list_egress_routes (the proxy-fetch introspection that only worked when the proxy carried one bottle's identity) — list-egress-routes is answered from the resolved policy. main() refuses to start without an orchestrator URL. Tests: a host-side fake resolver serves each test's Config through the real parse path (a small YAML-subset emitter round-trips route_to_yaml_dict); the response/websocket hooks stash it as request() would. Deletes the tests for the removed static-config, SIGHUP-reload, and single-tenant-passthrough paths; adds fail-closed-without-orchestrator coverage. Follow-up: gateway_init still forwards SIGHUP to the egress child (now dormant — no one sends it); the README still describes the docker backend's per-bottle topology. Both are outside the data-plane teardown. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UoEZHDjv84ChoZbozQERhJ
62 lines
2.3 KiB
Python
62 lines
2.3 KiB
Python
"""Unit: resolve_sandbox_root — source-IP-keyed git-gate repo namespace (PRD 0070).
|
|
|
|
The consolidated git-http backend serves every bottle from one process and
|
|
selects each request's repo root by the calling bottle's source IP. This is
|
|
the fail-closed selection logic, tested without a live server.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from bot_bottle.git_http_backend import resolve_sandbox_root
|
|
from bot_bottle.policy_resolver import PolicyResolveError
|
|
|
|
_BASE = Path("/git")
|
|
|
|
|
|
class _FakeResolver:
|
|
def __init__(self, bottle_id: str | None = None, raises: bool = False) -> None:
|
|
self._bottle_id = bottle_id
|
|
self._raises = raises
|
|
self.calls: list[tuple[str, str]] = []
|
|
|
|
def resolve_bottle_id(self, source_ip: str, identity_token: str = "") -> str | None:
|
|
self.calls.append((source_ip, identity_token))
|
|
if self._raises:
|
|
raise PolicyResolveError("orchestrator down")
|
|
return self._bottle_id
|
|
|
|
|
|
class TestResolveRepoRoot(unittest.TestCase):
|
|
def test_attributed_bottle_gets_namespaced_root(self) -> None:
|
|
root = resolve_sandbox_root(_FakeResolver(bottle_id="ab12cd34"), _BASE, "10.243.0.1")
|
|
self.assertEqual(Path("/git/ab12cd34"), root)
|
|
|
|
def test_unattributed_denies(self) -> None:
|
|
self.assertIsNone(resolve_sandbox_root(_FakeResolver(bottle_id=None), _BASE, "10.243.0.9"))
|
|
|
|
def test_empty_bottle_id_denies(self) -> None:
|
|
self.assertIsNone(resolve_sandbox_root(_FakeResolver(bottle_id=""), _BASE, "10.243.0.1"))
|
|
|
|
def test_resolver_error_denies(self) -> None:
|
|
# Orchestrator unreachable/errored must never serve repos.
|
|
self.assertIsNone(resolve_sandbox_root(_FakeResolver(raises=True), _BASE, "10.243.0.1"))
|
|
|
|
def test_namespace_escape_denies(self) -> None:
|
|
# A bottle_id that would climb out of the root is rejected (defense in
|
|
# depth — registry ids are token_hex, but never trust the namespace).
|
|
self.assertIsNone(
|
|
resolve_sandbox_root(_FakeResolver(bottle_id="../etc"), _BASE, "10.243.0.1")
|
|
)
|
|
|
|
def test_forwards_source_ip_and_token(self) -> None:
|
|
r = _FakeResolver(bottle_id="b1")
|
|
resolve_sandbox_root(r, _BASE, "10.243.0.1", "tok")
|
|
self.assertEqual(("10.243.0.1", "tok"), r.calls[0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|