AID-H-008.002: Byzantine-Robust Aggregation Rules

Version 1.20261015 · Actionable sub-technique

A class of statistical, non-cryptographic aggregation methods designed to protect the integrity of the global model in Federated Learning. These rules identify and mitigate the impact of outlier or malicious model updates from compromised clients (Byzantine actors) by using functions like median, trimmed mean, or distance-based scoring (e.g., Krum) to filter out or down-weight anomalous contributions before they can corrupt the final aggregated model.

Pillars: model

Lifecycle: building, operation

Scope boundary

Owns statistical filtering, down-weighting, or exclusion of anomalous and Byzantine federated updates before they corrupt the global model. It does not provide individual-contribution confidentiality.

AID-H-008.001: AID-H-008.002 controls malicious-update influence using per-client statistics or equivalent proofs; AID-H-008.001 hides each client's update under secure aggregation. Deploy both only through an explicitly compatible privacy-preserving robust protocol or trusted aggregation boundary.

AID-H-008.003: AID-H-008.002 limits the influence of accepted updates even from authorized but malicious clients; AID-H-008.003 decides whether the identity and exact update may enter the round. Authorized participation is not proof of benign contribution.

Implementation guidance

AID-H-008.002-G001

Replace standard Federated Averaging with robust statistical aggregators like Krum, Multi-Krum, or Trimmed Mean.
Concept:

Naive Federated Averaging (simple mean of all client updates) is fragile. A single malicious client can submit an extreme update to poison the global model. Byzantine-robust aggregators like Krum defend against this by down-weighting or outright ignoring outliers. Krum, for example, chooses the update that is most similar to the majority of other updates, assuming attackers are statistical outliers.

Implement the Krum Aggregation Function

The server computes pairwise distances between client updates, scores each update by how close it is to its nearest neighbors, and then selects the update with the lowest score. That chosen update becomes the global update for that round.

# File: hardening/robust_aggregators.py
import numpy as np

def krum_aggregator(client_updates, num_malicious):
    """Select one update only when the Krum Byzantine assumptions hold."""
    if isinstance(num_malicious, bool) or not isinstance(num_malicious, int) or num_malicious < 0:
        raise ValueError("num_malicious must be a non-negative integer")
    num_clients = len(client_updates)
    if num_clients < 2 * num_malicious + 3:
        raise ValueError("Krum requires n >= 2f + 3 authenticated clients")

    updates = [np.asarray(update, dtype=np.float64) for update in client_updates]
    reference_shape = updates[0].shape
    if not reference_shape or any(update.shape != reference_shape for update in updates):
        raise ValueError("client updates must have one identical non-empty shape")
    if any(not np.isfinite(update).all() for update in updates):
        raise ValueError("client updates contain NaN or infinity")

    stacked = np.stack(updates)
    deltas = stacked[:, None, ...] - stacked[None, :, ...]
    distances = np.sum(deltas * deltas, axis=tuple(range(2, deltas.ndim)))
    np.fill_diagonal(distances, np.inf)

    neighbor_count = num_clients - num_malicious - 2
    scores = np.array([
        np.partition(row, neighbor_count - 1)[:neighbor_count].sum()
        for row in distances
    ])
    if not np.isfinite(scores).all():
        raise ValueError("Krum scores are non-finite")
    best_client_index = int(np.argmin(scores))
    return updates[best_client_index].copy()

# --- Usage on FL server ---
# aggregated_update = krum_aggregator(received_updates, num_malicious=3)

Action: In any environment where you assume only a minority of clients are malicious, replace plain averaging with Krum, Multi-Krum, trimmed mean, or similar. You must define an upper bound on the number of malicious clients (f) to tune these defenses.

Keep the action safely bounded

Validate dimensionality, finite/bounded values and the algorithm's participant/Byzantine assumptions; calibrate with signed benign/attack simulations, run the pinned implementation over the complete cohort, bind selected/rejected updates and aggregate digest, and deny publication when assumptions or utility/security gates fail.

Verify safely

A separately credentialed verifier independently reads back or replays each round's complete authenticated update population and signed Krum, Multi-Krum, coordinate-trimmed-mean or other reviewed robust rule, exercises positive, negative, boundary and dependency-failure fixtures, and reconciles the complete eligible population.

AID-H-008.002-G002

Monitor the statistical properties of each client's updates over time to detect consistently anomalous actors.
Concept:

A one-time weird update might be noise. A client that repeatedly submits high-deviation or low-similarity updates across many rounds is more suspicious. Track basic stats (norm, cosine similarity to the global model) per client per round, then flag clients whose behavior is consistently out-of-family. Those clients can be down-weighted or blocked in future rounds.

Step 1: Persist per-client statistics in a server-side store

The server should write one row per client per round so the evidence survives process restarts and can be audited later. The example below uses SQLite for a simple, deterministic evidence trail; in production you can move the same schema to Postgres or your experiment metadata store.

# File: hardening/client_monitoring.py
from __future__ import annotations
import math
import os

import sqlite3
from pathlib import Path

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

def _required_positive_float(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive number")
    return value

UPDATE_MIN_NORM = _required_positive_float("VERIFIED_H008_UPDATE_MIN_NORM")

def _required_unit_interval(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or not 0 < value <= 1:
        raise RuntimeError(f"{name} must be in the interval (0, 1]")
    return value

UPDATE_MAX_SIMILARITY = _required_unit_interval("VERIFIED_H008_UPDATE_MAX_SIMILARITY")



DB_PATH = Path("state/fl_client_stats.db")


def init_stats_store(db_path: Path = DB_PATH) -> sqlite3.Connection:
    db_path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(db_path)
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS fl_client_stats (
            client_id TEXT NOT NULL,
            round_id INTEGER NOT NULL,
            update_norm REAL NOT NULL,
            similarity REAL NOT NULL,
            recorded_at DATETIME DEFAULT CURRENT_TIMESTAMP,
            PRIMARY KEY (client_id, round_id)
        )
        """
    )
    conn.commit()
    return conn


def log_update_stats(
    client_id: str,
    update_vector: np.ndarray,
    last_global_update_vector: np.ndarray,
    conn: sqlite3.Connection,
    round_id: int,
) -> dict:
    if not isinstance(client_id, str) or not client_id or type(round_id) is not int or round_id < 0:
        raise ValueError("client and round identity are invalid")
    update_vec = np.asarray(update_vector, dtype=np.float64).reshape(1, -1)
    baseline_vec = np.asarray(last_global_update_vector, dtype=np.float64).reshape(1, -1)
    if (update_vec.shape != baseline_vec.shape or update_vec.size == 0 or
            not np.isfinite(update_vec).all() or not np.isfinite(baseline_vec).all()):
        raise ValueError("update and baseline must be aligned finite vectors")

    update_norm = float(np.linalg.norm(update_vec))
    similarity = float(cosine_similarity(update_vec, baseline_vec)[0, 0])

    conn.execute(
        """
        INSERT INTO fl_client_stats (client_id, round_id, update_norm, similarity)
        VALUES (?, ?, ?, ?)
        """,
        (client_id, round_id, update_norm, similarity),
    )
    conn.commit()

    return {
        "client_id": client_id,
        "round_id": round_id,
        "norm": update_norm,
        "similarity": similarity,
    }


def get_recent_outliers(
    conn: sqlite3.Connection,
    *,
    min_norm: float,
    max_similarity: float,
) -> list[tuple]:
    cursor = conn.execute(
        """
        SELECT client_id, round_id, update_norm, similarity, recorded_at
        FROM fl_client_stats
        WHERE update_norm >= ? OR similarity <= ?
        ORDER BY round_id DESC, client_id ASC
        """,
        (min_norm, max_similarity),
    )
    return cursor.fetchall()


# Example usage on the FL server:
# conn = init_stats_store()
# record = log_update_stats("client-123", local_update, global_update, conn, round_id=42)
# suspicious_rows = get_recent_outliers(conn, min_norm=UPDATE_MIN_NORM, max_similarity=UPDATE_MAX_SIMILARITY)
Step 2: Review multi-round outliers before quarantine or exclusion

Do not key off one noisy round. Query the stored history and require repeated anomalies before you down-weight or remove a client from aggregation.

Action: Persist per-client norm and similarity every round, review the multi-round history, and use that durable evidence to support quarantine, reputation downgrade, or exclusion decisions.

Keep the action safely bounded

Evaluate all updates before aggregation, record score/reason and model version, quarantine findings for bounded review or secondary checks, and deny inclusion on confirmed policy breach or missing/error evidence; never exclude solely from an uncalibrated detector.

Verify safely

A separately credentialed verifier independently reads back or replays each authenticated per-client update under a signed instantaneous anomaly model, feature schema and calibrated threshold, exercises positive, negative, boundary and dependency-failure fixtures, and reconciles the complete eligible population.

AID-H-008.002-G003

Implement client reputation / trust scoring and use it to weight or exclude updates.
Concept:

Not all clients deserve equal influence. The FL server can maintain a reputation score for each client based on historical contribution quality (e.g. how aligned their updates were with the eventual global direction). During aggregation, weight high-reputation clients more, and either heavily down-weight or fully ignore low-reputation / repeatedly suspicious clients.

Implement Reputation-Based Weighted Aggregation

Below is a simple pattern: (1) maintain a reputation score per client, (2) compute a weighted average of updates using these scores, (3) update each client's reputation after each round based on how well they agreed with the final global update.

# File: hardening/reputation_aggregator.py
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

# Example server-side state:
# client_reputations = {
#   "client_123": 1.0,
#   "client_777": 0.6
# }

def reputation_weighted_aggregation(client_updates, client_ids, reputations):
    """
    Weighted average of client updates using their reputation scores.
    client_updates: list[np.array]
    client_ids: list[str]
    reputations: dict[str, float]
    """
    if not client_updates or len(client_updates) != len(client_ids):
        raise ValueError("client updates and IDs must be non-empty and aligned")
    if (len(client_ids) != len(set(client_ids)) or
            any(not isinstance(cid, str) or not cid for cid in client_ids)):
        raise ValueError("client IDs must be unique non-empty strings")
    missing = [cid for cid in client_ids if cid not in reputations]
    if missing:
        raise ValueError("reputation missing for admitted client")
    updates = np.stack([np.asarray(update, dtype=np.float64) for update in client_updates])
    if not np.all(np.isfinite(updates)):
        raise ValueError("client update contains a non-finite value")
    weights = np.array(
        [float(reputations[cid]) for cid in client_ids],
        dtype=np.float64,
    )
    if not np.all(np.isfinite(weights)) or np.any(weights < 0.0) or np.any(weights > 1.0):
        raise ValueError("reputation weights must be finite and in [0, 1]")
    weight_sum = float(weights.sum())
    if not np.isfinite(weight_sum) or weight_sum <= 0.0:
        raise ValueError("no client has a positive finite reputation weight")
    normalized_weights = weights / weight_sum

    # Weighted linear combination of updates
    weighted_avg = np.tensordot(normalized_weights, updates, axes=(0, 0))
    return weighted_avg

def update_reputations(global_update, client_updates, client_ids, reputations, learning_rate: float):
    """
    After producing the new global update, measure each client's 'quality'
    (e.g. cosine similarity to the final global update). Then adjust the
    client's reputation with a moving average.
    """
    if isinstance(learning_rate, bool) or not isinstance(learning_rate, (int, float)) or not np.isfinite(learning_rate) or not 0.0 <= learning_rate <= 1.0:
        raise ValueError("lr must be between zero and one")
    for i, cid in enumerate(client_ids):
        similarity = cosine_similarity(
            client_updates[i].reshape(1, -1),
            global_update.reshape(1, -1)
        )[0, 0]

        # Cosine similarity is [-1, 1], but a reputation weight must remain
        # non-negative. Treat anti-alignment as zero contribution quality.
        quality = max(0.0, min(1.0, float(similarity)))
        if cid not in reputations or isinstance(reputations[cid], bool):
            raise ValueError("reputation missing or invalid for admitted client")
        old_rep = float(reputations[cid])
        if not np.isfinite(old_rep) or not 0.0 <= old_rep <= 1.0:
            raise ValueError("stored reputation is outside [0, 1]")
        new_rep = (1 - learning_rate) * old_rep + learning_rate * quality
        reputations[cid] = max(0.0, min(1.0, new_rep))

    return reputations

Action: Maintain a per-client reputation score and use it as a weighting factor in aggregation. Over time, clients that repeatedly submit anomalous or low-alignment updates lose influence. This naturally rate-limits persistent Byzantine clients without requiring an immediate hard ban.

Verify safely

A separately credentialed verifier independently reads back or replays each durable cryptographic participant identity and its complete signed longitudinal event history across rounds, exercises positive, negative, boundary and dependency-failure fixtures, and reconciles the complete eligible population.

Open-source tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-H-008.002: Byzantine-Robust Aggregation Rules | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service