AID-H-007.007: Evaluator, Judge & Test-Harness Integrity
Pillars: model, infra, app
Lifecycle: building, validation
Scope boundary
Owns integrity and independence of the evaluator mechanism: judge model, grader and scoring code, rubric, judge prompts, dependencies, harness orchestration, threshold policy, and evidence-writer configuration. It does not own test-case sufficiency or adversarial coverage.
AID-H-007.004: AID-H-007.007 protects how captured behavior is scored and converted into a gate decision; AID-H-007.004 protects the evaluation cases, holdouts, corpus sufficiency, leakage boundaries, and result binding. Evaluator integrity and evaluation-data integrity must be verified independently.
AID-M-008: AID-H-007.007 protects the judge and harness mechanism; AID-M-008 owns the adversarial benchmark scenarios and their execution against the agent. A secure evaluator does not establish adequate attack coverage.
Implementation guidance
- AID-H-007.007-G001: Pin and sign one immutable evaluator-mechanism bundle containing judge, grader, rubric-template, dependency, harness, threshold-policy, and evidence-writer configuration.
- AID-H-007.007-G002: Separate evaluated-system identities from evaluator administration and evidence-writing identities.
- AID-H-007.007-G003: Run protected tamper tests and independently recompute the gate from signed frozen candidate-output records.
AID-H-007.007-G001
Verify the complete signed evaluator mechanism bundle
The signed manifest declares an immutable judge URI, protected mechanism roots, and every member digest. Evaluation cases and expected-result datasets are excluded because AID-H-007.004 owns their integrity and sufficiency.
Build the manifest from the exact bundle bytes
The release owner supplies a signature-verified, versioned profile containing the bundle ID and immutable judge URI. The builder verifies that profile from an in-memory snapshot and computes every member digest from the protected tree; no edit-time digest placeholder is accepted.
# File: tools/build_evaluator_bundle_manifest.py
from __future__ import annotations
import argparse, hashlib, json, math, os, re, subprocess, tempfile
from pathlib import Path, PurePosixPath
PROFILE_KEY = Path("/opt/aidefend/trust/evaluator-release-profile.pub")
PROFILE_FIELDS = {"schema_version", "profile_version", "bundle_id", "judge_uri"}
MEMBERS = (
("judge_adapter", "evaluator/judge_adapter.py"),
("grader", "evaluator/grader.py"),
("harness", "evaluator/harness.py"),
("evidence_writer", "evaluator/evidence_writer.py"),
("rubric", "rubric/safety.yaml"),
("threshold_policy", "policy/gate-thresholds.json"),
("dependency_lock", "dependencies/requirements.lock"),
)
JUDGE = re.compile(r"^oci://[^\s]+@sha256:[a-f0-9]{64}$")
BUNDLE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
def required_timeout() -> float:
value = float(os.environ["VERIFIED_H007_COSIGN_TIMEOUT_SECONDS"])
if not math.isfinite(value) or value <= 0:
raise ValueError("cosign timeout must be finite and positive")
return value
def strict_object(payload: bytes, label: str) -> dict:
def unique(pairs):
result = {}
for key, value in pairs:
if key in result: raise ValueError(f"duplicate {label} key: {key}")
result[key] = value
return result
def reject(value): raise ValueError(f"non-finite {label} number: {value}")
value = json.loads(payload.decode("utf-8", errors="strict"), object_pairs_hook=unique, parse_constant=reject)
if not isinstance(value, dict): raise ValueError(label + " must be an object")
return value
def write_private(path: Path, payload: bytes) -> None:
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o400)
with os.fdopen(descriptor, "wb") as handle: handle.write(payload); handle.flush(); os.fsync(handle.fileno())
def verify_profile(path: Path, signature: Path) -> tuple[dict, str]:
payload=path.read_bytes(); signature_bytes=signature.read_bytes(); key_bytes=PROFILE_KEY.read_bytes()
with tempfile.TemporaryDirectory(prefix="aidefend-evaluator-profile-") as directory:
root=Path(directory); root.chmod(0o700)
payload_snapshot=root/"profile.json"; signature_snapshot=root/"profile.sig"; key_snapshot=root/"profile.pub"
write_private(payload_snapshot,payload); write_private(signature_snapshot,signature_bytes); write_private(key_snapshot,key_bytes)
subprocess.run(["cosign","verify-blob","--key",str(key_snapshot),"--bundle",str(signature_snapshot),str(payload_snapshot)],check=True,capture_output=True,text=True,timeout=required_timeout())
profile = strict_object(payload, "evaluator release profile")
if not isinstance(profile, dict) or set(profile) != PROFILE_FIELDS:
raise ValueError("evaluator release profile schema differs")
if profile["schema_version"] != "aidefend.evaluator-release-profile.v1":
raise ValueError("unsupported evaluator release profile")
if not isinstance(profile["profile_version"], str) or not profile["profile_version"]:
raise ValueError("profile version is required")
if not isinstance(profile["bundle_id"], str) or not BUNDLE_ID.fullmatch(profile["bundle_id"]):
raise ValueError("bundle ID is invalid")
if not isinstance(profile["judge_uri"], str) or not JUDGE.fullmatch(profile["judge_uri"]):
raise ValueError("judge URI must be an immutable OCI digest reference")
return profile, hashlib.sha256(payload).hexdigest()
def checked_file(root: Path, relative: str) -> Path:
rel = PurePosixPath(relative)
cursor = root
for part in rel.parts:
cursor = cursor / part
if cursor.is_symlink():
raise ValueError(f"linked bundle path denied: {relative}")
path = root.joinpath(*rel.parts)
if not path.is_file():
raise ValueError(f"missing or non-file bundle member: {relative}")
resolved = path.resolve(strict=True)
if root not in resolved.parents:
raise ValueError(f"bundle member escaped root: {relative}")
return path
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--root", type=Path, required=True)
parser.add_argument("--profile", type=Path, required=True)
parser.add_argument("--profile-signature", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
root = args.root.resolve(strict=True)
if not root.is_dir(): raise ValueError("bundle root is not a directory")
profile, profile_sha256 = verify_profile(args.profile, args.profile_signature)
members = [
{"kind": kind, "path": relative, "sha256": sha256_file(checked_file(root, relative))}
for kind, relative in MEMBERS
]
manifest = {
"schema_version": "aidefend.evaluator-bundle.v1",
"bundle_id": profile["bundle_id"],
"judge_uri": profile["judge_uri"],
"protected_roots": ["evaluator", "rubric", "policy", "dependencies"],
"members": members,
}
output = args.output
output.parent.mkdir(parents=True, exist_ok=True)
payload = (json.dumps(manifest, sort_keys=True, separators=(",", ":")) + "\n").encode()
descriptor = os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(descriptor, "wb") as handle:
handle.write(payload); handle.flush(); os.fsync(handle.fileno())
print(json.dumps({"manifest_sha256": hashlib.sha256(payload).hexdigest(), "release_profile_sha256": profile_sha256, "member_count": len(members)}, sort_keys=True))
if __name__ == "__main__": main()Materialize and sign the manifest
set -euo pipefail
umask 077
: "${EVALUATOR_BUNDLE_SIGNING_KEY:?isolated signing key URI or path required}"
python tools/build_evaluator_bundle_manifest.py \
--root . \
--profile policy/evaluator-release-profile.json \
--profile-signature policy/evaluator-release-profile.json.sig \
--output evidence/evaluator-bundle.json
cosign sign-blob --yes --key "$EVALUATOR_BUNDLE_SIGNING_KEY" \
--bundle evidence/evaluator-bundle.sig \
evidence/evaluator-bundle.jsonApply and capture evidence
# File: tools/verify_evaluator_bundle.py
from __future__ import annotations
import argparse, hashlib, json, re
from pathlib import Path, PurePosixPath
SHA256=re.compile(r"^[a-f0-9]{64}$"); JUDGE=re.compile(r"^oci://[^\s]+@sha256:[a-f0-9]{64}$")
REQUIRED_KINDS={"judge_adapter":"evaluator","grader":"evaluator","harness":"evaluator","evidence_writer":"evaluator","rubric":"rubric","threshold_policy":"policy","dependency_lock":"dependencies"}
def strict_object(payload: bytes,label: str) -> dict:
def unique(pairs):
result={}
for key,value in pairs:
if key in result: raise ValueError(f"duplicate {label} key: {key}")
result[key]=value
return result
def reject(value): raise ValueError(f"non-finite {label} number: {value}")
value=json.loads(payload.decode("utf-8",errors="strict"),object_pairs_hook=unique,parse_constant=reject)
if not isinstance(value,dict): raise ValueError(label+" must be an object")
return value
def safe_rel(value: str) -> PurePosixPath:
path=PurePosixPath(value)
if path.is_absolute() or not path.parts or any(part in {"",".",".."} for part in path.parts): raise ValueError(f"unsafe bundle path: {value!r}")
return path
def link_like(path: Path) -> bool:
return path.is_symlink() or bool(getattr(path,"is_junction",lambda:False)())
def reject_root_links(path: Path,label: str) -> None:
cursor=path.absolute(); anchor=Path(cursor.anchor)
while True:
if link_like(cursor): raise ValueError(f"{label}: linked/junction path denied: {cursor}")
if cursor==anchor: return
cursor=cursor.parent
def reject_tree_links(path: Path,boundary: Path,label: str) -> None:
cursor=path.absolute(); stop=boundary.absolute()
while True:
if link_like(cursor): raise ValueError(f"{label}: linked/junction path denied: {cursor}")
if cursor==stop: return
if stop not in cursor.parents: raise ValueError(f"{label}: path escaped protected root: {cursor}")
cursor=cursor.parent
def main() -> None:
parser=argparse.ArgumentParser(); parser.add_argument("--root",type=Path,required=True); parser.add_argument("--manifest",type=Path,required=True); args=parser.parse_args()
root_input=args.root.absolute(); reject_root_links(root_input,"bundle root"); root=root_input.resolve(strict=True); manifest_raw=args.manifest.read_bytes(); manifest=strict_object(manifest_raw,"evaluator bundle")
if set(manifest)!={"schema_version","bundle_id","judge_uri","protected_roots","members"} or manifest["schema_version"]!="aidefend.evaluator-bundle.v1": raise ValueError("evaluator manifest schema differs")
if not isinstance(manifest["bundle_id"],str) or not manifest["bundle_id"] or not JUDGE.fullmatch(manifest["judge_uri"]): raise ValueError("bundle ID or immutable judge reference invalid")
roots=manifest["protected_roots"]
if not isinstance(roots,list) or not roots or len(roots)!=len(set(roots)) or set(roots)!=set(REQUIRED_KINDS.values()): raise ValueError("protected mechanism roots differ")
members=manifest["members"]
if not isinstance(members,list) or not members: raise ValueError("mechanism member set is empty")
expected={}; kinds=set()
for row in members:
if not isinstance(row,dict) or set(row)!={"kind","path","sha256"} or row.get("kind") not in REQUIRED_KINDS or not isinstance(row.get("sha256"),str) or not SHA256.fullmatch(row["sha256"]): raise ValueError("invalid member row")
if row["kind"] in kinds: raise ValueError(f"duplicate mechanism kind: {row['kind']}")
rel=safe_rel(row["path"])
if rel.parts[0]!=REQUIRED_KINDS[row["kind"]] or rel.as_posix() in expected: raise ValueError(f"misplaced or duplicate member: {rel}")
kinds.add(row["kind"]); expected[rel.as_posix()]=row["sha256"]
if kinds!=set(REQUIRED_KINDS): raise ValueError({"missing_mechanisms":sorted(set(REQUIRED_KINDS)-kinds),"extra_mechanisms":sorted(kinds-set(REQUIRED_KINDS))})
actual=set()
for declared in roots:
rel=safe_rel(declared); base=root.joinpath(*rel.parts)
if not base.exists() or not base.is_dir(): raise ValueError(f"missing or non-directory protected root: {rel}")
reject_tree_links(base,root,f"protected root {rel}")
root_members=0
for path in base.rglob("*"):
reject_tree_links(path,root,f"mechanism member {path}")
resolved=path.resolve(strict=True)
if root not in resolved.parents: raise ValueError(f"out-of-root member: {path}")
if path.is_dir(): continue
if not path.is_file(): raise ValueError(f"unsupported mechanism member type: {path}")
root_members+=1; actual.add(path.relative_to(root).as_posix())
if root_members==0: raise ValueError(f"empty protected root: {rel}")
if actual!=set(expected): raise ValueError({"missing":sorted(set(expected)-actual),"extra":sorted(actual-set(expected))})
for rel,want in expected.items():
got=hashlib.sha256((root/rel).read_bytes()).hexdigest()
if got!=want: raise ValueError(f"digest drift: {rel}")
print(json.dumps({"bundle_id":manifest["bundle_id"],"manifest_sha256":hashlib.sha256(manifest_raw).hexdigest(),"member_count":len(expected)},sort_keys=True))
if __name__=="__main__": main()Independent verification
Verify the manifest and immutable judge signature, enumerate all protected roots, and run tamper, missing/extra member, symlink/traversal, swapped-judge, and stale-signature negatives. The trap below restores the grader even when the negative test fails.
set -euo pipefail
umask 077
SNAPSHOT_DIR=$(mktemp -d)
BACKUP=$(mktemp)
cleanup() {
if [[ -s "$BACKUP" ]]; then cp -- "$BACKUP" evaluator/grader.py; fi
rm -f -- "$BACKUP"
rm -rf -- "$SNAPSHOT_DIR"
}
trap cleanup EXIT
chmod 0700 "$SNAPSHOT_DIR"
install -m 0400 -- evidence/evaluator-bundle.json "$SNAPSHOT_DIR/evaluator-bundle.json"
install -m 0400 -- evidence/evaluator-bundle.sig "$SNAPSHOT_DIR/evaluator-bundle.sig"
install -m 0400 -- keys/evaluator-bundle.pub "$SNAPSHOT_DIR/evaluator-bundle.pub"
install -m 0400 -- keys/judge.pub "$SNAPSHOT_DIR/judge.pub"
cosign verify-blob --key "$SNAPSHOT_DIR/evaluator-bundle.pub" \
--bundle "$SNAPSHOT_DIR/evaluator-bundle.sig" "$SNAPSHOT_DIR/evaluator-bundle.json"
python tools/verify_evaluator_bundle.py --root . --manifest "$SNAPSHOT_DIR/evaluator-bundle.json"
JUDGE_REF=$(jq -er '.judge_uri' "$SNAPSHOT_DIR/evaluator-bundle.json")
cosign verify --key "$SNAPSHOT_DIR/judge.pub" "$JUDGE_REF"
cp -- evaluator/grader.py "$BACKUP"
printf "\n# integrity-negative-test\n" >> evaluator/grader.py
if python tools/verify_evaluator_bundle.py --root . --manifest "$SNAPSHOT_DIR/evaluator-bundle.json"; then exit 1; fiFailure handling
A mutable or unsigned judge, missing or extra mechanism member, symlink or traversal, digest drift, stale signature, unknown schema, or verifier error blocks the release gate. Evaluation-case count is decided by AID-H-007.004, not this control.
AID-H-007.007-G002
Concept
Give the evaluator identity access only to its named result object. Do not grant create together with resourceNames: Kubernetes cannot constrain a create request by an existing resource name. Test authorization with server-side SubjectAccessReview objects so an API error, malformed response, or missing boolean decision is an ERROR, not a successful negative test.
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
name: evaluator-writer
namespace: ai-evaluation
rules:
- apiGroups: [""]
resources: ["configmaps"]
resourceNames: ["evaluation-results"]
verbs: ["get", "update"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
name: evaluator-writer
namespace: ai-evaluation
subjects:
- kind: ServiceAccount
name: evaluator-runner
namespace: ai-evaluation
roleRef:
apiGroup: rbac.authorization.k8s.io
kind: Role
name: evaluator-writerFail-closed authorization probes
#!/usr/bin/env bash
set -euo pipefail
kubectl apply -f deploy/evaluator-rbac.yaml
mkdir -p evidence/rbac
subject_access_review() {
expected="$1"
evidence_name="$2"
username="$3"
groups_json="$4"
verb="$5"
resource="$6"
resource_name="$7"
request="$(jq -nc \
--arg username "$username" \
--argjson groups "$groups_json" \
--arg verb "$verb" \
--arg resource "$resource" \
--arg name "$resource_name" \
'{
apiVersion:"authorization.k8s.io/v1",
kind:"SubjectAccessReview",
spec:{
user:$username,
groups:$groups,
resourceAttributes:{
group:"",
namespace:"ai-evaluation",
verb:$verb,
resource:$resource,
name:$name
}
}
}')"
response="$(printf '%s' "$request" | kubectl create \
--raw /apis/authorization.k8s.io/v1/subjectaccessreviews -f -)"
printf '%s\n' "$response" | jq -e '
.apiVersion == "authorization.k8s.io/v1" and
.kind == "SubjectAccessReview" and
(.status.allowed | type == "boolean")
' >/dev/null
actual="$(printf '%s\n' "$response" | jq -er '.status.allowed | tostring')"
printf '%s\n' "$response" > "evidence/rbac/$evidence_name.json"
if [[ "$actual" != "$expected" ]]; then
printf 'authorization mismatch: %s expected=%s actual=%s\n' \
"$evidence_name" "$expected" "$actual" >&2
exit 1
fi
}
runner="system:serviceaccount:ai-evaluation:evaluator-runner"
candidate="system:serviceaccount:ai-evaluation:candidate"
service_account_groups='["system:serviceaccounts","system:serviceaccounts:ai-evaluation","system:authenticated"]'
subject_access_review true runner-update "$runner" "$service_account_groups" update configmaps evaluation-results
subject_access_review false candidate-update "$candidate" "$service_account_groups" update configmaps evaluation-results
subject_access_review false candidate-rubric "$candidate" "$service_account_groups" get secrets evaluator-rubricVerification: Run the probes from a dedicated auditor identity allowed to create SubjectAccessReview resources but not to mutate the tested Role. The SAR must reproduce the authenticated service-account groups (system:serviceaccounts, the namespace-specific service-account group, and system:authenticated); otherwise group RoleBindings are invisible and a deny result is incomplete. Preserve the three response objects with the cluster identity, Role/RoleBinding resource versions, tested groups, and timestamp. A transport failure, RBAC API error, invalid JSON, absent status.allowed, or unexpected allow/deny decision must fail the release gate.
AID-H-007.007-G003
Recompute deterministic aggregation without demanding stochastic byte identity
AID-H-007.004 supplies the signed complete case manifest and frozen candidate outputs. Primary and independent evaluator identities score the same output digests. Deterministic aggregation must match exactly; stochastic judge scores use the signed policy's replicate count and tolerance, while final decisions must still agree.
Signed recomputation-policy input
The evaluator-policy authority supplies policy/evaluator-recompute.json with a concrete positive replicate count, a concrete finite nonnegative tolerance, and the exact bundle, case-manifest, and frozen-output-manifest digests. The workflow below pins the signature-verified bytes, validates the exact schema and typed values, and recomputes every signed result population; this file is an input artifact, not an edit-time template.
Apply and capture evidence
set -euo pipefail
rm -f evidence/evaluator-recompute.pass evidence/evaluator-recompute.pass.tmp evidence/evaluator-recompute.pass.sig
snapshot_dir="$(mktemp -d)"; export AIDEFEND_EVALUATOR_SNAPSHOT="$snapshot_dir"; trap 'rm -rf -- "$snapshot_dir"' EXIT
pin_blob() {
local key="$1" signature="$2" source="$3" destination="$4" output="$5" before after pinned
before="$(sha256sum -- "$source" | awk '{print $1}')" || return 1
cosign verify-blob --key "$key" --bundle "$signature" "$source" >/dev/null || return 1
cp -- "$source" "$destination" || return 1
after="$(sha256sum -- "$source" | awk '{print $1}')" || return 1
pinned="$(sha256sum -- "$destination" | awk '{print $1}')" || return 1
test "$before" = "$after" || return 1; test "$before" = "$pinned" || return 1
printf -v "$output" '%s' "$before"; export "$output"
}
pin_blob keys/evaluator-policy.pub policy/evaluator-recompute.sig policy/evaluator-recompute.json "$snapshot_dir/policy.json" AIDEFEND_RECOMPUTE_POLICY_SHA256
pin_blob keys/evaluator-bundle.pub evidence/evaluator-bundle.sig evidence/evaluator-bundle.json "$snapshot_dir/bundle.json" AIDEFEND_BUNDLE_SHA256
pin_blob keys/evaluation-cases.pub evidence/signed-case-manifest.sig evidence/signed-case-manifest.json "$snapshot_dir/cases.json" AIDEFEND_CASES_SHA256
pin_blob keys/frozen-outputs.pub evidence/frozen-output-manifest.sig evidence/frozen-output-manifest.json "$snapshot_dir/frozen.json" AIDEFEND_FROZEN_MANIFEST_SHA256
pin_blob keys/primary-evaluator.pub evidence/primary-results.sig evidence/primary-results.jsonl "$snapshot_dir/primary.jsonl" AIDEFEND_PRIMARY_SHA256
pin_blob keys/replay-evaluator.pub evidence/replay-results.sig evidence/replay-results.jsonl "$snapshot_dir/replay.jsonl" AIDEFEND_REPLAY_SHA256
python - <<'PY'
import hashlib, json, math, os, re
from pathlib import Path, PurePosixPath
SHA=re.compile(r"^[a-f0-9]{64}$"); DECISIONS={"PASS","FAIL","ERROR"}
POLICY_FIELDS={"schema_version","judge_mode","minimum_replicates","score_tolerance","allowed_decisions","bundle_sha256","case_manifest_sha256","frozen_output_manifest_sha256"}
RESULT_FIELDS={"case_id","output_sha256","decision","scores","bundle_sha256","case_manifest_sha256","frozen_output_manifest_sha256"}
snapshot=Path(os.environ["AIDEFEND_EVALUATOR_SNAPSHOT"])
def pinned(name,environment,json_value):
path=snapshot/name; raw=path.read_bytes(); expected=os.environ[environment]
if not SHA.fullmatch(expected) or hashlib.sha256(raw).hexdigest()!=expected or hashlib.sha256(path.read_bytes()).hexdigest()!=expected: raise ValueError(f"{name}: verified snapshot digest differs")
return json.loads(raw) if json_value else raw.decode("utf-8")
def safe_rel(value,label):
if not isinstance(value,str) or not value: raise ValueError(f"{label}: path missing")
rel=PurePosixPath(value)
if rel.is_absolute() or not rel.parts or any(part in {"",".",".."} for part in rel.parts): raise ValueError(f"{label}: unsafe path")
return rel
def link_like(path):
return path.is_symlink() or bool(getattr(path,"is_junction",lambda:False)())
def reject_root_links(path,label):
cursor=path.absolute(); anchor=Path(cursor.anchor)
while True:
if link_like(cursor): raise ValueError(f"{label}: linked/junction path denied: {cursor}")
if cursor==anchor: return
cursor=cursor.parent
def reject_tree_links(path,boundary,label):
cursor=path.absolute(); stop=boundary.absolute()
while True:
if link_like(cursor): raise ValueError(f"{label}: linked/junction path denied: {cursor}")
if cursor==stop: return
if stop not in cursor.parents: raise ValueError(f"{label}: path escaped frozen root: {cursor}")
cursor=cursor.parent
def load_results(text,label):
rows={}
for number,line in enumerate(text.splitlines(),1):
if not line.strip(): continue
row=json.loads(line)
if not isinstance(row,dict) or set(row)!=RESULT_FIELDS: raise ValueError(f"{label}:{number}: result schema differs")
key=row.get("case_id")
if not isinstance(key,str) or not key or key in rows: raise ValueError(f"{label}:{number}: missing/duplicate case_id")
for field in ("output_sha256","bundle_sha256","case_manifest_sha256","frozen_output_manifest_sha256"):
if not isinstance(row[field],str) or not SHA.fullmatch(row[field]): raise ValueError(f"{label}:{number}: invalid {field}")
if row["decision"] not in DECISIONS: raise ValueError(f"{label}:{number}: invalid decision")
scores=row["scores"]
if not isinstance(scores,list) or not scores or not all(not isinstance(value,bool) and isinstance(value,(int,float)) and math.isfinite(value) for value in scores): raise ValueError(f"{label}:{number}: invalid score population")
rows[key]=row
return rows
policy=pinned("policy.json","AIDEFEND_RECOMPUTE_POLICY_SHA256",True); bundle=pinned("bundle.json","AIDEFEND_BUNDLE_SHA256",True); case_manifest=pinned("cases.json","AIDEFEND_CASES_SHA256",True); frozen_manifest=pinned("frozen.json","AIDEFEND_FROZEN_MANIFEST_SHA256",True)
primary=load_results(pinned("primary.jsonl","AIDEFEND_PRIMARY_SHA256",False),"primary"); replay=load_results(pinned("replay.jsonl","AIDEFEND_REPLAY_SHA256",False),"replay")
if not isinstance(policy,dict) or set(policy)!=POLICY_FIELDS or policy.get("schema_version")!="aidefend.evaluator-recompute.v1": raise ValueError("recompute policy schema differs")
if not isinstance(bundle,dict) or bundle.get("schema_version")!="aidefend.evaluator-bundle.v1": raise ValueError("bundle schema differs")
for field,actual in (("bundle_sha256",os.environ["AIDEFEND_BUNDLE_SHA256"]),("case_manifest_sha256",os.environ["AIDEFEND_CASES_SHA256"]),("frozen_output_manifest_sha256",os.environ["AIDEFEND_FROZEN_MANIFEST_SHA256"])):
if not isinstance(policy[field],str) or not SHA.fullmatch(policy[field]) or policy[field]!=actual: raise ValueError(f"policy {field} binding differs")
if policy["allowed_decisions"]!=["PASS","FAIL","ERROR"]: raise ValueError("allowed decision enum differs")
mode=policy["judge_mode"]; minimum=policy["minimum_replicates"]; tolerance=policy["score_tolerance"]
if mode not in {"deterministic","stochastic"} or isinstance(minimum,bool) or not isinstance(minimum,int) or minimum<1 or isinstance(tolerance,bool) or not isinstance(tolerance,(int,float)) or not math.isfinite(tolerance) or tolerance<0: raise ValueError("invalid signed recompute policy")
if not isinstance(case_manifest,dict) or set(case_manifest)!={"schema_version","case_ids"} or case_manifest.get("schema_version")!="aidefend.evaluation-case-manifest.v1": raise ValueError("case manifest schema differs")
case_ids=case_manifest["case_ids"]
if not isinstance(case_ids,list) or not case_ids or len(case_ids)!=len(set(case_ids)) or not all(isinstance(key,str) and key for key in case_ids): raise ValueError("case population invalid")
cases=set(case_ids)
if not isinstance(frozen_manifest,dict) or set(frozen_manifest)!={"schema_version","outputs"} or frozen_manifest.get("schema_version")!="aidefend.frozen-output-manifest.v1": raise ValueError("frozen-output manifest schema differs")
outputs={}; expected_paths=set(); frozen_input=Path("evidence/frozen-outputs").absolute(); reject_root_links(frozen_input,"frozen-output root"); frozen_root=frozen_input.resolve(strict=True)
if not frozen_root.is_dir(): raise ValueError("frozen-output root invalid")
for number,row in enumerate(frozen_manifest["outputs"],1):
if not isinstance(row,dict) or set(row)!={"case_id","path","sha256"} or not isinstance(row.get("case_id"),str) or not row["case_id"] or row["case_id"] in outputs or not isinstance(row.get("sha256"),str) or not SHA.fullmatch(row["sha256"]): raise ValueError(f"frozen:{number}: schema, case, or digest differs")
rel=safe_rel(row["path"],f"frozen:{number}"); path=frozen_input.joinpath(*rel.parts); reject_tree_links(path,frozen_input,f"frozen:{number}")
resolved=path.resolve(strict=True)
if frozen_root not in resolved.parents or not resolved.is_file(): raise ValueError(f"frozen:{number}: output outside frozen root")
raw=resolved.read_bytes()
if hashlib.sha256(raw).hexdigest()!=row["sha256"] or hashlib.sha256(resolved.read_bytes()).hexdigest()!=row["sha256"]: raise ValueError(f"frozen:{number}: output bytes changed")
expected_paths.add(resolved); outputs[row["case_id"]]=row["sha256"]
actual_paths=set()
for path in frozen_input.rglob("*"):
reject_tree_links(path,frozen_input,f"frozen output tree member {path}")
if path.is_file(): actual_paths.add(path.resolve(strict=True))
if actual_paths!=expected_paths or set(outputs)!=cases or set(primary)!=cases or set(replay)!=cases: raise ValueError("frozen/output/result case population differs")
bindings={"bundle_sha256":policy["bundle_sha256"],"case_manifest_sha256":policy["case_manifest_sha256"],"frozen_output_manifest_sha256":policy["frozen_output_manifest_sha256"]}
for key in sorted(cases):
left,right=primary[key],replay[key]
if any(left[field]!=value or right[field]!=value for field,value in bindings.items()): raise ValueError(f"{key}: mechanism/input binding differs")
if left["output_sha256"]!=outputs[key] or right["output_sha256"]!=outputs[key] or left["decision"]!=right["decision"]: raise ValueError(f"{key}: frozen output or decision differs")
ls,rs=left["scores"],right["scores"]
if len(ls)<minimum or len(rs)<minimum: raise ValueError(f"{key}: insufficient replicate population")
if mode=="deterministic" and ls!=rs: raise ValueError(f"{key}: deterministic scores differ")
if mode=="stochastic" and abs(sum(ls)/len(ls)-sum(rs)/len(rs))>tolerance: raise ValueError(f"{key}: stochastic aggregate outside tolerance")
out=Path("evidence/evaluator-recompute.pass"); tmp=Path("evidence/evaluator-recompute.pass.tmp"); tmp.write_text("PASS\n",encoding="utf-8"); tmp.replace(out)
PY
cosign sign-blob --yes --key env://EVALUATOR_VERIFIER_KEY --bundle evidence/evaluator-recompute.pass.sig evidence/evaluator-recompute.passIndependent verification
Run the exact shell workflow so every policy, bundle, case, frozen-output, primary-result, and replay-result signature is verified and copied to a digest-pinned snapshot before Python reads it. Negative fixtures alter frozen bytes while keeping both results equal, use an arbitrary decision or non-SHA digest, inject bool/NaN/Infinity scores, omit or add frozen output files, and fail each signature in turn; all must prevent PASS and signing.
cosign verify-blob --key keys/evaluator-verifier.pub --bundle evidence/evaluator-recompute.pass.sig evidence/evaluator-recompute.pass
test "$(cat evidence/evaluator-recompute.pass)" = PASSFailure handling
A mechanism-signature failure, case-population mismatch, output-digest change, invalid score population, deterministic mismatch, stochastic aggregate outside signed tolerance, decision divergence, candidate privilege, or verifier error blocks promotion and preserves raw evidence.
Open-source tools
- MLflow
- in-toto
- slsa-verifier
- Cosign
- Inspect AI
- Git
- Kubernetes RBAC
- pytest
- Open Policy Agent (OPA)
Commercial / hosted tools
- Weights & Biases
- Amazon SageMaker AI Model Registry
- Databricks Models in Unity Catalog
- GitHub Enterprise Cloud
Threat mappings
MITRE ATLAS
- AML.T0010 AI Supply Chain Compromise
- AML.T0010.001 AI Supply Chain Compromise: AI Software
MAESTRO
- Manipulation of Evaluation Metrics (L5) (protecting evaluator judge, grader, rubric-template, dependency, harness, threshold-policy, and evidence-writer integrity)
OWASP LLM Top 10 2026
- LLM04:2026 Supply Chain
OWASP ML Top 10 2023
- ML06:2023 AI Supply Chain Attacks (protecting evaluator judge, grader, rubric-template, dependency, harness, threshold-policy, and evidence-writer integrity)
OWASP Top 10 for Agentic Applications 2026
- ASI04:2026 Agentic Supply Chain Vulnerabilities (protecting evaluator judge, grader, rubric-template, dependency, harness, threshold-policy, and evidence-writer integrity)
NIST Adversarial Machine Learning 2025
- N/A
Cisco Integrated AI Security and Safety Framework
- AISubtech-2.3.2 Verifier Manipulation
- AISubtech-2.3.3 Shortcut Acquisition
- AITech-2.3 Reward Hacking
Google Secure AI Framework 2.0 - Risks
- N/A
Databricks AI Security Framework 3.0
- Algorithms 5.1: Lack of tracking and reproducibility of experiments (protecting evaluator judge, grader, rubric-template, dependency, harness, threshold-policy, and evidence-writer integrity)
- Model 7.4: Source code control attack (protecting evaluator judge, grader, rubric-template, dependency, harness, threshold-policy, and evidence-writer integrity)