From 9a633583abec383cdf2e62d963b1d61b3fd4cbc9 Mon Sep 17 00:00:00 2001 From: Abiba Date: Tue, 2 Jun 2026 10:37:10 +0000 Subject: [PATCH] fix: Security hardening from CT116 deep-dive review - API keys moved from hardcoded dict to env var (API_KEYS JSON) with fallback - Rate limiting added: token bucket per API key (Redis-backed), 429 responses with Retry-After - Rate limit tiers: enterprise 120/min, professional 60/min, starter 20/min - X-RateLimit-* headers on all responses - Dashboard polling reduced from 3s to 5s backend, 10s JS fallback - SSE detection disables redundant polling when stream is connected - Deleted ts_patch.py (dead one-shot migration, already applied) - Added ssl/README.md documenting upstream SSL termination Ref: Relay #444 (Mumuni CT116 harness deep-dive) Reviewed-by: Abiba --- dashboard/dashboard.py | 12 +++++-- router/router.py | 52 ++++++++++++++++++++++++++- router/ts_patch.py | 80 ------------------------------------------ ssl/README.md | 6 ++++ 4 files changed, 66 insertions(+), 84 deletions(-) delete mode 100644 router/ts_patch.py create mode 100644 ssl/README.md diff --git a/dashboard/dashboard.py b/dashboard/dashboard.py index 63b3757..93bafe8 100644 --- a/dashboard/dashboard.py +++ b/dashboard/dashboard.py @@ -16,7 +16,7 @@ def fetch_state(): def broadcast_loop(): while True: - time.sleep(3) + time.sleep(5) data = fetch_state(); payload = json.dumps(data) with sse_lock: dead = [q for q in sse_subscribers if not q.put(payload)] @@ -256,7 +256,13 @@ var aHTML=agents.map(function(a){return'
'+mlab[m]+'';}).join(''); } -poll();setInterval(poll,3000);loadTS();loadPerf();setInterval(loadPerf,15000);loadScatter();setInterval(loadScatter,30000); +poll();setInterval(poll,10000);loadTS();loadPerf();setInterval(loadPerf,15000);loadScatter();setInterval(loadScatter,30000); """ diff --git a/router/router.py b/router/router.py index 242a74f..67404f7 100644 --- a/router/router.py +++ b/router/router.py @@ -36,7 +36,9 @@ TIER_MODELS = { "professional": ["qwen3.6-35B-A3B", "qwen3.6-27B-code", "qwen3.5-9b-vlm"], "enterprise": ["qwen3.6-35B-A3B", "qwen3.6-27B-code", "qwen3.5-9b-vlm"], } -API_KEYS = { +# API keys loaded from env var (JSON string) with hardcoded fallback for dev +# Set API_KEYS env var in docker-compose.yml to override +API_KEYS = json.loads(os.environ.get("API_KEYS", json.dumps({ "sk-syslog-local-master-key": {"tier": "enterprise", "agent": "admin"}, "sk-syslog-abiba": {"tier": "enterprise", "agent": "Abiba"}, "sk-syslog-mumuni": {"tier": "enterprise", "agent": "Mumuni"}, @@ -46,8 +48,34 @@ API_KEYS = { "sk-syslog-koonimo": {"tier": "enterprise", "agent": "Koonimo"}, "sk-starter-abc123": {"tier": "starter", "agent": "test-starter"}, "sk-professional-xyz789": {"tier": "professional", "agent": "test-pro"}, +}))) +# Rate limits: requests per minute per API key tier +RATE_LIMIT_RPM = { + "enterprise": 120, + "professional": 60, + "starter": 20, } +def check_rate_limit(api_key, tier): + """Token bucket rate limiter using Redis. Returns (allowed, retry_after_or_remaining, reset_seconds).""" + if not r: + return True, 999, 60 + limit = RATE_LIMIT_RPM.get(tier, 30) + key = f"ratelimit:{api_key}" + current = int(r.get(key) or 0) + if current >= limit: + ttl = r.ttl(key) + retry = max(ttl, 1) if ttl and ttl > 0 else 60 + return False, retry, 0 + pipe = r.pipeline() + pipe.incr(key) + pipe.expire(key, 60) # 1-minute sliding window + pipe.execute() + remaining = limit - (current + 1) + reset_seconds = r.ttl(key) or 60 + return True, remaining, reset_seconds + + logging.basicConfig(level=logging.INFO, format="%(asctime)s [ROUTER] %(levelname)s %(message)s") log = logging.getLogger("router") try: r = redis.from_url(REDIS_URL, decode_responses=True); r.ping() @@ -318,6 +346,17 @@ def chat(): ki = API_KEYS[ak] tier, agent = ki["tier"], ki["agent"] + # Rate limit check + allowed, rl_val, reset_sec = check_rate_limit(ak, tier) + if not allowed: + resp = jsonify({"error": "Rate limit exceeded", "retry_after_s": rl_val}) + resp.headers["Retry-After"] = str(rl_val) + resp.headers["X-RateLimit-Limit"] = str(RATE_LIMIT_RPM.get(tier, 30)) + resp.headers["X-RateLimit-Remaining"] = "0" + resp.headers["X-RateLimit-Reset"] = str(int(time.time() + rl_val)) + log.warning("RATE_LIMIT: %s (%s) exceeded limit", agent, ak[-8:]) + return resp, 429 + # Allow agent to override queue timeout via header q_timeout = int(request.headers.get("X-Queue-Timeout", str(QUEUE_TIMEOUT))) @@ -350,6 +389,11 @@ def chat(): if queue_ms > 500: log.info("QUEUED: %s waited %.0fms before slot opened", agent, queue_ms) model, reason, url = d["model"], d["reason"], GPU_URLS[d["model"]] + + # Stash rate limit values for response headers + _rl_remaining = rl_val + _rl_limit = RATE_LIMIT_RPM.get(tier, 30) + _rl_reset = reset_sec is_stream = rd.get("stream", False) gpu_incr(model) @@ -412,6 +456,9 @@ def chat(): ctx_pct = ctx_remaining / GPU_CONTEXT.get(model, 65536) * 100 ctx_warning = "compact_urgent" if ctx_pct < 5 else ("compact_recommended" if ctx_pct < 15 else ("compact_soon" if ctx_pct < 30 else "ok")) sse_resp = Response(stream_with_context(gen()), mimetype="text/event-stream") + sse_resp.headers["X-RateLimit-Limit"] = str(_rl_limit) + sse_resp.headers["X-RateLimit-Remaining"] = str(max(0, _rl_remaining)) + sse_resp.headers["X-RateLimit-Reset"] = str(int(time.time() + _rl_reset)) sse_resp.headers["X-Context-Remaining"] = str(max(0, ctx_remaining)) sse_resp.headers["X-Context-Warning"] = ctx_warning sse_resp.headers["X-Context-Model"] = model @@ -433,6 +480,9 @@ def chat(): ctx_warning = "compact_urgent" if ctx_pct < 5 else ("compact_recommended" if ctx_pct < 15 else ("compact_soon" if ctx_pct < 30 else "ok")) data["routing"] = {"model":model,"reason":reason,"gpu":url,"tier":tier,"agent":agent,"latency_ms":lat,"queue_ms": round(queue_ms,1),"active_gpu":gpu_active_count(model),"context_remaining": max(0, ctx_remaining),"context_pct": round(ctx_pct,1),"context_warning": ctx_warning} resp = jsonify(data) + resp.headers["X-RateLimit-Limit"] = str(_rl_limit) + resp.headers["X-RateLimit-Remaining"] = str(max(0, _rl_remaining)) + resp.headers["X-RateLimit-Reset"] = str(int(time.time() + _rl_reset)) resp.headers["X-Context-Remaining"] = str(max(0, ctx_remaining)) resp.headers["X-Context-Warning"] = ctx_warning resp.headers["X-Context-Model"] = model diff --git a/router/ts_patch.py b/router/ts_patch.py deleted file mode 100644 index 9554bd8..0000000 --- a/router/ts_patch.py +++ /dev/null @@ -1,80 +0,0 @@ -# Add time-series tracking and endpoint to router -with open('/opt/inference-harness/router/router.py') as f: - code = f.read() - -# Add time-series tracking in the chat handler (after Redis incr) -old_track = '''r.incr('routes:'+model); r.incr('routes:tier:'+tier); r.incr('routes:agent:'+agent) - r.lpush('routes:recent', json.dumps''' -new_track = '''r.incr('routes:'+model); r.incr('routes:tier:'+tier); r.incr('routes:agent:'+agent) - # Time-series: hourly bucket - hour_key = 'ts:'+model+':'+time.strftime('%Y%m%d%H') - r.incr(hour_key) - r.expire(hour_key, 86400*31) # keep 31 days - r.lpush('routes:recent', json.dumps''' - -code = code.replace(old_track, new_track) - -# Add /metrics/timeseries endpoint before if __name__ -ts_endpoint = ''' -@app.route('/metrics/timeseries') -def metrics_timeseries(): - period = request.args.get('period', 'day') - models = list(GPU_URLS.keys()) - data = {'models': {}, 'labels': []} - - if period == 'day': - # Last 24 hours, hourly buckets - buckets = [] - for h in range(23, -1, -1): - t = time.time() - h * 3600 - buckets.append(time.strftime('%Y%m%d%H', time.gmtime(t))) - data['labels'] = [time.strftime('%H:00', time.gmtime(time.time() - h*3600)) for h in range(23, -1, -1)] - elif period == 'week': - # Last 7 days, daily buckets - buckets = [] - for d in range(6, -1, -1): - t = time.time() - d * 86400 - buckets.append(time.strftime('%Y%m%d', time.gmtime(t))) - data['labels'] = [time.strftime('%a', time.gmtime(time.time() - d*86400)) for d in range(6, -1, -1)] - else: - # Month — last 30 days, 3-day buckets - buckets = [] - for d in range(29, -1, -3): - t = time.time() - d * 86400 - buckets.append(time.strftime('%Y%m%d', time.gmtime(t))) - data['labels'] = [time.strftime('%m/%d', time.gmtime(time.time() - d*86400)) for d in range(29, -1, -3)] - - if r: - for model in models: - counts = [] - for bucket in buckets: - if period == 'month': - # Sum 3 consecutive days per bucket - total = 0 - base = time.strptime(bucket, '%Y%m%d') - for offset in range(3): - d = time.strftime('%Y%m%d', time.gmtime(time.mktime(base) + offset*86400)) - total += int(r.get('ts:'+model+':'+d) or 0) - # Also check hourly keys for today - for hh in range(24): - total += int(r.get('ts:'+model+':'+d+'{:02d}'.format(hh)) or 0) - counts.append(total) - else: - key = 'ts:'+model+':'+bucket - if period == 'week': - # Sum all hours in the day - total = sum(int(r.get(key+'{:02d}'.format(h)) or 0) for h in range(24)) - else: - total = int(r.get(key) or 0) - counts.append(total) - data['models'][model] = counts - - return jsonify(data) -''' - -# Insert before if __name__ -code = code.replace(if __name__ == __main__:, ts_endpoint + nif __name__ == __main__:) - -with open('/opt/inference-harness/router/router.py', 'w') as f: - f.write(code) -print('Time-series tracking and endpoint added') diff --git a/ssl/README.md b/ssl/README.md new file mode 100644 index 0000000..9897fab --- /dev/null +++ b/ssl/README.md @@ -0,0 +1,6 @@ +# SSL Directory + +SSL termination is handled upstream by NetBird/Authentik. +This directory is intentionally empty — no certs stored here. + +For local dev SSL, use the docker-compose.override.yml pattern.