""" proxy_server.py Shared proxy implementation for the Status Checker tools. This is the same forwarding logic that was in launch.py / nectar_launch.py, factored out so local dev and the production systemd service run identical code, plus three additions: 1. API-key auth — every /proxy and /test call must send a valid X-Api-Key header (see apikeys.py). /health stays open for liveness checks. 2. Per-key rate limiting — sliding 60s window, default 60 req/min/key. 3. Audit logging — one JSON line per request to STATUS_CHECKER_AUDIT_LOG (who, what host, what transaction id, when, upstream result). Bind address, port, allowed hosts, rate limit, and log path are all overridable via environment variables so the same code runs unchanged locally (127.0.0.1) and in production (0.0.0.0 behind nginx). """ import json import os import time from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from threading import Lock from urllib.error import HTTPError from urllib.parse import urlparse, parse_qs from urllib.request import Request, urlopen import apikeys ALLOWED_HOSTS = [ 'api.kiwifinance.tech', 'api.nectargateway.com', 'api.westrapay.com', 'api.ganairepay.com', 'api.payswitch.finance', ] RATE_LIMIT_PER_MINUTE = int(os.environ.get('STATUS_CHECKER_RATE_LIMIT', '60')) AUDIT_LOG_PATH = os.environ.get( 'STATUS_CHECKER_AUDIT_LOG', os.path.join(os.path.dirname(os.path.abspath(__file__)), 'audit.log') ) _rate_lock = Lock() _rate_state = {} # label -> list[timestamps] def check_rate_limit(label): now = time.time() with _rate_lock: window = _rate_state.setdefault(label, []) cutoff = now - 60 while window and window[0] < cutoff: window.pop(0) if len(window) >= RATE_LIMIT_PER_MINUTE: return False window.append(now) return True _audit_lock = Lock() def audit_log(event): event['ts'] = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%fZ') line = json.dumps(event, separators=(',', ':')) try: os.makedirs(os.path.dirname(AUDIT_LOG_PATH) or '.', exist_ok=True) with _audit_lock: with open(AUDIT_LOG_PATH, 'a') as f: f.write(line + '\n') except OSError: pass print(line, flush=True) class ProxyHandler(BaseHTTPRequestHandler): protocol_version = 'HTTP/1.0' # deliberate: avoids keep-alive stacking bug def log_message(self, fmt, *a): print(f' {fmt % a}') def send_cors(self): self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type, X-Api-Key') self.send_header('Connection', 'close') def do_OPTIONS(self): self.send_response(200) self.send_cors() self.end_headers() def _json(self, code, obj): body = json.dumps(obj).encode() self.send_response(code) self.send_cors() self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) def _client_ip(self): return self.client_address[0] def _authenticate(self): """Returns label on success. Sends the 401 response itself and returns None on failure.""" raw_key = self.headers.get('X-Api-Key') label = apikeys.validate_key(raw_key) if not label: audit_log({'event': 'auth_failed', 'source_ip': self._client_ip(), 'path': self.path}) self._json(401, {'error': 'unauthorized', 'detail': 'missing or invalid X-Api-Key header'}) return None return label def do_GET(self): if self.path == '/health': return self._json(200, {'status': 'ok'}) if self.path == '/test': label = self._authenticate() if not label: return return self._json(200, {'status': 'ok', 'authenticated_as': label}) if not self.path.startswith('/proxy?'): return self._json(400, {'error': 'Use /proxy?url=TARGET'}) label = self._authenticate() if not label: return if not check_rate_limit(label): audit_log({'event': 'rate_limited', 'api_key_label': label, 'source_ip': self._client_ip()}) return self._json(429, {'error': 'rate_limited', 'detail': f'max {RATE_LIMIT_PER_MINUTE} requests/min'}) qs = parse_qs(urlparse(self.path).query) target = qs.get('url', [None])[0] if not target: return self._json(400, {'error': 'Missing url param'}) host = urlparse(target).hostname if host not in ALLOWED_HOSTS: audit_log({'event': 'host_rejected', 'api_key_label': label, 'source_ip': self._client_ip(), 'host': host}) return self._json(403, {'error': f'Host not allowed: {host}'}) txn_id = urlparse(target).path.rsplit('/', 1)[-1] print(f' → GET {target} (key: {label})') try: req = Request(target, headers={'User-Agent': 'NectarStatusChecker/1.0'}) with urlopen(req, timeout=15) as resp: body = resp.read() status = resp.status ct = resp.headers.get('Content-Type', 'application/json') self.send_response(status) self.send_cors() self.send_header('Content-Type', ct) self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) print(f' ✓ {status} ({len(body)}b)') audit_log({ 'event': 'status_check', 'api_key_label': label, 'source_ip': self._client_ip(), 'host': host, 'transaction_id': txn_id, 'upstream_status': status, }) except HTTPError as e: body = e.read() self.send_response(e.code) self.send_cors() self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) print(f' ✗ HTTP {e.code}') audit_log({ 'event': 'status_check', 'api_key_label': label, 'source_ip': self._client_ip(), 'host': host, 'transaction_id': txn_id, 'upstream_status': e.code, }) except Exception as e: self._json(502, {'error': str(e)}) print(f' ✗ {e}') audit_log({ 'event': 'upstream_error', 'api_key_label': label, 'source_ip': self._client_ip(), 'host': host, 'transaction_id': txn_id, 'error': str(e), }) def run(bind='127.0.0.1', port=5050): server = ThreadingHTTPServer((bind, port), ProxyHandler) print(f'Proxy listening on {bind}:{port}') print(f'Keystore: {apikeys.DEFAULT_STORE_PATH}') print(f'Audit log: {AUDIT_LOG_PATH}') try: server.serve_forever() except KeyboardInterrupt: server.shutdown() if __name__ == '__main__': import argparse parser = argparse.ArgumentParser() parser.add_argument('--bind', default=os.environ.get('STATUS_CHECKER_BIND', '127.0.0.1')) parser.add_argument('--port', type=int, default=int(os.environ.get('STATUS_CHECKER_PROXY_PORT', '5050'))) args = parser.parse_args() run(args.bind, args.port)