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 <abiba@sysloggh.com>
This commit is contained in:
Abiba
@@ -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'<div class="mb-2" style="font-size:11px"
|
||||
$('perf-agents').innerHTML=aHTML;
|
||||
}else{$('perf-agents').innerHTML='<div class="text-secondary small text-center py-3">-</div>';}
|
||||
}
|
||||
function poll(){fetch('/api/state').then(function(r){return r.json()}).then(function(data){render(data);$('connection-status').textContent='live';}).catch(function(){$('connection-status').textContent='reconnecting';});}
|
||||
var sseConnected = false;
|
||||
function poll(){if(sseConnected)return;fetch('/api/state').then(function(r){return r.json()}).then(function(data){render(data);$('connection-status').textContent='live';}).catch(function(){$('connection-status').textContent='reconnecting';});}
|
||||
// SSE connection tracking
|
||||
var es = new EventSource('/api/stream');
|
||||
es.onopen = function(){sseConnected = true; $('connection-status').textContent = 'live (SSE)';};
|
||||
es.onmessage = function(e){try{render(JSON.parse(e.data));}catch(ex){}};
|
||||
es.onerror = function(){sseConnected = false; $('connection-status').textContent = 'polling';};
|
||||
function loadScatter(){
|
||||
var m=$('scatter-model').value;
|
||||
fetch('/api/scatter?window=24&model='+m).then(function(r){return r.json()}).then(renderScatter).catch(function(){});
|
||||
@@ -293,7 +299,7 @@ el.innerHTML='<svg viewBox="-35 0 140 115" style="width:100%;height:200px">'+gri
|
||||
var models=[];pts.forEach(function(p){if(models.indexOf(p.model)===-1)models.push(p.model);});
|
||||
lg.innerHTML=models.map(function(m){return'<span class="d-flex align-items-center gap-1 small"><svg width="10" height="10"><circle cx="5" cy="5" r="3.5" fill="'+(mcol[m]||'#38bdf8')+'"/></svg>'+mlab[m]+'</span>';}).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);
|
||||
</script>
|
||||
</body>
|
||||
</html>"""
|
||||
|
||||
+51
-1
@@ -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
|
||||
|
||||
@@ -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')
|
||||
@@ -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.
|
||||
Reference in New Issue
Block a user