diff --git a/bot_bottle/egress/service.py b/bot_bottle/egress/service.py index 67670ffa..ffe2da49 100644 --- a/bot_bottle/egress/service.py +++ b/bot_bottle/egress/service.py @@ -130,7 +130,12 @@ def egress_forge_routes( (`/api/v1`): the proxy injects the Gitea `token` scheme using the value of the host env var named by the account's `token_secret`, resolved at launch from the host environment — the token never enters the bottle. Aliases that - canonicalize to the same host share a single route (deduplicated).""" + canonicalize to the same host share a single route (deduplicated). + + Composition (`resolve_forge_associations`) has already rejected referenced + aliases that share a host but disagree on origin/auth/token_secret, so this + host-dedup only ever collapses genuinely identical credentials — it never + silently discards a distinct one.""" out: list[EgressRoute] = [] seen_hosts: set[str] = set() for assoc in associations: diff --git a/bot_bottle/manifest/forge.py b/bot_bottle/manifest/forge.py index ae6fca20..554c435c 100644 --- a/bot_bottle/manifest/forge.py +++ b/bot_bottle/manifest/forge.py @@ -233,7 +233,7 @@ def resolve_forge_associations( ) by_alias.setdefault(alias, []).append(getattr(entry, "Name", "")) - return tuple( + associations = tuple( ResolvedForgeAssociation( account=forge_accounts[alias], repo_names=tuple(sorted(set(names))), @@ -241,6 +241,33 @@ def resolve_forge_associations( for alias, names in sorted(by_alias.items()) ) + # The egress proxy routes by host, so two referenced aliases that resolve + # to the same host must agree on the credential and target — otherwise the + # generated plan would authenticate every call to that host as whichever + # alias happened to win, silently acting as the wrong forge account. Fail + # closed here (before the bottle is created) rather than dropping a + # credential at route-synthesis time. + by_host: dict[str, ManifestForgeAccount] = {} + for assoc in associations: + acct = assoc.account + prev = by_host.get(acct.host) + if prev is None: + by_host[acct.host] = acct + continue + if (prev.origin, prev.api_prefix, prev.auth_type, prev.token_secret) != ( + acct.origin, acct.api_prefix, acct.auth_type, acct.token_secret + ): + raise ManifestError( + f"agent '{agent_name}' forge-accounts '{prev.alias}' and " + f"'{acct.alias}' both resolve to host '{acct.host}' but differ " + f"in origin/auth/token_secret. The egress proxy routes by host, " + f"so the intended credential would be ambiguous — every request " + f"to '{acct.host}' would authenticate as one account. Give the " + f"aliases distinct hosts, or make their url/auth identical." + ) + + return associations + def render_forge_guidance( associations: tuple[ResolvedForgeAssociation, ...], diff --git a/tests/unit/test_forge_identity.py b/tests/unit/test_forge_identity.py index 3edb4a83..d2474b84 100644 --- a/tests/unit/test_forge_identity.py +++ b/tests/unit/test_forge_identity.py @@ -253,6 +253,45 @@ class TestForgeAssociations(unittest.TestCase): idx = _index(repo_forge="didericis-gitea") # no forge-accounts at all self.assertIn("(none)", _error(idx.load_for_agent, "claude", ())) + def _two_alias_index(self, t1: str, t2: str) -> ManifestIndex: + """Two aliases on the SAME host, each referenced by a distinct repo.""" + def acct(token: str) -> dict[str, object]: + return { + "url": "https://same.example/api/v1", + "auth": {"type": "token", "token_secret": token}, + } + return ManifestIndex.from_json_obj({ + "bottles": {"bb": {"git-gate": {"repos": { + "r1": {"url": "ssh://git@h/x/r1.git", + "key": {"provider": "static", "path": "/k"}, + "forge": "g1"}, + "r2": {"url": "ssh://git@h/x/r2.git", + "key": {"provider": "static", "path": "/k"}, + "forge": "g2"}, + }}}}, + "agents": {"claude": {"bottle": "bb", "prompt": "", + "forge-accounts": {"g1": acct(t1), + "g2": acct(t2)}}}, + }) + + def test_conflicting_aliases_same_host_fail_closed(self) -> None: + # Two aliases on the same host with DIFFERENT tokens is ambiguous — + # the proxy routes by host, so it must fail before launch rather than + # silently authenticate every call as one account. + idx = self._two_alias_index("FIRST_TOKEN", "SECOND_TOKEN") + msg = _error(idx.load_for_agent, "claude", ()) + self.assertIn("both resolve to host 'same.example'", msg) + self.assertIn("ambiguous", msg) + + def test_matching_aliases_same_host_ok(self) -> None: + # Identical url/auth under two aliases is unambiguous: one route. + idx = self._two_alias_index("SAME_TOKEN", "SAME_TOKEN") + m = idx.load_for_agent("claude", ()) + self.assertEqual(2, len(m.forge_associations)) # both referenced + routes = egress_forge_routes(m.forge_associations) + self.assertEqual(1, len(routes)) # deduped to a single proxy route + self.assertEqual("SAME_TOKEN", routes[0].token_ref) + # --------------------------------------------------------------------------- # synthesized egress route (proxy-held credential)