From b85c4c24c2b17a388c424a1efdbac6540e876bec Mon Sep 17 00:00:00 2001 From: JS Ng Date: Tue, 6 Oct 2026 10:45:43 +0800 Subject: [PATCH 1/3] feat(tracing): fail-closed rate gate + breaker feed on ingest (#831) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Producer side of the audit ingest rate limit (audit enforcement landed in #835; SDK breaker in campus-api-python#98). Design: design comment on campus#538. - AuditClient traces.new feeds the campus_python.audit.ratelimit breaker on every ingest response before raising: 429 trips the bucket (Retry-After + tripped key from the body), 2xx clears expired trips, availability errors neutral. Observation is guarded — a breaker failure or stale SDK can never break ingestion. - New check_rate_gate() before_request hook (registered before start_span, so gated requests never open a span): when AUDIT_TRACING_FAIL_CLOSED=1, requests whose identity bucket is tripped get 503 + Retry-After at entry. Per-identity where identity is resolvable (app-level hook ordering), degrading to the producer's learned apikey= fallback bucket or the wildcard. Off by default — producers stay fail-open until flipped per the ratified rollout. - classroom (and any producer on campus.audit.middleware) inherits the gate with a campus-suite bump; no producer-side code needed there. AUDIT_TRACING_FAIL_CLOSED is a new deployment variable; document+flip per producer on dev as the #831 verification step. --- campus/audit/client/v1/traces.py | 31 +++++ campus/audit/middleware/__init__.py | 9 ++ campus/audit/middleware/tracing.py | 64 ++++++++++ tests/unit/audit/test_rate_gate.py | 178 ++++++++++++++++++++++++++++ 4 files changed, 282 insertions(+) create mode 100644 tests/unit/audit/test_rate_gate.py diff --git a/campus/audit/client/v1/traces.py b/campus/audit/client/v1/traces.py index 61d86dab..33c2a4ff 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,26 @@ from ..interface import SLASH, Resource, ResourceCollection, ResourceRoot +def _import_ratelimit(): + """Import the SDK breaker module lazily (it ships in campus_python). + + Kept behind a function so a missing/stale campus_python (older + deployments pre-#831) only degrades to no observation, never an + import error at module load. + """ + from campus_python.audit import ratelimit as campus_python_audit_ratelimit + + return campus_python_audit_ratelimit + + +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 +101,16 @@ 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 and + # can never break ingestion; the error still raises as normal. + with contextlib.suppress(Exception): + _import_ratelimit().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 d9b4f7bd..7c4eb9b9 100644 --- a/campus/audit/middleware/tracing.py +++ b/campus/audit/middleware/tracing.py @@ -221,6 +221,70 @@ 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 + + try: + import campus_python.audit.ratelimit as 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/tests/unit/audit/test_rate_gate.py b/tests/unit/audit/test_rate_gate.py new file mode 100644 index 00000000..20dd591b --- /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 builtins + + real_import = builtins.__import__ + + def blocked_import(name, *args, **kwargs): + if name.startswith("campus_python"): + raise ImportError("simulated stale SDK") + return real_import(name, *args, **kwargs) + + with mock.patch("builtins.__import__", side_effect=blocked_import): + traces.new({"span_id": "s1"}) # no raise + self.assertEqual(response.status_code, 201) + + +if __name__ == "__main__": + unittest.main() From c66de5e40719d4c965d8e19693d1524b3cebef41 Mon Sep 17 00:00:00 2001 From: JS Ng Date: Tue, 6 Oct 2026 10:50:14 +0800 Subject: [PATCH 2/3] fix(tracing): resolve SDK breaker via importlib (opaque optional import) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI type checks install campus-api-python from git main, which lacks campus_python.audit.ratelimit until campus-api-python#98 merges — the static imports failed pyright. importlib.import_module with a string name is opaque to type checkers and keeps the stale-SDK degradation runtime behavior under any install. Stale-SDK test now patches importlib.import_module to match. --- campus/audit/client/v1/traces.py | 30 ++++++++++++++++++------------ campus/audit/middleware/tracing.py | 7 ++++++- tests/unit/audit/test_rate_gate.py | 10 +++++----- 3 files changed, 29 insertions(+), 18 deletions(-) diff --git a/campus/audit/client/v1/traces.py b/campus/audit/client/v1/traces.py index 33c2a4ff..09c62f14 100644 --- a/campus/audit/client/v1/traces.py +++ b/campus/audit/client/v1/traces.py @@ -21,13 +21,17 @@ def _import_ratelimit(): """Import the SDK breaker module lazily (it ships in campus_python). - Kept behind a function so a missing/stale campus_python (older - deployments pre-#831) only degrades to no observation, never an - import error at module load. + 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. """ - from campus_python.audit import ratelimit as campus_python_audit_ratelimit + import importlib - return campus_python_audit_ratelimit + try: + return importlib.import_module("campus_python.audit.ratelimit") + except ImportError: + return None def _safe_json(response: JsonResponse) -> Any: @@ -103,14 +107,16 @@ def new(self, *spans: dict) -> JsonResponse: 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 and - # can never break ingestion; the error still raises as normal. + # body), 2xx clears expired trips. Observation is guarded — a + # breaker failure or a stale SDK can never break ingestion. with contextlib.suppress(Exception): - _import_ratelimit().breaker.observe( - response.status_code, - response.headers, - _safe_json(response), - ) + 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/tracing.py b/campus/audit/middleware/tracing.py index 7c4eb9b9..f0f1867d 100644 --- a/campus/audit/middleware/tracing.py +++ b/campus/audit/middleware/tracing.py @@ -245,8 +245,13 @@ def check_rate_gate() -> flask.Response | None: 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: - import campus_python.audit.ratelimit as ratelimit + ratelimit = importlib.import_module("campus_python.audit.ratelimit") except ImportError: logger.error( "AUDIT_TRACING_FAIL_CLOSED=1 but campus_python lacks " diff --git a/tests/unit/audit/test_rate_gate.py b/tests/unit/audit/test_rate_gate.py index 20dd591b..80333ef3 100644 --- a/tests/unit/audit/test_rate_gate.py +++ b/tests/unit/audit/test_rate_gate.py @@ -160,16 +160,16 @@ def test_stale_sdk_degrades_silently(self): traces, response = self._traces_with_response( 201, {"created": ["s1"]}, raises=False ) - import builtins + import importlib - real_import = builtins.__import__ + real_import_module = importlib.import_module def blocked_import(name, *args, **kwargs): - if name.startswith("campus_python"): + if name == "campus_python.audit.ratelimit": raise ImportError("simulated stale SDK") - return real_import(name, *args, **kwargs) + return real_import_module(name, *args, **kwargs) - with mock.patch("builtins.__import__", side_effect=blocked_import): + with mock.patch("importlib.import_module", side_effect=blocked_import): traces.new({"span_id": "s1"}) # no raise self.assertEqual(response.status_code, 201) From 7ebf4657951406ba2b9596f60a20ccc4c7f90025 Mon Sep 17 00:00:00 2001 From: JS Ng Date: Tue, 6 Oct 2026 10:55:18 +0800 Subject: [PATCH 3/3] chore(deps): pin campus-api-python to post-breaker main (#98) CI type checks and unit tests resolve campus_python.audit.ratelimit from the poetry.lock-pinned campus-api-python commit; bumping the lock to main@9caf58b (campus-api-python#98 merge) makes the breaker module visible to both, un-skipping the rate-gate suite on CI. --- poetry.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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"