121 lines
5.8 KiB
Python
121 lines
5.8 KiB
Python
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import get_current_user, get_db
|
|
from app.models.router import Router
|
|
from app.models.user import User
|
|
from app.schemas.router import RouterCreate, RouterResponse, RouterTestConnection, RouterUpdate
|
|
from app.services.router_service import router_service
|
|
from app.services.settings_service import settings_service
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def serialize_router(router: Router, global_settings) -> RouterResponse:
|
|
has_router_key = bool((router.ssh_key or '').strip())
|
|
has_global_key = bool((global_settings.global_ssh_key or '').strip())
|
|
router_user = (router.ssh_user or '').strip() or None
|
|
router_password = (router.ssh_password or '').strip() or None
|
|
default_swos_user = (global_settings.default_switchos_username or '').strip() or None
|
|
default_swos_password = (global_settings.default_switchos_password or '').strip() or None
|
|
effective_username = router_user
|
|
uses_global_switchos_credentials = False
|
|
has_effective_password = bool(router_password)
|
|
|
|
if router.device_type == 'switchos':
|
|
effective_username = router_user or default_swos_user
|
|
uses_global_switchos_credentials = bool(
|
|
(not router_user and default_swos_user) or (not router_password and default_swos_password)
|
|
)
|
|
has_effective_password = bool(router_password or default_swos_password)
|
|
|
|
payload = RouterResponse.model_validate(router, from_attributes=True).model_dump()
|
|
payload['effective_username'] = effective_username
|
|
payload['uses_global_ssh_key'] = router.device_type == 'routeros' and has_global_key and not has_router_key
|
|
payload['has_effective_ssh_key'] = router.device_type == 'routeros' and (has_router_key or has_global_key)
|
|
payload['uses_global_switchos_credentials'] = uses_global_switchos_credentials
|
|
payload['has_effective_password'] = has_effective_password
|
|
payload['supports_export'] = router.device_type == 'routeros'
|
|
payload['supports_restore_upload'] = router.device_type == 'routeros'
|
|
return RouterResponse.model_validate(payload)
|
|
|
|
|
|
@router.get('', response_model=list[RouterResponse])
|
|
def list_routers(current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
global_settings = settings_service.get_or_create(db)
|
|
routers = db.query(Router).filter(Router.owner_id == current_user.id).order_by(Router.created_at.desc()).all()
|
|
return [serialize_router(router, global_settings) for router in routers]
|
|
|
|
|
|
@router.post('', response_model=RouterResponse)
|
|
def create_router(payload: RouterCreate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
router_data = payload.model_dump()
|
|
if router_data.get('device_type') == 'switchos' and router_data.get('ssh_user') is None:
|
|
router_data['ssh_user'] = ''
|
|
router = Router(**router_data, owner_id=current_user.id)
|
|
db.add(router)
|
|
db.commit()
|
|
db.refresh(router)
|
|
global_settings = settings_service.get_or_create(db)
|
|
return serialize_router(router, global_settings)
|
|
|
|
|
|
@router.get('/{router_id}', response_model=RouterResponse)
|
|
def get_router(router_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
router = db.query(Router).filter(Router.id == router_id, Router.owner_id == current_user.id).first()
|
|
if not router:
|
|
raise HTTPException(status_code=404, detail='Device not found')
|
|
global_settings = settings_service.get_or_create(db)
|
|
return serialize_router(router, global_settings)
|
|
|
|
|
|
@router.put('/{router_id}', response_model=RouterResponse)
|
|
def update_router(router_id: int, payload: RouterUpdate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
router = db.query(Router).filter(Router.id == router_id, Router.owner_id == current_user.id).first()
|
|
if not router:
|
|
raise HTTPException(status_code=404, detail='Device not found')
|
|
changes = payload.model_dump(exclude_unset=True)
|
|
target_device_type = changes.get('device_type', router.device_type)
|
|
if target_device_type == 'switchos':
|
|
changes['ssh_key'] = None
|
|
if 'port' not in changes:
|
|
changes['port'] = 80
|
|
if changes.get('ssh_user') is None:
|
|
changes['ssh_user'] = ''
|
|
elif target_device_type == 'routeros' and 'port' not in changes and router.device_type != 'routeros':
|
|
changes['port'] = 22
|
|
if not changes.get('ssh_user'):
|
|
changes['ssh_user'] = router.ssh_user or 'admin'
|
|
for key, value in changes.items():
|
|
setattr(router, key, value)
|
|
db.add(router)
|
|
db.commit()
|
|
db.refresh(router)
|
|
global_settings = settings_service.get_or_create(db)
|
|
return serialize_router(router, global_settings)
|
|
|
|
|
|
@router.delete('/{router_id}')
|
|
def delete_router(router_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
router = db.query(Router).filter(Router.id == router_id, Router.owner_id == current_user.id).first()
|
|
if not router:
|
|
raise HTTPException(status_code=404, detail='Device not found')
|
|
for backup in list(router.backups):
|
|
path = Path(backup.file_path)
|
|
if path.exists():
|
|
path.unlink()
|
|
db.delete(router)
|
|
db.commit()
|
|
return {'message': 'Device deleted'}
|
|
|
|
|
|
@router.get('/{router_id}/test-connection', response_model=RouterTestConnection)
|
|
def test_connection(router_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
router = db.query(Router).filter(Router.id == router_id, Router.owner_id == current_user.id).first()
|
|
if not router:
|
|
raise HTTPException(status_code=404, detail='Device not found')
|
|
global_settings = settings_service.get_or_create(db)
|
|
return router_service.test_connection(db, router, global_settings)
|