AWS SigV4 API Request Flooding

AWS APIs authenticate requests using Signature Version 4 (SigV4), a keyed-HMAC scheme in which the caller signs a canonical request with a key derived from their AWS secret access key, the date, the region, and the service. Because every request must be individually signed, flooding an AWS endpoint is not as simple as replaying captured traffic — each request needs a fresh signature (or at least a fresh timestamp within the 5-minute validity window).

This page describes how to generate signed AWS requests at high volume from Python for load testing, rate-limit validation, and resilience research. Unauthorized flooding of systems you do not own is illegal under the Computer Fraud and Abuse Act and equivalents.

SigV4 signing recap

Per the AWS documentation, the SigV4 signing process has four steps:

  1. Create a canonical request — normalize HTTP method, URI, query string, headers, and payload hash.
  2. Create a string to sign — algorithm, timestamp, credential scope (date/region/service/aws4_request), and hash of the canonical request.
  3. Derive the signing key — successive HMAC-SHA256 rounds:
    kDate    = HMAC("AWS4" + SecretKey, Date)
    kRegion  = HMAC(kDate, Region)
    kService = HMAC(kRegion, Service)
    kSigning = HMAC(kService, "aws4_request")
    
  4. Compute the signatureHMAC(kSigning, string-to-sign), hex-encoded.

The signature goes into the Authorization header or the X-Amz-Signature query parameter.

Python implementation pattern

The requests-auth-aws-sigv4 library (or manual signing) lets Python’s requests library generate signed calls:

import requests
from requests_auth_aws_sigv4 import AWSSigV4
 
response = requests.post(
    url,
    headers={'Content-Type': 'application/json'},
    json=body,
    auth=AWSSigV4('service-name', region='us-west-2'),
    verify=False
)

To flood, wrap this in a threading pool with a synchronization gate so all threads fire simultaneously:

from threading import Thread, Event
from queue import Queue
import math, time
 
gate = Event()
response_codes = Queue()
 
def send_request(request_count, gate, response_codes):
    gate.wait()
    for _ in range(request_count):
        try:
            r = requests.post(url, auth=AWSSigV4(...), verify=False)
            response_codes.put(r.status_code)
        except requests.exceptions.ConnectionError:
            response_codes.put("Dropped")
 
threads = []
for _ in range(num_threads):
    t = Thread(target=send_request, args=(count_per_thread, gate, response_codes))
    t.start()
    threads.append(t)
 
gate.set()  # release all threads at once
for t in threads:
    t.join()

Key engineering details:

  • urllib3.disable_warnings(InsecureRequestWarning) — silences TLS warnings when testing against endpoints with self-signed or internal certificates.
  • Thread gate — ensures threads don’t drift; the flood starts simultaneously.
  • Queue — collects status codes thread-safely for post-test analysis.
  • TPS calculationtotal_requests / elapsed_time measures actual throughput.

Operational context

This technique was developed for testing AWS Directory Service (ds-data) rate limits, but the pattern applies to any SigV4-authenticated API: API Gateway, Lambda invoke URLs, S3, DynamoDB, etc.

AWS services implement rate limiting and throttling (typically returning 429 Too Many Requests or 503). A successful flood test measures:

  • Whether throttling engages
  • How quickly it engages
  • Whether the service degrades gracefully
  • Whether downstream resources are affected

Defenses

  • AWS WAF — rate-based rules on API Gateway / CloudFront
  • Usage plans — API Gateway throttling quotas
  • Service quotas — per-account, per-region request limits
  • Shield — DDoS protection for AWS resources
  • Monitoring — CloudWatch alarms on 4XXError, 5XXError, Latency, Count

See also

Sources