diff --git a/backend/motorsignal_api/alembic/versions/f1a2b3c4d5eb_add_provider_monthly_usage.py b/backend/motorsignal_api/alembic/versions/f1a2b3c4d5eb_add_provider_monthly_usage.py new file mode 100644 index 0000000..1805278 --- /dev/null +++ b/backend/motorsignal_api/alembic/versions/f1a2b3c4d5eb_add_provider_monthly_usage.py @@ -0,0 +1,48 @@ +"""add provider_monthly_usage table for atomic usage limits + +Revision ID: f1a2b3c4d5eb +Revises: f1a2b3c4d5ea +Create Date: 2026-10-10 +""" + +from alembic import op +import sqlalchemy as sa + +# revision identifiers, used by Alembic. +revision = 'f1a2b3c4d5eb' +down_revision = 'f1a2b3c4d5ea' +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "provider_monthly_usage", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("provider_id", sa.String(length=64), nullable=False), + sa.Column("period_month", sa.String(length=7), nullable=False), + sa.Column("requests_reserved", sa.Integer(), server_default="0", nullable=False), + sa.Column("requests_executed", sa.Integer(), server_default="0", nullable=False), + sa.Column("monthly_limit", sa.Integer(), server_default="500", nullable=False), + sa.Column("estimated_cost_usd", sa.Float(), server_default="0.0", nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("provider_id", "period_month", name="uq_provider_period"), + ) + op.create_index( + "ix_provider_monthly_usage_provider_id", + "provider_monthly_usage", + ["provider_id"], + ) + op.create_index( + "ix_provider_monthly_usage_period_month", + "provider_monthly_usage", + ["period_month"], + ) + + +def downgrade() -> None: + op.drop_index("ix_provider_monthly_usage_period_month", table_name="provider_monthly_usage") + op.drop_index("ix_provider_monthly_usage_provider_id", table_name="provider_monthly_usage") + op.drop_table("provider_monthly_usage") diff --git a/backend/motorsignal_api/app/auth.py b/backend/motorsignal_api/app/auth.py index 1b035f1..3cfc409 100644 --- a/backend/motorsignal_api/app/auth.py +++ b/backend/motorsignal_api/app/auth.py @@ -111,7 +111,7 @@ def _load_firebase_certs() -> dict[str, str]: def decode_firebase_id_token(token: str) -> dict: - project_id = os.environ["FIREBASE_PROJECT_ID"] + project_id = os.environ.get("FIREBASE_PROJECT_ID", "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") diff --git a/backend/motorsignal_api/app/models.py b/backend/motorsignal_api/app/models.py index ab6136d..058870c 100644 --- a/backend/motorsignal_api/app/models.py +++ b/backend/motorsignal_api/app/models.py @@ -475,3 +475,29 @@ class OpsCmsRevision(Base): document: Mapped["OpsCmsDocument"] = relationship(back_populates="revisions") + +# ============================================================ +# Road Alerts & External Provider Usage Models +# ============================================================ + + +class ProviderMonthlyUsage(Base): + __tablename__ = "provider_monthly_usage" + __table_args__ = ( + Index("ix_provider_period", "provider_id", "period_month", unique=True), + ) + + id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) + provider_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + period_month: Mapped[str] = mapped_column(String(7), nullable=False, index=True) # Format: YYYY-MM + requests_reserved: Mapped[int] = mapped_column(nullable=False, default=0) + requests_executed: Mapped[int] = mapped_column(nullable=False, default=0) + monthly_limit: Mapped[int] = mapped_column(nullable=False, default=500) + estimated_cost_usd: Mapped[float] = mapped_column(nullable=False, default=0.0) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=datetime.utcnow, nullable=False + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False + ) + diff --git a/backend/motorsignal_api/app/road_alerts/cache.py b/backend/motorsignal_api/app/road_alerts/cache.py index a191754..4d9d931 100644 --- a/backend/motorsignal_api/app/road_alerts/cache.py +++ b/backend/motorsignal_api/app/road_alerts/cache.py @@ -1,3 +1,4 @@ +import logging import math import os import time @@ -5,6 +6,8 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional +logger = logging.getLogger(__name__) + class GeographicCellCache: """ @@ -123,84 +126,287 @@ class PersistentUsageTracker: provider_id: str, monthly_budget_limit: int = 500, cost_per_request_usd: float = 0.005, + engine: Optional[Any] = None, + db_url: Optional[str] = None, state_file: Optional[Any] = None, ): - from pathlib import Path - import json - self.provider_id = provider_id self.monthly_budget_limit = monthly_budget_limit self.cost_per_request_usd = cost_per_request_usd + self._engine = engine + self._db_url = db_url + if state_file and not self._db_url and not self._engine: + self._db_url = f"sqlite:///{Path(state_file).resolve().as_posix()}" + self._table_ensured = False - if state_file is not None: - self.state_file = Path(state_file) - else: - base_dir = Path(os.environ.get("USAGE_STATE_DIR", Path(__file__).resolve().parent.parent.parent / "storage")) - self.state_file = base_dir / f"usage_{provider_id}.json" + def get_engine(self) -> Optional[Any]: + """ + Resolves the SQLAlchemy Engine for quota and budget persistence. + Returns None if store is unconfigured or unreachable (failing closed). + """ + if self._engine is not None: + self._ensure_table(self._engine) + return self._engine - from threading import Lock - self.lock = Lock() - self.current_period = datetime.now(timezone.utc).strftime("%Y-%m") - self.requests_used: int = 0 - self.total_lifetime_requests: int = 0 - self._load() + if self._db_url: + try: + from sqlalchemy import create_engine + self._engine = create_engine(self._db_url, pool_pre_ping=True) + self._ensure_table(self._engine) + return self._engine + except Exception as e: + logger.error("Failed to initialize engine from db_url: %s", e) + return None - def _load(self) -> None: + # Check DATABASE_URL environment variable + db_url_env = os.environ.get("DATABASE_URL") + if db_url_env: + try: + from sqlalchemy import create_engine + self._engine = create_engine(db_url_env, pool_pre_ping=True) + self._ensure_table(self._engine) + return self._engine + except Exception as e: + logger.error("Failed to initialize engine from DATABASE_URL: %s", e) + return None + + # Attempt resolving MotorSignal application database engine + if "DATABASE_USER" in os.environ and "DATABASE_NAME" in os.environ: + try: + from db import engine as app_engine + self._engine = app_engine + self._ensure_table(self._engine) + return self._engine + except Exception as e: + logger.error("Failed to load MotorSignal app.db.engine: %s", e) + return None + + return None + + def _ensure_table(self, engine: Any) -> None: + if self._table_ensured: + return try: - if self.state_file.exists(): - import json - with open(self.state_file, "r", encoding="utf-8") as f: - data = json.load(f) - saved_period = data.get("period") - if saved_period == self.current_period: - self.requests_used = int(data.get("requests_used", 0)) + from models import ProviderMonthlyUsage + ProviderMonthlyUsage.__table__.create(bind=engine, checkfirst=True) + self._table_ensured = True + except Exception as e: + logger.debug("Table check/creation completed: %s", e) + + def reserve_request(self, now: Optional[datetime] = None) -> bool: + """ + Atomically reserves 1 request slot against the monthly limit in the database. + Uses row-level atomic UPSERT with WHERE constraint to prevent concurrent + workers from exceeding the allowance. + FAILS CLOSED: Returns False if limit is reached or if database is unavailable. + """ + if self.monthly_budget_limit <= 0: + return False + + current_time = now or datetime.now(timezone.utc) + period = current_time.strftime("%Y-%m") + engine = self.get_engine() + if engine is None: + logger.error( + "Budget store unavailable for provider %s. Failing closed.", + self.provider_id, + ) + return False + + try: + from sqlalchemy import text + with engine.begin() as conn: + stmt = text(""" + INSERT INTO provider_monthly_usage ( + provider_id, period_month, requests_reserved, requests_executed, + monthly_limit, estimated_cost_usd, created_at, updated_at + ) + VALUES ( + :provider_id, :period, 1, 0, + :monthly_limit, 0.0, :now, :now + ) + ON CONFLICT (provider_id, period_month) + DO UPDATE SET + requests_reserved = provider_monthly_usage.requests_reserved + 1, + monthly_limit = :monthly_limit, + updated_at = :now + WHERE provider_monthly_usage.requests_reserved < :monthly_limit + RETURNING requests_reserved, monthly_limit + """) + result = conn.execute( + stmt, + { + "provider_id": self.provider_id, + "period": period, + "monthly_limit": self.monthly_budget_limit, + "now": current_time, + }, + ) + row = result.fetchone() + if row is not None: + return True else: - self.requests_used = 0 - self.total_lifetime_requests = int(data.get("lifetime_requests", self.requests_used)) - except Exception: - pass + logger.warning( + "Monthly budget reservation limit reached for %s in period %s (limit: %d).", + self.provider_id, + period, + self.monthly_budget_limit, + ) + return False + except Exception as e: + logger.error( + "Database error during reserve_request for %s: %s. Failing closed.", + self.provider_id, + e, + ) + return False + + def release_reservation(self, now: Optional[datetime] = None) -> None: + """ + Releases 1 unexecuted reserved slot if an upstream request was cancelled before sending. + """ + current_time = now or datetime.now(timezone.utc) + period = current_time.strftime("%Y-%m") + engine = self.get_engine() + if engine is None: + return - def _save(self) -> None: try: - import json - self.state_file.parent.mkdir(parents=True, exist_ok=True) - tmp_path = self.state_file.with_suffix(".tmp") - data = { - "provider_id": self.provider_id, - "period": self.current_period, - "requests_used": self.requests_used, - "lifetime_requests": self.total_lifetime_requests, - "monthly_budget_limit": self.monthly_budget_limit, - "estimated_cost_usd": round(self.requests_used * self.cost_per_request_usd, 4), - "updated_at": datetime.now(timezone.utc).isoformat(), - } - with open(tmp_path, "w", encoding="utf-8") as f: - json.dump(data, f, indent=2) - tmp_path.replace(self.state_file) + from sqlalchemy import text + with engine.begin() as conn: + stmt = text(""" + UPDATE provider_monthly_usage + SET requests_reserved = CASE WHEN requests_reserved > 0 THEN requests_reserved - 1 ELSE 0 END, + updated_at = :now + WHERE provider_id = :provider_id AND period_month = :period + """) + conn.execute( + stmt, + { + "provider_id": self.provider_id, + "period": period, + "now": current_time, + }, + ) + except Exception as e: + logger.error("Database error during release_reservation for %s: %s", self.provider_id, e) + + def record_execution(self, success: bool = True, now: Optional[datetime] = None) -> None: + """ + Records that an actual upstream HTTP request was executed. + Separately tracks executed requests and calculates estimated spend. + """ + current_time = now or datetime.now(timezone.utc) + period = current_time.strftime("%Y-%m") + engine = self.get_engine() + if engine is None: + return + + try: + from sqlalchemy import text + with engine.begin() as conn: + stmt = text(""" + UPDATE provider_monthly_usage + SET requests_executed = requests_executed + 1, + estimated_cost_usd = round((requests_executed + 1) * :cost_per_req, 4), + updated_at = :now + WHERE provider_id = :provider_id AND period_month = :period + """) + conn.execute( + stmt, + { + "provider_id": self.provider_id, + "period": period, + "cost_per_req": self.cost_per_request_usd, + "now": current_time, + }, + ) + except Exception as e: + logger.error("Database error during record_execution for %s: %s", self.provider_id, e) + + def record_request(self, now: Optional[datetime] = None) -> int: + """ + Backward-compatible helper reserving and recording 1 completed request. + """ + reserved = self.reserve_request(now=now) + if reserved: + self.record_execution(success=True, now=now) + return self.requests_used + + def can_request(self, now: Optional[datetime] = None) -> bool: + """ + Checks whether requests can be reserved for the current month. + Fails closed (returns False) if budget is exhausted or database is down. + """ + if self.monthly_budget_limit <= 0: + return False + current_time = now or datetime.now(timezone.utc) + period = current_time.strftime("%Y-%m") + engine = self.get_engine() + if engine is None: + logger.error("Budget store unavailable for %s. Failing closed.", self.provider_id) + return False + + try: + from sqlalchemy import text + with engine.connect() as conn: + stmt = text(""" + SELECT requests_reserved, monthly_limit + FROM provider_monthly_usage + WHERE provider_id = :provider_id AND period_month = :period + """) + row = conn.execute(stmt, {"provider_id": self.provider_id, "period": period}).fetchone() + if row is None: + return True + return int(row[0]) < int(row[1]) + except Exception as e: + logger.error("Database error during can_request for %s: %s. Failing closed.", self.provider_id, e) + return False + + def get_stats(self, now: Optional[datetime] = None) -> tuple[int, int, float]: + """ + Returns (requests_reserved, requests_executed, estimated_cost_usd) for the period. + """ + current_time = now or datetime.now(timezone.utc) + period = current_time.strftime("%Y-%m") + engine = self.get_engine() + if engine is None: + return 0, 0, 0.0 + + try: + from sqlalchemy import text + with engine.connect() as conn: + stmt = text(""" + SELECT requests_reserved, requests_executed, estimated_cost_usd + FROM provider_monthly_usage + WHERE provider_id = :provider_id AND period_month = :period + """) + row = conn.execute(stmt, {"provider_id": self.provider_id, "period": period}).fetchone() + if row is not None: + return int(row[0]), int(row[1]), float(row[2]) + return 0, 0, 0.0 except Exception: - pass + return 0, 0, 0.0 - def check_period(self) -> None: - now_period = datetime.now(timezone.utc).strftime("%Y-%m") - if now_period != self.current_period: - self.current_period = now_period - self.requests_used = 0 - self._save() + @property + def requests_used(self) -> int: + return self.get_stats()[1] - def can_request(self) -> bool: - with self.lock: - self.check_period() - return self.requests_used < self.monthly_budget_limit - - def record_request(self) -> int: - with self.lock: - self.check_period() - self.requests_used += 1 - self.total_lifetime_requests += 1 - self._save() - return self.requests_used + @property + def requests_reserved(self) -> int: + return self.get_stats()[0] @property def estimated_cost_usd(self) -> float: - return round(self.requests_used * self.cost_per_request_usd, 4) + return self.get_stats()[2] + + def dispose(self) -> None: + """ + Disposes the underlying SQLAlchemy engine and releases connection pools. + """ + if self._engine is not None: + try: + self._engine.dispose() + except Exception: + pass diff --git a/backend/motorsignal_api/app/road_alerts/router.py b/backend/motorsignal_api/app/road_alerts/router.py index e85313f..1716269 100644 --- a/backend/motorsignal_api/app/road_alerts/router.py +++ b/backend/motorsignal_api/app/road_alerts/router.py @@ -67,13 +67,37 @@ def verify_rate_limit(): ALLOWED_FIXTURES = {"zyla_rome_sample.json", "zyla_atlanta_sample.json"} +def is_production_environment() -> bool: + """ + Returns True if the runtime environment indicates production or release. + Prevents any test bypasses from executing in production builds. + """ + env_vars = [ + os.environ.get("MOTORSIGNAL_ENV", "").lower(), + os.environ.get("ENVIRONMENT", "").lower(), + os.environ.get("APP_ENV", "").lower(), + os.environ.get("NODE_ENV", "").lower(), + ] + return any(e in ("production", "prod", "staging", "release") for e in env_vars) + + +def is_test_token_allowed() -> bool: + """ + Test-token bypasses are strictly forbidden in production. + Only permitted in automated local test suites with explicit TESTING=true. + """ + if is_production_environment(): + return False + return os.environ.get("TESTING", "").lower() in ("true", "1") + + def verify_gateway_or_user_auth( authorization: Optional[str] = Header(None, alias="Authorization"), x_gateway_key: Optional[str] = Header(None, alias="X-MotorSignal-Gateway-Key"), ): """ Authenticates requests via either: - 1. A valid User Bearer Token (MotorSignal JWT access token or Firebase ID Token). + 1. A valid User Bearer Token (Firebase ID Token or MotorSignal JWT access token). 2. A matching server-side gateway secret (X-MotorSignal-Gateway-Key), if explicitly configured. FAILS CLOSED: If neither is provided or valid, raises HTTP 401 Unauthorized. """ @@ -82,21 +106,14 @@ def verify_gateway_or_user_auth( if required_secret and x_gateway_key == required_secret: return True - # 2. Bearer token check (MotorSignal access token or Firebase ID token) + # 2. Bearer token check (Firebase ID token or MotorSignal access token) if authorization: scheme, _, token = authorization.partition(" ") if scheme.lower() == "bearer" and token.strip(): token_val = token.strip() - # Try decoding MotorSignal access token - try: - from auth import decode_access_token - payload = decode_access_token(token_val) - if payload.get("type") in ("access", "test") or "sub" in payload: - return True - except Exception: - pass - # Try decoding Firebase ID token + # A. Try decoding Firebase ID token (cryptographically verified RS256 signature, + # audience, issuer, expiry, and subject) try: from auth import decode_firebase_id_token claims = decode_firebase_id_token(token_val) @@ -105,12 +122,19 @@ def verify_gateway_or_user_auth( except Exception: pass - # In test mode, accept simulated test/mock tokens - is_test_env = ( - os.environ.get("TESTING", "").lower() in ("true", "1") - or os.environ.get("ENVIRONMENT", "").lower() == "test" - ) - if is_test_env and (token_val.startswith("test_token_") or token_val.startswith("mock_token_")): + # B. Try decoding MotorSignal access token + try: + from auth import decode_access_token + payload = decode_access_token(token_val) + if payload.get("type") in ("access", "test") or "sub" in payload: + return True + except Exception: + pass + + # C. In non-production automated test environments only, accept mock test tokens + if is_test_token_allowed() and ( + token_val.startswith("test_token_") or token_val.startswith("mock_token_") + ): return True # Strict Fail Closed @@ -128,6 +152,8 @@ def verify_admin_auth( """ Requires admin privileges to modify operational provider switches. FAILS CLOSED: If admin key is missing or invalid, raises HTTP 403 Forbidden. + Verifies roles exclusively from trusted, cryptographically validated token claims + or server-configured admin secrets. Unverified user-supplied fields are never trusted. """ required_admin = os.environ.get("MOTORSIGNAL_ADMIN_KEY") if required_admin and x_admin_key == required_admin: @@ -137,20 +163,41 @@ def verify_admin_auth( scheme, _, token = authorization.partition(" ") if scheme.lower() == "bearer" and token.strip(): token_val = token.strip() + + # 1. Check cryptographically verified Firebase ID token claims + try: + from auth import decode_firebase_id_token + claims = decode_firebase_id_token(token_val) + # Roles MUST come from cryptographically signed custom claims in the token: + if claims.get("admin") is True or claims.get("role") in ("admin", "operator", "owner"): + return True + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Admin authorization required. Token lacks required administrator claims.", + ) + except HTTPException: + raise + except Exception: + pass + + # 2. Check cryptographically verified MotorSignal JWT access token claims try: from auth import decode_access_token payload = decode_access_token(token_val) role = payload.get("role", "") - if role in ("owner", "operator", "admin"): + if role in ("owner", "operator", "admin") or payload.get("is_admin") is True: return True + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Admin authorization required. Token lacks required administrator claims.", + ) + except HTTPException: + raise except Exception: pass - is_test_env = ( - os.environ.get("TESTING", "").lower() in ("true", "1") - or os.environ.get("ENVIRONMENT", "").lower() == "test" - ) - if is_test_env and token_val.startswith("test_admin_token_"): + # 3. In non-production automated test environments only, accept mock admin test tokens + if is_test_token_allowed() and token_val.startswith("test_admin_token_"): return True raise HTTPException( diff --git a/backend/motorsignal_api/app/road_alerts/zyla_adapter.py b/backend/motorsignal_api/app/road_alerts/zyla_adapter.py index aa555d5..6b7f690 100644 --- a/backend/motorsignal_api/app/road_alerts/zyla_adapter.py +++ b/backend/motorsignal_api/app/road_alerts/zyla_adapter.py @@ -53,7 +53,11 @@ class ZylaWazeAdapter: enabled: Optional[bool] = None, monthly_budget_limit: int = 500, cost_per_request_usd: float = 0.005, + engine: Optional[Any] = None, + db_url: Optional[str] = None, state_file: Optional[Path] = None, + usage_tracker: Optional[PersistentUsageTracker] = None, + fixture_mode: Optional[bool] = None, ): # Resolve config from environment or overrides self.api_key = api_key if api_key is not None else os.environ.get("ZYLA_API_KEY", "") @@ -62,17 +66,22 @@ class ZylaWazeAdapter: else: self.enabled = os.environ.get("ZYLA_ENABLED", "false").lower() in ("true", "1") - self.fixture_mode = os.environ.get("ZYLA_FIXTURE_MODE", "false").lower() in ("true", "1") + if fixture_mode is not None: + self.fixture_mode = fixture_mode + else: + self.fixture_mode = os.environ.get("ZYLA_FIXTURE_MODE", "false").lower() in ("true", "1") self.monthly_budget_limit = monthly_budget_limit self.cost_per_request_usd = cost_per_request_usd # Operational telemetry and durable billing safeguard self.cache = GeographicCellCache(cell_size_degrees=0.05, ttl_seconds=300.0) self.circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout_sec=300.0) - self.usage_tracker = PersistentUsageTracker( + self.usage_tracker = usage_tracker or PersistentUsageTracker( provider_id=self.PROVIDER_ID, monthly_budget_limit=monthly_budget_limit, cost_per_request_usd=cost_per_request_usd, + engine=engine, + db_url=db_url, state_file=state_file, ) self.request_count: int = 0 @@ -347,21 +356,25 @@ class ZylaWazeAdapter: logger.warning("Zyla Waze API key missing. Live calls prevented.") return [] + # Determine whether this is a real billable upstream request + is_billable = not (self.fixture_mode or bool(fixture_override)) + # 3. Circuit breaker check if not self.circuit_breaker.can_request() and not fixture_override: logger.warning("Zyla Waze circuit breaker is OPEN. Requests prevented.") return [] - # 4. Durable request budget safeguard - if not self.usage_tracker.can_request() and not fixture_override: - logger.warning("Zyla Waze monthly budget limit reached (%d).", self.monthly_budget_limit) - return [] + # 4. Durable atomic request budget safeguard + if is_billable: + if not self.usage_tracker.reserve_request(now=current_time): + logger.warning( + "Zyla Waze monthly budget limit reached (%d) or budget store unavailable.", + self.monthly_budget_limit, + ) + return [] # Increment tracked requests self.request_count += 1 - if not fixture_override: - self.usage_tracker.record_request() - try: # 5. Fixture handling or mock fallback @@ -375,6 +388,14 @@ class ZylaWazeAdapter: payload = self.load_fixture("zyla_atlanta_sample.json") else: # Live mode - Only executed when explicitly configured with valid key and enabled + if not self.api_key: + logger.warning("Zyla live request attempted without API key. Rejecting.") + self.invalid_filtered_count += 1 + self.error_count += 1 + if is_billable: + self.usage_tracker.release_reservation(now=current_time) + return [] + # We use urllib to avoid external dependencies import urllib.request import urllib.parse @@ -400,6 +421,10 @@ class ZylaWazeAdapter: body = response.read() payload = json.loads(body) + # Record successful billable execution + if is_billable: + self.usage_tracker.record_execution(success=True, now=current_time) + alerts = self.parse_payload(payload, current_time) self.cache.set(min_lat, min_lng, max_lat, max_lng, alerts) self.circuit_breaker.record_success() @@ -414,4 +439,6 @@ class ZylaWazeAdapter: is_429 = "429" in str(exc) 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) return [] diff --git a/backend/motorsignal_api/tests/test_road_alerts.py b/backend/motorsignal_api/tests/test_road_alerts.py index e9fe3ea..22e5537 100644 --- a/backend/motorsignal_api/tests/test_road_alerts.py +++ b/backend/motorsignal_api/tests/test_road_alerts.py @@ -1,10 +1,13 @@ +import concurrent.futures import json import os import sys import tempfile +import time import unittest -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from pathlib import Path +from typing import Optional # Add backend app directory and python packages to sys.path TEST_DIR = Path(__file__).resolve().parent @@ -14,6 +17,16 @@ PACKAGES_DIR = TEST_DIR.parent / ".python-packages" sys.path.insert(0, str(APP_DIR)) sys.path.insert(0, str(PACKAGES_DIR)) +os.environ.setdefault("JWT_SECRET", "local-test-secret-at-least-32-chars-long") +os.environ.setdefault("FIREBASE_PROJECT_ID", "motoapp-d22e0") + +import jwt +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.x509.oid import NameOID +import auth + from fastapi import HTTPException from road_alerts.cache import CircuitBreaker, GeographicCellCache, PersistentUsageTracker from road_alerts.ga511_adapter import Georgia511Adapter @@ -204,6 +217,9 @@ class TestZylaRoadAlertAdapter(unittest.TestCase): with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as tmp: tmp_path = Path(tmp.name) + tracker1 = None + tracker2 = None + tracker3 = None try: # 1. Process 1: Record 3 requests against a limit of 5 tracker1 = PersistentUsageTracker( @@ -245,8 +261,14 @@ class TestZylaRoadAlertAdapter(unittest.TestCase): self.assertEqual(tracker3.requests_used, 5) self.assertFalse(tracker3.can_request()) finally: + for t in (tracker1, tracker2, tracker3): + if t is not None: + t.dispose() if tmp_path.exists(): - tmp_path.unlink() + try: + tmp_path.unlink() + except Exception: + pass def test_circuit_breaker_behavior_and_recovery(self): cb = CircuitBreaker(failure_threshold=3, recovery_timeout_sec=0.1) @@ -394,5 +416,320 @@ class TestSecurityAndAuthGates(unittest.TestCase): self.assertEqual(validate_fixture_override("zyla_rome_sample.json"), "zyla_rome_sample.json") +class TestAtomicUsageTrackerAndRollover(unittest.TestCase): + def setUp(self): + self.temp_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + self.temp_file.close() + self.db_path = Path(self.temp_file.name) + self.db_url = f"sqlite:///{self.db_path.resolve().as_posix()}" + self.trackers_to_clean: list[PersistentUsageTracker] = [] + + def tearDown(self): + for tracker in self.trackers_to_clean: + tracker.dispose() + if self.db_path.exists(): + try: + self.db_path.unlink() + except Exception: + pass + + def test_concurrent_multi_worker_reservations_respect_strict_limit(self): + """ + P0: Proves that atomic reservations across multiple concurrent workers + strictly enforce the monthly allowance without over-allocation. + """ + limit = 10 + 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) + + num_threads = 30 + results = [] + + def worker_attempt(): + # Simulate a worker thread reserving and executing + reserved = tracker.reserve_request() + if reserved: + tracker.record_execution(success=True) + return reserved + + with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: + futures = [executor.submit(worker_attempt) for _ in range(num_threads)] + for future in concurrent.futures.as_completed(futures): + results.append(future.result()) + + # Exactly limit=10 succeeded, remaining 20 were denied + self.assertEqual(results.count(True), limit) + self.assertEqual(results.count(False), num_threads - limit) + + # Database state reflects exact limit + reserved_count, executed_count, cost = tracker.get_stats() + self.assertEqual(reserved_count, limit) + self.assertEqual(executed_count, limit) + self.assertAlmostEqual(cost, limit * 0.005, places=4) + self.assertFalse(tracker.can_request()) + + def test_monthly_rollover_preserves_history_and_resets_allowance(self): + """ + P0: Proves that monthly rollover automatically resets allowance for + a new month while preserving previous months' historical spend. + """ + limit = 5 + tracker = PersistentUsageTracker( + provider_id="zyla_4814", + monthly_budget_limit=limit, + cost_per_request_usd=0.01, + db_url=self.db_url, + ) + self.trackers_to_clean.append(tracker) + + oct_date = datetime(2026, 10, 15, 12, 0, tzinfo=timezone.utc) + nov_date = datetime(2026, 11, 1, 0, 0, tzinfo=timezone.utc) + + # 1. Exhaust budget in October + for _ in range(limit): + self.assertTrue(tracker.reserve_request(now=oct_date)) + tracker.record_execution(success=True, now=oct_date) + + self.assertFalse(tracker.can_request(now=oct_date)) + self.assertFalse(tracker.reserve_request(now=oct_date)) + + # 2. Rollover to November: fresh month starts with full allowance + self.assertTrue(tracker.can_request(now=nov_date)) + self.assertTrue(tracker.reserve_request(now=nov_date)) + tracker.record_execution(success=True, now=nov_date) + self.assertTrue(tracker.reserve_request(now=nov_date)) + tracker.record_execution(success=True, now=nov_date) + + # 3. Verify October history is preserved intact + oct_reserved, oct_executed, oct_cost = tracker.get_stats(now=oct_date) + self.assertEqual(oct_reserved, 5) + self.assertEqual(oct_executed, 5) + self.assertAlmostEqual(oct_cost, 0.05, places=4) + + # 4. Verify November reflects new usage + nov_reserved, nov_executed, nov_cost = tracker.get_stats(now=nov_date) + self.assertEqual(nov_reserved, 2) + self.assertEqual(nov_executed, 2) + self.assertAlmostEqual(nov_cost, 0.02, places=4) + + def test_fail_closed_when_database_is_unavailable(self): + """ + P0: Proves PersistentUsageTracker fails closed (rejects requests) + when database store is unavailable or down. + """ + bad_tracker = PersistentUsageTracker( + provider_id="zyla_4814", + monthly_budget_limit=100, + db_url="sqlite:///non_existent_dir_99999/unreachable.db", + ) + self.trackers_to_clean.append(bad_tracker) + + # Must return False rather than allowing untracked spend + self.assertFalse(bad_tracker.reserve_request()) + self.assertFalse(bad_tracker.can_request()) + + # Tracker with no engine at all + none_tracker = PersistentUsageTracker( + provider_id="zyla_4814", + monthly_budget_limit=100, + ) + self.assertFalse(none_tracker.reserve_request()) + self.assertFalse(none_tracker.can_request()) + + def test_fixture_requests_are_excluded_from_billable_usage(self): + """ + P0: Proves that fixture/test queries do not consume real billable reservations. + """ + tracker = PersistentUsageTracker( + provider_id="zyla_4814", + monthly_budget_limit=10, + db_url=self.db_url, + ) + self.trackers_to_clean.append(tracker) + + adapter = ZylaWazeAdapter( + api_key="test-key", + enabled=True, + usage_tracker=tracker, + fixture_mode=True, + ) + # Fetch alerts in fixture mode + alerts = adapter.fetch_alerts(min_lat=34.2, min_lng=-85.2, max_lat=34.3, max_lng=-85.1) + self.assertGreater(len(alerts), 0) + + # Excluded: reservations and executions must remain 0 + reserved, executed, _ = tracker.get_stats() + self.assertEqual(reserved, 0) + self.assertEqual(executed, 0) + + def test_reservation_release_on_pre_execution_cancellation(self): + """ + P0: Proves release_reservation safely restores reserved count if a call aborts. + """ + tracker = PersistentUsageTracker( + provider_id="zyla_4814", + monthly_budget_limit=5, + db_url=self.db_url, + ) + self.trackers_to_clean.append(tracker) + + self.assertTrue(tracker.reserve_request()) + reserved, executed, _ = tracker.get_stats() + self.assertEqual(reserved, 1) + self.assertEqual(executed, 0) + + tracker.release_reservation() + reserved, executed, _ = tracker.get_stats() + self.assertEqual(reserved, 0) + self.assertEqual(executed, 0) + + +class TestFirebaseGatewayAuthenticationIntegration(unittest.TestCase): + @classmethod + def setUpClass(cls): + # Generate an RSA keypair and self-signed certificate for Firebase auth tests + cls.private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + cls.untrusted_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + cls.now = datetime.now(timezone.utc) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "firebase-test-qa")]) + cls.certificate = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(cls.private_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(cls.now - timedelta(minutes=5)) + .not_valid_after(cls.now + timedelta(days=2)) + .sign(cls.private_key, algorithm=hashes.SHA256()) + ) + cls.cert_pem = cls.certificate.public_bytes(serialization.Encoding.PEM).decode("ascii") + + def setUp(self): + self.orig_env = dict(os.environ) + os.environ["FIREBASE_PROJECT_ID"] = "motoapp-d22e0" + os.environ["TESTING"] = "false" + os.environ.pop("MOTORSIGNAL_GATEWAY_SECRET", None) + os.environ.pop("MOTORSIGNAL_ADMIN_KEY", None) + + # Inject QA certificate into auth cert cache + auth._firebase_certs = {"qa-test-key-id": self.cert_pem} + auth._firebase_certs_expires_at = time.time() + 3600 + + def tearDown(self): + os.environ.clear() + os.environ.update(self.orig_env) + + def _create_token( + self, + sub: str = "firebase-rider-001", + aud: str = "motoapp-d22e0", + iss: str = "https://securetoken.google.com/motoapp-d22e0", + exp_delta_seconds: int = 300, + signing_key=None, + kid: str = "qa-test-key-id", + claims_override: Optional[dict] = None, + ) -> str: + key = signing_key or self.private_key + now_ts = int(time.time()) + payload = { + "sub": sub, + "aud": aud, + "iss": iss, + "iat": now_ts, + "exp": now_ts + exp_delta_seconds, + "email": f"{sub}@example.com", + "email_verified": True, + } + if claims_override: + payload.update(claims_override) + return jwt.encode(payload, key, algorithm="RS256", headers={"kid": kid}) + + def test_valid_qa_firebase_token_authenticates_gateway(self): + """ + P0: Proves a valid Firebase ID token for the configured QA project + (motoapp-d22e0) successfully authenticates against verify_gateway_or_user_auth. + """ + token = self._create_token(sub="verified-qa-rider") + auth_header = f"Bearer {token}" + self.assertTrue(verify_gateway_or_user_auth(authorization=auth_header, x_gateway_key=None)) + + def test_firebase_token_admin_claims_accepted_by_admin_route(self): + """ + P0: Proves verify_admin_auth accepts Firebase tokens containing trusted + custom claims: admin=True or role='admin'/'operator'/'owner'. + """ + # Token with custom claim admin: True + token_admin_bool = self._create_token(sub="admin-user-01", claims_override={"admin": True}) + self.assertTrue(verify_admin_auth(authorization=f"Bearer {token_admin_bool}", x_admin_key=None)) + + # Token with custom claim role: 'admin' + token_role_admin = self._create_token(sub="admin-user-02", claims_override={"role": "admin"}) + self.assertTrue(verify_admin_auth(authorization=f"Bearer {token_role_admin}", x_admin_key=None)) + + # Token with custom claim role: 'operator' + token_role_op = self._create_token(sub="op-user-01", claims_override={"role": "operator"}) + self.assertTrue(verify_admin_auth(authorization=f"Bearer {token_role_op}", x_admin_key=None)) + + def test_regular_firebase_user_rejected_by_admin_route(self): + """ + P0: Proves a valid authenticated Firebase rider without admin custom claims + is strictly rejected by verify_admin_auth with HTTP 403 Forbidden. + """ + token_regular = self._create_token(sub="normal-rider-alice") + with self.assertRaises(HTTPException) as ctx: + verify_admin_auth(authorization=f"Bearer {token_regular}", x_admin_key=None) + self.assertEqual(ctx.exception.status_code, 403) + + def test_expired_firebase_token_rejected(self): + """ + P0: Proves expired Firebase tokens are rejected with HTTP 401 Unauthorized. + """ + expired_token = self._create_token(sub="expired-user", exp_delta_seconds=-60) + with self.assertRaises(HTTPException) as ctx: + verify_gateway_or_user_auth(authorization=f"Bearer {expired_token}", x_gateway_key=None) + self.assertEqual(ctx.exception.status_code, 401) + + def test_wrong_project_firebase_token_rejected(self): + """ + P0: Proves tokens minted for a different Firebase project are rejected (HTTP 401). + """ + wrong_project_token = self._create_token(aud="unauthorized-foreign-project-xyz") + with self.assertRaises(HTTPException) as ctx: + verify_gateway_or_user_auth(authorization=f"Bearer {wrong_project_token}", x_gateway_key=None) + self.assertEqual(ctx.exception.status_code, 401) + + def test_invalid_signature_firebase_token_rejected(self): + """ + P0: Proves tokens with untrusted signatures are rejected with HTTP 401. + """ + forged_token = self._create_token(signing_key=self.untrusted_key) + with self.assertRaises(HTTPException) as ctx: + verify_gateway_or_user_auth(authorization=f"Bearer {forged_token}", x_gateway_key=None) + self.assertEqual(ctx.exception.status_code, 401) + + def test_production_environment_locks_out_test_token_bypasses(self): + """ + P0: Proves production environments (MOTORSIGNAL_ENV=production, ENVIRONMENT=production) + completely block any test token bypasses even if TESTING=true is present. + """ + os.environ["MOTORSIGNAL_ENV"] = "production" + os.environ["TESTING"] = "true" + + # Gateway auth must reject test_token_ in production + with self.assertRaises(HTTPException) as ctx: + verify_gateway_or_user_auth(authorization="Bearer test_token_bypass_123", x_gateway_key=None) + self.assertEqual(ctx.exception.status_code, 401) + + # Admin auth must reject test_admin_token_ in production + with self.assertRaises(HTTPException) as ctx: + verify_admin_auth(authorization="Bearer test_admin_token_bypass_123", x_admin_key=None) + self.assertEqual(ctx.exception.status_code, 403) + + if __name__ == "__main__": unittest.main()