AID-D-001.004: Active Prompt Integrity Check (Canary Tokens)

Version 1.20261015 · Actionable sub-technique

Issue a fresh, single-use prompt-integrity challenge in the system instruction and require a schema-constrained response to return it in a backend-only field that is stripped before release. Missing, mismatched, duplicated, expired, or visibly leaked challenge values are useful prompt-override and format-breakage findings. The challenge is visible to the model and is therefore not a secret, signature, authorization proof, or guarantee that the system prompt was followed; an injected instruction may copy it successfully. Treat this as a defense-in-depth detector and correlate it with content, session, and gate telemetry.

Pillars: app

Lifecycle: operation

Scope boundary

Owns single-use prompt-integrity challenge issuance and evaluation as a heuristic finding for prompt override, response-format breakage, or challenge leakage. The model-visible challenge is not a secret, signature, authorization proof, or guarantee of instruction compliance.

AID-H-006.001: AID-D-001.004 reports whether the isolated canary field is missing, changed, duplicated, expired, or leaked; AID-H-006.001 enforces the required output schema and rejects structurally invalid output. A returned canary is only a heuristic integrity signal, while schema enforcement is the deterministic release boundary.

AID-D-001.006: AID-D-001.004 creates a request-scoped integrity finding; AID-D-001.006 correlates gate and detector decisions over repeated requests, sessions, and identities. The canary checks one execution, while correlation identifies systematic probing or cumulative risk.

Implementation guidance

AID-D-001.004-G001

Issue a per-request prompt-integrity challenge and evaluate the isolated structured response field as a heuristic finding.
Detector boundary

Use this detector only on model and route versions whose signed capability profile supports strict structured output with a backend-only field. A returned challenge is a heuristic continuity signal, not cryptographic attestation: the model sees the value and an attacker may preserve or copy it. The detector emits finding_status; a separate Hardening control decides whether to block, retry, downgrade, or review.

Issue and verify a single-use challenge
from __future__ import annotations

# File: detection/prompt_integrity_challenge.py

import hashlib
import hmac
import json
import secrets
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Literal
from uuid import UUID

ControlStatus = Literal["PASS", "FAIL", "INSUFFICIENT_DATA", "NOT_APPLICABLE", "ERROR"]
FindingStatus = Literal["FINDING", "NO_FINDING", "UNKNOWN"]


@dataclass(frozen=True)
class Challenge:
    request_id: str
    value: str
    value_sha256: str
    issued_at: str
    expires_at: str


@dataclass(frozen=True)
class Result:
    schema_version: str
    control: str
    request_id: str
    control_status: ControlStatus
    finding_status: FindingStatus
    reason: str
    challenge_sha256: str
    response_sha256: str
    policy_digest: str
    capability_digest: str


def issue(request_id: str, ttl_seconds: int, now: datetime | None = None) -> Challenge:
    UUID(request_id)
    if ttl_seconds < 1:
        raise ValueError("ttl must come from signed policy and be positive")
    current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
    value = secrets.token_urlsafe(24)
    return Challenge(
        request_id=request_id,
        value=value,
        value_sha256=hashlib.sha256(value.encode("utf-8")).hexdigest(),
        issued_at=current.isoformat(),
        expires_at=(current + timedelta(seconds=ttl_seconds)).isoformat(),
    )


def system_fragment(challenge: Challenge) -> str:
    return (
        "Return exactly two JSON fields: answer and _integrity_challenge. "
        "Set _integrity_challenge to this single-use value: "
        + challenge.value
        + ". Never place the value inside answer."
    )


def evaluate(
    response_bytes: bytes,
    challenge: Challenge,
    *,
    policy_digest: str,
    capability_digest: str,
    now: datetime | None = None,
) -> Result:
    current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
    response_sha256 = hashlib.sha256(response_bytes).hexdigest()
    expires = datetime.fromisoformat(challenge.expires_at)
    try:
        parsed = json.loads(response_bytes.decode("utf-8", errors="strict"))
        exact_schema = (
            isinstance(parsed, dict)
            and set(parsed) == {"answer", "_integrity_challenge"}
            and isinstance(parsed.get("answer"), str)
            and isinstance(parsed.get("_integrity_challenge"), str)
        )
        leaked = exact_schema and challenge.value in parsed["answer"]
        matched = exact_schema and hmac.compare_digest(
            parsed["_integrity_challenge"], challenge.value
        )
        if current > expires:
            finding, reason = "FINDING", "expired challenge returned"
        elif not exact_schema:
            finding, reason = "FINDING", "structured response contract broken"
        elif leaked:
            finding, reason = "FINDING", "challenge leaked into visible answer"
        elif not matched:
            finding, reason = "FINDING", "challenge missing or mismatched"
        else:
            finding, reason = "NO_FINDING", "challenge matched and stayed isolated"
        return Result(
            schema_version="aidefend.d001004-result.v1",
            control="AID-D-001.004",
            request_id=challenge.request_id,
            control_status="PASS",
            finding_status=finding,
            reason=reason,
            challenge_sha256=challenge.value_sha256,
            response_sha256=response_sha256,
            policy_digest=policy_digest,
            capability_digest=capability_digest,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        return Result(
            schema_version="aidefend.d001004-result.v1",
            control="AID-D-001.004",
            request_id=challenge.request_id,
            control_status="PASS",
            finding_status="FINDING",
            reason="response was not valid UTF-8 JSON: " + type(exc).__name__,
            challenge_sha256=challenge.value_sha256,
            response_sha256=response_sha256,
            policy_digest=policy_digest,
            capability_digest=capability_digest,
        )
Verify complete coverage

Bind request IDs, issued challenges, response IDs, and terminal outcomes in a source-authenticated completeness manifest; duplicate, missing, timed-out, or uncorrelated requests remain in the denominator.

Verify safely

An independent verifier replays signed benign, direct-injection, indirect-injection, copied-challenge, missing-field, leaked-field, invalid-JSON, timeout, duplicate, and unsupported-capability fixtures through the real gateway path, reads evidence back, and compares canonical result digests.

Open-source tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-D-001.004: Active Prompt Integrity Check (Canary Tokens) | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service