From 32907213c48a275b5ad7eedd8183fffd1a10164f Mon Sep 17 00:00:00 2001 From: Developer Date: Sat, 10 Oct 2026 19:56:16 -0400 Subject: [PATCH] fix(alerts): exactly-once Zyla request accounting and Firebase QA/prod isolation (#31, #32) - Fixed Zyla request accounting to record each actual upstream HTTP attempt exactly once, eliminating double-counting when body parsing, JSON decoding, or normalization fails --- backend/motorsignal_api/app/auth.py | 19 +- .../motorsignal_api/app/road_alerts/cache.py | 4 + .../app/road_alerts/zyla_adapter.py | 35 ++- .../motorsignal_api/tests/test_road_alerts.py | 275 ++++++++++++++++++ 4 files changed, 322 insertions(+), 11 deletions(-) diff --git a/backend/motorsignal_api/app/auth.py b/backend/motorsignal_api/app/auth.py index 3cfc409..666172d 100644 --- a/backend/motorsignal_api/app/auth.py +++ b/backend/motorsignal_api/app/auth.py @@ -6,6 +6,7 @@ import time import urllib.request import uuid from datetime import datetime, timedelta, timezone +from typing import Optional import jwt from cryptography import x509 @@ -110,8 +111,14 @@ def _load_firebase_certs() -> dict[str, str]: return certs -def decode_firebase_id_token(token: str) -> dict: - project_id = os.environ.get("FIREBASE_PROJECT_ID", "motoapp-d22e0") +def decode_firebase_id_token(token: str, project_id: Optional[str] = None) -> dict: + target_project = project_id or os.environ.get("FIREBASE_PROJECT_ID") + if not target_project: + env = (os.environ.get("MOTORSIGNAL_ENV") or os.environ.get("ENVIRONMENT") or os.environ.get("APP_ENV") or "").lower() + if env in ("staging", "qa"): + target_project = "motorsignal-staging" + else: + target_project = "motoapp-d22e0" header = jwt.get_unverified_header(token) if header.get("alg") != "RS256" or not header.get("kid"): raise jwt.InvalidTokenError("Invalid Firebase token header") @@ -119,7 +126,13 @@ def decode_firebase_id_token(token: str) -> dict: if certificate_pem is None: raise jwt.InvalidTokenError("Unknown Firebase signing key") public_key = x509.load_pem_x509_certificate(certificate_pem.encode("ascii")).public_key() - claims = jwt.decode(token, public_key, algorithms=["RS256"], audience=project_id, issuer=f"https://securetoken.google.com/{project_id}") + claims = jwt.decode( + token, + public_key, + algorithms=["RS256"], + audience=target_project, + issuer=f"https://securetoken.google.com/{target_project}", + ) subject = claims.get("sub") if not isinstance(subject, str) or not subject or len(subject) > 128: raise jwt.InvalidTokenError("Invalid Firebase subject") diff --git a/backend/motorsignal_api/app/road_alerts/cache.py b/backend/motorsignal_api/app/road_alerts/cache.py index 4d9d931..5de6fc7 100644 --- a/backend/motorsignal_api/app/road_alerts/cache.py +++ b/backend/motorsignal_api/app/road_alerts/cache.py @@ -392,6 +392,10 @@ class PersistentUsageTracker: def requests_used(self) -> int: return self.get_stats()[1] + @property + def requests_executed(self) -> int: + return self.get_stats()[1] + @property def requests_reserved(self) -> int: return self.get_stats()[0] diff --git a/backend/motorsignal_api/app/road_alerts/zyla_adapter.py b/backend/motorsignal_api/app/road_alerts/zyla_adapter.py index 6b7f690..5b39523 100644 --- a/backend/motorsignal_api/app/road_alerts/zyla_adapter.py +++ b/backend/motorsignal_api/app/road_alerts/zyla_adapter.py @@ -376,6 +376,9 @@ class ZylaWazeAdapter: # Increment tracked requests self.request_count += 1 + upstream_attempted = False + upstream_recorded = False + try: # 5. Fixture handling or mock fallback if fixture_override: @@ -417,13 +420,24 @@ class ZylaWazeAdapter: "User-Agent": "MotorSignal-Backend/1.0", }, ) - with urllib.request.urlopen(req, timeout=8.0) as response: - body = response.read() - payload = json.loads(body) - # Record successful billable execution - if is_billable: - self.usage_tracker.record_execution(success=True, now=current_time) + upstream_attempted = True + try: + with urllib.request.urlopen(req, timeout=8.0) as response: + body = response.read() + # Upstream HTTP attempt succeeded; record exactly once + if is_billable and not upstream_recorded: + self.usage_tracker.record_execution(success=True, now=current_time) + upstream_recorded = True + except Exception as http_exc: + # Upstream HTTP attempt failed (429, 5xx, timeout, network); record exactly once + if is_billable and not upstream_recorded: + self.usage_tracker.record_execution(success=False, now=current_time) + upstream_recorded = True + raise http_exc + + # Parse JSON response (may raise json.JSONDecodeError if upstream returned malformed body) + payload = json.loads(body) alerts = self.parse_payload(payload, current_time) self.cache.set(min_lat, min_lng, max_lat, max_lng, alerts) @@ -436,9 +450,14 @@ class ZylaWazeAdapter: self.error_count += 1 self.last_error_at = current_time.isoformat() self.last_error_message = str(exc) - is_429 = "429" in str(exc) + is_429 = "429" in str(exc) or (hasattr(exc, "code") and getattr(exc, "code", None) == 429) self.circuit_breaker.record_failure(str(exc), is_rate_limit=is_429) logger.error("Zyla Waze API request failed: %s", exc) if is_billable: - self.usage_tracker.record_execution(success=False, now=current_time) + if upstream_attempted and not upstream_recorded: + self.usage_tracker.record_execution(success=False, now=current_time) + upstream_recorded = True + elif not upstream_attempted: + # Request aborted before making any network call; release the reserved slot + self.usage_tracker.release_reservation(now=current_time) return [] diff --git a/backend/motorsignal_api/tests/test_road_alerts.py b/backend/motorsignal_api/tests/test_road_alerts.py index 22e5537..0dcb568 100644 --- a/backend/motorsignal_api/tests/test_road_alerts.py +++ b/backend/motorsignal_api/tests/test_road_alerts.py @@ -8,6 +8,7 @@ import unittest from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Optional +from unittest.mock import MagicMock, patch # Add backend app directory and python packages to sys.path TEST_DIR = Path(__file__).resolve().parent @@ -588,6 +589,171 @@ class TestAtomicUsageTrackerAndRollover(unittest.TestCase): self.assertEqual(reserved, 0) self.assertEqual(executed, 0) + def test_malformed_json_response_does_not_double_count_request(self): + """ + P1 Accounting Bug Regression: Proves that when an upstream HTTP response + returns non-JSON / malformed data, the request is counted exactly once + and does NOT double-increment monthly counters or estimated spend. + """ + tracker = PersistentUsageTracker( + provider_id="zyla_4814", + monthly_budget_limit=10, + cost_per_request_usd=0.005, + db_url=self.db_url, + ) + self.trackers_to_clean.append(tracker) + + adapter = ZylaWazeAdapter( + api_key="real-test-key", + enabled=True, + usage_tracker=tracker, + fixture_mode=False, + ) + + mock_response = MagicMock() + mock_response.read.return_value = b"502 Bad GatewayBad Gateway" + mock_response.__enter__.return_value = mock_response + mock_response.__exit__.return_value = False + + with patch("urllib.request.urlopen", return_value=mock_response): + alerts = adapter.fetch_alerts(min_lat=34.2, min_lng=-85.2, max_lat=34.3, max_lng=-85.1) + + self.assertEqual(len(alerts), 0) + # Must be counted exactly once, NOT double-counted! + reserved, executed, cost = tracker.get_stats() + self.assertEqual(reserved, 1) + self.assertEqual(executed, 1) + self.assertAlmostEqual(cost, 0.005, places=4) + + def test_parsing_exception_in_payload_does_not_double_count_request(self): + """ + P1 Accounting Bug Regression: Proves that if parse_payload or alert normalization + raises an unhandled exception on valid JSON payload, the HTTP request is counted + exactly once. + """ + tracker = PersistentUsageTracker( + provider_id="zyla_4814", + monthly_budget_limit=10, + cost_per_request_usd=0.005, + db_url=self.db_url, + ) + self.trackers_to_clean.append(tracker) + + adapter = ZylaWazeAdapter( + api_key="real-test-key", + enabled=True, + usage_tracker=tracker, + fixture_mode=False, + ) + + mock_response = MagicMock() + mock_response.read.return_value = json.dumps({"alerts": []}).encode("utf-8") + mock_response.__enter__.return_value = mock_response + mock_response.__exit__.return_value = False + + with patch("urllib.request.urlopen", return_value=mock_response), \ + patch.object(adapter, "parse_payload", side_effect=RuntimeError("Parsing crash simulation")): + alerts = adapter.fetch_alerts(min_lat=34.2, min_lng=-85.2, max_lat=34.3, max_lng=-85.1) + + self.assertEqual(len(alerts), 0) + reserved, executed, cost = tracker.get_stats() + self.assertEqual(reserved, 1) + self.assertEqual(executed, 1) + self.assertAlmostEqual(cost, 0.005, places=4) + + def test_http_failures_and_timeouts_produce_accurate_counters(self): + """ + P1: Proves that HTTP 429 rate limit, 500 server error, and network timeouts + record exactly one request execution per network attempt, without overcounting. + """ + tracker = PersistentUsageTracker( + provider_id="zyla_4814", + monthly_budget_limit=10, + cost_per_request_usd=0.005, + db_url=self.db_url, + ) + self.trackers_to_clean.append(tracker) + + adapter = ZylaWazeAdapter( + api_key="real-test-key", + enabled=True, + usage_tracker=tracker, + fixture_mode=False, + ) + + import urllib.error + + # 1. Simulate HTTP 429 Too Many Requests + http_429 = urllib.error.HTTPError( + url="https://zylalabs.com/api", + code=429, + msg="Too Many Requests", + hdrs={}, + fp=None, + ) + with patch("urllib.request.urlopen", side_effect=http_429): + alerts = adapter.fetch_alerts(min_lat=34.2, min_lng=-85.2, max_lat=34.3, max_lng=-85.1) + + self.assertEqual(len(alerts), 0) + self.assertEqual(tracker.requests_reserved, 1) + self.assertEqual(tracker.requests_executed, 1) + self.assertEqual(adapter.circuit_breaker.state, "OPEN") + + # Reset circuit breaker for next test + adapter.circuit_breaker.reset() + + # 2. Simulate TimeoutError + timeout_err = TimeoutError("Connection timed out to Zyla gateway") + with patch("urllib.request.urlopen", side_effect=timeout_err): + alerts = adapter.fetch_alerts(min_lat=34.4, min_lng=-85.4, max_lat=34.5, max_lng=-85.3) + + self.assertEqual(len(alerts), 0) + self.assertEqual(tracker.requests_reserved, 2) + self.assertEqual(tracker.requests_executed, 2) + self.assertAlmostEqual(tracker.estimated_cost_usd, 0.010, places=4) + + def test_monthly_reservation_limit_strictly_enforced_against_upstream_calls(self): + """ + P0: Proves that once the monthly reservation cap is reached, no further upstream + HTTP network attempts are initiated. + """ + limit = 2 + tracker = PersistentUsageTracker( + provider_id="zyla_4814", + monthly_budget_limit=limit, + cost_per_request_usd=0.005, + db_url=self.db_url, + ) + self.trackers_to_clean.append(tracker) + + adapter = ZylaWazeAdapter( + api_key="real-test-key", + enabled=True, + usage_tracker=tracker, + fixture_mode=False, + ) + + mock_response = MagicMock() + mock_response.read.return_value = json.dumps({"alerts": []}).encode("utf-8") + mock_response.__enter__.return_value = mock_response + mock_response.__exit__.return_value = False + + with patch("urllib.request.urlopen", return_value=mock_response) as mock_urlopen: + # 1st call (within budget) + adapter.fetch_alerts(min_lat=34.0, min_lng=-85.0, max_lat=34.1, max_lng=-84.9) + # 2nd call (reaches budget) + adapter.fetch_alerts(min_lat=34.1, min_lng=-85.0, max_lat=34.2, max_lng=-84.9) + # 3rd call (exceeds budget) + adapter.fetch_alerts(min_lat=34.2, min_lng=-85.0, max_lat=34.3, max_lng=-84.9) + # 4th call (exceeds budget) + adapter.fetch_alerts(min_lat=34.3, min_lng=-85.0, max_lat=34.4, max_lng=-84.9) + + # Upstream HTTP call was made EXACTLY 2 times (never 3 or 4) + self.assertEqual(mock_urlopen.call_count, 2) + self.assertEqual(tracker.requests_reserved, 2) + self.assertEqual(tracker.requests_executed, 2) + self.assertFalse(tracker.can_request()) + class TestFirebaseGatewayAuthenticationIntegration(unittest.TestCase): @classmethod @@ -730,6 +896,115 @@ class TestFirebaseGatewayAuthenticationIntegration(unittest.TestCase): verify_admin_auth(authorization="Bearer test_admin_token_bypass_123", x_admin_key=None) self.assertEqual(ctx.exception.status_code, 403) + def test_qa_vs_production_firebase_project_id_isolation(self): + """ + P0: Proves project-specific Firebase token validation correctly isolates + QA (motorsignal-staging) from Production (motoapp-d22e0). + """ + qa_token = self._create_token( + sub="qa-rider-001", + aud="motorsignal-staging", + iss="https://securetoken.google.com/motorsignal-staging", + ) + prod_token = self._create_token( + sub="prod-rider-001", + aud="motoapp-d22e0", + iss="https://securetoken.google.com/motoapp-d22e0", + ) + + # 1. Under QA configuration: + os.environ["FIREBASE_PROJECT_ID"] = "motorsignal-staging" + # QA token succeeds + self.assertTrue(verify_gateway_or_user_auth(authorization=f"Bearer {qa_token}", x_gateway_key=None)) + # Production token is rejected with 401 + with self.assertRaises(HTTPException) as ctx: + verify_gateway_or_user_auth(authorization=f"Bearer {prod_token}", x_gateway_key=None) + self.assertEqual(ctx.exception.status_code, 401) + + # 2. Under Production configuration: + os.environ["FIREBASE_PROJECT_ID"] = "motoapp-d22e0" + # Production token succeeds + self.assertTrue(verify_gateway_or_user_auth(authorization=f"Bearer {prod_token}", x_gateway_key=None)) + # QA token is rejected with 401 + with self.assertRaises(HTTPException) as ctx: + verify_gateway_or_user_auth(authorization=f"Bearer {qa_token}", x_gateway_key=None) + self.assertEqual(ctx.exception.status_code, 401) + + def test_environment_driven_firebase_project_resolution(self): + """ + P0: Proves that when FIREBASE_PROJECT_ID is unconfigured, MOTORSIGNAL_ENV / + ENVIRONMENT selects the appropriate default (motorsignal-staging vs motoapp-d22e0). + """ + os.environ.pop("FIREBASE_PROJECT_ID", None) + + qa_token = self._create_token( + sub="qa-rider-002", + aud="motorsignal-staging", + iss="https://securetoken.google.com/motorsignal-staging", + ) + prod_token = self._create_token( + sub="prod-rider-002", + aud="motoapp-d22e0", + iss="https://securetoken.google.com/motoapp-d22e0", + ) + + # In staging/QA environment: + os.environ["MOTORSIGNAL_ENV"] = "staging" + claims_qa = auth.decode_firebase_id_token(qa_token) + self.assertEqual(claims_qa["sub"], "qa-rider-002") + + # In production environment: + os.environ["MOTORSIGNAL_ENV"] = "production" + claims_prod = auth.decode_firebase_id_token(prod_token) + self.assertEqual(claims_prod["sub"], "prod-rider-002") + + +class TestAlembicMigrationCleanApplication(unittest.TestCase): + def test_alembic_migration_generates_valid_schema_sql(self): + """ + Proves that Alembic migration f1a2b3c4d5eb applies cleanly and creates + the provider_monthly_usage table with its unique constraint and indexes. + """ + import io + from alembic.config import Config + from alembic import command + + ini_path = str(TEST_DIR.parent / "alembic.ini") + cfg = Config(ini_path) + + orig_user = os.environ.get("DATABASE_USER") + orig_pw = os.environ.get("DATABASE_PASSWORD") + orig_db = os.environ.get("DATABASE_NAME") + + os.environ["DATABASE_USER"] = "test_user" + os.environ["DATABASE_PASSWORD"] = "test_pass" + os.environ["DATABASE_NAME"] = "test_db" + + buf = io.StringIO() + try: + with patch("sys.stdout", buf): + command.upgrade(cfg, "head", sql=True) + sql_output = buf.getvalue() + finally: + if orig_user is not None: + os.environ["DATABASE_USER"] = orig_user + else: + os.environ.pop("DATABASE_USER", None) + if orig_pw is not None: + os.environ["DATABASE_PASSWORD"] = orig_pw + else: + os.environ.pop("DATABASE_PASSWORD", None) + if orig_db is not None: + os.environ["DATABASE_NAME"] = orig_db + else: + os.environ.pop("DATABASE_NAME", None) + + self.assertIn("CREATE TABLE provider_monthly_usage", sql_output) + self.assertIn("uq_provider_period", sql_output) + self.assertIn("ix_provider_monthly_usage_provider_id", sql_output) + self.assertIn("ix_provider_monthly_usage_period_month", sql_output) + self.assertIn("version_num='f1a2b3c4d5eb'", sql_output) + if __name__ == "__main__": unittest.main()