""" Routes for accountant security: step-up auth, session management, activity logs. """ from fastapi import APIRouter, Depends, HTTPException, Query, Request from sqlalchemy.orm import Session from typing import Optional from datetime import datetime from ...shared.config.database import get_db from ...shared.config.logging_config import get_logger from ...security.middleware.auth import authorize_roles, get_current_user from ...auth.models.user import User from ..services.accountant_security_service import accountant_security_service from ...shared.utils.response_helpers import success_response from ...auth.services.mfa_service import mfa_service logger = get_logger(__name__) router = APIRouter(prefix='/accountant/security', tags=['accountant-security']) @router.post('/step-up/verify') async def verify_step_up( request: Request, step_up_data: dict, current_user: User = Depends(authorize_roles('accountant')), db: Session = Depends(get_db) ): """Verify step-up authentication (MFA token or password re-entry).""" try: from ..models.accountant_session import AccountantSession mfa_token = step_up_data.get('mfa_token') password = step_up_data.get('password') session_token = step_up_data.get('session_token') if not session_token: # Try to get from header or cookie session_token = request.headers.get('X-Session-Token') or request.cookies.get('session_token') # If still no session token, try to find the most recent active session for this user. # If none exists (e.g., password-only admin without MFA), create a fresh session so # password-based step-up can proceed without forcing a full re-login. if not session_token: active_session = db.query(AccountantSession).filter( AccountantSession.user_id == current_user.id, AccountantSession.is_active == True, AccountantSession.expires_at > datetime.utcnow() ).order_by(AccountantSession.last_activity.desc()).first() if active_session: session_token = active_session.session_token else: new_session = accountant_security_service.create_session( db=db, user_id=current_user.id, ip_address=request.client.host if request.client else None, user_agent=request.headers.get('User-Agent') ) session_token = new_session.session_token # Verify MFA if token provided if mfa_token: try: is_valid = mfa_service.verify_mfa(db, current_user.id, mfa_token) if not is_valid: raise HTTPException(status_code=401, detail='Invalid MFA token') except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) # Or verify password if provided elif password: import bcrypt if not bcrypt.checkpw(password.encode('utf-8'), current_user.password.encode('utf-8')): raise HTTPException(status_code=401, detail='Invalid password') else: raise HTTPException(status_code=400, detail='Either mfa_token or password is required') # Complete step-up authentication success = accountant_security_service.complete_step_up( db=db, session_token=session_token, user_id=current_user.id ) if not success: raise HTTPException(status_code=400, detail='Failed to complete step-up authentication') # Log step-up activity client_ip = request.client.host if request.client else None user_agent = request.headers.get('User-Agent') accountant_security_service.log_activity( db=db, user_id=current_user.id, activity_type='step_up_authentication', activity_description='Step-up authentication completed', ip_address=client_ip, user_agent=user_agent, risk_level='low', metadata={'method': 'mfa' if mfa_token else 'password'} ) db.commit() return success_response( data={'step_up_completed': True}, message='Step-up authentication completed successfully' ) except HTTPException: raise except Exception as e: db.rollback() logger.error(f'Error verifying step-up: {str(e)}', exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.get('/sessions') async def get_active_sessions( current_user: User = Depends(authorize_roles('accountant')), db: Session = Depends(get_db) ): """Get active sessions for current user.""" try: from ..models.accountant_session import AccountantSession sessions = db.query(AccountantSession).filter( AccountantSession.user_id == current_user.id, AccountantSession.is_active == True ).order_by(AccountantSession.last_activity.desc()).all() session_list = [] for session in sessions: session_list.append({ 'id': session.id, 'ip_address': session.ip_address, 'user_agent': session.user_agent, 'country': session.country, 'city': session.city, 'last_activity': session.last_activity.isoformat() if session.last_activity else None, 'step_up_authenticated': session.step_up_authenticated, 'step_up_expires_at': session.step_up_expires_at.isoformat() if session.step_up_expires_at else None, 'created_at': session.created_at.isoformat() if session.created_at else None, 'expires_at': session.expires_at.isoformat() if session.expires_at else None }) return success_response(data={'sessions': session_list}) except Exception as e: logger.error(f'Error fetching sessions: {str(e)}', exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.post('/sessions/{session_id}/revoke') async def revoke_session( session_id: int, current_user: User = Depends(authorize_roles('accountant')), db: Session = Depends(get_db) ): """Revoke a specific session.""" try: from ..models.accountant_session import AccountantSession session = db.query(AccountantSession).filter( AccountantSession.id == session_id, AccountantSession.user_id == current_user.id ).first() if not session: raise HTTPException(status_code=404, detail='Session not found') session.is_active = False db.commit() return success_response(message='Session revoked successfully') except HTTPException: raise except Exception as e: db.rollback() logger.error(f'Error revoking session: {str(e)}', exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.post('/sessions/revoke-all') async def revoke_all_sessions( current_user: User = Depends(authorize_roles('accountant')), db: Session = Depends(get_db) ): """Revoke all active sessions for current user.""" try: count = accountant_security_service.revoke_all_user_sessions(db, current_user.id) db.commit() return success_response( data={'revoked_count': count}, message=f'Successfully revoked {count} active session(s)' ) except Exception as e: db.rollback() logger.error(f'Error revoking all sessions: {str(e)}', exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.get('/activity-logs') async def get_activity_logs( page: int = Query(1, ge=1), limit: int = Query(50, ge=1, le=100), risk_level: Optional[str] = Query(None), is_unusual: Optional[bool] = Query(None), current_user: User = Depends(authorize_roles('accountant')), db: Session = Depends(get_db) ): """Get activity logs for current user or all users (admin only).""" try: from ..models.accountant_activity_log import AccountantActivityLog from ...shared.utils.role_helpers import is_admin query = db.query(AccountantActivityLog) # Non-admins can only see their own logs if not is_admin(current_user, db): query = query.filter(AccountantActivityLog.user_id == current_user.id) if risk_level: query = query.filter(AccountantActivityLog.risk_level == risk_level) if is_unusual is not None: query = query.filter(AccountantActivityLog.is_unusual == is_unusual) total = query.count() offset = (page - 1) * limit logs = query.order_by(AccountantActivityLog.created_at.desc()).offset(offset).limit(limit).all() log_list = [] for log in logs: log_list.append({ 'id': log.id, 'user_id': log.user_id, 'activity_type': log.activity_type, 'activity_description': log.activity_description, 'ip_address': log.ip_address, 'country': log.country, 'city': log.city, 'risk_level': log.risk_level, 'is_unusual': log.is_unusual, 'metadata': log.activity_metadata, 'created_at': log.created_at.isoformat() if log.created_at else None }) return success_response(data={ 'logs': log_list, 'pagination': { 'total': total, 'page': page, 'limit': limit, 'total_pages': (total + limit - 1) // limit } }) except Exception as e: logger.error(f'Error fetching activity logs: {str(e)}', exc_info=True) raise HTTPException(status_code=500, detail=str(e)) @router.get('/mfa-status') async def get_mfa_status( current_user: User = Depends(authorize_roles('admin', 'accountant')), db: Session = Depends(get_db) ): """Get MFA status and enforcement requirements.""" try: requires_mfa = accountant_security_service.requires_mfa(current_user, db) is_enforced, reason = accountant_security_service.is_mfa_enforced(current_user, db) mfa_status = mfa_service.get_mfa_status(db, current_user.id) return success_response(data={ 'requires_mfa': requires_mfa, 'mfa_enabled': mfa_status['mfa_enabled'], 'is_enforced': is_enforced, 'enforcement_reason': reason, 'backup_codes_count': mfa_status['backup_codes_count'] }) except Exception as e: logger.error(f'Error getting MFA status: {str(e)}', exc_info=True) raise HTTPException(status_code=500, detail=str(e))