from fastapi import APIRouter, Depends, HTTPException, status from typing import List from app.core.auth import get_current_active_user, User from app.core.database import get_database from app.services.application_service import ApplicationService from app.services.user_service import UserService from app.schemas.application import ( ApplicationCreate, ApplicationUpdate, ApplicationResponse, ApplicationWithSecret ) router = APIRouter() def get_application_service(db=Depends(get_database)) -> ApplicationService: """Dependency to get application service""" return ApplicationService(db) def get_user_service(db=Depends(get_database)) -> UserService: """Dependency to get user service""" return UserService(db) @router.get("/", response_model=List[ApplicationResponse]) async def get_applications( current_user: User = Depends(get_current_active_user), app_service: ApplicationService = Depends(get_application_service), user_service: UserService = Depends(get_user_service) ): """ Get all OAuth2 applications - Admins can see all applications - Regular users can only see their own applications """ # Get current user from database user = await user_service.get_user_by_username(current_user.username) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) # Admins can see all, others only their own if current_user.role == "admin": applications = await app_service.get_applications() else: applications = await app_service.get_applications(owner_id=str(user.id)) return [ ApplicationResponse( _id=str(app.id), name=app.name, client_id=app.client_id, redirect_uris=app.redirect_uris, grant_types=app.grant_types, scopes=app.scopes, owner_id=app.owner_id, created_at=app.created_at, updated_at=app.updated_at ) for app in applications ] @router.get("/stats") async def get_application_stats( current_user: User = Depends(get_current_active_user), app_service: ApplicationService = Depends(get_application_service) ): """Get application statistics (admin only)""" if current_user.role != "admin": raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Only admins can view application statistics" ) stats = await app_service.get_application_stats() return stats @router.get("/{app_id}", response_model=ApplicationResponse) async def get_application( app_id: str, current_user: User = Depends(get_current_active_user), app_service: ApplicationService = Depends(get_application_service), user_service: UserService = Depends(get_user_service) ): """Get an application by ID""" app = await app_service.get_application_by_id(app_id) if not app: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Application with ID {app_id} not found" ) # Check ownership (admins can view all) user = await user_service.get_user_by_username(current_user.username) if current_user.role != "admin" and app.owner_id != str(user.id): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to view this application" ) return ApplicationResponse( _id=str(app.id), name=app.name, client_id=app.client_id, redirect_uris=app.redirect_uris, grant_types=app.grant_types, scopes=app.scopes, owner_id=app.owner_id, created_at=app.created_at, updated_at=app.updated_at ) @router.post("/", response_model=ApplicationWithSecret, status_code=status.HTTP_201_CREATED) async def create_application( app_data: ApplicationCreate, current_user: User = Depends(get_current_active_user), app_service: ApplicationService = Depends(get_application_service), user_service: UserService = Depends(get_user_service) ): """ Create a new OAuth2 application Returns the application with the client_secret (only shown once!) """ # Get current user from database user = await user_service.get_user_by_username(current_user.username) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) app, client_secret = await app_service.create_application( app_data=app_data, owner_id=str(user.id) ) return ApplicationWithSecret( _id=str(app.id), name=app.name, client_id=app.client_id, client_secret=client_secret, # Plain text secret (only shown once) redirect_uris=app.redirect_uris, grant_types=app.grant_types, scopes=app.scopes, owner_id=app.owner_id, created_at=app.created_at, updated_at=app.updated_at ) @router.put("/{app_id}", response_model=ApplicationResponse) async def update_application( app_id: str, app_data: ApplicationUpdate, current_user: User = Depends(get_current_active_user), app_service: ApplicationService = Depends(get_application_service), user_service: UserService = Depends(get_user_service) ): """Update an application""" app = await app_service.get_application_by_id(app_id) if not app: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Application with ID {app_id} not found" ) # Check ownership (admins can update all) user = await user_service.get_user_by_username(current_user.username) if current_user.role != "admin" and app.owner_id != str(user.id): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to update this application" ) updated_app = await app_service.update_application(app_id, app_data) if not updated_app: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Application with ID {app_id} not found" ) return ApplicationResponse( _id=str(updated_app.id), name=updated_app.name, client_id=updated_app.client_id, redirect_uris=updated_app.redirect_uris, grant_types=updated_app.grant_types, scopes=updated_app.scopes, owner_id=updated_app.owner_id, created_at=updated_app.created_at, updated_at=updated_app.updated_at ) @router.delete("/{app_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_application( app_id: str, current_user: User = Depends(get_current_active_user), app_service: ApplicationService = Depends(get_application_service), user_service: UserService = Depends(get_user_service) ): """Delete an application""" app = await app_service.get_application_by_id(app_id) if not app: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Application with ID {app_id} not found" ) # Check ownership (admins can delete all) user = await user_service.get_user_by_username(current_user.username) if current_user.role != "admin" and app.owner_id != str(user.id): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to delete this application" ) success = await app_service.delete_application(app_id) if not success: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Application with ID {app_id} not found" ) return None @router.post("/{app_id}/regenerate-secret", response_model=ApplicationWithSecret) async def regenerate_client_secret( app_id: str, current_user: User = Depends(get_current_active_user), app_service: ApplicationService = Depends(get_application_service), user_service: UserService = Depends(get_user_service) ): """ Regenerate client secret for an application Returns the application with the new client_secret (only shown once!) """ app = await app_service.get_application_by_id(app_id) if not app: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Application with ID {app_id} not found" ) # Check ownership (admins can regenerate all) user = await user_service.get_user_by_username(current_user.username) if current_user.role != "admin" and app.owner_id != str(user.id): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to regenerate secret for this application" ) result = await app_service.regenerate_client_secret(app_id) if not result: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Application with ID {app_id} not found" ) updated_app, new_secret = result return ApplicationWithSecret( _id=str(updated_app.id), name=updated_app.name, client_id=updated_app.client_id, client_secret=new_secret, # New plain text secret (only shown once) redirect_uris=updated_app.redirect_uris, grant_types=updated_app.grant_types, scopes=updated_app.scopes, owner_id=updated_app.owner_id, created_at=updated_app.created_at, updated_at=updated_app.updated_at )