feat(a2a): delegated-principal extension — carry the originating principal

agent-templates/a2a/authz.py

@@ -18,6 +18,8 @@
 from dataclasses import dataclass, field
 from enum import Enum
 
+from .delegation import Delegation, DelegationPolicy, SkillClass
+
 _REPO_NAME_RE = re.compile(r"^[A-Za-z0-9_.-]+$")
 _EXEC_PRINCIPAL_RE = re.compile(r"^Exec-[a-z0-9_-]+$")
 
@@ -37,6 +39,10 @@ class AuthContext:
     caller: str
     scopes: frozenset[str] = frozenset()
     authenticated: bool = True
+    #: `delegated-principal` extension claims, from an ALREADY-VERIFIED credential.
+    #: None = the caller sent none, which is exactly v1 and is only a problem for a
+    #: skill the callee classified `principal-required`.
+    delegation: Delegation | None = None
 
 
 @dataclass(frozen=True)
@@ -77,6 +83,9 @@ def authorize(
     *,
     skill_role: dict | None = None,
     skill_known: bool = True,
+    skill_key: str | None = None,
+    delegation_policy: DelegationPolicy | None = None,
+    permit_check=None,
 ) -> AuthzResult:
     """Run the callee-side decision procedure (authz.md §3).
 
@@ -87,6 +96,12 @@ def authorize(
       4. providesTo: ABSENT -> DENY, [] -> DENY, caller not in -> DENY.
       5. skill: unknown/unpublished-to-caller -> DENY; a2a.scopes present but token
          lacks them -> SCOPE_REQUIRED (the sole legitimate AUTH_REQUIRED).
+      6. delegated-principal extension, ONLY when ``delegation_policy`` is given.
+         Absent policy == the callee did not adopt the extension == exact v1.
+
+    Step 6 runs AFTER step 4 on purpose: channel authz still decides first, so a
+    caller outside ``providesTo`` is denied before any principal is consulted. The
+    extension narrows; it never widens.
     """
     if not ctx.authenticated or not valid_caller_identity(ctx.caller):
         return AuthzResult(Decision.DENY, "unauthenticated or invalid caller identity")
@@ -118,4 +133,85 @@ def authorize(
                 missing_scopes=missing,
             )
 
+    if delegation_policy is not None:
+        return _authorize_delegated(ctx, delegation_policy, skill_key, permit_check)
+
     return AuthzResult(Decision.ALLOW, "authorized")
+
+
+def _authorize_delegated(
+    ctx: AuthContext,
+    policy: DelegationPolicy,
+    skill_key: str | None,
+    permit_check,
+) -> AuthzResult:
+    """Extension step 6 (ext/delegated-principal/v1 spec.md §§3-5).
+
+    Every branch that cannot be evaluated is a DENY. There is no fail-open mode
+    and no flag that creates one: `DECISION_UNAVAILABLE` is never an allow.
+    """
+    if not skill_key:
+        return AuthzResult(
+            Decision.DENY,
+            "delegation policy present but no skill key to classify -> fail closed",
+        )
+
+    klass = policy.classify(skill_key)
+    if klass is None:
+        # spec.md §3: the closed set has no fourth bucket. Defaulting an
+        # unclassified skill to `delegable` would let a new skill acquire the
+        # weakest rule in the system by being written rather than decided.
+        return AuthzResult(
+            Decision.DENY,
+            f"skill {skill_key!r} is unclassified in delegation.skills -> fail closed",
+        )
+
+    if klass is SkillClass.NEVER_DELEGABLE:
+        return AuthzResult(
+            Decision.DENY,
+            f"skill {skill_key!r} is never-delegable; no principal reaches it via A2A",
+        )
+
+    if klass is SkillClass.DELEGABLE:
+        return AuthzResult(Decision.ALLOW, "authorized (delegable)")
+
+    # principal-required: BOTH terms of the intersection must allow (spec.md §4).
+    delegation = ctx.delegation
+    if delegation is None:
+        return AuthzResult(
+            Decision.DENY,
+            f"skill {skill_key!r} is principal-required but the call carries no "
+            f"verified subject",
+        )
+
+    # Term 2 first, because it needs no network call: the subject cannot reach past
+    # what the agent may broker, however privileged the subject is.
+    if not policy.brokerable_by(delegation.actor_chain, skill_key):
+        return AuthzResult(
+            Decision.DENY,
+            f"no actor in {delegation.actor_chain} may broker {skill_key!r}",
+        )
+
+    # Term 1: what the SUBJECT may do, decided by Permit via FuzeFront's Security
+    # API. A product never calls Permit directly.
+    if permit_check is None:
+        return AuthzResult(
+            Decision.DENY,
+            "principal-required skill but no authz client is wired -> cannot "
+            "evaluate the subject's permission -> fail closed",
+        )
+    try:
+        permitted = permit_check(delegation.subject, skill_key)
+    except Exception as exc:
+        # DECISION_UNAVAILABLE is a DENY. This is the branch that a fail-open
+        # implementation would turn into an ALLOW, so it is spelled out.
+        return AuthzResult(Decision.DENY, f"authz decision unavailable: {exc}")
+    if not permitted:
+        return AuthzResult(
+            Decision.DENY,
+            f"subject {delegation.subject!r} is not permitted {skill_key!r}",
+        )
+
+    # The agent cannot grant more than the subject has, AND the subject cannot
+    # reach past what the agent may broker. Both were required; neither alone.
+    return AuthzResult(Decision.ALLOW, "authorized (principal-required, intersection)")

agent-templates/a2a/delegation.py

@@ -0,0 +1,179 @@
+"""The `delegated-principal` A2A extension (contracts/a2a/ext/delegated-principal/v1).
+
+WHY THIS EXISTS. v1's `AuthContext.caller` is one string, and a caller identity is
+a bare repo name or an `Exec-*` principal (authz.md §2). There is no end-user
+subject anywhere in the model, so a CEO and a warehouse worker arriving through
+the same calling repo are indistinguishable to the callee: if `FuzeExecutive` is
+in `FuzePlan`'s `providesTo`, both pass identically.
+
+THE POD IS A DELEGATE, NEVER A PRINCIPAL. It holds no standing authority over
+product data; its workload credential authorizes exactly one thing — presenting
+delegated tokens. Every bit of data-plane authority arrives with the call. The
+alternative, giving the pod broad rights and having it check the caller first, is
+the confused deputy: the guard becomes a matter of remembering to look.
+
+This module is ADDITIVE. `Delegation` absent from an AuthContext means the callee
+did not adopt the extension, and `authorize()` behaves exactly as v1. Nothing
+here can make a v1 deployment stricter or looser than it was.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from enum import Enum
+
+EXTENSION_URI = "https://contracts.fuzefront.com/a2a/ext/delegated-principal/v1"
+
+_SUBJECT_RE = re.compile(r"^(user|service):[^\s:]+$")
+_ACTOR_RE = re.compile(r"^(repo|agent|service):[^\s:]+$")
+
+
+class SkillClass(str, Enum):
+    """Closed. There is deliberately no fourth member and no default.
+
+    An unclassified skill is DENIED, not treated as `DELEGABLE`. A default would
+    mean a newly written skill acquires the weakest rule in the system by being
+    written rather than by being decided — the same closed-set property the route
+    and OpenAPI gates enforce.
+    """
+
+    DELEGABLE = "delegable"
+    PRINCIPAL_REQUIRED = "principal-required"
+    NEVER_DELEGABLE = "never-delegable"
+
+
+class DelegationError(ValueError):
+    """A malformed policy. Raised at load, never carried into a decision."""
+
+
+@dataclass(frozen=True)
+class Delegation:
+    """Claims read from an ALREADY-VERIFIED credential.
+
+    Constructing this from an unverified token is the one way to misuse the
+    module: an unverified `sub` is worth less than no `sub`, because it looks
+    like authority. The caller of `parse_claims` verifies signature and `aud`
+    first — this module never sees a raw token and cannot check that for you.
+    """
+
+    #: RFC 8693 `sub` — the ORIGINATING principal, whom the work is ultimately for.
+    subject: str
+    #: RFC 8693 `act` chain, outermost first: immediate actor, then its actor.
+    actor_chain: tuple[str, ...] = field(default_factory=tuple)
+
+
+@dataclass(frozen=True)
+class DelegationPolicy:
+    """A callee's classification of its own skills, plus who may broker what.
+
+    Loaded from `.fuze/a2a-delegation.json` — see `from_manifest` for why it is a
+    separate file rather than a manifest key.
+    """
+
+    skills: dict[str, SkillClass]
+    brokerable: dict[str, frozenset[str]]
+    authz_base_url: str | None = None
+
+    @classmethod
+    def from_manifest(cls, block: dict | None) -> "DelegationPolicy | None":
+        """Build from `.fuze/a2a-delegation.json`, or None when the file is absent.
+
+        Its OWN file, not a key in the manifest's `a2a` block: v1's
+        manifest-a2a-extension.schema.json sets additionalProperties:false there,
+        so a `delegation` key would make every adopting repo's manifest fail v1
+        validation. An extension you must edit the frozen contract to adopt is a
+        version, not an extension.
+
+        None means "this callee did not adopt the extension" and restores exact v1
+        behaviour. A block that is PRESENT but malformed raises: half-configured
+        delegation is more dangerous than none, because it reads as protection.
+        """
+        if not block:
+            return None
+        if block.get("extension") != EXTENSION_URI:
+            raise DelegationError(
+                f"delegation block does not name {EXTENSION_URI}; refusing to guess "
+                f"which extension's rules apply"
+            )
+        raw_skills = block.get("skills")
+        if not isinstance(raw_skills, dict) or not raw_skills:
+            raise DelegationError("delegation.skills must be a non-empty object")
+        skills = {}
+        for key, value in raw_skills.items():
+            try:
+                skills[key] = SkillClass(value)
+            except ValueError as exc:
+                raise DelegationError(
+                    f"skill {key!r} has unknown class {value!r}; the classes are "
+                    f"{[c.value for c in SkillClass]} and there is no default"
+                ) from exc
+        raw_brokerable = block.get("brokerable")
+        if not isinstance(raw_brokerable, dict):
+            raise DelegationError("delegation.brokerable must be an object")
+        brokerable = {}
+        for actor, roles in raw_brokerable.items():
+            if not _ACTOR_RE.match(actor):
+                raise DelegationError(
+                    f"brokerable key {actor!r} is not a typed actor reference "
+                    f"(repo:/agent:/service:)"
+                )
+            if not isinstance(roles, list):
+                raise DelegationError(f"brokerable[{actor!r}] must be an array")
+            brokerable[actor] = frozenset(roles)
+        return cls(
+            skills=skills,
+            brokerable=brokerable,
+            authz_base_url=block.get("authzBaseUrl"),
+        )
+
+    def classify(self, skill: str) -> SkillClass | None:
+        """The skill's class, or None when unclassified. None is DENY, not a default."""
+        return self.skills.get(skill)
+
+    def brokerable_by(self, actor_chain: tuple[str, ...], skill: str) -> bool:
+        """May any actor in the chain broker this skill?
+
+        An actor absent from `brokerable` brokers NOTHING — absence is not a
+        wildcard. An explicit empty list says the same thing on purpose, which is
+        a meaningful statement in a way absence is not.
+        """
+        return any(
+            skill in self.brokerable.get(actor, frozenset()) for actor in actor_chain
+        )
+
+
+def parse_claims(claims: dict | None) -> Delegation | None:
+    """Delegation from VERIFIED credential claims, or None when absent.
+
+    Raises DelegationError on a malformed chain rather than dropping the bad
+    entry: an actor reference that cannot be evaluated must not be treated as
+    satisfied, and silently ignoring it is exactly that.
+    """
+    if not claims:
+        return None
+    subject = claims.get("sub")
+    if not subject:
+        return None
+    if not _SUBJECT_RE.match(subject):
+        raise DelegationError(
+            f"sub {subject!r} is not a typed principal (user:/service:); an untyped "
+            f"id cannot be resolved to a principal kind and so cannot be checked"
+        )
+    chain: list[str] = []
+    node = claims.get("act")
+    while isinstance(node, dict):
+        actor = node.get("sub")
+        if not actor or not _ACTOR_RE.match(actor or ""):
+            raise DelegationError(
+                f"actor {actor!r} is not a typed reference (repo:/agent:/service:); "
+                f"it cannot be matched against the brokerable set"
+            )
+        chain.append(actor)
+        node = node.get("act")
+    if not chain:
+        raise DelegationError(
+            "sub is present but the act chain is empty; a delegated call must name "
+            "the actor doing the delegating"
+        )
+    return Delegation(subject=subject, actor_chain=tuple(chain))