Scope boundary

Owns federation participant enrollment, attestation, revocation, role, model and round binding, update-digest proof, replay prevention, and admission of the exact update to the exact round. It does not prove the authorized client is benign.

AID-H-008.001: AID-H-008.003 decides who may submit which update; AID-H-008.001 protects accepted contributions from individual inspection. Round admission and contribution confidentiality are independent requirements.

AID-H-008.002: AID-H-008.003 rejects unknown, replayed, wrong-round, or unattested submitters; AID-H-008.002 mitigates malicious or outlier influence among updates that passed admission. Identity assurance does not establish statistical safety.

AID-H-004.003: AID-H-008.003 applies federated-learning-specific participant, model, round, and update binding; AID-H-004.003 supplies general peer identity, message integrity, replay, and task-protocol controls. Generic secure communication does not decide federated round membership.

Implementation guidance

• AID-H-008.003-G001: Enroll distinct participant workload identities with signed, revocable federation, model, role, public-key, and authoritative SPIRE-entry bindings.
• AID-H-008.003-G002: At the coordinator, independently verify a fresh SVID and a single-use proof bound to participant, federation, model digest, round, and update digest, then atomically consume its nonce before accepting the update.
• AID-H-008.003-G003: Reconcile the complete received-update ledger so every update is accepted or denied and every aggregator input has one valid admission receipt.

AID-H-008.003-G001

Create a revocable participant enrollment from policy values

Issue one SPIFFE identity per participant. Bind federation, immutable model digest, role, public JWK, organization approval, expiry, SPIRE entry readback digest, and selector digest. TTL and expiry come only from a verified versioned policy; never share a federation credential.

Concrete signed enrollment inputs

The federation authority supplies the final signed enrollment policy with concrete identity, selector, model/JWK, lifetime, and positive SPIRE-command timeout values; the organization authority supplies the separately signed approval. These are input artifacts, not edit-time templates. The enrollment job verifies both signatures, validates exact schemas and values, bounds both SPIRE CLI calls, and reads back the effective entry before signing the enrollment record.

Apply and capture evidence

set -euo pipefail
rm -f evidence/participant-enrollment.json evidence/participant-enrollment.tmp evidence/participant-enrollment.sig evidence/spire-entry-readback.json evidence/spire-entry-readback.raw.json
SNAPSHOT_DIR=$(mktemp -d)
trap 'rm -rf -- "$SNAPSHOT_DIR"' EXIT
install -m 0600 -- evidence/enrollment-policy.json "$SNAPSHOT_DIR/enrollment-policy.json"
install -m 0600 -- evidence/organization-approval.json "$SNAPSHOT_DIR/organization-approval.json"
export FL_ENROLLMENT_POLICY_SNAPSHOT="$SNAPSHOT_DIR/enrollment-policy.json"
export FL_ORGANIZATION_APPROVAL_SNAPSHOT="$SNAPSHOT_DIR/organization-approval.json"
cosign verify-blob --key keys/fl-policy.pub --bundle evidence/enrollment-policy.sig "$FL_ENROLLMENT_POLICY_SNAPSHOT"
cosign verify-blob --key keys/fl-organization.pub --bundle evidence/organization-approval.sig "$FL_ORGANIZATION_APPROVAL_SNAPSHOT"
python - <<'PY'
import hashlib, json, math, os, re, subprocess, time
from datetime import datetime, timezone
from pathlib import Path

SHA=re.compile(r"^[a-f0-9]{64}$"); IDENTIFIER=re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$"); SELECTOR=re.compile(r"^[A-Za-z0-9._-]+:[^\s:][^\s]*$"); B64URL=re.compile(r"^[A-Za-z0-9_-]{16,}$")
POLICY_FIELDS={"schema_version","enrollment_id","participant_id","spiffe_id","parent_id","selectors","federation_id","model_sha256","role","svid_ttl_seconds","max_enrollment_lifetime_seconds","command_timeout_seconds","public_jwk"}
APPROVAL_FIELDS={"schema_version","approval_id","organization_id","participant_id","federation_id","approved_roles","not_before","expires_at"}
def text(row,field,label,pattern=IDENTIFIER):
    value=row.get(field)
    if not isinstance(value,str) or not value or (pattern is not None and not pattern.fullmatch(value)): raise ValueError(f"{label}: invalid {field}")
    return value
def finite_int(value,label):
    if isinstance(value,bool) or not isinstance(value,(int,float)) or not math.isfinite(value) or value<=0 or int(value)!=value: raise ValueError(f"{label}: must be a positive finite integer")
    return int(value)
def instant(value,label):
    if not isinstance(value,str) or not value: raise ValueError(f"{label}: timestamp missing")
    parsed=datetime.fromisoformat(value.replace("Z","+00:00"))
    if parsed.tzinfo is None: raise ValueError(f"{label}: timestamp lacks timezone")
    epoch=parsed.timestamp()
    if not math.isfinite(epoch): raise ValueError(f"{label}: timestamp not finite")
    return parsed.astimezone(timezone.utc)
def canonical(value): return json.dumps(value,sort_keys=True,separators=(",",":"),ensure_ascii=False).encode("utf-8")
policy=json.loads(Path(os.environ["FL_ENROLLMENT_POLICY_SNAPSHOT"]).read_text(encoding="utf-8")); approval=json.loads(Path(os.environ["FL_ORGANIZATION_APPROVAL_SNAPSHOT"]).read_text(encoding="utf-8"))
if not isinstance(policy,dict) or set(policy)!=POLICY_FIELDS or policy.get("schema_version")!="aidefend.fl-enrollment-policy.v1": raise ValueError("enrollment policy schema differs")
if not isinstance(approval,dict) or set(approval)!=APPROVAL_FIELDS or approval.get("schema_version")!="aidefend.fl-organization-approval.v1": raise ValueError("organization approval schema differs")
enrollment_id=text(policy,"enrollment_id","policy"); participant=text(policy,"participant_id","policy"); spiffe=text(policy,"spiffe_id","policy",None); parent=text(policy,"parent_id","policy",None); federation=text(policy,"federation_id","policy"); role=text(policy,"role","policy")
if not spiffe.startswith("spiffe://") or not parent.startswith("spiffe://"): raise ValueError("SPIFFE IDs must be absolute spiffe URIs")
model=policy.get("model_sha256")
if not isinstance(model,str) or not SHA.fullmatch(model): raise ValueError("immutable model digest invalid")
selectors=policy.get("selectors")
if not isinstance(selectors,list) or not selectors or selectors!=sorted(set(selectors)) or not all(isinstance(value,str) and SELECTOR.fullmatch(value) for value in selectors): raise ValueError("selector set invalid")
jwk=policy.get("public_jwk")
if not isinstance(jwk,dict) or set(jwk)!={"kty","crv","x","y"} or jwk.get("kty")!="EC" or jwk.get("crv")!="P-256" or not B64URL.fullmatch(jwk.get("x", "")) or not B64URL.fullmatch(jwk.get("y", "")) or any("REPLACE" in value for value in jwk.values() if isinstance(value,str)): raise ValueError("public JWK invalid or placeholder")
if text(approval,"participant_id","approval")!=participant or text(approval,"federation_id","approval")!=federation: raise PermissionError("organization approval binding differs")
approval_id=text(approval,"approval_id","approval"); organization=text(approval,"organization_id","approval"); roles=approval.get("approved_roles")
if not isinstance(roles,list) or not roles or roles!=sorted(set(roles)) or not all(isinstance(value,str) and IDENTIFIER.fullmatch(value) for value in roles) or role not in roles: raise PermissionError("role lacks organization approval")
now=datetime.now(timezone.utc); not_before=instant(approval["not_before"],"approval.not_before"); approval_expiry=instant(approval["expires_at"],"approval.expires_at")
if not_before>now or approval_expiry<=now or approval_expiry<=not_before: raise PermissionError("organization approval inactive or expired")
ttl=finite_int(policy["svid_ttl_seconds"],"svid_ttl_seconds"); lifetime=finite_int(policy["max_enrollment_lifetime_seconds"],"max_enrollment_lifetime_seconds"); command_timeout=finite_int(policy["command_timeout_seconds"],"command_timeout_seconds")
approval_remaining=int((approval_expiry-now).total_seconds()); ttl=min(ttl,lifetime,approval_remaining)
if ttl<=0: raise PermissionError("no positive SVID lifetime remains under approval")
command=["spire-server","entry","create","-spiffeID",spiffe,"-parentID",parent,"-ttl",str(ttl)]
for selector in selectors: command.extend(["-selector",selector])
subprocess.run(command,check=True,timeout=command_timeout)
readback=subprocess.run(["spire-server","entry","show","-spiffeID",spiffe,"-output","json"],check=True,capture_output=True,timeout=command_timeout).stdout
Path("evidence/spire-entry-readback.raw.json").write_bytes(readback)
payload=json.loads(readback)
entries=payload.get("entries") if isinstance(payload,dict) and isinstance(payload.get("entries"),list) else payload if isinstance(payload,list) else [payload]
def normalized(entry):
    if not isinstance(entry,dict): raise ValueError("SPIRE entry must be an object")
    entry_id=entry.get("entry_id",entry.get("id")); entry_spiffe=entry.get("spiffe_id",entry.get("spiffeId")); entry_parent=entry.get("parent_id",entry.get("parentId")); raw_selectors=entry.get("selectors")
    if not isinstance(entry_id,str) or not entry_id or not isinstance(raw_selectors,list): raise ValueError("SPIRE readback fields missing")
    values=[]
    for item in raw_selectors:
        if isinstance(item,str): values.append(item)
        elif isinstance(item,dict) and set(item)>={"type","value"} and all(isinstance(item[key],str) and item[key] for key in ("type","value")): values.append(f"{item['type']}:{item['value']}")
        else: raise ValueError("SPIRE selector readback invalid")
    return {"entry_id":entry_id,"spiffe_id":entry_spiffe,"parent_id":entry_parent,"selectors":sorted(values)}
matches=[normalized(entry) for entry in entries if isinstance(entry,dict) and entry.get("spiffe_id",entry.get("spiffeId"))==spiffe]
if len(matches)!=1: raise ValueError("SPIRE readback must contain exactly one enrollment entry")
entry=matches[0]
if entry["spiffe_id"]!=spiffe or entry["parent_id"]!=parent or entry["selectors"]!=selectors: raise ValueError("SPIRE identity, parent, or selector readback differs")
readback_bytes=canonical(entry)+b"\n"; Path("evidence/spire-entry-readback.json").write_bytes(readback_bytes)
issued_at=now.isoformat().replace("+00:00","Z"); expires_at=min(approval_expiry,datetime.fromtimestamp(now.timestamp()+lifetime,timezone.utc)).isoformat().replace("+00:00","Z")
record={"schema_version":"aidefend.fl-enrollment.v1","enrollment_id":enrollment_id,"participant_id":participant,"spiffe_id":spiffe,"federation_id":federation,"model_sha256":model,"role":role,"status":"active","issued_at":issued_at,"expires_at":expires_at,"organization_approval_id":approval_id,"organization_id":organization,"public_jwk":jwk,"spire_entry_readback_sha256":hashlib.sha256(readback_bytes).hexdigest(),"selector_digest":hashlib.sha256(canonical(selectors)).hexdigest()}
if "REPLACE" in json.dumps(record): raise ValueError("generated enrollment retained a placeholder")
tmp=Path("evidence/participant-enrollment.tmp"); tmp.write_text(json.dumps(record,sort_keys=True,separators=(",",":"))+"\n",encoding="utf-8"); tmp.replace(Path("evidence/participant-enrollment.json"))
PY
cosign sign-blob --yes --key env://FL_ENROLLMENT_KEY --bundle evidence/participant-enrollment.sig evidence/participant-enrollment.json
cosign verify-blob --key keys/fl-enrollment.pub --bundle evidence/participant-enrollment.sig evidence/participant-enrollment.json

Independent verification

A federation auditor independently exports the exact SPIRE entry and selectors, recomputes the readback and selector digests, verifies organization approval and the current revocation registry, and tests unknown, expired, revoked, wrong-role, shared-key, and duplicate-participant enrollments.

spire-server entry show -spiffeID spiffe://example.org/fl/hospital-a/trainer-01 -output json > evidence/verifier-spire-entry.json
sha256sum evidence/verifier-spire-entry.json
cosign verify-blob --key keys/fl-enrollment.pub --bundle evidence/participant-enrollment.sig evidence/participant-enrollment.json

Failure handling

Missing organization approval, shared identity or key, mutable model alias, SPIRE selector/readback mismatch, expired or revoked enrollment, invalid policy value, signature error, or roster mismatch denies enrollment and update admission.

AID-H-008.003-G002

Verify one round-bound participant proof

Verify signed policy and enrollment first. The coordinator validates the peer certificate chain and exact SPIFFE URI, the enrollment's current SPIRE readback binding and revocation state, the enrolled JWS key and algorithm, exact round/update claims, freshness, and atomic one-time nonce use. It never trusts caller identity or attestation booleans.

Signed federation-round policy input

The federation-policy authority supplies evidence/round-policy.json with concrete positive verification/Redis/proof-age limits and a concrete nonnegative future-skew limit from the versioned round profile. The workflow verifies its detached signature before the coordinator validates the exact schema, numeric types, digest/round bindings, and complete participant evidence; this file is an input artifact, not an edit-time template.

Apply and capture evidence

# File: coordinator/verify_round_admission.py
from __future__ import annotations
import argparse, hashlib, json, math, re, subprocess, time
from datetime import datetime
from pathlib import Path
from cryptography import x509
from cryptography.x509.oid import ExtensionOID
from jwcrypto import jwk, jws
import redis
SHA=re.compile(r"^[a-f0-9]{64}$"); NONCE=re.compile(r"^[A-Za-z0-9_-]{16,128}$")
def load(path: Path) -> dict: return json.loads(path.read_text(encoding="utf-8"))
def epoch(value: str) -> float:
    if not isinstance(value,str) or not value: raise ValueError("timestamp missing")
    parsed=datetime.fromisoformat(value.replace("Z","+00:00"))
    if parsed.tzinfo is None: raise ValueError("timestamp lacks timezone")
    result=parsed.timestamp()
    if not math.isfinite(result): raise ValueError("timestamp not finite")
    return result
def number(value,positive=True):
    if isinstance(value,bool) or not isinstance(value,(int,float)) or not math.isfinite(value) or (value<=0 if positive else value<0): raise ValueError("invalid numeric policy value")
    return float(value)
def main() -> None:
    parser=argparse.ArgumentParser()
    for name in ("policy","enrollment","organization-approval","enrollment-roster","peer-cert","ca-file","token","update","spire-readback"): parser.add_argument("--"+name,type=Path,required=True)
    parser.add_argument("--redis-url",required=True); args=parser.parse_args(); policy,enrollment=load(args.policy),load(args.enrollment); approval=load(args.organization_approval); roster=load(args.enrollment_roster); now=time.time()
    required_policy={"schema_version","federation_id","model_sha256","participant_roster_sha256","round_id","allowed_roles","allowed_jws_algorithms","verification_timeout_seconds","redis_timeout_seconds","max_future_skew_seconds","max_proof_age_seconds","nonce_key_prefix","revoked_enrollment_ids"}
    required_enrollment={"schema_version","enrollment_id","participant_id","spiffe_id","federation_id","model_sha256","role","status","issued_at","expires_at","organization_approval_id","organization_id","public_jwk","spire_entry_readback_sha256","selector_digest"}
    if set(policy)!=required_policy or policy["schema_version"]!="aidefend.fl-round-policy.v1": raise ValueError("round policy schema differs")
    if set(enrollment)!=required_enrollment or enrollment["schema_version"]!="aidefend.fl-enrollment.v1": raise ValueError("enrollment schema differs")
    def strings(value,label,allow_empty=False):
        if not isinstance(value,list) or (not value and not allow_empty) or len(value)!=len(set(value)) or not all(isinstance(item,str) and item for item in value): raise ValueError(f"{label}: must be a unique string list")
        return value
    for field in ("federation_id","model_sha256","participant_roster_sha256","round_id","nonce_key_prefix"):
        if not isinstance(policy.get(field),str) or not policy[field]: raise ValueError(f"policy {field} invalid")
    if not SHA.fullmatch(policy["model_sha256"]) or not SHA.fullmatch(policy["participant_roster_sha256"]): raise ValueError("policy digest invalid")
    roles=strings(policy["allowed_roles"],"allowed_roles"); algorithms=strings(policy["allowed_jws_algorithms"],"allowed_jws_algorithms"); revoked=strings(policy["revoked_enrollment_ids"],"revoked_enrollment_ids",True)
    if not all(re.fullmatch(r"[A-Za-z0-9._:-]+",value) for value in roles+algorithms+revoked): raise ValueError("role, algorithm, or revocation identifier invalid")
    timeout=number(policy["verification_timeout_seconds"]); redis_timeout=number(policy["redis_timeout_seconds"]); future=number(policy["max_future_skew_seconds"],False); proof_age=number(policy["max_proof_age_seconds"])
    for field in ("enrollment_id","participant_id","spiffe_id","federation_id","role","status","issued_at","expires_at","organization_approval_id","organization_id"):
        if not isinstance(enrollment.get(field),str) or not enrollment[field]: raise ValueError(f"enrollment {field} invalid")
    if enrollment["role"] not in roles or enrollment["enrollment_id"] in revoked or enrollment["status"]!="active" or not SHA.fullmatch(enrollment.get("model_sha256","")): raise PermissionError("role, digest, status, or revocation policy denied")
    public_jwk=enrollment.get("public_jwk")
    if not isinstance(public_jwk,dict) or set(public_jwk)!={"kty","crv","x","y"} or public_jwk.get("kty")!="EC" or public_jwk.get("crv")!="P-256" or not all(isinstance(public_jwk.get(field),str) and public_jwk[field] for field in ("x","y")): raise ValueError("enrollment public JWK schema differs")
    readback_bytes=args.spire_readback.read_bytes()
    if not SHA.fullmatch(enrollment.get("spire_entry_readback_sha256","")) or hashlib.sha256(readback_bytes).hexdigest()!=enrollment["spire_entry_readback_sha256"]: raise PermissionError("SPIRE entry readback digest differs")
    readback=json.loads(readback_bytes)
    if not isinstance(readback,dict) or set(readback)!={"entry_id","spiffe_id","parent_id","selectors"} or readback.get("spiffe_id")!=enrollment["spiffe_id"]: raise PermissionError("SPIRE entry readback schema or identity differs")
    selectors=strings(readback["selectors"],"SPIRE selectors")
    selector_digest=hashlib.sha256(json.dumps(sorted(selectors),sort_keys=True,separators=(",",":")).encode()).hexdigest()
    if not SHA.fullmatch(enrollment.get("selector_digest","")) or selector_digest!=enrollment["selector_digest"]: raise PermissionError("SPIRE selector binding differs")
    if not isinstance(approval,dict) or set(approval)!={"schema_version","approval_id","organization_id","participant_id","federation_id","approved_roles","not_before","expires_at"} or approval.get("schema_version")!="aidefend.fl-organization-approval.v1": raise ValueError("organization approval schema differs")
    approved_roles=strings(approval["approved_roles"],"approved_roles")
    if approval.get("approval_id")!=enrollment["organization_approval_id"] or approval.get("organization_id")!=enrollment["organization_id"] or approval.get("participant_id")!=enrollment["participant_id"] or approval.get("federation_id")!=enrollment["federation_id"] or enrollment["role"] not in approved_roles or epoch(approval["not_before"])>now or epoch(approval["expires_at"])<=now: raise PermissionError("organization approval missing, inactive, or mismatched")
    if hashlib.sha256(args.enrollment_roster.read_bytes()).hexdigest()!=policy["participant_roster_sha256"]: raise PermissionError("signed participant roster digest differs")
    if not isinstance(roster,dict) or set(roster)!={"schema_version","federation_id","entries"} or roster.get("schema_version")!="aidefend.fl-enrollment-roster.v1" or roster.get("federation_id")!=policy["federation_id"] or not isinstance(roster.get("entries"),list) or not roster["entries"]: raise ValueError("participant roster schema differs")
    roster_rows=[]
    for index,row in enumerate(roster["entries"],1):
        if not isinstance(row,dict) or set(row)!={"enrollment_id","participant_id","spiffe_id","public_jwk_thumbprint"} or not all(isinstance(value,str) and value for value in row.values()) or not SHA.fullmatch(row["public_jwk_thumbprint"]): raise ValueError(f"roster:{index}: row schema differs")
        roster_rows.append(row)
    for field in ("enrollment_id","participant_id","spiffe_id","public_jwk_thumbprint"):
        values=[row[field] for row in roster_rows]
        if len(values)!=len(set(values)): raise PermissionError(f"roster shared/duplicate {field}")
    thumbprint=hashlib.sha256(json.dumps(public_jwk,sort_keys=True,separators=(",",":")).encode()).hexdigest()
    matches=[row for row in roster_rows if row["enrollment_id"]==enrollment["enrollment_id"]]
    if len(matches)!=1 or matches[0]!={"enrollment_id":enrollment["enrollment_id"],"participant_id":enrollment["participant_id"],"spiffe_id":enrollment["spiffe_id"],"public_jwk_thumbprint":thumbprint}: raise PermissionError("participant roster binding differs")
    subprocess.run(["openssl","verify","-CAfile",str(args.ca_file),str(args.peer_cert)],check=True,capture_output=True,text=True,timeout=timeout)
    cert=x509.load_pem_x509_certificate(args.peer_cert.read_bytes()); uris=cert.extensions.get_extension_for_oid(ExtensionOID.SUBJECT_ALTERNATIVE_NAME).value.get_values_for_type(x509.UniformResourceIdentifier)
    if uris!=[enrollment["spiffe_id"]] or not(cert.not_valid_before_utc.timestamp()<=now<cert.not_valid_after_utc.timestamp()): raise PermissionError("peer SPIFFE identity or SVID lifetime differs")
    if epoch(enrollment["issued_at"])>now+future or epoch(enrollment["expires_at"])<=now or epoch(enrollment["expires_at"])<=epoch(enrollment["issued_at"]): raise PermissionError("enrollment lifetime invalid or expired")
    for field in ("federation_id","model_sha256"):
        if enrollment[field]!=policy[field]: raise PermissionError(f"enrollment {field} differs")
    token=jws.JWS(); token.deserialize(args.token.read_text(encoding="utf-8").strip())
    if token.jose_header.get("alg") not in algorithms: raise PermissionError("JWS algorithm denied")
    token.verify(jwk.JWK(**enrollment["public_jwk"])); claims=json.loads(token.payload.decode("utf-8")); required={"participant_id","spiffe_id","federation_id","model_sha256","round_id","update_sha256","nonce","iat","exp"}
    if set(claims)!=required or not NONCE.fullmatch(claims.get("nonce","")): raise PermissionError("round-proof schema or nonce differs")
    expected={"participant_id":enrollment["participant_id"],"spiffe_id":enrollment["spiffe_id"],"federation_id":policy["federation_id"],"model_sha256":policy["model_sha256"],"round_id":policy["round_id"],"update_sha256":hashlib.sha256(args.update.read_bytes()).hexdigest()}
    if any(claims[key]!=value for key,value in expected.items()): raise PermissionError("round-proof binding differs")
    if any(isinstance(claims[key],bool) or not isinstance(claims[key],(int,float)) or not math.isfinite(claims[key]) for key in ("iat","exp")) or claims["iat"]>now+future or claims["exp"]<=now or claims["exp"]<=claims["iat"] or now-claims["iat"]>proof_age: raise PermissionError("round proof stale, future-dated, non-finite, or expired")
    client=redis.Redis.from_url(args.redis_url,decode_responses=True,socket_connect_timeout=redis_timeout,socket_timeout=redis_timeout)
    key=f"{policy['nonce_key_prefix']}:{policy['round_id']}:{claims['nonce']}"; lua="if redis.call('GET',KEYS[1])~=ARGV[1] then return 0 end redis.call('SET',KEYS[1],ARGV[2],'KEEPTTL') return 1"
    used=client.eval(lua,1,key,"issued:"+enrollment["participant_id"],"consumed:"+claims["update_sha256"])
    if used!=1: raise PermissionError("nonce unknown, expired, participant-mismatched, or replayed")
    print(json.dumps({**expected,"enrollment_id":enrollment["enrollment_id"]},sort_keys=True))
if __name__=="__main__": main()

Independent verification

Verify policy, enrollment, organization-approval, and complete-roster signatures before running the verifier, then rerun the same token as an executable replay negative. Also test missing approval, shared key, duplicate participant/enrollment/SPIFFE ID, selector or readback change, revoked enrollment, wrong SAN/key/role/model/round/update digest, NaN/Infinity time claims, malformed role/algorithm/revocation lists, and Redis unavailable.

set -euo pipefail
SNAPSHOT_DIR=$(mktemp -d)
trap 'rm -rf -- "$SNAPSHOT_DIR"' EXIT
for INPUT in evidence/round-policy.json evidence/participant-enrollment.json evidence/organization-approval.json evidence/enrollment-roster.json evidence/participant.crt keys/spiffe-bundle.pem evidence/round-proof.jws evidence/update.bin evidence/spire-entry-readback.json; do
  install -m 0600 -- "$INPUT" "$SNAPSHOT_DIR/$(basename "$INPUT")"
done
cosign verify-blob --key keys/fl-policy.pub --bundle evidence/round-policy.sig "$SNAPSHOT_DIR/round-policy.json"
cosign verify-blob --key keys/fl-enrollment.pub --bundle evidence/participant-enrollment.sig "$SNAPSHOT_DIR/participant-enrollment.json"
cosign verify-blob --key keys/fl-organization.pub --bundle evidence/organization-approval.sig "$SNAPSHOT_DIR/organization-approval.json"
cosign verify-blob --key keys/fl-roster.pub --bundle evidence/enrollment-roster.sig "$SNAPSHOT_DIR/enrollment-roster.json"
python coordinator/verify_round_admission.py --policy "$SNAPSHOT_DIR/round-policy.json" --enrollment "$SNAPSHOT_DIR/participant-enrollment.json" --organization-approval "$SNAPSHOT_DIR/organization-approval.json" --enrollment-roster "$SNAPSHOT_DIR/enrollment-roster.json" --peer-cert "$SNAPSHOT_DIR/participant.crt" --ca-file "$SNAPSHOT_DIR/spiffe-bundle.pem" --token "$SNAPSHOT_DIR/round-proof.jws" --update "$SNAPSHOT_DIR/update.bin" --spire-readback "$SNAPSHOT_DIR/spire-entry-readback.json" --redis-url redis://redis.fl.svc:6379/0
if python coordinator/verify_round_admission.py --policy "$SNAPSHOT_DIR/round-policy.json" --enrollment "$SNAPSHOT_DIR/participant-enrollment.json" --organization-approval "$SNAPSHOT_DIR/organization-approval.json" --enrollment-roster "$SNAPSHOT_DIR/enrollment-roster.json" --peer-cert "$SNAPSHOT_DIR/participant.crt" --ca-file "$SNAPSHOT_DIR/spiffe-bundle.pem" --token "$SNAPSHOT_DIR/round-proof.jws" --update "$SNAPSHOT_DIR/update.bin" --spire-readback "$SNAPSHOT_DIR/spire-entry-readback.json" --redis-url redis://redis.fl.svc:6379/0; then exit 1; fi

Failure handling

Fail closed on CA, certificate, SPIRE readback, JWK/JWS, policy schema, roster, role, revocation, clock, Redis, nonce, update binding, or coordinator errors. A timeout or ambiguous submission remains unadmitted until authoritative readback resolves it.

AID-H-008.003-G003

Permit dropout and robust filtering without admitting unreceipted updates

Eligible and admitted participants may legitimately not submit, and AID-H-008.002 may filter an accepted update. Require admissions to be a subset of eligible identities, one accept or deny receipt per received update, unique accepted participant/round/update bindings, and aggregation inputs to be a digest-matching subset of accepted receipts.

Concrete configuration

{"round_id":"round-0042","eligible":"evidence/eligible.json","admissions":"evidence/admissions.jsonl","received":"evidence/received-updates.jsonl","receipts":"evidence/update-receipts.jsonl","aggregation":"evidence/aggregation-inputs.jsonl"}

Apply and capture evidence

# File: verifier/verify_fl_round_population.py
import json
from pathlib import Path
def load_jsonl(path,key):
    rows={}
    for number,line in enumerate(Path(path).read_text().splitlines(),1):
        if not line.strip(): continue
        row=json.loads(line); value=row.get(key)
        if not isinstance(value,str) or not value or value in rows: raise ValueError(f"{path}:{number}: missing/duplicate {key}")
        rows[value]=row
    return rows
policy=json.loads(Path("evidence/round-population-policy.json").read_text()); round_id=policy["round_id"]
eligible=json.loads(Path("evidence/eligible.json").read_text())
if not isinstance(eligible,list) or len(eligible)!=len(set(eligible)): raise ValueError("eligible roster invalid")
eligible=set(eligible); admissions=load_jsonl("evidence/admissions.jsonl","participant_id")
if not set(admissions)<=eligible or any(row.get("round_id")!=round_id or row.get("decision")!="ACCEPTED" for row in admissions.values()): raise ValueError("admission roster differs")
received=load_jsonl("evidence/received-updates.jsonl","update_id"); receipts=load_jsonl("evidence/update-receipts.jsonl","update_id"); aggregation=load_jsonl("evidence/aggregation-inputs.jsonl","update_id")
if set(received)!=set(receipts): raise ValueError({"receipt_missing":sorted(set(received)-set(receipts)),"receipt_extra":sorted(set(receipts)-set(received))})
accepted={}; bindings=set()
for update_id,row in received.items():
    receipt=receipts[update_id]; participant=row.get("participant_id"); digest=row.get("update_sha256")
    if participant not in admissions or row.get("round_id")!=round_id or receipt.get("participant_id")!=participant or receipt.get("round_id")!=round_id or receipt.get("update_sha256")!=digest or receipt.get("decision") not in {"ACCEPTED","DENIED"}: raise ValueError(f"{update_id}: receipt binding differs")
    if receipt["decision"]=="ACCEPTED":
        binding=(participant,round_id,digest)
        if binding in bindings: raise ValueError(f"{update_id}: duplicate accepted participant/round/update binding")
        bindings.add(binding); accepted[update_id]=receipt
if not set(aggregation)<=set(accepted): raise ValueError("denied or unreceipted update entered aggregation")
for update_id,row in aggregation.items():
    if row.get("participant_id")!=accepted[update_id]["participant_id"] or row.get("round_id")!=round_id or row.get("update_sha256")!=accepted[update_id]["update_sha256"]: raise ValueError(f"{update_id}: aggregation digest binding differs")
out=Path("evidence/round-admission.pass"); tmp=Path("evidence/round-admission.pass.tmp"); tmp.write_text("PASS\n"); tmp.replace(out)

Independent verification

A separately credentialed coordinator auditor regenerates all five populations and reruns the verifier. Negative fixtures cover an ineligible participant, duplicate update, missing or duplicate receipt, wrong round or digest, denied or unreceipted aggregation input, and altered aggregation digest. Positive fixtures include legitimate participant dropout and an accepted update filtered by AID-H-008.002.

set -euo pipefail
rm -f evidence/round-admission.pass evidence/round-admission.pass.tmp
python verifier/verify_fl_round_population.py
sha256sum policy/fl-round-policy.json evidence/eligible.json evidence/admissions.jsonl evidence/received-updates.jsonl evidence/update-receipts.jsonl evidence/aggregation-inputs.jsonl evidence/round-admission.pass > evidence/round-admission.sha256

Failure handling

A failed export, invalid roster, missing or duplicate receipt, wrong participant/round/update binding, denied or unreceipted aggregator input, digest mismatch, or verifier disagreement blocks the round. Legitimate dropout and policy-authorized robust filtering do not fail this admission control.

Open-source tools

• Flower (Federated Learning Framework)
• NVIDIA FLARE
• FATE (Federated AI Technology Enabler)
• PySyft (OpenMined)
• SPIRE
• pyca/cryptography
• jwcrypto
• Valkey
• Redis 8+ (AGPL-3.0)
• Keylime

Commercial / hosted tools

• Microsoft Azure Attestation
• Intel Trust Authority
• Google Cloud Confidential Computing attestation
• AWS Nitro Enclaves attestation

Threat mappings

MITRE ATLAS

• AML.T0073 Impersonation (rejecting unauthorized, impersonated, revoked, replayed, wrong-round, wrong-model, or unattested federated participants without claiming that authorized clients are benign)

MAESTRO

• N/A

OWASP LLM Top 10 2026

• LLM05:2026 Data and Model Poisoning

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-14.1.2 Insufficient Access Controls (Restricts federated round admission to approved and unrevoked participants)
• AISubtech-4.3.4 Replay Exploitation (Rejects replayed federated updates; agentic communication is not assumed)
• AITech-6.1 Training Data Poisoning

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• N/A