diff --git a/campus/audit/client/v1/traces.py b/campus/audit/client/v1/traces.py index 61d86dab..09c62f14 100644 --- a/campus/audit/client/v1/traces.py +++ b/campus/audit/client/v1/traces.py @@ -10,6 +10,7 @@ /audit/v1/traces/search → Search traces """ +import contextlib from typing import Any, Optional from campus.common.http.interface import JsonClient, JsonResponse @@ -17,6 +18,30 @@ from ..interface import SLASH, Resource, ResourceCollection, ResourceRoot +def _import_ratelimit(): + """Import the SDK breaker module lazily (it ships in campus_python). + + Resolved via importlib with a string name: the module is optional + (a stale campus_python predating #831 lacks it), and opaque + resolution keeps type checkers happy against older SDK installs. + Returns None when unavailable — callers degrade to no observation. + """ + import importlib + + try: + return importlib.import_module("campus_python.audit.ratelimit") + except ImportError: + return None + + +def _safe_json(response: JsonResponse) -> Any: + """Best-effort JSON parse for breaker observation.""" + try: + return response.json() + except Exception: + return None + + class Traces(ResourceCollection): """Resource for trace spans collection. @@ -80,6 +105,18 @@ def new(self, *spans: dict) -> JsonResponse: client.traces.new(span1, span2, span3) """ response = self.client.post(self.make_path(), json={"spans": list(spans)}) + # Feed the ingest circuit breaker before raising (#831): audit + # 429s trip the bucket (Retry-After + tripped key from the + # body), 2xx clears expired trips. Observation is guarded — a + # breaker failure or a stale SDK can never break ingestion. + with contextlib.suppress(Exception): + ratelimit_module = _import_ratelimit() + if ratelimit_module is not None: + ratelimit_module.breaker.observe( + response.status_code, + response.headers, + _safe_json(response), + ) response.raise_for_status() return response diff --git a/campus/audit/middleware/__init__.py b/campus/audit/middleware/__init__.py index d9998570..0b50640c 100644 --- a/campus/audit/middleware/__init__.py +++ b/campus/audit/middleware/__init__.py @@ -27,6 +27,15 @@ def init_app(app: flask.Flask) -> None: """ from . import tracing + @app.before_request + def audit_rate_gate(): + """Fail-closed gate for audit ingest rate limiting (#831). + + Registered before start_span so a gated (503) request never + opens a span. No-op unless AUDIT_TRACING_FAIL_CLOSED=1. + """ + return tracing.check_rate_gate() + @app.before_request def start_span(): """Start a root span for each incoming request. diff --git a/campus/audit/middleware/tracing.py b/campus/audit/middleware/tracing.py index a98e4641..324c2ed9 100644 --- a/campus/audit/middleware/tracing.py +++ b/campus/audit/middleware/tracing.py @@ -221,6 +221,75 @@ def is_static_request() -> bool: return flask.request.path == "/favicon.ico" +def check_rate_gate() -> flask.Response | None: + """Fail-closed gate for the audit ingest rate limit (#831). + + When AUDIT_TRACING_FAIL_CLOSED=1 (opt-in per deployment), requests + whose identity bucket is tripped in the SDK's ingest circuit + breaker (campus_python.audit.ratelimit — fed by this client's 429s) + are short-circuited with 503 + Retry-After before any view runs. + Off by default: producers run fail-open until flipped, per the + ratified rollout (design comment on campus#538). + + The gate is per-identity when identity is already resolvable at + request entry (app-level hooks run before blueprint-level ones, so + e.g. campus.auth's session resolution has run); it degrades to the + producer's learned apikey= fallback bucket (coarse) otherwise. + + Registered BEFORE start_span so a gated request never opens a span + (end_span no-ops without span state — no wasted ingest against a + bucket that audit is currently rejecting anyway). + """ + from campus.common import env + + if env.get("AUDIT_TRACING_FAIL_CLOSED") != "1": + return None + + # importlib with a string name: the module is optional (a stale + # campus_python predating #831 lacks it) and opaque resolution + # keeps type checkers happy against older SDK installs. + import importlib + + try: + ratelimit = importlib.import_module("campus_python.audit.ratelimit") + except ImportError: + logger.error( + "AUDIT_TRACING_FAIL_CLOSED=1 but campus_python lacks " + "audit.ratelimit (stale SDK); rate gate disabled" + ) + return None + + retry_after = ratelimit.breaker.check_request( + client_id=( + _identity_id(getattr(flask.g, "current_client", None)) + or getattr(flask.g, "client_id", None) + ), + user_id=( + _identity_id(getattr(flask.g, "current_user", None)) + or getattr(flask.g, "user_id", None) + ), + ) + if retry_after is None: + return None + + logger.warning( + "audit ingest breaker tripped; failing request closed " + "(retry_after=%ds)", + retry_after, + ) + response = flask.jsonify({ + "error": { + "code": "RATE_LIMITED", + "message": f"Rate limit exceeded; retry after {retry_after}s", + "details": {}, + "request_id": None, + } + }) + response.status_code = 503 + response.headers["Retry-After"] = str(max(1, int(retry_after))) + return response + + def start_span() -> None: """Start a span for the incoming request. diff --git a/poetry.lock b/poetry.lock index d604361f..95022c87 100644 --- a/poetry.lock +++ b/poetry.lock @@ -632,7 +632,7 @@ flask = "^3.0.0" type = "git" url = "https://github.com/nyjc-computing/campus-api-python.git" reference = "main" -resolved_reference = "e85cf61ae619695cbae28a15fffcbcfea30580dd" +resolved_reference = "9caf58b84ade2c804dc83688527f2525f2954701" [[package]] name = "certifi" diff --git a/tests/unit/audit/test_rate_gate.py b/tests/unit/audit/test_rate_gate.py new file mode 100644 index 00000000..80333ef3 --- /dev/null +++ b/tests/unit/audit/test_rate_gate.py @@ -0,0 +1,178 @@ +"""Unit tests for the audit ingest rate gate + breaker feed (#831). + +The producer side of campus.audit's per-identity ingest rate limit +(campus PR #835): the local AuditClient's traces.new feeds the SDK +circuit breaker on every ingest response, and the tracing middleware's +fail-closed gate (AUDIT_TRACING_FAIL_CLOSED=1) 503s matching requests +at entry. +""" + +import os +import unittest +from unittest import mock + +import flask +from campus_python.audit import ratelimit + + +def _set_env(test: unittest.TestCase, name: str, value: str) -> None: + """Set an env var for the duration of a test.""" + os.environ[name] = value + test.addCleanup(os.environ.pop, name, None) + + +def _429_body(bucket: str) -> dict: + """Error envelope as emitted by campus.audit #835.""" + return {"error": {"code": "RATE_LIMITED", "details": {"bucket": bucket}}} + + +def _make_app(**identity) -> flask.Flask: + """Flask app with the tracing middleware and a stub identity hook. + + The identity stub is registered BEFORE middleware.init_app so the + gate (a later before_request hook) observes it — mirroring how + app-level hooks registered earlier run first. + """ + app = flask.Flask(__name__) + app.config["TESTING"] = True + + @app.before_request + def stub_identity(): + for key, value in identity.items(): + setattr(flask.g, key, value) + + from campus.audit.middleware import init_app as init_tracing + + init_tracing(app) + + @app.route("/") + def index(): + return "ok" + + return app + + +class TestRateGate(unittest.TestCase): + """The fail-closed gate short-circuits tripped buckets at entry.""" + + def setUp(self): + ratelimit.breaker.reset() + self.addCleanup(ratelimit.breaker.reset) + + def test_flag_off_is_fail_open_even_when_tripped(self): + _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "") + ratelimit.breaker.observe(429, {"Retry-After": "60"}, _429_body("user=u1")) + client = _make_app().test_client() + response = client.get("/") + self.assertEqual(response.status_code, 200) + + def test_tripped_identity_bucket_returns_503(self): + _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "1") + ratelimit.breaker.observe(429, {"Retry-After": "60"}, _429_body("user=u1")) + client = _make_app(user_id="u1").test_client() + response = client.get("/") + self.assertEqual(response.status_code, 503) + self.assertIn("Retry-After", response.headers) + self.assertGreaterEqual(int(response.headers["Retry-After"]), 1) + data = response.get_json() + self.assertEqual(data["error"]["code"], "RATE_LIMITED") + + def test_other_identities_unaffected(self): + _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "1") + ratelimit.breaker.observe(429, {"Retry-After": "60"}, _429_body("user=u1")) + client = _make_app(user_id="u2").test_client() + self.assertEqual(client.get("/").status_code, 200) + + def test_identityless_uses_learned_fallback_bucket(self): + _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "1") + ratelimit.breaker.observe( + 429, {"Retry-After": "60"}, _429_body("apikey=k-123") + ) + client = _make_app().test_client() + self.assertEqual(client.get("/").status_code, 503) + + def test_identityless_before_learning_uses_wildcard(self): + _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "1") + ratelimit.breaker.observe( + 429, {"Retry-After": "60"}, _429_body(ratelimit.WILDCARD) + ) + client = _make_app().test_client() + self.assertEqual(client.get("/").status_code, 503) + + def test_availability_error_does_not_trip_gate(self): + _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "1") + # 503 from audit is an availability error: neutral. + ratelimit.breaker.observe(503, None, None) + client = _make_app().test_client() + self.assertEqual(client.get("/").status_code, 200) + + def test_verifying_state_gate_is_open(self): + _set_env(self, "AUDIT_TRACING_FAIL_CLOSED", "1") + ratelimit.breaker.observe(429, {"Retry-After": "60"}, _429_body("user=u1")) + # Force the cooldown into the past (verifying): gate opens... + ratelimit.breaker._tripped["user=u1"]["until"] -= 10_000 + client = _make_app(user_id="u1").test_client() + self.assertEqual(client.get("/").status_code, 200) + # ...and a 2xx ingest fully clears the trip. + ratelimit.breaker.observe(201, None, None) + self.assertEqual(ratelimit.breaker._tripped, {}) + + +class TestClientObserveHook(unittest.TestCase): + """AuditClient traces.new feeds the SDK breaker (#831).""" + + def setUp(self): + ratelimit.breaker.reset() + self.addCleanup(ratelimit.breaker.reset) + + def _traces_with_response(self, status: int, body: dict, raises: bool): + from campus.audit.client.v1.traces import Traces + + response = mock.Mock() + response.status_code = status + response.headers = {"Retry-After": "30"} + response.json.return_value = body + if raises: + response.raise_for_status.side_effect = RuntimeError(status) + client = mock.Mock() + client.post.return_value = response + return Traces(client=client, root=mock.Mock()), response + + def test_429_trips_breaker_and_still_raises(self): + traces, _ = self._traces_with_response( + 429, _429_body("client=c1;user=u1"), raises=True + ) + with self.assertRaises(RuntimeError): + traces.new({"span_id": "s1"}) + self.assertIsNotNone( + ratelimit.breaker.check_request(client_id="c1", user_id="u1") + ) + + def test_201_does_not_trip(self): + traces, _ = self._traces_with_response( + 201, {"created": ["s1"]}, raises=False + ) + traces.new({"span_id": "s1"}) + self.assertEqual(ratelimit.breaker._tripped, {}) + + def test_stale_sdk_degrades_silently(self): + """A missing campus_python.audit.ratelimit must not break ingest.""" + traces, response = self._traces_with_response( + 201, {"created": ["s1"]}, raises=False + ) + import importlib + + real_import_module = importlib.import_module + + def blocked_import(name, *args, **kwargs): + if name == "campus_python.audit.ratelimit": + raise ImportError("simulated stale SDK") + return real_import_module(name, *args, **kwargs) + + with mock.patch("importlib.import_module", side_effect=blocked_import): + traces.new({"span_id": "s1"}) # no raise + self.assertEqual(response.status_code, 201) + + +if __name__ == "__main__": + unittest.main()