121 lines
4.2 KiB
Python
121 lines
4.2 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import AuthSession, require_admin_session, require_client_session
|
|
from app.core.config import settings
|
|
from app.core.security import issue_token
|
|
from app.db.session import get_db
|
|
from app.models.client_access import ClientAccount
|
|
from app.services.client_access_service import get_client_user_by_email, module_access_map
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
email: str
|
|
password: str
|
|
|
|
|
|
class SessionResponse(BaseModel):
|
|
name: str
|
|
email: str
|
|
role: str
|
|
tenant_id: str | None = None
|
|
client_role: str | None = None
|
|
user_id: int | None = None
|
|
client_account_id: int | None = None
|
|
module_permissions: dict[str, str] = Field(default_factory=dict)
|
|
token: str
|
|
|
|
|
|
def _build_session_response(
|
|
*,
|
|
name: str,
|
|
email: str,
|
|
role: str,
|
|
tenant_id: str | None = None,
|
|
client_role: str | None = None,
|
|
user_id: int | None = None,
|
|
client_account_id: int | None = None,
|
|
module_permissions: dict[str, str] | None = None,
|
|
) -> SessionResponse:
|
|
token = issue_token(
|
|
{
|
|
"name": name,
|
|
"email": email,
|
|
"role": role,
|
|
"tenant_id": tenant_id,
|
|
"client_role": client_role,
|
|
"user_id": user_id,
|
|
"client_account_id": client_account_id,
|
|
}
|
|
)
|
|
return SessionResponse(
|
|
name=name,
|
|
email=email,
|
|
role=role,
|
|
tenant_id=tenant_id,
|
|
client_role=client_role,
|
|
user_id=user_id,
|
|
client_account_id=client_account_id,
|
|
module_permissions=module_permissions or {},
|
|
token=token,
|
|
)
|
|
|
|
|
|
@router.post("/client/login", response_model=SessionResponse)
|
|
def client_login(payload: LoginRequest, db: Session = Depends(get_db)):
|
|
if payload.password != settings.client_password:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid client email or password")
|
|
|
|
user = get_client_user_by_email(db, email=payload.email.strip().lower())
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid client email or password")
|
|
|
|
client_account = db.scalar(select(ClientAccount).where(ClientAccount.id == user.client_account_id))
|
|
if client_account is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Client account is not configured for this user")
|
|
|
|
return _build_session_response(
|
|
name=user.full_name,
|
|
email=user.email,
|
|
role="client",
|
|
tenant_id=client_account.tenant_id,
|
|
client_role=user.role,
|
|
user_id=user.id,
|
|
client_account_id=client_account.id,
|
|
module_permissions=module_access_map(user),
|
|
)
|
|
|
|
|
|
@router.post("/admin/login", response_model=SessionResponse)
|
|
def admin_login(payload: LoginRequest):
|
|
if payload.email.strip().lower() != settings.admin_email.lower() or payload.password != settings.admin_password:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid admin email or password")
|
|
|
|
return _build_session_response(name=settings.admin_name, email=settings.admin_email, role="admin")
|
|
|
|
|
|
@router.get("/client/session", response_model=SessionResponse)
|
|
def read_client_session(session: AuthSession = Depends(require_client_session), db: Session = Depends(get_db)):
|
|
user = get_client_user_by_email(db, email=session.email, tenant_id=session.tenant_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Client user session is no longer available")
|
|
return _build_session_response(
|
|
name=user.full_name,
|
|
email=user.email,
|
|
role=session.role,
|
|
tenant_id=session.tenant_id,
|
|
client_role=user.role,
|
|
user_id=user.id,
|
|
client_account_id=user.client_account_id,
|
|
module_permissions=module_access_map(user),
|
|
)
|
|
|
|
|
|
@router.get("/admin/session", response_model=SessionResponse)
|
|
def read_admin_session(session: AuthSession = Depends(require_admin_session)):
|
|
return _build_session_response(name=session.name, email=session.email, role=session.role)
|