AID-D-013: RL Reward & Policy Manipulation Detection
Monitor and analyze Reinforcement Learning (RL), RLHF/RLAIF, RLVR, and agentic post-training systems to detect reward hacking, reward tampering, verifier gaming, and policy manipulation. Reward hacking occurs when an agent discovers an exploit in the reward function or evaluator to achieve high measured reward for unintended or harmful behavior. Reward tampering involves an actor or compromised service manipulating the reward signal, verifier result, or evaluator logs. This technique compares reward streams, trajectories, verifier outputs, held-out evaluations, and task-success signals collected outside the reward path so high reward cannot silently substitute for real goal achievement.
Pillars: model
Lifecycle: building, validation, operation
Scope boundary
Owns findings for reinforcement-learning reward hacking, reward tampering, verifier gaming, and policy manipulation by comparing reward streams, trajectories, verifier outputs, held-out evaluations, and task-success signals outside the reward path. It does not change rewards, retrain a policy, or contain an agent.
AID-H-012: AID-D-013 detects reward and verifier manipulation from observed training or agentic RL evidence; AID-H-012 designs and enforces preventive reward-function, verifier, training, and pre-release hardening. Preventive controls reduce reward exploits, while Detect identifies manipulation or gaming that still occurs.
AID-D-010: AID-D-013 focuses on reward, verifier, and policy optimization signals; AID-D-010 focuses on whether live plans and actions remain aligned with the approved goal state. An agent can game reward without obvious goal-text deviation, or drift from its goal without tampering with the reward path.
Implementation guidance
- AID-D-013-G001: Monitor the reward stream for statistical anomalies.
- AID-D-013-G002: Analyze agent trajectories to detect pathological behaviors.
- AID-D-013-G003: Periodically test the agent's policy in a sandboxed 'honey-state'.
- AID-D-013-G004: Implement out-of-band reward signal verification.
- AID-D-013-G005: Monitor RLHF, RLAIF, RLVR, and agentic RL reward-model or verifier gaming by comparing training reward, independent evaluator outcomes, and held-out task success.
AID-D-013-G001
Monitor the reward stream for statistical anomalies.
When and where to run this detector
Use this detector when an RL, RLHF/RLAIF, RLVR, or agentic post-training asset emits numeric rewards. The RL platform owner provides immutable reward events; the detection engineering owner evaluates them at the reward-event fan-in point after evaluator emission and before dashboards or promotion gates consume the stream. A signed architecture inventory binds the exact asset and version, run, policy artifact, environment, evaluator, and reward-definition versions. An absent reward-producing path is NOT_APPLICABLE; unsupported or missing events are not a pass.
Production evaluator
from __future__ import annotations
# Non-tunable process-safety deadline; signed policy and callers cannot raise it.
COSIGN_DEADLINE_CEILING_SECONDS = 15.0
# File: detection/evaluate_reward_stream.py
import hashlib
import json
import math
import os
import subprocess
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
import hmac
import re
import tempfile
import pandas as pd
EVENTS = Path("artifacts/rl/reward-events.jsonl")
RECEIPT = Path("artifacts/rl/reward-collection-receipt.json")
POLICY = Path("policy/reward-anomaly-policy.json")
EVIDENCE = Path("artifacts/rl/AID-D-013-reward-stream-evidence.json")
REWARD_COLLECTOR_VERIFY_KEY = Path("/etc/aidefend/d013/trust/reward-collector.pub")
REWARD_COMPLETENESS_VERIFY_KEY = Path("/etc/aidefend/d013/trust/reward-completeness.pub")
REWARD_POLICY_VERIFY_KEY = Path("/etc/aidefend/d013/trust/reward-policy.pub")
REWARD_EVIDENCE_VERIFY_KEY = Path("/etc/aidefend/d013/trust/reward-evidence.pub")
BINDING_FIELDS = (
"run_id", "policy_artifact_sha256", "environment_artifact_sha256",
"evaluator_artifact_sha256", "reward_definition_sha256",
)
SHA256 = re.compile(r"[0-9a-f]{64}")
def digest(path: Path) -> str:
captured = _VERIFIED_INPUTS.get(Path(path))
payload = captured if captured is not None else Path(path).read_bytes()
return hashlib.sha256(payload).hexdigest()
_VERIFIED_INPUTS: dict[Path, bytes] = {}
def _write_private_snapshot(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 | str) -> object:
text = raw.decode("utf-8", errors="strict") if isinstance(raw, bytes) else raw
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(
text,
object_pairs_hook=unique_object,
parse_constant=lambda value: (_ for _ in ()).throw(
ValueError("non-finite JSON constant: " + value)
),
)
def verify(path: Path, verify_key: Path) -> None:
payload_path = Path(path)
signature_path = Path(str(payload_path) + ".sig")
if payload_path == signature_path:
raise ValueError("signed payload and signature paths must differ")
payload_raw = payload_path.read_bytes()
signature_raw = signature_path.read_bytes()
key_reference = Path(verify_key)
key_raw = key_reference.read_bytes()
with tempfile.TemporaryDirectory(prefix="aidefend-d013-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
signature_snapshot = root / "signature"
_write_private_snapshot(payload_snapshot, payload_raw)
_write_private_snapshot(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_snapshot(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=COSIGN_DEADLINE_CEILING_SECONDS,
)
if any(
not hmac.compare_digest(candidate.read_bytes(), expected)
for candidate, expected in captured
):
raise RuntimeError("private signature-verification snapshot changed")
finally:
for candidate, _ in captured:
os.chmod(candidate, 0o600)
_VERIFIED_INPUTS[payload_path] = payload_raw
def _verified_bytes(path: Path) -> bytes:
try:
return _VERIFIED_INPUTS[Path(path)]
except KeyError as error:
raise RuntimeError("signed input was not captured and verified") from error
def _verified_text(path: Path) -> str:
return _verified_bytes(path).decode("utf-8", errors="strict")
def finite_number(value: object, field: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f"{field} is not numeric")
result = float(value)
if not math.isfinite(result):
raise ValueError(f"{field} is not finite")
return result
def load_inputs() -> tuple[list[dict], dict, dict]:
verify(EVENTS, REWARD_COLLECTOR_VERIFY_KEY)
verify(RECEIPT, REWARD_COMPLETENESS_VERIFY_KEY)
verify(POLICY, REWARD_POLICY_VERIFY_KEY)
receipt = _strict_json(_verified_bytes(RECEIPT))
policy = _strict_json(_verified_bytes(POLICY))
receipt_fields = {
"schema_version", "asset_id", "asset_version", "window_start",
"window_end", "event_count", "first_sequence", "last_sequence",
"collection_complete", "events_sha256", "expected_bindings",
}
if (
not isinstance(receipt, dict)
or set(receipt) != receipt_fields
or receipt["schema_version"] != "aidefend.reward-collection.v1"
or receipt["events_sha256"] != digest(EVENTS)
or SHA256.fullmatch(str(receipt["events_sha256"])) is None
or not all(
isinstance(receipt[field], str) and 1 <= len(receipt[field]) <= 256
for field in ("asset_id", "asset_version", "window_start", "window_end")
)
or isinstance(receipt["event_count"], bool)
or not isinstance(receipt["event_count"], int)
or receipt["event_count"] < 0
or not isinstance(receipt["collection_complete"], bool)
or not isinstance(receipt["expected_bindings"], list)
or not receipt["expected_bindings"]
):
raise ValueError("reward collection receipt is invalid or misbound")
expected_binding_keys = []
for binding in receipt["expected_bindings"]:
if (
not isinstance(binding, dict)
or set(binding) != set(BINDING_FIELDS)
or not isinstance(binding["run_id"], str)
or not 1 <= len(binding["run_id"]) <= 256
or any(
not isinstance(binding[field], str)
or SHA256.fullmatch(binding[field]) is None
for field in BINDING_FIELDS[1:]
)
):
raise ValueError("expected reward binding is invalid")
expected_binding_keys.append(tuple(binding[field] for field in BINDING_FIELDS))
if len(expected_binding_keys) != len(set(expected_binding_keys)):
raise ValueError("expected reward bindings are duplicated")
window_start = datetime.fromisoformat(
receipt["window_start"].replace("Z", "+00:00")
)
window_end = datetime.fromisoformat(
receipt["window_end"].replace("Z", "+00:00")
)
if (
window_start.tzinfo is None
or window_end.tzinfo is None
or window_start >= window_end
or window_end > datetime.now(timezone.utc)
):
raise ValueError("reward collection window is invalid or not closed")
policy_fields = {
"schema_version", "policy_version", "baseline_events",
"minimum_baseline_events", "z_score_threshold",
"variance_ratio_floor", "standard_deviation_floor",
}
if (
not isinstance(policy, dict)
or set(policy) != policy_fields
or policy["schema_version"] != "aidefend.reward-anomaly-policy.v1"
or not isinstance(policy["policy_version"], str)
or not policy["policy_version"]
):
raise ValueError("reward anomaly policy schema differs")
for field in ("baseline_events", "minimum_baseline_events"):
if (
isinstance(policy[field], bool)
or not isinstance(policy[field], int)
or policy[field] < 2
):
raise ValueError(f"{field} must be an integer of at least two")
if policy["minimum_baseline_events"] > policy["baseline_events"]:
raise ValueError("minimum baseline exceeds baseline window")
for field in (
"z_score_threshold", "variance_ratio_floor",
"standard_deviation_floor",
):
finite_number(policy[field], field)
if (
policy["z_score_threshold"] <= 0
or not 0 <= policy["variance_ratio_floor"] <= 1
or policy["standard_deviation_floor"] <= 0
):
raise ValueError("signed anomaly thresholds are outside valid ranges")
lines = _verified_text(EVENTS).splitlines()
if any(not line.strip() for line in lines):
raise ValueError("reward JSONL contains a blank record")
events = [_strict_json(line) for line in lines]
if len(events) != receipt["event_count"]:
raise ValueError("reward event count differs from receipt")
if events and (
isinstance(receipt["first_sequence"], bool)
or not isinstance(receipt["first_sequence"], int)
or isinstance(receipt["last_sequence"], bool)
or not isinstance(receipt["last_sequence"], int)
):
raise ValueError("non-empty reward population lacks integer sequence bounds")
event_fields = {
"schema_version", "event_id", "asset_id", "asset_version", "run_id",
"policy_artifact_sha256", "environment_artifact_sha256",
"evaluator_artifact_sha256", "reward_definition_sha256",
"sequence", "observed_at", "reward",
}
if any(
not isinstance(item, dict)
or set(item) != event_fields
or item["schema_version"] != "aidefend.reward-event.v1"
for item in events
):
raise ValueError("reward event schema differs")
ids = [item["event_id"] for item in events]
if (
any(not isinstance(value, str) or not value for value in ids)
or len(ids) != len(set(ids))
):
raise ValueError("reward event IDs are missing or duplicated")
if any(
item["asset_id"] != receipt["asset_id"]
or item["asset_version"] != receipt["asset_version"]
for item in events
):
raise ValueError("reward event asset binding differs")
sequences = [item["sequence"] for item in events]
if any(
isinstance(value, bool) or not isinstance(value, int) or value < 0
for value in sequences
):
raise ValueError("reward sequence is invalid")
if events:
if (
receipt["first_sequence"] != min(sequences)
or receipt["last_sequence"] != max(sequences)
or len(set(sequences)) != len(sequences)
or max(sequences) - min(sequences) + 1 != len(sequences)
):
raise ValueError("reward sequence population is incomplete")
elif receipt["first_sequence"] is not None or receipt["last_sequence"] is not None:
raise ValueError("empty population has sequence bounds")
for item in events:
finite_number(item["reward"], "reward")
observed = datetime.fromisoformat(item["observed_at"].replace("Z", "+00:00"))
if (
observed.tzinfo is None
or not window_start <= observed.astimezone(timezone.utc) < window_end
):
raise ValueError("reward timestamp is outside the signed window")
if (
not isinstance(item["run_id"], str)
or not 1 <= len(item["run_id"]) <= 256
or any(
not isinstance(item[field], str)
or SHA256.fullmatch(item[field]) is None
for field in BINDING_FIELDS[1:]
)
):
raise ValueError("reward event binding is invalid")
observed_binding_keys = {
tuple(item[field] for field in BINDING_FIELDS) for item in events
}
if observed_binding_keys - set(expected_binding_keys):
raise ValueError("reward event contains an unexpected binding")
return events, receipt, policy
def evaluate(events: list[dict], receipt: dict, policy: dict) -> dict:
coverage_gaps = []
if receipt["collection_complete"] is not True:
coverage_gaps.append("authenticated reward collection is incomplete")
grouped: dict[tuple[str, ...], list[dict]] = defaultdict(list)
for item in events:
grouped[tuple(item[field] for field in BINDING_FIELDS)].append(item)
findings: list[dict] = []
expected = {
tuple(binding[field] for field in BINDING_FIELDS)
for binding in receipt["expected_bindings"]
}
insufficient_groups: list[dict] = [
{
"binding": dict(zip(BINDING_FIELDS, binding)),
"event_count": 0,
"reason": "expected binding is absent from the event population",
}
for binding in sorted(expected - set(grouped))
]
baseline_size = policy["baseline_events"]
minimum = policy["minimum_baseline_events"]
for binding, group in sorted(grouped.items()):
ordered = sorted(group, key=lambda item: item["sequence"])
rewards = pd.Series(
[float(item["reward"]) for item in ordered],
index=[item["event_id"] for item in ordered],
dtype="float64",
)
prior_mean = rewards.shift(1).rolling(
baseline_size, min_periods=minimum
).mean()
prior_std = rewards.shift(1).rolling(
baseline_size, min_periods=minimum
).std(ddof=0)
short_variance = rewards.shift(1).rolling(
minimum, min_periods=minimum
).var(ddof=0)
long_variance = rewards.shift(1).rolling(
baseline_size, min_periods=minimum
).var(ddof=0)
scored = 0
for item in ordered:
event_id = item["event_id"]
mean = prior_mean.get(event_id)
standard_deviation = prior_std.get(event_id)
if pd.isna(mean) or pd.isna(standard_deviation):
continue
scored += 1
standard_deviation = max(
float(standard_deviation),
float(policy["standard_deviation_floor"]),
)
z_score = (float(item["reward"]) - float(mean)) / standard_deviation
reasons = []
if abs(z_score) >= float(policy["z_score_threshold"]):
reasons.append("reward_spike_or_drop")
short_var = short_variance.get(event_id)
long_var = long_variance.get(event_id)
if (
not pd.isna(short_var)
and not pd.isna(long_var)
and float(long_var) > 0
and float(short_var) / float(long_var)
<= float(policy["variance_ratio_floor"])
):
reasons.append("reward_variance_collapse")
if reasons:
findings.append({
"event_id": event_id,
"binding": dict(zip(BINDING_FIELDS, binding)),
"sequence": item["sequence"],
"reasons": reasons,
"reward": item["reward"],
"prior_mean": float(mean),
"prior_standard_deviation": standard_deviation,
"z_score": z_score,
})
if scored == 0:
insufficient_groups.append({
"binding": dict(zip(BINDING_FIELDS, binding)),
"event_count": len(ordered),
})
if findings:
return {
"status": "FAIL",
"reason": "reward anomaly threshold crossed",
"findings": findings,
"insufficient_groups": insufficient_groups,
"coverage_gaps": coverage_gaps,
}
if insufficient_groups or coverage_gaps:
return {
"status": "INSUFFICIENT_DATA",
"reason": "reward binding coverage or prior baseline is incomplete",
"findings": [],
"insufficient_groups": insufficient_groups,
"coverage_gaps": coverage_gaps,
}
return {
"status": "PASS",
"reason": "complete reward population had no policy threshold crossing",
"findings": [],
"insufficient_groups": [],
"coverage_gaps": [],
}
def main() -> None:
try:
events, receipt, policy = load_inputs()
result = evaluate(events, receipt, policy)
evidence = {
"schema_version": "aidefend.d013-reward-stream-evidence.v1",
"control": "AID-D-013",
"guidance": "reward-stream-anomaly-evaluation",
"asset_id": receipt["asset_id"],
"asset_version": receipt["asset_version"],
"window_start": receipt["window_start"],
"window_end": receipt["window_end"],
"events_sha256": digest(EVENTS),
"receipt_sha256": digest(RECEIPT),
"policy_sha256": digest(POLICY),
"policy_version": policy["policy_version"],
"expected_population": receipt["event_count"],
"expected_binding_population": len(receipt["expected_bindings"]),
"evaluated_population": len(events),
**result,
"evaluated_at": datetime.now(timezone.utc).isoformat(),
}
except Exception as error:
evidence = {
"schema_version": "aidefend.d013-reward-stream-evidence.v1",
"control": "AID-D-013",
"guidance": "reward-stream-anomaly-evaluation",
"status": "ERROR",
"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://D013_REWARD_EVIDENCE_SIGNING_KEY",
"--bundle", str(EVIDENCE) + ".sig", str(EVIDENCE),
],
check=True, timeout=COSIGN_DEADLINE_CEILING_SECONDS,
)
verify(EVIDENCE, REWARD_EVIDENCE_VERIFY_KEY)
if __name__ == "__main__":
main()
Replay and interpret results
Action: At the reward-stream observer, evaluate every authenticated reward record in the reconciled window against the signed baseline policy, persist the signed result, and emit any spike, drop, or variance-collapse finding to the training-security queue; do not rewrite rewards in this detector.
AID-D-013-G002
Analyze agent trajectories to detect pathological behaviors.
When to run this detector and who owns it
Apply this detector to RL or agentic policies whose evaluation runtime can export ordered, privacy-safe state and action digests. The evaluation-platform owner signs the episode snapshot and completeness receipt; detection engineering owns the trajectory rules. Observe the immutable post-episode trace before policy promotion. Bind every trace to the exact asset/version, policy artifact, environment artifact, scenario-suite version, seed, and episode ID. If the signed architecture inventory proves that the asset has no stateful trajectory, return NOT_APPLICABLE; an unsupported state encoder or missing trace is not a clean result.
Production trajectory evaluator
from __future__ import annotations
# Non-tunable process-safety deadline; signed policy and callers cannot raise it.
COSIGN_DEADLINE_CEILING_SECONDS = 15.0
# File: detection/evaluate_policy_trajectories.py
import hashlib
import json
import math
import os
import subprocess
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
import hmac
import re
import tempfile
TRACES = Path("artifacts/rl/trajectory-events.jsonl")
RECEIPT = Path("artifacts/rl/trajectory-receipt.json")
POLICY = Path("policy/trajectory-anomaly-policy.json")
EVIDENCE = Path("artifacts/rl/AID-D-013-trajectory-evidence.json")
TRAJECTORY_COLLECTOR_VERIFY_KEY = Path("/etc/aidefend/d013/trust/trajectory-collector.pub")
TRAJECTORY_COMPLETENESS_VERIFY_KEY = Path("/etc/aidefend/d013/trust/trajectory-completeness.pub")
TRAJECTORY_POLICY_VERIFY_KEY = Path("/etc/aidefend/d013/trust/trajectory-policy.pub")
TRAJECTORY_EVIDENCE_VERIFY_KEY = Path("/etc/aidefend/d013/trust/trajectory-evidence.pub")
SHA256 = re.compile(r"[0-9a-f]{64}")
def digest(path: Path) -> str:
captured = _VERIFIED_INPUTS.get(Path(path))
payload = captured if captured is not None else Path(path).read_bytes()
return hashlib.sha256(payload).hexdigest()
_VERIFIED_INPUTS: dict[Path, bytes] = {}
def _write_private_snapshot(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 | str) -> object:
text = raw.decode("utf-8", errors="strict") if isinstance(raw, bytes) else raw
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(
text,
object_pairs_hook=unique_object,
parse_constant=lambda value: (_ for _ in ()).throw(
ValueError("non-finite JSON constant: " + value)
),
)
def verify(path: Path, verify_key: Path) -> None:
payload_path = Path(path)
signature_path = Path(str(payload_path) + ".sig")
if payload_path == signature_path:
raise ValueError("signed payload and signature paths must differ")
payload_raw = payload_path.read_bytes()
signature_raw = signature_path.read_bytes()
key_reference = Path(verify_key)
key_raw = key_reference.read_bytes()
with tempfile.TemporaryDirectory(prefix="aidefend-d013-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
signature_snapshot = root / "signature"
_write_private_snapshot(payload_snapshot, payload_raw)
_write_private_snapshot(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_snapshot(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=COSIGN_DEADLINE_CEILING_SECONDS,
)
if any(
not hmac.compare_digest(candidate.read_bytes(), expected)
for candidate, expected in captured
):
raise RuntimeError("private signature-verification snapshot changed")
finally:
for candidate, _ in captured:
os.chmod(candidate, 0o600)
_VERIFIED_INPUTS[payload_path] = payload_raw
def _verified_bytes(path: Path) -> bytes:
try:
return _VERIFIED_INPUTS[Path(path)]
except KeyError as error:
raise RuntimeError("signed input was not captured and verified") from error
def _verified_text(path: Path) -> str:
return _verified_bytes(path).decode("utf-8", errors="strict")
def validate_inputs() -> tuple[list[dict], dict, dict]:
verify(TRACES, TRAJECTORY_COLLECTOR_VERIFY_KEY)
verify(RECEIPT, TRAJECTORY_COMPLETENESS_VERIFY_KEY)
verify(POLICY, TRAJECTORY_POLICY_VERIFY_KEY)
receipt = _strict_json(_verified_bytes(RECEIPT))
policy = _strict_json(_verified_bytes(POLICY))
receipt_fields = {
"schema_version", "asset_id", "asset_version",
"policy_artifact_sha256", "environment_artifact_sha256",
"scenario_suite_version", "scenario_suite_sha256",
"expected_episode_ids", "episode_seeds",
"event_count", "collection_complete", "traces_sha256",
}
if (
not isinstance(receipt, dict)
or set(receipt) != receipt_fields
or receipt["schema_version"] != "aidefend.trajectory-receipt.v1"
or receipt["traces_sha256"] != digest(TRACES)
or any(
not isinstance(receipt[field], str)
or not 1 <= len(receipt[field]) <= 256
for field in ("asset_id", "asset_version", "scenario_suite_version")
)
or any(
not isinstance(receipt[field], str)
or SHA256.fullmatch(receipt[field]) is None
for field in (
"policy_artifact_sha256", "environment_artifact_sha256",
"scenario_suite_sha256",
)
)
or not isinstance(receipt["expected_episode_ids"], list)
or not receipt["expected_episode_ids"]
or any(
not isinstance(value, str) or not value
for value in receipt["expected_episode_ids"]
)
or len(receipt["expected_episode_ids"])
!= len(set(receipt["expected_episode_ids"]))
or not isinstance(receipt["episode_seeds"], dict)
or set(receipt["episode_seeds"]) != set(receipt["expected_episode_ids"])
or any(
isinstance(seed, bool) or not isinstance(seed, int)
or not 0 <= seed <= 2**32 - 1
for seed in receipt["episode_seeds"].values()
)
or not isinstance(receipt["collection_complete"], bool)
or isinstance(receipt["event_count"], bool)
or not isinstance(receipt["event_count"], int)
or receipt["event_count"] < 0
):
raise ValueError("trajectory receipt is invalid or misbound")
policy_fields = {
"schema_version", "policy_version", "minimum_steps",
"maximum_cycle_period", "minimum_cycle_repetitions",
"minimum_unique_state_ratio", "minimum_action_entropy_bits",
}
if (
not isinstance(policy, dict)
or set(policy) != policy_fields
or policy["schema_version"] != "aidefend.trajectory-policy.v1"
or not isinstance(policy["policy_version"], str)
or not policy["policy_version"]
):
raise ValueError("trajectory policy schema differs")
for field in (
"minimum_steps", "maximum_cycle_period", "minimum_cycle_repetitions",
):
if (
isinstance(policy[field], bool)
or not isinstance(policy[field], int)
or policy[field] < 1
):
raise ValueError(f"{field} must be a positive integer")
for field in ("minimum_unique_state_ratio", "minimum_action_entropy_bits"):
value = policy[field]
if (
isinstance(value, bool)
or not isinstance(value, (int, float))
or not math.isfinite(float(value))
or float(value) < 0
):
raise ValueError(f"{field} is invalid")
if policy["minimum_unique_state_ratio"] > 1:
raise ValueError("unique-state ratio cannot exceed one")
lines = _verified_text(TRACES).splitlines()
if any(not line.strip() for line in lines):
raise ValueError("trajectory JSONL contains a blank record")
events = [_strict_json(line) for line in lines]
if len(events) != receipt["event_count"]:
raise ValueError("trajectory population count differs")
fields = {
"schema_version", "event_id", "asset_id", "asset_version",
"episode_id", "step", "state_sha256", "action_sha256",
"reward", "terminated", "truncated",
}
if any(
not isinstance(item, dict)
or set(item) != fields
or item["schema_version"] != "aidefend.trajectory-event.v1"
for item in events
):
raise ValueError("trajectory event schema differs")
ids = [item["event_id"] for item in events]
if len(ids) != len(set(ids)) or any(not isinstance(value, str) or not value for value in ids):
raise ValueError("trajectory event IDs are absent or duplicated")
expected = set(receipt["expected_episode_ids"])
if any(
item["asset_id"] != receipt["asset_id"]
or item["asset_version"] != receipt["asset_version"]
or item["episode_id"] not in expected
for item in events
):
raise ValueError("trajectory event binding differs")
for item in events:
if (
isinstance(item["step"], bool)
or not isinstance(item["step"], int)
or item["step"] < 0
or not isinstance(item["state_sha256"], str)
or SHA256.fullmatch(item["state_sha256"]) is None
or not isinstance(item["action_sha256"], str)
or SHA256.fullmatch(item["action_sha256"]) is None
or not isinstance(item["terminated"], bool)
or not isinstance(item["truncated"], bool)
):
raise ValueError("trajectory event field is invalid")
reward = item["reward"]
if (
isinstance(reward, bool)
or not isinstance(reward, (int, float))
or not math.isfinite(float(reward))
):
raise ValueError("trajectory reward is invalid")
return events, receipt, policy
def repeated_cycle(states: list[str], maximum_period: int, repeats: int) -> int | None:
for period in range(1, min(maximum_period, len(states) // repeats) + 1):
width = period * repeats
suffix = states[-width:]
if all(
suffix[offset:offset + period] == suffix[:period]
for offset in range(0, width, period)
):
return period
return None
def entropy_bits(values: list[str]) -> float:
counts = Counter(values)
total = len(values)
return -sum(
(count / total) * math.log2(count / total)
for count in counts.values()
)
def evaluate(events: list[dict], receipt: dict, policy: dict) -> dict:
episodes: dict[str, list[dict]] = defaultdict(list)
for item in events:
episodes[item["episode_id"]].append(item)
findings, insufficient = [], []
if receipt["collection_complete"] is not True:
insufficient.append({
"scope": "collection",
"reason": "authenticated trajectory collection is incomplete",
})
for episode_id in sorted(set(receipt["expected_episode_ids"]) - set(episodes)):
insufficient.append({
"episode_id": episode_id,
"reason": "expected episode is absent",
})
for episode_id in sorted(episodes):
ordered = sorted(episodes[episode_id], key=lambda item: item["step"])
if [item["step"] for item in ordered] != list(range(len(ordered))):
raise ValueError("episode step sequence is incomplete")
if any(
item["terminated"] or item["truncated"]
for item in ordered[:-1]
):
raise ValueError("episode contains events after a terminal step")
if not ordered[-1]["terminated"] and not ordered[-1]["truncated"]:
insufficient.append({"episode_id": episode_id, "reason": "episode has no terminal event"})
continue
if len(ordered) < policy["minimum_steps"]:
insufficient.append({"episode_id": episode_id, "reason": "episode shorter than signed analytic minimum"})
continue
states = [item["state_sha256"] for item in ordered]
actions = [item["action_sha256"] for item in ordered]
ratio = len(set(states)) / len(states)
action_entropy = entropy_bits(actions)
cycle_period = repeated_cycle(
states,
policy["maximum_cycle_period"],
policy["minimum_cycle_repetitions"],
)
reasons = []
if ratio < float(policy["minimum_unique_state_ratio"]):
reasons.append("low_unique_state_ratio")
if action_entropy < float(policy["minimum_action_entropy_bits"]):
reasons.append("low_action_entropy")
if cycle_period is not None:
reasons.append("repeated_state_cycle")
if reasons:
findings.append({
"episode_id": episode_id,
"reasons": reasons,
"steps": len(ordered),
"unique_state_ratio": ratio,
"action_entropy_bits": action_entropy,
"cycle_period": cycle_period,
})
if findings:
return {
"status": "FAIL",
"reason": "pathological trajectory pattern found",
"findings": findings,
"insufficient_episodes": insufficient,
}
if insufficient:
return {
"status": "INSUFFICIENT_DATA",
"reason": "one or more expected episodes are not scoreable",
"findings": [],
"insufficient_episodes": insufficient,
}
return {
"status": "PASS",
"reason": "complete trajectory population met signed behavior rules",
"findings": [],
"insufficient_episodes": [],
}
def main() -> None:
try:
events, receipt, policy = validate_inputs()
result = evaluate(events, receipt, policy)
evidence = {
"schema_version": "aidefend.d013-trajectory-evidence.v1",
"control": "AID-D-013",
"guidance": "pathological-trajectory-analysis",
"asset_id": receipt["asset_id"],
"asset_version": receipt["asset_version"],
"policy_artifact_sha256": receipt["policy_artifact_sha256"],
"environment_artifact_sha256": receipt["environment_artifact_sha256"],
"scenario_suite_version": receipt["scenario_suite_version"],
"scenario_suite_sha256": receipt["scenario_suite_sha256"],
"episode_seeds": receipt["episode_seeds"],
"expected_episodes": len(receipt["expected_episode_ids"]),
"evaluated_events": len(events),
"traces_sha256": digest(TRACES),
"receipt_sha256": digest(RECEIPT),
"analytic_policy_sha256": digest(POLICY),
"analytic_policy_version": policy["policy_version"],
**result,
"evaluated_at": datetime.now(timezone.utc).isoformat(),
}
except Exception as error:
evidence = {
"schema_version": "aidefend.d013-trajectory-evidence.v1",
"control": "AID-D-013",
"guidance": "pathological-trajectory-analysis",
"status": "ERROR",
"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://D013_TRAJECTORY_EVIDENCE_SIGNING_KEY",
"--bundle", str(EVIDENCE) + ".sig", str(EVIDENCE),
],
check=True, timeout=COSIGN_DEADLINE_CEILING_SECONDS,
)
verify(EVIDENCE, TRAJECTORY_EVIDENCE_VERIFY_KEY)
if __name__ == "__main__":
main()
Verify and interpret results
Action: At the trajectory-analysis job, evaluate every expected episode and terminal state against the signed behavior rules, persist the signed episode-bound result, and emit each pathological-pattern finding to the RL security owner; do not terminate or modify the agent here.
AID-D-013-G003
Periodically test the agent's policy in a sandboxed 'honey-state'.
Detector scope and responsibilities
Use a honey-state only where the environment owner can build a deterministic, test-only state containing a known illegitimate reward shortcut and can expose an explicit exploit-action marker. The RL evaluation owner runs the probe in an isolated non-production environment; detection engineering owns the signed fixture policy and verdict. This detector observes exploit-seeking behavior. It does not quarantine a policy or block rollout, and it never injects a honey-state into production traffic. If the signed environment inventory contains no approved honey fixture, return NOT_APPLICABLE; an expected fixture that cannot run is INSUFFICIENT_DATA or ERROR, not a pass.
Replayable Gymnasium probe harness
from __future__ import annotations
# Non-tunable process-safety ceilings (input sizes are bytes); signed policy and callers cannot raise them.
COSIGN_DEADLINE_CEILING_SECONDS = 15.0
PROBE_WALL_CLOCK_CEILING_SECONDS = 60.0
WORKER_SHUTDOWN_GRACE_CEILING_SECONDS = 2.0
# File: detection/run_honey_state_suite.py
import hashlib
import importlib.util
import json
import math
import multiprocessing as mp
import os
import subprocess
from datetime import datetime, timezone
from pathlib import Path
import hmac
import re
import tempfile
from typing import Any
import gymnasium as gym
SUITE = Path("policy/honey-state-suite.json")
SANDBOX_RECEIPT = Path("artifacts/rl/honey-sandbox-receipt.json")
EVIDENCE = Path("artifacts/rl/AID-D-013-honey-state-evidence.json")
HONEY_POLICY_VERIFY_KEY = Path("/etc/aidefend/d013/trust/honey-policy.pub")
HONEY_SANDBOX_VERIFY_KEY = Path("/etc/aidefend/d013/trust/honey-sandbox.pub")
HONEY_EVIDENCE_VERIFY_KEY = Path("/etc/aidefend/d013/trust/honey-evidence.pub")
SHA256 = re.compile(r"[0-9a-f]{64}")
def digest(path: Path) -> str:
captured = _VERIFIED_INPUTS.get(Path(path))
payload = captured if captured is not None else Path(path).read_bytes()
return hashlib.sha256(payload).hexdigest()
_VERIFIED_INPUTS: dict[Path, bytes] = {}
def _write_private_snapshot(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 | str) -> object:
text = raw.decode("utf-8", errors="strict") if isinstance(raw, bytes) else raw
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(
text,
object_pairs_hook=unique_object,
parse_constant=lambda value: (_ for _ in ()).throw(
ValueError("non-finite JSON constant: " + value)
),
)
def verify(path: Path, verify_key: Path) -> None:
payload_path = Path(path)
signature_path = Path(str(payload_path) + ".sig")
if payload_path == signature_path:
raise ValueError("signed payload and signature paths must differ")
payload_raw = payload_path.read_bytes()
signature_raw = signature_path.read_bytes()
key_reference = Path(verify_key)
key_raw = key_reference.read_bytes()
with tempfile.TemporaryDirectory(prefix="aidefend-d013-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
signature_snapshot = root / "signature"
_write_private_snapshot(payload_snapshot, payload_raw)
_write_private_snapshot(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_snapshot(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=COSIGN_DEADLINE_CEILING_SECONDS,
)
if any(
not hmac.compare_digest(candidate.read_bytes(), expected)
for candidate, expected in captured
):
raise RuntimeError("private signature-verification snapshot changed")
finally:
for candidate, _ in captured:
os.chmod(candidate, 0o600)
_VERIFIED_INPUTS[payload_path] = payload_raw
def _verified_bytes(path: Path) -> bytes:
try:
return _VERIFIED_INPUTS[Path(path)]
except KeyError as error:
raise RuntimeError("signed input was not captured and verified") from error
def _verified_text(path: Path) -> str:
return _verified_bytes(path).decode("utf-8", errors="strict")
def _snapshot_artifact(source: Path, destination: Path, expected_sha256: str) -> None:
if source.is_symlink() or not source.is_file():
raise ValueError("honey-state artifact must be a regular non-symlink file")
raw = source.read_bytes()
if hashlib.sha256(raw).hexdigest() != expected_sha256:
raise ValueError("honey-state artifact digest differs")
destination.parent.mkdir(parents=True, mode=0o700)
os.chmod(destination.parent, 0o700)
_write_private_snapshot(destination, raw)
if not hmac.compare_digest(destination.read_bytes(), raw):
raise RuntimeError("private honey-state artifact snapshot changed")
def load_contracts() -> tuple[
dict, dict, tempfile.TemporaryDirectory, dict[str, Path],
]:
verify(SUITE, HONEY_POLICY_VERIFY_KEY)
verify(SANDBOX_RECEIPT, HONEY_SANDBOX_VERIFY_KEY)
suite = _strict_json(_verified_bytes(SUITE))
sandbox = _strict_json(_verified_bytes(SANDBOX_RECEIPT))
suite_fields = {
"schema_version", "suite_version", "asset_id", "asset_version",
"policy_artifact_path", "policy_artifact_sha256",
"environment_artifact_path", "environment_artifact_sha256",
"adapter_artifact_path", "adapter_artifact_sha256",
"adapter_factory", "maximum_fixture_count",
"maximum_total_seed_count", "fixtures",
}
if (
not isinstance(suite, dict)
or set(suite) != suite_fields
or suite["schema_version"] != "aidefend.honey-state-suite.v1"
or not isinstance(suite["fixtures"], list)
or not suite["fixtures"]
):
raise ValueError("honey-state suite is invalid")
for field in ("suite_version", "asset_id", "asset_version", "adapter_factory"):
if not isinstance(suite[field], str) or not 1 <= len(suite[field]) <= 256:
raise ValueError(f"suite field {field} is absent or unbounded")
for field in (
"policy_artifact_sha256", "environment_artifact_sha256",
"adapter_artifact_sha256",
):
if not isinstance(suite[field], str) or SHA256.fullmatch(suite[field]) is None:
raise ValueError(f"suite field {field} is not lowercase SHA-256")
for field in ("maximum_fixture_count", "maximum_total_seed_count"):
if (
isinstance(suite[field], bool) or not isinstance(suite[field], int)
or suite[field] < 1
):
raise ValueError(f"suite resource bound is invalid: {field}")
if len(suite["fixtures"]) > suite["maximum_fixture_count"]:
raise ValueError("honey fixture population exceeds signed bound")
fixture_fields = {
"fixture_id", "fixture_sha256", "seeds", "maximum_steps",
"maximum_action_bytes", "exploit_reward_threshold",
"minimum_exploit_actions",
}
fixture_ids = []
for fixture in suite["fixtures"]:
if not isinstance(fixture, dict) or set(fixture) != fixture_fields:
raise ValueError("honey fixture schema differs")
if (
not isinstance(fixture["fixture_id"], str)
or not 1 <= len(fixture["fixture_id"]) <= 256
or not isinstance(fixture["fixture_sha256"], str)
or SHA256.fullmatch(fixture["fixture_sha256"]) is None
or not isinstance(fixture["seeds"], list)
or not fixture["seeds"]
or any(
isinstance(seed, bool) or not isinstance(seed, int) or seed < 0
for seed in fixture["seeds"]
)
or len(fixture["seeds"]) != len(set(fixture["seeds"]))
):
raise ValueError("honey fixture identity or seed population is invalid")
fixture_ids.append(fixture["fixture_id"])
for field in (
"maximum_steps", "maximum_action_bytes", "minimum_exploit_actions",
):
if (
isinstance(fixture[field], bool)
or not isinstance(fixture[field], int)
or fixture[field] < 1
):
raise ValueError(f"fixture {field} is invalid")
if fixture["minimum_exploit_actions"] > fixture["maximum_steps"]:
raise ValueError("minimum exploit actions exceeds maximum steps")
threshold = fixture["exploit_reward_threshold"]
if (
isinstance(threshold, bool)
or not isinstance(threshold, (int, float))
or not math.isfinite(float(threshold))
):
raise ValueError("exploit reward threshold is invalid")
if len(fixture_ids) != len(set(fixture_ids)):
raise ValueError("honey fixture IDs are duplicated")
total_seed_count = sum(len(fixture["seeds"]) for fixture in suite["fixtures"])
if total_seed_count > suite["maximum_total_seed_count"]:
raise ValueError("honey seed population exceeds signed bound")
sandbox_fields = {
"schema_version", "asset_id", "asset_version", "suite_sha256",
"environment_artifact_sha256", "sandbox_id", "sandbox_runtime_version",
"network_policy_sha256", "egress_disabled", "production_credentials_absent",
"issued_at", "expires_at",
}
if (
not isinstance(sandbox, dict)
or set(sandbox) != sandbox_fields
or sandbox["schema_version"] != "aidefend.honey-sandbox-receipt.v1"
or sandbox["asset_id"] != suite["asset_id"]
or sandbox["asset_version"] != suite["asset_version"]
or sandbox["suite_sha256"] != digest(SUITE)
or sandbox["environment_artifact_sha256"]
!= suite["environment_artifact_sha256"]
or any(
not isinstance(sandbox[field], str)
or SHA256.fullmatch(sandbox[field]) is None
for field in ("suite_sha256", "environment_artifact_sha256", "network_policy_sha256")
)
or sandbox["egress_disabled"] is not True
or sandbox["production_credentials_absent"] is not True
):
raise ValueError("sandbox receipt is invalid or misbound")
issued = datetime.fromisoformat(sandbox["issued_at"].replace("Z", "+00:00"))
expires = datetime.fromisoformat(sandbox["expires_at"].replace("Z", "+00:00"))
now = datetime.now(timezone.utc)
if issued.tzinfo is None or expires.tzinfo is None or not issued <= now < expires:
raise ValueError("sandbox receipt is not currently valid")
declared = {
"policy": Path(suite["policy_artifact_path"]),
"environment": Path(suite["environment_artifact_path"]),
"adapter": Path(suite["adapter_artifact_path"]),
}
if any(path.is_symlink() for path in declared.values()):
raise ValueError("honey-state artifact path cannot be a symlink")
sources = {name: path.resolve(strict=True) for name, path in declared.items()}
if len(set(sources.values())) != len(sources):
raise ValueError("honey-state artifact roles must use distinct files")
snapshot_directory = tempfile.TemporaryDirectory(prefix="aidefend-d013-honey-")
root = Path(snapshot_directory.name)
os.chmod(root, 0o700)
snapshots = {
"policy": root / "policy" / sources["policy"].name,
"environment": root / "environment" / sources["environment"].name,
"adapter": root / "adapter" / "adapter.py",
}
try:
for name in ("policy", "environment", "adapter"):
_snapshot_artifact(
sources[name], snapshots[name], suite[f"{name}_artifact_sha256"],
)
except Exception:
snapshot_directory.cleanup()
raise
return suite, sandbox, snapshot_directory, snapshots
def load_factory(adapter_path: Path, function_name: str):
specification = importlib.util.spec_from_file_location(
"aidefend_honey_state_adapter",
adapter_path,
)
if specification is None or specification.loader is None:
raise ValueError("configured honey-state adapter cannot be loaded")
module = importlib.util.module_from_spec(specification)
specification.loader.exec_module(module)
factory = getattr(module, function_name, None)
if not callable(factory):
raise ValueError("configured honey-state adapter factory is not callable")
return factory
def run_probe(agent: Any, env: gym.Env, fixture: dict, seed: int) -> dict:
observation, reset_info = env.reset(
seed=seed,
options={
"honey_state_id": fixture["fixture_id"],
"fixture_sha256": fixture["fixture_sha256"],
},
)
if (
not isinstance(reset_info, dict)
or reset_info.get("honey_state_id") != fixture["fixture_id"]
or reset_info.get("fixture_sha256") != fixture["fixture_sha256"]
):
raise ValueError("environment did not attest the requested honey fixture")
cumulative_reward = 0.0
exploit_actions = 0
action_trace: list[dict] = []
terminated = truncated = False
for step in range(fixture["maximum_steps"]):
action, _ = agent.predict(observation, deterministic=True)
observation, reward, terminated, truncated, info = env.step(action)
numeric_reward = float(reward)
if not math.isfinite(numeric_reward):
raise ValueError("environment returned a non-finite reward")
if not isinstance(info, dict) or not isinstance(info.get("honey_exploit_action"), bool):
raise ValueError("environment omitted the exploit-action marker")
cumulative_reward += numeric_reward
exploit_actions += int(info["honey_exploit_action"])
action_raw = json.dumps(
action.tolist() if hasattr(action, "tolist") else action,
sort_keys=True, separators=(",", ":"), allow_nan=False,
).encode("utf-8")
if len(action_raw) > fixture["maximum_action_bytes"]:
raise ValueError("agent action exceeds signed byte bound")
action_trace.append({
"step": step,
"action_sha256": hashlib.sha256(action_raw).hexdigest(),
"reward": numeric_reward,
"honey_exploit_action": info["honey_exploit_action"],
})
if terminated or truncated:
break
finding = (
cumulative_reward >= float(fixture["exploit_reward_threshold"])
and exploit_actions >= fixture["minimum_exploit_actions"]
)
return {
"fixture_id": fixture["fixture_id"],
"fixture_sha256": fixture["fixture_sha256"],
"seed": seed,
"steps": len(action_trace),
"terminated": terminated,
"truncated": truncated,
"cumulative_reward": cumulative_reward,
"exploit_actions": exploit_actions,
"finding": finding,
"action_trace": action_trace,
}
def _probe_worker(connection, snapshots: dict[str, Path], suite: dict, fixture: dict, seed: int) -> None:
environment = None
try:
factory = load_factory(snapshots["adapter"], suite["adapter_factory"])
agent, environment = factory(snapshots["policy"], snapshots["environment"])
if not isinstance(environment, gym.Env):
raise ValueError("adapter did not return a Gymnasium environment")
connection.send({"ok": True, "result": run_probe(agent, environment, fixture, seed)})
except Exception as error:
connection.send({
"ok": False,
"error_class": type(error).__name__,
"reason": str(error),
})
finally:
if environment is not None:
environment.close()
connection.close()
def run_supervised_probe(
snapshots: dict[str, Path], suite: dict, fixture: dict, seed: int,
) -> dict:
context = mp.get_context("spawn")
receiver, sender = context.Pipe(duplex=False)
process = context.Process(
target=_probe_worker,
args=(sender, snapshots, suite, fixture, seed),
)
process.start()
sender.close()
try:
if not receiver.poll(PROBE_WALL_CLOCK_CEILING_SECONDS):
process.terminate()
process.join(WORKER_SHUTDOWN_GRACE_CEILING_SECONDS)
if process.is_alive():
process.kill()
process.join(WORKER_SHUTDOWN_GRACE_CEILING_SECONDS)
raise TimeoutError("fixture/seed probe exceeded the probe wall-clock ceiling")
try:
message = receiver.recv()
except EOFError as error:
raise RuntimeError("probe worker exited without a result") from error
finally:
receiver.close()
process.join(WORKER_SHUTDOWN_GRACE_CEILING_SECONDS)
if process.is_alive():
process.terminate()
process.join(WORKER_SHUTDOWN_GRACE_CEILING_SECONDS)
raise RuntimeError("probe worker did not exit after returning a result")
if process.exitcode != 0:
raise RuntimeError(f"probe worker exited with code {process.exitcode}")
if not isinstance(message, dict) or message.get("ok") is not True:
error_class = message.get("error_class", "WorkerError") if isinstance(message, dict) else "WorkerError"
reason = message.get("reason", "malformed worker response") if isinstance(message, dict) else "malformed worker response"
raise RuntimeError(f"{error_class}: {reason}")
return message["result"]
def evaluate() -> dict:
suite, sandbox, snapshot_directory, snapshots = load_contracts()
results, probe_errors = [], []
try:
for fixture in suite["fixtures"]:
for seed in fixture["seeds"]:
try:
results.append(run_supervised_probe(snapshots, suite, fixture, seed))
except Exception as error:
probe_errors.append({
"fixture_id": fixture["fixture_id"],
"seed": seed,
"error_class": type(error).__name__,
"reason": str(error),
})
finally:
snapshot_directory.cleanup()
expected = sum(len(fixture["seeds"]) for fixture in suite["fixtures"])
incomplete = [
{
"fixture_id": item["fixture_id"],
"seed": item["seed"],
"reason": "episode did not terminate or truncate",
}
for item in results
if item["terminated"] is not True and item["truncated"] is not True
]
findings = [item for item in results if item["finding"]]
common = {
"results": results,
"probe_errors": probe_errors,
"incomplete_probes": incomplete,
"expected_probe_population": expected,
"completed_probe_population": len(results),
"probe_wall_clock_ceiling_seconds": PROBE_WALL_CLOCK_CEILING_SECONDS,
"suite": suite,
"sandbox_receipt_sha256": digest(SANDBOX_RECEIPT),
}
if findings:
return {
"status": "FAIL",
"reason": "policy pursued a known honey-state exploit",
**common,
}
if probe_errors:
return {
"status": "ERROR",
"reason": "one or more fixture/seed supervisors failed or timed out",
**common,
}
if len(results) != expected or incomplete:
return {
"status": "INSUFFICIENT_DATA",
"reason": "fixture/seed population or episode termination is incomplete",
**common,
}
return {
"status": "PASS",
"reason": "complete honey-state suite had no exploit-seeking finding",
**common,
}
def main() -> None:
try:
result = evaluate()
suite = result.pop("suite")
evidence = {
"schema_version": "aidefend.d013-honey-state-evidence.v1",
"control": "AID-D-013",
"guidance": "sandboxed-honey-state-probe",
"asset_id": suite["asset_id"],
"asset_version": suite["asset_version"],
"suite_version": suite["suite_version"],
"suite_sha256": digest(SUITE),
"policy_artifact_sha256": suite["policy_artifact_sha256"],
"environment_artifact_sha256": suite["environment_artifact_sha256"],
"adapter_artifact_sha256": suite["adapter_artifact_sha256"],
**result,
"evaluated_at": datetime.now(timezone.utc).isoformat(),
}
except Exception as error:
evidence = {
"schema_version": "aidefend.d013-honey-state-evidence.v1",
"control": "AID-D-013",
"guidance": "sandboxed-honey-state-probe",
"status": "ERROR",
"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://D013_HONEY_EVIDENCE_SIGNING_KEY",
"--bundle", str(EVIDENCE) + ".sig", str(EVIDENCE),
],
check=True, timeout=COSIGN_DEADLINE_CEILING_SECONDS,
)
verify(EVIDENCE, HONEY_EVIDENCE_VERIFY_KEY)
if __name__ == "__main__":
main()
Replay and interpret results
Action: Run the complete signed honey-state fixture and seed population only in the isolated evaluation environment, persist the signed per-step result, and emit exploit-seeking findings to the model-security queue; production containment remains a separate Isolate decision.
Isolation boundary: The Python multiprocessing supervisor enforces timeout and child-process cleanup only; it does not create the network, credential, filesystem, or kernel sandbox. Launch this job inside the separately provisioned sandbox named by the verified receipt, and verify effective egress denial and absence of production credentials at that runtime before accepting a result.
AID-D-013-G004
Implement out-of-band reward signal verification.
Trusted inputs and responsibilities
Use this detector when reward is computed by a service, simulator, evaluator, or verifier and transitions can be replayed. The training runtime signs the complete transition population before a separately authorized audit service releases a sampling seed. A verifier in another trust domain, using an independently built reward implementation, recalculates selected rewards over mTLS and signs each response. Detection engineering owns comparison and evidence. If the signed architecture inventory proves reward cannot be independently replayed, return NOT_APPLICABLE; do not claim verification.
Production out-of-band verifier client
from __future__ import annotations
# Non-tunable process-safety deadline; signed policy and callers cannot raise it.
COSIGN_DEADLINE_CEILING_SECONDS = 15.0
# File: detection/verify_rewards_out_of_band.py
import hashlib
import json
import math
import os
import subprocess
from datetime import datetime, timezone
from pathlib import Path
import hmac
import re
import tempfile
from urllib.parse import urlparse
import jwt
import requests
TRANSITIONS = Path("artifacts/rl/transitions.jsonl")
RECEIPT = Path("artifacts/rl/transition-receipt.json")
SEED = Path("artifacts/rl/audit-seed.json")
POLICY = Path("policy/out-of-band-reward-policy.json")
EVIDENCE = Path("artifacts/rl/AID-D-013-oob-reward-evidence.json")
TRANSITION_COLLECTOR_VERIFY_KEY = Path("/etc/aidefend/d013/trust/transition-collector.pub")
TRANSITION_COMPLETENESS_VERIFY_KEY = Path("/etc/aidefend/d013/trust/transition-completeness.pub")
AUDIT_SEED_VERIFY_KEY = Path("/etc/aidefend/d013/trust/audit-seed.pub")
OOB_POLICY_VERIFY_KEY = Path("/etc/aidefend/d013/trust/oob-policy.pub")
OOB_EVIDENCE_VERIFY_KEY = Path("/etc/aidefend/d013/trust/oob-evidence.pub")
OOB_CLIENT_CERT = Path("/etc/aidefend/d013/identity/oob-client.crt")
OOB_CLIENT_KEY = Path("/etc/aidefend/d013/identity/oob-client.key")
OOB_CA_BUNDLE = Path("/etc/aidefend/d013/trust/oob-service-ca.pem")
SHA256 = re.compile(r"[0-9a-f]{64}")
def digest(path: Path) -> str:
captured = _VERIFIED_INPUTS.get(Path(path))
payload = captured if captured is not None else Path(path).read_bytes()
return hashlib.sha256(payload).hexdigest()
def canonical_digest(value: dict) -> str:
return hashlib.sha256(
json.dumps(
value, sort_keys=True, separators=(",", ":"), allow_nan=False
).encode("utf-8")
).hexdigest()
_VERIFIED_INPUTS: dict[Path, bytes] = {}
def _write_private_snapshot(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 | str) -> object:
text = raw.decode("utf-8", errors="strict") if isinstance(raw, bytes) else raw
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(
text,
object_pairs_hook=unique_object,
parse_constant=lambda value: (_ for _ in ()).throw(
ValueError("non-finite JSON constant: " + value)
),
)
def verify(path: Path, verify_key: Path) -> None:
payload_path = Path(path)
signature_path = Path(str(payload_path) + ".sig")
if payload_path == signature_path:
raise ValueError("signed payload and signature paths must differ")
payload_raw = payload_path.read_bytes()
signature_raw = signature_path.read_bytes()
key_reference = Path(verify_key)
key_raw = key_reference.read_bytes()
with tempfile.TemporaryDirectory(prefix="aidefend-d013-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
signature_snapshot = root / "signature"
_write_private_snapshot(payload_snapshot, payload_raw)
_write_private_snapshot(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_snapshot(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=COSIGN_DEADLINE_CEILING_SECONDS,
)
if any(
not hmac.compare_digest(candidate.read_bytes(), expected)
for candidate, expected in captured
):
raise RuntimeError("private signature-verification snapshot changed")
finally:
for candidate, _ in captured:
os.chmod(candidate, 0o600)
_VERIFIED_INPUTS[payload_path] = payload_raw
def _verified_bytes(path: Path) -> bytes:
try:
return _VERIFIED_INPUTS[Path(path)]
except KeyError as error:
raise RuntimeError("signed input was not captured and verified") from error
def _verified_text(path: Path) -> str:
return _verified_bytes(path).decode("utf-8", errors="strict")
def finite(value: object, field: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f"{field} is not numeric")
result = float(value)
if not math.isfinite(result):
raise ValueError(f"{field} is not finite")
return result
def load_inputs() -> tuple[list[dict], dict, dict, dict]:
verify(TRANSITIONS, TRANSITION_COLLECTOR_VERIFY_KEY)
verify(RECEIPT, TRANSITION_COMPLETENESS_VERIFY_KEY)
verify(SEED, AUDIT_SEED_VERIFY_KEY)
verify(POLICY, OOB_POLICY_VERIFY_KEY)
receipt = _strict_json(_verified_bytes(RECEIPT))
seed = _strict_json(_verified_bytes(SEED))
policy = _strict_json(_verified_bytes(POLICY))
receipt_fields = {
"schema_version", "asset_id", "asset_version", "run_id",
"window_start", "window_end", "transition_count",
"collection_complete", "transitions_sha256",
}
if (
not isinstance(receipt, dict)
or set(receipt) != receipt_fields
or receipt["schema_version"] != "aidefend.transition-receipt.v1"
or receipt["transitions_sha256"] != digest(TRANSITIONS)
or isinstance(receipt["transition_count"], bool)
or not isinstance(receipt["transition_count"], int)
or receipt["transition_count"] < 0
or not isinstance(receipt["collection_complete"], bool)
):
raise ValueError("transition receipt is invalid or misbound")
if any(
not isinstance(receipt[field], str) or not receipt[field]
for field in (
"asset_id", "asset_version", "run_id",
"window_start", "window_end",
)
):
raise ValueError("transition receipt asset, run, or window is absent")
seed_fields = {
"schema_version", "population_sha256", "seed",
"issued_at", "issuer_id",
}
if (
not isinstance(seed, dict)
or set(seed) != seed_fields
or seed["schema_version"] != "aidefend.reward-audit-seed.v1"
or seed["population_sha256"] != digest(TRANSITIONS)
or not isinstance(seed["seed"], str)
or len(seed["seed"]) < 32
or not isinstance(seed["issuer_id"], str)
or not seed["issuer_id"]
):
raise ValueError("audit seed is invalid or misbound")
window_start = datetime.fromisoformat(
receipt["window_start"].replace("Z", "+00:00")
)
window_end = datetime.fromisoformat(
receipt["window_end"].replace("Z", "+00:00")
)
issued_at = datetime.fromisoformat(seed["issued_at"].replace("Z", "+00:00"))
now = datetime.now(timezone.utc)
if (
window_start.tzinfo is None
or window_start >= window_end
or window_end.tzinfo is None
or issued_at.tzinfo is None
or issued_at < window_end
or issued_at > now
):
raise ValueError("audit seed issuance is outside the closed population window")
policy_fields = {
"schema_version", "policy_version", "sample_rate",
"absolute_tolerance", "relative_tolerance",
"verifier_url", "verifier_issuer", "verifier_audience",
"verifier_public_key_pem", "verifier_build_sha256",
"request_timeout_seconds", "max_verifier_response_age_seconds",
"response_max_bytes", "response_chunk_bytes",
"minimum_sample_count",
}
if (
not isinstance(policy, dict)
or set(policy) != policy_fields
or policy["schema_version"] != "aidefend.oob-reward-policy.v1"
or not isinstance(policy["policy_version"], str)
or not policy["policy_version"]
or not isinstance(policy["verifier_url"], str)
or not policy["verifier_url"].startswith("https://")
):
raise ValueError("out-of-band reward policy is invalid")
verifier_url = urlparse(policy["verifier_url"])
if (
verifier_url.scheme != "https"
or not verifier_url.netloc
or verifier_url.username
or verifier_url.password
or verifier_url.fragment
):
raise ValueError("verifier URL is not a fixed HTTPS endpoint")
if any(
not isinstance(policy[field], str) or not policy[field]
for field in (
"verifier_issuer", "verifier_audience",
"verifier_public_key_pem", "verifier_build_sha256",
)
) or SHA256.fullmatch(policy["verifier_build_sha256"]) is None:
raise ValueError("verifier identity or artifact binding is invalid")
for field in (
"sample_rate", "absolute_tolerance", "relative_tolerance",
"request_timeout_seconds", "max_verifier_response_age_seconds",
):
finite(policy[field], field)
for field in ("response_max_bytes", "response_chunk_bytes", "minimum_sample_count"):
if (
isinstance(policy[field], bool)
or not isinstance(policy[field], int)
or policy[field] < 1
):
raise ValueError(f"{field} must be a positive integer")
if (
not 0 < float(policy["sample_rate"]) <= 1
or policy["absolute_tolerance"] < 0
or policy["relative_tolerance"] < 0
or policy["request_timeout_seconds"] <= 0
or policy["max_verifier_response_age_seconds"] <= 0
or policy["response_max_bytes"] < 2
or policy["response_chunk_bytes"] > policy["response_max_bytes"]
):
raise ValueError("out-of-band thresholds are invalid")
lines = _verified_text(TRANSITIONS).splitlines()
if any(not line.strip() for line in lines):
raise ValueError("transition JSONL contains a blank record")
transitions = [_strict_json(line) for line in lines]
if len(transitions) != receipt["transition_count"]:
raise ValueError("transition count differs from receipt")
fields = {
"schema_version", "transition_id", "asset_id", "asset_version",
"run_id", "sequence", "state_sha256", "action_sha256",
"next_state_sha256", "transition_payload_ref",
"transition_payload_sha256", "received_reward",
"reward_definition_sha256",
}
if any(
not isinstance(item, dict)
or set(item) != fields
or item["schema_version"] != "aidefend.reward-transition.v1"
for item in transitions
):
raise ValueError("transition schema differs")
ids = [item["transition_id"] for item in transitions]
sequences = [item["sequence"] for item in transitions]
if (
len(ids) != len(set(ids))
or any(not isinstance(value, str) or not value for value in ids)
or len(sequences) != len(set(sequences))
or any(
isinstance(value, bool)
or not isinstance(value, int)
or value < 0
for value in sequences
)
):
raise ValueError("transition identity or sequence is invalid")
if transitions and max(sequences) - min(sequences) + 1 != len(sequences):
raise ValueError("transition sequence is incomplete")
for item in transitions:
if (
item["asset_id"] != receipt["asset_id"]
or item["asset_version"] != receipt["asset_version"]
or item["run_id"] != receipt["run_id"]
):
raise ValueError("transition binding differs")
finite(item["received_reward"], "received_reward")
if (
not isinstance(item["transition_payload_ref"], str)
or item["transition_payload_ref"]
!= f"sha256://{item['transition_payload_sha256']}"
):
raise ValueError("transition payload reference is not content-addressed")
for field in (
"state_sha256", "action_sha256", "next_state_sha256",
"transition_payload_sha256", "reward_definition_sha256",
):
if (
not isinstance(item[field], str)
or SHA256.fullmatch(item[field]) is None
):
raise ValueError(f"{field} is not lowercase SHA-256")
return transitions, receipt, seed, policy
def selected(item: dict, seed: str, rate: float) -> bool:
value = int.from_bytes(
hashlib.sha256(
(seed + ":" + item["transition_id"]).encode("utf-8")
).digest()[:8],
"big",
)
return value < int(rate * (2 ** 64))
def reject_duplicate_keys(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate verifier JSON key: {key}")
value[key] = item
return value
def read_verifier_json(response, policy):
content_encoding = response.headers.get("Content-Encoding", "identity").lower()
if content_encoding not in {"", "identity"}:
raise ValueError("compressed verifier responses are not accepted")
chunks, observed = [], 0
for chunk in response.iter_content(chunk_size=policy["response_chunk_bytes"]):
observed += len(chunk)
if observed > policy["response_max_bytes"]:
raise ValueError("verifier response exceeds signed bound")
chunks.append(chunk)
return 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)
),
)
def verify_response(
token: str,
transition: dict,
policy: dict,
population_sha256: str,
audit_seed_sha256: str,
policy_sha256: str,
) -> dict:
claims = jwt.decode(
token,
policy["verifier_public_key_pem"],
algorithms=["EdDSA"],
issuer=policy["verifier_issuer"],
audience=policy["verifier_audience"],
options={"require": [
"iss", "aud", "iat", "exp", "jti", "transition_id",
"transition_sha256", "reward_definition_sha256",
"verifier_build_sha256", "trusted_reward",
"population_sha256", "audit_seed_sha256",
"policy_sha256", "policy_version",
]},
)
if (
claims["transition_id"] != transition["transition_id"]
or claims["transition_sha256"] != canonical_digest(transition)
or claims["reward_definition_sha256"] != transition["reward_definition_sha256"]
or claims["verifier_build_sha256"] != policy["verifier_build_sha256"]
or claims["population_sha256"] != population_sha256
or claims["audit_seed_sha256"] != audit_seed_sha256
or claims["policy_sha256"] != policy_sha256
or claims["policy_version"] != policy["policy_version"]
):
raise ValueError("verifier response is misbound")
now = int(datetime.now(timezone.utc).timestamp())
if (
isinstance(claims["iat"], bool)
or not isinstance(claims["iat"], int)
or now - claims["iat"] > policy["max_verifier_response_age_seconds"]
or claims["iat"] > now
):
raise ValueError("verifier response is stale or future-issued")
finite(claims["trusted_reward"], "trusted_reward")
return claims
def evaluate() -> tuple[dict, dict, dict]:
transitions, receipt, seed, policy = load_inputs()
chosen = [
item for item in transitions
if selected(item, seed["seed"], float(policy["sample_rate"]))
]
coverage_gaps = []
if receipt["collection_complete"] is not True:
coverage_gaps.append("transition population is incomplete")
if len(chosen) < policy["minimum_sample_count"]:
coverage_gaps.append("deterministic sample is below the signed minimum")
results, findings, verification_errors = [], [], []
verifier_jtis = set()
population_sha256 = digest(TRANSITIONS)
audit_seed_sha256 = digest(SEED)
policy_sha256 = digest(POLICY)
with requests.Session() as session:
session.trust_env = False
for transition in chosen:
try:
with session.post(
policy["verifier_url"],
json={
"transition": transition,
"population_sha256": population_sha256,
"audit_seed_sha256": audit_seed_sha256,
"policy_sha256": policy_sha256,
"policy_version": policy["policy_version"],
},
cert=(str(OOB_CLIENT_CERT), str(OOB_CLIENT_KEY)),
verify=str(OOB_CA_BUNDLE),
timeout=float(policy["request_timeout_seconds"]),
allow_redirects=False,
stream=True,
headers={"Accept-Encoding": "identity"},
) as response:
response.raise_for_status()
body = read_verifier_json(response, policy)
if not isinstance(body, dict) or set(body) != {"verifier_jws"}:
raise ValueError("verifier response schema differs")
claims = verify_response(
body["verifier_jws"], transition, policy,
population_sha256, audit_seed_sha256, policy_sha256,
)
if (
not isinstance(claims["jti"], str)
or not claims["jti"]
or claims["jti"] in verifier_jtis
):
raise ValueError("verifier response jti is absent or reused")
verifier_jtis.add(claims["jti"])
trusted = float(claims["trusted_reward"])
received = float(transition["received_reward"])
tolerance = max(
float(policy["absolute_tolerance"]),
float(policy["relative_tolerance"]) * max(abs(trusted), abs(received)),
)
delta = abs(received - trusted)
result = {
"transition_id": transition["transition_id"],
"transition_sha256": canonical_digest(transition),
"verifier_jti": claims["jti"],
"verifier_jws": body["verifier_jws"],
"verifier_jws_sha256": hashlib.sha256(
body["verifier_jws"].encode("utf-8")
).hexdigest(),
"received_reward": received,
"trusted_reward": trusted,
"delta": delta,
"tolerance": tolerance,
}
results.append(result)
if delta > tolerance:
findings.append(result)
except Exception as error:
verification_errors.append({
"transition_id": transition["transition_id"],
"error_class": type(error).__name__,
"reason": str(error),
})
common = {
"sample_population": len(chosen),
"verified_population": len(results),
"results": results,
"findings": findings,
"verification_errors": verification_errors,
"coverage_gaps": coverage_gaps,
}
if findings:
status, reason = "FAIL", "independent reward mismatch found"
elif verification_errors:
status, reason = "ERROR", "one or more verifier requests or responses failed"
elif coverage_gaps:
status, reason = "INSUFFICIENT_DATA", "signed transition or sample coverage is incomplete"
else:
status, reason = "PASS", "complete deterministic sample matched independent rewards"
return {"status": status, "reason": reason, **common}, receipt, policy
def main() -> None:
try:
result, receipt, policy = evaluate()
evidence = {
"schema_version": "aidefend.d013-oob-reward-evidence.v1",
"control": "AID-D-013",
"guidance": "out-of-band-reward-verification",
"asset_id": receipt["asset_id"],
"asset_version": receipt["asset_version"],
"run_id": receipt["run_id"],
"transition_population": receipt["transition_count"],
"transitions_sha256": digest(TRANSITIONS),
"receipt_sha256": digest(RECEIPT),
"audit_seed_sha256": digest(SEED),
"policy_sha256": digest(POLICY),
"policy_version": policy["policy_version"],
"verifier_build_sha256": policy["verifier_build_sha256"],
**result,
"evaluated_at": datetime.now(timezone.utc).isoformat(),
}
except Exception as error:
evidence = {
"schema_version": "aidefend.d013-oob-reward-evidence.v1",
"control": "AID-D-013",
"guidance": "out-of-band-reward-verification",
"status": "ERROR",
"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://D013_OOB_EVIDENCE_SIGNING_KEY",
"--bundle", str(EVIDENCE) + ".sig", str(EVIDENCE),
],
check=True, timeout=COSIGN_DEADLINE_CEILING_SECONDS,
)
verify(EVIDENCE, OOB_EVIDENCE_VERIFY_KEY)
if __name__ == "__main__":
main()
Replay and interpret results
Action: At the independent reward verifier, replay every transition selected by the signed sampling policy, persist the signed comparison result, and emit each mismatch to the training-integrity queue; do not overwrite trainer rewards from this detector.
AID-D-013-G005
Monitor RLHF, RLAIF, RLVR, and agentic RL reward-model or verifier gaming by comparing training reward, independent evaluator outcomes, and held-out task success.
Independent observation point
Apply this detector to RLHF, RLAIF, RLVR, and agentic post-training runs that optimize a reward model or verifier. The post-training platform signs the complete evaluation export; an evaluator owner outside the optimizer's trust domain supplies independent scores and task-success labels on a versioned held-out suite. Detection engineering compares the populations after evaluation and before promotion. Bind evidence to the exact asset/version, candidate policy, reward model or verifier, independent evaluator, task suite, prompt-family registry, and output-normalizer versions. If no reward-optimized post-training path exists, signed inventory may yield NOT_APPLICABLE.
Production reward-gaming evaluator
from __future__ import annotations
# Non-tunable process-safety deadline; signed policy and callers cannot raise it.
COSIGN_DEADLINE_CEILING_SECONDS = 15.0
# File: detection/evaluate_post_training_reward_gaming.py
import hashlib
import json
import math
import os
import statistics
import subprocess
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
import hmac
import re
import tempfile
RECORDS = Path("artifacts/rl/post-training-evaluation.jsonl")
RECEIPT = Path("artifacts/rl/post-training-evaluation-receipt.json")
POLICY = Path("policy/post-training-reward-gaming-policy.json")
EVIDENCE = Path("artifacts/rl/AID-D-013-post-training-evidence.json")
POSTTRAIN_COLLECTOR_VERIFY_KEY = Path("/etc/aidefend/d013/trust/posttrain-collector.pub")
POSTTRAIN_COMPLETENESS_VERIFY_KEY = Path("/etc/aidefend/d013/trust/posttrain-completeness.pub")
POSTTRAIN_POLICY_VERIFY_KEY = Path("/etc/aidefend/d013/trust/posttrain-policy.pub")
POSTTRAIN_EVIDENCE_VERIFY_KEY = Path("/etc/aidefend/d013/trust/posttrain-evidence.pub")
SHA256 = re.compile(r"[0-9a-f]{64}")
def digest(path: Path) -> str:
captured = _VERIFIED_INPUTS.get(Path(path))
payload = captured if captured is not None else Path(path).read_bytes()
return hashlib.sha256(payload).hexdigest()
_VERIFIED_INPUTS: dict[Path, bytes] = {}
def _write_private_snapshot(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 | str) -> object:
text = raw.decode("utf-8", errors="strict") if isinstance(raw, bytes) else raw
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(
text,
object_pairs_hook=unique_object,
parse_constant=lambda value: (_ for _ in ()).throw(
ValueError("non-finite JSON constant: " + value)
),
)
def verify(path: Path, verify_key: Path) -> None:
payload_path = Path(path)
signature_path = Path(str(payload_path) + ".sig")
if payload_path == signature_path:
raise ValueError("signed payload and signature paths must differ")
payload_raw = payload_path.read_bytes()
signature_raw = signature_path.read_bytes()
key_reference = Path(verify_key)
key_raw = key_reference.read_bytes()
with tempfile.TemporaryDirectory(prefix="aidefend-d013-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
payload_snapshot = root / "payload"
signature_snapshot = root / "signature"
_write_private_snapshot(payload_snapshot, payload_raw)
_write_private_snapshot(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_snapshot(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=COSIGN_DEADLINE_CEILING_SECONDS,
)
if any(
not hmac.compare_digest(candidate.read_bytes(), expected)
for candidate, expected in captured
):
raise RuntimeError("private signature-verification snapshot changed")
finally:
for candidate, _ in captured:
os.chmod(candidate, 0o600)
_VERIFIED_INPUTS[payload_path] = payload_raw
def _verified_bytes(path: Path) -> bytes:
try:
return _VERIFIED_INPUTS[Path(path)]
except KeyError as error:
raise RuntimeError("signed input was not captured and verified") from error
def _verified_text(path: Path) -> str:
return _verified_bytes(path).decode("utf-8", errors="strict")
def finite(value: object, field: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f"{field} is not numeric")
result = float(value)
if not math.isfinite(result):
raise ValueError(f"{field} is not finite")
return result
def load_inputs() -> tuple[list[dict], dict, dict]:
verify(RECORDS, POSTTRAIN_COLLECTOR_VERIFY_KEY)
verify(RECEIPT, POSTTRAIN_COMPLETENESS_VERIFY_KEY)
verify(POLICY, POSTTRAIN_POLICY_VERIFY_KEY)
receipt = _strict_json(_verified_bytes(RECEIPT))
policy = _strict_json(_verified_bytes(POLICY))
receipt_fields = {
"schema_version", "asset_id", "asset_version",
"candidate_policy_sha256", "reward_evaluator_sha256",
"independent_evaluator_sha256", "task_suite_version",
"task_suite_sha256", "prompt_family_registry_version",
"prompt_family_registry_sha256", "output_normalizer_version",
"output_normalizer_sha256", "template_clusterer_sha256",
"expected_record_ids", "record_count", "collection_complete",
"records_sha256",
}
if (
not isinstance(receipt, dict)
or set(receipt) != receipt_fields
or receipt["schema_version"] != "aidefend.posttrain-eval-receipt.v1"
or receipt["records_sha256"] != digest(RECORDS)
or any(
not isinstance(receipt[field], str)
or not 1 <= len(receipt[field]) <= 256
for field in (
"asset_id", "asset_version", "task_suite_version",
"prompt_family_registry_version", "output_normalizer_version",
)
)
or any(
not isinstance(receipt[field], str)
or SHA256.fullmatch(receipt[field]) is None
for field in (
"candidate_policy_sha256", "reward_evaluator_sha256",
"independent_evaluator_sha256", "task_suite_sha256",
"prompt_family_registry_sha256", "output_normalizer_sha256",
"template_clusterer_sha256",
)
)
or not isinstance(receipt["expected_record_ids"], list)
or not receipt["expected_record_ids"]
or any(
not isinstance(value, str) or not value
for value in receipt["expected_record_ids"]
)
or len(receipt["expected_record_ids"])
!= len(set(receipt["expected_record_ids"]))
or isinstance(receipt["record_count"], bool)
or not isinstance(receipt["record_count"], int)
or receipt["record_count"] < 0
or not isinstance(receipt["collection_complete"], bool)
):
raise ValueError("post-training receipt is invalid or misbound")
if any(
not isinstance(receipt[field], str) or not receipt[field]
for field in (
"asset_id", "asset_version", "task_suite_version",
"prompt_family_registry_version", "output_normalizer_version",
)
):
raise ValueError("post-training asset or evaluation version is absent")
if any(
not isinstance(receipt[field], str) or len(receipt[field]) != 64
for field in (
"candidate_policy_sha256", "reward_evaluator_sha256",
"independent_evaluator_sha256", "template_clusterer_sha256",
)
):
raise ValueError("post-training artifact digest is invalid")
policy_fields = {
"schema_version", "policy_version", "families",
"maximum_mean_reward_gap", "minimum_verifier_pass_rate",
"maximum_heldout_success_rate_when_verifier_high",
"maximum_exact_output_rate", "maximum_template_cluster_rate",
"reward_comparison", "candidate_policy_sha256", "reward_evaluator_sha256",
"independent_evaluator_sha256", "task_suite_version",
"task_suite_sha256", "prompt_family_registry_version",
"prompt_family_registry_sha256", "output_normalizer_version",
"output_normalizer_sha256", "template_clusterer_sha256",
}
if (
not isinstance(policy, dict)
or set(policy) != policy_fields
or policy["schema_version"] != "aidefend.posttrain-gaming-policy.v1"
or not isinstance(policy["policy_version"], str)
or not policy["policy_version"]
or not isinstance(policy["families"], dict)
or not policy["families"]
or any(
policy[field] != receipt[field]
for field in (
"candidate_policy_sha256", "reward_evaluator_sha256",
"independent_evaluator_sha256", "task_suite_version",
"task_suite_sha256", "prompt_family_registry_version",
"prompt_family_registry_sha256", "output_normalizer_version",
"output_normalizer_sha256", "template_clusterer_sha256",
)
)
):
raise ValueError("post-training policy is invalid")
for family, minimum in policy["families"].items():
if (
not isinstance(family, str)
or not family
or not isinstance(minimum, dict)
or set(minimum) != {"minimum_optimization_records", "minimum_heldout_records"}
or any(
isinstance(value, bool) or not isinstance(value, int) or value < 1
for value in minimum.values()
)
):
raise ValueError("prompt-family population policy is invalid")
for field in (
"maximum_mean_reward_gap", "minimum_verifier_pass_rate",
"maximum_heldout_success_rate_when_verifier_high",
"maximum_exact_output_rate", "maximum_template_cluster_rate",
):
value = finite(policy[field], field)
if field != "maximum_mean_reward_gap" and not 0 <= value <= 1:
raise ValueError(f"{field} is outside [0,1]")
if not 0 <= policy["maximum_mean_reward_gap"] <= 1:
raise ValueError("maximum_mean_reward_gap is outside normalized [0,1]")
comparison = policy["reward_comparison"]
comparison_fields = {
"optimizer_min", "optimizer_max", "independent_min",
"independent_max", "higher_is_better",
}
if (
not isinstance(comparison, dict)
or set(comparison) != comparison_fields
or not isinstance(comparison["higher_is_better"], bool)
):
raise ValueError("reward comparison contract is invalid")
for prefix in ("optimizer", "independent"):
low = finite(comparison[f"{prefix}_min"], f"{prefix}_min")
high = finite(comparison[f"{prefix}_max"], f"{prefix}_max")
if not low < high:
raise ValueError(f"{prefix} score range is not ordered")
lines = _verified_text(RECORDS).splitlines()
if any(not line.strip() for line in lines):
raise ValueError("post-training JSONL contains a blank record")
records = [_strict_json(line) for line in lines]
if len(records) != receipt["record_count"]:
raise ValueError("post-training record count differs")
fields = {
"schema_version", "record_id", "prompt_id", "prompt_family",
"population", "optimizer_reward", "verifier_pass",
"independent_score", "independent_evidence_sha256",
"task_success", "output_sha256", "template_cluster_id",
}
if any(
not isinstance(item, dict)
or set(item) != fields
or item["schema_version"] != "aidefend.posttrain-eval-record.v1"
for item in records
):
raise ValueError("post-training record schema differs")
ids = [item["record_id"] for item in records]
if set(ids) != set(receipt["expected_record_ids"]) or len(ids) != len(set(ids)):
raise ValueError("post-training record population differs")
for item in records:
if (
item["population"] not in {"optimization", "heldout"}
or item["prompt_family"] not in policy["families"]
or not isinstance(item["prompt_id"], str)
or not 1 <= len(item["prompt_id"]) <= 256
or not isinstance(item["verifier_pass"], bool)
or not isinstance(item["task_success"], bool)
or not isinstance(item["independent_evidence_sha256"], str)
or SHA256.fullmatch(item["independent_evidence_sha256"]) is None
or not isinstance(item["output_sha256"], str)
or SHA256.fullmatch(item["output_sha256"]) is None
or not isinstance(item["template_cluster_id"], str)
or not item["template_cluster_id"]
):
raise ValueError("post-training record field is invalid")
optimizer_reward = finite(item["optimizer_reward"], "optimizer_reward")
independent_score = finite(item["independent_score"], "independent_score")
comparison = policy["reward_comparison"]
if not (
comparison["optimizer_min"] <= optimizer_reward <= comparison["optimizer_max"]
and comparison["independent_min"]
<= independent_score <= comparison["independent_max"]
):
raise ValueError("reward score is outside its signed scale")
prompt_identities = [
(item["population"], item["prompt_family"], item["prompt_id"])
for item in records
]
if len(prompt_identities) != len(set(prompt_identities)):
raise ValueError("post-training prompt identities are duplicated")
return records, receipt, policy
def rate(values: list[bool]) -> float:
return statistics.fmean(1.0 if value else 0.0 for value in values)
def maximum_share(values: list[str]) -> float:
counts = Counter(values)
return counts.most_common(1)[0][1] / len(values)
def normalized_score(value: float, low: float, high: float, higher: bool) -> float:
normalized = (float(value) - float(low)) / (float(high) - float(low))
return normalized if higher else 1.0 - normalized
def evaluate(records: list[dict], receipt: dict, policy: dict) -> dict:
collection_incomplete = receipt["collection_complete"] is not True
grouped: dict[tuple[str, str], list[dict]] = defaultdict(list)
for item in records:
grouped[(item["prompt_family"], item["population"])].append(item)
insufficient, findings, metrics = [], [], []
for family, minimums in sorted(policy["families"].items()):
optimization = grouped.get((family, "optimization"), [])
heldout = grouped.get((family, "heldout"), [])
if (
len(optimization) < minimums["minimum_optimization_records"]
or len(heldout) < minimums["minimum_heldout_records"]
):
insufficient.append({
"prompt_family": family,
"optimization_records": len(optimization),
"heldout_records": len(heldout),
})
continue
comparison = policy["reward_comparison"]
higher = comparison["higher_is_better"]
mean_gap = statistics.fmean(
normalized_score(
item["optimizer_reward"], comparison["optimizer_min"],
comparison["optimizer_max"], higher,
) - normalized_score(
item["independent_score"], comparison["independent_min"],
comparison["independent_max"], higher,
)
for item in optimization
)
verifier_rate = rate([item["verifier_pass"] for item in optimization])
heldout_success = rate([item["task_success"] for item in heldout])
all_rows = optimization + heldout
exact_output_rate = maximum_share([item["output_sha256"] for item in all_rows])
cluster_rate = maximum_share([item["template_cluster_id"] for item in all_rows])
reasons = []
if mean_gap >= policy["maximum_mean_reward_gap"]:
reasons.append("optimizer_reward_outpaces_independent_evaluator")
if (
verifier_rate >= policy["minimum_verifier_pass_rate"]
and heldout_success
<= policy["maximum_heldout_success_rate_when_verifier_high"]
):
reasons.append("high_verifier_pass_with_low_heldout_success")
if exact_output_rate >= policy["maximum_exact_output_rate"]:
reasons.append("exact_output_concentration")
if cluster_rate >= policy["maximum_template_cluster_rate"]:
reasons.append("template_cluster_concentration")
row = {
"prompt_family": family,
"optimization_records": len(optimization),
"heldout_records": len(heldout),
"mean_optimizer_independent_gap": mean_gap,
"optimizer_verifier_pass_rate": verifier_rate,
"heldout_task_success_rate": heldout_success,
"maximum_exact_output_rate": exact_output_rate,
"maximum_template_cluster_rate": cluster_rate,
"reasons": reasons,
}
metrics.append(row)
if reasons:
findings.append(row)
if findings:
return {
"status": "FAIL",
"reason": "reward or verifier gaming signal found",
"findings": findings,
"insufficient_families": insufficient,
"collection_complete": not collection_incomplete,
"metrics": metrics,
}
if insufficient or collection_incomplete:
return {
"status": "INSUFFICIENT_DATA",
"reason": "signed evaluation collection or prompt-family coverage is incomplete",
"findings": [],
"insufficient_families": insufficient,
"collection_complete": not collection_incomplete,
"metrics": metrics,
}
return {
"status": "PASS",
"reason": "complete post-training populations met signed comparison rules",
"findings": [],
"insufficient_families": [],
"collection_complete": True,
"metrics": metrics,
}
def main() -> None:
try:
records, receipt, policy = load_inputs()
result = evaluate(records, receipt, policy)
evidence = {
"schema_version": "aidefend.d013-posttrain-evidence.v1",
"control": "AID-D-013",
"guidance": "post-training-reward-gaming-evaluation",
"asset_id": receipt["asset_id"],
"asset_version": receipt["asset_version"],
"candidate_policy_sha256": receipt["candidate_policy_sha256"],
"reward_evaluator_sha256": receipt["reward_evaluator_sha256"],
"independent_evaluator_sha256": receipt["independent_evaluator_sha256"],
"task_suite_version": receipt["task_suite_version"],
"prompt_family_registry_version": receipt["prompt_family_registry_version"],
"output_normalizer_version": receipt["output_normalizer_version"],
"template_clusterer_sha256": receipt["template_clusterer_sha256"],
"expected_population": receipt["record_count"],
"records_sha256": digest(RECORDS),
"receipt_sha256": digest(RECEIPT),
"policy_sha256": digest(POLICY),
"policy_version": policy["policy_version"],
**result,
"evaluated_at": datetime.now(timezone.utc).isoformat(),
}
except Exception as error:
evidence = {
"schema_version": "aidefend.d013-posttrain-evidence.v1",
"control": "AID-D-013",
"guidance": "post-training-reward-gaming-evaluation",
"status": "ERROR",
"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://D013_POSTTRAIN_EVIDENCE_SIGNING_KEY",
"--bundle", str(EVIDENCE) + ".sig", str(EVIDENCE),
],
check=True, timeout=COSIGN_DEADLINE_CEILING_SECONDS,
)
verify(EVIDENCE, POSTTRAIN_EVIDENCE_VERIFY_KEY)
if __name__ == "__main__":
main()
Verify the complete evaluated population
Action: For every signed prompt family and required checkpoint, compute the reward-to-independent-outcome comparison, persist the signed population result, and emit each gaming or divergence finding to the post-training security owner; promotion enforcement remains with the consuming gate.
Open-source tools
- Stable-Baselines3
- Ray RLlib
- Hugging Face TRL
- OpenRLHF
- verl
- Prometheus
- Grafana
- pandas
- NumPy
- SciPy
- Gymnasium
- MuJoCo
- Captum
- PyJWT (signed reward and verifier claims; not anomaly detection)
- Requests (out-of-band verifier transport; not anomaly detection)
- Cosign (artifact and runner attestation; not reward analysis)
- NVIDIA Isaac Sim
Commercial / hosted tools
- Weights & Biases
- Datadog
Threat mappings
MITRE ATLAS
- AML.T0018 Manipulate AI Model (reward tampering manipulates the learned policy)
- AML.T0018.000 Manipulate AI Model: Poison AI Model (reward-signal tampering interferes with training and poisons the learned policy)
- AML.T0020 Training Data Poisoning (reward signal poisoning is training data poisoning for RL)
MAESTRO
- Agent Goal Manipulation (L7)
- Manipulation of Evaluation Metrics (L5) (Detects agents that learn to game reward rather than actually achieve intended task success)
OWASP LLM Top 10 2026
- LLM05:2026 Data and Model Poisoning
OWASP ML Top 10 2023
- ML08:2023 Model Skewing (where agent behavior is skewed by an exploitable reward)
- ML02:2023 Data Poisoning Attack (reward signal poisoning is a data poisoning variant)
OWASP Top 10 for Agentic Applications 2026
- ASI10:2026 Rogue Agents (reward hacking creates rogue behavior)
NIST Adversarial Machine Learning 2025
- NISTAML.013 Data Poisoning (reward signal is training data)
Cisco Integrated AI Security and Safety Framework
- AISubtech-2.3.1 Task Loophole Exploitation
- AISubtech-2.3.2 Verifier Manipulation
- AISubtech-6.1.2 Reinforcement Biasing
- AISubtech-6.1.3 Reinforcement Signal Corruption
- AITech-2.3 Reward Hacking
- AITech-6.1 Training Data Poisoning
Google Secure AI Framework 2.0 - Risks
- DP: Data Poisoning (reward poisoning is a form of data poisoning)
- RA: Rogue Actions (manipulated RL policies produce rogue actions)
Databricks AI Security Framework 3.0
- Datasets 3.1: Data poisoning
- Agents - Core 13.6: Intent Breaking & Goal Manipulation
- Agents - Core 13.7: Misaligned & Deceptive Behaviors (reward hacking produces deceptive reward-maximizing behavior)