Check an AI Agent's Identity, Owner and Consent Before It Reads Personal Data

pip install cryptography

"""Check an AI agent's identity, owner and data scope before it reads personal data.

A minimal pattern, not a product. Requires: pip install cryptography
"""
import json
import time
from dataclasses import dataclass

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
    Ed25519PublicKey,
)
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat


def raw_public(key: Ed25519PublicKey) -> str:
    return key.public_bytes(Encoding.Raw, PublicFormat.Raw).hex()


def canonical(data: dict) -> bytes:
    return json.dumps(data, sort_keys=True, separators=(",", ":")).encode()


# 1. The issuer: whoever your company trusts to vouch for agents.
issuer_key = Ed25519PrivateKey.generate()
ISSUER_PUBLIC = issuer_key.public_key()


def issue_credential(agent_public: Ed25519PublicKey, claims: dict) -> dict:
    body = {**claims, "agent_public_key": raw_public(agent_public)}
    return {"body": body, "issuer_sig": issuer_key.sign(canonical(body)).hex()}


# 2. The gate: runs before any personal data leaves your system.
REVOKED_AGENTS = set()


@dataclass
class Decision:
    allowed: bool
    reason: str


def check_request(credential: dict, request: dict, request_sig: str) -> Decision:
    body = credential["body"]

    # Is the credential genuine?
    try:
        ISSUER_PUBLIC.verify(bytes.fromhex(credential["issuer_sig"]), canonical(body))
    except InvalidSignature:
        return Decision(False, "credential not signed by a trusted issuer")

    if body["agent_id"] in REVOKED_AGENTS:
        return Decision(False, "agent credential revoked")
    if body["expires_at"] < time.time():
        return Decision(False, "credential expired")

    # Who is it? The caller must prove it holds the agent's private key.
    agent_public = Ed25519PublicKey.from_public_bytes(
        bytes.fromhex(body["agent_public_key"])
    )
    try:
        agent_public.verify(bytes.fromhex(request_sig), canonical(request))
    except InvalidSignature:
        return Decision(False, "caller does not hold this agent's key")

    # Who owns it? No named legal owner, no personal data.
    if not body.get("owner_legal_entity"):
        return Decision(False, "no accountable legal owner")

    # What may it touch? Purpose and consent must match the request.
    if request["purpose"] not in body["allowed_purposes"]:
        return Decision(False, f"purpose '{request['purpose']}' not in scope")
    if not request.get("consent_ref"):
        return Decision(False, "no consent reference for this personal data")

    return Decision(True, f"{body['agent_id']} acting for {body['owner_legal_entity']}")


def audit(credential: dict, request: dict, decision: Decision) -> str:
    # One line per decision: which agent, which owner, which data, which consent.
    return json.dumps({
        "ts": int(time.time()),
        "agent_id": credential["body"]["agent_id"],
        "owner": credential["body"].get("owner_legal_entity"),
        "purpose": request["purpose"],
        "consent_ref": request.get("consent_ref"),
        "allowed": decision.allowed,
        "reason": decision.reason,
    })


if __name__ == "__main__":
    agent_key = Ed25519PrivateKey.generate()
    cred = issue_credential(agent_key.public_key(), {
        "agent_id": "support-agent-07",
        "owner_legal_entity": "Example Retail Pvt Ltd",
        "dpdp_role": "processor for a Data Fiduciary",
        "allowed_purposes": ["order_support"],
        "expires_at": time.time() + 3600,
    })

    def signed(req, key=agent_key):
        return req, key.sign(canonical(req)).hex()

    cases = {
        "valid request": signed({"purpose": "order_support", "consent_ref": "CM-2026-1188"}),
        "wrong purpose": signed({"purpose": "marketing", "consent_ref": "CM-2026-1188"}),
        "no consent": signed({"purpose": "order_support", "consent_ref": ""}),
        "impostor key": signed({"purpose": "order_support", "consent_ref": "CM-2026-1188"},
                               Ed25519PrivateKey.generate()),
    }
    for name, (req, sig) in cases.items():
        print(f"{name:14} -> {audit(cred, req, check_request(cred, req, sig))}")

    REVOKED_AGENTS.add("support-agent-07")
    req, sig = cases["valid request"]
    print(f"{'after revoke':14} -> {check_request(cred, req, sig).reason}")

valid request  -> {"ts": 1794628987, "agent_id": "support-agent-07", "owner": "Example Retail Pvt Ltd", "purpose": "order_support", "consent_ref": "CM-2026-1188", "allowed": true, "reason": "support-agent-07 acting for Example Retail Pvt Ltd"}
wrong purpose  -> {"ts": 1794628987, "agent_id": "support-agent-07", "owner": "Example Retail Pvt Ltd", "purpose": "marketing", "consent_ref": "CM-2026-1188", "allowed": false, "reason": "purpose 'marketing' not in scope"}
no consent     -> {"ts": 1794628987, "agent_id": "support-agent-07", "owner": "Example Retail Pvt Ltd", "purpose": "order_support", "consent_ref": "", "allowed": false, "reason": "no consent reference for this personal data"}
impostor key   -> {"ts": 1794628987, "agent_id": "support-agent-07", "owner": "Example Retail Pvt Ltd", "purpose": "order_support", "consent_ref": "CM-2026-1188", "allowed": false, "reason": "caller does not hold this agent's key"}
after revoke   -> agent credential revoked