#!/usr/bin/env python3 """Local, synthetic HTTP rate-limit exercise. Python 3.10+ standard library only. SPDX-License-Identifier: MIT Not a production server or a distributed limiter. No remote target is accepted. """ from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone import hashlib from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json from pathlib import Path import platform import secrets import sys import threading from urllib.error import HTTPError from urllib.request import build_opener, ProxyHandler, Request def main(): if len(sys.argv) != 1: raise SystemExit('Usage: python3 run.py > results.json; local fixture only') checks = [] lock = threading.Lock() # The test controls time directly; the HTTP API cannot change it. state = {'now': 0, 'counts': {}, 'work': 0} identities = { 'alice': ('alpha', 'alice', True), 'anna': ('alpha', 'anna', True), 'bob': ('beta', 'bob', True), 'partner': ('alpha', 'partner', False), } tokens = {who: secrets.token_urlsafe(24) for who in identities} principals = {tokens[who]: value for who, value in identities.items()} class Handler(BaseHTTPRequestHandler): def log_message(self, *args): pass def reply(self, status, code, retry=None): payload = json.dumps({'code': code}).encode() self.send_response(status) self.send_header('Content-Type', 'application/json') self.send_header('Cache-Control', 'no-store') self.send_header('Content-Length', str(len(payload))) if retry is not None: self.send_header('Retry-After', str(retry)) self.end_headers() self.wfile.write(payload) def do_POST(self): if self.path != '/export': return self.reply(404, 'not_found') principal = principals.get(self.headers.get('Authorization', '').removeprefix('Bearer ')) if principal is None: return self.reply(401, 'authentication_required') tenant, user, can_export = principal if not can_export: return self.reply(403, 'export_forbidden') # Deliberately bounded HTTP fixture, not a general HTTP parser. if self.headers.get('Transfer-Encoding'): return self.reply(400, 'unsupported_encoding') lengths = self.headers.get_all('Content-Length', []) if len(lengths) != 1 or not lengths[0].isdigit(): return self.reply(400, 'invalid_length') length = int(lengths[0]) if length > 1024: return self.reply(413, 'payload_too_large') try: body = json.loads(self.rfile.read(length)) except (ValueError, UnicodeError): return self.reply(400, 'invalid_json') if not isinstance(body, dict) or set(body) != {'records'}: return self.reply(400, 'invalid_input') if type(body['records']) is not int or not 1 <= body['records'] <= 100: return self.reply(400, 'invalid_record_limit') # Both quotas are checked and charged in one critical section. # Tenant/user come from the trusted fixture session, never from a request header. with lock: window = state['now'] // 60 keys = [('user', tenant, user, window), ('tenant', tenant, window)] if any(state['counts'].get(key, 0) >= cap for key, cap in zip(keys, (3, 5))): retry = 60 - state['now'] % 60 status, code = 429, 'quota_exceeded' else: for key in keys: state['counts'][key] = state['counts'].get(key, 0) + 1 state['work'] += 1 # Stands in for one accepted expensive operation. status, code, retry = 200, 'export_accepted', None self.reply(status, code, retry) class Server(ThreadingHTTPServer): request_queue_size = 32 daemon_threads = True server = Server(('127.0.0.1', 0), Handler) base = f'http://127.0.0.1:{server.server_port}' thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() def request(who=None, payload=None, extra=None): headers = {'Content-Type': 'application/json', **(extra or {})} if who: headers['Authorization'] = 'Bearer ' + tokens[who] body = b'{"records":10}' if payload is None else payload req = Request(base + '/export', data=body, headers=headers, method='POST') opener = build_opener(ProxyHandler({})) try: response = opener.open(req, timeout=10) except HTTPError as error: response = error with response: return response.status, dict(response.headers), json.loads(response.read()) def expect(label, result, status, code): actual, headers, body = result if (actual, body.get('code')) != (status, code): raise RuntimeError(f'{label}: unexpected status/body {actual}/{body}') if headers.get('Cache-Control') != 'no-store': raise RuntimeError(f'{label}: response can be cached') if status == 429 and headers.get('Retry-After') != str(60 - state['now'] % 60): raise RuntimeError(f'{label}: incorrect Retry-After') checks.append({'case': label, 'status': actual, 'code': code, 'passed': True}) try: expect('anonymous rejected', request(), 401, 'authentication_required') expect('partner cannot export', request('partner'), 403, 'export_forbidden') expect('payload bounded before work', request('alice', b' ' * 1025), 413, 'payload_too_large') expect('batch size bounded', request('alice', b'{"records":101}'), 400, 'invalid_record_limit') expect('boolean is not a record count', request('alice', b'{"records":true}'), 400, 'invalid_record_limit') expect('tenant cannot be supplied in body', request('alice', b'{"records":1,"tenant":"beta"}'), 400, 'invalid_input') for index in range(3): expect(f'alice allowed {index + 1}', request('alice'), 200, 'export_accepted') expect('alice user limit', request('alice'), 429, 'quota_exceeded') expect('spoofed tenant and IP do not bypass quota', request('alice', extra={ 'X-Tenant': 'beta', 'X-Forwarded-For': '192.0.2.99'}), 429, 'quota_exceeded') for index in range(2): expect(f'anna shared tenant budget {index + 1}', request('anna'), 200, 'export_accepted') expect('alpha tenant limit', request('anna'), 429, 'quota_exceeded') expect('beta remains available', request('bob'), 200, 'export_accepted') state['now'] = 15 expect('retry delay falls with controlled clock', request('alice'), 429, 'quota_exceeded') if state['work'] != 6: raise RuntimeError('Rejected requests executed expensive work') checks.append({'case': 'only six accepted operations executed', 'passed': True}) state['now'] = 60 expect('new window accepts again', request('alice'), 200, 'export_accepted') state['now'] = 120 before = state['work'] barrier = threading.Barrier(10) def simultaneous(_): barrier.wait(timeout=10) return request('alice') with ThreadPoolExecutor(max_workers=10) as pool: results = list(pool.map(simultaneous, range(10))) statuses = sorted(result[0] for result in results) if statuses != [200] * 3 + [429] * 7 or state['work'] - before != 3: raise RuntimeError('Concurrent quota exceeded or rejection executed work') checks.append({'case': 'ten simultaneous requests: exactly three accepted', 'statuses': statuses, 'operations_executed': 3, 'passed': True}) print(json.dumps({ 'recorded_at': datetime.now(timezone.utc).isoformat(), 'python': platform.python_version(), 'scope': 'Local HTTP fixture, one process, synthetic sessions, controlled clock. No production auth, distributed counter or load certification.', 'policy': {'window_seconds': 60, 'user_limit': 3, 'tenant_limit': 5, 'max_payload_bytes': 1024, 'max_records': 100}, 'source_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), 'passed': len(checks), 'checks': checks, }, indent=2)) finally: server.shutdown() server.server_close() thread.join(timeout=10) if __name__ == '__main__': main()