didericis/bot-bottle
1
0
Fork 0
Code Issues 34 Pull Requests 11 Actions Packages Releases Wiki Activity
Files
25113fae92961e015a8603967e250e27528dd3bb
bot-bottle/tests/unit/test_forge_identity.py
T
didericis-claude 25113fae92 fix(forge): fail closed on aliases colliding on a routable host 
Codex review (PR #520, P1): egress_forge_routes deduplicated referenced
forge accounts by hostname and silently dropped every account after the
first. Two aliases with different token_secret on the same host would let
all API calls to that host authenticate as whichever alias won the dedup —
acting as the wrong forge account and breaking the per-alias identity
guarantee.

Fail closed at composition (before the bottle is created): when two
referenced aliases resolve to the same host but disagree on
origin/auth/token_secret, resolve_forge_associations raises ManifestError.
Identical url/auth under two aliases still dedups to one route. The proxy
routes by host, so an ambiguous credential cannot be selected safely.

Regression tests cover both the conflicting-tokens (raise) and
identical-config (single route) cases; forge.py stays at 100% coverage.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-27 02:06:10 +00:00

427 lines
17 KiB
Python
Raw Blame History

"""Unit: trusted agent forge identity & guidance
(PRD prd-new-trusted-agent-forge-identity).
Covers the net-new surface: agent `author`, agent `forge-accounts` (Gitea API
URL validation + host token reference), the repo→forge association resolved at
composition, the synthesized proxy-held egress route, and the generated,
non-secret prompt guidance. Legacy git-gate.user / cwd-agent behavior lives in
the manifest test modules.
"""
from __future__ import annotations
import os
import tempfile
import unittest
from pathlib import Path
from typing import Callable
from unittest.mock import patch
from bot_bottle.egress import (
egress_forge_routes,
egress_render_routes,
egress_resolve_token_values,
egress_routes_for_bottle,
egress_token_env_map,
)
from bot_bottle.manifest import ManifestError, ManifestIndex
from bot_bottle.manifest.forge import (
ManifestForgeAccount,
canonicalize_forge_url,
render_forge_guidance,
)
GITEA = {
"url": "https://gitea.dideric.is/api/v1",
"auth": {"type": "token", "token_secret": "GITEA_CLAUDE_TOKEN"},
}
def _repo(*, forge: str | None = None) -> dict[str, object]:
entry: dict[str, object] = {
"url": "ssh://git@100.78.141.42:30009/didericis/bot-bottle.git",
"key": {"provider": "static", "path": "/k"},
}
if forge is not None:
entry["forge"] = forge
return entry
def _index(
*,
author: dict[str, object] | None = None,
forge_accounts: dict[str, object] | None = None,
repo_forge: str | None = None,
) -> ManifestIndex:
"""Build an eager index with one agent 'claude' + bottle 'bb'."""
agent: dict[str, object] = {"bottle": "bb", "prompt": ""}
if author is not None:
agent["author"] = author
if forge_accounts is not None:
agent["forge-accounts"] = forge_accounts
bottle: dict[str, object] = {
"git-gate": {"repos": {"bot-bottle": _repo(forge=repo_forge)}}
}
return ManifestIndex.from_json_obj(
{"bottles": {"bb": bottle}, "agents": {"claude": agent}}
)
def _error(
callable_: Callable[..., object], *args: object, **kwargs: object,
) -> str:
try:
callable_(*args, **kwargs)
except ManifestError as e:
return str(e)
raise AssertionError("expected ManifestError was not raised")
# ---------------------------------------------------------------------------
# author
# ---------------------------------------------------------------------------
class TestAuthor(unittest.TestCase):
def test_populates_git_identity(self) -> None:
idx = _index(author={"name": "didericis-claude", "email": "e+c@x.is"})
m = idx.load_for_agent("claude", ())
self.assertEqual("didericis-claude", m.bottle.git_user.name)
self.assertEqual("e+c@x.is", m.bottle.git_user.email)
self.assertEqual(
"name=didericis-claude, email=e+c@x.is",
m.git_identity_summary(),
)
def test_absent_author_no_identity(self) -> None:
m = _index().load_for_agent("claude", ())
self.assertTrue(m.bottle.git_user.is_empty())
self.assertIsNone(m.git_identity_summary())
def test_name_required(self) -> None:
msg = _error(_index, author={"email": "e@x.is"})
self.assertIn("author.name must be a non-empty string", msg)
def test_email_required(self) -> None:
msg = _error(_index, author={"name": "n"})
self.assertIn("author.email must be a non-empty string", msg)
def test_email_rejects_whitespace(self) -> None:
msg = _error(_index, author={"name": "n", "email": "a b@x.is"})
self.assertIn("must not contain whitespace", msg)
def test_name_rejects_newline(self) -> None:
msg = _error(_index, author={"name": "a\nb", "email": "e@x.is"})
self.assertIn("author.name must not contain newlines", msg)
def test_unknown_key(self) -> None:
msg = _error(
_index, author={"name": "n", "email": "e@x.is", "role": "x"}
)
self.assertIn("author has unknown key", msg)
# ---------------------------------------------------------------------------
# forge-accounts URL validation
# ---------------------------------------------------------------------------
class TestForgeUrl(unittest.TestCase):
def test_canonical_ok(self) -> None:
canonical, origin, host, prefix = canonicalize_forge_url(
"a", "g", "https://Gitea.Dideric.is/api/v1/"
)
# trailing slash normalized; host lowercased.
self.assertEqual("https://gitea.dideric.is/api/v1", canonical)
self.assertEqual("https://gitea.dideric.is", origin)
self.assertEqual("gitea.dideric.is", host)
self.assertEqual("/api/v1", prefix)
def test_port_preserved(self) -> None:
canonical, origin, _host, _ = canonicalize_forge_url(
"a", "g", "https://gitea.local:3000/api/v1"
)
self.assertEqual("https://gitea.local:3000/api/v1", canonical)
self.assertEqual("https://gitea.local:3000", origin)
def test_http_fails(self) -> None:
self.assertIn("must use https", _error(
canonicalize_forge_url, "a", "g", "http://gitea/api/v1"))
def test_userinfo_fails(self) -> None:
self.assertIn("userinfo", _error(
canonicalize_forge_url, "a", "g", "https://u:p@gitea/api/v1"))
def test_query_fails(self) -> None:
self.assertIn("query string", _error(
canonicalize_forge_url, "a", "g", "https://gitea/api/v1?x=1"))
def test_fragment_fails(self) -> None:
self.assertIn("fragment", _error(
canonicalize_forge_url, "a", "g", "https://gitea/api/v1#x"))
def test_missing_host_fails(self) -> None:
self.assertIn("hostname", _error(
canonicalize_forge_url, "a", "g", "https:///api/v1"))
def test_bad_path_fails(self) -> None:
self.assertIn("Gitea API base", _error(
canonicalize_forge_url, "a", "g", "https://gitea/api/v2"))
def test_non_string_fails(self) -> None:
self.assertIn("required", _error(
canonicalize_forge_url, "a", "g", None))
def test_malformed_url_fails(self) -> None:
# An unparseable URL (invalid IPv6 literal) trips urlsplit's ValueError.
self.assertIn("not a valid URL", _error(
canonicalize_forge_url, "a", "g", "https://[/api/v1"))
class TestForgeAccount(unittest.TestCase):
def test_parses(self) -> None:
acct = ManifestForgeAccount.from_dict("a", "didericis-gitea", GITEA)
self.assertEqual("didericis-gitea", acct.alias)
self.assertEqual("gitea.dideric.is", acct.host)
self.assertEqual("token", acct.auth_type)
self.assertEqual("GITEA_CLAUDE_TOKEN", acct.token_secret)
def test_non_kebab_alias_fails(self) -> None:
self.assertIn("valid alias", _error(
ManifestForgeAccount.from_dict, "a", "Bad_Alias", GITEA))
def test_auth_required(self) -> None:
self.assertIn("missing required 'auth'", _error(
ManifestForgeAccount.from_dict, "a", "g",
{"url": "https://gitea/api/v1"}))
def test_bad_auth_type_fails(self) -> None:
self.assertIn("auth.type must be", _error(
ManifestForgeAccount.from_dict, "a", "g",
{"url": "https://gitea/api/v1",
"auth": {"type": "basic", "token_secret": "T"}}))
def test_token_secret_required(self) -> None:
self.assertIn("token_secret must be", _error(
ManifestForgeAccount.from_dict, "a", "g",
{"url": "https://gitea/api/v1", "auth": {"type": "token"}}))
def test_unknown_key_fails(self) -> None:
self.assertIn("unknown key", _error(
ManifestForgeAccount.from_dict, "a", "g",
{**GITEA, "bogus": 1}))
def test_unknown_auth_key_fails(self) -> None:
self.assertIn("auth has unknown key", _error(
ManifestForgeAccount.from_dict, "a", "g",
{"url": "https://gitea/api/v1",
"auth": {"type": "token", "token_secret": "T", "bogus": 1}}))
# ---------------------------------------------------------------------------
# repo -> forge association (composition)
# ---------------------------------------------------------------------------
class TestForgeAssociations(unittest.TestCase):
def test_resolves_when_repo_references_alias(self) -> None:
idx = _index(
forge_accounts={"didericis-gitea": GITEA},
repo_forge="didericis-gitea",
)
m = idx.load_for_agent("claude", ())
self.assertEqual(1, len(m.forge_associations))
assoc = m.forge_associations[0]
self.assertEqual("didericis-gitea", assoc.alias)
self.assertEqual(("bot-bottle",), assoc.repo_names)
def test_unreferenced_account_yields_no_association(self) -> None:
idx = _index(forge_accounts={"didericis-gitea": GITEA}) # repo has no forge
m = idx.load_for_agent("claude", ())
self.assertEqual((), m.forge_associations)
def test_unknown_alias_fails_closed(self) -> None:
idx = _index(
forge_accounts={"didericis-gitea": GITEA}, repo_forge="nope"
)
msg = _error(idx.load_for_agent, "claude", ())
self.assertIn("references forge alias 'nope'", msg)
self.assertIn("not defined on agent", msg)
def test_forge_without_account_fails_closed(self) -> None:
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)
# ---------------------------------------------------------------------------
class TestForgeEgressRoutes(unittest.TestCase):
def _assoc(self):
idx = _index(
forge_accounts={"didericis-gitea": GITEA},
repo_forge="didericis-gitea",
)
return idx.load_for_agent("claude", ())
def test_route_shape(self) -> None:
routes = egress_forge_routes(self._assoc().forge_associations)
self.assertEqual(1, len(routes))
r = routes[0]
self.assertEqual("gitea.dideric.is", r.host)
self.assertEqual("token", r.auth_scheme)
self.assertEqual("GITEA_CLAUDE_TOKEN", r.token_ref)
self.assertTrue(r.inspect)
# scoped to the API prefix.
self.assertEqual("/api/v1", r.matches[0].paths[0].value)
def test_token_slot_and_resolution(self) -> None:
m = self._assoc()
forge_routes = egress_forge_routes(m.forge_associations)
routes = egress_routes_for_bottle(m.bottle, (), forge_routes)
token_map = egress_token_env_map(routes)
# one slot mapping the proxy env slot -> host env var name.
self.assertEqual({"EGRESS_TOKEN_0": "GITEA_CLAUDE_TOKEN"}, token_map)
resolved = egress_resolve_token_values(
token_map, {"GITEA_CLAUDE_TOKEN": "sekret-value"}
)
self.assertEqual({"EGRESS_TOKEN_0": "sekret-value"}, resolved)
def test_rendered_routes_hide_secret_name_and_value(self) -> None:
m = self._assoc()
routes = egress_routes_for_bottle(
m.bottle, (), egress_forge_routes(m.forge_associations)
)
rendered = egress_render_routes(routes)
self.assertIn("gitea.dideric.is", rendered)
self.assertIn("EGRESS_TOKEN_0", rendered) # slot, not the host var name
self.assertNotIn("GITEA_CLAUDE_TOKEN", rendered)
self.assertNotIn("sekret-value", rendered)
def test_dedup_by_host(self) -> None:
# Two repos referencing the same alias share one route/credential.
idx = ManifestIndex.from_json_obj({
"bottles": {"bb": {"git-gate": {"repos": {
"r1": _repo(forge="g"),
"r2": {
"url": "ssh://git@h/x/other.git",
"key": {"provider": "static", "path": "/k"},
"forge": "g",
},
}}}},
"agents": {"claude": {"bottle": "bb", "prompt": "",
"forge-accounts": {"g": GITEA}}},
})
m = idx.load_for_agent("claude", ())
routes = egress_forge_routes(m.forge_associations)
self.assertEqual(1, len(routes))
# ---------------------------------------------------------------------------
# generated prompt guidance
# ---------------------------------------------------------------------------
class TestForgeGuidance(unittest.TestCase):
def _assoc(self):
idx = _index(
forge_accounts={"didericis-gitea": GITEA},
repo_forge="didericis-gitea",
)
return idx.load_for_agent("claude", ()).forge_associations
def test_empty_when_no_associations(self) -> None:
self.assertEqual("", render_forge_guidance(()))
def test_names_account_repo_and_workflow(self) -> None:
text = render_forge_guidance(self._assoc())
self.assertIn("didericis-gitea", text)
self.assertIn("https://gitea.dideric.is/api/v1", text)
self.assertIn("bot-bottle", text)
# branch-backed PR workflow + AGit prohibition.
self.assertIn("refs/heads/", text)
self.assertIn("refs/for/*", text)
self.assertIn("pull request", text)
def test_no_token_secret_name_or_value(self) -> None:
text = render_forge_guidance(self._assoc())
self.assertNotIn("GITEA_CLAUDE_TOKEN", text)
def test_prompt_file_appends_guidance(self) -> None:
from bot_bottle.backend.resolve_common import prepare_agent_state_dir
idx = _index(
author={"name": "n", "email": "e@x.is"},
forge_accounts={"didericis-gitea": GITEA},
repo_forge="didericis-gitea",
)
# Give the agent a real prompt body.
m = idx.load_for_agent("claude", ())
from dataclasses import replace
m = replace(m, agent=replace(m.agent, prompt="BASE PROMPT"))
with tempfile.TemporaryDirectory() as td:
with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": td}):
_dir, prompt_file = prepare_agent_state_dir("slug1", m)
body = Path(prompt_file).read_text()
self.assertIn("BASE PROMPT", body)
self.assertIn("Forge access", body)
self.assertIn("https://gitea.dideric.is/api/v1", body)
def test_prompt_file_no_guidance_without_forge(self) -> None:
from bot_bottle.backend.resolve_common import prepare_agent_state_dir
m = _index(author={"name": "n", "email": "e@x.is"}).load_for_agent(
"claude", ()
)
with tempfile.TemporaryDirectory() as td:
with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": td}):
_dir, prompt_file = prepare_agent_state_dir("slug2", m)
body = Path(prompt_file).read_text()
self.assertNotIn("Forge access", body)
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink