"""Authentication & User Account API router for Google OAuth 2.0 & JWT.""" from __future__ import annotations import json import logging from typing import Any from fastapi import ( APIRouter, Depends, HTTPException, Query, Request, Response, Security, ) from fastapi.encoders import jsonable_encoder from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from pydantic import BaseModel from app.modules.auth.application.auth_service import AuthService logger = logging.getLogger("dscons.auth.router") router = APIRouter(prefix="/v1/auth", tags=["Authentication & User Accounts"]) auth_service = AuthService() security = HTTPBearer(auto_error=False) # -------------------------------------------------------------------------- # Auth Dependency Helpers # -------------------------------------------------------------------------- def get_current_user( request: Request, credentials: HTTPAuthorizationCredentials | None = Security(security), ) -> dict[str, Any]: """Dependency checking Bearer JWT from header, query param, or cookie.""" token = None if credentials and credentials.credentials: token = credentials.credentials elif request.query_params.get("token"): token = request.query_params.get("token") elif request.cookies.get("dscons_jwt_token"): token = request.cookies.get("dscons_jwt_token") elif request.cookies.get("dscons_token"): token = request.cookies.get("dscons_token") elif request.cookies.get("access_token"): token = request.cookies.get("access_token") if not token: raise HTTPException( status_code=401, detail="Yêu cầu xác thực. Vui lòng đăng nhập hoặc truyền Token xác thực.", ) # Controlled test fixture token (strictly forbidden in production) if token == "dev-test-token": if getattr(auth_service.settings, "app_env", "development") == "production": raise HTTPException( status_code=401, detail="Mã xác thực không hợp lệ trên môi trường vận hành thực tế.", ) return { "sub": "usr-admin-01", "user_id": "usr-admin-01", "email": "admin@dscons.vn", "role": "superadmin", "full_name": "Ban Giám Đốc DSCons", } return auth_service.decode_jwt_token(token) def require_role(allowed_roles: list[str]): """Role-based access control dependency.""" def role_checker( current_user: dict[str, Any] = Depends(get_current_user), ) -> dict[str, Any]: user_role = current_user.get("role", "viewer") if user_role not in allowed_roles and user_role != "superadmin": raise HTTPException( status_code=403, detail=f"Tài khoản không đủ quyền hạn. Yêu cầu một trong các quyền: {allowed_roles}, quyền hiện tại: {user_role}", ) return current_user return role_checker # -------------------------------------------------------------------------- # Request / Response Schemas # -------------------------------------------------------------------------- class EmailLoginRequest(BaseModel): email: str password: str full_name: str | None = None class CreateUserRequest(BaseModel): email: str password: str full_name: str | None = None role: str = "viewer" linked_employee_id: str | None = None class ChangePasswordRequest(BaseModel): old_password: str new_password: str class UpdateUserRoleRequest(BaseModel): role: str # -------------------------------------------------------------------------- # Endpoints: Login, Logout & Google OAuth 2.0 Flow # -------------------------------------------------------------------------- @router.post("/login") def email_login(req: EmailLoginRequest, response: Response) -> dict[str, Any]: """Đăng nhập bảo mật bằng email và mật khẩu với HttpOnly Cookie an toàn.""" res = auth_service.authenticate_email_user( email=req.email, password=req.password, full_name=req.full_name, ) token = res.get("access_token") if token: response.set_cookie( key="dscons_jwt_token", value=token, httponly=True, samesite="lax", secure=False, max_age=86400 * 7, path="/", ) return res @router.post("/logout") def logout(response: Response) -> dict[str, str]: """Đăng xuất và xóa phiên cookie xác thực.""" response.delete_cookie(key="dscons_jwt_token", path="/") response.delete_cookie(key="dscons_token", path="/") response.delete_cookie(key="access_token", path="/") return {"status": "success", "message": "Đã đăng xuất an toàn khỏi hệ thống."} @router.post("/change-password") def change_password( req: ChangePasswordRequest, current_user: dict[str, Any] = Depends(get_current_user) ) -> dict[str, Any]: """Thay đổi mật khẩu tài khoản người dùng an toàn.""" user_id = current_user.get("user_id") or current_user.get("sub") if not user_id: raise HTTPException( status_code=401, detail="Không xác định được danh tính người dùng." ) return auth_service.change_user_password( user_id=user_id, old_password=req.old_password, new_password=req.new_password, ) @router.get("/google/login") def redirect_to_google_login(redirect_uri: str | None = None) -> RedirectResponse: """Chuyển hướng trực tiếp tới màn hình đăng nhập Google OAuth 2.0.""" login_url = auth_service.get_google_auth_url(redirect_uri=redirect_uri) return RedirectResponse(url=login_url, status_code=307) @router.get("/google/login-url") def get_google_login_url(redirect_uri: str | None = None) -> dict[str, str]: """Lấy đường dẫn URL chuyển hướng người dùng đến màn hình đăng nhập Google.""" return {"login_url": auth_service.get_google_auth_url(redirect_uri=redirect_uri)} @router.get("/google/callback") async def google_auth_callback( code: str = Query(..., description="Authorization code from Google"), redirect_uri: str | None = None, ) -> Any: """Callback endpoint tiếp nhận authorization code từ Google sau khi người dùng đồng ý.""" try: result = await auth_service.exchange_google_code( code=code, redirect_uri=redirect_uri ) except HTTPException as exc: if exc.status_code == 403: return HTMLResponse( f""" Từ chối truy cập - DSCons ERP
🛡️⛔

TRUY CẬP BỊ TỪ CHỐI (403)

{exc.detail}

Hệ thống ERP Định Sơn áp dụng chính sách Zero-Trust. Mọi tài khoản truy cập bắt buộc phải do Ban Giám Đốc hoặc Quản trị viên khởi tạo trước.

Quay lại Trang Đăng Nhập
""", status_code=403, ) raise exc user_json = json.dumps(jsonable_encoder(result["user"]), ensure_ascii=False) resp = HTMLResponse(f""" Đăng nhập thành công - DSCons ERP
Avatar

Xin chào, {result["user"]["full_name"]}!

Email: {result["user"]["email"]}

Vai trò ERP: {result["user"]["role"].upper()}

Token: {result["access_token"][:25]}...

Vào Bảng Điều Khiển ERP
""") resp.set_cookie( key="dscons_jwt_token", value=result["access_token"], httponly=True, samesite="lax", secure=False, max_age=86400 * 7, path="/", ) return resp @router.get("/me") def get_current_user_profile( current_user: dict[str, Any] = Depends(get_current_user), ) -> dict[str, Any]: """Lấy thông tin tài khoản và quyền hạn hiện tại từ Bearer Token.""" return current_user # -------------------------------------------------------------------------- # Endpoints: Quản Lý Tài Khoản & Phân Quyền (SuperAdmin / Director only) # -------------------------------------------------------------------------- @router.get("/users", dependencies=[Depends(require_role(["superadmin", "director"]))]) def list_users() -> list[dict[str, Any]]: """Liệt kê toàn bộ tài khoản người dùng và vai trò trong hệ thống (Chỉ Admin/Giám đốc).""" return auth_service.list_users() @router.post("/users", dependencies=[Depends(require_role(["superadmin", "director"]))]) def create_user(req: CreateUserRequest) -> dict[str, Any]: """Tạo tài khoản người dùng mới có phân quyền (Chỉ SuperAdmin/Giám đốc).""" return auth_service.create_user_account( email=req.email, password=req.password, full_name=req.full_name, role=req.role, linked_employee_id=req.linked_employee_id, ) @router.put( "/users/{user_id}/role", dependencies=[Depends(require_role(["superadmin"]))] ) def update_user_role(user_id: str, req: UpdateUserRoleRequest) -> dict[str, Any]: """Phân quyền vai trò cho người dùng (Chỉ SuperAdmin).""" return auth_service.update_user_role(user_id=user_id, new_role=req.role)