AID-D-009.001: Source-Grounded Claim Verification & Confidence State

Version 1.20261015 · Actionable sub-technique

Verify high-impact factual claims against policy-approved independent and authoritative sources, preserve contradictions and abstentions, and attach an evidence-derived verification state and calibrated confidence for triage. This control applies to a single agent or multi-agent system and does not treat model confidence as truth.

Pillars: app, data

Lifecycle: operation

Scope boundary

Owns verification state and calibrated confidence for high-impact factual claims based on policy-approved independent and authoritative sources, preserving contradiction, uncertainty, abstention, and source evidence.

AID-D-009.002: AID-D-009.001 determines the evidence-grounded state of a claim; AID-D-009.002 detects how a contradicted, unverified, inconclusive, or poisoned claim propagates across agents, messages, memory, and delegation. Claim truth state and propagation lineage are separate findings, and peer repetition never upgrades the truth state.

Implementation guidance

AID-D-009.001-G001

Verify high-impact agent assertions against multiple independent sources and emit a structured verification decision.
Define authority, independence, freshness, and source population

The signed verification policy names the exact required sources, each source's authority scope and independence group, the minimum supporting groups, and the maximum observation age. Mutually authenticated adapters return signed observations; generic web results cannot become authoritative by choosing a provider label.

Verify and classify every observation
from __future__ import annotations

# File: verification/multi_source.py

import hashlib
import json
import math
import re
from collections import defaultdict
from dataclasses import dataclass
from typing import Literal

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

Verdict = Literal["SUPPORTS", "CONTRADICTS", "ABSTAINS"]
SHA256 = re.compile(r"^[0-9a-f]{64}$")


def _is_sha256(value: object) -> bool:
    return isinstance(value, str) and SHA256.fullmatch(value) is not None


@dataclass(frozen=True)
class TrustedSource:
    independence_group: str
    authority_scopes: frozenset[str]
    verification_key: Ed25519PublicKey


@dataclass(frozen=True)
class SignedObservation:
    source_id: str
    evidence: bytes
    signature: bytes


def _valid_observation_shape(value: object) -> bool:
    return (
        isinstance(value, SignedObservation)
        and isinstance(value.source_id, str)
        and bool(value.source_id)
        and isinstance(value.evidence, bytes)
        and bool(value.evidence)
        and isinstance(value.signature, bytes)
        and bool(value.signature)
    )


def _strict_object(pairs: list[tuple[str, object]]) -> dict:
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError("duplicate source-evidence key: " + key)
        value[key] = item
    return value


def parse_observation(
    item: SignedObservation,
    *,
    trusted_sources: dict[str, TrustedSource],
    expected_claim_sha256: str,
    expected_scope: str,
    evaluated_at_unix: float,
    maximum_age_seconds: int,
) -> dict:
    if (
        not _valid_observation_shape(item)
        or not isinstance(trusted_sources, dict)
        or not _is_sha256(expected_claim_sha256)
        or not isinstance(expected_scope, str)
        or not expected_scope
        or isinstance(evaluated_at_unix, bool)
        or not isinstance(evaluated_at_unix, (int, float))
        or not math.isfinite(float(evaluated_at_unix))
        or isinstance(maximum_age_seconds, bool)
        or not isinstance(maximum_age_seconds, int)
        or maximum_age_seconds < 1
    ):
        raise ValueError("observation verification inputs are invalid")
    trusted = trusted_sources.get(item.source_id)
    if trusted is None or expected_scope not in trusted.authority_scopes:
        raise ValueError("source is not authoritative for the requested scope")
    trusted.verification_key.verify(item.signature, item.evidence)
    value = json.loads(
        item.evidence.decode("utf-8", errors="strict"),
        object_pairs_hook=_strict_object,
        parse_constant=lambda item: (_ for _ in ()).throw(
            ValueError("non-finite source value")
        ),
    )
    required = {
        "schema_version", "source_id", "claim_sha256", "authority_scope",
        "observed_at_unix", "valid_until_unix", "verdict", "normalized_value",
    }
    if not isinstance(value, dict) or set(value) != required:
        raise ValueError("source evidence schema mismatch")
    observed_at = value["observed_at_unix"]
    valid_until = value["valid_until_unix"]
    if (
        value["schema_version"] != "aidefend.source_observation.v2"
        or value["source_id"] != item.source_id
        or value["claim_sha256"] != expected_claim_sha256
        or value["authority_scope"] != expected_scope
        or not isinstance(value["verdict"], str)
        or value["verdict"] not in {"SUPPORTS", "CONTRADICTS", "ABSTAINS"}
        or isinstance(observed_at, bool)
        or not isinstance(observed_at, (int, float))
        or isinstance(valid_until, bool)
        or not isinstance(valid_until, (int, float))
        or not math.isfinite(float(observed_at))
        or not math.isfinite(float(valid_until))
    ):
        raise ValueError("source evidence is misbound or malformed")
    normalized = value["normalized_value"]
    if value["verdict"] == "ABSTAINS":
        if normalized is not None:
            raise ValueError("abstention must not carry a normalized value")
    elif not isinstance(normalized, str) or not normalized:
        raise ValueError("support or contradiction lacks a normalized value")
    fresh = (
        float(observed_at) <= evaluated_at_unix
        and evaluated_at_unix - float(observed_at) <= maximum_age_seconds
        and float(valid_until) >= evaluated_at_unix
    )
    return {
        **value,
        "fresh": fresh,
        "independence_group": trusted.independence_group,
        "evidence_sha256": hashlib.sha256(item.evidence).hexdigest(),
    }


def verify_claim(
    observations: list[SignedObservation],
    *,
    trusted_sources: dict[str, TrustedSource],
    required_source_ids: tuple[str, ...],
    expected_claim_sha256: str,
    expected_scope: str,
    evaluated_at_unix: float,
    maximum_age_seconds: int,
    required_independent_groups: int,
    policy_version: str,
) -> dict:
    if (
        not isinstance(observations, list)
        or any(not _valid_observation_shape(item) for item in observations)
    ):
        raise ValueError("source observations are malformed")
    if not isinstance(trusted_sources, dict):
        raise ValueError("trusted source registry is malformed")
    if (
        not isinstance(required_source_ids, tuple)
        or not required_source_ids
        or any(not isinstance(item, str) or not item for item in required_source_ids)
    ):
        raise ValueError("required source identifiers are malformed")
    if (
        isinstance(required_independent_groups, bool)
        or not isinstance(required_independent_groups, int)
        or required_independent_groups < 2
        or isinstance(maximum_age_seconds, bool)
        or not isinstance(maximum_age_seconds, int)
        or maximum_age_seconds < 1
        or isinstance(evaluated_at_unix, bool)
        or not isinstance(evaluated_at_unix, (int, float))
        or not math.isfinite(float(evaluated_at_unix))
    ):
        raise ValueError("verification thresholds or evaluation time are invalid")
    if (
        not isinstance(policy_version, str)
        or not policy_version
        or not isinstance(expected_scope, str)
        or not expected_scope
        or not _is_sha256(expected_claim_sha256)
    ):
        raise ValueError("verification policy binding is malformed")

    required_sources = set(required_source_ids)
    observed_sources = [item.source_id for item in observations]
    if (
        len(required_sources) != len(required_source_ids)
        or len(observed_sources) != len(set(observed_sources))
        or set(observed_sources) - required_sources
    ):
        raise ValueError("source population is duplicated or outside policy")
    missing = sorted(required_sources - set(observed_sources))
    parsed = [
        parse_observation(
            item,
            trusted_sources=trusted_sources,
            expected_claim_sha256=expected_claim_sha256,
            expected_scope=expected_scope,
            evaluated_at_unix=float(evaluated_at_unix),
            maximum_age_seconds=maximum_age_seconds,
        )
        for item in observations
    ]
    fresh = [item for item in parsed if item["fresh"]]
    stale = sorted(item["source_id"] for item in parsed if not item["fresh"])
    supports = [item for item in fresh if item["verdict"] == "SUPPORTS"]
    contradictions = [item for item in fresh if item["verdict"] == "CONTRADICTS"]
    abstentions = [item for item in fresh if item["verdict"] == "ABSTAINS"]
    support_groups: dict[str, set[str]] = defaultdict(set)
    for item in supports:
        support_groups[item["independence_group"]].add(item["normalized_value"])
    conflicting_support = any(len(values) != 1 for values in support_groups.values())
    supported_values = {
        next(iter(values)) for values in support_groups.values() if len(values) == 1
    }

    if contradictions or conflicting_support or len(supported_values) > 1:
        decision = "CONTRADICTED"
    elif (
        not missing
        and not stale
        and len(support_groups) >= required_independent_groups
        and len(supported_values) == 1
    ):
        decision = "VERIFIED"
    else:
        decision = "INCONCLUSIVE"
    return {
        "schema_version": "aidefend.multi_source_verification.v2",
        "decision": decision,
        "claim_sha256": expected_claim_sha256,
        "authority_scope": expected_scope,
        "policy_version": policy_version,
        "required_source_ids": sorted(required_sources),
        "missing_source_ids": missing,
        "stale_source_ids": stale,
        "abstaining_source_ids": sorted(item["source_id"] for item in abstentions),
        "supporting_independence_groups": sorted(support_groups),
        "contradicting_source_ids": sorted(
            item["source_id"] for item in contradictions
        ),
        "observed_values": sorted(supported_values),
        "evidence_sha256": sorted(item["evidence_sha256"] for item in parsed),
    }
Replay and decision semantics

Derive expected_claim_sha256 from the immutable canonical claim, not from an agent-provided digest. Signature failure, malformed evidence, wrong scope, unexpected sources, and duplicate sources are verifier errors; missing, unavailable, stale, abstaining, or insufficient independent sources remain INCONCLUSIVE; a fresh explicit contradiction or conflicting authoritative value is CONTRADICTED. Replay supported, contradicted, stale, correlated-mirror, unavailable, abstaining, temporal-change, wrong-scope, and duplicate-source fixtures under a separate identity.

AID-D-009.001-G002

Check claim classes against authoritative source systems before marking them verified.
Read and authenticate the exact authoritative record

Load the endpoint, tenant/resource scope, freshness rule, response bounds, receipt issuer/audience/key, and mTLS material from a signature-verified registry profile. The lookup must bind the response and its signed receipt to the exact claim, tenant, resource, requested as-of instant, and returned version. A transport, schema, receipt, freshness, or binding failure is unavailable evidence, never a verified claim.

# File: verification/authoritative_lookup.py
from __future__ import annotations

import hashlib
import json
import math
import ssl
from datetime import datetime, timezone
from urllib.parse import urlsplit

import httpx
import jwt


def reject_duplicate_keys(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError(f'duplicate authoritative-response key: {key}')
        value[key] = item
    return value


def canonical(value: dict) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(',', ':'), allow_nan=False).encode()


def lookup(*, profile: dict, claim_id: str, tenant_id: str,
           resource_id: str, as_of: str) -> dict:
    required_profile = {
        'policy_version', 'policy_sha256', 'endpoint', 'connect_timeout_seconds',
        'read_timeout_seconds', 'response_max_bytes', 'response_chunk_bytes',
        'ca_file', 'client_cert', 'client_key', 'receipt_public_key',
        'receipt_issuer', 'receipt_audience', 'maximum_receipt_age_seconds',
    }
    if not isinstance(profile, dict) or set(profile) != required_profile:
        raise ValueError('verified authoritative-source profile schema differs')
    for field in ('connect_timeout_seconds', 'read_timeout_seconds', 'maximum_receipt_age_seconds'):
        value = profile[field]
        if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(float(value)) or value <= 0:
            raise ValueError(f'invalid signed value: {field}')
    maximum = profile['response_max_bytes']
    chunk = profile['response_chunk_bytes']
    if (isinstance(maximum, bool) or not isinstance(maximum, int) or maximum < 2
            or isinstance(chunk, bool) or not isinstance(chunk, int)
            or chunk < 1 or chunk > maximum):
        raise ValueError('signed response bounds are invalid')
    parsed = urlsplit(profile['endpoint'])
    if (parsed.scheme != 'https' or not parsed.hostname or parsed.username
            or parsed.password or parsed.query or parsed.fragment):
        raise ValueError('authoritative endpoint is not one canonical HTTPS URL')
    if not all(isinstance(value, str) and value for value in (claim_id, tenant_id, resource_id, as_of)):
        raise ValueError('exact claim and resource scope is required')

    tls = ssl.create_default_context(cafile=profile['ca_file'])
    tls.load_cert_chain(profile['client_cert'], profile['client_key'])
    timeout = httpx.Timeout(float(profile['read_timeout_seconds']),
                            connect=float(profile['connect_timeout_seconds']))
    with httpx.Client(verify=tls, timeout=timeout, follow_redirects=False) as client:
        with client.stream(
            'GET', profile['endpoint'],
            params={'claim_id': claim_id, 'tenant_id': tenant_id,
                    'resource_id': resource_id, 'as_of': as_of},
            headers={'Accept': 'application/json', 'Accept-Encoding': 'identity'},
        ) as response:
            if 300 <= response.status_code < 400:
                raise RuntimeError('authoritative-source redirect is forbidden')
            response.raise_for_status()
            if response.headers.get('Content-Encoding', 'identity').lower() != 'identity':
                raise RuntimeError('encoded authoritative response is forbidden')
            chunks, observed = [], 0
            for item in response.iter_bytes(chunk):
                observed += len(item)
                if observed > maximum:
                    raise RuntimeError('authoritative response exceeds signed bound')
                chunks.append(item)
    payload = json.loads(
        b''.join(chunks).decode('utf-8', errors='strict'),
        object_pairs_hook=reject_duplicate_keys, parse_constant=lambda value: (_ for _ in ()).throw(ValueError('non-finite JSON constant: ' + value)),
    )
    fields = {'claim_id', 'tenant_id', 'resource_id', 'as_of',
              'version', 'value', 'receipt_jws'}
    if not isinstance(payload, dict) or set(payload) != fields:
        raise RuntimeError('authoritative response schema differs')
    if (payload['claim_id'], payload['tenant_id'], payload['resource_id'], payload['as_of']) != (
            claim_id, tenant_id, resource_id, as_of) or not payload['version']:
        raise RuntimeError('authoritative response scope binding differs')
    signed_payload = {key: value for key, value in payload.items() if key != 'receipt_jws'}
    payload_sha256 = hashlib.sha256(canonical(signed_payload)).hexdigest()
    claims = jwt.decode(
        payload['receipt_jws'], profile['receipt_public_key'], algorithms=['EdDSA'],
        issuer=profile['receipt_issuer'], audience=profile['receipt_audience'],
        options={'require': ['iss', 'aud', 'iat', 'exp', 'jti', 'claim_id',
                             'tenant_id', 'resource_id', 'as_of', 'version',
                             'payload_sha256']},
    )
    if any(claims[field] != signed_payload[field] for field in (
            'claim_id', 'tenant_id', 'resource_id', 'as_of', 'version')):
        raise RuntimeError('authoritative receipt scope differs')
    if claims['payload_sha256'] != payload_sha256:
        raise RuntimeError('authoritative receipt digest differs')
    now = datetime.now(timezone.utc).timestamp()
    if isinstance(claims['iat'], bool) or not isinstance(claims['iat'], int) or not 0 <= now - claims['iat'] <= profile['maximum_receipt_age_seconds']:
        raise RuntimeError('authoritative receipt is future-issued or stale')
    return {**signed_payload, 'receipt_jti': claims['jti'],
            'response_sha256': payload_sha256,
            'policy_version': profile['policy_version'],
            'policy_sha256': profile['policy_sha256']}
Independent replay

A separate verifier identity rereads the same tenant/resource/as-of record or authenticates the retained immutable response, verifies the receipt independently, and recomputes the typed comparison and freshness decision. Test wrong tenant, wrong resource, stale cache, schema change, missing record, version race, redirect, oversized body, duplicate keys, TLS failure, and receipt misbinding.

AID-D-009.001-G003

Attach confidence and verification state to agent-generated facts so low-confidence assertions are routed for deeper review instead of being treated as verified.
Before you begin

Owned by the factual-assurance telemetry team for agent-generated facts that require a calibrated confidence and verification-state envelope for triage.

Production implementation

Validate the immutable claim and model confidence, calibrate only within a supported stratum, validate any upstream source/peer/authoritative receipts by identity/scope/version/freshness/evidence digest, compute VERIFIED, CONTRADICTED, UNVERIFIED or INCONCLUSIVE independently of numeric confidence, and emit the bounded triage annotation. High confidence without valid source evidence remains UNVERIFIED, and no action is approved or blocked.

Independent verification

An independent runner replays the exact inputs and receipts, recomputes calibration and state transition, runs high-confidence-false, low-confidence-supported, stale-receipt, unsupported-stratum and contradictory-evidence fixtures, and reconciles every eligible claim/annotation with the workflow ledger.

Executable verification-state gate
# File: verification/claim_state.py
from __future__ import annotations

import math


def claim_state(
    *, calibrated_confidence: float,
    verification_decision: str,
    verified_threshold: float,
    review_threshold: float,
) -> dict:
    values = (calibrated_confidence, verified_threshold, review_threshold)
    if any(isinstance(value, bool) or not math.isfinite(value) for value in values):
        raise ValueError("confidence policy is invalid")
    if not 0 <= review_threshold <= verified_threshold <= 1:
        raise ValueError("confidence thresholds are inconsistent")
    if verification_decision == "VERIFIED" and calibrated_confidence >= verified_threshold:
        state = "VERIFIED"
    elif calibrated_confidence >= review_threshold:
        state = "REVIEW_REQUIRED"
    else:
        state = "UNVERIFIED"
    return {
        "state": state,
        "confidence": calibrated_confidence,
        "verification_decision": verification_decision,
    }

Only VERIFIED claims may receive a verified label. Preserve the calibration-set digest, verifier receipts, policy version, confidence, and state so an independent evaluator can recompute the transition.

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-009.001: Source-Grounded Claim Verification & Confidence State | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service