tegwick
00145d705e
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06eaf-3425-7f92-a0c2-bb4aa4faebe4
103 lines
4.2 KiB
Python
103 lines
4.2 KiB
Python
"""Operator access assist — render structured handoff for a credential need.
|
|
|
|
The `warden access` front door (WP-0014) resolves a need to a `RouteEntry` and
|
|
renders its **structured handoff**: how the caller authenticates to the owning
|
|
subsystem, the owner-side path template, the command skeleton to run *as the
|
|
caller*, and the policy check the fetch path gates on.
|
|
|
|
This module is **pure**: it expands templates and reports gate status. It never
|
|
fetches, holds, or logs a secret value — that boundary is the whole point of the
|
|
assist layer. Proxy execution (`--fetch`/`--exec`) lives in the CLI/T3 lane and
|
|
reuses `expand_handoff` to build the command it runs as the caller.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import dataclass
|
|
from typing import Mapping, Optional
|
|
|
|
from warden.config import ConfigError, load_config
|
|
from warden.routing.models import RouteEntry
|
|
|
|
|
|
# KeyCape KEY-WP-0009-T03 is the issued coding-agent identity available today.
|
|
# This is intentionally an exact subject allowlist, not a guess based on every
|
|
# ``service:*`` identity. OpenBao validates and enforces the credential; this
|
|
# module only recognizes the already-issued subject for the advisory CLI guard.
|
|
ISSUED_AGENT_SUBJECTS = frozenset({"service:codex:railiance-platform"})
|
|
|
|
|
|
@dataclass
|
|
class ExpandedHandoff:
|
|
"""Handoff templates with `<domain>` substituted when a domain is supplied.
|
|
|
|
Remaining placeholders (`<workload>`, `<bundle>`, `<FIELD>`) are intentionally
|
|
left for the caller/owner to fill — ops-warden does not invent owner-side names.
|
|
"""
|
|
|
|
auth_method: Optional[str]
|
|
path_template: Optional[str]
|
|
fetch_command: Optional[str]
|
|
policy_ref: Optional[str]
|
|
exec_capable: bool
|
|
|
|
|
|
def agent_read_boundary_identity(
|
|
environ: Mapping[str, str] | None = None,
|
|
) -> str | None:
|
|
"""Return the issued/fallback agent marker used by the advisory read guard.
|
|
|
|
``WARDEN_POLICY_SUBJECT`` carries the principal identity used by the policy
|
|
request. When it names an issued coding-agent subject, prefer it over the
|
|
self-declared legacy marker. This function does not validate a token or
|
|
render an authorization decision; OpenBao's agent policy is the enforced
|
|
boundary. ``WARDEN_AGENT_ID`` remains a fail-toward-safety fallback.
|
|
"""
|
|
env = os.environ if environ is None else environ
|
|
issued_subject = str(env.get("WARDEN_POLICY_SUBJECT") or "").strip()
|
|
if issued_subject in ISSUED_AGENT_SUBJECTS:
|
|
return issued_subject
|
|
fallback = str(env.get("WARDEN_AGENT_ID") or "").strip()
|
|
return fallback or None
|
|
|
|
|
|
def _sub_domain(value: Optional[str], domain: Optional[str]) -> Optional[str]:
|
|
if value and domain:
|
|
return value.replace("<domain>", domain)
|
|
return value
|
|
|
|
|
|
def expand_handoff(entry: RouteEntry, domain: Optional[str] = None) -> ExpandedHandoff:
|
|
"""Expand an entry's handoff templates for display or proxy.
|
|
|
|
The catalog `fetch_command` may reference the literal token ``<path_template>``;
|
|
we inline the entry's ``path_template`` so the rendered command is self-contained,
|
|
then substitute ``<domain>`` across every field when a domain is given.
|
|
"""
|
|
path = entry.path_template
|
|
fetch = entry.fetch_command
|
|
if fetch and path and "<path_template>" in fetch:
|
|
fetch = fetch.replace("<path_template>", path)
|
|
|
|
return ExpandedHandoff(
|
|
auth_method=_sub_domain(entry.auth_method, domain),
|
|
path_template=_sub_domain(path, domain),
|
|
fetch_command=_sub_domain(fetch, domain),
|
|
policy_ref=_sub_domain(entry.policy_ref, domain),
|
|
exec_capable=entry.exec_capable,
|
|
)
|
|
|
|
|
|
def policy_gate_status() -> str:
|
|
"""One-line description of whether the flex-auth gate is enforced for fetches.
|
|
|
|
Advisory output only — never raises. The proxy lane (T3) is what actually runs
|
|
the gate before fetching; here we just report the configured posture.
|
|
"""
|
|
try:
|
|
cfg = load_config()
|
|
except ConfigError:
|
|
return "advisory — no warden.yaml (caller identity; gate not enforced)"
|
|
if cfg.policy.flex_auth_url:
|
|
return f"zone-aware — flex-auth at {cfg.policy.flex_auth_url}"
|
|
return "zone-aware — evaluator unconfigured; unknown-zone fail_open applies"
|