busya
c6189876b3
Auto-fixed categories: - F401: unused imports removed - I001: import blocks sorted - W293: trailing whitespace stripped - UP035: deprecated typing imports replaced - SIM: simplify suggestions applied - ARG: unused args prefixed with underscore - T201: print statements removed - F841: unused variables removed - RUF059: unpacked variables prefixed Remaining ~1500 unfixable errors (C901, B904, N806, E402) require manual work. Backend smoke tests: 13/13 passed.
142 lines
5.4 KiB
Python
Executable File
142 lines
5.4 KiB
Python
Executable File
# #region AuthApi [C:3] [TYPE Module] [SEMANTICS fastapi, auth, api]
|
|
#
|
|
# @BRIEF Authentication API endpoints.
|
|
# @LAYER: API
|
|
# @RELATION DEPENDS_ON -> [AuthService]
|
|
# @RELATION DEPENDS_ON -> [get_auth_db]
|
|
# @RELATION DEPENDS_ON -> [get_current_user]
|
|
# @RELATION DEPENDS_ON -> [is_adfs_configured]
|
|
# @INVARIANT: All auth endpoints must return consistent error codes.
|
|
|
|
import starlette.requests
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..core.auth.logger import log_security_event
|
|
from ..core.auth.oauth import is_adfs_configured, oauth
|
|
from ..core.database import get_auth_db
|
|
from ..core.logger import belief_scope
|
|
from ..dependencies import get_current_user
|
|
from ..schemas.auth import Token
|
|
from ..schemas.auth import User as UserSchema
|
|
from ..services.auth_service import AuthService
|
|
|
|
# #region router [C:1] [TYPE Variable]
|
|
# @RELATION DEPENDS_ON -> [fastapi.APIRouter]
|
|
# @BRIEF APIRouter instance for authentication routes.
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
# #endregion router
|
|
|
|
|
|
# #region login_for_access_token [C:3] [TYPE Function]
|
|
# @BRIEF Authenticates a user and returns a JWT access token.
|
|
# @PRE: form_data contains username and password.
|
|
# @POST: Returns a Token object on success.
|
|
# @RELATION CALLS -> [AuthService.authenticate_user]
|
|
# @RELATION CALLS -> [AuthService.create_session]
|
|
@router.post("/login", response_model=Token)
|
|
async def login_for_access_token(
|
|
form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_auth_db)
|
|
):
|
|
with belief_scope("api.auth.login"):
|
|
auth_service = AuthService(db)
|
|
user = auth_service.authenticate_user(form_data.username, form_data.password)
|
|
if not user:
|
|
log_security_event(
|
|
"LOGIN_FAILED", form_data.username, {"reason": "Invalid credentials"}
|
|
)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Incorrect username or password",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
log_security_event("LOGIN_SUCCESS", user.username, {"source": "LOCAL"})
|
|
return auth_service.create_session(user)
|
|
|
|
|
|
# #endregion login_for_access_token
|
|
|
|
|
|
# #region read_users_me [C:3] [TYPE Function]
|
|
# @BRIEF Retrieves the profile of the currently authenticated user.
|
|
# @PRE: Valid JWT token provided.
|
|
# @POST: Returns the current user's data.
|
|
# @RELATION DEPENDS_ON -> [get_current_user]
|
|
@router.get("/me", response_model=UserSchema)
|
|
async def read_users_me(current_user: UserSchema = Depends(get_current_user)):
|
|
with belief_scope("api.auth.me"):
|
|
return current_user
|
|
|
|
|
|
# #endregion read_users_me
|
|
|
|
|
|
# #region logout [C:3] [TYPE Function]
|
|
# @BRIEF Logs out the current user (placeholder for session revocation).
|
|
# @PRE: Valid JWT token provided.
|
|
# @POST: Returns success message.
|
|
# @RELATION DEPENDS_ON -> [get_current_user]
|
|
@router.post("/logout")
|
|
async def logout(current_user: UserSchema = Depends(get_current_user)):
|
|
with belief_scope("api.auth.logout"):
|
|
log_security_event("LOGOUT", current_user.username)
|
|
# In a stateless JWT setup, client-side token deletion is primary.
|
|
# Server-side revocation (blacklisting) can be added here if needed.
|
|
return {"message": "Successfully logged out"}
|
|
|
|
|
|
# #endregion logout
|
|
|
|
|
|
# #region login_adfs [C:3] [TYPE Function]
|
|
# @BRIEF Initiates the ADFS OIDC login flow.
|
|
# @POST: Redirects the user to ADFS.
|
|
# @RELATION USES -> [is_adfs_configured]
|
|
@router.get("/login/adfs")
|
|
async def login_adfs(request: starlette.requests.Request):
|
|
with belief_scope("api.auth.login_adfs"):
|
|
if not is_adfs_configured():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="ADFS is not configured. Please set ADFS_CLIENT_ID, ADFS_CLIENT_SECRET, and ADFS_METADATA_URL environment variables.",
|
|
)
|
|
redirect_uri = request.url_for("auth_callback_adfs")
|
|
return await oauth.adfs.authorize_redirect(request, str(redirect_uri))
|
|
|
|
|
|
# #endregion login_adfs
|
|
|
|
|
|
# #region auth_callback_adfs [C:3] [TYPE Function]
|
|
# @BRIEF Handles the callback from ADFS after successful authentication.
|
|
# @POST: Provisions user JIT and returns session token.
|
|
# @RELATION DEPENDS_ON -> [is_adfs_configured]
|
|
# @RELATION CALLS -> [AuthService.provision_adfs_user]
|
|
# @RELATION CALLS -> [AuthService.create_session]
|
|
@router.get("/callback/adfs", name="auth_callback_adfs")
|
|
async def auth_callback_adfs(
|
|
request: starlette.requests.Request, db: Session = Depends(get_auth_db)
|
|
):
|
|
with belief_scope("api.auth.callback_adfs"):
|
|
if not is_adfs_configured():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="ADFS is not configured. Please set ADFS_CLIENT_ID, ADFS_CLIENT_SECRET, and ADFS_METADATA_URL environment variables.",
|
|
)
|
|
token = await oauth.adfs.authorize_access_token(request)
|
|
user_info = token.get("userinfo")
|
|
if not user_info:
|
|
raise HTTPException(
|
|
status_code=400, detail="Failed to retrieve user info from ADFS"
|
|
)
|
|
|
|
auth_service = AuthService(db)
|
|
user = auth_service.provision_adfs_user(user_info)
|
|
return auth_service.create_session(user)
|
|
|
|
|
|
# #endregion auth_callback_adfs
|
|
|
|
# #endregion AuthApi
|