diff --git a/humane_proxy/config.yaml b/humane_proxy/config.yaml index 2280510..c2bd3de 100644 --- a/humane_proxy/config.yaml +++ b/humane_proxy/config.yaml @@ -312,6 +312,22 @@ escalation: rate_limit_max: 3 rate_limit_window_hours: 1 + # Per-IP ALERT rate-limit layer, checked on every request. session_id is + # caller-supplied and unauthenticated, so the per-session quota above can + # be bypassed by rotating session_id on every request. This layer is + # keyed on the real TCP peer address instead (not attacker-controlled), + # so rotating session_id no longer helps. Set ip_rate_limit_max to 0 to + # disable it. + ip_rate_limit_max: 10 + ip_rate_limit_window_seconds: 60 + + # Global ALERT rate-limit — outer ceiling, last line of defense. Keyed on + # nothing (counts every alert regardless of session_id or IP), so it also + # catches the distributed case (many real IPs) the per-IP layer alone + # can't stop. Set global_rate_limit_max to 0 to disable it. + global_rate_limit_max: 100 + global_rate_limit_window_seconds: 60 + webhooks: slack_url: "" discord_url: "" @@ -331,4 +347,3 @@ escalation: from: "" # List of recipient addresses to: [] - diff --git a/humane_proxy/escalation/router.py b/humane_proxy/escalation/router.py index 5adc9ea..566c2c6 100644 --- a/humane_proxy/escalation/router.py +++ b/humane_proxy/escalation/router.py @@ -4,12 +4,132 @@ import asyncio import logging +import threading +import time +from collections import deque from humane_proxy.config import get_config from humane_proxy.escalation.local_db import check_rate_limit, log_escalation logger = logging.getLogger("humane_proxy.escalation") +# --------------------------------------------------------------------------- +# Alert-rate backstops: per-IP layer + global outer ceiling +# --------------------------------------------------------------------------- +# The per-session limiter in `check_rate_limit()` is keyed entirely on +# `session_id`, which is caller-supplied and unauthenticated (see +# middleware/interceptor.py — it's read straight off the request body with +# no validation). That means the per-session quota can be trivially +# defeated by rotating `session_id` on every request: each "new" session +# gets its own fresh quota, so an attacker can trigger unlimited operator +# alerts (Slack/Discord/Teams/PagerDuty/email) even though each individual +# session never exceeds its own limit. +# +# Two backstops sit on top of the per-session check, evaluated in order: +# +# 1. Per-IP layer (checked on every request). Keyed on `request.client.host` +# — the raw TCP peer address, NOT any caller-supplied header (X-Forwarded-For +# etc. are trivially spoofable and are intentionally not trusted here). +# Rotating session_id doesn't help an attacker anymore, since a single +# real network origin still has one IP-scoped quota. +# +# 2. Global outer ceiling (last line of defense). Keyed on nothing — +# counts every alert regardless of session_id or IP. Catches the +# distributed case (many real IPs, e.g. a botnet) where the per-IP +# layer alone wouldn't help, at the cost of being a blunt, shared cap. +# +# Both are in-process (sliding window over a deque of timestamps, guarded +# by a lock). For multi-process/multi-worker deployments each is per-process, +# not shared across the fleet; a Redis-backed version would be needed for a +# hard cap there. Tracked as a possible follow-up — this still closes the +# single-process bypass, which is the exploitable case for the default +# (non-Redis) deployment most users run. + +_MAX_TRACKED_IPS = 10_000 # bound memory: cap distinct IP buckets tracked + +_global_alert_lock = threading.Lock() +_global_alert_timestamps: deque[float] = deque() + +_ip_alert_lock = threading.Lock() +_ip_alert_timestamps: dict[str, deque[float]] = {} + + +def _reset_global_rate_limit() -> None: + """Clear in-process global-limiter state. Test-only helper.""" + with _global_alert_lock: + _global_alert_timestamps.clear() + + +def _reset_ip_rate_limit() -> None: + """Clear in-process IP-limiter state. Test-only helper.""" + with _ip_alert_lock: + _ip_alert_timestamps.clear() + + +def _ip_rate_limit_allows(client_ip: str | None) -> bool: + """Return True if firing another alert stays within this IP's quota. + + Config keys (under ``escalation:``): + - ``ip_rate_limit_max`` (default 10) + - ``ip_rate_limit_window_seconds`` (default 60) + + Set ``ip_rate_limit_max`` to ``0`` to disable this layer. + """ + cfg = get_config() + esc_cfg = cfg.get("escalation", {}) or {} + max_alerts = esc_cfg.get("ip_rate_limit_max", 10) + window_s = esc_cfg.get("ip_rate_limit_window_seconds", 60) + + if not max_alerts or max_alerts <= 0: + return True # layer disabled + + client_ip = client_ip or "unknown" + now = time.monotonic() + with _ip_alert_lock: + bucket = _ip_alert_timestamps.get(client_ip) + if bucket is None: + if len(_ip_alert_timestamps) >= _MAX_TRACKED_IPS: + # Bound memory under sustained traffic from many distinct + # IPs: evict the oldest-inserted bucket to make room. + _ip_alert_timestamps.pop(next(iter(_ip_alert_timestamps))) + bucket = deque() + _ip_alert_timestamps[client_ip] = bucket + + while bucket and now - bucket[0] > window_s: + bucket.popleft() + if len(bucket) >= max_alerts: + return False + bucket.append(now) + return True + + +def _global_rate_limit_allows() -> bool: + """Return True if firing another alert stays within the global ceiling. + + Config keys (under ``escalation:``): + - ``global_rate_limit_max`` (default 100) + - ``global_rate_limit_window_seconds`` (default 60) + + Set ``global_rate_limit_max`` to ``0`` to disable this backstop. + """ + cfg = get_config() + esc_cfg = cfg.get("escalation", {}) or {} + max_alerts = esc_cfg.get("global_rate_limit_max", 100) + window_s = esc_cfg.get("global_rate_limit_window_seconds", 60) + + if not max_alerts or max_alerts <= 0: + return True # backstop disabled + + now = time.monotonic() + with _global_alert_lock: + while _global_alert_timestamps and now - _global_alert_timestamps[0] > window_s: + _global_alert_timestamps.popleft() + if len(_global_alert_timestamps) >= max_alerts: + return False + _global_alert_timestamps.append(now) + return True + + # --------------------------------------------------------------------------- # International crisis resource database # --------------------------------------------------------------------------- @@ -156,6 +276,7 @@ def escalate( message_hash: str | None = None, stage_reached: int = 1, reasoning: str | None = None, + client_ip: str | None = None, ) -> dict: """Handle a flagged interaction. @@ -163,7 +284,8 @@ def escalate( ---- 1. Persist the event to the audit log — always. The audit trail must be complete; only *alerting* is rate limited. - 2. Check the per-session alert rate limit. + 2. Check the per-session alert rate limit, then the per-IP layer, then + the global outer ceiling. 3. If within quota → emit a CRITICAL log and fire webhooks. 4. Return a result dict indicating the outcome (``alerted`` says whether operator notifications went out). @@ -184,6 +306,11 @@ def escalate( Which pipeline stage (1, 2, or 3) produced the final result. reasoning: Stage-3 reasoning string (if available). + client_ip: + Raw TCP peer address of the caller (``request.client.host``), used + to key the per-IP alert-rate backstop. Optional for backward + compatibility with existing callers; falls back to a shared + ``"unknown"`` bucket when omitted. Returns ------- @@ -196,7 +323,20 @@ def escalate( # --- Alert rate-limit check (before logging so the quota is counted # against events already persisted in the window) --- - alerts_allowed = check_rate_limit(session_id) + # Three independent, ordered checks — each only runs if the previous + # one passed, so a request already blocked by an earlier layer never + # consumes quota from a later one: + # 1. Per-session quota (existing) — defeated by rotating session_id. + # 2. Per-IP layer (new) — keyed on the real TCP peer address, so + # rotating session_id alone can no longer bypass rate limiting. + # 3. Global outer ceiling (existing) — catches the distributed case + # (many real IPs) that the per-IP layer alone can't stop. + session_limit_ok = check_rate_limit(session_id) + ip_limit_ok = _ip_rate_limit_allows(client_ip) if session_limit_ok else True + global_limit_ok = ( + _global_rate_limit_allows() if (session_limit_ok and ip_limit_ok) else True + ) + alerts_allowed = session_limit_ok and ip_limit_ok and global_limit_ok # --- Persist — always, with failure protection. Rate limiting only # applies to operator alerts; suppressing audit records would blind @@ -221,15 +361,24 @@ def escalate( } if not alerts_allowed: + if not session_limit_ok: + reason = "logged_alerts_rate_limited" + limit_label = "session" + elif not ip_limit_ok: + reason = "logged_alerts_ip_rate_limited" + limit_label = "ip" + else: + reason = "logged_alerts_globally_rate_limited" + limit_label = "GLOBAL" logger.warning( - "[RATE-LIMITED] session=%s category=%s risk_score=%.2f — " + "[RATE-LIMITED:%s] session=%s client_ip=%s category=%s risk_score=%.2f — " "event logged, alerts suppressed (quota exhausted)", - session_id, category, risk_score, + limit_label, session_id, client_ip or "unknown", category, risk_score, ) return { "escalated": True, "alerted": False, - "reason": "logged_alerts_rate_limited", + "reason": reason, "session_id": session_id, "category": category, "risk_score": risk_score, diff --git a/humane_proxy/middleware/interceptor.py b/humane_proxy/middleware/interceptor.py index 8fbbfad..4f8283e 100644 --- a/humane_proxy/middleware/interceptor.py +++ b/humane_proxy/middleware/interceptor.py @@ -95,11 +95,20 @@ async def _lifespan(app: FastAPI) -> AsyncGenerator[None, None]: ) +def _client_ip(request: Request) -> str: + """Return the raw TCP peer address for this connection. + + Intentionally does NOT trust X-Forwarded-For / X-Real-IP headers — those + are attacker-controlled unless a trusted reverse proxy is guaranteed to + overwrite them, which this app doesn't assume. ``request.client.host`` + is the actual TCP peer and can't be spoofed via request headers/body. + """ + return request.client.host if request.client else "unknown" + + def _resolve_session_id(payload: dict[str, Any], request: Request) -> str: """Return the session_id from the payload, falling back to the client IP.""" - return payload.get("session_id") or ( - request.client.host if request.client else "unknown" - ) + return payload.get("session_id") or _client_ip(request) def _extract_last_user_message(payload: dict[str, Any]) -> str: @@ -167,6 +176,7 @@ async def chat(request: Request) -> JSONResponse: message_hash=result.message_hash, stage_reached=cls.stage, reasoning=cls.reasoning, + client_ip=_client_ip(request), ) # Self-harm: return care response instead of generic flagged message. @@ -248,4 +258,4 @@ async def chat(request: Request) -> JSONResponse: return _Response( status_code=503, content={"status": "error", "message": "Upstream LLM unavailable."}, - ) \ No newline at end of file + ) diff --git a/tests/test_router.py b/tests/test_router.py index dbbeeef..6180868 100644 --- a/tests/test_router.py +++ b/tests/test_router.py @@ -2,9 +2,27 @@ from unittest.mock import patch +import pytest + from humane_proxy.escalation.router import escalate +@pytest.fixture(autouse=True) +def _reset_alert_backstops(): + """Isolate the in-process IP/global limiter state between tests. + + Both live as module-level counters in router.py, so without this, + tests that don't pass client_ip share a single "unknown" IP bucket and + can bleed into each other's quotas depending on run order. + """ + from humane_proxy.escalation.router import _reset_global_rate_limit, _reset_ip_rate_limit + _reset_global_rate_limit() + _reset_ip_rate_limit() + yield + _reset_global_rate_limit() + _reset_ip_rate_limit() + + class TestEscalation: def test_basic_escalation(self): result = escalate("esc-sess", 0.95, ["trigger1"], "self_harm") @@ -60,6 +78,215 @@ def test_no_webhooks_fired_when_rate_limited(self): mock_fire.assert_not_called() +class TestGlobalRateLimitBackstop: + """Regression: rotating session_id (AND client_ip) must not bypass + alert rate limiting entirely — the global outer ceiling still caps it. + + Previously, check_rate_limit() was keyed only on session_id, which is + caller-supplied and unauthenticated (interceptor.py reads it straight + off the request body with no validation). An attacker could send a + fresh session_id on every request and get unlimited operator alerts. + + ip_rate_limit_max is disabled (0) in these tests to isolate the global + layer specifically — see TestIpRateLimitBackstop for that layer. + """ + + def _cfg(self, global_max: int, window_s: int = 60): + return { + "escalation": { + "ip_rate_limit_max": 0, + "global_rate_limit_max": global_max, + "global_rate_limit_window_seconds": window_s, + "webhooks": {}, + } + } + + def test_rotating_session_id_no_longer_bypasses_rate_limit(self): + from humane_proxy.escalation import router as router_mod + + with patch.object(router_mod, "get_config", return_value=self._cfg(global_max=5)): + results = [ + escalate(f"attacker-session-{i}", 0.95, ["t"], "self_harm") + for i in range(20) + ] + + alerted = [r for r in results if r["alerted"]] + suppressed = [r for r in results if not r["alerted"]] + # Every request used a brand-new session_id (own fresh per-session + # quota each time), yet the global ceiling still caps total alerts. + assert len(alerted) == 5 + assert len(suppressed) == 15 + assert all(r["reason"] == "logged_alerts_globally_rate_limited" for r in suppressed) + # Audit trail is still complete for every event, same guarantee as + # the existing per-session limiter. + assert all(r["escalated"] is True for r in results) + + def test_global_limit_disabled_when_zero(self): + from humane_proxy.escalation import router as router_mod + + with patch.object(router_mod, "get_config", return_value=self._cfg(global_max=0)): + results = [ + escalate(f"disabled-check-session-{i}", 0.95, ["t"], "self_harm") + for i in range(10) + ] + assert all(r["alerted"] is True for r in results) + + def test_global_window_expires_and_allows_more(self): + from humane_proxy.escalation import router as router_mod + + with patch.object(router_mod, "get_config", return_value=self._cfg(global_max=2, window_s=60)): + results = [ + escalate(f"window-session-{i}", 0.95, ["t"], "self_harm") + for i in range(4) + ] + assert sum(1 for r in results if r["alerted"]) == 2 + + # Simulate the window elapsing by backdating recorded timestamps. + with router_mod._global_alert_lock: + router_mod._global_alert_timestamps.clear() + + with patch.object(router_mod, "get_config", return_value=self._cfg(global_max=2, window_s=60)): + result = escalate("window-session-after-reset", 0.95, ["t"], "self_harm") + assert result["alerted"] is True + + def test_session_limited_event_does_not_consume_global_slot(self): + """A request already blocked by its own session quota shouldn't + also burn a global-quota slot — it was never going to alert.""" + from humane_proxy.escalation import router as router_mod + from humane_proxy.storage.factory import get_store + + # The per-session cap lives on the already-instantiated storage + # singleton (set once from real config at process start) — read + # the live value rather than assuming one. + session_cap = get_store()._rate_limit_max + + with patch.object(router_mod, "get_config", return_value=self._cfg(global_max=5)): + sid = "same-session-repeated" + for _ in range(session_cap): + escalate(sid, 0.95, ["t"], "self_harm") # burn its own session quota + for _ in range(10): + result = escalate(sid, 0.95, ["t"], "self_harm") + assert result["reason"] == "logged_alerts_rate_limited" + + # Global quota (5) should be untouched by the 10 session-limited + # calls above — a fresh session can still alert. + fresh = escalate("fresh-session", 0.95, ["t"], "self_harm") + assert fresh["alerted"] is True + + +class TestIpRateLimitBackstop: + """The per-IP layer: keyed on client_ip, checked on every request. + + This is the layer that actually neutralizes session_id rotation from a + single network origin — the global ceiling (above) is the last-resort + catch-all for a *distributed* attacker with many real IPs. + + global_rate_limit_max is set generously high in these tests to isolate + the IP layer specifically. + """ + + def _cfg(self, ip_max: int, window_s: int = 60): + return { + "escalation": { + "ip_rate_limit_max": ip_max, + "ip_rate_limit_window_seconds": window_s, + "global_rate_limit_max": 10_000, + "webhooks": {}, + } + } + + def test_rotating_session_id_same_ip_still_capped(self): + """Same attacker IP, fresh session_id every request — the per-IP + layer catches what the per-session limiter alone would miss.""" + from humane_proxy.escalation import router as router_mod + + with patch.object(router_mod, "get_config", return_value=self._cfg(ip_max=4)): + results = [ + escalate( + f"attacker-session-{i}", 0.95, ["t"], "self_harm", + client_ip="203.0.113.7", + ) + for i in range(10) + ] + + alerted = [r for r in results if r["alerted"]] + suppressed = [r for r in results if not r["alerted"]] + assert len(alerted) == 4 + assert len(suppressed) == 6 + assert all(r["reason"] == "logged_alerts_ip_rate_limited" for r in suppressed) + assert all(r["escalated"] is True for r in results) # audit trail intact + + def test_different_ips_each_get_their_own_quota(self): + """Two distinct real IPs each get their own fresh per-IP quota — + expected/legitimate behavior; the global ceiling is the backstop + for the distributed case, not this layer.""" + from humane_proxy.escalation import router as router_mod + + with patch.object(router_mod, "get_config", return_value=self._cfg(ip_max=2)): + results_a = [ + escalate(f"sess-a-{i}", 0.95, ["t"], "self_harm", client_ip="203.0.113.7") + for i in range(2) + ] + results_b = [ + escalate(f"sess-b-{i}", 0.95, ["t"], "self_harm", client_ip="198.51.100.42") + for i in range(2) + ] + assert all(r["alerted"] for r in results_a) + assert all(r["alerted"] for r in results_b) + + def test_ip_limit_disabled_when_zero(self): + from humane_proxy.escalation import router as router_mod + + with patch.object(router_mod, "get_config", return_value=self._cfg(ip_max=0)): + results = [ + escalate( + f"disabled-check-{i}", 0.95, ["t"], "self_harm", + client_ip="203.0.113.7", + ) + for i in range(15) + ] + assert all(r["alerted"] is True for r in results) + + def test_missing_client_ip_falls_back_to_shared_unknown_bucket(self): + """Backward compatibility: callers that don't pass client_ip (e.g. + any external caller of escalate() written before this change) + still work, sharing one 'unknown' bucket rather than crashing.""" + from humane_proxy.escalation import router as router_mod + + with patch.object(router_mod, "get_config", return_value=self._cfg(ip_max=3)): + results = [ + escalate(f"no-ip-sess-{i}", 0.95, ["t"], "self_harm") + for i in range(5) + ] + assert sum(1 for r in results if r["alerted"]) == 3 + + def test_session_limited_event_does_not_consume_ip_slot(self): + """Mirrors the equivalent global-layer test: a request already + blocked by its own session quota shouldn't burn an IP-quota slot.""" + from humane_proxy.escalation import router as router_mod + from humane_proxy.storage.factory import get_store + + session_cap = get_store()._rate_limit_max + ip = "203.0.113.7" + # IP quota must have headroom beyond session_cap, or burning the + # session's own quota (below) would itself exhaust the IP bucket + # and confound the thing this test is isolating. + ip_max = session_cap + 5 + + with patch.object(router_mod, "get_config", return_value=self._cfg(ip_max=ip_max)): + sid = "same-session-repeated-ip-test" + for _ in range(session_cap): + escalate(sid, 0.95, ["t"], "self_harm", client_ip=ip) + for _ in range(10): + result = escalate(sid, 0.95, ["t"], "self_harm", client_ip=ip) + assert result["reason"] == "logged_alerts_rate_limited" + + # IP quota should be untouched by the 10 session-limited calls + # above — a fresh session from the same IP can still alert. + fresh = escalate("fresh-session-ip-test", 0.95, ["t"], "self_harm", client_ip=ip) + assert fresh["alerted"] is True + + class TestDbFailure: def test_db_failure_graceful(self): with patch(