AID-I-006: Federated Client Quarantine & Exclusion
Quarantine and exclude federated-learning clients whose cryptographically bound identities have been suspended or denied by an authoritative incident-response control plane. Enforce the deny decision at enrollment, client selection, update ingestion, training, retraining, and recovery/unlearning boundaries so a blocked participant cannot rejoin under the same identity or continue contributing cached updates.
Pillars: model
Lifecycle: response
Scope boundary
Owns authoritative, identity-bound federated-client quarantine decisions, signed deny-state distribution, and exclusion enforcement at enrollment, selection, update ingestion, training, retraining, and recovery. Statistical clustering and reputation may nominate investigation candidates but cannot independently prove malice or authorize durable quarantine.
AID-H-008.002: AID-I-006 excludes clients already named in a verified quarantine decision; AID-H-008.002 robustly aggregates updates from the accepted client population. Byzantine-robust aggregation is not identity-bound eviction, and quarantine does not define how accepted updates are combined or made robust to residual outliers.
AID-R-001.002: AID-I-006 produces and enforces the signed federated-client and update exclusion population; AID-R-001.002 consumes that population when building sanitized recovery-training inputs and validating a newly remediated model. Recovery retraining must not recreate the quarantine decision, and successful exclusion does not prove that the replacement model passed remediation evaluation.
Implementation guidance
- AID-I-006-G001: Generate non-authoritative investigation candidates by clustering historical model updates; do not quarantine from clustering alone.
- AID-I-006-G002: Operate the independent quarantine authority as a signed, mutually authenticated decision service.
- AID-I-006-G003: Exclude contributions from clients named in the signature-verified authoritative quarantine bundle during unlearning or retraining.
- AID-I-006-G004: Apply signed quarantine admission before passing accepted updates to the Byzantine-robust aggregation control owned by AID-H-008.002.
- AID-I-006-G005: Use reputation only to place a one-round non-authoritative safety hold and nominate the authenticated client to the quarantine authority.
- AID-I-006-G006: Verify a canonical, context-bound signed update envelope after quarantine admission and consume its update ID/nonce exactly once.
- AID-I-006-G007: Enforce signature-verified quarantine state from an authoritative deny service, using only a bounded-staleness last-known-good cache during short outages.
AID-I-006-G001
Generate non-authoritative investigation candidates by clustering historical model updates; do not quarantine from clustering alone.
Nomination-only analysis boundary
Clustering may nominate updates for independent investigation; it must not change enrollment, admission, quarantine state, aggregation, recovery, or training. Build one immutable, authenticated same-model/same-feature cohort. Algorithm, distance metric, normalization, parameters, minimum cohort coverage, and output limits come from a signature-verified analysis policy. There is no universal DBSCAN eps or min_samples.
Reproducible nomination module
# File: isolate/fl_candidate_analysis.py
# Dependencies: numpy>=2,<3, scikit-learn>=1.5,<2
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from typing import Mapping, Sequence, Callable
import numpy as np
import sklearn
from sklearn.cluster import DBSCAN
@dataclass(frozen=True)
class PolicyVerifier:
"""Required authenticated runtime adapter callables; no fallback implementation."""
verify_analysis_policy: Callable[[str], Mapping[str, object]]
@dataclass(frozen=True)
class EvidenceClient:
"""Required authenticated runtime adapter callables; no fallback implementation."""
append: Callable[[Mapping[str, object]], Mapping[str, object]]
read: Callable[[str], Mapping[str, object]]
def canonical_sha256(value: object) -> str:
data = json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=True
).encode("utf-8")
return hashlib.sha256(data).hexdigest()
def nominate(
*,
cohort: Sequence[Mapping[str, object]],
cohort_manifest_sha256: str,
policy_jws: str,
policy_verifier: PolicyVerifier,
evidence: EvidenceClient,
) -> Mapping[str, object]:
policy = policy_verifier.verify_analysis_policy(policy_jws)
if policy["algorithm"] != "DBSCAN":
raise ValueError("this module implements only signed-policy DBSCAN")
for field in (
"expected_cohort_count", "minimum_samples",
"feature_count", "maximum_candidates",
):
value = policy.get(field)
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
raise ValueError("signed analysis integer is invalid: " + field)
expected_count = policy["expected_cohort_count"]
if len(cohort) != expected_count or len(cohort) < policy["minimum_samples"]:
raise ValueError("cohort population is incomplete or too small")
update_ids = [str(item["update_id"]) for item in cohort]
if len(set(update_ids)) != len(update_ids):
raise ValueError("cohort contains duplicate update IDs")
if canonical_sha256([
{
"update_id": item["update_id"],
"client_id": item["authenticated_client_id"],
"model_id": item["model_id"],
"feature_schema_sha256": item["feature_schema_sha256"],
"vector_sha256": item["vector_sha256"],
}
for item in cohort
]) != cohort_manifest_sha256:
raise ValueError("cohort manifest binding mismatch")
model_ids = {str(item["model_id"]) for item in cohort}
schemas = {str(item["feature_schema_sha256"]) for item in cohort}
if model_ids != {str(policy["model_id"])} or schemas != {
str(policy["feature_schema_sha256"])
}:
raise ValueError("mixed model or feature cohort")
for item in cohort:
vector = item.get("finite_feature_vector")
if (
not isinstance(vector, list)
or canonical_sha256(vector) != str(item.get("vector_sha256", ""))
):
raise ValueError("feature vector does not match its claimed digest")
vectors = np.asarray(
[item["finite_feature_vector"] for item in cohort],
dtype=np.dtype(str(policy["dtype"])),
)
if (
vectors.ndim != 2
or vectors.shape[1] != policy["feature_count"]
or not np.isfinite(vectors).all()
):
raise ValueError("cohort is not one finite rectangular feature matrix")
if policy["normalization"] == "SIGNED_ZSCORE":
mean = np.asarray(policy["feature_mean"], dtype=vectors.dtype)
scale = np.asarray(policy["feature_scale"], dtype=vectors.dtype)
if (
mean.shape != (vectors.shape[1],)
or scale.shape != (vectors.shape[1],)
or not np.isfinite(mean).all()
or not np.isfinite(scale).all()
or np.any(scale <= 0)
):
raise ValueError("invalid signed normalization vectors")
vectors = (vectors - mean) / scale
elif policy["normalization"] != "NONE":
raise ValueError("unsupported signed normalization")
eps = policy.get("eps")
if isinstance(eps, bool) or not isinstance(eps, (int, float)):
raise ValueError("signed DBSCAN epsilon must be numeric")
eps = float(eps)
if not np.isfinite(eps) or eps <= 0:
raise ValueError("signed DBSCAN epsilon must be finite and positive")
clustering = DBSCAN(
eps=eps,
min_samples=policy["minimum_samples"],
metric=str(policy["metric"]),
algorithm=str(policy["search_algorithm"]),
).fit(vectors)
candidate_rows = [
{
"update_id": cohort[index]["update_id"],
"authenticated_client_id": cohort[index]["authenticated_client_id"],
"label": int(label),
}
for index, label in enumerate(clustering.labels_)
if int(label) == -1
]
if len(candidate_rows) > policy["maximum_candidates"]:
raise ValueError("candidate population exceeds signed review capacity")
event = {
"schema_version": 1,
"control_id": "AID-I-006",
"guidance_ref": "AID-I-006-G001",
"decision": "NOMINATE_FOR_INDEPENDENT_REVIEW_ONLY",
"cohort_manifest_sha256": cohort_manifest_sha256,
"cohort_count": len(cohort),
"matrix_sha256": hashlib.sha256(
np.ascontiguousarray(vectors).tobytes(order="C")
).hexdigest(),
"policy_sha256": policy["policy_sha256"],
"scikit_learn_version": sklearn.__version__,
"algorithm_parameters": {
"eps": policy["eps"],
"minimum_samples": policy["minimum_samples"],
"metric": policy["metric"],
"search_algorithm": policy["search_algorithm"],
"normalization": policy["normalization"],
},
"candidate_count": len(candidate_rows),
"candidate_population_sha256": canonical_sha256(candidate_rows),
"candidates": candidate_rows,
"authoritative_state_changed": False,
"outcome": "PASS",
}
event_sha256 = canonical_sha256(event)
receipt = evidence.append(event)
readback = evidence.read(str(receipt["receipt_id"]))
if (
receipt.get("event_sha256") != event_sha256
or readback.get("event_sha256") != event_sha256
):
raise RuntimeError("evidence receipt/readback does not bind the exact event")
return {**event, "evidence_receipt_id": receipt["receipt_id"]}
Integration and verification
The cohort loader must derive client identity from authenticated transport/admission receipts, verify every immutable vector digest and finite shape, and bind the exact snapshot/population root. The policy verifier pins issuer, audience, type, algorithm, key status, applicability, monotonic version, and all numeric parameters; reject NaN/infinity and unsupported metric/normalization combinations. Submit candidate evidence to the independent authority through its nomination API. Independently rerun the analysis from the immutable cohort and prove, by reading the current signed quarantine bundle and admission logs, that this candidate list changed no authoritative state. A reproducible empty candidate set is valid PASS; incomplete coverage is INSUFFICIENT_DATA; clustering/evidence faults are ERROR.
AID-I-006-G002
Operate the independent quarantine authority as a signed, mutually authenticated decision service.
Independent decision authority
The training coordinator can nominate evidence but cannot write deny state. Run the authority under a separate workload/admin identity and database. Every request uses directly verified mTLS/SPIFFE context supplied by mandatory server middleware—not a caller-controlled HTTP identity header. Only registered federation scopes produce bundles. Nomination is non-authoritative; a separate incident-response decision writer consumes action-bound approval and advances the bundle version transactionally.
Authoritative PostgreSQL state
CREATE TABLE fl_registered_federation (
authority_id text NOT NULL,
environment_id text NOT NULL,
federation_id text NOT NULL,
model_id text NOT NULL,
bundle_version bigint NOT NULL CHECK (bundle_version BETWEEN 1 AND 9007199254740991),
policy_sha256 text NOT NULL,
PRIMARY KEY (authority_id, environment_id, federation_id)
);
CREATE TABLE fl_quarantine_nomination (
receipt_id text PRIMARY KEY,
caller_spiffe_id text NOT NULL,
idempotency_key_sha256 text NOT NULL,
nomination_sha256 text NOT NULL,
nomination_json jsonb NOT NULL,
receipt_json jsonb NOT NULL,
created_at timestamptz NOT NULL,
UNIQUE (caller_spiffe_id, idempotency_key_sha256)
);
CREATE TABLE fl_quarantine_decision (
authority_id text NOT NULL,
environment_id text NOT NULL,
federation_id text NOT NULL,
client_id text NOT NULL,
active boolean NOT NULL,
decision_id text NOT NULL,
approval_receipt_sha256 text NOT NULL,
evidence_manifest_sha256 text NOT NULL,
effective_bundle_version bigint NOT NULL,
decided_at timestamptz NOT NULL,
PRIMARY KEY (
authority_id, environment_id, federation_id, client_id
),
UNIQUE (decision_id),
FOREIGN KEY (authority_id, environment_id, federation_id)
REFERENCES fl_registered_federation
(authority_id, environment_id, federation_id)
);
mTLS nomination and signed bundle service
# File: isolate/fl_quarantine_authority_service.py
# Dependencies: fastapi>=0.115,<1, pydantic>=2.9,<3,
# psycopg[binary,pool]>=3.2,<4, jcs>=0.2,<1, PyNaCl>=1.5,<2
from __future__ import annotations
import hashlib as _aidefend_policy_hashlib
import json as _aidefend_policy_json
import math as _aidefend_policy_math
import os as _aidefend_policy_os
import subprocess as _aidefend_policy_subprocess
import tempfile as _aidefend_policy_tempfile
from pathlib import Path as _AidefendPolicyPath
def _aidefend_reject_duplicate_keys(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate signed-policy JSON key: {key}")
value[key] = item
return value
def _aidefend_reject_nonfinite(value):
raise ValueError(f"non-finite signed-policy JSON value: {value}")
def _aidefend_load_signed_policy(*, path_env, signature_env, key_env, schema_version):
policy_path = _AidefendPolicyPath(_aidefend_policy_os.environ[path_env])
signature_path = _AidefendPolicyPath(_aidefend_policy_os.environ[signature_env])
verify_timeout = float(_aidefend_policy_os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
if not _aidefend_policy_math.isfinite(verify_timeout) or verify_timeout <= 0:
raise RuntimeError("policy-verification timeout is invalid")
payload = policy_path.read_bytes()
signature = signature_path.read_bytes()
if not payload or not signature:
raise RuntimeError("signed policy payload or signature is empty")
with _aidefend_policy_tempfile.TemporaryDirectory(prefix="aidefend-policy-") as directory:
root = _AidefendPolicyPath(directory)
_aidefend_policy_os.chmod(root, 0o700)
payload_path = root / "policy.json"
signature_path = root / "policy.sig"
payload_path.write_bytes(payload)
signature_path.write_bytes(signature)
_aidefend_policy_os.chmod(payload_path, 0o400)
_aidefend_policy_os.chmod(signature_path, 0o400)
_aidefend_policy_subprocess.run(
["cosign", "verify-blob", "--key", _aidefend_policy_os.environ[key_env],
"--bundle", str(signature_path), str(payload_path)],
check=True, capture_output=True, text=True, timeout=verify_timeout,
)
verified = payload_path.read_bytes()
if verified != payload:
raise RuntimeError("verified policy snapshot changed")
policy = _aidefend_policy_json.loads(
verified.decode("utf-8", errors="strict"),
object_pairs_hook=_aidefend_reject_duplicate_keys,
parse_constant=_aidefend_reject_nonfinite,
)
if (not isinstance(policy, dict) or policy.get("schema_version") != schema_version
or not isinstance(policy.get("policy_version"), str) or not policy["policy_version"]):
raise RuntimeError("signature-verified policy schema or version is invalid")
if "_aidefend_verified_sha256" in policy:
raise RuntimeError("signed policy uses a reserved evidence field")
policy["_aidefend_verified_sha256"] = _aidefend_policy_hashlib.sha256(verified).hexdigest()
return policy
import base64
import hashlib
import os
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Protocol
from typing import Literal
import jcs
from fastapi import FastAPI, Header, HTTPException, Request
from nacl.signing import SigningKey
from psycopg_pool import ConnectionPool
from psycopg.types.json import Jsonb
from pydantic import BaseModel, Field, field_validator
AUTHORITY_ID = os.environ["FL_QUARANTINE_AUTHORITY_ID"]
ENVIRONMENT_ID = os.environ["FL_QUARANTINE_ENVIRONMENT_ID"]
POLICY_SHA256 = os.environ["FL_QUARANTINE_POLICY_SHA256"]
BUNDLE_TTL_SECONDS = int(
os.environ["FL_QUARANTINE_BUNDLE_TTL_SECONDS"]
)
if BUNDLE_TTL_SECONDS <= 0:
raise RuntimeError("invalid signed-policy bundle TTL")
SIGNING_KEY = SigningKey(
bytes.fromhex(
open(
os.environ["FL_QUARANTINE_SIGNING_KEY_PATH"],
"rt",
encoding="ascii",
).read().strip()
)
)
FL_NOMINATION_POLICY = _aidefend_load_signed_policy(
path_env="I006_NOMINATION_POLICY_PATH",
signature_env="I006_NOMINATION_POLICY_SIGNATURE",
key_env="I006_NOMINATION_POLICY_VERIFY_KEY",
schema_version="aidefend.fl-quarantine-nomination-policy.v1",
)
if (
POLICY_SHA256 != "sha256:" + FL_NOMINATION_POLICY["_aidefend_verified_sha256"]
or FL_NOMINATION_POLICY.get("bundle_ttl_seconds") != BUNDLE_TTL_SECONDS
):
raise RuntimeError("quarantine policy digest or bundle TTL projection mismatch")
_FL_BOUND_NAMES = (
"database_pool_min_size", "database_pool_max_size",
"federation_id_min_length", "federation_id_max_length",
"client_id_min_length", "client_id_max_length",
"idempotency_key_min_length", "idempotency_key_max_length",
"maximum_isolated_clients",
)
FL_BOUNDS = {name: FL_NOMINATION_POLICY.get(name) for name in _FL_BOUND_NAMES}
if any(
isinstance(value, bool) or not isinstance(value, int) or value < 1
for value in FL_BOUNDS.values()
):
raise RuntimeError("signed FL nomination bounds are invalid")
for minimum_name, maximum_name in (
("database_pool_min_size", "database_pool_max_size"),
("federation_id_min_length", "federation_id_max_length"),
("client_id_min_length", "client_id_max_length"),
("idempotency_key_min_length", "idempotency_key_max_length"),
):
if FL_BOUNDS[minimum_name] > FL_BOUNDS[maximum_name]:
raise RuntimeError("signed FL nomination minimum exceeds maximum")
pool = ConnectionPool(
os.environ["FL_QUARANTINE_DATABASE_URL"],
min_size=FL_BOUNDS["database_pool_min_size"],
max_size=FL_BOUNDS["database_pool_max_size"],
kwargs={"autocommit": False},
)
app = FastAPI()
@dataclass(frozen=True)
class VerifiedWorkload:
spiffe_id: str
trust_domain: str
peer_certificate_sha256: str
class Nomination(BaseModel):
schema_version: Literal["aidefend.fl_quarantine_nomination.v1"]
federation_id: str = Field(min_length=FL_BOUNDS["federation_id_min_length"], max_length=FL_BOUNDS["federation_id_max_length"])
client_id: str = Field(min_length=FL_BOUNDS["client_id_min_length"], max_length=FL_BOUNDS["client_id_max_length"])
evidence_sha256: str = Field(pattern=r"^sha256:[a-f0-9]{64}$")
requested_action: Literal["independent_review"]
@field_validator("federation_id", "client_id")
@classmethod
def safe_identifier(cls, value: str) -> str:
allowed = set(
"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._:/-"
)
if any(character not in allowed for character in value):
raise ValueError("identifier contains unsupported characters")
return value
def canonical(value: object) -> bytes:
return bytes(jcs.canonicalize(value))
def sha256(value: bytes) -> str:
return "sha256:" + hashlib.sha256(value).hexdigest()
def sign_statement(value: dict) -> str:
signature = SIGNING_KEY.sign(canonical(value)).signature
return base64.b64encode(signature).decode("ascii")
def require_verified_workload(request: Request) -> VerifiedWorkload:
# Set only by direct mTLS/SPIFFE middleware after peer-certificate
# validation and trust-domain mapping. No HTTP header fallback exists.
workload = getattr(request.state, "verified_workload", None)
if (
not isinstance(workload, VerifiedWorkload)
or not workload.spiffe_id.startswith(
"spiffe://" + workload.trust_domain + "/"
)
or not workload.peer_certificate_sha256.startswith("sha256:")
):
raise HTTPException(
status_code=401, detail="verified workload identity required"
)
return workload
def require_federation_entitlement(
workload: VerifiedWorkload, federation_id: str, role: str
) -> None:
entitlements = FL_NOMINATION_POLICY.get("federation_entitlements")
if (
not isinstance(entitlements, list)
or any(
not isinstance(item, dict)
or set(item) != {"caller_spiffe_id", "federation_id", "role"}
for item in entitlements
)
):
raise RuntimeError("signed federation entitlement policy is invalid")
expected = {
"caller_spiffe_id": workload.spiffe_id,
"federation_id": federation_id,
"role": role,
}
if expected not in entitlements:
raise HTTPException(status_code=403, detail="federation caller is not entitled")
def require_registered_scope(conn, federation_id: str) -> tuple:
row = conn.execute(
"SELECT bundle_version, policy_sha256, model_id "
"FROM fl_registered_federation "
"WHERE authority_id=%s AND environment_id=%s "
"AND federation_id=%s FOR SHARE",
(AUTHORITY_ID, ENVIRONMENT_ID, federation_id),
).fetchone()
if row is None or row[1] != POLICY_SHA256:
raise HTTPException(
status_code=404, detail="federation scope is not registered"
)
return row
@app.post("/v1/nominations")
def create_nomination(
request: Request,
body: Nomination,
idempotency_key: str = Header(
alias="Idempotency-Key", min_length=FL_BOUNDS["idempotency_key_min_length"], max_length=FL_BOUNDS["idempotency_key_max_length"]
),
):
workload = require_verified_workload(request)
nomination = body.model_dump()
nomination_sha256 = sha256(canonical(nomination))
key_sha256 = sha256(idempotency_key.encode("utf-8"))
receipt_id = sha256(canonical({
"caller_spiffe_id": workload.spiffe_id,
"idempotency_key_sha256": key_sha256,
}))
now = datetime.now(timezone.utc)
unsigned_receipt = {
"schema_version":
"aidefend.fl_quarantine_nomination_receipt.v1",
"receipt_id": receipt_id,
"authority_id": AUTHORITY_ID,
"environment_id": ENVIRONMENT_ID,
"federation_id": body.federation_id,
"caller_spiffe_id": workload.spiffe_id,
"peer_certificate_sha256":
workload.peer_certificate_sha256,
"nomination_sha256": nomination_sha256,
"policy_sha256": POLICY_SHA256,
"accepted_for": "INDEPENDENT_REVIEW_ONLY",
"created_at": now.isoformat(),
}
receipt = {
**unsigned_receipt,
"signature_b64": sign_statement(unsigned_receipt),
}
with pool.connection() as conn, conn.transaction():
require_federation_entitlement(workload, body.federation_id, "NOMINATE")
require_registered_scope(conn, body.federation_id)
conn.execute(
"INSERT INTO fl_quarantine_nomination "
"(receipt_id, caller_spiffe_id, idempotency_key_sha256, "
"nomination_sha256, nomination_json, receipt_json, created_at) "
"VALUES (%s,%s,%s,%s,%s,%s,%s) "
"ON CONFLICT (caller_spiffe_id, idempotency_key_sha256) "
"DO NOTHING",
(
receipt_id,
workload.spiffe_id,
key_sha256,
nomination_sha256,
Jsonb(nomination),
Jsonb(receipt),
now,
),
)
stored = conn.execute(
"SELECT nomination_sha256, receipt_json "
"FROM fl_quarantine_nomination "
"WHERE caller_spiffe_id=%s AND idempotency_key_sha256=%s",
(workload.spiffe_id, key_sha256),
).fetchone()
if stored is None or stored[0] != nomination_sha256:
raise HTTPException(
status_code=409,
detail="idempotency key reused for another nomination",
)
return stored[1]
@app.get("/v1/deny-bundle")
def deny_bundle(
request: Request,
authority_id: str,
environment_id: str,
federation_id: str,
):
workload = require_verified_workload(request)
if authority_id != AUTHORITY_ID or environment_id != ENVIRONMENT_ID:
raise HTTPException(
status_code=403, detail="authority/environment mismatch"
)
with pool.connection() as conn, conn.transaction():
require_federation_entitlement(workload, federation_id, "READ_BUNDLE")
version, policy_sha256, model_id = require_registered_scope(
conn, federation_id
)
rows = conn.execute(
"SELECT client_id FROM fl_quarantine_decision "
"WHERE authority_id=%s AND environment_id=%s "
"AND federation_id=%s AND active=true "
"AND effective_bundle_version <= %s "
"ORDER BY client_id LIMIT %s",
(
AUTHORITY_ID,
ENVIRONMENT_ID,
federation_id,
version,
FL_BOUNDS["maximum_isolated_clients"] + 1,
),
).fetchall()
isolated_clients = [row[0] for row in rows]
if len(isolated_clients) > FL_BOUNDS["maximum_isolated_clients"]:
raise HTTPException(status_code=503, detail="deny population exceeds signed bound")
if isinstance(version, bool) or not isinstance(version, int) or not 1 <= version <= 9007199254740991:
raise HTTPException(status_code=503, detail="bundle version exceeds exact JCS integer range")
now = datetime.now(timezone.utc)
payload = {
"schema_version": "aidefend.fl_quarantine.v1",
"authority_id": AUTHORITY_ID,
"environment_id": ENVIRONMENT_ID,
"federation_id": federation_id,
"model_id": model_id,
"version": version,
"issued_at": now.isoformat(),
"expires_at": (
now + timedelta(seconds=BUNDLE_TTL_SECONDS)
).isoformat(),
"policy_sha256": policy_sha256,
"isolated_clients": isolated_clients,
"isolated_population_sha256": sha256(
canonical(isolated_clients)
),
}
return {
"payload": payload,
"signature_b64": sign_statement(payload),
}
Decision-writer transaction
The separate incident-response writer verifies a one-time approval bound to authority/environment/federation, authenticated client ID, action (DENY or RESTORE), evidence manifest, current bundle version, policy, and approvers. In one serializable transaction it locks the federation row, advances bundle_version, upserts exactly one decision with that new version, and writes an append-only decision/outbox event. It never accepts a coordinator-generated decision or clustering/reputation result as approval. The signing key is read-only to the bundle service (prefer a signing service/HSM in high-assurance deployments); database roles separate nomination, decision, and bundle reads.
Verification
Deploy mandatory direct mTLS middleware and ensure the proxy strips identity headers; validate trust domain and workload entitlement for nomination versus bundle read. Pin JCS, key ID/registry version, policy digest, database migration and service build in evidence. Independently query registered scope, version and exact sorted client population, replay the Ed25519 signature, and test unregistered federation, caller header spoofing, idempotency mutation/race, policy/version rollback, same-version mutation, decision-writer concurrency, database/signing outage, wrong trust domain, and expiry. A nomination receipt proves only accepted review evidence. Only a current signature-verified bundle governs durable quarantine.
Version identifies the complete quarantine decision, not the fetch timestamp. G007 computes a stable JCS digest excluding only issued_at/expires_at and verifies the signed, fresh envelope separately. Advance bundle_version in the same transaction whenever scope, model, policy or isolated-client membership changes; do not advance it merely to refresh envelope lifetime. Keep version within the exact JCS integer range.
AID-I-006-G003
Exclude contributions from clients named in the signature-verified authoritative quarantine bundle during unlearning or retraining.
Concept:
Recovery filtering must consume the same signature-verified, scope-bound quarantine state as online admission. Never accept a caller-supplied exclusion list, a clustering result, or an unsigned incident spreadsheet. The recovery receipt must bind the authoritative bundle version and payload digest so an auditor can prove exactly which deny state governed the job. The request path must not inject its own evidence callback; use one root-owned, peer-pinned evidence client and verify its detached signature locally.
Step 1: Pin and cryptographically verify the evidence service
# File: isolate/signed_evidence.py
from __future__ import annotations
import base64
import hashlib
import json
import re
from pathlib import Path
from dataclasses import dataclass
from typing import Final, Callable
from nacl.exceptions import BadSignatureError
from nacl.signing import VerifyKey
from isolate.evidence_transport import connect_pinned_evidence_sink
@dataclass(frozen=True)
class SignedEvidenceSink:
"""Required authenticated runtime adapter callables; no fallback implementation."""
append_signed: Callable[[dict], dict]
RECEIPT_FIELDS = {
"schema_version", "receipt_id", "event_sha256", "signature_b64",
}
RECEIPT_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/-]{0,255}")
VERIFY_KEY = VerifyKey(bytes.fromhex(
Path("/etc/federated-learning/evidence-service.pub").read_text().strip()
))
TRUSTED_SINK: Final[SignedEvidenceSink] = connect_pinned_evidence_sink(
unix_socket="/run/aidefend/fl-evidence.sock",
expected_peer_uid=0,
expected_service_identity="aidefend-fl-evidence-prod",
)
def canonical(value: object) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
def append_verified_event(event: dict) -> dict:
if (
not isinstance(event, dict)
or not isinstance(event.get("schema_version"), str)
or not event["schema_version"]
):
raise RuntimeError("signed evidence event schema is missing")
event_bytes = canonical(event)
event_sha256 = "sha256:" + hashlib.sha256(event_bytes).hexdigest()
receipt = TRUSTED_SINK.append_signed(event)
if (
not isinstance(receipt, dict)
or set(receipt) != RECEIPT_FIELDS
or receipt.get("schema_version") != "aidefend.fl-evidence-receipt.v1"
or receipt.get("event_sha256") != event_sha256
or not isinstance(receipt.get("receipt_id"), str)
or RECEIPT_ID.fullmatch(receipt["receipt_id"]) is None
or not isinstance(receipt.get("signature_b64"), str)
):
raise RuntimeError("evidence receipt schema or event binding is invalid")
statement = canonical({
"schema_version": receipt["schema_version"],
"receipt_id": receipt["receipt_id"],
"event_sha256": event_sha256,
"event": event,
})
try:
signature = base64.b64decode(receipt["signature_b64"], validate=True)
VERIFY_KEY.verify(statement, signature)
except (BadSignatureError, ValueError, TypeError) as exc:
raise RuntimeError("evidence service signature is invalid") from exc
return receipt
Step 2: Build a recovery cohort from verified quarantine state
# File: isolate/fl_unlearning_isolation.py
from __future__ import annotations
import hashlib
import json
from isolate.fl_quarantine_authority import load_quarantine_state
from isolate.signed_evidence import append_verified_event
def canonical_sha256(value: object) -> str:
encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
return "sha256:" + hashlib.sha256(encoded).hexdigest()
def prepare_verified_recovery_cohort(
full_historical_dataset: list[tuple[str, object]],
historical_dataset_root_sha256: str,
expected_federation_id: str,
expected_model_id: str,
model_release_sha256: str,
recovery_job_id: str,
) -> tuple[list[tuple[str, object]], dict]:
for name, digest in {
"historical_dataset_root_sha256": historical_dataset_root_sha256,
"model_release_sha256": model_release_sha256,
}.items():
if (
not isinstance(digest, str)
or not digest.startswith("sha256:")
or len(digest) != 71
or any(ch not in "0123456789abcdef" for ch in digest[7:])
):
raise ValueError(f"invalid immutable digest: {name}")
# load_quarantine_state verifies signature, pinned authority/environment/
# federation/model scope, freshness, and the scope-specific monotonic high-water.
denied_clients, bundle, bundle_payload_sha256, source = load_quarantine_state(
expected_federation_id=expected_federation_id,
expected_model_id=expected_model_id,
)
clean_dataset = [
(client_id, sample)
for client_id, sample in full_historical_dataset
if client_id not in denied_clients
]
included_client_ids = sorted({client_id for client_id, _ in clean_dataset})
excluded_client_ids = sorted(
{client_id for client_id, _ in full_historical_dataset} & denied_clients
)
record = {
"schema_version": "aidefend.fl_recovery_exclusion.v1",
"recovery_job_id": recovery_job_id,
"authority_id": bundle["authority_id"],
"environment_id": bundle["environment_id"],
"federation_id": bundle["federation_id"],
"model_id": expected_model_id,
"model_release_sha256": model_release_sha256,
"historical_dataset_root_sha256": historical_dataset_root_sha256,
"quarantine_bundle_version": bundle["version"],
"quarantine_bundle_payload_sha256": bundle_payload_sha256,
"quarantine_state_source": source,
"included_client_ids_sha256": canonical_sha256(included_client_ids),
"excluded_client_ids": excluded_client_ids,
"input_record_count": len(full_historical_dataset),
"output_record_count": len(clean_dataset),
}
record_sha256 = canonical_sha256(record)
sink_receipt = append_verified_event(record)
if sink_receipt["event_sha256"] != record_sha256:
raise RuntimeError("signed recovery receipt does not bind the exact record")
return clean_dataset, sink_receipt
Action: Derive exclusions only from load_quarantine_state(), fail the job when signature, scope, freshness, LKG, or high-water validation fails, and require an externally signed append-only receipt over the exact bundle digest/version and recovery cohort before unlearning or retraining starts.
AID-I-006-G004
Apply signed quarantine admission before passing accepted updates to the Byzantine-robust aggregation control owned by AID-H-008.002.
Concept:
This technique must not define a second robust-aggregation algorithm. AID-H-008.002 owns Byzantine-resilient aggregation for updates that remain admissible. AID-I-006 verifies identity and the signed update, enforces the current scope-bound quarantine bundle, and obtains a signed append-only admission receipt before an update reaches that aggregator.
Quarantine-to-aggregation integration contract
# File: isolate/fl_quarantine_aggregation_boundary.py
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Callable
from uuid import uuid4
import numpy as np
import redis
from isolate.fl_quarantine_authority import DenyAuthorityUnavailable
from isolate.signed_evidence import append_verified_event
from isolate.verify_signed_update import ClientKeyRegistry, UpdateRejected, verify_and_consume_update
@dataclass(frozen=True)
class AuthenticatedSubmission:
# Construct only after mTLS/workload-identity authentication.
authenticated_client_id: str
envelope: dict
@dataclass(frozen=True)
class AcceptedUpdate:
client_id: str
round_id: str
update_id: str
delta: np.ndarray
quarantine_bundle_version: int
quarantine_bundle_payload_sha256: str
admission_receipt_id: str
@dataclass(frozen=True)
class H008002Aggregator:
"""Required authenticated runtime adapter callables; no fallback implementation."""
aggregate: Callable[[list[AcceptedUpdate]], np.ndarray]
def append_admission_event(event: dict) -> dict:
immutable_event = {
"schema_version": "aidefend.fl_admission.v1",
"event_id": str(uuid4()),
"observed_at": datetime.now(timezone.utc).isoformat(),
**event,
}
return append_verified_event(immutable_event)
def run_federated_round(
submissions: list[AuthenticatedSubmission],
expected_federation_id: str,
expected_model_id: str,
expected_round_id: str,
key_registry: ClientKeyRegistry,
replay_store: redis.Redis,
h008002_aggregator: H008002Aggregator,
) -> np.ndarray:
accepted: list[AcceptedUpdate] = []
for submission in submissions:
try:
delta, evidence = verify_and_consume_update(
envelope=submission.envelope,
authenticated_client_id=submission.authenticated_client_id,
expected_federation_id=expected_federation_id,
expected_model_id=expected_model_id,
expected_round_id=expected_round_id,
key_registry=key_registry,
replay_store=replay_store,
)
except DenyAuthorityUnavailable:
append_admission_event({
"client_id": submission.authenticated_client_id,
"federation_id": expected_federation_id,
"model_id": expected_model_id,
"round_id": expected_round_id,
"decision": "pause_round",
"reason": "quarantine_authority_unavailable_or_invalid",
})
raise
except (UpdateRejected, PermissionError) as exc:
append_admission_event({
"client_id": submission.authenticated_client_id,
"federation_id": expected_federation_id,
"model_id": expected_model_id,
"round_id": expected_round_id,
"decision": "deny",
"reason": type(exc).__name__,
})
continue
receipt = append_admission_event({
**evidence,
"decision": "admit_to_h008002",
})
accepted.append(AcceptedUpdate(
client_id=evidence["client_id"],
round_id=evidence["round_id"],
update_id=evidence["update_id"],
delta=delta,
quarantine_bundle_version=evidence["quarantine_bundle_version"],
quarantine_bundle_payload_sha256=evidence["quarantine_bundle_payload_sha256"],
admission_receipt_id=receipt["receipt_id"],
))
if not accepted:
raise RuntimeError("no fully verified updates with signed admission receipts; pause the round")
return h008002_aggregator.aggregate(accepted)
Action: Route every submission only through verify_and_consume_update(). Require the external append-only evidence service to return a signature and digest-matching receipt before admission; evidence-service, authority, LKG, high-water, or replay-store failure pauses the round. Keep aggregation and poisoning-resilience tests in AID-H-008.002.
AID-I-006-G005
Use reputation only to place a one-round non-authoritative safety hold and nominate the authenticated client to the quarantine authority.
Bounded decision
Reputation is a fallible signed signal. It may hold an already authenticated and enrolled client out of exactly one identified round and nominate immutable evidence; it must never add to the authoritative deny bundle, survive into another round, or supply a default score for an unknown client. Threshold, score/model version, eligible client cohort, maximum candidate population, and missing-signal behavior come from signed policy. If authority state is unavailable, #7 decides whether the whole round pauses; this module must not invent a local durable deny.
Durable one-round ledger
CREATE TABLE fl_reputation_round_decision (
environment_id text NOT NULL,
federation_id text NOT NULL,
round_id text NOT NULL,
authenticated_client_id text NOT NULL,
admission_receipt_id text NOT NULL,
reputation_signal_sha256 bytea NOT NULL
CHECK (octet_length(reputation_signal_sha256) = 32),
score numeric NOT NULL,
score_model_version text NOT NULL,
policy_sha256 bytea NOT NULL CHECK (octet_length(policy_sha256) = 32),
decision text NOT NULL CHECK (
decision IN ('NO_HOLD', 'HOLD_THIS_ROUND', 'INSUFFICIENT_DATA')
),
nomination_id text,
nomination_receipt_sha256 bytea,
created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
PRIMARY KEY (
environment_id, federation_id, round_id, authenticated_client_id
),
CHECK (
(nomination_id IS NULL) =
(nomination_receipt_sha256 IS NULL)
)
);
-- A new round has a different primary key and receives no inherited HOLD.
-- Admission checks only the exact current round:
SELECT decision
FROM fl_reputation_round_decision
WHERE environment_id = $1
AND federation_id = $2
AND round_id = $3
AND authenticated_client_id = $4;
Evaluation module
# File: isolate/fl_reputation_round_hold.py
from __future__ import annotations
import hashlib
import json
from decimal import Decimal
from dataclasses import dataclass
from typing import Mapping, Callable
@dataclass(frozen=True)
class SignalVerifier:
"""Required authenticated runtime adapter callables; no fallback implementation."""
verify: Callable[[str, str, Mapping[str, object], Mapping[str, object], Mapping[str, object]], Mapping[str, object]]
@dataclass(frozen=True)
class RoundLedger:
"""Required authenticated runtime adapter callables; no fallback implementation."""
reserve_once: Callable[[Mapping[str, object]], Mapping[str, object]]
attach_nomination: Callable[[str, Mapping[str, object]], None]
@dataclass(frozen=True)
class QuarantineAuthority:
"""Required authenticated runtime adapter callables; no fallback implementation."""
nominate: Callable[[str, Mapping[str, object]], Mapping[str, object]]
@dataclass(frozen=True)
class EvidenceClient:
"""Required authenticated runtime adapter callables; no fallback implementation."""
append: Callable[[Mapping[str, object]], Mapping[str, object]]
read: Callable[[str], Mapping[str, object]]
def canonical_sha256(value: object) -> str:
data = json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=True
).encode("utf-8")
return hashlib.sha256(data).hexdigest()
def evaluate_for_round(
*,
authenticated_client_id: str,
admission_receipt: Mapping[str, object],
round_context: Mapping[str, object],
signal_jws: str,
policy: Mapping[str, object],
verifier: SignalVerifier,
ledger: RoundLedger,
authority: QuarantineAuthority,
evidence: EvidenceClient,
) -> Mapping[str, object]:
signal = verifier.verify(
signal_jws,
authenticated_client_id=authenticated_client_id,
admission_receipt=admission_receipt,
round_context=round_context,
policy=policy,
)
if signal.get("policy_sha256") != policy.get("policy_sha256"):
raise ValueError("reputation signal policy binding mismatch")
score = Decimal(str(signal["score"]))
threshold = Decimal(str(policy["hold_below_score"]))
if not score.is_finite() or not threshold.is_finite():
raise ValueError("reputation score and threshold must be finite")
decision_name = (
"HOLD_THIS_ROUND" if score < threshold else "NO_HOLD"
)
decision = {
"environment_id": round_context["environment_id"],
"federation_id": round_context["federation_id"],
"round_id": round_context["round_id"],
"authenticated_client_id": authenticated_client_id,
"admission_receipt_id": admission_receipt["receipt_id"],
"reputation_signal_sha256": canonical_sha256(signal),
"score": str(score),
"score_model_version": signal["score_model_version"],
"policy_sha256": policy["policy_sha256"],
"decision": decision_name,
"authoritative_quarantine_changed": False,
}
reservation = ledger.reserve_once(decision)
nomination = None
if decision_name == "HOLD_THIS_ROUND":
nomination_evidence = {
**decision,
"requested_action": "INDEPENDENT_REVIEW",
"evidence_manifest_sha256": signal["evidence_manifest_sha256"],
}
nomination = authority.nominate(
idempotency_key=str(reservation["reservation_id"]),
evidence=nomination_evidence,
)
ledger.attach_nomination(
str(reservation["reservation_id"]), nomination
)
event = {
**decision,
"reservation_id": reservation["reservation_id"],
"nomination": nomination,
"outcome": "PASS",
}
event_sha256 = canonical_sha256(event)
receipt = evidence.append(event)
readback = evidence.read(str(receipt["receipt_id"]))
if (
receipt.get("event_sha256") != event_sha256
or readback.get("event_sha256") != event_sha256
):
raise RuntimeError("evidence receipt/readback does not bind the exact event")
return {**event, "evidence_receipt_id": receipt["receipt_id"]}
Closure evidence
The verifier must pin signal/policy issuer, audience, type, algorithm, key state, scope, client, model/version, exact round, finite score and immutable evidence manifest; the admission receipt proves identity/enrollment and must be independent of reputation. The ledger uses a conditional insert and returns the same result on retry. The authority response must be signed and bind the nomination, scope, round, client and evidence digest. At round close, sign the complete admitted/held client population and independently verify that each hold appears only for that round. At the next round, prove there is no inherited local hold and compare the current authority bundle; only that bundle can impose durable exclusion. A missing/invalid score or nomination/evidence outage is not PASS; follow signed missing-signal/fail-safe policy without converting it to local quarantine.
AID-I-006-G006
Verify a canonical, context-bound signed update envelope after quarantine admission and consume its update ID/nonce exactly once.
Concept:
A signature over raw vector bytes does not bind an update to a client, federation, model, round, shape, dtype, or time window. Require a canonical envelope that binds that context plus a one-time update ID and nonce. Derive client identity from authenticated transport, enforce the pinned federation's signed quarantine state, then atomically consume the replay key before aggregation.
Signed update envelope
{
"payload": {
"schema_version": "aidefend.fl_update.v1",
"client_id": "spiffe://example.org/fl/client-a",
"federation_id": "fraud-model-us",
"model_id": "fraud-model-v42",
"round_id": "round-000184",
"update_id": "upd-7d946ce4-89ac-4ce9-b871-2d746efe9a75",
"nonce": "qSfCDzaW81mSgYtTgS5mNQ",
"issued_at": "2026-07-10T18:00:00Z",
"expires_at": "2026-07-10T18:05:00Z",
"dtype": "float32",
"shape": [4096],
"delta_b64": "BASE64_FLOAT32_LITTLE_ENDIAN_BYTES"
},
"signature_b64": "BASE64_ED25519_SIGNATURE"
}Server-side verification and replay consumption
# File: isolate/verify_signed_update.py
from __future__ import annotations
import base64
import hashlib
import json
import math
import os
import re
from datetime import datetime, timedelta, timezone
from dataclasses import dataclass
from typing import Callable
import numpy as np
import redis
import jcs
from nacl.exceptions import BadSignatureError
from nacl.signing import VerifyKey
from isolate.fl_quarantine_authority import require_client_allowed
UPDATE_POLICY_SHA256 = os.environ["FL_UPDATE_POLICY_SHA256"]
MAX_ELEMENTS = int(os.environ["FL_UPDATE_MAX_ELEMENTS"])
MAX_RANK = int(os.environ["FL_UPDATE_MAX_RANK"])
MAX_ENVELOPE_LIFETIME = timedelta(
seconds=int(os.environ["FL_UPDATE_MAX_LIFETIME_SECONDS"])
)
MAX_CLOCK_SKEW = timedelta(
seconds=int(os.environ["FL_UPDATE_MAX_CLOCK_SKEW_SECONDS"])
)
if (
MAX_ELEMENTS <= 0
or MAX_RANK <= 0
or MAX_ENVELOPE_LIFETIME.total_seconds() <= 0
or MAX_CLOCK_SKEW.total_seconds() < 0
):
raise RuntimeError("invalid signature-verified update-policy bounds")
ID_VALUE = re.compile(r"[A-Za-z0-9._:/-]{1,200}")
NONCE = re.compile(r"[A-Za-z0-9_-]{16,128}")
PAYLOAD_FIELDS = {
"schema_version", "client_id", "federation_id", "model_id", "round_id",
"update_id", "nonce", "issued_at", "expires_at", "dtype", "shape", "delta_b64",
}
class UpdateRejected(PermissionError):
pass
@dataclass(frozen=True)
class ClientKeyRegistry:
"""Required authenticated runtime adapter callables; no fallback implementation."""
active_verify_key: Callable[[str, datetime, str], tuple[bytes, str, str]]
def parse_time(value: object) -> datetime:
if not isinstance(value, str):
raise UpdateRejected("timestamp must be a string")
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError as exc:
raise UpdateRejected("timestamp is not ISO-8601") from exc
if parsed.tzinfo is None:
raise UpdateRejected("timestamp must include a timezone")
return parsed
def verify_and_consume_update(
envelope: dict,
authenticated_client_id: str,
expected_federation_id: str,
expected_model_id: str,
expected_round_id: str,
key_registry: ClientKeyRegistry,
replay_store: redis.Redis,
) -> tuple[np.ndarray, dict]:
quarantine = require_client_allowed(
client_id=authenticated_client_id,
expected_federation_id=expected_federation_id,
expected_model_id=expected_model_id,
)
if (
not isinstance(envelope, dict)
or set(envelope) != {"payload", "signature_b64"}
or not isinstance(envelope.get("payload"), dict)
):
raise UpdateRejected("invalid envelope shape")
payload = envelope["payload"]
if set(payload) != PAYLOAD_FIELDS:
raise UpdateRejected("missing or unexpected signed payload fields")
if payload["schema_version"] != "aidefend.fl_update.v1":
raise UpdateRejected("unsupported update schema")
if payload["client_id"] != authenticated_client_id:
raise UpdateRejected("signed client differs from authenticated client")
expected_context = {
"federation_id": expected_federation_id,
"model_id": expected_model_id,
"round_id": expected_round_id,
}
for field, expected in expected_context.items():
if payload[field] != expected:
raise UpdateRejected(f"signed update context mismatch: {field}")
for field in ("client_id", "federation_id", "model_id", "round_id", "update_id"):
if not isinstance(payload[field], str) or ID_VALUE.fullmatch(payload[field]) is None:
raise UpdateRejected(f"invalid identifier: {field}")
if not isinstance(payload["nonce"], str) or NONCE.fullmatch(payload["nonce"]) is None:
raise UpdateRejected("invalid nonce")
now = datetime.now(timezone.utc)
issued_at = parse_time(payload["issued_at"])
expires_at = parse_time(payload["expires_at"])
if issued_at > now + MAX_CLOCK_SKEW or expires_at <= now:
raise UpdateRejected("update time window invalid")
if expires_at <= issued_at or expires_at - issued_at > MAX_ENVELOPE_LIFETIME:
raise UpdateRejected("update lifetime exceeds policy")
if payload["dtype"] != "float32":
raise UpdateRejected("only canonical float32 updates are accepted")
shape = payload["shape"]
if not isinstance(shape, list) or not 1 <= len(shape) <= MAX_RANK:
raise UpdateRejected("invalid update rank")
if any(isinstance(item, bool) or not isinstance(item, int) or item < 1 for item in shape):
raise UpdateRejected("invalid update dimensions")
elements = math.prod(shape)
if elements > MAX_ELEMENTS:
raise UpdateRejected("update exceeds element limit")
if not isinstance(payload["delta_b64"], str) or len(payload["delta_b64"]) > 4 * ((elements * 4 + 2) // 3):
raise UpdateRejected("encoded update size exceeds signed shape")
canonical = bytes(jcs.canonicalize(payload))
key_bytes, key_id, key_registry_version = (
key_registry.active_verify_key(
authenticated_client_id=authenticated_client_id,
issued_at=issued_at,
policy_sha256=UPDATE_POLICY_SHA256,
)
)
try:
if (
not isinstance(envelope["signature_b64"], str)
or len(envelope["signature_b64"]) > 128
):
raise ValueError("update signature size invalid")
signature = base64.b64decode(envelope["signature_b64"], validate=True)
if len(signature) != 64:
raise ValueError("update signature length invalid")
VerifyKey(key_bytes).verify(canonical, signature)
delta_bytes = base64.b64decode(payload["delta_b64"], validate=True)
except (BadSignatureError, ValueError, TypeError) as exc:
raise UpdateRejected("signature or base64 encoding invalid") from exc
if len(delta_bytes) != elements * np.dtype("<f4").itemsize:
raise UpdateRejected("signed shape does not match delta byte length")
delta = np.frombuffer(delta_bytes, dtype="<f4").reshape(shape).copy()
if not np.isfinite(delta).all():
raise UpdateRejected("signed update contains NaN or infinity")
ttl = max(1, math.ceil((expires_at - now).total_seconds()))
replay_scope = json.dumps({
"schema_version": "aidefend.fl_update_replay_scope.v1",
"authority_id": quarantine["authority_id"],
"environment_id": quarantine["environment_id"],
"federation_id": quarantine["federation_id"],
"model_id": expected_model_id,
"client_id": authenticated_client_id,
"round_id": expected_round_id,
"update_id": payload["update_id"],
"nonce": payload["nonce"],
}, sort_keys=True, separators=(",", ":")).encode("utf-8")
replay_key = "fl-update-consumed:" + hashlib.sha256(replay_scope).hexdigest()
if not replay_store.set(replay_key, expected_round_id, nx=True, ex=ttl):
raise UpdateRejected("update replay detected")
evidence = {
"client_id": authenticated_client_id,
"authority_id": quarantine["authority_id"],
"environment_id": quarantine["environment_id"],
"federation_id": quarantine["federation_id"],
"model_id": expected_model_id,
"round_id": expected_round_id,
"update_id": payload["update_id"],
"update_payload_sha256": "sha256:" + hashlib.sha256(canonical).hexdigest(),
"client_update_key_id": key_id,
"client_key_registry_version": key_registry_version,
"quarantine_bundle_version": quarantine["bundle_version"],
"quarantine_bundle_payload_sha256": quarantine["bundle_payload_sha256"],
"quarantine_state_source": quarantine["source"],
"update_policy_sha256": UPDATE_POLICY_SHA256,
"replay_key_consumed": True,
}
return delta, evidence
Action: Derive authenticated_client_id outside request data, require the coordinator's pinned federation to match both the signed update and the signed quarantine bundle, verify finite shape/dtype/time/signature, and consume the fully scoped update ID/nonce with Redis/Valkey SET NX. Preserve the update-payload and quarantine-bundle digests in the signed admission receipt.
Load one exact signed update policy
Replace the environment-derived UPDATE_POLICY_SHA256, element/rank bounds, lifetime, and clock-skew constants above with one verified object. Environment variables may identify the immutable object and a bootstrap byte ceiling; they must not supply policy values.
# File: isolate/verified_update_policy.py
import base64
import hashlib
import json
import os
import re
import stat
from datetime import timedelta
from pathlib import Path
import jcs
from nacl.signing import VerifyKey
POLICY_ROOT = Path("/etc/aidefend/fl-update-policies")
POLICY_REF = os.environ["FL_UPDATE_POLICY_REF"]
POLICY_MAX_BYTES = int(os.environ["FL_UPDATE_POLICY_MAX_BYTES"])
if (
re.fullmatch(r"sha256:[0-9a-f]{64}", POLICY_REF) is None
or POLICY_MAX_BYTES < 2
or POLICY_MAX_BYTES > 1_048_576
):
raise RuntimeError("immutable update-policy bootstrap is invalid")
def regular_bytes(path: Path, maximum: int) -> bytes:
if isinstance(maximum, bool) or not isinstance(maximum, int) or maximum < 1:
raise ValueError("maximum must be a positive integer")
required = ("O_CLOEXEC", "O_NOFOLLOW")
if any(not hasattr(os, name) for name in required):
raise OSError("required no-follow file primitives are unavailable")
descriptor = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
try:
before = os.fstat(descriptor)
if not stat.S_ISREG(before.st_mode) or before.st_uid != 0 or before.st_mode & 0o022:
raise PermissionError("update-policy object ownership or mode invalid")
if before.st_size < 1 or before.st_size > maximum:
raise ValueError("update-policy object size invalid")
raw = os.read(descriptor, before.st_size + 1)
after = os.fstat(descriptor)
identity = lambda value: (
value.st_dev, value.st_ino, value.st_size,
value.st_mtime_ns, value.st_ctime_ns,
)
if len(raw) != before.st_size or identity(before) != identity(after):
raise OSError("update-policy object changed during read")
return raw
finally:
os.close(descriptor)
def unique_object(pairs):
result = {}
for key, value in pairs:
if key in result:
raise RuntimeError("duplicate update-policy JSON member")
result[key] = value
return result
def reject_constant(value):
raise RuntimeError("non-finite update-policy value: " + value)
raw = regular_bytes(
POLICY_ROOT / (POLICY_REF.removeprefix("sha256:") + ".signed.json"),
POLICY_MAX_BYTES,
)
if "sha256:" + hashlib.sha256(raw).hexdigest() != POLICY_REF:
raise RuntimeError("immutable update-policy object digest mismatch")
envelope = json.loads(
raw.decode("utf-8", errors="strict"),
object_pairs_hook=unique_object,
parse_constant=reject_constant,
)
if not isinstance(envelope, dict) or set(envelope) != {"payload", "signature_b64"}:
raise RuntimeError("update-policy envelope differs")
payload = envelope["payload"]
required = {
"schema_version", "policy_version", "signing_key_id",
"max_elements", "max_rank", "max_lifetime_seconds",
"max_clock_skew_seconds",
}
if not isinstance(payload, dict) or set(payload) != required:
raise RuntimeError("update-policy payload fields differ")
policy_bytes = bytes(jcs.canonicalize(payload))
signature = base64.b64decode(envelope["signature_b64"], validate=True)
verify_key = VerifyKey(regular_bytes(
Path("/etc/aidefend/trust/fl-update-policy-root.ed25519"), 32
))
verify_key.verify(policy_bytes, signature)
if (
payload["schema_version"] != "aidefend.fl-update-policy.v1"
or not isinstance(payload["policy_version"], str)
or not payload["policy_version"]
or not isinstance(payload["signing_key_id"], str)
or not payload["signing_key_id"]
):
raise RuntimeError("update-policy identity is invalid")
for field in ("max_elements", "max_rank", "max_lifetime_seconds"):
if isinstance(payload[field], bool) or not isinstance(payload[field], int) or payload[field] < 1:
raise RuntimeError("signed update-policy bound is invalid: " + field)
if (
isinstance(payload["max_clock_skew_seconds"], bool)
or not isinstance(payload["max_clock_skew_seconds"], int)
or payload["max_clock_skew_seconds"] < 0
):
raise RuntimeError("signed update-policy clock skew is invalid")
UPDATE_POLICY_SHA256 = "sha256:" + hashlib.sha256(policy_bytes).hexdigest()
MAX_ELEMENTS = payload["max_elements"]
MAX_RANK = payload["max_rank"]
MAX_ENVELOPE_LIFETIME = timedelta(seconds=payload["max_lifetime_seconds"])
MAX_CLOCK_SKEW = timedelta(seconds=payload["max_clock_skew_seconds"])Load this module once for one admission operation and carry POLICY_REF, policy_version, signing_key_id, and UPDATE_POLICY_SHA256 into the key-registry lookup and admission receipt. Rotation selects a different immutable ref; it never mutates bytes at the same ref.
AID-I-006-G007
Enforce signature-verified quarantine state from an authoritative deny service, using only a bounded-staleness last-known-good cache during short outages.
Concept:
An empty or unavailable deny store must never mean “allow everyone.” Bind every signed bundle to one pinned authority, deployment environment, and federation; a valid bundle from another scope is an authorization failure. Cache and monotonic high-water keys are derived from that same scope, so one federation cannot roll back or substitute another federation's state. A fresh last-known-good bundle may cover only a short authority outage; after its bounded staleness or signed expiry, the round pauses.
Signed, scope-bound bundle shape
{
"payload": {
"schema_version": "aidefend.fl_quarantine.v1",
"authority_id": "fl-quarantine-prod-v1",
"environment_id": "prod-us",
"federation_id": "fraud-model-us",
"version": 42,
"issued_at": "2026-07-10T18:00:00Z",
"expires_at": "2026-07-10T18:20:00Z",
"isolated_clients": ["spiffe://example.org/fl/client-z"]
},
"signature_b64": "BASE64_ED25519_SIGNATURE"
}Fetch, verify, cache, and enforce
# File: isolate/fl_quarantine_authority.py
from __future__ import annotations
import base64
import hashlib
import json
import math
import os
import re
import secrets
import stat
from datetime import datetime, timedelta, timezone
from pathlib import Path
import jcs
import redis
import requests
from nacl.exceptions import BadSignatureError
from nacl.signing import VerifyKey
# The deployment composition root verifies a signed, versioned policy and
# projects these mandatory values into a root-owned environment. There are no
# framework defaults. Record POLICY_SHA256 in every admission receipt.
POLICY_VERSION = os.environ["FL_QUARANTINE_POLICY_VERSION"]
POLICY_SHA256 = os.environ["FL_QUARANTINE_POLICY_SHA256"]
if (
not POLICY_VERSION
or re.fullmatch(r"sha256:[0-9a-f]{64}", POLICY_SHA256) is None
):
raise RuntimeError("invalid signed quarantine policy identity")
EXPECTED_AUTHORITY_ID = os.environ["FL_QUARANTINE_AUTHORITY_ID"]
EXPECTED_ENVIRONMENT_ID = os.environ["FL_QUARANTINE_ENVIRONMENT_ID"]
EXPECTED_FEDERATION_ID = os.environ["FL_QUARANTINE_FEDERATION_ID"]
EXPECTED_MODEL_ID = os.environ["FL_QUARANTINE_MODEL_ID"]
AUTHORITY_URL = os.environ["FL_QUARANTINE_AUTHORITY_URL"]
if not AUTHORITY_URL.startswith("https://"):
raise RuntimeError("quarantine authority endpoint must use HTTPS")
CLIENT_CERT_PATH = os.environ["FL_QUARANTINE_CLIENT_CERT_PATH"]
CLIENT_KEY_PATH = os.environ["FL_QUARANTINE_CLIENT_KEY_PATH"]
CA_BUNDLE_PATH = os.environ["FL_QUARANTINE_CA_BUNDLE_PATH"]
REQUEST_TIMEOUT_SECONDS = float(
os.environ["FL_QUARANTINE_REQUEST_TIMEOUT_SECONDS"]
)
SCOPE_CANONICAL = "|".join([
EXPECTED_AUTHORITY_ID,
EXPECTED_ENVIRONMENT_ID,
EXPECTED_FEDERATION_ID,
EXPECTED_MODEL_ID,
])
SCOPE_KEY = hashlib.sha256(SCOPE_CANONICAL.encode("utf-8")).hexdigest()
LKG_PATH = Path(f"/var/lib/fl-coordinator/quarantine-lkg-{SCOPE_KEY}.json")
HIGH_WATER_KEY = f"aidefend:fl-quarantine:accepted-high-water:{SCOPE_KEY}"
VERIFY_KEY = VerifyKey(
bytes.fromhex(
Path(os.environ["FL_QUARANTINE_AUTHORITY_KEY_PATH"]).read_text().strip()
)
)
STATE_STORE = redis.Redis.from_url(
os.environ["QUARANTINE_STATE_REDIS_URL"],
decode_responses=True,
ssl_cert_reqs="required",
)
MAX_STALENESS = timedelta(
seconds=int(os.environ["FL_QUARANTINE_MAX_STALENESS_SECONDS"])
)
MAX_BUNDLE_LIFETIME = timedelta(
seconds=int(os.environ["FL_QUARANTINE_MAX_BUNDLE_LIFETIME_SECONDS"])
)
MAX_FUTURE_SKEW = timedelta(
seconds=int(os.environ["FL_QUARANTINE_MAX_FUTURE_SKEW_SECONDS"])
)
MAX_RESPONSE_BYTES = int(os.environ["FL_QUARANTINE_MAX_RESPONSE_BYTES"])
STREAM_CHUNK_BYTES = int(os.environ["FL_QUARANTINE_STREAM_CHUNK_BYTES"])
if (
REQUEST_TIMEOUT_SECONDS <= 0
or MAX_RESPONSE_BYTES <= 0
or STREAM_CHUNK_BYTES <= 0
or MAX_STALENESS.total_seconds() <= 0
or MAX_BUNDLE_LIFETIME.total_seconds() <= 0
or MAX_FUTURE_SKEW.total_seconds() < 0
):
raise RuntimeError("invalid signed-policy quarantine bounds")
CLIENT_ID = re.compile(r"^[A-Za-z0-9._:/-]{1,300}$")
ADVANCE_HIGH_WATER = """
local current_version = redis.call('HGET', KEYS[1], 'version')
local current_digest = redis.call('HGET', KEYS[1], 'decision_sha256')
if current_version then
if ARGV[1] < current_version then return -1 end
if ARGV[1] == current_version and current_digest ~= ARGV[2] then return -2 end
end
if (not current_version) or ARGV[1] > current_version then
redis.call('HSET', KEYS[1], 'version', ARGV[1], 'decision_sha256', ARGV[2])
return 1
end
return 0
"""
class DenyAuthorityUnavailable(RuntimeError):
pass
def reject_duplicate_members(pairs: list[tuple[str, object]]) -> dict:
result = {}
for key, value in pairs:
if key in result:
raise DenyAuthorityUnavailable(f"duplicate JSON member: {key}")
result[key] = value
return result
def reject_nonfinite(value: str) -> None:
raise DenyAuthorityUnavailable(f"non-finite JSON number: {value}")
def parse_strict_json(raw: bytes) -> object:
try:
text = raw.decode("utf-8", errors="strict")
return json.loads(
text,
object_pairs_hook=reject_duplicate_members,
parse_constant=reject_nonfinite,
)
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise DenyAuthorityUnavailable("quarantine JSON is invalid") from exc
def parse_time(value: object) -> datetime:
if not isinstance(value, str):
raise DenyAuthorityUnavailable("quarantine timestamp must be a string")
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError as exc:
raise DenyAuthorityUnavailable("quarantine timestamp is not ISO-8601") from exc
if parsed.tzinfo is None:
raise DenyAuthorityUnavailable("quarantine timestamp must include a timezone")
return parsed
def verify_bundle(
bundle: dict,
now: datetime,
) -> tuple[set[str], dict, str]:
if (
not isinstance(bundle, dict)
or set(bundle) != {"payload", "signature_b64"}
or not isinstance(bundle.get("payload"), dict)
):
raise DenyAuthorityUnavailable("quarantine bundle envelope invalid")
payload = bundle["payload"]
required = {
"schema_version",
"authority_id",
"environment_id",
"federation_id",
"model_id",
"version",
"issued_at",
"expires_at",
"policy_sha256",
"isolated_clients",
"isolated_population_sha256",
}
if set(payload) != required or payload.get("schema_version") != "aidefend.fl_quarantine.v1":
raise DenyAuthorityUnavailable("quarantine bundle schema invalid")
pinned_scope = {
"authority_id": EXPECTED_AUTHORITY_ID,
"environment_id": EXPECTED_ENVIRONMENT_ID,
"federation_id": EXPECTED_FEDERATION_ID,
"model_id": EXPECTED_MODEL_ID,
}
for field, expected in pinned_scope.items():
if payload.get(field) != expected:
raise DenyAuthorityUnavailable(f"quarantine bundle scope mismatch: {field}")
version = payload.get("version")
if (
isinstance(version, bool)
or not isinstance(version, int)
or not 1 <= version <= 9_007_199_254_740_991
):
raise DenyAuthorityUnavailable("quarantine bundle version invalid")
clients = payload.get("isolated_clients")
if (
not isinstance(clients, list)
or any(
not isinstance(item, str) or CLIENT_ID.fullmatch(item) is None
for item in clients
)
or clients != sorted(set(clients))
):
raise DenyAuthorityUnavailable("isolated client list invalid or duplicated")
client_population = bytes(jcs.canonicalize(clients))
expected_population_sha256 = (
"sha256:" + hashlib.sha256(client_population).hexdigest()
)
if (
payload["policy_sha256"] != POLICY_SHA256
or payload["isolated_population_sha256"]
!= expected_population_sha256
):
raise DenyAuthorityUnavailable("quarantine policy or population binding mismatch")
canonical = bytes(jcs.canonicalize(payload))
try:
if (
not isinstance(bundle["signature_b64"], str)
or len(bundle["signature_b64"]) > 128
):
raise ValueError("quarantine signature size invalid")
signature = base64.b64decode(bundle["signature_b64"], validate=True)
if len(signature) != 64:
raise ValueError("quarantine signature length invalid")
VERIFY_KEY.verify(canonical, signature)
except (BadSignatureError, ValueError, TypeError, KeyError) as exc:
raise DenyAuthorityUnavailable("quarantine bundle signature invalid") from exc
issued_at = parse_time(payload["issued_at"])
expires_at = parse_time(payload["expires_at"])
if issued_at > now + MAX_FUTURE_SKEW:
raise DenyAuthorityUnavailable("quarantine bundle issued in the future")
if (
expires_at <= issued_at
or expires_at - issued_at > MAX_BUNDLE_LIFETIME
or now >= expires_at
or now - issued_at > MAX_STALENESS
):
raise DenyAuthorityUnavailable("quarantine bundle is stale or overlong")
payload_sha256 = "sha256:" + hashlib.sha256(canonical).hexdigest()
return set(clients), payload, payload_sha256
def decision_sha256(payload: dict) -> str:
# Envelope time can refresh without changing the versioned decision.
# All scope, policy, model and exact population fields remain committed.
decision = {key: value for key, value in payload.items()
if key not in {"issued_at", "expires_at"}}
return "sha256:" + hashlib.sha256(bytes(jcs.canonicalize(decision))).hexdigest()
def advance_high_water(payload: dict, payload_sha256: str) -> None:
result = int(
STATE_STORE.eval(
ADVANCE_HIGH_WATER,
1,
HIGH_WATER_KEY,
f"{payload['version']:020d}",
decision_sha256(payload),
)
)
if result == -1:
raise DenyAuthorityUnavailable("signed quarantine bundle rollback detected")
if result == -2:
raise DenyAuthorityUnavailable("same quarantine version has different decision state")
def require_lkg_at_high_water(payload: dict, payload_sha256: str) -> None:
state = STATE_STORE.hgetall(HIGH_WATER_KEY)
if not state:
raise DenyAuthorityUnavailable("no scope-specific quarantine high-water state")
if (
state.get("version") != f"{payload['version']:020d}"
or state.get("decision_sha256") != decision_sha256(payload)
):
raise DenyAuthorityUnavailable("LKG is not this scope's accepted high-water bundle")
def write_lkg(bundle: dict) -> None:
LKG_PATH.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
directory = LKG_PATH.parent.lstat()
if (
not stat.S_ISDIR(directory.st_mode)
or stat.S_ISLNK(directory.st_mode)
or directory.st_mode & 0o077
):
raise DenyAuthorityUnavailable("LKG directory is not protected")
raw = json.dumps(
bundle, sort_keys=True, separators=(",", ":"), allow_nan=False
).encode("utf-8")
if len(raw) > MAX_RESPONSE_BYTES:
raise DenyAuthorityUnavailable("LKG exceeds size policy")
tmp = LKG_PATH.with_name(
LKG_PATH.name + "." + secrets.token_hex(16) + ".tmp"
)
descriptor = os.open(
tmp,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
0o600,
)
try:
with os.fdopen(descriptor, "wb", closefd=True) as handle:
handle.write(raw)
handle.flush()
os.fsync(handle.fileno())
os.replace(tmp, LKG_PATH)
directory_fd = os.open(
LKG_PATH.parent, os.O_RDONLY | os.O_DIRECTORY
)
try:
os.fsync(directory_fd)
finally:
os.close(directory_fd)
finally:
if tmp.exists():
tmp.unlink()
def load_verified_lkg(now: datetime) -> tuple[set[str], dict, str, str]:
try:
descriptor = os.open(
LKG_PATH, os.O_RDONLY | os.O_NOFOLLOW
)
with os.fdopen(descriptor, "rb", closefd=True) as handle:
metadata = os.fstat(handle.fileno())
if (
not stat.S_ISREG(metadata.st_mode)
or metadata.st_nlink != 1
or metadata.st_mode & 0o077
or metadata.st_size > MAX_RESPONSE_BYTES
):
raise DenyAuthorityUnavailable(
"scoped LKG file identity or permissions invalid"
)
raw = handle.read(MAX_RESPONSE_BYTES + 1)
if len(raw) > MAX_RESPONSE_BYTES:
raise DenyAuthorityUnavailable("scoped LKG exceeds size policy")
bundle = parse_strict_json(raw)
except OSError as exc:
raise DenyAuthorityUnavailable("scoped LKG cannot be read") from exc
denied, payload, payload_sha256 = verify_bundle(bundle, now)
require_lkg_at_high_water(payload, payload_sha256)
return denied, payload, payload_sha256, "last_known_good"
def load_quarantine_state(
expected_federation_id: str,
expected_model_id: str,
) -> tuple[set[str], dict, str, str]:
if expected_federation_id != EXPECTED_FEDERATION_ID:
raise DenyAuthorityUnavailable("coordinator requested an unpinned federation")
if expected_model_id != EXPECTED_MODEL_ID:
raise DenyAuthorityUnavailable("coordinator requested an unpinned model")
try:
with requests.get(
AUTHORITY_URL,
params={
"authority_id": EXPECTED_AUTHORITY_ID,
"environment_id": EXPECTED_ENVIRONMENT_ID,
"federation_id": EXPECTED_FEDERATION_ID,
"model_id": EXPECTED_MODEL_ID,
},
cert=(CLIENT_CERT_PATH, CLIENT_KEY_PATH),
verify=CA_BUNDLE_PATH,
timeout=REQUEST_TIMEOUT_SECONDS,
stream=True,
allow_redirects=False,
) as response:
response.raise_for_status()
declared = response.headers.get("content-length")
if declared is not None and int(declared) > MAX_RESPONSE_BYTES:
raise DenyAuthorityUnavailable(
"quarantine response exceeds size limit"
)
raw = bytearray()
for chunk in response.iter_content(chunk_size=STREAM_CHUNK_BYTES):
raw.extend(chunk)
if len(raw) > MAX_RESPONSE_BYTES:
raise DenyAuthorityUnavailable(
"quarantine response exceeds size limit"
)
bundle = parse_strict_json(bytes(raw))
now = datetime.now(timezone.utc)
denied, payload, payload_sha256 = verify_bundle(bundle, now)
# Redis/Valkey failure intentionally propagates and pauses the round.
advance_high_water(payload, payload_sha256)
write_lkg(bundle)
return denied, payload, payload_sha256, "authority"
except (
requests.RequestException,
json.JSONDecodeError,
OSError,
ValueError,
DenyAuthorityUnavailable,
):
return load_verified_lkg(datetime.now(timezone.utc))
def require_client_allowed(
client_id: str, expected_federation_id: str, expected_model_id: str
) -> dict:
if not isinstance(client_id, str) or CLIENT_ID.fullmatch(client_id) is None:
raise PermissionError("invalid federated client identity")
denied, payload, payload_sha256, source = load_quarantine_state(
expected_federation_id=expected_federation_id,
expected_model_id=expected_model_id,
)
if client_id in denied:
raise PermissionError(f"federated client quarantined: {client_id}")
return {
"client_id": client_id,
"authority_id": payload["authority_id"],
"environment_id": payload["environment_id"],
"federation_id": payload["federation_id"],
"bundle_version": payload["version"],
"bundle_payload_sha256": payload_sha256,
"policy_version": POLICY_VERSION,
"policy_sha256": POLICY_SHA256,
"issued_at": payload["issued_at"],
"source": source,
"decision": "allow",
}
# Call before selection and again before accepting each update. State-store,
# scope, signature, freshness, or LKG errors propagate and pause the round.
Operational evidence
Emit the pinned authority/environment/federation scope, bundle version, signed-payload digest, issued time, source (authority or last_known_good), authenticated client, and decision in the signed admission receipt. Alert when LKG age crosses half of MAX_STALENESS; pause on authority, signature, scope, freshness, Redis/Valkey, or rollback-state failure.
Action: Pin the scope and authority public key outside client/model control, derive separate LKG/high-water names from that scope, and test cross-authority, cross-environment, cross-federation substitution as well as outage, cache corruption, expiry, clock skew, lower-version replay, and same-version/different-payload replay.
Use jcs>=0.2,<1 in both issuer and consumer. Refreshing signed issued_at/expires_at preserves the stable decision digest at the same version; any scope, policy, model or client-population mutation at that version is rejected. Full envelope digests remain evidence identities and every envelope still passes signature and freshness checks. On upgrade, a high-water row containing only payload_sha256 fails closed: pause admission, independently read and verify the current authoritative decision at a version no lower than the stored version, then transactionally replace the old digest field with its decision_sha256 while preserving the version fence. Never clear or expire the high-water store to recover service.
Open-source tools
- Flower
- FATE
- PySyft
- NVIDIA FLARE
- scikit-learn
- PyTorch
- TensorFlow
- SPIRE
- PyNaCl
- Valkey
- Redis 8+ AGPLv3 source build
- Substra
- SubstraFL
- FedML
Threat mappings
MITRE ATLAS
- AML.T0020 Training Data Poisoning
- AML.T0018 Manipulate AI Model (malicious FL participants directly manipulate the model via updates)
MAESTRO
- Data Poisoning (L2)
- Data Poisoning (Training Phase) (L1)
- Backdoor Attacks (L1) (FL poisoning can introduce model backdoors)
- Supply Chain Attacks (Cross-Layer) (FL participants form part of the training supply chain)
OWASP LLM Top 10 2026
- LLM05:2026 Data and Model Poisoning
OWASP ML Top 10 2023
- ML02:2023 Data Poisoning Attack
- ML10:2023 Model Poisoning
OWASP Top 10 for Agentic Applications 2026
- N/A
NIST Adversarial Machine Learning 2025
- NISTAML.013 Data Poisoning
- NISTAML.023 Backdoor Poisoning
- NISTAML.024 Targeted Poisoning
- NISTAML.011 Model Poisoning (Availability)
- NISTAML.026 Model Poisoning (Integrity) (isolating malicious FL participants prevents integrity-targeted poisoning)
- NISTAML.021 Clean-label Backdoor
Cisco Integrated AI Security and Safety Framework
- AISubtech-14.1.2 Insufficient Access Controls (authenticated federation admission enforces current client quarantine)
- AISubtech-9.2.2 Backdoors and Trojans (authoritatively quarantined federated contributors cannot keep seeding backdoors)
- AITech-6.1 Training Data Poisoning (known quarantined clients and their historical contributions are excluded from training)
Google Secure AI Framework 2.0 - Risks
- DP: Data Poisoning (isolating malicious FL participants prevents data poisoning)
Databricks AI Security Framework 3.0
- Datasets 3.1: Data poisoning (isolating malicious participants prevents training data poisoning)
- Datasets 3.3: Label flipping (isolating malicious participants prevents label flipping attacks)
- Model 7.1: Backdoor machine learning / Trojaned model (isolating malicious FL clients prevents backdoor insertion)