From fbe297d22679fd8900fa709a334cd2e764f4e8fc Mon Sep 17 00:00:00 2001 From: Nishtha Sharma Date: Thu, 9 Jul 2026 14:19:18 +0530 Subject: [PATCH 1/7] Implement global alert rate limiting mechanism Added a global alert rate limiting mechanism to prevent abuse by rotating session IDs. This includes a new function to check global rate limits and integrates it into the existing alert logging process. --- humane_proxy/escalation/router.py | 82 +++++++++++++++++++++++++++++-- 1 file changed, 78 insertions(+), 4 deletions(-) diff --git a/humane_proxy/escalation/router.py b/humane_proxy/escalation/router.py index 5adc9ea..b2ec058 100644 --- a/humane_proxy/escalation/router.py +++ b/humane_proxy/escalation/router.py @@ -4,12 +4,73 @@ 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") +# --------------------------------------------------------------------------- +# Global alert-rate backstop +# --------------------------------------------------------------------------- +# 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. +# +# This backstop counts alerts *regardless of session_id*, so rotating the +# ID cannot bypass it. It's a ceiling on top of the existing per-session +# limiter, not a replacement for it — both still apply. +# +# In-process only (a sliding window over a deque of timestamps, guarded by +# a lock). For multi-process/multi-worker deployments this ceiling is +# per-process, not global across the fleet; a shared backend (Redis) would +# be needed for a hard global 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. +_global_alert_lock = threading.Lock() +_global_alert_timestamps: deque[float] = deque() + + +def _reset_global_rate_limit() -> None: + """Clear in-process global-limiter state. Test-only helper.""" + with _global_alert_lock: + _global_alert_timestamps.clear() + + +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 # --------------------------------------------------------------------------- @@ -196,7 +257,14 @@ 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) + # Two independent checks: + # 1. Per-session quota (existing) — defeated by rotating session_id. + # 2. Global backstop (new) — cannot be defeated that way, since it + # doesn't key off session_id at all. Short-circuits so a session + # that's already over its own quota never consumes a global slot. + session_limit_ok = check_rate_limit(session_id) + global_limit_ok = _global_rate_limit_allows() if session_limit_ok else True + alerts_allowed = session_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 +289,21 @@ def escalate( } if not alerts_allowed: + reason = ( + "logged_alerts_rate_limited" + if not session_limit_ok + else "logged_alerts_globally_rate_limited" + ) + limit_label = "per-session" if not session_limit_ok else "GLOBAL" logger.warning( - "[RATE-LIMITED] session=%s category=%s risk_score=%.2f — " + "[RATE-LIMITED:%s] session=%s category=%s risk_score=%.2f — " "event logged, alerts suppressed (quota exhausted)", - session_id, category, risk_score, + limit_label, session_id, 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, From c5608fb486c831c39c5854c444c30b4917975ad5 Mon Sep 17 00:00:00 2001 From: Nishtha Sharma Date: Thu, 9 Jul 2026 14:21:53 +0530 Subject: [PATCH 2/7] Add regression tests for global rate limiting Added regression tests for global rate limiting to ensure that rotating session IDs do not bypass alert limits and that global limits function correctly under various conditions. --- tests/test_router.py | 99 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 99 insertions(+) diff --git a/tests/test_router.py b/tests/test_router.py index dbbeeef..c54c235 100644 --- a/tests/test_router.py +++ b/tests/test_router.py @@ -60,6 +60,105 @@ def test_no_webhooks_fired_when_rate_limited(self): mock_fire.assert_not_called() +class TestGlobalRateLimitBackstop: + """Regression: rotating session_id must NOT bypass alert rate limiting. + + 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. + """ + + def _cfg(self, global_max: int, window_s: int = 60): + return { + "escalation": { + "global_rate_limit_max": global_max, + "global_rate_limit_window_seconds": window_s, + "webhooks": {}, + } + } + + def setup_method(self): + from humane_proxy.escalation.router import _reset_global_rate_limit + _reset_global_rate_limit() + + def teardown_method(self): + from humane_proxy.escalation.router import _reset_global_rate_limit + _reset_global_rate_limit() + + 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 TestDbFailure: def test_db_failure_graceful(self): with patch( From 37d0f652c348e0a67f1c29fdfe080363544555a7 Mon Sep 17 00:00:00 2001 From: Nishtha Sharma Date: Thu, 9 Jul 2026 14:25:00 +0530 Subject: [PATCH 3/7] Add global rate limit settings to config.yaml Added global rate limit configuration for alerts. --- humane_proxy/config.yaml | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/humane_proxy/config.yaml b/humane_proxy/config.yaml index 2280510..95bed39 100644 --- a/humane_proxy/config.yaml +++ b/humane_proxy/config.yaml @@ -312,6 +312,14 @@ escalation: rate_limit_max: 3 rate_limit_window_hours: 1 + # Global ALERT rate-limit backstop, independent of session_id. + # session_id is caller-supplied and unauthenticated, so the per-session + # quota above can be bypassed by rotating session_id on every request. + # This ceiling counts alerts across ALL sessions and cannot be bypassed + # that way. 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 +339,3 @@ escalation: from: "" # List of recipient addresses to: [] - From 925a5b90d0273846cfe0e91f32f8833ce254ee04 Mon Sep 17 00:00:00 2001 From: Nishtha Sharma Date: Mon, 13 Jul 2026 08:26:56 +0530 Subject: [PATCH 4/7] Add client IP retrieval function and update session ID logic --- humane_proxy/middleware/interceptor.py | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) 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 + ) From 47f16e2ff36fe87dd95e537855558b489f2a48fe Mon Sep 17 00:00:00 2001 From: Nishtha Sharma Date: Mon, 13 Jul 2026 08:27:28 +0530 Subject: [PATCH 5/7] Enhance alert rate limiting with IP and global checks Refactor alert rate limiting logic to include per-IP and global ceiling checks. Update comments for clarity on rate limiting mechanisms. --- humane_proxy/escalation/router.py | 125 ++++++++++++++++++++++++------ 1 file changed, 100 insertions(+), 25 deletions(-) diff --git a/humane_proxy/escalation/router.py b/humane_proxy/escalation/router.py index b2ec058..566c2c6 100644 --- a/humane_proxy/escalation/router.py +++ b/humane_proxy/escalation/router.py @@ -14,7 +14,7 @@ logger = logging.getLogger("humane_proxy.escalation") # --------------------------------------------------------------------------- -# Global alert-rate backstop +# 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 @@ -25,19 +25,34 @@ # alerts (Slack/Discord/Teams/PagerDuty/email) even though each individual # session never exceeds its own limit. # -# This backstop counts alerts *regardless of session_id*, so rotating the -# ID cannot bypass it. It's a ceiling on top of the existing per-session -# limiter, not a replacement for it — both still apply. +# Two backstops sit on top of the per-session check, evaluated in order: # -# In-process only (a sliding window over a deque of timestamps, guarded by -# a lock). For multi-process/multi-worker deployments this ceiling is -# per-process, not global across the fleet; a shared backend (Redis) would -# be needed for a hard global 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. +# 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.""" @@ -45,6 +60,49 @@ def _reset_global_rate_limit() -> None: _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. @@ -71,6 +129,7 @@ def _global_rate_limit_allows() -> bool: _global_alert_timestamps.append(now) return True + # --------------------------------------------------------------------------- # International crisis resource database # --------------------------------------------------------------------------- @@ -217,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. @@ -224,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). @@ -245,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 ------- @@ -257,14 +323,20 @@ def escalate( # --- Alert rate-limit check (before logging so the quota is counted # against events already persisted in the window) --- - # Two independent checks: + # 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. Global backstop (new) — cannot be defeated that way, since it - # doesn't key off session_id at all. Short-circuits so a session - # that's already over its own quota never consumes a global slot. + # 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) - global_limit_ok = _global_rate_limit_allows() if session_limit_ok else True - alerts_allowed = session_limit_ok and global_limit_ok + 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 @@ -289,16 +361,19 @@ def escalate( } if not alerts_allowed: - reason = ( - "logged_alerts_rate_limited" - if not session_limit_ok - else "logged_alerts_globally_rate_limited" - ) - limit_label = "per-session" if not session_limit_ok else "GLOBAL" + 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:%s] 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)", - limit_label, session_id, category, risk_score, + limit_label, session_id, client_ip or "unknown", category, risk_score, ) return { "escalated": True, From 86dedcbf62bb99999b5a03f08db0bbeddee2d2c8 Mon Sep 17 00:00:00 2001 From: Nishtha Sharma Date: Mon, 13 Jul 2026 08:28:11 +0530 Subject: [PATCH 6/7] Refactor tests for rate limiting with pytest fixtures Added pytest fixtures to reset rate limiters between tests and refactored test methods to use the new fixture. Improved test coverage for rate limiting behavior with respect to session IDs and client IPs. --- tests/test_router.py | 146 ++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 137 insertions(+), 9 deletions(-) diff --git a/tests/test_router.py b/tests/test_router.py index c54c235..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") @@ -61,31 +79,28 @@ def test_no_webhooks_fired_when_rate_limited(self): class TestGlobalRateLimitBackstop: - """Regression: rotating session_id must NOT bypass alert rate limiting. + """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 setup_method(self): - from humane_proxy.escalation.router import _reset_global_rate_limit - _reset_global_rate_limit() - - def teardown_method(self): - from humane_proxy.escalation.router import _reset_global_rate_limit - _reset_global_rate_limit() - def test_rotating_session_id_no_longer_bypasses_rate_limit(self): from humane_proxy.escalation import router as router_mod @@ -159,6 +174,119 @@ def test_session_limited_event_does_not_consume_global_slot(self): 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( From 2e97429946644f23463c84d6a6a95c91c82582a6 Mon Sep 17 00:00:00 2001 From: Nishtha Sharma Date: Mon, 13 Jul 2026 08:28:36 +0530 Subject: [PATCH 7/7] Implement per-IP alert rate limiting Added per-IP alert rate-limiting configuration. --- humane_proxy/config.yaml | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/humane_proxy/config.yaml b/humane_proxy/config.yaml index 95bed39..c2bd3de 100644 --- a/humane_proxy/config.yaml +++ b/humane_proxy/config.yaml @@ -312,11 +312,19 @@ escalation: rate_limit_max: 3 rate_limit_window_hours: 1 - # Global ALERT rate-limit backstop, independent of session_id. - # session_id is caller-supplied and unauthenticated, so the per-session - # quota above can be bypassed by rotating session_id on every request. - # This ceiling counts alerts across ALL sessions and cannot be bypassed - # that way. Set global_rate_limit_max to 0 to disable it. + # 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