didericis-claude
91f6cbabdc
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>
313 lines
13 KiB
Python
313 lines
13 KiB
Python
"""Agent-owned author identity and forge accounts (PRD prd-new-trusted-agent-forge-identity).
|
|
|
|
`ManifestAuthor` is the agent's git author identity (name/email) — it moved off
|
|
`git-gate.user` (PRD 0027 / ADR 0002) and onto the trusted agent definition.
|
|
|
|
`ManifestForgeAccount` maps a **forge alias** to a canonical HTTPS Gitea API
|
|
origin plus the *name* of a host env var holding the operator-provided API
|
|
token. The token value never lives on the manifest; the host resolves it only
|
|
when a selected bottle repository references the alias, and hands it to the
|
|
egress proxy — never to the bottle.
|
|
|
|
`ResolvedForgeAssociation` is the composed view: one distinct forge alias that
|
|
at least one selected git-gate repository points at, with the repository names
|
|
that reference it. The proxy-provisioning and prompt-guidance steps consume it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import urllib.parse
|
|
from dataclasses import dataclass
|
|
|
|
from .schema import is_valid_entity_name
|
|
from .util import ManifestError, as_json_object
|
|
|
|
# Only the Gitea `/api/v1` base is supported today. A future provider adds a
|
|
# validator + auth scheme + guidance renderer in code rather than accepting a
|
|
# repository-supplied prompt or path (PRD non-goal: provider-generic prompts).
|
|
GITEA_API_PREFIX = "/api/v1"
|
|
|
|
# Auth schemes an agent forge account may declare. Gitea uses `token`.
|
|
FORGE_AUTH_TYPES = ("token",)
|
|
|
|
|
|
def canonicalize_forge_url(
|
|
agent_name: str, alias: str, url: object,
|
|
) -> tuple[str, str, str, str]:
|
|
"""Validate and canonicalize a forge account's API base URL.
|
|
|
|
Returns `(canonical, origin, host, api_prefix)` where `origin` is
|
|
`https://host[:port]` and `canonical` is `origin + api_prefix`.
|
|
|
|
Fails closed on: non-string, non-`https`, embedded userinfo, query, or
|
|
fragment, a missing host, or any path other than the supported Gitea API
|
|
base (`/api/v1`, trailing slash tolerated)."""
|
|
label = f"agent '{agent_name}' forge-accounts.{alias}.url"
|
|
if not isinstance(url, str) or not url:
|
|
raise ManifestError(f"{label} is required (non-empty string)")
|
|
try:
|
|
parts = urllib.parse.urlsplit(url)
|
|
except ValueError as e:
|
|
raise ManifestError(f"{label} is not a valid URL ({e}): {url!r}") from e
|
|
|
|
if parts.scheme != "https":
|
|
raise ManifestError(
|
|
f"{label} must use https (got scheme {parts.scheme!r} in {url!r})"
|
|
)
|
|
if parts.username or parts.password:
|
|
raise ManifestError(
|
|
f"{label} must not embed userinfo (user:pass@); the token is held "
|
|
f"host-side via auth.token_secret. Got {url!r}"
|
|
)
|
|
if parts.query:
|
|
raise ManifestError(f"{label} must not contain a query string: {url!r}")
|
|
if parts.fragment:
|
|
raise ManifestError(f"{label} must not contain a fragment: {url!r}")
|
|
|
|
host = parts.hostname
|
|
if not host:
|
|
raise ManifestError(f"{label} must include a hostname: {url!r}")
|
|
|
|
path = parts.path.rstrip("/")
|
|
if path != GITEA_API_PREFIX:
|
|
raise ManifestError(
|
|
f"{label} path must be the Gitea API base '{GITEA_API_PREFIX}' "
|
|
f"(got {parts.path!r} in {url!r}); other providers and API paths "
|
|
f"are not yet supported."
|
|
)
|
|
|
|
host = host.lower()
|
|
netloc = host if parts.port is None else f"{host}:{parts.port}"
|
|
origin = f"https://{netloc}"
|
|
return f"{origin}{path}", origin, host, path
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManifestAuthor:
|
|
"""The agent's git author identity. Both fields are required and
|
|
non-empty — an author that is declared must fully identify the agent.
|
|
Populates `user.name` / `user.email` inside the bottle."""
|
|
|
|
name: str
|
|
email: str
|
|
|
|
@classmethod
|
|
def from_dict(cls, agent_name: str, raw: object) -> "ManifestAuthor":
|
|
d = as_json_object(raw, f"agent '{agent_name}' author")
|
|
for k in d:
|
|
if k not in {"name", "email"}:
|
|
raise ManifestError(
|
|
f"agent '{agent_name}' author has unknown key {k!r}; "
|
|
f"allowed: name, email"
|
|
)
|
|
name = d.get("name")
|
|
if not isinstance(name, str) or not name:
|
|
raise ManifestError(
|
|
f"agent '{agent_name}' author.name must be a non-empty string"
|
|
)
|
|
email = d.get("email")
|
|
if not isinstance(email, str) or not email:
|
|
raise ManifestError(
|
|
f"agent '{agent_name}' author.email must be a non-empty string"
|
|
)
|
|
# Git identity validation: reject values that would break the
|
|
# `git config user.email` line or smuggle a second config directive.
|
|
if any(c in email for c in ("\n", "\r", " ", "\t")):
|
|
raise ManifestError(
|
|
f"agent '{agent_name}' author.email must not contain whitespace "
|
|
f"or newlines (got {email!r})"
|
|
)
|
|
if any(c in name for c in ("\n", "\r")):
|
|
raise ManifestError(
|
|
f"agent '{agent_name}' author.name must not contain newlines "
|
|
f"(got {name!r})"
|
|
)
|
|
return cls(name=name, email=email)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManifestForgeAccount:
|
|
"""One forge alias on an agent: a canonical Gitea API origin plus the
|
|
host env var name that holds the agent-specific API token. The
|
|
authenticated account is whoever owns that token — there is no separate
|
|
account-name field."""
|
|
|
|
alias: str
|
|
url: str # canonical https base incl. /api/v1, no trailing slash
|
|
origin: str # https://host[:port]
|
|
host: str # lowercased hostname
|
|
api_prefix: str # /api/v1
|
|
auth_type: str # "token"
|
|
token_secret: str # host env var name holding the API token
|
|
|
|
@classmethod
|
|
def from_dict(
|
|
cls, agent_name: str, alias: str, raw: object,
|
|
) -> "ManifestForgeAccount":
|
|
if not is_valid_entity_name(alias):
|
|
raise ManifestError(
|
|
f"agent '{agent_name}' forge-accounts key {alias!r} is not a "
|
|
f"valid alias; must match [a-z][a-z0-9-]*"
|
|
)
|
|
label = f"agent '{agent_name}' forge-accounts.{alias}"
|
|
d = as_json_object(raw, label)
|
|
for k in d:
|
|
if k not in {"url", "auth"}:
|
|
raise ManifestError(
|
|
f"{label} has unknown key {k!r}; allowed: url, auth"
|
|
)
|
|
|
|
canonical, origin, host, api_prefix = canonicalize_forge_url(
|
|
agent_name, alias, d.get("url"),
|
|
)
|
|
|
|
if "auth" not in d:
|
|
raise ManifestError(f"{label} missing required 'auth' block")
|
|
auth = as_json_object(d.get("auth"), f"{label}.auth")
|
|
for k in auth:
|
|
if k not in {"type", "token_secret"}:
|
|
raise ManifestError(
|
|
f"{label}.auth has unknown key {k!r}; allowed: type, "
|
|
f"token_secret"
|
|
)
|
|
auth_type = auth.get("type")
|
|
if auth_type not in FORGE_AUTH_TYPES:
|
|
raise ManifestError(
|
|
f"{label}.auth.type must be one of "
|
|
f"{', '.join(FORGE_AUTH_TYPES)} (got {auth_type!r})"
|
|
)
|
|
token_secret = auth.get("token_secret")
|
|
if not isinstance(token_secret, str) or not token_secret:
|
|
raise ManifestError(
|
|
f"{label}.auth.token_secret must be a non-empty string (the "
|
|
f"name of a host env var holding the API token)"
|
|
)
|
|
return cls(
|
|
alias=alias,
|
|
url=canonical,
|
|
origin=origin,
|
|
host=host,
|
|
api_prefix=api_prefix,
|
|
auth_type=str(auth_type),
|
|
token_secret=token_secret,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ResolvedForgeAssociation:
|
|
"""A distinct forge alias referenced by one or more selected git-gate
|
|
repositories. `repo_names` lists the git-gate repo names pointing at
|
|
`account.alias` (sorted, deduplicated)."""
|
|
|
|
account: ManifestForgeAccount
|
|
repo_names: tuple[str, ...]
|
|
|
|
@property
|
|
def alias(self) -> str:
|
|
return self.account.alias
|
|
|
|
|
|
def resolve_forge_associations(
|
|
agent_name: str,
|
|
forge_accounts: "dict[str, ManifestForgeAccount]",
|
|
git_entries: "tuple[object, ...]",
|
|
) -> tuple[ResolvedForgeAssociation, ...]:
|
|
"""Compose the agent's forge accounts with the effective bottle's
|
|
git-gate repos. Each git entry that declares a `forge` alias must match a
|
|
forge account on the selected agent — otherwise fail closed before launch.
|
|
|
|
Returns one association per distinct referenced alias, carrying the repo
|
|
names that reference it. Unreferenced accounts produce nothing (no token
|
|
is resolved and no route is created for them)."""
|
|
by_alias: dict[str, list[str]] = {}
|
|
for entry in git_entries:
|
|
alias = getattr(entry, "Forge", "")
|
|
if not alias:
|
|
continue
|
|
if alias not in forge_accounts:
|
|
available = ", ".join(sorted(forge_accounts)) or "(none)"
|
|
raise ManifestError(
|
|
f"git-gate.repos['{getattr(entry, 'Name', '?')}'].forge "
|
|
f"references forge alias {alias!r}, which is not defined on "
|
|
f"agent '{agent_name}'. Available forge-accounts: {available}"
|
|
)
|
|
by_alias.setdefault(alias, []).append(getattr(entry, "Name", ""))
|
|
|
|
associations = tuple(
|
|
ResolvedForgeAssociation(
|
|
account=forge_accounts[alias],
|
|
repo_names=tuple(sorted(set(names))),
|
|
)
|
|
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, ...],
|
|
) -> str:
|
|
"""Render the non-secret, provider-specific forge workflow guidance
|
|
appended to the agent's system prompt for associated repositories only.
|
|
|
|
Derived entirely from validated typed fields — never repository-supplied
|
|
Markdown. Contains neither the token value nor its `token_secret` name.
|
|
Returns "" when there are no associations (no section is generated)."""
|
|
if not associations:
|
|
return ""
|
|
lines: list[str] = [
|
|
"## Forge access (managed by bot-bottle)",
|
|
"",
|
|
"bot-bottle authenticates the forge API requests below for you "
|
|
"through the egress proxy. Do not read, print, or manually attach "
|
|
"an authorization token — the proxy injects it. Never place a token "
|
|
"in a request.",
|
|
]
|
|
for assoc in associations:
|
|
acct = assoc.account
|
|
for repo in assoc.repo_names:
|
|
lines.extend([
|
|
"",
|
|
f"### Repository `{repo}` (forge alias `{acct.alias}`)",
|
|
f"- Forge API base URL: `{acct.url}`.",
|
|
f"- This git-gate repository (`{repo}`) is tied to forge "
|
|
f"`{acct.alias}`; use its API for forge actions on this repo.",
|
|
f"- Call the API at `{acct.url}` over HTTPS through the proxy. "
|
|
"The proxy adds authentication; you never supply a token "
|
|
"yourself.",
|
|
"- Git pushes still use the git-gate remote, not the API.",
|
|
"- To propose changes: push a normal branch "
|
|
"(`refs/heads/<branch>`) through the git-gate remote, then open "
|
|
"a branch-backed pull request through the API.",
|
|
"- Do NOT push AGit review refs (`refs/for/*`, `refs/draft/*`, "
|
|
"`refs/for-review/*`); they are prohibited here.",
|
|
"- Use the API for reviews and comments, and verify the "
|
|
"returned object state before claiming the task is complete.",
|
|
])
|
|
return "\n".join(lines) + "\n"
|