60 lines
1.7 KiB
Python
60 lines
1.7 KiB
Python
from fastapi import APIRouter
|
|
from shared.firebase import get_db
|
|
from auth.models import LoginRequest, TokenResponse
|
|
from auth.utils import verify_password, create_access_token
|
|
from shared.exceptions import AuthenticationError
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
|
|
|
|
@router.post("/login", response_model=TokenResponse)
|
|
async def login(body: LoginRequest):
|
|
db = get_db()
|
|
if not db:
|
|
raise AuthenticationError("Service unavailable")
|
|
|
|
users_ref = db.collection("admin_users")
|
|
query = users_ref.where("email", "==", body.email).limit(1).get()
|
|
|
|
if not query:
|
|
raise AuthenticationError("Invalid email or password")
|
|
|
|
doc = query[0]
|
|
user_data = doc.to_dict()
|
|
|
|
if not user_data.get("is_active", True):
|
|
raise AuthenticationError("Account is disabled")
|
|
|
|
if not verify_password(body.password, user_data["hashed_password"]):
|
|
raise AuthenticationError("Invalid email or password")
|
|
|
|
role = user_data["role"]
|
|
# Map legacy roles to new roles
|
|
role_mapping = {
|
|
"superadmin": "sysadmin",
|
|
"melody_editor": "editor",
|
|
"device_manager": "editor",
|
|
"user_manager": "editor",
|
|
"viewer": "user",
|
|
}
|
|
role = role_mapping.get(role, role)
|
|
|
|
token = create_access_token({
|
|
"sub": doc.id,
|
|
"email": user_data["email"],
|
|
"role": role,
|
|
"name": user_data["name"],
|
|
})
|
|
|
|
# Get permissions for editor/user roles
|
|
permissions = None
|
|
if role in ("editor", "user"):
|
|
permissions = user_data.get("permissions")
|
|
|
|
return TokenResponse(
|
|
access_token=token,
|
|
role=role,
|
|
name=user_data["name"],
|
|
permissions=permissions,
|
|
)
|