byteforge-hmac

A Python library for HMAC-based HTTP request authentication with built-in timestamp validation and replay attack protection.

Features

• HMAC-SHA256 Signature Verification - Cryptographically secure request authentication
• Timestamp Validation - Configurable tolerance window to prevent stale requests
• Replay Attack Protection - Nonce tracking to prevent request replay
• Server & Client Components - Complete solution for both sides of authentication
• Flexible Secret Management - Pluggable secret provider architecture
• Framework Agnostic - Works with Flask, Django, FastAPI, or any Python web framework

Installation

pip install byteforge-hmac

For development with Flask examples:

pip install byteforge-hmac[dev]

Quick Start

Server-Side: Protecting API Endpoints

from flask import Flask, request, jsonify
from byteforge_hmac import (
    HMACAuthenticator,
    DictSecretProvider,
    AuthHeaderParser
)

app = Flask(__name__)

# Initialize with your client secrets
# In production, use a database or secret management service
secrets = {
    'client_123': 'secret_key_abc',
    'client_456': 'secret_key_xyz'
}

secret_provider = DictSecretProvider(secrets)
authenticator = HMACAuthenticator(
    secret_provider=secret_provider,
    timestamp_tolerance=300  # 5 minutes
)

@app.route('/api/protected', methods=['GET', 'POST'])
def protected_endpoint():
    # Parse the Authorization header
    auth_header = request.headers.get('Authorization', '')
    auth_request = AuthHeaderParser.parse(auth_header)

    if not auth_request:
        return jsonify({'error': 'Unauthorized'}), 401

    # Extract request details
    method = request.method
    path = request.path
    body = request.get_data(as_text=True) or ''

    # Authenticate the request
    if not authenticator.authenticate(auth_request, method, path, body):
        return jsonify({'error': 'Authentication failed'}), 403

    # Request is authenticated - proceed with business logic
    return jsonify({'status': 'success', 'data': 'Protected resource'})

Client-Side: Making Authenticated Requests

from byteforge_hmac import HMACClient

# Initialize the client
client = HMACClient(
    client_id='client_123',
    secret_key='secret_key_abc',
    base_url='https://api.example.com'
)

# Make authenticated GET request
response = client.get('/api/protected')
print(response.json())

# Make authenticated POST request with data
data = {'name': 'example', 'value': 42}
response = client.post('/api/protected', data=data)
print(response.json())

# Other HTTP methods are also supported
response = client.put('/api/resource', data={'update': 'value'})
response = client.delete('/api/resource')

How It Works

Authentication Flow

1. Client generates a signature:

   • Creates a Unix timestamp
   • Generates a unique nonce (UUID)
   • Computes HMAC-SHA256 signature over: {method}\n{path}\n{timestamp}\n{nonce}\n{body}
   • Sends request with Authorization header

2. Server validates the request:

   • Timestamp Check: Ensures request is within tolerance window (prevents stale requests)
   • Replay Check: Verifies nonce hasn't been seen before (prevents replay attacks)
   • Signature Verification: Recomputes signature and compares using constant-time comparison

Authorization Header Format

Authorization: HMAC client_id="client_123",timestamp="1234567890",nonce="uuid-string",signature="hex-signature"

Signature Calculation

The HMAC-SHA256 signature is calculated over the following message format:

{HTTP_METHOD}\n{PATH}\n{TIMESTAMP}\n{NONCE}\n{BODY}

Example for POST /api/data with body {"key":"value"}:

POST\n/api/data\n1234567890\nuuid-here\n{"key":"value"}

Server-Side Usage

Custom Secret Provider

Implement your own secret provider to integrate with databases or secret management services:

from byteforge_hmac import SecretProvider
from typing import Optional

class DatabaseSecretProvider(SecretProvider):
    def __init__(self, db_connection):
        self.db = db_connection

    def get_secret(self, client_id: str) -> Optional[str]:
        # Query your database
        result = self.db.query(
            "SELECT secret_key FROM clients WHERE client_id = %s",
            (client_id,)
        )
        return result[0] if result else None

# Use it with the authenticator
secret_provider = DatabaseSecretProvider(db_connection)
authenticator = HMACAuthenticator(secret_provider=secret_provider)

Configuration Options

authenticator = HMACAuthenticator(
    secret_provider=secret_provider,
    timestamp_tolerance=300,  # Time tolerance in seconds (default: 300)
    nonce_storage=None       # Optional: any NonceStorage backend.
                             # Defaults to DictNonceStorage (process-local).
)

Persistent Nonce Storage

⚠️ Required for Production: the default DictNonceStorage lives inside a single process. It is atomic across threads, but two gunicorn workers or two containers each keep their own copy, so a captured request replays cleanly against a worker that has not seen it. Any deployment with more than one process needs a shared backend.

A backend implements exactly one method:

def put_if_absent(self, key: str, value: int, ttl_seconds: int) -> bool

It must store the key and report whether it was already present, as a single atomic operation. Returning True means the nonce is new; False means a replay. Every real backend has this primitive: Redis SET NX EX, memcached add, INSERT ... ON CONFLICT DO NOTHING RETURNING.

import redis
from byteforge_hmac import HMACAuthenticator

class RedisNonceStorage:
    """Shared nonce storage using Redis."""

    def __init__(self, redis_client):
        self.redis = redis_client

    def put_if_absent(self, key: str, value: int, ttl_seconds: int) -> bool:
        # SET key value NX EX ttl -- one round trip, atomic in Redis itself.
        # Returns None (falsey) when the key already exists.
        return bool(self.redis.set(key, value, nx=True, ex=ttl_seconds))

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

authenticator = HMACAuthenticator(
    secret_provider=secret_provider,
    timestamp_tolerance=300,
    nonce_storage=RedisNonceStorage(redis_client)
)

Honour the ttl_seconds you are passed; do not apply a TTL of your own. ReplayProtector computes the exact remaining lifetime per request (timestamp + tolerance - now), so the nonce is forgotten at precisely the moment TimestampValidator stops accepting the request. The value is not a constant, and it is not the tolerance: TimestampValidator compares with abs(), so a request timestamped 300 seconds in the future is accepted now and stays acceptable for another 600 seconds. A backend hardcoding setex(key, 300, value) forgets that nonce halfway through its live window, which is a replay hole.

Upgrading from 0.1.x

Backends that implement only __contains__ and __setitem__ still work. Constructing a ReplayProtector over one logs a warning to byteforge_hmac.replay_protector and raises a DeprecationWarning — once, not per request. Both, because Python ignores DeprecationWarning by default outside __main__, so under gunicorn the warning alone would reach nobody. That fallback is not atomic — see below — and will be removed in 0.3.0. Replace the two methods with put_if_absent; items(), __getitem__ and __delitem__ shims are no longer needed at all.

Passing a plain {} as nonce_storage also takes the deprecated path. Use DictNonceStorage instead:

from byteforge_hmac import DictNonceStorage

authenticator = HMACAuthenticator(
    secret_provider=secret_provider,
    nonce_storage=DictNonceStorage()   # or omit entirely -- this is the default
)

Framework Integration Examples

Django

from django.http import JsonResponse
from byteforge_hmac import HMACAuthenticator, AuthHeaderParser

def protected_view(request):
    auth_header = request.META.get('HTTP_AUTHORIZATION', '')
    auth_request = AuthHeaderParser.parse(auth_header)

    if not auth_request:
        return JsonResponse({'error': 'Unauthorized'}, status=401)

    method = request.method
    path = request.path
    body = request.body.decode('utf-8') if request.body else ''

    if not authenticator.authenticate(auth_request, method, path, body):
        return JsonResponse({'error': 'Authentication failed'}, status=403)

    return JsonResponse({'status': 'success'})

FastAPI

from fastapi import FastAPI, Request, HTTPException, Depends
from byteforge_hmac import HMACAuthenticator, AuthHeaderParser

app = FastAPI()

async def verify_hmac(request: Request):
    auth_header = request.headers.get('authorization', '')
    auth_request = AuthHeaderParser.parse(auth_header)

    if not auth_request:
        raise HTTPException(status_code=401, detail="Unauthorized")

    # Read body
    body = await request.body()
    body_str = body.decode('utf-8') if body else ''

    if not authenticator.authenticate(
        auth_request,
        request.method,
        request.url.path,
        body_str
    ):
        raise HTTPException(status_code=403, detail="Authentication failed")

    return auth_request

@app.post("/api/protected")
async def protected_endpoint(auth_request = Depends(verify_hmac)):
    return {"status": "success", "client_id": auth_request.client_id}

Client-Side Usage

Basic Client Usage

from byteforge_hmac import HMACClient

client = HMACClient(
    client_id='your_client_id',
    secret_key='your_secret_key',
    base_url='https://api.example.com'
)

# GET request
response = client.get('/api/users')

# POST request with JSON data
response = client.post('/api/users', data={'name': 'John', 'email': 'john@example.com'})

# PUT request
response = client.put('/api/users/123', data={'name': 'Jane'})

# DELETE request
response = client.delete('/api/users/123')

Advanced Client Usage

# Pass additional requests library arguments
response = client.get(
    '/api/data',
    params={'page': 1, 'limit': 10},
    timeout=30
)

# Custom headers (Authorization header is automatically added)
response = client.post(
    '/api/data',
    data={'key': 'value'},
    headers={'X-Custom-Header': 'custom-value'}
)

# Using the generic request method
response = client.request(
    'PATCH',
    '/api/resource',
    data={'field': 'updated'}
)

Manual Signature Generation

If you need to generate signatures manually without using HMACClient:

import hmac
import hashlib
import time
import uuid

def generate_hmac_signature(secret_key, method, path, timestamp, nonce, body=''):
    message = f"{method}\n{path}\n{timestamp}\n{nonce}\n{body}"
    signature = hmac.new(
        secret_key.encode('utf-8'),
        message.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
    return signature

# Generate components
timestamp = str(int(time.time()))
nonce = str(uuid.uuid4())
signature = generate_hmac_signature(
    'your_secret_key',
    'GET',
    '/api/data',
    timestamp,
    nonce
)

# Create Authorization header
auth_header = f'HMAC client_id="your_client",timestamp="{timestamp}",nonce="{nonce}",signature="{signature}"'

Security Considerations

Timestamp Tolerance

The timestamp_tolerance parameter defines how old a request can be before it's rejected. Consider:

• Shorter tolerance (e.g., 60 seconds): More secure but requires tighter clock synchronization
• Longer tolerance (e.g., 300 seconds): More forgiving of clock drift but larger replay window
• Default is 300 seconds (5 minutes)

Nonce Storage

⚠️ Replay protection is only as shared as your storage.

The default DictNonceStorage serialises concurrent callers with a lock, so a nonce cannot be accepted twice within one process. It is not shared between processes. Under gunicorn with four workers you have four independent nonce stores, and a captured request replays against whichever worker has not seen it. That is a deployment property this library cannot fix for you.

Storage options:

• DictNonceStorage (default): single-process deployments, development, and tests. Thread-safe, bounded, process-local.
• Redis / Memcached: required for anything running more than one process. Implement put_if_absent on SET NX EX / add — both are atomic.
• Database: INSERT ... ON CONFLICT DO NOTHING RETURNING is atomic and needs no explicit locking. A SELECT followed by an INSERT is not, and reintroduces exactly the race this release closed.

Whatever you use, put_if_absent must be genuinely atomic. A backend that checks and then stores in two operations lets two workers both accept the same captured request, which is the failure the single-primitive protocol exists to prevent.

See "Persistent Nonce Storage" above for a Redis implementation and the ttl_seconds contract.

Secret Key Management

• Never hardcode secrets in your application code
• Use environment variables or secret management services (AWS Secrets Manager, HashiCorp Vault, etc.)
• Rotate keys periodically
• Use cryptographically strong random keys (at least 32 bytes of entropy)

TLS/SSL Encryption

⚠️ IMPORTANT: This library does NOT provide encryption

HMAC authentication provides:

• ✅ Request authentication (proves who sent it)
• ✅ Request integrity (detects tampering)
• ❌ NO encryption of request/response data

Production Deployment Requirements:

Your application MUST be deployed behind a TLS-enabled reverse proxy (nginx, Apache, AWS ALB, etc.) to ensure:

• Request/response confidentiality
• Protection against man-in-the-middle attacks
• Server authentication

Recommended Architecture:

Internet → [Nginx with TLS] → [Your Python App with HMAC Auth]

Example nginx configuration:

server {
    listen 443 ssl http2;
    server_name api.example.com;

    ssl_certificate /path/to/cert.pem;
    ssl_certificate_key /path/to/key.pem;

    location / {
        proxy_pass http://localhost:5001;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

Without TLS, request bodies and secrets are transmitted in plaintext over the network.

Testing

Run the included test server and client:

# Terminal 1: Start the test server
python test_server.py

# Terminal 2: Run the test client
python test_client.py

The test server runs on http://localhost:5001 with these test credentials:

• Client ID: test_client_1, Secret: secret_key_123
• Client ID: test_client_2, Secret: another_secret_456

API Reference

Server Components

HMACAuthenticator

Main authenticator class that coordinates all validation steps.

HMACAuthenticator(
    secret_provider: SecretProvider,
    timestamp_tolerance: int = 300,
    nonce_storage: Optional[NonceStorage] = None
)

Methods:

• authenticate(auth_request, method, path, body='') -> bool: Perform complete authentication

AuthHeaderParser

Parses HMAC authorization headers.

Methods:

• parse(auth_header: str) -> Optional[AuthRequest]: Parse Authorization header

SecretProvider

Abstract base class for retrieving client secrets.

Methods:

• get_secret(client_id: str) -> Optional[str]: Get secret for a client

DictSecretProvider

Dictionary-based secret provider for testing/simple use cases.

DictSecretProvider(secrets: Dict[str, str])

NonceStorage

Protocol a nonce storage backend must satisfy.

put_if_absent(key: str, value: int, ttl_seconds: int) -> bool

Store key if and only if it is absent, atomically, and report which happened: True if stored (new nonce), False if it already existed (replay). value is the request's unix timestamp — store it if your backend requires a payload, ignore it otherwise; nothing reads it back. ttl_seconds is computed per request and must be honoured as given.

DictNonceStorage

Process-local in-memory nonce storage. The default when nonce_storage is omitted. Thread-safe via a lock; expired entries are swept on write so memory stays bounded. Not shared between processes — use Redis or a database for multi-worker deployments.

DictNonceStorage()

ReplayProtector

Checks and records nonces against a NonceStorage.

ReplayProtector(storage: NonceStorage)

Methods:

• check_and_store(client_id, nonce, timestamp, tolerance_seconds=300) -> bool: True if the nonce is new, False if a replay was detected.

Client Components

HMACClient

Client for making HMAC-authenticated HTTP requests.

HMACClient(
    client_id: str,
    secret_key: str,
    base_url: str = 'http://localhost:5001'
)

Methods:

• get(path, **kwargs) -> requests.Response
• post(path, data=None, **kwargs) -> requests.Response
• put(path, data=None, **kwargs) -> requests.Response
• delete(path, **kwargs) -> requests.Response
• request(method, path, data=None, **kwargs) -> requests.Response

Models

AuthRequest

Data model for parsed authentication requests.

Attributes:

• client_id: str
• timestamp: str
• nonce: str
• signature: str

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

License

MIT License

Author

Jason Byteforge (@jmazzahacks)