From ea75fab3b421c492d5d2822980d122c48851fefd Mon Sep 17 00:00:00 2001 From: didericis Date: Tue, 14 Jul 2026 02:35:13 -0400 Subject: [PATCH] fix(orchestrator): supersede a prior active bottle at the same source IP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit register() used INSERT OR REPLACE keyed on bottle_id, so re-registering an existing source_ip with a new bottle_id left TWO active rows for that IP. by_source_ip fail-closes on ambiguity (>1 active -> None), so the reused IP was then bricked to a 403 on egress resolve — even for hosts its own policy allowed. Happy path (fresh IPs, teardown on stop) never hit it, but IP reuse, crash recovery, or a persisted registry DB across an orchestrator restart (now that ensure_running always recreates the container) all could. register() now deletes any other active row at the same source_ip before insert; the fail-closed ambiguity guard stays as defense in depth. Adds a unit test for the supersede, and a docker integration test (test_multitenant_isolation) that drives two bottles through one shared gateway and asserts each gets only its own injected token and its own allowlist — the core PRD 0070 multi-tenancy invariant, previously only verified by hand. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck --- bot_bottle/orchestrator/registry.py | 15 +- .../integration/test_multitenant_isolation.py | 154 ++++++++++++++++++ tests/unit/test_orchestrator_registry.py | 39 ++++- 3 files changed, 204 insertions(+), 4 deletions(-) create mode 100644 tests/integration/test_multitenant_isolation.py diff --git a/bot_bottle/orchestrator/registry.py b/bot_bottle/orchestrator/registry.py index fc2e770..dc200a3 100644 --- a/bot_bottle/orchestrator/registry.py +++ b/bot_bottle/orchestrator/registry.py @@ -149,7 +149,15 @@ class RegistryStore(DbStore): policy: str = "", ) -> BottleRecord: """Register (or replace) a bottle. Mints a bottle_id / identity token - when not supplied. Live — this is the control-plane reload path.""" + when not supplied. Live — this is the control-plane reload path. + + Supersedes any *other* active bottle at the same source IP first: + `by_source_ip` fail-closes on ambiguity (>1 active row → None), so a + leftover row at a reused IP — from an untorn-down bottle, crash + recovery, or a persisted DB — would otherwise *brick* the new bottle's + attribution. The new registration is authoritative for its IP; the + stale rows go. INSERT OR REPLACE handles a same-`bottle_id` reload in + place (its own row is exempt from the sweep).""" rec = BottleRecord( bottle_id=bottle_id or secrets.token_hex(8), source_ip=source_ip, @@ -160,6 +168,11 @@ class RegistryStore(DbStore): policy=policy, ) with self._connect() as conn: + conn.execute( + "DELETE FROM orchestrator_bottles " + "WHERE source_ip = ? AND state = 'active' AND bottle_id != ?", + (rec.source_ip, rec.bottle_id), + ) conn.execute( "INSERT OR REPLACE INTO orchestrator_bottles " "(bottle_id, source_ip, identity_token, state, created_at, metadata, policy) " diff --git a/tests/integration/test_multitenant_isolation.py b/tests/integration/test_multitenant_isolation.py new file mode 100644 index 0000000..c2d8e47 --- /dev/null +++ b/tests/integration/test_multitenant_isolation.py @@ -0,0 +1,154 @@ +"""Integration: two bottles share one gateway; source-IP attribution keeps +their egress tokens *and* allowlists isolated (PRD 0070 multi-tenancy). + +The whole premise of the consolidation is one shared gateway for every +bottle, isolated only by the unspoofable source IP. A single-bottle test +can't catch cross-bottle token bleed or an allowlist leak — this brings up +the real orchestrator + gateway and drives two bottles through the one +gateway to prove each gets exactly its own token and its own routes. + +Gated on a reachable Docker daemon and builds the bundle image, so it runs +with the other docker integration tests, not in unit CI. Uses the real +fixed gateway/orchestrator names (that's what it's testing), so it can't run +concurrently with a live per-host gateway; it points the orchestrator at a +throwaway BOT_BOTTLE_ROOT for a clean registry and tears everything down. +""" + +from __future__ import annotations + +import os +import subprocess +import tempfile +import unittest +from pathlib import Path + +from bot_bottle.backend.docker.consolidated_launch import ( + _network_cidr, + _network_container_ips, +) +from bot_bottle.backend.docker.egress import EGRESS_PORT +from bot_bottle.backend.docker.gateway_net import next_free_ip +from bot_bottle.orchestrator.client import OrchestratorClient +from bot_bottle.orchestrator.gateway import GATEWAY_IMAGE, GATEWAY_NAME, GATEWAY_NETWORK +from bot_bottle.orchestrator.lifecycle import OrchestratorService +from tests._docker import skip_unless_docker + +# One upstream reachable under two names; reflects the Authorization header so +# the probe can see exactly what the gateway injected (or didn't). +_ECHO_NAME = "bot-bottle-mtitest-echo" +_ECHO_ALIASES = ("echo-shared", "echo-bonly") +_ECHO_SRC = ( + "from http.server import BaseHTTPRequestHandler, HTTPServer\n" + "class H(BaseHTTPRequestHandler):\n" + " def do_GET(s):\n" + " a=s.headers.get('Authorization','NONE'); s.send_response(200); s.end_headers()\n" + " s.wfile.write(('AUTH='+a).encode())\n" + " def log_message(s,*a): pass\n" + "HTTPServer(('0.0.0.0',80),H).serve_forever()\n" +) + +# A: echo-shared only, authed. B: echo-shared authed + echo-bonly open. +_POLICY_A = ( + 'routes:\n' + ' - host: echo-shared\n' + ' auth_scheme: "Bearer"\n' + ' token_env: "EGRESS_TOKEN_0"\n' +) +_POLICY_B = _POLICY_A + ' - host: echo-bonly\n' +_TOKEN_A = "sk-AAA-alice" +_TOKEN_B = "sk-BBB-bob" + +# Client probe: open http:/// through the gateway proxy from a container +# pinned to the bottle's source IP, printing " ". Uses only the +# bundle image's python3 — no dependency on the agent image. +_PROBE_SRC = ( + "import sys,urllib.request,urllib.error\n" + "op=urllib.request.build_opener(urllib.request.ProxyHandler({'http':sys.argv[1]}))\n" + "try:\n" + " r=op.open('http://'+sys.argv[2]+'/',timeout=8)\n" + " sys.stdout.write(str(r.status)+' '+r.read().decode())\n" + "except urllib.error.HTTPError as e:\n" + " sys.stdout.write(str(e.code)+' '+e.read().decode())\n" +) + + +@skip_unless_docker() +class TestMultitenantIsolation(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + # Throwaway root → a clean registry DB, independent of the host's. + self.svc = OrchestratorService(host_root=Path(self._tmp.name)) + self.addCleanup(self._teardown_docker) + # ensure_running builds the bundle image (slow on a cold cache) and + # brings up the shared network + gateway + orchestrator. + url = self.svc.ensure_running(startup_timeout=300) + self.client = OrchestratorClient(url) + self.gw_ip = self._container_ip(GATEWAY_NAME) + self._run_echo() + + def _teardown_docker(self) -> None: + subprocess.run(["docker", "rm", "--force", _ECHO_NAME], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False) + self.svc.stop() + subprocess.run(["docker", "network", "rm", GATEWAY_NETWORK], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False) + # The orchestrator container wrote the registry DB as root into the + # throwaway root; chown it back so the (non-root) tempdir cleanup can + # remove it. + subprocess.run( + ["docker", "run", "--rm", "-v", f"{self._tmp.name}:/r", + "--entrypoint", "chown", GATEWAY_IMAGE, "-R", + f"{os.getuid()}:{os.getgid()}", "/r"], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False) + + @staticmethod + def _container_ip(name: str) -> str: + proc = subprocess.run( + ["docker", "inspect", "-f", + '{{(index .NetworkSettings.Networks "' + GATEWAY_NETWORK + '").IPAddress}}', name], + stdout=subprocess.PIPE, text=True, check=True, + ) + return proc.stdout.strip() + + def _run_echo(self) -> None: + argv = ["docker", "run", "--detach", "--name", _ECHO_NAME, + "--network", GATEWAY_NETWORK] + for alias in _ECHO_ALIASES: + argv += ["--network-alias", alias] + argv += ["--entrypoint", "python3", GATEWAY_IMAGE, "-c", _ECHO_SRC] + proc = subprocess.run(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True, check=False) + self.assertEqual(0, proc.returncode, f"echo upstream failed: {proc.stderr}") + + def _free_ip(self, extra_taken: list[str]) -> str: + taken = _network_container_ips(GATEWAY_NETWORK) + extra_taken + return next_free_ip(_network_cidr(GATEWAY_NETWORK), taken) + + def _probe(self, source_ip: str, host: str) -> str: + proc = subprocess.run( + ["docker", "run", "--rm", "--network", GATEWAY_NETWORK, "--ip", source_ip, + "--entrypoint", "python3", GATEWAY_IMAGE, "-c", _PROBE_SRC, + f"http://{self.gw_ip}:{EGRESS_PORT}", host], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, timeout=90, + ) + return proc.stdout.strip() + + def test_two_bottles_share_gateway_with_isolated_tokens_and_allowlists(self) -> None: + ip_a = self._free_ip([]) + ip_b = self._free_ip([ip_a]) + self.client.register_bottle(ip_a, policy=_POLICY_A, tokens={"EGRESS_TOKEN_0": _TOKEN_A}) + self.client.register_bottle(ip_b, policy=_POLICY_B, tokens={"EGRESS_TOKEN_0": _TOKEN_B}) + + # Each bottle gets its OWN token injected on the shared route — no bleed. + self.assertEqual(f"200 AUTH=Bearer {_TOKEN_A}", self._probe(ip_a, "echo-shared")) + self.assertEqual(f"200 AUTH=Bearer {_TOKEN_B}", self._probe(ip_b, "echo-shared")) + + # Allowlist is per-bottle: echo-bonly is only in B's policy. + self.assertTrue(self._probe(ip_a, "echo-bonly").startswith("403"), # fail-closed for A + "A reached a host outside its allowlist") + self.assertEqual("200 AUTH=NONE", self._probe(ip_b, "echo-bonly")) # allowed, unauthed for B + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_orchestrator_registry.py b/tests/unit/test_orchestrator_registry.py index 16f9dc2..58751ab 100644 --- a/tests/unit/test_orchestrator_registry.py +++ b/tests/unit/test_orchestrator_registry.py @@ -2,6 +2,7 @@ from __future__ import annotations +import sqlite3 import tempfile import unittest from pathlib import Path @@ -23,6 +24,20 @@ class TestRegistryStore(unittest.TestCase): def tearDown(self) -> None: self._tmp.cleanup() + def _force_active_row(self, source_ip: str, bottle_id: str) -> None: + """Insert a raw active row, bypassing `register`'s same-IP supersede — + the only way to manufacture the ambiguity the fail-closed guard exists + for (crash-orphaned / externally-injected duplicate).""" + conn = sqlite3.connect(self.db) + conn.execute( + "INSERT INTO orchestrator_bottles " + "(bottle_id, source_ip, identity_token, state, created_at, metadata, policy) " + "VALUES (?, ?, ?, 'active', 0.0, '', '')", + (bottle_id, source_ip, "tok-" + bottle_id), + ) + conn.commit() + conn.close() + def test_register_mints_id_and_token(self) -> None: rec = self.store.register("10.243.0.1") self.assertTrue(rec.bottle_id) @@ -76,11 +91,29 @@ class TestRegistryStore(unittest.TestCase): def test_attribute_ambiguous_ip_denied(self) -> None: # Two active bottles on one source IP is a misconfiguration — deny - # rather than guess (fail-closed), even with a valid token. + # rather than guess (fail-closed), even with a valid token. (Forced + # directly: `register` now supersedes, so this can only arise from an + # orphaned/injected row.) a = self.store.register("10.243.0.1", bottle_id="a") - self.store.register("10.243.0.1", bottle_id="b") + self._force_active_row("10.243.0.1", "b") self.assertIsNone(self.store.attribute("10.243.0.1", a.identity_token)) + def test_reregister_same_source_ip_supersedes(self) -> None: + # Re-registering a source IP (reused IP / crash recovery / persisted + # DB) must supersede the prior active bottle, not add a second active + # row that would fail-closed the whole IP to 403. + a = self.store.register("10.243.0.1", bottle_id="a", policy="A") + b = self.store.register("10.243.0.1", bottle_id="b", policy="B") + # exactly one active row at that IP, and it's the new bottle + got = self.store.by_source_ip("10.243.0.1") + assert got is not None + self.assertEqual("b", got.bottle_id) + self.assertEqual("B", got.policy) + # the superseded bottle is gone and its token no longer attributes + self.assertIsNone(self.store.get("a")) + self.assertIsNone(self.store.attribute("10.243.0.1", a.identity_token)) + self.assertIsNotNone(self.store.attribute("10.243.0.1", b.identity_token)) + def test_state_persists_across_reopen(self) -> None: rec = self.store.register("10.243.0.1") reopened = RegistryStore(self.db) @@ -130,7 +163,7 @@ class TestRegistryStore(unittest.TestCase): def test_by_source_ip_unknown_or_ambiguous_denied(self) -> None: self.assertIsNone(self.store.by_source_ip("10.243.9.9")) # unknown self.store.register("10.243.0.1", bottle_id="a") - self.store.register("10.243.0.1", bottle_id="b") + self._force_active_row("10.243.0.1", "b") # orphaned duplicate self.assertIsNone(self.store.by_source_ip("10.243.0.1")) # ambiguous