GitHub - dualeai/hpke-http: End-to-end encryption for HTTP APIs using RFC 9180 HPKE. Drop-in middleware for FastAPI, aiohttp, and httpx.

uv add "hpke-http[fastapi]"       # Server
uv add "hpke-http[aiohttp]"       # Client (aiohttp)
uv add "hpke-http[httpx]"         # Client (httpx)
uv add "hpke-http[fastapi,zstd]"  # + zstd compression (gzip fallback included)

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

app = FastAPI()

async def resolve_psk(scope: dict) -> tuple[bytes, bytes]:
    psk_id = scope.get("hpke_psk_id")
    record = await db.lookup_by_derived_id(psk_id)
    if record is None:
        raise HTTPException(401, "Unknown API key")  # Forwarded to client
    scope["tenant_id"] = record["tenant_id"]
    return (record["psk"], psk_id)

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

@app.post("/users")
async def create_user(request: Request):
    data = await request.json()  # Decrypted by middleware
    return {"id": 123, "name": data["name"]}  # Encrypted by middleware

@app.get("/users/{user_id}")
async def get_user(request: Request):
    return {"id": 123, "name": "Alice"}  # Encrypted by middleware

@app.post("/chat")
async def chat(request: Request):
    data = await request.json()

    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")

import hashlib
import aiohttp
from hpke_http.middleware.aiohttp import HPKEClientSession

# Derive PSK ID from API key (see "PSK Authentication" section)
psk_id = hashlib.sha256(api_key).digest()

async with HPKEClientSession(
    base_url="https://api.example.com",
    psk=api_key,        # >= 32 bytes
    psk_id=psk_id,      # Derived from key, not tenant ID
    # compress=True,           # Compression (zstd preferred, gzip fallback)
    # require_encryption=True, # Raise if server responds unencrypted
    # release_encrypted=True,  # Free encrypted bytes after decryption (saves memory)
) as session:
    # POST with JSON body
    async with session.post("/users", json={"name": "Alice"}) as resp:
        user = await resp.json()

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

    # GET (bodyless) - response is still encrypted
    async with session.get("/users/123") as resp:
        user = await resp.json()

    # File upload - streams with O(chunk_size) memory
    form = aiohttp.FormData()
    form.add_field("file", open("large.pdf", "rb"), filename="large.pdf")
    async with session.post("/upload", data=form) as resp:
        result = await resp.json()

import hashlib
from hpke_http.middleware.httpx import HPKEAsyncClient

# Derive PSK ID from API key (see "PSK Authentication" section)
psk_id = hashlib.sha256(api_key).digest()

async with HPKEAsyncClient(
    base_url="https://api.example.com",
    psk=api_key,        # >= 32 bytes
    psk_id=psk_id,      # Derived from key, not tenant ID
    # compress=True,           # Compression (zstd preferred, gzip fallback)
    # require_encryption=True, # Raise if server responds unencrypted
    # release_encrypted=True,  # Free encrypted bytes after decryption (saves memory)
) as client:
    # POST with JSON body
    resp = await client.post("/users", json={"name": "Alice"})
    user = resp.json()

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

    # GET (bodyless) - response is still encrypted
    resp = await client.get("/users/123")
    user = resp.json()

    # File upload - streams with O(chunk_size) memory
    resp = await client.post("/upload", files={"file": open("large.pdf", "rb")})
    result = resp.json()

import hashlib

api_key = b"sk_live_7f3a9c..."  # Your API key (>= 32 bytes)
# Derive PSK ID from the key itself
psk_id = hashlib.sha256(api_key).digest()

async with HPKEClientSession(
    base_url="https://api.example.com",
    psk=api_key,
    psk_id=psk_id,
) as client:
    await client.post("/api", json=data)

import hashlib
from starlette.exceptions import HTTPException

# Key creation: store derived_id → {psk, tenant_id}
derived_id = hashlib.sha256(api_key).digest()
db.store(derived_id, {"psk": api_key, "tenant_id": tenant_id})

# psk_resolver: lookup by derived_id from header
async def resolve_psk(scope: dict) -> tuple[bytes, bytes]:
    derived_id = scope.get("hpke_psk_id")
    record = await db.lookup(derived_id)
    if record is None:
        raise HTTPException(401, "Unknown API key")
    scope["tenant_id"] = record["tenant_id"]
    return (record["psk"], derived_id)

from starlette.exceptions import HTTPException

async def resolve_psk(scope: dict) -> tuple[bytes, bytes]:
    psk_id = scope.get("hpke_psk_id")

    # Token revoked — tell the client exactly what happened
    record = await db.lookup(psk_id)
    if record is None:
        raise HTTPException(401, "Unknown API key")
    if record["revoked"]:
        raise HTTPException(401, "API key revoked")

    # Authorization check — different status code
    if not record["scopes"].issuperset(required_scopes):
        raise HTTPException(403, "Insufficient permissions")

    # Backend unavailable — signal transient failure
    if not await auth_service.healthy():
        raise HTTPException(503, "Auth service unavailable")

    return (record["psk"], psk_id)

raise HTTPException(
    401,
    "Bearer token required",
    headers={"WWW-Authenticate": "Bearer"},
)

Headers:
  X-HPKE-Enc: <base64url(32B ephemeral key)>
  X-HPKE-Stream: <base64url(4B session salt)>
  X-HPKE-PSK-ID: <base64url(derived key ID, 32B recommended)>

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

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

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

# PSK ID must be derived from the key (see "PSK Authentication" section)
psk_id = hashlib.sha256(api_key).digest()
HPKEClientSession(psk=api_key, psk_id=psk_id)   # Correct

# 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

┌─────────────────────────────────────────────────────────────┐
│ TLS Encrypted (transport)                                   │
│  ┌───────────────────────────────────────────────────────┐  │
│  │ HTTP Layer (visible to CDN/LB/proxies)                │  │
│  │  • Method: POST                                       │  │
│  │  • URL: /api/chat                                     │  │
│  │  • Headers: Authorization, X-HPKE-*, Content-Type     │  │
│  │  ┌─────────────────────────────────────────────────┐  │  │
│  │  │ HPKE Encrypted (end-to-end)                     │  │  │
│  │  │  • Request body: {"prompt": "Hello"}            │  │  │
│  │  │  • Response body: {"response": "Hi!"}           │  │  │
│  │  │  • SSE events: event: done\ndata: {...}\n\n     │  │  │
│  │  └─────────────────────────────────────────────────┘  │  │
│  └───────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────┘

from hpke_http.hpke import seal_psk, open_psk

# pk_r: recipient public key, sk_r: recipient secret key
# psk/psk_id: pre-shared key and identifier, aad: additional authenticated data
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)

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