bot-bottle/bot_bottle/manifest/forge.py

"""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", ""))

    return tuple(
        ResolvedForgeAssociation(
            account=forge_accounts[alias],
            repo_names=tuple(sorted(set(names))),
        )
        for alias, names in sorted(by_alias.items())
    )


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"