fix: retire legacy admin authority
This commit is contained in:
Developer
parent
304d2db108
commit
87f2289fbc
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
Loading…
Reference in New Issue