AID-H-012.004: Secure Reward Signal Integrity & Transport

Version 1.20261015 · Actionable sub-technique

Protect the integrity and authenticity of the reward signal itself. In many real systems, the reward is computed by an external service (simulation backend, telemetry service, rules arbiter). If an attacker can tamper with that reward in transit, the agent will learn a malicious or useless policy. This subtechnique treats the reward channel like a critical security surface: mutually authenticated transport plus cryptographic signing of the reward payload.

Pillars: model

Lifecycle: operation

Scope boundary

Owns authenticity, integrity, freshness, and service identity for externally computed reward payloads and their transport before the reward is consumed for learning.

AID-H-012.001: AID-H-012.004 protects the reward value from transport or service tampering; AID-H-012.001 determines whether the reward function itself encodes the intended objective. An authentic reward can still be badly designed, and a sound reward is useless if attackers can alter it in transit.

AID-H-004.005: AID-H-012.004 is the primary owner for an externally computed reward payload's producer identity, integrity, freshness, and learning-consumption binding; AID-H-004.005 supplies a protocol-independent signed-envelope primitive when intermediaries break transport trust. The same signed reward evidence must not earn a second PASS under the generic message control.

Implementation guidance

AID-H-012.004-G001: Secure the reward channel using mutual TLS (mTLS) and per-service identities.
AID-H-012.004-G002: Digitally sign the reward payload and verify it inside the agent before using it for learning.

AID-H-012.004-G001

Secure the reward channel using mutual TLS (mTLS) and per-service identities.

Apply this control when reward computation crosses a network boundary between the trainer/agent and a reward-oracle workload. If reward computation is in the same process and no reward transport exists, return NOT_APPLICABLE; application-layer transition binding in the sibling guidance may still apply. This control owns authenticated transport and workload identity at the reward-service boundary. AID-H-004 supplies the enterprise workload identity root; its PASS does not prove that this exact reward route enforces mTLS.

Enforce exact Istio identities and fail-closed authorization

# File: deploy/rl-system/reward-mtls.yaml
apiVersion: security.istio.io/v1
kind: PeerAuthentication
metadata:
  name: reward-oracle-strict-mtls
  namespace: rl-system
spec:
  selector:
    matchLabels:
      app.kubernetes.io/name: reward-oracle
  mtls:
    mode: STRICT
---
apiVersion: security.istio.io/v1
kind: AuthorizationPolicy
metadata:
  name: reward-oracle-only-approved-trainer
  namespace: rl-system
spec:
  selector:
    matchLabels:
      app.kubernetes.io/name: reward-oracle
  action: ALLOW
  rules:
    - from:
        - source:
            principals:
              - cluster.local/ns/rl-system/sa/rl-trainer
      to:
        - operation:
            methods: ["POST"]
            paths: ["/v1/reward"]
---
apiVersion: networking.istio.io/v1
kind: DestinationRule
metadata:
  name: reward-oracle-istio-mtls
  namespace: rl-system
spec:
  host: reward-oracle.rl-system.svc.cluster.local
  trafficPolicy:
    tls:
      mode: ISTIO_MUTUAL

Run the trainer as service account rl-trainer and the oracle as a different least-privilege service account. Admission policy must reject either Deployment if its service account, namespace, mesh injection/ambient enrollment, container image digest, or required label differs from the signed release manifest. Keep the reward endpoint off public ingress; route the client only to the exact service DNS name and require the application-layer signed transition contract from the sibling guidance.

Verification, evidence, and outcomes

Before a training run, wait for the three resources to report accepted/reconciled state, then capture their canonical specs and resource UIDs, the signed release manifest, trust-domain and root-bundle identifiers, exact trainer/oracle service accounts and image digests, proxy/workload enrollment readback, and reward-route access logs containing source principal, destination workload, route, response, and mTLS indicator. Independently test: the approved trainer succeeds; a different service account is denied; a plaintext client is denied; a wrong path/method is denied; and an expired or untrusted workload certificate is denied. Re-read the effective proxy/waypoint configuration rather than trusting submitted YAML.

Action: Bind the exact trainer and reward-oracle workloads to separate identities, enforce strict mTLS plus principal/path authorization, and block training until independent effective-config readback and negative identity tests return PASS.

AID-H-012.004-G002

Digitally sign the reward payload and verify it inside the agent before using it for learning.

Apply this control whenever a reward crosses a process, queue, storage, or network boundary before policy update. Use an asymmetric Ed25519 JWS so the trainer holds only verification keys and cannot mint oracle rewards; a shared HMAC key would let every verifier forge producer messages. If the reward is computed and consumed atomically inside one measured process with no serialization boundary, return NOT_APPLICABLE. Transport mTLS from the sibling guidance is defense in depth and does not prove transition binding.

Issue and verify an exact short-lived reward token

# File: rl_comms/signed_reward.py
from __future__ import annotations

import hashlib
import math
import re
import time
from collections.abc import Callable, Mapping
from typing import Any

import jwt
from jwt import InvalidTokenError


HEX64 = re.compile(r"^[0-9a-f]{64}$")
REQUIRED_CLAIMS = {
    "iss",
    "aud",
    "iat",
    "exp",
    "jti",
    "schema_version",
    "episode_id",
    "transition_id",
    "state_before_sha256",
    "state_after_sha256",
    "reward_policy_sha256",
    "reward",
}


class RewardDenied(RuntimeError):
    pass


def issue_reward_token(
    *,
    private_key_pem: bytes,
    key_id: str,
    issuer: str,
    audience: str,
    episode_id: str,
    transition_id: str,
    state_before_sha256: str,
    state_after_sha256: str,
    reward_policy_sha256: str,
    reward: float,
    ttl_seconds: int,
    maximum_ttl_seconds: int,
) -> str:
    if not all([key_id, issuer, audience, episode_id, transition_id]):
        raise ValueError("reward identity fields are required")
    if not all(
        HEX64.fullmatch(value)
        for value in [state_before_sha256, state_after_sha256, reward_policy_sha256]
    ):
        raise ValueError("state and reward-policy digests must be lowercase SHA-256")
    if isinstance(reward, bool) or not isinstance(reward, (int, float)) or not math.isfinite(reward):
        raise ValueError("reward must be finite")
    if (
        isinstance(ttl_seconds, bool)
        or isinstance(maximum_ttl_seconds, bool)
        or not isinstance(ttl_seconds, int)
        or not isinstance(maximum_ttl_seconds, int)
        or not 0 < ttl_seconds <= maximum_ttl_seconds
    ):
        raise ValueError("token TTL exceeds the signature-verified policy ceiling")

    now = int(time.time())
    transition_key = f"{episode_id}:{transition_id}"
    claims = {
        "iss": issuer,
        "aud": audience,
        "iat": now,
        "exp": now + ttl_seconds,
        "jti": hashlib.sha256(transition_key.encode()).hexdigest(),
        "schema_version": "aidefend.signed-reward.v1",
        "episode_id": episode_id,
        "transition_id": transition_id,
        "state_before_sha256": state_before_sha256,
        "state_after_sha256": state_after_sha256,
        "reward_policy_sha256": reward_policy_sha256,
        "reward": float(reward),
    }
    return jwt.encode(
        claims,
        private_key_pem,
        algorithm="EdDSA",
        headers={"kid": key_id, "typ": "JWT"},
    )


def verify_reward_token(
    *,
    token: str,
    public_keys_by_id: Mapping[str, bytes],
    expected_issuer: str,
    expected_audience: str,
    expected_episode_id: str,
    expected_transition_id: str,
    expected_state_before_sha256: str,
    expected_state_after_sha256: str,
    expected_reward_policy_sha256: str,
    leeway_seconds: int,
    maximum_leeway_seconds: int,
    maximum_token_ttl_seconds: int,
    claim_once: Callable[[str, int], bool],
) -> tuple[float, dict[str, Any]]:
    if (
        not token
        or isinstance(leeway_seconds, bool)
        or isinstance(maximum_leeway_seconds, bool)
        or isinstance(maximum_token_ttl_seconds, bool)
        or not isinstance(leeway_seconds, int)
        or not isinstance(maximum_leeway_seconds, int)
        or not isinstance(maximum_token_ttl_seconds, int)
        or not 0 <= leeway_seconds <= maximum_leeway_seconds
        or maximum_token_ttl_seconds <= 0
    ):
        raise RewardDenied("token or signed-policy leeway is invalid")
    try:
        header = jwt.get_unverified_header(token)
    except InvalidTokenError as exc:
        raise RewardDenied("malformed JWS header") from exc
    if set(header) != {"alg", "kid", "typ"}:
        raise RewardDenied("unexpected protected header fields")
    if header["alg"] != "EdDSA" or header["typ"] != "JWT":
        raise RewardDenied("algorithm or token type is not allowed")
    public_key = public_keys_by_id.get(header["kid"])
    if public_key is None:
        raise RewardDenied("signing key ID is not in the pinned active registry")

    try:
        claims = jwt.decode(
            token,
            public_key,
            algorithms=["EdDSA"],
            issuer=expected_issuer,
            audience=expected_audience,
            leeway=leeway_seconds,
            options={
                "require": ["iss", "aud", "iat", "exp", "jti"],
                "verify_signature": True,
                "verify_exp": True,
                "verify_iat": True,
                "verify_iss": True,
                "verify_aud": True,
            },
        )
    except InvalidTokenError as exc:
        raise RewardDenied("JWS signature or registered claim validation failed") from exc
    if set(claims) != REQUIRED_CLAIMS:
        raise RewardDenied("reward claim schema is not exact")
    if claims["schema_version"] != "aidefend.signed-reward.v1":
        raise RewardDenied("unsupported reward schema")
    if (
        isinstance(claims["iat"], bool)
        or isinstance(claims["exp"], bool)
        or not isinstance(claims["iat"], int)
        or not isinstance(claims["exp"], int)
        or not 0 < claims["exp"] - claims["iat"] <= maximum_token_ttl_seconds
    ):
        raise RewardDenied("reward lifetime exceeds the signature-verified policy ceiling")
    expected = {
        "episode_id": expected_episode_id,
        "transition_id": expected_transition_id,
        "state_before_sha256": expected_state_before_sha256,
        "state_after_sha256": expected_state_after_sha256,
        "reward_policy_sha256": expected_reward_policy_sha256,
    }
    if any(claims[name] != value for name, value in expected.items()):
        raise RewardDenied("reward does not bind the live transition and policy")
    if not all(
        HEX64.fullmatch(claims[name])
        for name in [
            "state_before_sha256",
            "state_after_sha256",
            "reward_policy_sha256",
            "jti",
        ]
    ):
        raise RewardDenied("reward digest claim is malformed")
    reward = claims["reward"]
    if isinstance(reward, bool) or not isinstance(reward, (int, float)) or not math.isfinite(reward):
        raise RewardDenied("reward is non-finite")
    if not claim_once(claims["jti"], int(claims["exp"])):
        raise RewardDenied("reward transition was already claimed")

    receipt = {
        "schema_version": "aidefend.reward-verification.v1",
        "status": "accepted",
        "key_id": header["kid"],
        "jti": claims["jti"],
        "episode_id": claims["episode_id"],
        "transition_id": claims["transition_id"],
        "reward_policy_sha256": claims["reward_policy_sha256"],
        "token_sha256": hashlib.sha256(token.encode()).hexdigest(),
        "expires_at": int(claims["exp"]),
    }
    return float(reward), receipt

Atomicity, verification, evidence, and outcomes

Load the oracle private key from a non-exportable signing service or tightly scoped secret manager and distribute only pinned public keys plus activation/revocation windows to trainers. Implement claim_once as a durable atomic unique insert keyed by jti with expiry, never an in-process set. Commit that claim and the accepted reward receipt before the corresponding transition enters the training buffer; on any denial, do not update the policy. Retain the signed key registry and verification receipt, oracle/trainer image digests, token bytes, live state and transition digests, atomic-claim result, buffer readback, and key-rotation/revocation tests.

Action: Sign every reward with the oracle's Ed25519 key, bind it to the exact episode, transition, states, reward policy, audience, and lifetime, atomically reject replay, and block training unless independent negative tests and buffer readback return PASS.

Open-source tools

Istio
PyJWT
pyca/cryptography
Cosign
SPIRE (workload identity issuance; mTLS dataplane required)

Threat mappings

MITRE ATLAS

AML.T0020 Training Data Poisoning (authenticated reward signals prevent one training-feedback tampering path)

MAESTRO

Data Tampering (L2) (reward signal tampering in transit or storage)

OWASP LLM Top 10 2026

LLM05:2026 Data and Model Poisoning

OWASP ML Top 10 2023

ML08:2023 Model Skewing

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

NISTAML.013 Data Poisoning (reward signals are integrity-protected training inputs)
NISTAML.037 Training Data Attacks (covers RLHF feedback poisoning)

Cisco Integrated AI Security and Safety Framework

AISubtech-14.1.2 Insufficient Access Controls (Authenticates and authorizes the exact trainer-to-reward-oracle service path)
AISubtech-4.3.4 Replay Exploitation (Rejects replayed reward messages; agentic communication is not assumed)
AISubtech-6.1.3 Reinforcement Signal Corruption
AITech-16.1 Eavesdropping
AITech-6.1 Training Data Poisoning

Google Secure AI Framework 2.0 - Risks

DP: Data Poisoning (reward signals are integrity-protected training inputs)

Databricks AI Security Framework 3.0

Datasets 3.1: Data poisoning (authenticated reward-feedback transport)