AID-D-003.007: Sensitive Attribute Inference Detection & Escalation

Version 1.20261015 · Actionable sub-technique

Detect model outputs that derive or infer protected sensitive attributes such as political affiliation, religious beliefs, sexual orientation, gender identity, health conditions, disability status, genetic information, trade union membership, ethnic origin, or immigration status from combinations of non-sensitive or semi-sensitive signals, then emit review, suppression-routing, or policy-escalation findings for downstream controls.

Sensitive-attribute inference may be accurate and still violate purpose limitation, anti-discrimination, or special-category data handling rules. Coverage includes protected-attribute inference testing before deployment, runtime inference findings, multi-turn context-combination scoring, and cross-tool attribute-derivation findings.

Pillars: data, app

Lifecycle: operation

Scope boundary

Owns findings for protected sensitive attributes inferred from combinations of non-sensitive or semi-sensitive signals, including multi-turn and cross-tool derivation. It does not suppress, redact, block, define use-case policy, or perform pre-release fairness evaluation.

AID-D-003.002: AID-D-003.007 detects inferential derivation of protected attributes; AID-D-003.002 detects direct exposure of PII, secrets, private records, or memorized data. An inferred conclusion may contain no leaked sensitive token and therefore needs a distinct control path.

AID-D-003.005: AID-D-003.007 owns the category-specific sensitive-inference finding; AID-D-003.005 owns the broader multi-turn context, intent-drift, and invariant-breach ledger. Session risk supplies accumulated evidence but does not by itself classify the output as a protected-attribute inference.

AID-M-007.003: AID-D-003.007 detects sensitive-attribute inference during operation; AID-M-007.003 evaluates pre-release protected-group outcome fairness and disparity over a governed population. Inference-time privacy findings and population-level fairness metrics address different harms and evidence sets.

Warning

Regulatory & Ethical Critical

Accurate sensitive-attribute inference can still violate purpose limitation, anti-discrimination, or special-category data rules.

Implementation guidance

AID-D-003.007-G001

Deploy a protected-attribute inference detector in the output guardrail pipeline and emit structured findings when generated text appears to infer protected attributes from non-sensitive signals.
Concept:

This is a detect-side classifier that evaluates generated text for inferred protected attributes, such as health, pregnancy, religion, ethnicity, or political affiliation. It does not suppress, rewrite, or block the answer itself. Its output is a structured finding that downstream response or hardening controls can use to decide whether to redact, escalate, or require human review.

Step 1: Load a real detector with category-specific thresholds
from __future__ import annotations

# File: detect/protected_attribute_detector.py

import hashlib as _aidefend_policy_hashlib
import hmac as _aidefend_policy_hmac
import json as _aidefend_policy_json
import math as _aidefend_policy_math
import os as _aidefend_policy_os
import re as _aidefend_policy_re
import subprocess as _aidefend_policy_subprocess
import tempfile as _aidefend_policy_tempfile
from pathlib import Path as _AidefendPolicyPath


def _aidefend_write_private(path, payload):
    descriptor = _aidefend_policy_os.open(
        path,
        _aidefend_policy_os.O_WRONLY
        | _aidefend_policy_os.O_CREAT
        | _aidefend_policy_os.O_EXCL,
        0o600,
    )
    with _aidefend_policy_os.fdopen(descriptor, "wb") as stream:
        stream.write(payload)
        stream.flush()
        _aidefend_policy_os.fsync(stream.fileno())
    _aidefend_policy_os.chmod(path, 0o400)


def _aidefend_strict_json(raw):
    def reject_duplicate_keys(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError("duplicate JSON key: " + key)
            value[key] = item
        return value

    return _aidefend_policy_json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=reject_duplicate_keys,
        parse_constant=lambda value: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + value)
        ),
    )


def _aidefend_load_signed_policy(*, path_env, signature_env, key_env, schema_version):
    policy_path = _AidefendPolicyPath(_aidefend_policy_os.environ[path_env])
    signature_path = _AidefendPolicyPath(_aidefend_policy_os.environ[signature_env])
    if policy_path == signature_path:
        raise RuntimeError("policy and signature paths must differ")
    policy_raw = policy_path.read_bytes()
    signature_raw = signature_path.read_bytes()
    key_reference = _aidefend_policy_os.environ[key_env]
    remote_key = bool(_aidefend_policy_re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*:(?://)?", key_reference
    )) and not key_reference.startswith("-----BEGIN ")
    key_raw = None if remote_key else (
        key_reference.encode("utf-8")
        if key_reference.startswith("-----BEGIN ")
        else _AidefendPolicyPath(key_reference).read_bytes()
    )
    verify_timeout = float(_aidefend_policy_os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not _aidefend_policy_math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("policy-verification timeout from the versioned runtime profile is invalid")
    with _aidefend_policy_tempfile.TemporaryDirectory(
        prefix="aidefend-signed-policy-"
    ) as directory:
        root = _AidefendPolicyPath(directory)
        _aidefend_policy_os.chmod(root, 0o700)
        policy_snapshot = root / "policy.json"
        signature_snapshot = root / "policy.sig"
        _aidefend_write_private(policy_snapshot, policy_raw)
        _aidefend_write_private(signature_snapshot, signature_raw)
        captured = [(policy_snapshot, policy_raw), (signature_snapshot, signature_raw)]
        key_argument = key_reference
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_write_private(key_snapshot, key_raw)
            captured.append((key_snapshot, key_raw))
            key_argument = str(key_snapshot)
        try:
            _aidefend_policy_subprocess.run(
                [
                    "cosign", "verify-blob", "--key", key_argument,
                    "--bundle", str(signature_snapshot), str(policy_snapshot),
                ],
                check=True, capture_output=True, text=True, timeout=verify_timeout,
            )
            if any(
                not _aidefend_policy_hmac.compare_digest(path.read_bytes(), expected)
                for path, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
            policy = _aidefend_strict_json(policy_raw)
        finally:
            for path, _ in captured:
                _aidefend_policy_os.chmod(path, 0o600)
    if (
        not isinstance(policy, dict)
        or policy.get("schema_version") != schema_version
        or not isinstance(policy.get("policy_version"), str)
        or not policy["policy_version"]
    ):
        raise RuntimeError("signature-verified policy schema or version is invalid")
    return policy, _aidefend_policy_hashlib.sha256(policy_raw).hexdigest()


from transformers import pipeline

PROTECTED_ATTRIBUTE_POLICY, POLICY_SHA256 = _aidefend_load_signed_policy(
    path_env="D003007_POLICY_PATH",
    signature_env="D003007_POLICY_SIGNATURE",
    key_env="D003007_POLICY_VERIFY_KEY",
    schema_version="aidefend.protected-attribute-detection-policy.v1",
)
MODEL_ID = PROTECTED_ATTRIBUTE_POLICY.get("model_id")
MODEL_REVISION = PROTECTED_ATTRIBUTE_POLICY.get("model_revision")
POLICY_VERSION = PROTECTED_ATTRIBUTE_POLICY["policy_version"]
MAX_TOKEN_LENGTH = PROTECTED_ATTRIBUTE_POLICY.get("max_token_length")
CATEGORY_PROMPTS = PROTECTED_ATTRIBUTE_POLICY.get("category_prompts")
THRESHOLDS = PROTECTED_ATTRIBUTE_POLICY.get("category_thresholds")
if (
    not isinstance(MODEL_ID, str) or not MODEL_ID
    or not isinstance(MODEL_REVISION, str)
    or _aidefend_policy_re.fullmatch(r"[0-9a-f]{40}", MODEL_REVISION) is None
    or isinstance(MAX_TOKEN_LENGTH, bool) or not isinstance(MAX_TOKEN_LENGTH, int)
    or MAX_TOKEN_LENGTH < 1
    or not isinstance(CATEGORY_PROMPTS, dict) or not CATEGORY_PROMPTS
    or not isinstance(THRESHOLDS, dict) or set(THRESHOLDS) != set(CATEGORY_PROMPTS)
    or any(not isinstance(key, str) or not key for key in CATEGORY_PROMPTS)
    or any(not isinstance(value, str) or not value for value in CATEGORY_PROMPTS.values())
    or any(isinstance(value, bool) or not isinstance(value, (int, float))
           or not 0 <= float(value) <= 1 for value in THRESHOLDS.values())
):
    raise RuntimeError("signed protected-attribute detector policy is invalid")

classifier = pipeline(
    task="zero-shot-classification",
    model=MODEL_ID,
    revision=MODEL_REVISION,
    trust_remote_code=False,
    truncation=True,
    max_length=MAX_TOKEN_LENGTH,
)


def score_output(answer_text: str) -> list[dict[str, float | str]]:
    result = classifier(
        answer_text,
        candidate_labels=list(CATEGORY_PROMPTS.values()),
        multi_label=True,
    )
    labels, scores = result["labels"], result["scores"]
    if (len(labels) != len(scores) or len(labels) != len(CATEGORY_PROMPTS)
            or len(set(labels)) != len(labels)
            or set(labels) != set(CATEGORY_PROMPTS.values())
            or any(isinstance(v, bool) or not isinstance(v, (int, float))
                   or not _aidefend_policy_math.isfinite(v) or not 0 <= v <= 1 for v in scores)):
        raise ValueError("incomplete or invalid detector category scores")
    scores_by_prompt = {
        label: float(score)
        for label, score in zip(result["labels"], result["scores"])
    }

    hits = []
    for category, prompt in CATEGORY_PROMPTS.items():
        score = scores_by_prompt.get(prompt, 0.0)
        if score >= THRESHOLDS[category]:
            hits.append({"label": category, "score": round(score, 4)})
    return hits
Step 2: Emit a finding record instead of mutating the model output
from __future__ import annotations

# File: detect/protected_attribute_findings.py

import hashlib
from datetime import datetime, timezone

from protected_attribute_detector import (
    MODEL_ID, MODEL_REVISION, POLICY_VERSION, POLICY_SHA256, score_output,
)


def evaluate_answer(request_id: str, session_id: str, answer_text: str) -> dict[str, object]:
    if not all(isinstance(v, str) and v for v in (request_id, session_id)) or not isinstance(answer_text, str):
        raise ValueError("server-bound request, session and exact answer text required")
    binding = {"policy_version": POLICY_VERSION, "policy_sha256": POLICY_SHA256,
               "detector_model": MODEL_ID, "detector_revision": MODEL_REVISION,
               "answer_sha256": hashlib.sha256(answer_text.encode("utf-8")).hexdigest()}
    try:
        hits = score_output(answer_text)
    except Exception as error:
        return {**binding, "schema_version": "aidefend.protected-inference-finding.v2",
                "event_type": "protected_attribute_inference_finding",
                "request_id": request_id, "session_id": session_id,
                "ts": datetime.now(timezone.utc).isoformat(),
                "coverage": "ERROR", "finding_status": "UNKNOWN",
                "detected": None, "risk_score": None, "categories": [],
                "recommended_action": "detector_error_review", "reason_code": type(error).__name__}
    max_score = max((hit["score"] for hit in hits), default=0.0)
    return {
        **binding,
        "schema_version": "aidefend.protected-inference-finding.v2",
        "coverage": "PASS",
        "finding_status": "FINDING" if hits else "NO_FINDING",
        "event_type": "protected_attribute_inference_finding",
        "ts": datetime.now(timezone.utc).isoformat(),
        "request_id": request_id,
        "session_id": session_id,
        "detected": bool(hits),
        "risk_score": round(max_score, 4),
        "categories": sorted(hit["label"] for hit in hits),
        "recommended_action": "review_or_redact" if hits else "none",
    }

Action: Run this detector after candidate generation and before user delivery, bind the finding to the exact answer digest, request/session, signature-verified policy bytes, and immutable detector revision, persist it, and hand the result to a separate output policy or HITL control. Keeping the detector and the enforcement path separate gives AIDEFEND clean evidence boundaries for “can detect inferred sensitive disclosures” versus “can actively suppress them.”

Verify safely

An independent verifier must replay signed explicit-inference, implicit-inference, benign, quotation, negation, multilingual, ambiguous, unsupported, and detector-error fixtures and compare category scores, evidence spans, and findings, read evidence back, verify signatures and digests, and compare the canonical result digest.

AID-D-003.007-G002

Run proactive protected-attribute inference tests before deployment and on every model or prompt update, using structured multi-turn scenarios that try to elicit sensitive profiling from non-sensitive inputs.
Load a signed, non-empty scenario manifest

The suite loader verifies the manifest signature and returns its canonical bytes and parsed cases. The runner recomputes the digest, rejects an empty or duplicate population, preserves each typed multi-turn role, and distinguishes a security regression from a model or detector execution error.

Run the complete suite through the real path
from __future__ import annotations

# File: evals/protected_inference_suite.py

import hashlib
from collections.abc import Callable
from dataclasses import dataclass

InvokeModel = Callable[[list[dict]], dict]
EvaluateOutput = Callable[[str, str], dict]
ALLOWED_ROLES = {"system", "developer", "user", "assistant", "tool"}
EXPECTED_FINDINGS = {"FINDING", "NO_FINDING"}


@dataclass(frozen=True)
class VerifiedSuite:
    canonical_bytes: bytes
    cases: tuple[dict, ...]


def run_suite(
    suite: VerifiedSuite,
    *,
    invoke_model: InvokeModel,
    evaluate_output: EvaluateOutput,
    model_revision: str,
    detector_revision: str,
    expected_suite_sha256: str,
) -> dict:
    observed_suite_sha256 = hashlib.sha256(suite.canonical_bytes).hexdigest()
    if observed_suite_sha256 != expected_suite_sha256:
        raise ValueError("suite bytes differ from the signed deployment manifest")
    cases = list(suite.cases)
    if not cases:
        raise ValueError("protected-inference suite is empty")
    scenario_ids = [case.get("scenario_id") for case in cases]
    if any(not isinstance(value, str) or not value for value in scenario_ids):
        raise ValueError("scenario ID is absent")
    if len(scenario_ids) != len(set(scenario_ids)):
        raise ValueError("scenario ID is duplicated")

    results = []
    execution_errors = []
    for case in cases:
        required = {"scenario_id", "target", "turns", "expected_finding_status"}
        if set(case) != required or case["expected_finding_status"] not in EXPECTED_FINDINGS:
            raise ValueError("scenario schema or expected result is invalid")
        if not isinstance(case["target"], str) or not case["target"].strip():
            raise ValueError("scenario target is absent")
        if not isinstance(case["turns"], list) or not case["turns"]:
            raise ValueError("scenario turn population is empty")
        messages = []
        for turn in case["turns"]:
            if (
                not isinstance(turn, dict)
                or set(turn) != {"role", "content"}
                or turn["role"] not in ALLOWED_ROLES
                or not isinstance(turn["content"], str)
                or not turn["content"]
            ):
                raise ValueError("typed multi-turn message is invalid")
            messages.append({"role": turn["role"], "content": turn["content"]})

        try:
            response = invoke_model(messages)
            if set(response) != {"response_id", "text", "model_revision"}:
                raise ValueError("model adapter response schema mismatch")
            if response["model_revision"] != model_revision:
                raise ValueError("model revision drifted during the suite")
            response_sha256 = hashlib.sha256(response["text"].encode("utf-8")).hexdigest()
            detector = evaluate_output(response["response_id"], response["text"])
            evaluated_targets = detector.get("evaluated_targets")
            finding_targets = detector.get("finding_targets")
            if (
                detector.get("detector_revision") != detector_revision
                or detector.get("response_sha256") != response_sha256
                or detector.get("control_status") != "PASS"
                or detector.get("finding_status") not in EXPECTED_FINDINGS
                or not isinstance(evaluated_targets, list)
                or len(evaluated_targets) != len(set(evaluated_targets))
                or any(not isinstance(item, str) or not item for item in evaluated_targets)
                or not isinstance(finding_targets, list)
                or len(finding_targets) != len(set(finding_targets))
                or any(item not in evaluated_targets for item in finding_targets)
                or case["target"] not in evaluated_targets
                or detector["finding_status"] != ("FINDING" if finding_targets else "NO_FINDING")
            ):
                raise ValueError("detector receipt is erroneous, incomplete, or bound to another response")
            observed_target_status = (
                "FINDING" if case["target"] in finding_targets else "NO_FINDING"
            )
            passed = observed_target_status == case["expected_finding_status"]
            results.append({
                "scenario_id": case["scenario_id"],
                "target": case["target"],
                "response_id": response["response_id"],
                "response_sha256": response_sha256,
                "observed_finding_status": observed_target_status,
                "all_finding_targets": finding_targets,
                "expected_finding_status": case["expected_finding_status"],
                "passed": passed,
            })
        except Exception as error:
            execution_errors.append({
                "scenario_id": case["scenario_id"],
                "error_class": type(error).__name__,
            })

    if execution_errors:
        control_status, finding_status = "ERROR", "UNKNOWN"
    else:
        failed = [item["scenario_id"] for item in results if not item["passed"]]
        control_status = "PASS"
        finding_status = "FINDING" if failed else "NO_FINDING"
    return {
        "schema_version": "aidefend.protected_inference_suite.v2",
        "control_status": control_status,
        "finding_status": finding_status,
        "suite_sha256": observed_suite_sha256,
        "model_revision": model_revision,
        "detector_revision": detector_revision,
        "expected_scenario_count": len(cases),
        "evaluated_scenario_count": len(results),
        "execution_errors": execution_errors,
        "results": results,
    }
Verify and retain the result

Canonicalize and sign the scenario manifest before the verified loader constructs VerifiedSuite. Replay safe, direct, indirect, hedged, quoted, negated, multilingual, multi-turn, timeout, abstention, malformed-result, wrong-role, empty-suite, revision-mismatch, target-mismatch, and unrelated-finding fixtures. A separate runner re-verifies the manifest and reconciles every scenario ID and result. Retain response digests rather than raw sensitive outputs in the general report.

AID-D-003.007-G003

Maintain a session-level context-combination risk score and emit threshold-crossing findings when accumulated signals make sensitive inference feasible.
Bind scoring to a signed expected-turn manifest

Keep raw sensitive text inside the authorized category detector. Before scoring a closed window, load a signature-verified manifest from the authoritative conversation ledger. It binds tenant, session, window, exact expected turn IDs, count, and population root. Every pinned category-detector receipt binds the same identity, turn ID, event time, detector revision, content digest, and category hits. Missing turns therefore remain visible instead of disappearing from a process-local counter.

from __future__ import annotations

# File: detect/context_combination_risk.py

import hashlib
import json
import math
import re
from dataclasses import dataclass
from typing import Mapping, Sequence

HEX64 = re.compile(r"^[0-9a-f]{64}$")


@dataclass(frozen=True)
class VerifiedTurnManifest:
    tenant_id: str
    session_id: str
    window_start_unix: float
    window_end_unix: float
    expected_turn_ids: tuple[str, ...]
    expected_turn_count: int
    expected_turn_root_sha256: str
    manifest_sha256: str


@dataclass(frozen=True)
class VerifiedCategoryReceipt:
    tenant_id: str
    session_id: str
    turn_id: str
    observed_at_unix: float
    detector_revision: str
    content_hmac_sha256: str
    categories: tuple[str, ...]
    receipt_sha256: str


@dataclass(frozen=True)
class ContextRiskPolicy:
    policy_version: str
    detector_revision: str
    category_weights: Mapping[str, float]
    combination_weights: Mapping[tuple[str, ...], float]
    decay_half_life_seconds: float
    combination_activity_floor: float
    finding_threshold: float


def population_root(turn_ids: Sequence[str]) -> str:
    raw = json.dumps(sorted(turn_ids), ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()


def evaluate_session(*, tenant_id: str, session_id: str,
                     manifest: VerifiedTurnManifest,
                     receipts: Sequence[VerifiedCategoryReceipt],
                     policy: ContextRiskPolicy, as_of_unix: float) -> dict:
    expected_ids = tuple(manifest.expected_turn_ids)
    if (
        not tenant_id or not session_id or not policy.policy_version
        or manifest.tenant_id != tenant_id or manifest.session_id != session_id
        or not math.isfinite(manifest.window_start_unix)
        or not math.isfinite(manifest.window_end_unix)
        or manifest.window_start_unix >= manifest.window_end_unix
        or manifest.window_end_unix > as_of_unix
        or isinstance(manifest.expected_turn_count, bool)
        or manifest.expected_turn_count < 1
        or manifest.expected_turn_count != len(expected_ids)
        or len(expected_ids) != len(set(expected_ids))
        or any(not isinstance(item, str) or not item for item in expected_ids)
        or HEX64.fullmatch(manifest.expected_turn_root_sha256) is None
        or HEX64.fullmatch(manifest.manifest_sha256) is None
        or population_root(expected_ids) != manifest.expected_turn_root_sha256
    ):
        raise ValueError("verified expected-turn manifest is malformed or misbound")

    half_life = float(policy.decay_half_life_seconds)
    activity_floor = float(policy.combination_activity_floor)
    threshold = float(policy.finding_threshold)
    weights = {str(name): float(value) for name, value in policy.category_weights.items()}
    if (
        not math.isfinite(as_of_unix) or not math.isfinite(half_life)
        or not math.isfinite(activity_floor) or not math.isfinite(threshold)
        or half_life <= 0 or activity_floor <= 0 or threshold < 0
        or not weights or any(not name or not math.isfinite(value) or value < 0
                              for name, value in weights.items())
    ):
        raise ValueError("signed scoring policy is invalid")

    seen_turns = set()
    ordered = sorted(receipts, key=lambda item: (item.observed_at_unix, item.turn_id))
    for receipt in ordered:
        if (
            receipt.tenant_id != tenant_id or receipt.session_id != session_id
            or receipt.detector_revision != policy.detector_revision
            or not receipt.turn_id or receipt.turn_id in seen_turns
            or HEX64.fullmatch(receipt.content_hmac_sha256) is None
            or HEX64.fullmatch(receipt.receipt_sha256) is None
            or not math.isfinite(receipt.observed_at_unix)
            or not manifest.window_start_unix <= receipt.observed_at_unix < manifest.window_end_unix
            or len(receipt.categories) != len(set(receipt.categories))
            or set(receipt.categories) - set(weights)
        ):
            raise ValueError("category receipt is duplicated, malformed, or misbound")
        seen_turns.add(receipt.turn_id)

    expected_set = set(expected_ids)
    missing = sorted(expected_set - seen_turns)
    unexpected = sorted(seen_turns - expected_set)
    observed_root = population_root(tuple(seen_turns))
    population_evidence = {
        "manifest_sha256": manifest.manifest_sha256,
        "expected_turn_count": manifest.expected_turn_count,
        "observed_turn_count": len(seen_turns),
        "expected_turn_root_sha256": manifest.expected_turn_root_sha256,
        "observed_turn_root_sha256": observed_root,
    }
    if missing or unexpected:
        return {
            "schema_version": "aidefend.context_combination_finding.v3",
            "control_status": "INSUFFICIENT_DATA", "finding_status": "UNKNOWN",
            "tenant_id": tenant_id, "session_id": session_id,
            "policy_version": policy.policy_version, **population_evidence,
            "missing_turn_ids": missing, "unexpected_turn_ids": unexpected,
            "reason": "expected and observed turn populations do not reconcile",
        }

    scores = {name: 0.0 for name in weights}
    for receipt in ordered:
        decay = 0.5 ** ((as_of_unix - receipt.observed_at_unix) / half_life)
        for category in receipt.categories:
            scores[category] += weights[category] * decay
    active = {name for name, score in scores.items() if score >= activity_floor}
    hits, combination_score, seen_combinations = [], 0.0, set()
    for members, raw_weight in policy.combination_weights.items():
        member_tuple, weight = tuple(sorted(members)), float(raw_weight)
        if (
            not member_tuple or len(member_tuple) != len(set(member_tuple))
            or set(member_tuple) - set(weights) or not math.isfinite(weight)
            or weight < 0 or member_tuple in seen_combinations
        ):
            raise ValueError("signed category combination is invalid or duplicated")
        seen_combinations.add(member_tuple)
        if set(member_tuple).issubset(active):
            hits.append(member_tuple)
            combination_score += weight
    risk_score = sum(scores.values()) + combination_score
    return {
        "schema_version": "aidefend.context_combination_finding.v3",
        "control_status": "PASS",
        "finding_status": "FINDING" if risk_score >= threshold else "NO_FINDING",
        "tenant_id": tenant_id, "session_id": session_id,
        "policy_version": policy.policy_version,
        "detector_revision": policy.detector_revision,
        **population_evidence, "population_reconciled": True,
        "as_of_unix": as_of_unix,
        "category_scores": {key: round(value, 6) for key, value in scores.items()},
        "combination_hits": hits, "risk_score": round(risk_score, 6),
        "threshold": threshold,
    }
Verification and privacy boundary

The manifest verifier, category-receipt verifier, and signed-policy loader are startup dependencies, not request-provided callbacks. Convert malformed signed inputs or verifier failures to ERROR at the runner boundary. Rebuild each result from the append-only conversation ledger, independently regenerate the expected-turn manifest, and compare exact IDs, count, population root, session binding, detector revision, decay time, and policy version. Replay one missing middle turn, several missing turns, an unexpected turn, duplicate, late, cross-tenant, unsupported-category, restart, threshold-boundary, multilingual, and detector-error fixtures. No partial receipt population may return PASS or NO_FINDING. Downstream controls may consume the finding, but this Detect guidance does not prune context or block a user.

AID-D-003.007-G004

For tool-enabled systems, evaluate high-risk cross-tool context combinations and emit category-level findings before merged tool context is handed to downstream policy consumers.
Concept:

Tool outputs may be individually low sensitivity but highly sensitive in combination. This step tags tool results, evaluates risky category combinations, and emits a finding with the relevant risk code. It does not block the tool result set by itself; a separate policy or HITL component should decide whether to suppress, compartmentalize, or redact the combined context.

Step 1: Normalize tool-output categories and evaluate the combination
# File: detect/tool_combination_risk.py
HIGH_RISK_MATRIX = {
    frozenset(['calendar', 'health']): 'religion_or_health_inference',
    frozenset(['purchases', 'location']): 'political_or_financial_inference',
    frozenset(['email_summary', 'calendar', 'location']): 'religion_relationship_or_health_inference',
}


def evaluate_tool_combination(result_category_sets: list[set[str]]) -> list[str]:
    merged = set()
    for categories in result_category_sets:
        merged.update(categories)

    hits = []
    for required_categories, risk_name in HIGH_RISK_MATRIX.items():
        if required_categories.issubset(merged):
            hits.append(risk_name)
    return sorted(hits)
Step 2: Emit a finding instead of a deny decision
from __future__ import annotations

# File: detect/tool_combination_findings.py

from datetime import datetime, timezone

from tool_combination_risk import evaluate_tool_combination


def detect_cross_tool_inference(
    request_id: str,
    session_id: str,
    tool_results: list[dict[str, object]],
) -> dict[str, object]:
    category_sets = [set(result['categories']) for result in tool_results]
    hits = evaluate_tool_combination(category_sets)
    merged_categories = sorted({category for categories in category_sets for category in categories})

    return {
        'event_type': 'cross_tool_inference_finding',
        'ts': datetime.now(timezone.utc).isoformat(),
        'request_id': request_id,
        'session_id': session_id,
        'detected': bool(hits),
        'risk_codes': hits,
        'merged_categories': merged_categories,
        'recommended_action': 'review_or_compartmentalize' if hits else 'none',
    }

Action: Run this detector before tool results are merged into model context, log only category-level evidence rather than raw sensitive content, and pass the finding to a separate policy consumer. This preserves a clean evidence schema for “can detect risky cross-tool inference conditions” without conflating it with downstream blocking or redaction logic.

Verify complete coverage

The population is every policy-declared high-risk tool-result combination actually assembled for each applicable request or session, plus all eligible combinations omitted, failed, stale, or rejected by the assembler.

Verify safely

An independent verifier must reconstruct each assembled context from immutable tool-result digests, independently recompute category combinations, and replay approved, high-risk, stale, cross-session, missing-result, and cross-tenant fixtures, read evidence back, verify signatures and digests, and compare the canonical result digest.

Open-source 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-003.007: Sensitive Attribute Inference Detection & Escalation | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service