didericis
08aef30d87
First slice of git-gate consolidation: make the smart-HTTP git backend serve every bottle from one process, selecting each request's repo root by the calling bottle's source IP — the same attribution invariant + resolver the multi-tenant egress addon uses. `git daemon` can't source-IP-route per connection, so the consolidated gateway serves git-gate over this HTTP backend (the transport firecracker/macOS already use); wiring the docker path onto it lands with the launch-integration slice. - policy_resolver: factor out `_post_resolve`; add `resolve_bottle_id` (source IP -> bottle id, same fail-closed 403->None contract as `resolve`) — the git-gate has no policy blob to parse, the bottle *is* the namespace. - git_http_backend: `resolve_repo_root(resolver, base, source_ip, token)` — single-tenant passthrough when no resolver; else `<base>/<bottle_id>`, fail-closed on unattributed / resolver error / namespace escape. `BOT_BOTTLE_ORCHESTRATOR_URL` (same env as egress) flips the shared gateway multi-tenant; the identity header is read for attribution and never forwarded to the CGI. Per-repo creds + hooks scope by repo dir, so isolating the root per bottle isolates its creds too. Single-tenant path unchanged (existing real-git-push tests green). New unit coverage for the resolver + repo-root selection matrix. Full unit suite green (1690 tests; the 13 test_sidecar_init /bin/sleep errors are the pre-existing NixOS-local noise). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
66 lines
2.5 KiB
Python
66 lines
2.5 KiB
Python
"""Unit: resolve_repo_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_repo_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_single_tenant_passthrough(self) -> None:
|
|
# No resolver → the flat base root, unchanged (legacy per-bottle mode).
|
|
self.assertEqual(_BASE, resolve_repo_root(None, _BASE, "10.243.0.1"))
|
|
|
|
def test_attributed_bottle_gets_namespaced_root(self) -> None:
|
|
root = resolve_repo_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_repo_root(_FakeResolver(bottle_id=None), _BASE, "10.243.0.9"))
|
|
|
|
def test_empty_bottle_id_denies(self) -> None:
|
|
self.assertIsNone(resolve_repo_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_repo_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_repo_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_repo_root(r, _BASE, "10.243.0.1", "tok")
|
|
self.assertEqual(("10.243.0.1", "tok"), r.calls[0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|