From 87f2289fbc73ca59ca6725cf20468eb7b0b8652c Mon Sep 17 00:00:00 2001 From: Developer Date: Fri, 11 Sep 2026 03:42:29 -0400 Subject: [PATCH] fix: retire legacy admin authority --- backend/motorsignal_api/app/main.py | 337 ---------------- .../tests/test_legacy_admin_retirement.py | 377 ++++++++++++++++++ 2 files changed, 377 insertions(+), 337 deletions(-) create mode 100644 backend/motorsignal_api/tests/test_legacy_admin_retirement.py diff --git a/backend/motorsignal_api/app/main.py b/backend/motorsignal_api/app/main.py index a8f7be3..ba95988 100644 --- a/backend/motorsignal_api/app/main.py +++ b/backend/motorsignal_api/app/main.py @@ -104,28 +104,6 @@ class SessionResponse(BaseModel): current: bool -class AdminUserCreate(BaseModel): - email: str - password: str = Field(min_length=12, max_length=128) - role: str = "user" - - -class AdminUserUpdate(BaseModel): - email: str | None = None - role: str | None = None - is_active: bool | None = None - email_verified: bool | None = None - - -class VehicleCreate(BaseModel): - user_id: uuid.UUID - nickname: str = Field(min_length=1, max_length=120) - make: str = Field(min_length=1, max_length=120) - model: str = Field(min_length=1, max_length=120) - year: int | None = Field(default=None, ge=1885, le=2100) - vin: str | None = Field(default=None, min_length=11, max_length=17) - - class VehicleUpdate(BaseModel): nickname: str | None = Field(default=None, min_length=1, max_length=120) make: str | None = Field(default=None, min_length=1, max_length=120) @@ -134,26 +112,6 @@ class VehicleUpdate(BaseModel): vin: str | None = Field(default=None, min_length=11, max_length=17) -class RideCreate(BaseModel): - user_id: uuid.UUID - vehicle_id: uuid.UUID | None = None - started_at: datetime - ended_at: datetime | None = None - distance_miles: float | None = Field(default=None, ge=0) - top_speed_mph: float | None = Field(default=None, ge=0) - avg_speed_mph: float | None = Field(default=None, ge=0) - status: str = Field(default="completed", min_length=1, max_length=32) - - -class RideUpdate(BaseModel): - vehicle_id: uuid.UUID | None = None - ended_at: datetime | None = None - distance_miles: float | None = Field(default=None, ge=0) - top_speed_mph: float | None = Field(default=None, ge=0) - avg_speed_mph: float | None = Field(default=None, ge=0) - status: str | None = Field(default=None, min_length=1, max_length=32) - - class UserVehicleCreate(BaseModel): nickname: str = Field(min_length=1, max_length=120) make: str = Field(min_length=1, max_length=120) @@ -187,19 +145,6 @@ class TelemetryBatch(BaseModel): points: list[TelemetryPointInput] = Field(min_length=1, max_length=1000) -class SubscriptionUpdate(BaseModel): - plan: str = Field(min_length=1, max_length=32) - status: str = Field(min_length=1, max_length=32) - current_period_start: datetime | None = None - current_period_end: datetime | None = None - - -class EntitlementCreate(BaseModel): - key: str = Field(min_length=1, max_length=96) - value: dict | None = None - expires_at: datetime | None = None - - def get_db(): db = DbSession(engine) @@ -343,13 +288,6 @@ def get_current_user(identity: tuple[User, Session] = Depends(get_current_identi return identity[0] -def require_roles(*roles: str): - def dependency(user: User = Depends(get_current_user)) -> User: - if user.role not in roles: - raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions") - return user - return dependency - @app.get("/api/v1/health") def health(): @@ -692,16 +630,6 @@ def logout_all(identity: tuple[User, Session] = Depends(get_current_identity), d return None -@app.get("/api/v1/admin/audit-events") -def list_audit_events(user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): - events = db.scalars(select(AuditEvent).order_by(AuditEvent.created_at.desc()).limit(100)).all() - return [{"id": str(event.id), "event_type": event.event_type, "actor_user_id": str(event.actor_user_id) if event.actor_user_id else None, "target_user_id": str(event.target_user_id) if event.target_user_id else None, "created_at": event.created_at, "metadata": event.metadata_json} for event in events] - - -def user_payload(user: User) -> dict: - return {"id": str(user.id), "email": user.email, "role": user.role, "is_active": user.is_active, "email_verified": user.email_verified, "created_at": user.created_at} - - def vehicle_payload(vehicle: Vehicle) -> dict: return {"id": str(vehicle.id), "user_id": str(vehicle.user_id), "nickname": vehicle.nickname, "make": vehicle.make, "model": vehicle.model, "year": vehicle.year, "vin": vehicle.vin, "created_at": vehicle.created_at, "updated_at": vehicle.updated_at} @@ -710,216 +638,6 @@ def ride_payload(ride: Ride) -> dict: return {"id": str(ride.id), "user_id": str(ride.user_id), "vehicle_id": str(ride.vehicle_id) if ride.vehicle_id else None, "started_at": ride.started_at, "ended_at": ride.ended_at, "distance_miles": ride.distance_miles, "top_speed_mph": ride.top_speed_mph, "avg_speed_mph": ride.avg_speed_mph, "status": ride.status, "created_at": ride.created_at} -def validate_role(role: str) -> str: - if role not in {"user", "support", "admin"}: - raise HTTPException(status_code=400, detail="Invalid role") - return role - - -@app.get("/api/v1/admin/users") -def list_users(page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=100), search: str | None = None, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): - query = select(User) - count_query = select(func.count()).select_from(User) - if search: - pattern = f"%{search.strip().lower()}%" - condition = or_(func.lower(User.email).like(pattern), User.role == search.strip().lower()) - query = query.where(condition) - count_query = count_query.where(condition) - total = db.scalar(count_query) or 0 - users = db.scalars(query.order_by(User.created_at.desc()).offset((page - 1) * page_size).limit(page_size)).all() - return {"items": [user_payload(item) for item in users], "page": page, "page_size": page_size, "total": total} - - -@app.post("/api/v1/admin/users", status_code=201) -def create_user(request: AdminUserCreate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): - email = normalize_email(request.email) - try: - email = normalize_email(validate_email(email, check_deliverability=False).normalized) - except EmailNotValidError: - raise HTTPException(status_code=400, detail="Invalid email address") - if db.scalar(select(User).where(User.email == email)) is not None: - raise HTTPException(status_code=409, detail="Account already exists") - user = User(email=email, password_hash=hash_password(request.password), role=validate_role(request.role)) - db.add(user) - db.flush() - audit(db, "admin.user_created", actor_user_id=actor.id, target_user_id=user.id) - db.commit() - db.refresh(user) - return user_payload(user) - - -@app.get("/api/v1/admin/users/{user_id}") -def get_user(user_id: uuid.UUID, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): - target = db.get(User, user_id) - if target is None: - raise HTTPException(status_code=404, detail="User not found") - return user_payload(target) - - -@app.patch("/api/v1/admin/users/{user_id}") -def update_user(user_id: uuid.UUID, request: AdminUserUpdate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): - target = db.get(User, user_id) - if target is None: - raise HTTPException(status_code=404, detail="User not found") - changes = request.model_dump(exclude_unset=True) - if "email" in changes: - try: - changes["email"] = normalize_email(validate_email(changes["email"], check_deliverability=False).normalized) - except EmailNotValidError: - raise HTTPException(status_code=400, detail="Invalid email address") - duplicate = db.scalar(select(User).where(User.email == changes["email"], User.id != target.id)) - if duplicate is not None: - raise HTTPException(status_code=409, detail="Account already exists") - if "role" in changes: - changes["role"] = validate_role(changes["role"]) - for field, value in changes.items(): - setattr(target, field, value) - audit(db, "admin.user_updated", actor_user_id=actor.id, target_user_id=target.id, metadata={"fields": list(changes)}) - db.commit() - db.refresh(target) - return user_payload(target) - - -@app.delete("/api/v1/admin/users/{user_id}", status_code=204) -def deactivate_user(user_id: uuid.UUID, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): - target = db.get(User, user_id) - if target is None: - raise HTTPException(status_code=404, detail="User not found") - target.is_active = False - now = datetime.now(timezone.utc) - db.query(Session).filter(Session.user_id == target.id, Session.revoked_at.is_(None)).update({Session.revoked_at: now}, synchronize_session=False) - audit(db, "admin.user_deactivated", actor_user_id=actor.id, target_user_id=target.id) - db.commit() - return None - - -@app.get("/api/v1/admin/vehicles") -def list_vehicles(page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=100), user_id: uuid.UUID | None = None, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): - query = select(Vehicle) - count_query = select(func.count()).select_from(Vehicle) - if user_id: - query = query.where(Vehicle.user_id == user_id) - count_query = count_query.where(Vehicle.user_id == user_id) - total = db.scalar(count_query) or 0 - vehicles = db.scalars(query.order_by(Vehicle.created_at.desc()).offset((page - 1) * page_size).limit(page_size)).all() - return {"items": [vehicle_payload(item) for item in vehicles], "page": page, "page_size": page_size, "total": total} - - -@app.post("/api/v1/admin/vehicles", status_code=201) -def create_vehicle(request: VehicleCreate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): - if db.get(User, request.user_id) is None: - raise HTTPException(status_code=404, detail="User not found") - vehicle = Vehicle(**request.model_dump()) - db.add(vehicle) - db.flush() - audit(db, "admin.vehicle_created", actor_user_id=actor.id, target_user_id=vehicle.user_id, metadata={"vehicle_id": str(vehicle.id)}) - db.commit() - db.refresh(vehicle) - return vehicle_payload(vehicle) - - -@app.get("/api/v1/admin/vehicles/{vehicle_id}") -def get_vehicle(vehicle_id: uuid.UUID, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): - vehicle = db.get(Vehicle, vehicle_id) - if vehicle is None: - raise HTTPException(status_code=404, detail="Vehicle not found") - return vehicle_payload(vehicle) - - -@app.patch("/api/v1/admin/vehicles/{vehicle_id}") -def update_vehicle(vehicle_id: uuid.UUID, request: VehicleUpdate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): - vehicle = db.get(Vehicle, vehicle_id) - if vehicle is None: - raise HTTPException(status_code=404, detail="Vehicle not found") - changes = request.model_dump(exclude_unset=True) - for field, value in changes.items(): - setattr(vehicle, field, value) - audit(db, "admin.vehicle_updated", actor_user_id=actor.id, target_user_id=vehicle.user_id, metadata={"vehicle_id": str(vehicle.id), "fields": list(changes)}) - db.commit() - db.refresh(vehicle) - return vehicle_payload(vehicle) - - -@app.delete("/api/v1/admin/vehicles/{vehicle_id}", status_code=204) -def delete_vehicle(vehicle_id: uuid.UUID, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): - vehicle = db.get(Vehicle, vehicle_id) - if vehicle is None: - raise HTTPException(status_code=404, detail="Vehicle not found") - audit(db, "admin.vehicle_deleted", actor_user_id=actor.id, target_user_id=vehicle.user_id, metadata={"vehicle_id": str(vehicle.id)}) - db.delete(vehicle) - db.commit() - return None - - -@app.get("/api/v1/admin/rides") -def list_rides(page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=100), user_id: uuid.UUID | None = None, vehicle_id: uuid.UUID | None = None, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): - query = select(Ride) - count_query = select(func.count()).select_from(Ride) - if user_id: - query = query.where(Ride.user_id == user_id) - count_query = count_query.where(Ride.user_id == user_id) - if vehicle_id: - query = query.where(Ride.vehicle_id == vehicle_id) - count_query = count_query.where(Ride.vehicle_id == vehicle_id) - total = db.scalar(count_query) or 0 - rides = db.scalars(query.order_by(Ride.started_at.desc()).offset((page - 1) * page_size).limit(page_size)).all() - return {"items": [ride_payload(item) for item in rides], "page": page, "page_size": page_size, "total": total} - - -@app.post("/api/v1/admin/rides", status_code=201) -def create_ride(request: RideCreate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): - if db.get(User, request.user_id) is None: - raise HTTPException(status_code=404, detail="User not found") - if request.vehicle_id is not None: - vehicle = db.get(Vehicle, request.vehicle_id) - if vehicle is None or vehicle.user_id != request.user_id: - raise HTTPException(status_code=400, detail="Vehicle does not belong to user") - ride = Ride(**request.model_dump()) - db.add(ride) - db.flush() - audit(db, "admin.ride_created", actor_user_id=actor.id, target_user_id=ride.user_id, metadata={"ride_id": str(ride.id)}) - db.commit() - db.refresh(ride) - return ride_payload(ride) - - -@app.get("/api/v1/admin/rides/{ride_id}") -def get_ride(ride_id: uuid.UUID, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): - ride = db.get(Ride, ride_id) - if ride is None: - raise HTTPException(status_code=404, detail="Ride not found") - return ride_payload(ride) - - -@app.patch("/api/v1/admin/rides/{ride_id}") -def update_ride(ride_id: uuid.UUID, request: RideUpdate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): - ride = db.get(Ride, ride_id) - if ride is None: - raise HTTPException(status_code=404, detail="Ride not found") - changes = request.model_dump(exclude_unset=True) - if "vehicle_id" in changes and changes["vehicle_id"] is not None: - vehicle = db.get(Vehicle, changes["vehicle_id"]) - if vehicle is None or vehicle.user_id != ride.user_id: - raise HTTPException(status_code=400, detail="Vehicle does not belong to user") - for field, value in changes.items(): - setattr(ride, field, value) - audit(db, "admin.ride_updated", actor_user_id=actor.id, target_user_id=ride.user_id, metadata={"ride_id": str(ride.id), "fields": list(changes)}) - db.commit() - db.refresh(ride) - return ride_payload(ride) - - -@app.delete("/api/v1/admin/rides/{ride_id}", status_code=204) -def archive_ride(ride_id: uuid.UUID, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): - ride = db.get(Ride, ride_id) - if ride is None: - raise HTTPException(status_code=404, detail="Ride not found") - ride.status = "archived" - audit(db, "admin.ride_archived", actor_user_id=actor.id, target_user_id=ride.user_id, metadata={"ride_id": str(ride.id)}) - db.commit() - return None - - @app.get("/api/v1/vehicles") def list_my_vehicles(user: User = Depends(get_current_user), db: DbSession = Depends(get_db)): vehicles = db.scalars(select(Vehicle).where(Vehicle.user_id == user.id).order_by(Vehicle.created_at.desc())).all() @@ -1043,20 +761,6 @@ def my_telemetry_summary(ride_id: uuid.UUID, user: User = Depends(get_current_us return telemetry_summary(ride_id, db) -@app.get("/api/v1/admin/rides/{ride_id}/telemetry") -def list_admin_telemetry(ride_id: uuid.UUID, limit: int = Query(default=1000, ge=1, le=5000), user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): - if db.get(Ride, ride_id) is None: - raise HTTPException(status_code=404, detail="Ride not found") - points = db.scalars(select(TelemetryPoint).where(TelemetryPoint.ride_id == ride_id).order_by(TelemetryPoint.recorded_at.asc()).limit(limit)).all() - return {"items": [telemetry_payload(point) for point in points], "count": len(points)} - - -@app.get("/api/v1/admin/rides/{ride_id}/telemetry/summary") -def admin_telemetry_summary(ride_id: uuid.UUID, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): - if db.get(Ride, ride_id) is None: - raise HTTPException(status_code=404, detail="Ride not found") - return telemetry_summary(ride_id, db) - def subscription_payload(subscription: Subscription) -> dict: return {"id": str(subscription.id), "user_id": str(subscription.user_id), "plan": subscription.plan, "status": subscription.status, "current_period_start": subscription.current_period_start, "current_period_end": subscription.current_period_end, "created_at": subscription.created_at, "updated_at": subscription.updated_at} @@ -1089,47 +793,6 @@ def list_my_entitlements(user: User = Depends(get_current_user), db: DbSession = return [entitlement_payload(item) for item in entitlements if item.expires_at is None or item.expires_at > datetime.now(timezone.utc)] -@app.get("/api/v1/admin/subscriptions") -def list_subscriptions(page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=100), user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): - total = db.scalar(select(func.count()).select_from(Subscription)) or 0 - subscriptions = db.scalars(select(Subscription).order_by(Subscription.updated_at.desc()).offset((page - 1) * page_size).limit(page_size)).all() - return {"items": [subscription_payload(item) for item in subscriptions], "page": page, "page_size": page_size, "total": total} - - -@app.patch("/api/v1/admin/users/{user_id}/subscription") -def update_subscription(user_id: uuid.UUID, request: SubscriptionUpdate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): - if db.get(User, user_id) is None: - raise HTTPException(status_code=404, detail="User not found") - subscription = get_or_create_subscription(user_id, db) - subscription.plan = request.plan - subscription.status = request.status - subscription.current_period_start = request.current_period_start - subscription.current_period_end = request.current_period_end - audit(db, "admin.subscription_updated", actor_user_id=actor.id, target_user_id=user_id, metadata={"plan": request.plan, "status": request.status}) - db.commit() - db.refresh(subscription) - return subscription_payload(subscription) - - -@app.get("/api/v1/admin/users/{user_id}/entitlements") -def list_entitlements(user_id: uuid.UUID, user: User = Depends(require_roles("admin", "support")), db: DbSession = Depends(get_db)): - if db.get(User, user_id) is None: - raise HTTPException(status_code=404, detail="User not found") - return [entitlement_payload(item) for item in db.scalars(select(Entitlement).where(Entitlement.user_id == user_id).order_by(Entitlement.created_at.desc())).all()] - - -@app.post("/api/v1/admin/users/{user_id}/entitlements", status_code=201) -def create_entitlement(user_id: uuid.UUID, request: EntitlementCreate, actor: User = Depends(require_roles("admin")), db: DbSession = Depends(get_db)): - if db.get(User, user_id) is None: - raise HTTPException(status_code=404, detail="User not found") - entitlement = Entitlement(user_id=user_id, **request.model_dump()) - db.add(entitlement) - db.flush() - audit(db, "admin.entitlement_created", actor_user_id=actor.id, target_user_id=user_id, metadata={"key": entitlement.key}) - db.commit() - db.refresh(entitlement) - return entitlement_payload(entitlement) - # ============================================================ # Operations (OPS-3B / OPS-3C / OPS-4) Platform Endpoints diff --git a/backend/motorsignal_api/tests/test_legacy_admin_retirement.py b/backend/motorsignal_api/tests/test_legacy_admin_retirement.py new file mode 100644 index 0000000..ee42b54 --- /dev/null +++ b/backend/motorsignal_api/tests/test_legacy_admin_retirement.py @@ -0,0 +1,377 @@ +import hashlib +import os +import sys +import unittest +import uuid +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +ROOT = os.path.dirname(os.path.dirname(__file__)) +sys.path.insert(0, os.path.join(ROOT, ".python-packages")) +sys.path.insert(0, os.path.join(ROOT, "app")) + +os.environ.setdefault("JWT_SECRET", "local-test-secret-at-least-32-chars-long-for-hmac") +os.environ["FIREBASE_PROJECT_ID"] = "motoapp-d22e0" +os.environ["DATABASE_USER"] = "test" +os.environ["DATABASE_PASSWORD"] = "test" +os.environ["DATABASE_NAME"] = "test" + +from sqlalchemy import create_engine, event, select, text +from sqlalchemy.orm import Session as DbSession +from starlette.testclient import TestClient + +import main +from main import app, get_db +from auth import create_access_token +from models import Base, Operator, OpsAuditEvent, OpsIncident, OpsSession, Session, User + + +def hash_token(raw_token: str) -> str: + return hashlib.sha256(raw_token.encode()).hexdigest() + + +class TestLegacyAdminRetirement(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.db_path = os.path.join(ROOT, "tests", "test_retirement.db") + cls.ops_db_path = os.path.join(ROOT, "tests", "test_retirement_ops.db") + for p in (cls.db_path, cls.ops_db_path): + if os.path.exists(p): + try: + os.remove(p) + except OSError: + pass + + cls.engine = create_engine(f"sqlite:///{cls.db_path}") + + cls.seq_val = 0 + + @event.listens_for(cls.engine, "connect") + def attach_ops(dbapi_con, con_record): + dbapi_con.execute(f"ATTACH DATABASE '{cls.ops_db_path}' AS ops") + + def sqlite_nextval(name): + cls.seq_val += 1 + return cls.seq_val + + dbapi_con.create_function("nextval", 1, sqlite_nextval) + + Base.metadata.create_all(cls.engine) + + def override_get_db(): + db = DbSession(cls.engine) + try: + yield db + finally: + db.close() + + app.dependency_overrides[get_db] = override_get_db + cls.client = TestClient(app, raise_server_exceptions=False) + cls.patcher = patch("main.check_database", return_value=True) + cls.patcher.start() + + @classmethod + def tearDownClass(cls): + cls.patcher.stop() + app.dependency_overrides.clear() + cls.engine.dispose() + for p in (cls.db_path, cls.ops_db_path): + if os.path.exists(p): + try: + os.remove(p) + except OSError: + pass + + def setUp(self): + with DbSession(self.engine) as db: + db.query(Session).delete() + db.query(User).delete() + db.query(OpsAuditEvent).delete() + db.query(OpsSession).delete() + db.query(Operator).delete() + db.commit() + + def _create_user(self, email: str, role: str) -> tuple[User, str]: + with DbSession(self.engine) as db: + user = User( + id=uuid.uuid4(), + email=email, + password_hash="test-hash", + role=role, + is_active=True, + email_verified=True, + ) + db.add(user) + db.flush() + + session = Session( + id=uuid.uuid4(), + user_id=user.id, + refresh_token_hash="test-refresh-hash", + expires_at=datetime.now(timezone.utc) + timedelta(days=7), + ) + db.add(session) + db.commit() + db.refresh(user) + db.refresh(session) + + token = create_access_token(user.id, session.id) + return user, token + + def _create_operator(self, identity: str, role: str) -> tuple[Operator, str]: + op_id = uuid.uuid4() + raw_token = f"raw-ops-token-{uuid.uuid4()}" + now = datetime.now(timezone.utc) + expires = now + timedelta(hours=8) + + with DbSession(self.engine) as db: + op = Operator( + id=op_id, + identity=identity, + role=role, + status="active", + created_at=now, + ) + db.add(op) + db.flush() + + ops_session = OpsSession( + id=uuid.uuid4(), + operator_id=op.id, + session_token_hash=hash_token(raw_token), + created_at=now, + last_activity_at=now, + expires_at=expires, + ) + db.add(ops_session) + db.commit() + db.refresh(op) + return op, raw_token + + # ============================================================ + # 1. PROGRAMMATIC ROUTE TABLE AUDIT + # ============================================================ + + def test_fastapi_route_table_contains_zero_admin_routes(self): + """Assert zero registered routes start with /api/v1/admin or contain /admin.""" + admin_routes = [ + r.path for r in app.routes + if getattr(r, "path", "").startswith("/api/v1/admin") + ] + self.assertEqual( + admin_routes, + [], + f"FastAPI route table still contains admin routes: {admin_routes}", + ) + self.assertEqual(len(admin_routes), 0) + + # ============================================================ + # 2. UNAUTHENTICATED PROBES -> 404 (ROUTING ABSENCE) + # ============================================================ + + def test_unauthenticated_requests_to_legacy_admin_routes_return_404(self): + """All former legacy admin routes must return 404, not 401/403.""" + test_uuid = str(uuid.uuid4()) + legacy_endpoints = [ + ("GET", "/api/v1/admin/users"), + ("POST", "/api/v1/admin/users"), + ("GET", f"/api/v1/admin/users/{test_uuid}"), + ("PATCH", f"/api/v1/admin/users/{test_uuid}"), + ("DELETE", f"/api/v1/admin/users/{test_uuid}"), + ("GET", "/api/v1/admin/vehicles"), + ("POST", "/api/v1/admin/vehicles"), + ("GET", f"/api/v1/admin/vehicles/{test_uuid}"), + ("PATCH", f"/api/v1/admin/vehicles/{test_uuid}"), + ("DELETE", f"/api/v1/admin/vehicles/{test_uuid}"), + ("GET", "/api/v1/admin/rides"), + ("POST", "/api/v1/admin/rides"), + ("GET", f"/api/v1/admin/rides/{test_uuid}"), + ("PATCH", f"/api/v1/admin/rides/{test_uuid}"), + ("DELETE", f"/api/v1/admin/rides/{test_uuid}"), + ("GET", f"/api/v1/admin/rides/{test_uuid}/telemetry"), + ("GET", f"/api/v1/admin/rides/{test_uuid}/telemetry/summary"), + ("GET", "/api/v1/admin/audit-events"), + ("GET", "/api/v1/admin/subscriptions"), + ("PATCH", f"/api/v1/admin/users/{test_uuid}/subscription"), + ("GET", f"/api/v1/admin/users/{test_uuid}/entitlements"), + ("POST", f"/api/v1/admin/users/{test_uuid}/entitlements"), + ] + + for method, path in legacy_endpoints: + response = self.client.request(method, path) + self.assertEqual( + response.status_code, + 404, + f"Expected 404 for unauthenticated {method} {path}, got {response.status_code}", + ) + + # ============================================================ + # 3. FORMER PRIVILEGED IDENTITIES -> 404 (ROUTING ABSENCE) + # ============================================================ + + def test_legacy_admin_identity_returns_404_on_legacy_admin_routes(self): + """User with role='admin' receives 404 on legacy admin paths (absence, not denial).""" + _, admin_token = self._create_user("admin@motorsignal.test", "admin") + headers = {"Authorization": f"Bearer {admin_token}"} + test_uuid = str(uuid.uuid4()) + + representative_paths = [ + ("GET", "/api/v1/admin/users"), + ("GET", "/api/v1/admin/vehicles"), + ("GET", "/api/v1/admin/rides"), + ("GET", "/api/v1/admin/audit-events"), + ("GET", "/api/v1/admin/subscriptions"), + ("GET", f"/api/v1/admin/users/{test_uuid}"), + ("GET", f"/api/v1/admin/rides/{test_uuid}/telemetry"), + ("GET", f"/api/v1/admin/users/{test_uuid}/entitlements"), + ] + + for method, path in representative_paths: + response = self.client.request(method, path, headers=headers) + self.assertEqual( + response.status_code, + 404, + f"Expected 404 for admin token on {method} {path}, got {response.status_code}", + ) + + def test_legacy_support_identity_returns_404_on_legacy_admin_routes(self): + """User with role='support' receives 404 on legacy admin paths (absence, not denial).""" + _, support_token = self._create_user("support@motorsignal.test", "support") + headers = {"Authorization": f"Bearer {support_token}"} + + representative_paths = [ + ("GET", "/api/v1/admin/users"), + ("GET", "/api/v1/admin/vehicles"), + ("GET", "/api/v1/admin/rides"), + ("GET", "/api/v1/admin/audit-events"), + ("GET", "/api/v1/admin/subscriptions"), + ] + + for method, path in representative_paths: + response = self.client.request(method, path, headers=headers) + self.assertEqual( + response.status_code, + 404, + f"Expected 404 for support token on {method} {path}, got {response.status_code}", + ) + + def test_standard_user_identity_returns_404_on_legacy_admin_routes(self): + """User with role='user' receives 404 on legacy admin paths.""" + _, user_token = self._create_user("rider@motorsignal.test", "user") + headers = {"Authorization": f"Bearer {user_token}"} + + representative_paths = [ + ("GET", "/api/v1/admin/users"), + ("GET", "/api/v1/admin/vehicles"), + ("GET", "/api/v1/admin/rides"), + ("GET", "/api/v1/admin/audit-events"), + ("GET", "/api/v1/admin/subscriptions"), + ] + + for method, path in representative_paths: + response = self.client.request(method, path, headers=headers) + self.assertEqual( + response.status_code, + 404, + f"Expected 404 for user token on {method} {path}, got {response.status_code}", + ) + + # ============================================================ + # 4. NEGATIVE PRIVILEGE TESTS AGAINST /api/v1/ops/* + # ============================================================ + + def test_legacy_admin_user_cannot_access_ops_endpoints(self): + """User with public.users.role='admin' has NO Operations authority.""" + _, admin_jwt = self._create_user("admin@motorsignal.test", "admin") + headers = {"Authorization": f"Bearer {admin_jwt}"} + + ops_endpoints = [ + ("GET", "/api/v1/ops/me"), + ("GET", "/api/v1/ops/audit-events"), + ("GET", "/api/v1/ops/incidents"), + ("POST", "/api/v1/ops/incidents"), + ("DELETE", "/api/v1/ops/sessions/current"), + ] + + for method, path in ops_endpoints: + response = self.client.request(method, path, headers=headers) + self.assertEqual( + response.status_code, + 401, + f"Legacy admin JWT unexpectedly gained access ({response.status_code}) to {method} {path}", + ) + + def test_legacy_support_user_cannot_access_ops_endpoints(self): + """User with public.users.role='support' has NO Operations authority.""" + _, support_jwt = self._create_user("support@motorsignal.test", "support") + headers = {"Authorization": f"Bearer {support_jwt}"} + + for path in ("/api/v1/ops/me", "/api/v1/ops/audit-events", "/api/v1/ops/incidents"): + response = self.client.get(path, headers=headers) + self.assertEqual( + response.status_code, + 401, + f"Legacy support JWT unexpectedly gained access to {path}", + ) + + def test_standard_rider_user_cannot_access_ops_endpoints(self): + """User with public.users.role='user' has NO Operations authority.""" + _, user_jwt = self._create_user("rider@motorsignal.test", "user") + headers = {"Authorization": f"Bearer {user_jwt}"} + + for path in ("/api/v1/ops/me", "/api/v1/ops/audit-events", "/api/v1/ops/incidents"): + response = self.client.get(path, headers=headers) + self.assertEqual( + response.status_code, + 401, + f"Rider JWT unexpectedly gained access to {path}", + ) + + def test_firebase_claims_cannot_access_ops_endpoints(self): + """Firebase tokens / claims have NO Operations authority.""" + headers = {"Authorization": "Bearer eyJhbGciOiJSUzI1NiIsImtpZCI6InRlc3QifQ.eyJzdWIiOiJmYi0xMjMifQ.sig"} + for path in ("/api/v1/ops/me", "/api/v1/ops/audit-events", "/api/v1/ops/incidents"): + response = self.client.get(path, headers=headers) + self.assertEqual( + response.status_code, + 401, + f"Firebase token unexpectedly gained access to {path}", + ) + + # ============================================================ + # 5. OPS AUTHORITY IS PRESERVED AND OPERATIONAL + # ============================================================ + + def test_valid_operator_can_access_ops_me_and_health(self): + """Operator role in ops.operators succeeds on /ops/me and /ops/health.""" + _, op_token = self._create_operator("operator1@motorsignal.internal", "operator") + headers = {"Authorization": f"Bearer {op_token}"} + + me_res = self.client.get("/api/v1/ops/me", headers=headers) + self.assertEqual(me_res.status_code, 200) + self.assertEqual(me_res.json()["role"], "operator") + + health_res = self.client.get("/api/v1/ops/health", headers=headers) + self.assertEqual(health_res.status_code, 200) + self.assertEqual(health_res.json()["status"], "available") + + def test_operator_role_forbidden_from_owner_only_audit_events(self): + """Operator role receives 403 on owner-only /ops/audit-events.""" + _, op_token = self._create_operator("operator1@motorsignal.internal", "operator") + headers = {"Authorization": f"Bearer {op_token}"} + + audit_res = self.client.get("/api/v1/ops/audit-events", headers=headers) + self.assertEqual(audit_res.status_code, 403) + + def test_owner_role_can_access_ops_audit_events(self): + """Owner role in ops.operators succeeds on /ops/audit-events.""" + _, owner_token = self._create_operator("owner1@motorsignal.internal", "owner") + headers = {"Authorization": f"Bearer {owner_token}"} + + audit_res = self.client.get("/api/v1/ops/audit-events", headers=headers) + self.assertEqual(audit_res.status_code, 200) + self.assertIn("items", audit_res.json()) + + +if __name__ == "__main__": + unittest.main()