Fix two Datalog injection authorization bypasses; release 0.4.0

CHANGELOG.md

@@ -2,6 +2,62 @@
 
 All notable changes to AIP (Agent Identity Protocol) are documented here.
 
+## [0.4.0] - 2026-08-23
+
+### Security
+
+Two authorization bypasses, both caused by building Biscuit Datalog with string
+interpolation. Anyone running 0.3.0 or earlier should upgrade.
+
+- **Authorizer injection in `ChainedToken.authorize()`.** The requested tool name
+  was interpolated into the authorizer source. A tool name that closed its own
+  string could append an in-allowlist `tool()` fact and `allow if true`, so a
+  token granting only `tool:search` authorized any tool. This required no
+  malicious token, no forged signature and no control of the issuer, only
+  influence over the tool name being checked. In an MCP binding that name comes
+  straight from the agent's JSON-RPC `params.name`. The tool is now validated and
+  bound as a Datalog parameter rather than interpolated.
+- **Block-encoder injection in `create_authority()` and `delegate()`.** The
+  issuer, principal, scope, delegator, delegate and context values were
+  interpolated into block Datalog, so a breakout in any of them injected facts
+  into a block that was then signed. An injected `tool()` fact satisfied the
+  authority block's own scope check for any tool, an injected `time()` fact
+  defeated expiry, and an injected `max_depth()` ahead of the real one was what
+  the depth check read. All six fields are now validated at the API boundary.
+
+`aip_token.canonical.validate_term()` rejects the double quote, the backslash and
+control characters, which are the only characters that can escape a Datalog
+string context. Semicolons, parentheses, brackets, commas, `//`, spaces, colons
+and non-ASCII characters remain valid, so ordinary identifiers and free-form
+context strings are unaffected.
+
+### Fixed
+
+- `SimplePolicy.to_datalog()` emitted `check if budget($b), $b <= N` and
+  `check if depth($d), $d <= N` in both the Python and Rust SDKs. Section 3.4.1
+  of draft-prakash-aip-01 requires these as the facts `budget_ceiling(N)` and
+  `max_depth(N)`, and states that implementations MUST NOT emit a check over
+  budget: such a check binds to whatever budget facts are in scope at evaluation
+  rather than to the block's own ceiling, so it either rejects valid chains or
+  passes unconditionally. Both dimensions are verified structurally in step V4.
+  The chained-token builders were already correct; only `SimplePolicy` was wrong.
+
+### Changed
+
+- The canonical Section 3.4.1 encoding now lives in one module per language
+  (`aip_token.canonical`, `aip_token::canonical`). It had been implemented twice
+  in each SDK, which is how `SimplePolicy` drifted from the chained builder.
+- New cross-language conformance vectors compare the Rust and Python renderings
+  byte for byte, and both SDKs now carry regression tests for the two bypasses.
+
+### Breaking
+
+- `SimplePolicy.to_datalog()` output has changed. It now emits the canonical
+  Section 3.4.1 form. Callers embedding the old output in their own Datalog
+  should re-render.
+- Identifiers, scopes and context strings containing a double quote, a backslash
+  or control characters are now rejected with `token_malformed`.
+
 ## [0.3.0] - 2026-05-08
 
 ### Added

python/aip_agents/pyproject.toml

@@ -26,7 +26,7 @@ classifiers = [
     "Topic :: Software Development :: Libraries",
 ]
 dependencies = [
-    "agent-identity-protocol>=0.2.0",
+    "agent-identity-protocol>=0.4.0",
 ]
 
 [project.optional-dependencies]

python/aip_token/canonical.py

@@ -0,0 +1,119 @@
+"""Canonical Simple-profile block encoding (draft-prakash-aip-01 Section 3.4.1).
+
+Every caller that renders AIP Datalog goes through this module. The encoding is
+normative: Section 3.4.1 requires implementations of the Simple profile to emit
+exactly these forms, and two independent renderings of the same intent drift
+into tokens that no other implementation can verify.
+
+Two rules from Section 3.4.1 are easy to get wrong and are enforced here by
+construction:
+
+*   Depth and budget are carried as FACTS, never as checks. A check of the form
+    ``check if budget($b), $b <= N`` binds to whatever budget facts are in scope
+    during evaluation, which is a different question from whether this block's
+    ceiling narrows its parent's. Encoding either as a check therefore rejects
+    valid chains, or passes unconditionally when the verifier supplies a
+    satisfying ambient fact. Both are verified structurally in step V4 instead.
+
+*   A block's scope constraint is ONE self-contained check. A rule-based
+    encoding does not attenuate, because a delegation block's narrower rule is
+    unioned with an ancestor's broader rule of the same name rather than
+    replacing it, and the broader one then satisfies the delegation's own check.
+
+This module deliberately has no dependency on ``biscuit_auth`` so that callers
+needing only the Datalog text do not pull in the optional extra.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from aip_token.error import TokenError
+
+_STR_FACT = '{name}("{value}");\n'
+
+_TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
+
+#: Characters that let a value escape the Datalog string context it is written
+#: into. A double quote closes the string outright. A backslash reprints as an
+#: escape sequence that this implementation's block parsers do not decode. A
+#: control character (newline included) splits one fact across lines, which the
+#: line-oriented parsers in the attenuation walk then read as two facts.
+#:
+#: Nothing else needs excluding. Semicolons, parentheses, brackets, commas,
+#: ``//``, spaces, colons and non-ASCII characters are all inert inside a
+#: quoted term and round-trip unambiguously, so identifiers and free-form
+#: context strings keep working.
+_FORBIDDEN_CHARS = frozenset('"\\')
+
+
+def validate_term(value: str, field: str) -> str:
+    """Reject a string that could escape the Datalog string context.
+
+    AIP block and authorizer sources are Datalog text. A value carrying a
+    double quote can close its own term and append arbitrary facts, checks or
+    policies to a block that is then signed, or to the authorizer that decides
+    a request. Both are authorization bypasses, so this is enforced at every
+    boundary where a caller-supplied string becomes Datalog rather than left to
+    the caller.
+    """
+    if not isinstance(value, str):
+        raise TokenError(f"{field} must be a string", "token_malformed")
+    for char in value:
+        if char in _FORBIDDEN_CHARS or ord(char) < 0x20 or ord(char) == 0x7F:
+            raise TokenError(
+                f"{field} contains a character that is not permitted in an AIP "
+                f"Datalog term: {char!r}",
+                "token_malformed",
+            )
+    return value
+
+
+def str_fact(name: str, value: str) -> str:
+    """Render a string-valued fact, e.g. ``identity("aip:web:example.com");``."""
+    validate_term(value, name)
+    return _STR_FACT.format(name=name, value=value)
+
+
+def int_fact(name: str, value: int) -> str:
+    """Render an integer-valued fact, e.g. ``max_depth(3);``."""
+    return f"{name}({value});\n"
+
+
+def scope_list(scopes: list[str]) -> str:
+    """Render a scope allowlist as a Datalog set literal."""
+    return ", ".join(f'"{s}"' for s in scopes)
+
+
+def tool_check(scopes: list[str]) -> str:
+    """Render a block's scope constraint as one self-contained check.
+
+    An all-exact scope set produces the Simple profile form (spec/aip-tokens.md
+    section 7.1). A set containing a wildcard produces the Standard profile
+    form (section 7.2), where patterns become ``starts_with`` clauses joined to
+    the exact-match clause with ``or``.
+    """
+    for scope in scopes:
+        validate_term(scope, "scope")
+
+    exact = [s for s in scopes if not s.endswith("*")]
+    wildcards = [s for s in scopes if s.endswith("*")]
+
+    clauses = []
+    if exact:
+        clauses.append(f"tool($t), [{scope_list(exact)}].contains($t)")
+    for pattern in wildcards:
+        prefix = pattern[:-1]
+        if prefix:
+            clauses.append(f'tool($t), $t.starts_with("{prefix}")')
+        else:
+            clauses.append("tool($t)")
+
+    if not clauses:
+        return ""
+    return "check if " + " or ".join(clauses) + ";\n"
+
+
+def time_check(expiry: datetime) -> str:
+    """Render a block's expiry constraint."""
+    return f"check if time($t), $t <= {expiry.strftime(_TIME_FORMAT)};\n"

python/aip_token/chained.py

@@ -20,12 +20,20 @@
     BiscuitBuilder,
     BlockBuilder,
     Fact,
+    Policy,
     PrivateKey as BiscuitPrivateKey,
     PublicKey as BiscuitPublicKey,
     Rule,
 )
 
 from aip_core.crypto import KeyPair
+from aip_token.canonical import (
+    int_fact,
+    str_fact,
+    time_check,
+    tool_check,
+    validate_term,
+)
 from aip_token.error import TokenError
 
 # biscuit-python >= 0.4 requires an Algorithm argument for from_bytes();
@@ -53,44 +61,8 @@ def _biscuit_public_key(raw_bytes: bytes) -> BiscuitPublicKey:
 
 
 
-# --- Canonical block encoding (draft-prakash-aip-01 Section 3.4.1) ----------
-
-_STR_FACT = "{name}(\"{value}\");\n"
-
-
-def _scope_list(scopes: list[str]) -> str:
-    """Render a scope allowlist as a Datalog set literal."""
-    return ", ".join(f'"{s}"' for s in scopes)
-
-
-def _tool_check(scopes: list[str]) -> str:
-    """Render a block's scope constraint as one self-contained check.
-
-    An all-exact scope set produces the Simple profile form (spec/aip-tokens.md
-    section 7.1). A set containing a wildcard produces the Standard profile
-    form (section 7.2), where patterns become `starts_with` clauses joined to
-    the exact-match clause with `or`. The check MUST be self-contained: a
-    rule-based encoding
-    does not attenuate, because a delegation block's narrower rule is unioned
-    with the authority block's broader rule of the same name rather than
-    replacing it, and the broader one then satisfies the delegation's check.
-    """
-    exact = [s for s in scopes if not s.endswith("*")]
-    wildcards = [s for s in scopes if s.endswith("*")]
-
-    clauses = []
-    if exact:
-        clauses.append(f"tool($t), [{_scope_list(exact)}].contains($t)")
-    for pattern in wildcards:
-        prefix = pattern[:-1]
-        if prefix:
-            clauses.append(f'tool($t), $t.starts_with("{prefix}")')
-        else:
-            clauses.append("tool($t)")
-
-    if not clauses:
-        return ""
-    return "check if " + " or ".join(clauses) + ";\n"
+# Canonical block encoding lives in aip_token.canonical so that the chained
+# builder and the standalone SimplePolicy renderer cannot drift apart.
 
 
 def _covers(pattern: str, scope: str) -> bool:
@@ -207,17 +179,17 @@ def create_authority(
 
         expiry = datetime.now(tz=timezone.utc) + timedelta(seconds=ttl_seconds)
 
-        facts = _STR_FACT.format(name="identity", value=issuer)
+        facts = str_fact("identity", issuer)
         if principal is not None:
-            facts += _STR_FACT.format(name="principal", value=principal)
+            facts += str_fact("principal", principal)
         for scope in scopes:
-            facts += _STR_FACT.format(name="right", value=scope)
-        facts += f"max_depth({max_depth});\n"
+            facts += str_fact("right", scope)
+        facts += int_fact("max_depth", max_depth)
         if budget_cents is not None:
-            facts += f"budget_ceiling({budget_cents});\n"
+            facts += int_fact("budget_ceiling", budget_cents)
         if scopes:
-            facts += _tool_check(scopes)
-        facts += f'check if time($t), $t <= {expiry.strftime("%Y-%m-%dT%H:%M:%SZ")};\n'
+            facts += tool_check(scopes)
+        facts += time_check(expiry)
 
         biscuit = BiscuitBuilder(facts).build(_biscuit_private_key(keypair))
         return ChainedToken(biscuit, issuer, max_depth, keypair.public_key_bytes(), depth=0)
@@ -265,15 +237,15 @@ def delegate(
                     "budget_exceeded",
                 )
 
-        block = _STR_FACT.format(name="delegator", value=delegator)
-        block += _STR_FACT.format(name="delegate", value=delegate)
-        block += _STR_FACT.format(name="context", value=context)
+        block = str_fact("delegator", delegator)
+        block += str_fact("delegate", delegate)
+        block += str_fact("context", context)
         if principal is not None:
-            block += _STR_FACT.format(name="principal", value=principal)
+            block += str_fact("principal", principal)
         if budget_cents is not None:
-            block += f"budget_ceiling({budget_cents});\n"
+            block += int_fact("budget_ceiling", budget_cents)
         if scopes:
-            block += _tool_check(scopes)
+            block += tool_check(scopes)
 
         new_biscuit = self._biscuit.append(BlockBuilder(block))
         return ChainedToken(
@@ -312,15 +284,26 @@ def authorize(self, tool: str, root_public_key_bytes: bytes) -> None:
         # V6: policy evaluation. Ambient facts are tool, time and depth only.
         # Supplying a budget fact here would make any budget check evaluate
         # against verifier-chosen data; budget is handled in V4.
+        #
+        # The requested tool is the one input to this step that the caller
+        # controls, and in an MCP binding it arrives straight from the agent's
+        # JSON-RPC params. It is validated and then bound as a Datalog
+        # parameter, never interpolated into the source: a tool name that
+        # closed its own string could append `allow if true` to this very
+        # authorizer and pass any check the token carries.
+        validate_term(tool, "tool")
+
         now = datetime.now(tz=timezone.utc)
-        auth_code = (
-            f'tool("{tool}");\n'
-            f'time({now.strftime("%Y-%m-%dT%H:%M:%SZ")});\n'
-            f"depth({self._depth});\n"
-            f'allow if tool("{tool}");\n'
-        )
         try:
-            AuthorizerBuilder(auth_code).build(verified).authorize()
+            builder = AuthorizerBuilder(
+                f'time({now.strftime("%Y-%m-%dT%H:%M:%SZ")});\n'
+                f"depth({self._depth});\n"
+            )
+            builder.add_fact(Fact("tool({t})", {"t": tool}))
+            builder.add_policy(Policy("allow if tool({t})", {"t": tool}))
+            builder.build(verified).authorize()
+        except TokenError:
+            raise
         except Exception as exc:
             raise TokenError(f"insufficient scope: {exc}", "scope_insufficient") from exc
 

python/aip_token/policy.py

@@ -5,10 +5,22 @@
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 
+from aip_token.canonical import int_fact, time_check, tool_check
+
 
 @dataclass
 class SimplePolicy:
-    """A simple policy that generates Datalog checks for Biscuit tokens.
+    """A Simple-profile policy rendered as canonical AIP Datalog.
+
+    Emits the forms mandated by draft-prakash-aip-01 Section 3.4.1, in the
+    order given there: ``max_depth`` and ``budget_ceiling`` as facts, followed
+    by the self-contained tool check and the expiry check.
+
+    Depth and budget are facts, not checks. Section 3.4.1 states that
+    implementations MUST NOT emit a check statement over budget, because such a
+    check binds to whatever budget facts are in scope at evaluation rather than
+    to this block's ceiling. Both dimensions are verified structurally in step
+    V4 of Section 4 instead. See :mod:`aip_token.canonical`.
 
     Budget is expressed in integer cents (Biscuit has no float support).
     """
@@ -19,17 +31,14 @@ class SimplePolicy:
     ttl_seconds: int | None = None
 
     def to_datalog(self) -> str:
-        rules: list[str] = []
-        if self.tools:
-            tool_list = ", ".join(f'"{t}"' for t in self.tools)
-            rules.append(f"check if tool($tool), [{tool_list}].contains($tool);")
-        if self.budget_cents is not None:
-            rules.append(f"check if budget($b), $b <= {self.budget_cents};")
+        parts: list[str] = []
         if self.max_depth is not None:
-            rules.append(f"check if depth($d), $d <= {self.max_depth};")
+            parts.append(int_fact("max_depth", self.max_depth))
+        if self.budget_cents is not None:
+            parts.append(int_fact("budget_ceiling", self.budget_cents))
+        if self.tools:
+            parts.append(tool_check(self.tools))
         if self.ttl_seconds is not None:
             expiry = datetime.now(tz=timezone.utc) + timedelta(seconds=self.ttl_seconds)
-            rules.append(
-                f'check if time($t), $t <= {expiry.strftime("%Y-%m-%dT%H:%M:%SZ")};'
-            )
-        return "\n".join(rules)
+            parts.append(time_check(expiry))
+        return "".join(parts)

python/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "agent-identity-protocol"
-version = "0.3.0"
+version = "0.4.0"
 description = "Verifiable cryptographic identity and delegation for AI agents across MCP and A2A"
 readme = "README.md"
 license = "Apache-2.0"