feat: complete local layer model v0.7 conformance work
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06eaf-3425-7f92-a0c2-bb4aa4faebe4
This commit is contained in:
tegwick 2026-09-05 01:19:48 +02:00
parent 89b117f640
commit 00145d705e
13 changed files with 279 additions and 27 deletions

View file

@ -12,13 +12,21 @@ 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 Optional
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.
@ -34,6 +42,25 @@ class ExpandedHandoff:
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)

View file

@ -2,7 +2,6 @@
from __future__ import annotations
import json
import os
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Annotated, List, Optional
@ -1309,7 +1308,9 @@ def _access_proxy(
# T04 — agent identity on a high-risk lane: never stream raw secret data.
# Agents may use sanctioned transports (--out / --exec / --wrap / --fingerprint).
agent_id = os.environ.get("WARDEN_AGENT_ID", "").strip()
from warden.access import agent_read_boundary_identity
agent_id = agent_read_boundary_identity()
raw_value_stream = (
not is_login and not do_exec and not wrap and not out_path and not fingerprint
)