How can I implement HTTP request signing for secure APIs?

HTTP request signing is a crucial security mechanism that ensures API requests are authentic and haven't been tampered with during transmission. This cryptographic technique creates a unique signature for each request based on the request content and a secret key, providing both authentication and integrity verification.

Understanding HTTP Request Signing

Request signing works by generating a cryptographic hash of the request components (headers, body, timestamp) using a shared secret key. The server can then verify the signature to ensure the request came from an authorized source and hasn't been modified.

Common Signing Algorithms

The most widely used algorithms for HTTP request signing include:

• HMAC-SHA256: Hash-based Message Authentication Code using SHA-256
• RSA with SHA-256: Public key cryptography for asymmetric signing
• ECDSA: Elliptic Curve Digital Signature Algorithm

HMAC-SHA256 Implementation

HMAC-SHA256 is the most common approach for API request signing due to its simplicity and security.

Python Implementation

import hmac
import hashlib
import base64
import json
from datetime import datetime
import requests

class APIRequestSigner:
    def __init__(self, access_key, secret_key):
        self.access_key = access_key
        self.secret_key = secret_key.encode('utf-8')

    def create_signature(self, method, url, headers, body=None):
        # Create timestamp
        timestamp = datetime.utcnow().strftime('%Y%m%dT%H%M%SZ')

        # Prepare string to sign
        string_to_sign = self._create_string_to_sign(
            method, url, headers, body, timestamp
        )

        # Generate HMAC-SHA256 signature
        signature = hmac.new(
            self.secret_key,
            string_to_sign.encode('utf-8'),
            hashlib.sha256
        ).digest()

        # Base64 encode the signature
        encoded_signature = base64.b64encode(signature).decode('utf-8')

        return encoded_signature, timestamp

    def _create_string_to_sign(self, method, url, headers, body, timestamp):
        # Canonical headers (sorted by key)
        canonical_headers = []
        for key in sorted(headers.keys()):
            canonical_headers.append(f"{key.lower()}:{headers[key]}")

        # Body hash
        body_hash = hashlib.sha256(
            (body or '').encode('utf-8')
        ).hexdigest()

        # Combine components
        string_to_sign = '\n'.join([
            method.upper(),
            url,
            '\n'.join(canonical_headers),
            body_hash,
            timestamp
        ])

        return string_to_sign

    def sign_request(self, method, url, headers=None, body=None):
        headers = headers or {}

        # Add required headers
        headers.update({
            'X-API-Key': self.access_key,
            'Content-Type': 'application/json'
        })

        # Create signature
        signature, timestamp = self.create_signature(method, url, headers, body)

        # Add authorization header
        headers['Authorization'] = f'HMAC-SHA256 {signature}'
        headers['X-Timestamp'] = timestamp

        return headers

# Usage example
signer = APIRequestSigner('your-access-key', 'your-secret-key')

# Sign a GET request
headers = signer.sign_request('GET', 'https://api.example.com/data')
response = requests.get('https://api.example.com/data', headers=headers)

# Sign a POST request with body
data = {'key': 'value'}
body = json.dumps(data)
headers = signer.sign_request('POST', 'https://api.example.com/create', body=body)
response = requests.post('https://api.example.com/create', headers=headers, data=body)

JavaScript Implementation

const crypto = require('crypto');
const axios = require('axios');

class APIRequestSigner {
    constructor(accessKey, secretKey) {
        this.accessKey = accessKey;
        this.secretKey = secretKey;
    }

    createSignature(method, url, headers, body = null) {
        // Create timestamp
        const timestamp = new Date().toISOString().replace(/[:-]|\.\d{3}/g, '');

        // Create string to sign
        const stringToSign = this.createStringToSign(method, url, headers, body, timestamp);

        // Generate HMAC-SHA256 signature
        const signature = crypto
            .createHmac('sha256', this.secretKey)
            .update(stringToSign)
            .digest('base64');

        return { signature, timestamp };
    }

    createStringToSign(method, url, headers, body, timestamp) {
        // Canonical headers (sorted by key)
        const canonicalHeaders = Object.keys(headers)
            .sort()
            .map(key => `${key.toLowerCase()}:${headers[key]}`)
            .join('\n');

        // Body hash
        const bodyHash = crypto
            .createHash('sha256')
            .update(body || '')
            .digest('hex');

        // Combine components
        return [
            method.toUpperCase(),
            url,
            canonicalHeaders,
            bodyHash,
            timestamp
        ].join('\n');
    }

    signRequest(method, url, headers = {}, body = null) {
        // Add required headers
        headers = {
            ...headers,
            'X-API-Key': this.accessKey,
            'Content-Type': 'application/json'
        };

        // Create signature
        const { signature, timestamp } = this.createSignature(method, url, headers, body);

        // Add authorization headers
        headers['Authorization'] = `HMAC-SHA256 ${signature}`;
        headers['X-Timestamp'] = timestamp;

        return headers;
    }
}

// Usage example
const signer = new APIRequestSigner('your-access-key', 'your-secret-key');

// Sign a GET request
async function makeGetRequest() {
    const headers = signer.signRequest('GET', 'https://api.example.com/data');
    const response = await axios.get('https://api.example.com/data', { headers });
    return response.data;
}

// Sign a POST request with body
async function makePostRequest(data) {
    const body = JSON.stringify(data);
    const headers = signer.signRequest('POST', 'https://api.example.com/create', {}, body);
    const response = await axios.post('https://api.example.com/create', body, { headers });
    return response.data;
}

AWS Signature Version 4

For AWS services, you'll need to implement the AWS Signature Version 4 algorithm:

import hmac
import hashlib
import urllib.parse
from datetime import datetime

class AWSSignatureV4:
    def __init__(self, access_key, secret_key, region, service):
        self.access_key = access_key
        self.secret_key = secret_key
        self.region = region
        self.service = service

    def sign_request(self, method, url, headers, body=''):
        # Parse URL
        parsed_url = urllib.parse.urlparse(url)

        # Create canonical request
        canonical_request = self._create_canonical_request(
            method, parsed_url, headers, body
        )

        # Create string to sign
        timestamp = datetime.utcnow()
        string_to_sign = self._create_string_to_sign(canonical_request, timestamp)

        # Calculate signature
        signing_key = self._get_signature_key(timestamp)
        signature = hmac.new(signing_key, string_to_sign.encode('utf-8'), hashlib.sha256).hexdigest()

        # Create authorization header
        credential = f"{self.access_key}/{timestamp.strftime('%Y%m%d')}/{self.region}/{self.service}/aws4_request"
        signed_headers = ';'.join(sorted(headers.keys()))

        authorization = f"AWS4-HMAC-SHA256 Credential={credential}, SignedHeaders={signed_headers}, Signature={signature}"

        return authorization

    def _create_canonical_request(self, method, parsed_url, headers, body):
        canonical_uri = parsed_url.path or '/'
        canonical_querystring = parsed_url.query or ''
        canonical_headers = '\n'.join([f"{k.lower()}:{v}" for k, v in sorted(headers.items())]) + '\n'
        signed_headers = ';'.join(sorted(headers.keys()))
        payload_hash = hashlib.sha256(body.encode('utf-8')).hexdigest()

        return '\n'.join([
            method.upper(),
            canonical_uri,
            canonical_querystring,
            canonical_headers,
            signed_headers,
            payload_hash
        ])

    def _create_string_to_sign(self, canonical_request, timestamp):
        algorithm = 'AWS4-HMAC-SHA256'
        credential_scope = f"{timestamp.strftime('%Y%m%d')}/{self.region}/{self.service}/aws4_request"
        canonical_request_hash = hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()

        return '\n'.join([
            algorithm,
            timestamp.strftime('%Y%m%dT%H%M%SZ'),
            credential_scope,
            canonical_request_hash
        ])

    def _get_signature_key(self, timestamp):
        date_key = hmac.new(f"AWS4{self.secret_key}".encode('utf-8'), timestamp.strftime('%Y%m%d').encode('utf-8'), hashlib.sha256).digest()
        region_key = hmac.new(date_key, self.region.encode('utf-8'), hashlib.sha256).digest()
        service_key = hmac.new(region_key, self.service.encode('utf-8'), hashlib.sha256).digest()
        signing_key = hmac.new(service_key, b'aws4_request', hashlib.sha256).digest()
        return signing_key

JWT Token Signing

JSON Web Tokens (JWT) provide another approach to request signing:

import jwt
import time
from datetime import datetime, timedelta

class JWTSigner:
    def __init__(self, secret_key, algorithm='HS256'):
        self.secret_key = secret_key
        self.algorithm = algorithm

    def create_token(self, payload, expires_in=3600):
        # Add standard claims
        now = datetime.utcnow()
        payload.update({
            'iat': now,  # Issued at
            'exp': now + timedelta(seconds=expires_in),  # Expiration
            'nbf': now   # Not before
        })

        # Create JWT token
        token = jwt.encode(payload, self.secret_key, algorithm=self.algorithm)
        return token

    def sign_request(self, user_id, permissions=None):
        payload = {
            'sub': user_id,  # Subject
            'permissions': permissions or []
        }

        return self.create_token(payload)

# Usage
jwt_signer = JWTSigner('your-secret-key')
token = jwt_signer.sign_request('user123', ['read', 'write'])

# Add to request headers
headers = {
    'Authorization': f'Bearer {token}',
    'Content-Type': 'application/json'
}

Best Practices for Request Signing

1. Use HTTPS Always

Request signing doesn't encrypt the data; it only provides authentication and integrity. Always use HTTPS to protect against eavesdropping.

2. Include Timestamps

Always include timestamps in your signatures to prevent replay attacks:

def add_timestamp_validation(self, timestamp_header):
    request_time = datetime.strptime(timestamp_header, '%Y%m%dT%H%M%SZ')
    current_time = datetime.utcnow()

    # Reject requests older than 5 minutes
    if abs((current_time - request_time).seconds) > 300:
        raise ValueError("Request timestamp too old")

3. Secure Key Storage

Never hardcode secret keys in your application:

import os
from cryptography.fernet import Fernet

# Use environment variables
SECRET_KEY = os.environ.get('API_SECRET_KEY')

# Or encrypted storage
def load_encrypted_key():
    cipher_suite = Fernet(os.environ.get('ENCRYPTION_KEY'))
    with open('encrypted_key.txt', 'rb') as f:
        encrypted_key = f.read()
    return cipher_suite.decrypt(encrypted_key).decode()

4. Implement Rate Limiting

Combine request signing with rate limiting to prevent abuse:

from functools import wraps
import time

def rate_limit(max_requests=100, window=3600):
    def decorator(func):
        request_counts = {}

        @wraps(func)
        def wrapper(*args, **kwargs):
            client_id = kwargs.get('client_id')
            current_time = time.time()

            # Clean old entries
            request_counts[client_id] = [
                req_time for req_time in request_counts.get(client_id, [])
                if current_time - req_time < window
            ]

            # Check rate limit
            if len(request_counts[client_id]) >= max_requests:
                raise Exception("Rate limit exceeded")

            request_counts[client_id].append(current_time)
            return func(*args, **kwargs)

        return wrapper
    return decorator

Server-Side Verification

On the server side, you'll need to verify the signatures:

def verify_request_signature(request, secret_key):
    # Extract signature from headers
    auth_header = request.headers.get('Authorization', '')
    if not auth_header.startswith('HMAC-SHA256 '):
        return False

    received_signature = auth_header[12:]  # Remove 'HMAC-SHA256 '
    timestamp = request.headers.get('X-Timestamp')

    # Recreate the signature
    signer = APIRequestSigner('', secret_key)
    expected_signature, _ = signer.create_signature(
        request.method,
        request.url,
        dict(request.headers),
        request.body
    )

    # Compare signatures securely
    return hmac.compare_digest(received_signature, expected_signature)

Testing Request Signing

Always test your implementation thoroughly:

import unittest

class TestRequestSigning(unittest.TestCase):
    def setUp(self):
        self.signer = APIRequestSigner('test-key', 'test-secret')

    def test_signature_consistency(self):
        # Same request should produce same signature
        headers1 = self.signer.sign_request('GET', 'https://api.test.com/data')
        headers2 = self.signer.sign_request('GET', 'https://api.test.com/data')

        # Remove timestamp for comparison
        sig1 = headers1['Authorization']
        sig2 = headers2['Authorization']

        # Signatures should be different due to timestamp
        self.assertNotEqual(sig1, sig2)

    def test_signature_with_body(self):
        body = '{"test": "data"}'
        headers = self.signer.sign_request('POST', 'https://api.test.com/create', body=body)

        self.assertIn('Authorization', headers)
        self.assertIn('X-Timestamp', headers)

if __name__ == '__main__':
    unittest.main()

Debugging Signed Requests

When implementing request signing, debugging can be challenging. Consider when monitoring network requests in Puppeteer or working with complex authentication flows, having proper logging is essential:

import logging

def debug_signature_creation(self, method, url, headers, body):
    logger = logging.getLogger(__name__)

    logger.debug(f"Signing request: {method} {url}")
    logger.debug(f"Headers: {headers}")
    logger.debug(f"Body hash: {hashlib.sha256((body or '').encode()).hexdigest()}")

    string_to_sign = self._create_string_to_sign(method, url, headers, body, timestamp)
    logger.debug(f"String to sign: {repr(string_to_sign)}")

    return string_to_sign

Conclusion

HTTP request signing is essential for securing API communications. Whether you choose HMAC-SHA256 for simplicity, AWS Signature V4 for AWS services, or JWT for stateless authentication, the key is consistent implementation and proper security practices. Remember to always use HTTPS, include timestamps, store keys securely, and thoroughly test your implementation.

When implementing complex authentication systems, especially those involving browser automation for testing signed requests, understanding how to handle authentication in Puppeteer becomes crucial for end-to-end testing scenarios.

The examples provided give you a solid foundation for implementing request signing in your applications, but always adapt them to your specific security requirements and compliance needs.