RFC-0014: ALPE Anti-Replay Hardening — nonce + iat in the JWE Protected Header

• Status: Implemented
• Author(s): CalyXx
• Created: 2026-09-15
• Updated: 2026-09-21

---

1. Summary

This RFC specifies the design and implementation of replay-attack prevention for Framework M's Application-Layer Payload Encryption (ALPE) subsystem. When payload encryption is active (selective, metadata, or strict modes), every JWE envelope produced by the frontend must include two additional claims in the JWE Protected Header:

1. iat — Unix epoch timestamp (seconds) recording when the envelope was created.
2. nonce — RFC 4122 v4 UUID that uniquely identifies the envelope.

Because the protected header is base64url-encoded and fed into AES-GCM as Additional Authenticated Data (AAD), any in-transit modification to iat or nonce invalidates the 16-byte GCM authentication tag and causes decryption to fail before the server even evaluates the anti-replay logic.

On the server side, PayloadDecryptionMiddleware is extended with a two-step verification pipeline:

1. Clock-skew check — Reject envelopes whose iat deviates more than max_clock_skew_seconds (default 60 s) from the server's current time, without touching storage.
2. Nonce deduplication — Atomically acquire the nonce in a short-lived store (Redis or PostgreSQL). Reject duplicate nonces; prune entries after their TTL.

The feature is zero-config by default: enforce_anti_replay is false when mode is disabled or selective, and is automatically enabled (but still overridable) when mode = "strict".

---

2. Motivation

2.1 The Replay-Attack Window in Zero-Trust Deployments

In strict zero-trust deployments, ALPE encrypts every mutating HTTP request body end-to-end. Without replay protection, a captured JWE token remains valid indefinitely:

Attacker captures: POST /api/v1/auth/login
  body = { protected: "...", encrypted_key: "...", ... }

Replay 60 minutes later → Server decrypts identical payload → Login succeeds

Intermediate proxies, APM log collectors, and WAFs that operate in man-in-the-middle positions (common in enterprise Kubernetes clusters) can log the raw encrypted body. An insider with access to proxy logs could replay a captured login JWE envelope hours later using any HTTP client.

2.2 Why iat + nonce Together?

Using only a random nonce without an expiry timestamp forces the server to store every nonce forever — causing unbounded storage bloat. Using only an iat timestamp guards against stale replays but not replays within the valid window.

Pairing iat with nonce and using iat as the TTL anchor solves both:

Client sends JWE with { iat: 1726225000, nonce: "uuid-1234" } in Protected Header
                                   │
                                   ▼
          ┌──────────────────────────────────────────────────┐
          │ Step 1: Clock Skew Check                         │
          │ Is |Server Time - iat| <= 60 seconds?            │
          └────────┬────────────────────────────────┬────────┘
                   │ YES                            │ NO (older or future)
                   ▼                                ▼
          ┌───────────────────────────┐    ┌───────────────────────────┐
          │ Step 2: Nonce Check       │    │ ❌ Reject immediately     │
          │ Has "uuid-1234" been seen │    │ (zero storage lookup)     │
          │ in the last 60 seconds?   │    └───────────────────────────┘
          └────────┬──────────────────┘
                   │ NO (first time seen)
                   ▼
          ┌───────────────────────────────────┐
          │ ✅ Store nonce with TTL = 60 s    │
          │ → Forward decrypted body to route │
          └───────────────────────────────────┘

Result: The nonce store holds entries only for the 60-second sliding window, completely eliminating storage bloat while making replay within that window cryptographically impossible.

2.3 Alignment with Framework M's Zero-Cliff Philosophy

Deployment Mode	enforce_anti_replay default	Why
disabled	false	No encryption → no replay surface. Standard TLS + session cookies suffice.
selective	false	Auth routes rely on single-use server tokens (OTP, magic links) already.
metadata	false (operator opt-in)	Field-level PII encryption; operator chooses whether to add replay protection.
strict	true (auto-enabled)	Full zero-trust posture. Insider replay via proxy logs is a genuine threat.

The default configuration requires zero additional infrastructure for indie and SaaS deployments. Strict-mode deployments, which already configure Redis for caching, get anti-replay hardening automatically.

---

3. High-Level Architecture

┌───────────────────────────────────────────────────────────────────────────────────┐
│                        FRONTEND  (libs/framework-m-desk)                          │
│                                                                                   │
│  crypto.ts :: encryptPayload()                                                    │
│                                                                                   │
│  const header = buildJweHeader();   // called fresh on every envelope             │
│  // {                                                                             │
│  //   alg:   "RSA-OAEP-256",                                                      │
│  //   enc:   "A256GCM",                                                           │
│  //   cty:   "application/json",                                                  │
│  //   iat:   Math.floor(Date.now() / 1000),  <- Unix seconds                     │
│  //   nonce: crypto.randomUUID(),            <- RFC 4122 v4 UUID                 │
│  // }                                                                             │
│                                                                                   │
│  protected_header = base64url(JSON.stringify(header))                            │
│  [protected header is AAD in AES-GCM; tamper invalidates the 16-byte auth tag]  │
└────────────────────────────────────────┬──────────────────────────────────────────┘
                                         │  POST /api/v1/resource
                                         │  Content-Type: application/jose
                                         │  Body: { protected, encrypted_key, iv,
                                         │          ciphertext, tag }
                                         ▼
┌───────────────────────────────────────────────────────────────────────────────────┐
│                  ASGI MIDDLEWARE STACK  (libs/framework-m-standard)               │
│                                                                                   │
│  PayloadDecryptionMiddleware (priority=15)                                        │
│                                                                                   │
│  1. Decrypt JWE envelope → obtain plaintext + decode protected header             │
│  2. Clock-skew check:  |now() - iat| <= max_clock_skew_seconds                   │
│     → 400 "Payload timestamp expired or invalid"                                  │
│  3. Nonce deduplication via NonceStoreProtocol                                    │
│     → 400 "Replay attack detected: duplicate nonce"                               │
│  4. Rewrite Content-Type → application/json, forward plaintext body              │
└──────────────┬──────────────────────────────────────┬─────────────────────────────┘
               │                                      │
               ▼ (REDIS_URL configured)               ▼ (Indie / no Redis)
┌──────────────────────────┐           ┌──────────────────────────────────────────┐
│  RedisNonceAdapter        │           │  DatabaseNonceAdapter                    │
│                           │           │                                          │
│  SET nonce:alpe:<uuid> 1  │           │  INSERT INTO alpe_nonces (nonce,         │
│  NX EX <clock_skew_secs>  │           │    expires_at)                           │
│                           │           │  ON CONFLICT (nonce) DO NOTHING          │
│  O(1), zero DB overhead   │           │  + opportunistic expiry pruning          │
└──────────────────────────┘           └──────────────────────────────────────────┘

---

4. Detailed Design

4.1 Frontend — libs/framework-m-desk

4.1.1 crypto.ts — JWE Header Extension

The existing static JWE_HEADER constant in crypto.ts (libs/framework-m-desk/src/crypto.ts) must be replaced with a per-call factory function that stamps a fresh iat and nonce for every envelope:

// BEFORE (static header — replay-vulnerable)
const JWE_HEADER = { alg: "RSA-OAEP-256", enc: "A256GCM", cty: "application/json" };

// AFTER (per-call factory — replay-proof)
function buildJweHeader(): Record<string, unknown> {
  return {
    alg:   "RSA-OAEP-256",
    enc:   "A256GCM",
    cty:   "application/json",
    iat:   Math.floor(Date.now() / 1000),
    nonce: crypto.randomUUID(),
  };
}

The encryptPayload() function is updated to:

1. Call buildJweHeader() on every invocation.
2. Pass encodedHeader as additionalData in crypto.subtle.encrypt() so the GCM tag covers the protected header (AAD alignment, see §4.4).

export async function encryptPayload(
  body: string,
  url: string,
  method = "POST",
  schema?: JSONSchema
): Promise<EncryptedPayload> {
  // ... mode/route decision logic unchanged ...

  const header = buildJweHeader();
  const encodedHeader = base64UrlEncode(
    toArrayBuffer(new TextEncoder().encode(JSON.stringify(header)))
  );

  const aad = new TextEncoder().encode(encodedHeader);

  const encryptedBuffer = await crypto.subtle.encrypt(
    { name: "AES-GCM", iv, additionalData: aad },  // <-- AAD added
    aesKey,
    plaintext
  );

  const encryptedBytes = new Uint8Array(encryptedBuffer);
  const ciphertext = encryptedBytes.slice(0, -16);
  const tag = encryptedBytes.slice(-16);

  const envelope = {
    protected:     encodedHeader,
    encrypted_key: base64UrlEncode(wrappedKey),
    iv:            base64UrlEncode(toArrayBuffer(iv)),
    ciphertext:    base64UrlEncode(toArrayBuffer(ciphertext)),
    tag:           base64UrlEncode(toArrayBuffer(tag)),
  };

  return { body: JSON.stringify(envelope), contentType: "application/jose" };
}

4.1.2 New Unit Tests — __tests__/crypto.test.ts

• iat is present in the decoded protected header and within ±5 s of Date.now() / 1000.
• nonce is a valid RFC 4122 v4 UUID string.
• Two sequential encryptPayload() calls produce different iat/nonce values (no accidental caching of the header).

---

4.2 Core Protocol — libs/framework-m-core

4.2.1 NonceStoreProtocol

New file: framework_m_core/interfaces/nonce.py

"""Nonce Store Protocol (Port).

Defines the interface that nonce store adapters must implement for
ALPE (Application-Layer Payload Encryption) anti-replay protection.
"""

from __future__ import annotations

from typing import Protocol, runtime_checkable


@runtime_checkable
class NonceStoreProtocol(Protocol):
    """Port for atomic nonce acquisition and deduplication.

    Implementations guarantee that acquire() is atomic and idempotent:
    - Returns True  on the FIRST call for a given nonce (nonce registered).
    - Returns False on ANY subsequent call within the TTL window (replay).

    Implementations MUST auto-expire entries after `ttl_seconds`.
    """

    async def acquire(self, nonce: str, ttl_seconds: int) -> bool:
        """Attempt to atomically register a nonce.

        Args:
            nonce: A unique identifier string (RFC 4122 v4 UUID recommended).
            ttl_seconds: Lifetime in seconds before the nonce record expires.

        Returns:
            True  if the nonce was newly registered (first time seen).
            False if the nonce already exists (replay detected).
        """
        ...

    async def close(self) -> None:
        """Release any resources held by the adapter (connections, etc.)."""
        ...


__all__ = ["NonceStoreProtocol"]

---

4.3 Standard Adapters — libs/framework-m-standard

4.3.1 RedisNonceAdapter

New file: framework_m_standard/adapters/crypto/nonce/redis.py

Uses an atomic Redis SET … NX EX command — a single network round-trip that is both the existence check and the registration:

"""Redis-backed Nonce Store Adapter for ALPE Anti-Replay Protection."""

from __future__ import annotations

from typing import Any

from framework_m_core.interfaces.nonce import NonceStoreProtocol

NONCE_KEY_PREFIX = "nonce:alpe:"


class RedisNonceAdapter:
    """Redis adapter implementing NonceStoreProtocol.

    Uses SET <key> 1 NX EX <ttl> for atomic check-and-set:
    - NX  ensures the SET is a no-op if the key already exists.
    - EX  attaches a TTL so Redis auto-expires the entry after the window.

    This adapter requires zero database overhead and O(1) latency.

    Args:
        redis_client: An async Redis client (e.g. redis.asyncio.Redis).
    """

    def __init__(self, redis_client: Any) -> None:
        self._redis = redis_client

    async def acquire(self, nonce: str, ttl_seconds: int) -> bool:
        """Atomically register nonce. Returns True on first acquisition."""
        key = f"{NONCE_KEY_PREFIX}{nonce}"
        result = await self._redis.set(key, "1", nx=True, ex=ttl_seconds)
        # redis-py returns True on successful SET NX, None if key existed
        return bool(result)

    async def close(self) -> None:
        """Close the underlying Redis connection."""
        await self._redis.aclose()


# Runtime protocol check
assert isinstance(RedisNonceAdapter.__new__(RedisNonceAdapter), NonceStoreProtocol)

__all__ = ["RedisNonceAdapter"]

4.3.2 DatabaseNonceAdapter

New file: framework_m_standard/adapters/crypto/nonce/database.py

Uses INSERT … ON CONFLICT DO NOTHING. The rowcount discriminates first-use from replay:

"""Database-backed Nonce Store Adapter for ALPE Anti-Replay Protection.

Architectural Note:
This adapter uses SQLAlchemy Core directly, bypassing ORM overhead.
It mirrors the pattern established by DatabaseIdempotencyAdapter for
consistency across system-level middleware components.
"""

from __future__ import annotations

import logging
import random
from collections.abc import Callable
from datetime import UTC, datetime, timedelta

import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession

from framework_m_core.interfaces.nonce import NonceStoreProtocol

logger = logging.getLogger(__name__)

_CLEANUP_PROBABILITY = 0.02  # 2% chance of opportunistic cleanup per write


class DatabaseNonceAdapter:
    """PostgreSQL / SQLite nonce store adapter implementing NonceStoreProtocol.

    Uses INSERT ... ON CONFLICT (nonce) DO NOTHING to atomically register
    nonces. rowcount == 1 means first-use (True); 0 means replay (False).

    Opportunistic cleanup sweeps expired rows on ~2% of writes, avoiding the
    need for a dedicated pruning background job.

    Args:
        session_factory: Callable that returns an AsyncSession context manager.
        table: Optional pre-resolved SQLAlchemy Table. If omitted, resolved
               lazily from TableRegistry on first use.
    """

    def __init__(
        self,
        session_factory: Callable[[], AsyncSession],
        table: sa.Table | None = None,
    ) -> None:
        self._session_factory = session_factory
        self._table: sa.Table | None = table

    def _get_table(self) -> sa.Table:
        if self._table is None:
            from framework_m_standard.adapters.db.table_registry import TableRegistry

            self._table = TableRegistry().get_table("AlpeNonce")
        return self._table

    async def acquire(self, nonce: str, ttl_seconds: int) -> bool:
        """Atomically register nonce. Returns True on first acquisition."""
        table = self._get_table()
        now = datetime.now(UTC)
        expires_at = now + timedelta(seconds=ttl_seconds)

        async with self._session_factory() as session:
            try:
                bind = session.get_bind()
                dialect_name = getattr(getattr(bind, "dialect", None), "name", "sqlite")

                if dialect_name == "postgresql":
                    from sqlalchemy.dialects.postgresql import insert as pg_insert

                    stmt = (
                        pg_insert(table)
                        .values(nonce=nonce, created_at=now, expires_at=expires_at)
                        .on_conflict_do_nothing(index_elements=["nonce"])
                    )
                else:
                    # SQLite: INSERT OR IGNORE
                    stmt = (
                        sa.insert(table)
                        .prefix_with("OR IGNORE")
                        .values(nonce=nonce, created_at=now, expires_at=expires_at)
                    )

                result = await session.execute(stmt)
                await session.commit()
                acquired = (result.rowcount or 0) > 0

            except Exception:
                logger.exception("Nonce acquisition failed for nonce=%s", nonce)
                # Fail open: transient DB errors should not block all traffic
                return True

            # Opportunistic pruning (~2% of writes)
            if random.random() < _CLEANUP_PROBABILITY:
                try:
                    del_stmt = sa.delete(table).where(table.c.expires_at < now)
                    await session.execute(del_stmt)
                    await session.commit()
                except Exception:
                    logger.debug("Opportunistic nonce pruning failed (non-critical)")

        return acquired

    async def close(self) -> None:
        """No persistent resources to release."""
        pass


assert isinstance(
    DatabaseNonceAdapter.__new__(DatabaseNonceAdapter), NonceStoreProtocol
)

__all__ = ["DatabaseNonceAdapter"]

4.3.3 AlpeNonce DocType

New file: framework_m_core/doctypes/alpe_nonce.py

"""ALPE Nonce DocType.

System table used by DatabaseNonceAdapter to track consumed JWE nonces within
the anti-replay sliding window. This table has no auto-generated REST API
endpoints and bypasses row-level security.
"""

from datetime import datetime

from pydantic import Field

from framework_m_core.domain.base_doctype import BaseDocType


class AlpeNonce(BaseDocType):
    """Stores consumed ALPE nonces for replay-attack detection.

    Entries are automatically pruned after `expires_at` via opportunistic
    cleanup in DatabaseNonceAdapter or a scheduled maintenance sweep.
    """

    nonce: str = Field(
        description="RFC 4122 v4 UUID nonce from the JWE Protected Header.",
        max_length=64,
    )
    created_at: datetime = Field(
        description="Server timestamp when the nonce was first consumed.",
    )
    expires_at: datetime = Field(
        description="Expiry timestamp after which this record may be pruned.",
    )

    class Meta:
        table_name = "alpe_nonces"
        api_resource = False
        show_in_desk = False
        apply_rls = False
        requires_auth = False
        indexes = [
            {"fields": ["nonce"]},
            {"fields": ["expires_at"], "name": "ix_alpe_nonces_expires_at"},
        ]

The SQL DDL for PostgreSQL (leveraging RFC-0013's IndexSpec partial index support):

CREATE TABLE alpe_nonces (
    nonce       VARCHAR(64)                  PRIMARY KEY,
    created_at  TIMESTAMP WITH TIME ZONE     NOT NULL,
    expires_at  TIMESTAMP WITH TIME ZONE     NOT NULL
);

CREATE INDEX ix_alpe_nonces_expires_at ON alpe_nonces (expires_at);

Open question: BaseDocType provides id (UUID PK), name, creation, modified, and other standard columns. Using nonce as the PK directly requires either a RawDocType base class or an explicit ON CONFLICT on the nonce unique column rather than the PK. See §10, Q1.

---

4.4 PayloadEncryptionConfig Extension

config.py (libs/framework-m-standard/src/framework_m_standard/adapters/web/config.py) receives two new fields on PayloadEncryptionConfig:

class PayloadEncryptionConfig(BaseModel):
    # ... existing fields unchanged ...

    # ── Anti-Replay Hardening ────────────────────────────────────────────────
    max_clock_skew_seconds: int = Field(
        default=60,
        ge=10,
        le=300,
        description=(
            "Maximum tolerated deviation in seconds between the client-supplied "
            "`iat` claim and the server's current time. Requests outside this "
            "window are rejected immediately without a nonce store lookup."
        ),
    )
    enforce_anti_replay: bool | None = Field(
        default=None,
        description=(
            "Enable iat + nonce anti-replay enforcement. "
            "When None (default), derived from `mode`: True for 'strict', "
            "False otherwise. Set explicitly to override."
        ),
    )

    @model_validator(mode="after")
    def resolve_anti_replay_default(self) -> PayloadEncryptionConfig:
        """Auto-enable anti-replay for strict mode when not explicitly set."""
        if self.enforce_anti_replay is None:
            self.enforce_anti_replay = self.mode == "strict"
        return self

TOML configuration for a zero-trust deployment:

[web.payload_encryption]
mode = "strict"
jwks_uri = "/api/v1/security/jwks"
max_clock_skew_seconds = 60
# enforce_anti_replay defaults to true when mode = "strict"

---

4.5 PayloadDecryptionMiddleware Extension

payload_decryption.py (libs/framework-m-standard/src/framework_m_standard/adapters/web/middleware/payload_decryption.py) gains three new capabilities:

1. Extract and decode the protected field from the envelope post-decryption.
2. Validate the iat claim against the server clock.
3. Acquire the nonce via the injected NonceStoreProtocol adapter.

class PayloadDecryptionMiddleware:
    """ASGI middleware that decrypts encrypted request bodies.

    When `enforce_anti_replay` is True, also validates the `iat` claim in the
    JWE Protected Header and rejects duplicate nonces via the injected nonce
    store.
    """

    priority = 15

    def __init__(
        self,
        app: ASGIApp,
        crypto: PayloadCryptoProtocol,
        enabled: bool = True,
        nonce_store: NonceStoreProtocol | None = None,
        max_clock_skew_seconds: int = 60,
        enforce_anti_replay: bool = False,
    ) -> None:
        self.app = app
        self.crypto = crypto
        self.enabled = enabled
        self.nonce_store = nonce_store
        self.max_clock_skew_seconds = max_clock_skew_seconds
        # Only enforce if a nonce store is actually wired
        self.enforce_anti_replay = enforce_anti_replay and nonce_store is not None

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        # ... existing early-exit guards unchanged (non-HTTP, safe methods,
        #     wrong content-type) ...

        body = await self._read_body(receive)
        if not body:
            await self.app(scope, receive, send)
            return

        try:
            plaintext = self.crypto.decrypt_payload(body)
        except Exception as exc:
            logger.warning("Payload decryption failed: %s", exc)
            await self._send_error_response(send, str(exc))
            return

        # ── Anti-Replay Pipeline ─────────────────────────────────────────────
        if self.enforce_anti_replay:
            try:
                envelope = json.loads(body)
                header = self._decode_protected_header(envelope.get("protected", ""))
            except Exception as exc:
                logger.warning("Could not decode JWE protected header: %s", exc)
                await self._send_error_response(send, "Malformed JWE protected header")
                return

            # Step 1: Clock-skew check
            iat = header.get("iat")
            if not isinstance(iat, (int, float)):
                await self._send_error_response(
                    send, "Payload timestamp (iat) missing or invalid"
                )
                return

            now = time.time()
            if abs(now - float(iat)) > self.max_clock_skew_seconds:
                logger.warning(
                    "Payload iat rejected: delta=%.1fs, max=%ds",
                    abs(now - float(iat)),
                    self.max_clock_skew_seconds,
                )
                await self._send_error_response(
                    send, "Payload timestamp expired or invalid"
                )
                return

            # Step 2: Nonce deduplication
            nonce = header.get("nonce")
            if not isinstance(nonce, str) or not nonce:
                await self._send_error_response(
                    send, "Payload nonce missing or invalid"
                )
                return

            acquired = await self.nonce_store.acquire(
                nonce, self.max_clock_skew_seconds
            )
            if not acquired:
                logger.warning(
                    "Replay attack detected: duplicate nonce=%s", nonce[:8] + "..."
                )
                await self._send_error_response(
                    send, "Replay attack detected: duplicate nonce"
                )
                return

        # ... existing Content-Type rewrite + app forwarding unchanged ...

    @staticmethod
    def _decode_protected_header(encoded: str) -> dict:
        """Base64url-decode and JSON-parse the JWE Protected Header."""
        import base64

        padding = 4 - (len(encoded) % 4)
        if padding != 4:
            encoded += "=" * padding
        raw = base64.urlsafe_b64decode(encoded.encode("ascii"))
        return json.loads(raw)

---

4.6 Dependency Injection — Nonce Store Factory

The NonceStoreProtocol adapter is wired in the application container, following the same pattern as RedisIdempotencyAdapter / DatabaseIdempotencyAdapter:

# framework_m_standard/adapters/web/app.py (or lifespan.py)


def _build_nonce_store(config: PayloadEncryptionConfig) -> NonceStoreProtocol | None:
    """Return the appropriate nonce store based on config and infra availability."""
    if not config.enforce_anti_replay:
        return None

    redis_url = os.environ.get("REDIS_URL")
    if redis_url:
        import redis.asyncio as aioredis
        from framework_m_standard.adapters.crypto.nonce.redis import RedisNonceAdapter

        client = aioredis.from_url(redis_url, decode_responses=True)
        return RedisNonceAdapter(client)

    from framework_m_standard.adapters.crypto.nonce.database import DatabaseNonceAdapter

    return DatabaseNonceAdapter(session_factory=get_session_factory())

middleware_registry.py is updated to pass nonce_store, max_clock_skew_seconds, and enforce_anti_replay to PayloadDecryptionMiddleware.

---

4.7 JWEPayloadCrypto — AAD Alignment

The current jwe_crypto.py (libs/framework-m-standard/src/framework_m_standard/adapters/crypto/jwe_crypto.py) passes None as the AAD argument to AESGCM.encrypt() / AESGCM.decrypt(). For iat and nonce to be tamper-proof, the base64url-encoded protected header must be supplied as AAD:

# encrypt_payload — pass protected header as AAD
aad = _b64url_encode(json.dumps(header).encode("utf-8")).encode("ascii")
ciphertext_with_tag = aesgcm.encrypt(iv, plaintext, aad)

# decrypt_payload — reconstruct AAD from envelope
aad = envelope["protected"].encode("ascii")
plaintext = aesgcm.decrypt(iv, ciphertext + tag, aad)

Warning: This is a wire-format breaking change. Envelopes produced by the old frontend cannot be decrypted by the new backend. Frontend and backend must be deployed atomically. Use enforce_anti_replay = false temporarily during a staged rollout if an atomic cut-over is not possible.

---

5. Sequence Diagram

sequenceDiagram
    autonumber
    participant Client as Frontend (WebCrypto)
    participant Middleware as PayloadDecryptionMiddleware
    participant Store as NonceStoreProtocol
    participant Route as Route Handler

    Client->>Middleware: POST /api/v1/resource<br/>Content-Type: application/jose<br/>protected contains {alg,enc,cty,iat,nonce}

    Middleware->>Middleware: Decrypt JWE envelope (RSA-OAEP + AES-GCM)<br/>Decode protected header

    rect rgb(245, 245, 245)
        note right of Middleware: Step 1 — Clock-Skew Check
        Middleware->>Middleware: Compute |now() - iat|
        alt iat expired or in future (> max_clock_skew_seconds)
            Middleware-->>Client: 400 Bad Request<br/>"Payload timestamp expired or invalid"
        end
    end

    rect rgb(245, 245, 245)
        note right of Middleware: Step 2 — Nonce Deduplication
        Middleware->>Store: acquire(nonce, ttl=max_clock_skew_seconds)
        alt Nonce already exists (replay)
            Store-->>Middleware: False
            Middleware-->>Client: 400 Bad Request<br/>"Replay attack detected: duplicate nonce"
        else First time seen
            Store-->>Middleware: True
        end
    end

    Middleware->>Route: Forward decrypted body (Content-Type: application/json)
    Route-->>Client: 200 OK

---

6. Security Analysis

6.1 Threat Model

Threat	Mitigated By	Residual Risk
Replay within TLS session	TLS sequence numbers	None — TLS prevents this natively.
Replay via proxy log capture	iat + nonce sliding window	None within max_clock_skew_seconds.
In-transit iat/nonce tampering	AES-GCM AAD (GCM tag covers protected header)	None — tag mismatch causes decryption failure.
Clock manipulation (future iat)	Clock-skew upper bound enforced server-side	Attacker can extend validity by at most max_clock_skew_seconds (60 s).
UUID collision	RFC 4122 v4 provides 2^122 entropy	Negligible (< 2^-60 per 60-second window).
Redis/DB nonce store compromise	Nonces have no value without the RSA private key	Key management is out of scope for this RFC.
DoS via nonce store flooding	Entries TTL-expire; Redis uses bounded memory	Monitor with Redis MEMORY USAGE; LRU eviction policy recommended.

6.2 Fail-Open vs. Fail-Closed

DatabaseNonceAdapter.acquire() returns True (allow) on unexpected database errors rather than False (block). This is a deliberate availability-over- security trade-off: a transient DB error should not cause a complete API outage. Operators requiring hard fail-closed behaviour can subclass the adapter and raise an exception instead, which the middleware will convert to a 503.

---

7. Database Dialect Compatibility Matrix

Feature	PostgreSQL 14+	SQLite 3.35+	Notes
INSERT … ON CONFLICT DO NOTHING	✅ Full	✅ OR IGNORE	Primary deduplication mechanism
TIMESTAMP WITH TIME ZONE	✅ Full	✅ UTC ISO	Used for expires_at
CREATE INDEX … WHERE (partial indexes)	✅ Full	✅ (≥ 3.8)	Covered by RFC-0013 IndexSpec.where
SET … NX EX (Redis atomic check-set)	N/A	N/A	Redis only; zero DB involvement

---

8. Implementation Plan

Phase 1 — Core Protocol Definitions (libs/framework-m-core)

• Create framework_m_core/interfaces/nonce.py with NonceStoreProtocol.
• Create framework_m_core/doctypes/alpe_nonce.py with AlpeNonce DocType.
• Register AlpeNonce in framework_m_core/doctypes/__init__.py.

Phase 2 — Standard Adapters (libs/framework-m-standard)

• Create framework_m_standard/adapters/crypto/nonce/__init__.py.
• Implement RedisNonceAdapter in framework_m_standard/adapters/crypto/nonce/redis.py.
• Implement DatabaseNonceAdapter in framework_m_standard/adapters/crypto/nonce/database.py.

Phase 3 — Configuration (libs/framework-m-standard)

• Add max_clock_skew_seconds: int and enforce_anti_replay: bool | None to PayloadEncryptionConfig in framework_m_standard/adapters/web/config.py.
• Add @model_validator to resolve enforce_anti_replay from mode.

Phase 4 — AAD Alignment (libs/framework-m-standard)

• Update JWEPayloadCrypto.encrypt_payload() to supply base64url-encoded protected header as AES-GCM aad.
• Update JWEPayloadCrypto.decrypt_payload() to pass envelope["protected"] as aad bytes.
• Refactor JWE_HEADER constant into a build_jwe_header(iat, nonce) helper for server-side test tooling.

Phase 5 — Middleware Extension (libs/framework-m-standard)

• Extend PayloadDecryptionMiddleware.__init__ with nonce_store, max_clock_skew_seconds, and enforce_anti_replay parameters.
• Implement _decode_protected_header() static method.
• Implement iat clock-skew check and nonce deduplication in __call__.
• Update middleware_registry.py to wire the nonce store from DI container.
• Implement _build_nonce_store() factory in lifespan.py / app.py.

Phase 6 — Frontend (libs/framework-m-desk)

• Replace static JWE_HEADER constant with buildJweHeader() factory in crypto.ts.
• Update encryptPayload() to pass encodedHeader as additionalData to crypto.subtle.encrypt().

Phase 7 — Test Suite

• Backend unit test_nonce_redis_adapter.py: acquire() returns True on first call, False on duplicate; correct TTL.
• Backend unit test_nonce_database_adapter.py: Same assertions against in-memory SQLite; verify opportunistic pruning.
• Backend unit test_payload_decryption_middleware.py (parametrized):
  • Valid iat + unique nonce → 200 OK.
  • iat > 60 s in the past → 400.
  • iat > 60 s in the future → 400.
  • Duplicate nonce within window → 400.
  • Missing iat → 400.
  • Missing nonce → 400.
  • enforce_anti_replay = False → all pass through (no checks).
• Backend unit test_jwe_crypto_aad.py: Tampered protected header causes decrypt_payload() to raise ValueError.
• Frontend unit crypto.test.ts: iat populated; nonce is valid UUID; sequential calls differ.
• Integration test_alpe_anti_replay_e2e.py: Tests against live Redis and live PostgreSQL nonce backends.

---

9. Drawbacks

• Breaking wire-format change: The AAD extension (§4.7) requires simultaneous deployment of updated frontend (crypto.ts) and backend (jwe_crypto.py). A mixed-version deployment causes decryption failures for all encrypted requests. Operators must perform an atomic upgrade or use enforce_anti_replay = false temporarily.
• Latency increase in strict mode: Each request now incurs one additional asynchronous I/O operation (Redis SET or DB INSERT). On a local Redis instance, this is typically 0.3–1 ms. Applications with sub-5 ms P99 latency targets should measure and accept this explicitly.
• SQLite rowcount reliability: rowcount for OR IGNORE may return -1 in some aiosqlite versions. The adapter must handle this edge case. See §10, Q2.

---

10. Unresolved Questions

1. AlpeNonce primary key strategy: BaseDocType provides id (UUID), name, creation, modified, and standard columns. AlpeNonce ideally uses nonce as its PK directly for ON CONFLICT efficiency. Does Framework M support a RawDocType base with a custom PK, or should AlpeNonce extend BaseDocType with a separate nonce unique-indexed column?

2. OR IGNORE + rowcount on SQLite: INSERT OR IGNORE rowcount behaviour (-1 vs 0 vs 1) varies across aiosqlite / pysqlite versions. Should DatabaseNonceAdapter perform a follow-up SELECT COUNT(*) on -1 for correctness, or is fail-open acceptable in that branch?

3. Opportunistic pruning vs. ScheduledJob: Should AlpeNonce expired-row cleanup also be registered as a Framework M ScheduledJob running every 5 minutes, or is 2%-probability opportunistic cleanup sufficient?

4. max_clock_skew_seconds upper bound: Is le=300 (5 minutes) acceptable for zero-trust environments, or should the ceiling be tighter (e.g., 120 s)?

5. nonce format enforcement: Should the middleware validate that nonce conforms to RFC 4122 v4 UUID format (regex or uuid.UUID(..., version=4)) to prevent DoS via maliciously large nonce strings, or is any non-empty string acceptable?

---

11. Alternatives Considered

Alternative	Pros	Cons	Reason for Rejection
Nonce-only (no iat)	Simpler header	Must store all nonces forever; infinite DB bloat	Incompatible with zero-cliff philosophy
iat-only (no nonce)	Zero storage overhead	Does not prevent replay within max_clock_skew_seconds window	Leaves up to 120 s replay window; insufficient for strict mode
JTI in JWT wrapper around JWE	Standard RFC 7519 approach	Requires a full JWT outer layer; increases payload size & complexity	ALPE already uses JWE; JWT wrapper is redundant overhead
HTTP Replay-Nonce header (ACME-style)	Server-issued nonces guarantee freshness	Requires a prefetch round-trip before every mutating request	Violates Framework M's single-round-trip principle
Dedicated HMAC request-signing	Strong replay protection	Requires symmetric key distribution to the frontend	Incompatible with the RSA public-key-only frontend model

---

12. Migration & Backwards Compatibility

• Breaking changes: AAD extension (§4.7) is a wire-format breaking change. Old-format envelopes will fail decryption on the new backend. Deploy frontend and backend atomically, or use enforce_anti_replay = false during a staged rollout.
• Non-strict modes: enforce_anti_replay defaults to false for disabled / selective / metadata. Existing deployments are completely unaffected — no migration required.
• New database table: alpe_nonces is auto-created on first startup via Framework M's standard schema migration flow. No manual DDL is required.
• Redis namespacing: The nonce:alpe: key prefix does not conflict with any existing Framework M Redis keys.

---

13. References

• RFC 4122 — UUID v4
• RFC 7515 — JSON Web Signature
• RFC 7516 — JSON Web Encryption (JWE)
• NIST SP 800-38D — AES-GCM AAD
• ACME RFC 8555 — Replay-Nonce Header Pattern
• Framework M RFC-0013 — Database Operations & Concurrency
• libs/framework-m-standard/src/framework_m_standard/adapters/web/middleware/payload_decryption.py — Current middleware implementation
• libs/framework-m-standard/src/framework_m_standard/adapters/crypto/jwe_crypto.py — Current JWE adapter
• libs/framework-m-desk/src/crypto.ts — Current frontend ALPE client
• libs/framework-m-standard/src/framework_m_standard/adapters/web/config.py — Current web configuration