AID-D-016.002: Reputation Scoring & Containment Escalation Signals
Score agent reputation continuously using signals such as unknown identity, signature failures, abnormal tool usage, and egress anomalies, then emit auditable escalation recommendations when thresholds are crossed.
Pillars: infra, app, data
Lifecycle: operation
Scope boundary
Owns deduplicated, time-windowed, auditable agent reputation scoring and escalation evidence. It stops at scoring and recommendation and does not execute quarantine, eviction, or restoration.
AID-D-016.001: AID-D-016.002 scores reputation from discovery and other detector signals; AID-D-016.001 owns the specific unknown-identity and new-edge finding. A new identity can inform reputation but must not be treated as the complete risk score.
AID-I-003: AID-D-016.002 emits reputation and containment-escalation evidence; AID-I-003 applies quarantine, throttling, or safe-mode containment. A threshold crossing is a recommendation until an authorized Isolate control enforces it.
Implementation guidance
- AID-D-016.002-G001: Compute an auditable agent reputation score from structured security signals using time-window aggregation, de-duplication, and time decay; emit escalation events when thresholds are crossed.
AID-D-016.002-G001
Compute an auditable agent reputation score from structured security signals using time-window aggregation, de-duplication, and time decay; emit escalation events when thresholds are crossed.
Signed scoring contract
Consume the complete immutable signal window from the security-event fan-in point. A separately signed completeness receipt binds required sources, per-source counts/checkpoints, window, record count, and JSONL digest. A signed policy owns the allowed signal types, weights, confidence bounds, window/half-life, thresholds, and policy version. There is no default weight: an unknown event type produces an UNMAPPED_SIGNAL integrity finding and a non-scoreable ERROR, never a clean score or containment recommendation.
Production scorer with conflict-safe de-duplication
from __future__ import annotations
import math as _aidefend_math
import os as _aidefend_os
# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
value = _aidefend_os.environ.get(name)
if value is None or not value.strip():
raise RuntimeError(f"required runtime-profile field is absent: {name}")
return value.strip()
def _aidefend_positive_float(name):
try:
value = float(_aidefend_required(name))
except ValueError as error:
raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
if not _aidefend_math.isfinite(value) or value <= 0:
raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
return value
def _aidefend_positive_int(name):
try:
value = int(_aidefend_required(name))
except ValueError as error:
raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
if value < 1:
raise RuntimeError(f"runtime-profile field must be positive: {name}")
return value
def _aidefend_nonnegative_int(name):
try:
value = int(_aidefend_required(name))
except ValueError as error:
raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
if value < 0:
raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
return value
def _aidefend_fraction(name):
value = _aidefend_positive_float(name)
if value > 1:
raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
return value
RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
REPUTATION_ANALYSIS_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("D016002_REPUTATION_COMMAND_TIMEOUT_SECONDS")
# file: detection/score_agent_reputation.py
import hashlib, hmac, json, math, os, re, subprocess, tempfile
from collections import Counter
from datetime import datetime, timezone
from pathlib import Path
SIGNALS = Path("artifacts/reputation/security-signals.jsonl")
RECEIPT = Path("artifacts/reputation/collection-receipt.json")
POLICY = Path("policy/agent-reputation-policy.json")
EVIDENCE = Path("artifacts/reputation/AID-D-016.002-evidence.json")
_VERIFIED_INPUTS: dict[Path, bytes] = {}
def file_digest(path: Path) -> str:
payload = _VERIFIED_INPUTS.get(path)
if payload is None:
payload = path.read_bytes()
return hashlib.sha256(payload).hexdigest()
def canonical_digest(value: dict) -> str:
payload = json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
return hashlib.sha256(payload).hexdigest()
def _write_private(path: Path, payload: bytes) -> None:
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(descriptor, "wb") as stream:
stream.write(payload)
stream.flush()
os.fsync(stream.fileno())
os.chmod(path, 0o400)
def _strict_json(raw: bytes) -> object:
def unique_object(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate JSON key: " + key)
value[key] = item
return value
return json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique_object,
parse_constant=lambda token: (_ for _ in ()).throw(
ValueError("non-finite JSON constant: " + token)
),
)
def _strict_jsonl(raw: bytes) -> list[object]:
text = raw.decode("utf-8", errors="strict")
lines = text.splitlines()
if not lines or any(not line.strip() for line in lines):
raise ValueError("signal JSONL is empty or contains a blank record")
return [_strict_json(line.encode("utf-8")) for line in lines]
def verify(path: Path, key_env: str) -> bytes:
signature_path = Path(str(path) + ".sig")
if path == signature_path:
raise RuntimeError("signed payload and signature paths collide")
payload_raw = path.read_bytes()
signature_raw = signature_path.read_bytes()
key_reference = os.environ[key_env]
remote_key = bool(re.match(
r"^[A-Za-z][A-Za-z0-9+.-]*://", key_reference
)) or key_reference.startswith("pkcs11:")
key_raw = None if remote_key else (
key_reference.encode("utf-8")
if key_reference.startswith("-----BEGIN ")
else Path(key_reference).read_bytes()
)
with tempfile.TemporaryDirectory(prefix="aidefend-d016002-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
signature_snapshot = root / "payload.sig"
_write_private(payload_snapshot, payload_raw)
_write_private(signature_snapshot, signature_raw)
captured = [(payload_snapshot, payload_raw), (signature_snapshot, signature_raw)]
key_argument = key_reference
if key_raw is not None:
key_snapshot = root / "verify-key.pub"
_write_private(key_snapshot, key_raw)
captured.append((key_snapshot, key_raw))
key_argument = str(key_snapshot)
try:
subprocess.run(
["cosign", "verify-blob", "--key", key_argument,
"--bundle", str(signature_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True,
timeout=REPUTATION_ANALYSIS_COMMAND_TIMEOUT_SECONDS,
)
if any(
not hmac.compare_digest(snapshot.read_bytes(), expected)
for snapshot, expected in captured
):
raise RuntimeError("private signature-verification snapshot changed")
finally:
for snapshot, _ in captured:
os.chmod(snapshot, 0o600)
_VERIFIED_INPUTS[path] = payload_raw
return payload_raw
def timestamp(value: object) -> datetime:
if not isinstance(value, str):
raise ValueError("timestamp is absent")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.tzinfo is None:
raise ValueError("timestamp lacks timezone")
return parsed.astimezone(timezone.utc)
def finite(value: object, field: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f"{field} is not numeric")
number = float(value)
if not math.isfinite(number):
raise ValueError(f"{field} is not finite")
return number
def load() -> tuple[list[dict], dict, dict, datetime, datetime]:
signal_raw = verify(SIGNALS, "D016002_COLLECTOR_VERIFY_KEY")
receipt_raw = verify(RECEIPT, "D016002_COMPLETENESS_VERIFY_KEY")
policy_raw = verify(POLICY, "D016002_POLICY_VERIFY_KEY")
receipt = _strict_json(receipt_raw)
policy = _strict_json(policy_raw)
receipt_fields = {"schema_version", "asset_id", "asset_version",
"agent_inventory_version", "expected_agent_ids",
"window_start", "window_end", "required_sources",
"source_counts", "source_checkpoints", "collection_complete",
"record_count", "signals_sha256"}
if (not isinstance(receipt, dict) or set(receipt) != receipt_fields
or receipt["schema_version"] != "aidefend.reputation-receipt.v1"
or receipt["signals_sha256"] != file_digest(SIGNALS)
or not all(isinstance(receipt[field], str) and receipt[field]
for field in ("asset_id", "asset_version",
"agent_inventory_version"))
or not isinstance(receipt["expected_agent_ids"], list)
or any(not isinstance(value, str) or not value
for value in receipt["expected_agent_ids"])
or len(receipt["expected_agent_ids"])
!= len(set(receipt["expected_agent_ids"]))
or not isinstance(receipt["required_sources"], list)
or not receipt["required_sources"]
or any(not isinstance(value, str) or not value
for value in receipt["required_sources"])
or len(receipt["required_sources"]) != len(set(receipt["required_sources"]))
or not isinstance(receipt["source_counts"], dict)
or set(receipt["source_counts"]) != set(receipt["required_sources"])
or any(isinstance(value, bool) or not isinstance(value, int) or value < 0
for value in receipt["source_counts"].values())
or not isinstance(receipt["source_checkpoints"], dict)
or set(receipt["source_checkpoints"]) != set(receipt["required_sources"])
or not isinstance(receipt["collection_complete"], bool)
or isinstance(receipt["record_count"], bool)
or not isinstance(receipt["record_count"], int) or receipt["record_count"] < 0):
raise ValueError("completeness receipt is invalid or misbound")
policy_fields = {"schema_version", "policy_version", "signal_weights", "confidence_min",
"confidence_max", "window_seconds", "half_life_seconds",
"investigate_threshold", "containment_review_threshold",
"maximum_evidence_age_seconds"}
if not isinstance(policy, dict) or set(policy) != policy_fields or policy["schema_version"] != "aidefend.agent-reputation-policy.v1":
raise ValueError("policy schema differs")
if not isinstance(policy["policy_version"], str) or not policy["policy_version"]:
raise ValueError("policy version is absent")
if not isinstance(policy["signal_weights"], dict) or not policy["signal_weights"]:
raise ValueError("signal weight map is empty")
for event_type, weight in policy["signal_weights"].items():
if not isinstance(event_type, str) or not event_type or finite(weight, event_type) <= 0:
raise ValueError("signal weight map is invalid")
minimum, maximum = finite(policy["confidence_min"], "confidence_min"), finite(policy["confidence_max"], "confidence_max")
if not 0 <= minimum <= maximum <= 1:
raise ValueError("confidence bounds are invalid")
if (
isinstance(policy["window_seconds"], bool)
or not isinstance(policy["window_seconds"], int)
or policy["window_seconds"] < 1
):
raise ValueError("window_seconds must be a positive integer")
if (
isinstance(policy["maximum_evidence_age_seconds"], bool)
or not isinstance(policy["maximum_evidence_age_seconds"], int)
or policy["maximum_evidence_age_seconds"] < 1
):
raise ValueError("maximum_evidence_age_seconds must be a positive integer")
for field in ("half_life_seconds", "investigate_threshold", "containment_review_threshold"):
if finite(policy[field], field) <= 0:
raise ValueError(f"{field} must be positive")
if policy["investigate_threshold"] >= policy["containment_review_threshold"]:
raise ValueError("threshold order is invalid")
signals = _strict_jsonl(signal_raw)
if len(signals) != receipt["record_count"]:
raise ValueError("signal population differs")
event_fields = {"schema_version", "event_id", "dedupe_key", "agent_id", "event_type",
"observed_at", "source", "source_sequence", "source_event_id",
"confidence", "payload_sha256"}
if any(not isinstance(item, dict) or set(item) != event_fields
or item["schema_version"] != "aidefend.reputation-signal.v1" for item in signals):
raise ValueError("signal schema differs")
ids = [item["event_id"] for item in signals]
if any(not isinstance(value, str) or not value for value in ids) or len(set(ids)) != len(ids):
raise ValueError("event IDs are absent or duplicated")
start, end = timestamp(receipt["window_start"]), timestamp(receipt["window_end"])
if start >= end or int((end - start).total_seconds()) != int(policy["window_seconds"]):
raise ValueError("window differs from signed policy")
now = datetime.now(timezone.utc)
if (
end > now
or (now - end).total_seconds()
> policy["maximum_evidence_age_seconds"]
):
raise ValueError("reputation window is future-dated or stale")
for item in signals:
for field in ("dedupe_key", "agent_id", "event_type", "source", "source_event_id", "payload_sha256"):
if not isinstance(item[field], str) or not item[field]:
raise ValueError(f"{field} is absent")
if item["source"] not in receipt["required_sources"] or not start <= timestamp(item["observed_at"]) < end:
raise ValueError("signal source or time is outside receipt")
if item["agent_id"] not in set(receipt["expected_agent_ids"]):
raise ValueError("signal agent is outside signed inventory")
if (isinstance(item["source_sequence"], bool)
or not isinstance(item["source_sequence"], int)
or item["source_sequence"] < 0):
raise ValueError("signal source sequence is invalid")
if len(item["payload_sha256"]) != 64:
raise ValueError("signal payload digest is invalid")
confidence = finite(item["confidence"], "confidence")
if not minimum <= confidence <= maximum:
raise ValueError("confidence is outside signed bounds")
source_event_ids = [
(item["source"], item["source_event_id"])
for item in signals
]
if len(source_event_ids) != len(set(source_event_ids)):
raise ValueError("source event identity is duplicated")
observed_source_counts = {
source: sum(1 for item in signals if item["source"] == source)
for source in receipt["required_sources"]
}
if observed_source_counts != receipt["source_counts"]:
raise ValueError("source population differs from completeness receipt")
checkpoint_fields = {"first", "last", "complete"}
for source, value in receipt["source_checkpoints"].items():
if not isinstance(value, dict) or set(value) != checkpoint_fields or not isinstance(value["complete"], bool):
raise ValueError("source checkpoint schema differs")
sequences = sorted(item["source_sequence"] for item in signals if item["source"] == source)
if sequences:
if (isinstance(value["first"], bool) or not isinstance(value["first"], int)
or isinstance(value["last"], bool) or not isinstance(value["last"], int)
or sequences != list(range(value["first"], value["last"] + 1))):
raise ValueError("source sequence population is incomplete")
elif value["first"] is not None or value["last"] is not None:
raise ValueError("empty source has non-empty checkpoint bounds")
return signals, receipt, policy, start, end
def deduplicate(signals: list[dict]) -> tuple[list[dict], list[dict]]:
groups: dict[str, list[dict]] = {}
conflicts = []
for item in signals:
canonical_key = canonical_digest({
"agent_id": item["agent_id"],
"event_type": item["event_type"],
"source_event_id": item["source_event_id"],
"payload_sha256": item["payload_sha256"],
})
if item["dedupe_key"] != canonical_key:
conflicts.append({
"dedupe_key": item["dedupe_key"],
"event_ids": [item["event_id"]],
"reason": "producer key differs from canonical event identity",
})
continue
groups.setdefault(canonical_key, []).append(item)
unique = []
material_fields = ("agent_id", "event_type", "observed_at", "source",
"source_sequence", "source_event_id", "confidence",
"payload_sha256")
for key, group in sorted(groups.items()):
fingerprints = {canonical_digest({field: item[field] for field in material_fields})
for item in group}
if len(fingerprints) != 1:
conflicts.append({"dedupe_key": key, "event_ids": sorted(item["event_id"] for item in group)})
continue
unique.append(sorted(group, key=lambda item: item["event_id"])[0])
return unique, conflicts
def score() -> dict:
signals, receipt, policy, start, end = load()
if not receipt["expected_agent_ids"] and not signals:
return {"status": "INSUFFICIENT_DATA",
"reason": "empty collector population does not prove no-agent applicability",
"dedupe_conflicts": [], "integrity_findings": [],
"raw_population": 0, "deduplicated_population": 0,
"recommendation": None, "contributions": [], "agent_scores": {}}
if receipt["collection_complete"] is not True or any(
value["complete"] is not True for value in receipt["source_checkpoints"].values()):
return {"status": "INSUFFICIENT_DATA",
"reason": "required source/checkpoint coverage is incomplete",
"dedupe_conflicts": [], "integrity_findings": [],
"raw_population": len(signals), "deduplicated_population": 0,
"recommendation": None, "contributions": []}
unique, conflicts = deduplicate(signals)
unmapped = [{"event_id": item["event_id"], "event_type": item["event_type"],
"finding": "UNMAPPED_SIGNAL"} for item in unique
if item["event_type"] not in policy["signal_weights"]]
if conflicts or unmapped:
return {"status": "ERROR", "reason": "dedupe conflict or unmapped signal prevents scoring",
"dedupe_conflicts": conflicts, "integrity_findings": unmapped,
"raw_population": len(signals), "deduplicated_population": len(unique),
"recommendation": None, "contributions": []}
now = end
contributions = []
totals = {agent_id: 0.0 for agent_id in receipt["expected_agent_ids"]}
half_life = float(policy["half_life_seconds"])
for item in unique:
age = max(0.0, (now - timestamp(item["observed_at"])).total_seconds())
contribution = float(policy["signal_weights"][item["event_type"]]) * float(item["confidence"]) * math.pow(0.5, age / half_life)
totals[item["agent_id"]] = totals.get(item["agent_id"], 0.0) + contribution
contributions.append({"event_id": item["event_id"], "dedupe_key": item["dedupe_key"],
"agent_id": item["agent_id"], "event_type": item["event_type"],
"age_seconds": age, "contribution": round(contribution, 6)})
def recommendation_for(value: float) -> str:
if value >= policy["containment_review_threshold"]:
return "ESCALATE_CONTAINMENT_REVIEW"
if value >= policy["investigate_threshold"]:
return "INVESTIGATE"
return "OBSERVE"
agent_outcomes = {
agent_id: {
"score": round(value, 6),
"recommendation": recommendation_for(value),
}
for agent_id, value in sorted(totals.items())
}
escalated = sorted(
agent_id for agent_id, outcome in agent_outcomes.items()
if outcome["recommendation"] != "OBSERVE"
)
aggregate = (
"ESCALATE_CONTAINMENT_REVIEW"
if any(
outcome["recommendation"] == "ESCALATE_CONTAINMENT_REVIEW"
for outcome in agent_outcomes.values()
)
else "INVESTIGATE" if escalated else "OBSERVE"
)
return {"status": "FAIL" if escalated else "PASS",
"reason": "one or more agent scores crossed a review threshold" if escalated else "no signed agent score crossed a threshold",
"recommendation": aggregate, "escalated_agent_ids": escalated,
"agent_outcomes": agent_outcomes,
"agent_scores": {key: value["score"] for key, value in agent_outcomes.items()},
"contributions": contributions, "dedupe_conflicts": [], "integrity_findings": [],
"raw_population": len(signals), "deduplicated_population": len(unique)}
def main() -> None:
try:
result = score()
receipt = _strict_json(_VERIFIED_INPUTS[RECEIPT])
policy = _strict_json(_VERIFIED_INPUTS[POLICY])
evidence = {"schema_version": "aidefend.d016002-evidence.v1",
"control": "AID-D-016.002", **result,
"asset_id": receipt["asset_id"],
"asset_version": receipt["asset_version"],
"agent_inventory_version": receipt["agent_inventory_version"],
"expected_agent_population": len(receipt["expected_agent_ids"]),
"window_start": receipt["window_start"], "window_end": receipt["window_end"],
"signals_sha256": file_digest(SIGNALS), "receipt_sha256": file_digest(RECEIPT),
"policy_sha256": file_digest(POLICY), "policy_version": policy["policy_version"],
"evaluated_at": datetime.now(timezone.utc).isoformat()}
except Exception as error:
evidence = {"schema_version": "aidefend.d016002-evidence.v1",
"control": "AID-D-016.002", "status": "ERROR", "recommendation": None,
"reason": f"{type(error).__name__}: {error}",
"evaluated_at": datetime.now(timezone.utc).isoformat()}
EVIDENCE.parent.mkdir(parents=True, exist_ok=True)
EVIDENCE.write_text(json.dumps(evidence, sort_keys=True, separators=(",", ":")) + "\n", encoding="utf-8")
subprocess.run(["cosign", "sign-blob", "--yes", "--key", "env://D016002_EVIDENCE_SIGNING_KEY",
"--bundle", str(EVIDENCE) + ".sig", str(EVIDENCE)], check=True, timeout=REPUTATION_ANALYSIS_COMMAND_TIMEOUT_SECONDS)
verify(EVIDENCE, "D016002_EVIDENCE_VERIFY_KEY")
if __name__ == "__main__":
main()
Observation point, boundary, and findings
Security telemetry owners sign normalized source events and checkpoints; the agent-governance owner signs the complete agent inventory; detection engineering owns weights, decay, thresholds, and scoring at the security-event fan-in point. Bind each result to the exact asset/version and inventory version. Re-verify source artifacts and recompute the score locally; treat detector and collector results as authenticated inputs that still require local recomputation.
Action: At the reputation evaluator, reconcile and de-duplicate the complete signed signal window, compute the policy-bound score and recommendation, persist the signed replay bundle, and emit threshold-crossing recommendations to the containment decision point; this detector does not isolate an agent.
Open-source tools
- OpenTelemetry
- Prometheus
- Apache Kafka
- Valkey
- Redis 8+ AGPLv3 source build
- Falco
- Cosign
Commercial / hosted tools
- Palo Alto Networks Prisma Cloud
- Splunk Enterprise Security
- Datadog
- Microsoft Sentinel
Threat mappings
MITRE ATLAS
- AML.T0108 AI Agent (reputation scoring detects C2-controlled agents)
MAESTRO
- Compromised Agents (L7) (reputation scoring surfaces compromised agents for response)
- Agent Identity Attack (L7)
OWASP LLM Top 10 2026
- N/A
OWASP ML Top 10 2023
- N/A
OWASP Top 10 for Agentic Applications 2026
- ASI10:2026 Rogue Agents
- ASI03:2026 Identity and Privilege Abuse
- ASI02:2026 Tool Misuse and Exploitation (reputation scoring correlates authenticated tool-misuse signals)
NIST Adversarial Machine Learning 2025
- N/A
Cisco Integrated AI Security and Safety Framework
- N/A
Google Secure AI Framework 2.0 - Risks
- RA: Rogue Actions (reputation scoring produces proportional escalation signals for rogue agents)
- IIC: Insecure Integrated Component (reputation scoring surfaces compromised components for follow-on containment)
Databricks AI Security Framework 3.0
- Agents - Core 13.13: Rogue Agents in Multi-Agent Systems
- Agents - Core 13.7: Misaligned & Deceptive Behaviors
- Agents - Core 13.9: Identity Spoofing & Impersonation