AID-D-001.001: Per-Prompt Content, Intent & Obfuscation Analysis

Version 1.20261015 · Actionable sub-technique

Analyze each inbound prompt for harmful content, prompt injection, jailbreaking, instruction bypass, intent or goal switching, privilege-escalation intent, and obfuscation such as layered encoding, homoglyphs, mixed scripts, or high-entropy payloads. Use a pinned and calibrated detector family appropriate to the modality and language, and emit a synchronous structured finding bound to the detector revision, policy version, scores, categories, and input digest.

Pillars: app

Lifecycle: operation

Scope boundary

Owns synchronous per-prompt findings for harmful content, injection, jailbreak, instruction bypass, intent switching, privilege-escalation intent, and bounded obfuscation analysis. It emits evidence but does not authorize or reject the prompt.

AID-H-002.002: AID-D-001.001 analyzes the prompt and emits calibrated content, intent, and obfuscation findings; AID-H-002.002 consumes those findings and owns sanitize, block, downgrade, review, or admission behavior. A detector verdict is policy input rather than the admission decision.

AID-D-001.003: AID-D-001.001 analyzes explicit content, intent, and recovered obfuscation views; AID-D-001.003 scores statistical distance or density in a governed embedding space. Semantic or lexical policy signals and vector-distribution anomalies are complementary findings with different baselines and failure modes.

AID-D-001.006: AID-D-001.001 produces a finding for the current prompt; AID-D-001.006 correlates prompt-gate verdict telemetry across requests, sessions, and identities. Per-request analysis identifies the immediate signal, while correlation identifies cumulative probing and repeated near misses.

Implementation guidance

AID-D-001.001-G001

Run a pinned, calibrated secondary classifier for prompt content, intent switching, instruction bypass, goal manipulation, and privilege-escalation signals, and emit a structured finding without making the admission decision.
Concept

This detector scores one prompt and publishes evidence. It does not reject the request. Load the exact classifier revision and category thresholds from a versioned deployment policy, preserve a content digest rather than raw prompt text, and return FINDING, NO_FINDING, or ERROR. The consuming AID-H-002.002 policy decides whether an error or finding blocks, sanitizes, downgrades, or routes the request for review.

Build a pinned classifier and evidence contract
from __future__ import annotations

# File: detect/prompt_intent_detector.py

import hashlib
import json
import os
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Literal

from transformers import pipeline


ControlStatus = Literal["PASS", "ERROR"]
FindingStatus = Literal["FINDING", "NO_FINDING", "UNKNOWN"]
MODEL_ID = os.environ["PROMPT_DETECTOR_MODEL_ID"]
MODEL_REVISION = os.environ["PROMPT_DETECTOR_MODEL_REVISION"]
POLICY_PATH = Path(os.environ["PROMPT_DETECTOR_POLICY_PATH"])
classifier = pipeline(
    "text-classification",
    model=MODEL_ID,
    tokenizer=MODEL_ID,
    revision=MODEL_REVISION,
    top_k=None,
    truncation=True,
)


@dataclass(frozen=True)
class PromptFinding:
    schema_version: str
    control_id: str
    request_id: str
    prompt_sha256: str
    control_status: ControlStatus
    finding_status: FindingStatus
    categories: list[str]
    scores: dict[str, float]
    model_id: str
    model_revision: str
    policy_version: str
    error_class: str | None = None


def _load_policy() -> dict:
    policy = json.loads(POLICY_PATH.read_text(encoding="utf-8"))
    required = {"policy_version", "category_thresholds"}
    if set(policy) != required or not isinstance(policy["category_thresholds"], dict):
        raise ValueError("prompt detector policy schema mismatch")
    for category, threshold in policy["category_thresholds"].items():
        if not isinstance(category, str) or not isinstance(threshold, (int, float)):
            raise ValueError("invalid prompt detector threshold")
        if not 0.0 <= float(threshold) <= 1.0:
            raise ValueError("prompt detector threshold outside [0,1]")
    return policy


def _flatten(raw: object) -> list[dict]:
    if isinstance(raw, list) and raw and isinstance(raw[0], list):
        raw = raw[0]
    if not isinstance(raw, list):
        raise ValueError("unexpected classifier result shape")
    return [item for item in raw if isinstance(item, dict)]


def detect_prompt(request_id: str, prompt: str) -> PromptFinding:
    digest = hashlib.sha256(prompt.encode("utf-8")).hexdigest()
    try:
        policy = _load_policy()
        predictions = _flatten(classifier(prompt))
        scores = {
            str(item["label"]): round(float(item["score"]), 6)
            for item in predictions
            if "label" in item and "score" in item
        }
        matched = sorted(
            category
            for category, threshold in policy["category_thresholds"].items()
            if scores.get(category, 0.0) >= float(threshold)
        )
        return PromptFinding(
            schema_version="aidefend.prompt_finding.v1",
            control_id="AID-D-001.001",
            request_id=request_id,
            prompt_sha256=digest,
            control_status="PASS",
            finding_status="FINDING" if matched else "NO_FINDING",
            categories=matched,
            scores=scores,
            model_id=MODEL_ID,
            model_revision=MODEL_REVISION,
            policy_version=policy["policy_version"],
        )
    except Exception as exc:
        return PromptFinding(
            schema_version="aidefend.prompt_finding.v1",
            control_id="AID-D-001.001",
            request_id=request_id,
            prompt_sha256=digest,
            control_status="ERROR",
            finding_status="UNKNOWN",
            categories=[],
            scores={},
            model_id=MODEL_ID,
            model_revision=MODEL_REVISION,
            policy_version="unavailable",
            error_class=type(exc).__name__,
        )


def publish_finding(request_id: str, prompt: str, sink) -> None:
    sink(asdict(detect_prompt(request_id, prompt)))
Promotion and verification

Calibrate category thresholds on a versioned corpus containing benign, harmful, prompt-injection, jailbreak, goal-switching, privilege-escalation, multilingual, and unsupported-language segments. Record per-segment false-positive and false-negative rates, model/revision, policy digest, corpus digest, and shadow-versus-production comparison. Replay known positive, benign, timeout, malformed-output, and unsupported-input cases and verify the exact status and evidence fields.

Action: Publish the finding synchronously before prompt admission. Do not include an allow field and do not log raw prompt text. Configure AID-H-002.002 to consume FINDING, NO_FINDING, and ERROR under its own signed policy.

AID-D-001.001-G002

Recursively decode bounded encoding layers, normalize Unicode, and score every recovered view for obfuscation and prompt-policy signals without logging or executing decoded content.
Concept

Attackers split payloads across Base64, hex, Unicode escapes, mixed scripts, or multiple nested layers. Decode only recognized complete candidates under strict depth, candidate, byte, and time budgets. Analyze both the raw and recovered views, store hashes instead of recovered text, and treat a decoding or detector failure as ERROR rather than a clean result.

Bounded detector
from __future__ import annotations

# File: detect/obfuscation_detector.py

import base64
import binascii
import hashlib
import re
import time
import unicodedata
from collections import deque
from dataclasses import dataclass
from typing import Callable, Literal


BASE64_RE = re.compile(r"(?<![A-Za-z0-9+/])(?:[A-Za-z0-9+/]{4}){4,}(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?(?![A-Za-z0-9+/=])")
HEX_RE = re.compile(r"(?<![0-9a-fA-F])(?:[0-9a-fA-F]{2}){8,}(?![0-9a-fA-F])")
UNICODE_ESCAPE_RE = re.compile(r"(?:\u[0-9a-fA-F]{4}|\x[0-9a-fA-F]{2}){4,}")


@dataclass(frozen=True)
class DetectorPolicy:
    policy_version: str
    max_input_chars: int
    max_depth: int
    max_candidates: int
    max_decoded_bytes: int
    max_elapsed_ms: int
    finding_score: int


def _decode(candidate: str, encoding: str) -> str | None:
    try:
        if encoding == "base64":
            raw = base64.b64decode(candidate, validate=True)
            return raw.decode("utf-8")
        if encoding == "hex":
            return binascii.unhexlify(candidate).decode("utf-8")
        if encoding == "unicode_escape":
            return candidate.encode("ascii").decode("unicode_escape")
    except (UnicodeError, binascii.Error, ValueError):
        return None
    return None


def _candidates(text: str) -> list[tuple[str, str]]:
    found: list[tuple[str, str]] = []
    for encoding, pattern in (
        ("base64", BASE64_RE),
        ("hex", HEX_RE),
        ("unicode_escape", UNICODE_ESCAPE_RE),
    ):
        found.extend((encoding, match.group(0)) for match in pattern.finditer(text))
    return found


def detect_obfuscation(
    prompt: str,
    request_id: str,
    policy: DetectorPolicy,
    content_detector: Callable[[str], dict],
) -> dict:
    started = time.monotonic()
    if not isinstance(prompt, str) or len(prompt) > policy.max_input_chars:
        return {
            "control_id": "AID-D-001.001",
            "request_id": request_id,
            "control_status": "ERROR",
            "finding_status": "UNKNOWN",
            "reason": "unsupported_or_oversized_input",
            "policy_version": policy.policy_version,
        }

    queue = deque([(unicodedata.normalize("NFKC", prompt), 0, "raw")])
    seen: set[str] = set()
    evidence: list[dict] = []
    decoded_bytes = 0
    score = 0

    try:
        while queue:
            if (time.monotonic() - started) * 1000 > policy.max_elapsed_ms:
                raise TimeoutError("obfuscation budget exceeded")
            text, depth, encoding = queue.popleft()
            if text in seen:
                continue
            seen.add(text)
            detector_result = content_detector(text)
            if (
                not isinstance(detector_result, dict)
                or detector_result.get("control_status") != "PASS"
                or detector_result.get("finding_status")
                not in {"FINDING", "NO_FINDING"}
            ):
                raise RuntimeError("content detector dependency failed")
            if detector_result["finding_status"] == "FINDING":
                score += 40
            if encoding != "raw":
                score += 15
            evidence.append({
                "encoding": encoding,
                "depth": depth,
                "content_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
                "content_detector_control_status": detector_result["control_status"],
                "content_detector_finding_status": detector_result["finding_status"],
            })

            if depth >= policy.max_depth:
                continue
            for child_encoding, candidate in _candidates(text):
                if len(evidence) + len(queue) >= policy.max_candidates:
                    break
                decoded = _decode(candidate, child_encoding)
                if decoded is None:
                    continue
                decoded = unicodedata.normalize("NFKC", decoded)
                decoded_bytes += len(decoded.encode("utf-8"))
                if decoded_bytes > policy.max_decoded_bytes:
                    raise ValueError("decoded byte budget exceeded")
                queue.append((decoded, depth + 1, child_encoding))

        return {
            "control_id": "AID-D-001.001",
            "request_id": request_id,
            "control_status": "PASS",
            "finding_status": "FINDING" if score >= policy.finding_score else "NO_FINDING",
            "score": score,
            "policy_version": policy.policy_version,
            "views": evidence,
        }
    except Exception as exc:
        return {
            "control_id": "AID-D-001.001",
            "request_id": request_id,
            "control_status": "ERROR",
            "finding_status": "UNKNOWN",
            "reason": type(exc).__name__,
            "policy_version": policy.policy_version,
            "views": evidence,
        }
Verification and handoff

Replay nested Base64, hex, Unicode escapes, homoglyph normalization, oversized input, candidate explosion, timeout, detector outage, benign encoded data, and true injection cases. Verify resource limits, status semantics, evidence hashes, and that no decoded payload reaches logs or execution. Load the policy values from a signed versioned artifact.

Action: Publish the evidence before context assembly. AID-H-002.002 owns the fail-closed admission response; this detector never emits allow or executes decoded content.

Open-source tools

Source-available / open-weight tools

Commercial / hosted 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.001: Per-Prompt Content, Intent & Obfuscation Analysis | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service