Spaces:
Sleeping
Sleeping
| """ | |
| Admin Authentication API | |
| Handles admin login and user management | |
| """ | |
| from fastapi import APIRouter, Depends, HTTPException, status, Request | |
| from sqlalchemy.orm import Session | |
| from pydantic import BaseModel | |
| from datetime import datetime, timedelta | |
| from typing import Optional | |
| import bcrypt | |
| import jwt | |
| import uuid | |
| from app.models.database import get_db | |
| from app.models.orm import AdminUserModel | |
| from app.models.domain import AdminUser | |
| from app.core.config import settings | |
| from app.services.audit_service import get_audit_service | |
| router = APIRouter() | |
| # Admin JWT settings | |
| ADMIN_JWT_SECRET = settings.SECRET_KEY + "_admin" # Separate secret for admin tokens | |
| ADMIN_JWT_ALGORITHM = "HS256" | |
| ADMIN_JWT_EXPIRATION_HOURS = 8 | |
| class AdminRegisterRequest(BaseModel): | |
| username: str | |
| email: str | |
| password: str | |
| registration_secret: str # Secret key to prevent unauthorized admin creation | |
| class AdminLoginRequest(BaseModel): | |
| username: str | |
| password: str | |
| class AdminLoginResponse(BaseModel): | |
| access_token: str | |
| token_type: str = "bearer" | |
| admin_id: str | |
| username: str | |
| role: str | |
| def hash_password(password: str) -> str: | |
| """Hash a password using bcrypt""" | |
| return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') | |
| def verify_password(password: str, hashed_password: str) -> bool: | |
| """Verify a password against a hash""" | |
| return bcrypt.checkpw(password.encode('utf-8'), hashed_password.encode('utf-8')) | |
| def create_admin_token(admin_id: str, username: str, role: str) -> str: | |
| """Create a JWT token for admin user""" | |
| payload = { | |
| "sub": admin_id, | |
| "username": username, | |
| "role": role, | |
| "type": "admin", | |
| "exp": datetime.utcnow() + timedelta(hours=ADMIN_JWT_EXPIRATION_HOURS) | |
| } | |
| return jwt.encode(payload, ADMIN_JWT_SECRET, algorithm=ADMIN_JWT_ALGORITHM) | |
| def register_admin( | |
| request: AdminRegisterRequest, | |
| http_request: Request, | |
| db: Session = Depends(get_db) | |
| ): | |
| """ | |
| Register the first admin user (requires secret key) | |
| This endpoint is protected by a registration secret to prevent unauthorized admin creation. | |
| After the first admin is created, additional admins should be created through the admin UI. | |
| """ | |
| # Verify registration secret | |
| if request.registration_secret != settings.ADMIN_REGISTRATION_SECRET: | |
| raise HTTPException( | |
| status_code=status.HTTP_403_FORBIDDEN, | |
| detail="Invalid registration secret" | |
| ) | |
| # Check if username or email already exists | |
| existing = db.query(AdminUserModel).filter( | |
| (AdminUserModel.username == request.username) | | |
| (AdminUserModel.email == request.email) | |
| ).first() | |
| if existing: | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail="Username or email already exists" | |
| ) | |
| # Create admin user | |
| admin_id = str(uuid.uuid4()) | |
| admin_user = AdminUserModel( | |
| id=admin_id, | |
| username=request.username, | |
| email=request.email, | |
| hashed_password=hash_password(request.password), | |
| role="admin", | |
| created_at=datetime.utcnow(), | |
| is_active=True | |
| ) | |
| db.add(admin_user) | |
| db.commit() | |
| db.refresh(admin_user) | |
| # Log the registration | |
| audit_service = get_audit_service() | |
| audit_service.log_action( | |
| db=db, | |
| actor_type="admin", | |
| actor_id=admin_id, | |
| action="admin_registered", | |
| details={"username": request.username, "email": request.email}, | |
| ip_address=http_request.client.host if http_request.client else None | |
| ) | |
| # Create token | |
| token = create_admin_token(admin_id, request.username, "admin") | |
| return AdminLoginResponse( | |
| access_token=token, | |
| admin_id=admin_id, | |
| username=request.username, | |
| role="admin" | |
| ) | |
| def login_admin( | |
| request: AdminLoginRequest, | |
| http_request: Request, | |
| db: Session = Depends(get_db) | |
| ): | |
| """Admin login endpoint""" | |
| # Find admin user | |
| admin_user = db.query(AdminUserModel).filter( | |
| AdminUserModel.username == request.username | |
| ).first() | |
| if not admin_user or not verify_password(request.password, admin_user.hashed_password): | |
| raise HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Invalid username or password" | |
| ) | |
| if not admin_user.is_active: | |
| raise HTTPException( | |
| status_code=status.HTTP_403_FORBIDDEN, | |
| detail="Admin account is deactivated" | |
| ) | |
| # Update last login | |
| admin_user.last_login = datetime.utcnow() | |
| db.commit() | |
| # Log the login | |
| audit_service = get_audit_service() | |
| audit_service.log_action( | |
| db=db, | |
| actor_type="admin", | |
| actor_id=admin_user.id, | |
| action="admin_login", | |
| details={"username": request.username}, | |
| ip_address=http_request.client.host if http_request.client else None | |
| ) | |
| # Create token | |
| token = create_admin_token(admin_user.id, admin_user.username, admin_user.role) | |
| return AdminLoginResponse( | |
| access_token=token, | |
| admin_id=admin_user.id, | |
| username=admin_user.username, | |
| role=admin_user.role | |
| ) | |
| def get_current_admin( | |
| admin_user: AdminUserModel = Depends(lambda: None) # Will be replaced with proper dependency | |
| ): | |
| """Get current admin user info""" | |
| # TODO: Implement get_admin_user dependency | |
| raise HTTPException(status_code=501, detail="Not implemented yet") | |