agent-credential-broker/agent_credential_broker/policy.py

"""Item-scoped policy derived from non-sensitive ``agent.*`` custom fields.

Recognized fields (all optional except agent.ref):

    agent.ref                 required. stable, unique, exact-match key
    agent.adapters            csv whitelist, e.g. "browser,http,git"
    agent.allowed_hosts       csv hostnames (ssh + host-level checks)
    agent.allowed_urls        csv origins or origin+path prefixes
    agent.auth_type           bearer | basic | none        (http)
    agent.confirmation        AUTO | SESSION | EVERY_USE | DENY
                              default: EVERY_USE (fail-safe)
    agent.username_selector   browser field selector
    agent.password_selector   browser field selector
    agent.submit_selector     browser field selector
    agent.ssh_user            ssh login override (else vault username)
    agent.ssh_key_path        existing local key file to use instead of
                              password auth (no key material moves)
    agent.cli_profile         restrict this credential to one profile id
    agent.insecure_http       "true" permits http:// targets (discouraged)

Policy is enforced daemon-side. The agent never sees or controls it.
"""

from __future__ import annotations

import urllib.parse
from dataclasses import dataclass, field

from .errors import adapter_denied, origin_denied, invalid_argument

CONFIRMATION_LEVELS = {"AUTO", "SESSION", "EVERY_USE", "DENY"}

SUPPORTED_ADAPTERS = {"browser", "http", "ssh", "git", "cli-profile"}


def _csv(value: str | None) -> list[str]:
    if not value:
        return []
    return [v.strip() for v in value.split(",") if v.strip()]


def normalize_url(url: str) -> urllib.parse.ParseResult:
    p = urllib.parse.urlparse(url if "://" in url else "https://" + url)
    if p.scheme not in ("https", "http", "ssh", "git"):
        raise invalid_argument(f"unsupported URL scheme {p.scheme!r}")
    if not p.hostname:
        raise invalid_argument(f"URL has no host: {url!r}")
    return p


def origin_of(url: str) -> str:
    p = normalize_url(url)
    port = f":{p.port}" if p.port else ""
    return f"{p.scheme}://{p.hostname}{port}"


@dataclass
class CredentialPolicy:
    ref: str
    item_id: str
    name: str = ""
    adapters: set = field(default_factory=set)
    allowed_hosts: set = field(default_factory=set)
    allowed_urls: list = field(default_factory=list)
    auth_type: str = "basic"
    confirmation: str = "EVERY_USE"
    username_selector: str | None = None
    password_selector: str | None = None
    submit_selector: str | None = None
    ssh_user: str | None = None
    ssh_key_path: str | None = None
    cli_profile: str | None = None
    insecure_http: bool = False
    uris: list = field(default_factory=list)

    def safe_metadata(self) -> dict:
        """Metadata that is safe to show the agent (no values)."""
        return {
            "ref": self.ref,
            "name": self.name,
            "adapters": sorted(self.adapters),
            "allowed_hosts": sorted(self.allowed_hosts),
            "allowed_urls": list(self.allowed_urls),
            "confirmation": self.confirmation,
            "auth_type": self.auth_type,
            "cli_profile": self.cli_profile,
            "uris": list(self.uris),
        }


def _parse_allowed_url(raw: str) -> urllib.parse.ParseResult:
    p = normalize_url(raw)
    return p


def policy_for(item: dict) -> CredentialPolicy | None:
    fields = item.get("fields") or {}
    ref = fields.get("agent.ref")
    if not ref:
        return None
    adapters = {a for a in _csv(fields.get("agent.adapters"))}
    unknown = adapters - SUPPORTED_ADAPTERS
    if unknown:
        adapters -= unknown
    confirmation = (fields.get("agent.confirmation") or "EVERY_USE").upper()
    if confirmation not in CONFIRMATION_LEVELS:
        confirmation = "EVERY_USE"
    auth_type = (fields.get("agent.auth_type") or "basic").lower()
    if auth_type not in ("bearer", "basic", "none"):
        auth_type = "basic"
    allowed_urls = []
    for raw in _csv(fields.get("agent.allowed_urls")) + list(item.get("uris") or []):
        try:
            p = _parse_allowed_url(raw)
            allowed_urls.append(origin_of(raw) + (p.path.rstrip("/") or ""))
        except Exception:
            continue
    return CredentialPolicy(
        ref=ref.strip(),
        item_id=item.get("id") or "",
        name=item.get("name") or "",
        adapters=adapters,
        allowed_hosts={h.lower() for h in _csv(fields.get("agent.allowed_hosts"))},
        allowed_urls=allowed_urls,
        auth_type=auth_type,
        confirmation=confirmation,
        username_selector=fields.get("agent.username_selector"),
        password_selector=fields.get("agent.password_selector"),
        submit_selector=fields.get("agent.submit_selector"),
        ssh_user=fields.get("agent.ssh_user"),
        ssh_key_path=fields.get("agent.ssh_key_path"),
        cli_profile=fields.get("agent.cli_profile"),
        insecure_http=(fields.get("agent.insecure_http") or "").lower() == "true",
        uris=list(item.get("uris") or []),
    )


def require_adapter(policy: CredentialPolicy, adapter: str):
    if adapter not in policy.adapters:
        raise adapter_denied(policy.ref, adapter)


def _default_port(scheme: str) -> int:
    return {"https": 443, "http": 80, "ssh": 22, "git": 9418}.get(scheme, -1)


def require_url_allowed(policy: CredentialPolicy, url: str, *, require_https=True):
    """Origin+path-prefix check against agent.allowed_urls / item URIs."""
    p = normalize_url(url)
    if require_https and p.scheme == "http" and not policy.insecure_http:
        raise origin_denied(policy.ref, url)
    cand_origin = f"{p.scheme}://{p.hostname}:{p.port or _default_port(p.scheme)}"
    cand = cand_origin + (p.path.rstrip("/") or "")
    for allowed in policy.allowed_urls:
        ap = normalize_url(allowed)
        a_origin = f"{ap.scheme}://{ap.hostname}:{ap.port or _default_port(ap.scheme)}"
        if a_origin != cand_origin:
            continue
        a_path = ap.path.rstrip("/")
        if not a_path or cand == a_origin + a_path or cand.startswith(a_origin + a_path + "/"):
            return
    raise origin_denied(policy.ref, url)


def require_host_allowed(policy: CredentialPolicy, host: str):
    if host.lower() not in policy.allowed_hosts:
        raise origin_denied(policy.ref, host)