generated from coulomb/repo-seed
tegwick
2c513864bc
Adds `warden access <need> [--domain X] [--json]`: resolves a credential need against the routing catalog and renders the structured handoff (owner, auth method, path template, command skeleton, policy gate status, proxy hint). SSH lane points at `warden sign`; routed lanes end "warden advises, the owner vends". New pure warden/access.py module (expand_handoff, policy_gate_status) reused by the T3 proxy lane. JSON output is stable and secret-free. tests/test_access.py added. 157 passed, lint clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
77 lines
2.9 KiB
Python
77 lines
2.9 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
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
from warden.config import ConfigError, load_config
|
|
from warden.routing.models import RouteEntry
|
|
|
|
|
|
@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 _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.enabled:
|
|
return f"enforced — flex-auth at {cfg.policy.flex_auth_url}"
|
|
return "advisory — policy.enabled=false (gate ships with flex-auth deploy)"
|