85 lines
3.0 KiB
Python
85 lines
3.0 KiB
Python
"""Auth router — simple JWT for admin login + customer registration.
|
|
|
|
POST /api/auth/login (admin, env ADMIN_EMAIL / ADMIN_PASSWORD)
|
|
POST /api/auth/register (customer)
|
|
|
|
TODO(security): hash ADMIN_PASSWORD (bcrypt/argon2) instead of plain compare —
|
|
acceptable for MVP per task instructions. Customers table has no password
|
|
column yet, so registration is identity-only.
|
|
"""
|
|
import os
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import jwt
|
|
from fastapi import APIRouter, Depends, Header, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.models import Customer
|
|
from app.schemas import CustomerRegister, LoginRequest, TokenResponse
|
|
|
|
router = APIRouter()
|
|
|
|
SECRET_KEY = os.getenv("SECRET_KEY", "polaris-dev-secret-change-me")
|
|
ALGORITHM = "HS256"
|
|
TOKEN_TTL_HOURS = 24
|
|
|
|
|
|
def _make_token(email: str, role: str) -> str:
|
|
now = datetime.now(timezone.utc)
|
|
payload = {
|
|
"sub": email,
|
|
"role": role,
|
|
"iat": now,
|
|
"exp": now + timedelta(hours=TOKEN_TTL_HOURS),
|
|
}
|
|
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
|
|
|
|
|
|
@router.post("/login", response_model=TokenResponse)
|
|
def login(body: LoginRequest):
|
|
admin_email = os.getenv("ADMIN_EMAIL", "admin@polaris.local")
|
|
# contract mentions ADMIN_PASSWORD_HASH; MVP uses plain compare (see TODO above)
|
|
admin_password = os.getenv("ADMIN_PASSWORD") or os.getenv("ADMIN_PASSWORD_HASH", "admin")
|
|
|
|
if body.email.strip().lower() != admin_email.strip().lower():
|
|
raise HTTPException(status_code=401, detail="invalid credentials")
|
|
if body.password != admin_password:
|
|
raise HTTPException(status_code=401, detail="invalid credentials")
|
|
|
|
return TokenResponse(access_token=_make_token(body.email, "admin"))
|
|
|
|
|
|
@router.post("/register")
|
|
def register(body: CustomerRegister, db: Session = Depends(get_db)):
|
|
email = body.email.strip().lower()
|
|
existing = db.query(Customer).filter(Customer.email == email).first()
|
|
if existing is not None:
|
|
raise HTTPException(status_code=409, detail="email already registered")
|
|
|
|
customer = Customer(email=email, name=body.name, shipping_addr=body.shipping_addr or {})
|
|
db.add(customer)
|
|
db.commit()
|
|
db.refresh(customer)
|
|
|
|
return {
|
|
"id": str(customer.id),
|
|
"email": customer.email,
|
|
"name": customer.name,
|
|
"access_token": _make_token(customer.email, "customer"),
|
|
"token_type": "bearer",
|
|
}
|
|
|
|
|
|
def require_admin(authorization: str = Header(default="")):
|
|
"""FastAPI dependency — requires a valid admin JWT in the Authorization header."""
|
|
if not authorization.startswith("Bearer "):
|
|
raise HTTPException(status_code=401, detail="missing bearer token")
|
|
try:
|
|
payload = jwt.decode(authorization[7:], SECRET_KEY, algorithms=[ALGORITHM])
|
|
except jwt.PyJWTError:
|
|
raise HTTPException(status_code=401, detail="invalid token")
|
|
if payload.get("role") != "admin":
|
|
raise HTTPException(status_code=403, detail="admin role required")
|
|
return payload
|