hpke-http 1.0.1

End-to-end encryption for HTTP APIs using RFC 9180 HPKE

hpke-http

End-to-end encryption for HTTP APIs using RFC 9180 HPKE.

Highlights

• Transparent - Drop-in middleware, no application code changes
• E2E encryption - Protects data even with TLS termination at CDN/LB
• PSK binding - Each request cryptographically bound to API key
• Replay protection - Counter-based nonces prevent replay attacks
• RFC 9180 compliant - Auditable, interoperable standard

Installation

uv add "hpke-http[fastapi]"       # Server
uv add "hpke-http[aiohttp]"       # Client
uv add "hpke-http[fastapi,zstd]"  # + compression

Quick Start

Both standard JSON requests and SSE streaming are transparently encrypted.

Server (FastAPI)

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from hpke_http.middleware.fastapi import HPKEMiddleware
from hpke_http.constants import KemId

app = FastAPI()

async def resolve_psk(scope: dict) -> tuple[bytes, bytes]:
    api_key = dict(scope["headers"]).get(b"authorization", b"").decode()
    return (api_key.encode(), (await lookup_tenant(api_key)).encode())

app.add_middleware(
    HPKEMiddleware,
    private_keys={KemId.DHKEM_X25519_HKDF_SHA256: private_key},
    psk_resolver=resolve_psk,
)

# Standard JSON endpoint - encryption is automatic
@app.post("/users")
async def create_user(request: Request):
    data = await request.json()  # Decrypted automatically
    return {"id": 123, "name": data["name"]}  # Encrypted automatically

# SSE streaming endpoint - encryption is automatic
@app.post("/chat")
async def chat(request: Request):
    data = await request.json()  # Decrypted automatically

    async def generate():
        yield b"event: progress\ndata: {\"step\": 1}\n\n"
        yield b"event: complete\ndata: {\"result\": \"done\"}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

Client (aiohttp)

from hpke_http.middleware.aiohttp import HPKEClientSession

async with HPKEClientSession(
    base_url="https://api.example.com",
    psk=api_key,        # >= 32 bytes
    psk_id=tenant_id,
) as session:
    # Standard JSON request - encryption is automatic
    resp = await session.post("/users", json={"name": "Alice"})
    user = await resp.json()  # Decrypted automatically
    print(user)  # {"id": 123, "name": "Alice"}

    # SSE streaming request - encryption is automatic
    resp = await session.post("/chat", json={"prompt": "Hello"})
    async for chunk in session.iter_sse(resp):
        print(chunk)  # b"event: progress\ndata: {...}\n\n"

Documentation

• RFC 9180 - HPKE
• RFC 7748 - X25519
• RFC 5869 - HKDF
• RFC 8439 - ChaCha20-Poly1305
• RFC 8878 - Zstandard (optional compression)

Security

Uses OpenSSL constant-time implementations via cryptography library.

• Security Policy - Vulnerability reporting
• SBOM - CycloneDX attached to releases

Contributing

Contributions welcome! Please open an issue first to discuss changes.

make install      # Setup venv
make test         # Run tests
make lint         # Format and lint

License

Apache-2.0

---

Technical Details

Cipher Suite

Component	Algorithm	ID
KEM	DHKEM(X25519, HKDF-SHA256)	0x0020
KDF	HKDF-SHA256	0x0001
AEAD	ChaCha20-Poly1305	0x0003
Mode	PSK	0x01

Wire Format

Request/Response (Chunked Binary)

Headers:
  X-HPKE-Enc: <base64url(32B ephemeral key)>
  X-HPKE-Stream: <base64url(4B session salt)>

Body (repeating chunks):
┌───────────┬────────────┬─────────────────────────────────┐
│ Length(4B)│ Counter(4B)│ Ciphertext (N + 16B tag)        │
│ big-end   │ big-end    │ encrypted: encoding_id || data  │
└───────────┴────────────┴─────────────────────────────────┘
Overhead: 24B/chunk (4B length + 4B counter + 16B tag)

SSE Event

event: enc
data: <base64(counter_be32 || ciphertext)>
Decrypted: raw SSE chunk (e.g., "event: progress\ndata: {...}\n\n")

Uses standard base64 (not base64url) - SSE data fields allow +/= characters.

Auto-Encryption

The middleware automatically encrypts all responses when the request was encrypted:

• Standard responses (JSON, HTML, etc.) - Uses chunked binary format (RawFormat)
• SSE responses - Uses base64-encoded SSE events (SSEFormat)

Response type is detected via Content-Type header:

• text/event-stream → SSE format (requires media_type="text/event-stream")
• Everything else → Binary chunked format

Compression (Optional)

Zstd compression reduces bandwidth by 40-95% for JSON/text.

HPKEMiddleware(..., compress=True)      # Server
HPKEClientSession(..., compress=True)   # Client

Choice	Rationale
Compress-then-encrypt	Encrypted data is incompressible
Zstd (RFC 8878)	Best ratio/speed, Python 3.14 native
64B threshold	Smaller payloads skip compression

Pitfalls

# PSK too short
HPKEClientSession(psk=b"short")                 # InvalidPSKError
HPKEClientSession(psk=secrets.token_bytes(32))  # >= 32 bytes

# SSE missing content-type (won't use SSE format)
return StreamingResponse(gen())                                  # Binary format (wrong for SSE)
return StreamingResponse(gen(), media_type="text/event-stream")  # SSE format (correct)

# Standard responses work automatically - no special handling needed
return {"data": "value"}  # Auto-encrypted as binary chunks

Limits

Resource	Limit	Applies to
HPKE messages/context	2^96-1	All
Chunks/session	2^32-1	All
PSK minimum	32 bytes	All
Chunk size	64KB	All
Binary chunk overhead	24B (length + counter + tag)	Requests & standard responses
SSE event buffer	64MB (configurable)	SSE only

Note: SSE is text-only (UTF-8). Binary data must be base64-encoded (+33% overhead).

Low-Level API

from hpke_http.hpke import seal_psk, open_psk

enc, ct = seal_psk(pk_r, b"info", psk, psk_id, b"aad", b"plaintext")
pt = open_psk(enc, sk_r, b"info", psk, psk_id, b"aad", ct)