130 lines
4.2 KiB
Python
130 lines
4.2 KiB
Python
"""HTTP mapping for typed miner controls."""
|
|
|
|
from typing import Annotated
|
|
from urllib.parse import urlsplit
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
|
|
from app.identity import require_permission
|
|
from app.miner_control.dependencies import get_miner_control_service
|
|
from app.miner_control.exceptions import (
|
|
ControlledMinerNotFound,
|
|
MinerControlConflict,
|
|
MinerControlUnavailable,
|
|
)
|
|
from app.miner_control.models import (
|
|
EmptyControlRequest,
|
|
MinerControlAcceptedResponse,
|
|
MinerControlResponse,
|
|
PerformanceModeRequest,
|
|
)
|
|
from app.miner_control.service import MinerControlService
|
|
|
|
|
|
router = APIRouter(prefix="/api/miners", tags=["miner-controls"])
|
|
ControlService = Annotated[MinerControlService, Depends(get_miner_control_service)]
|
|
ControlViewer = Annotated[dict, Depends(require_permission("miners.view"))]
|
|
ControlUser = Annotated[dict, Depends(require_permission("miners.control"))]
|
|
|
|
|
|
def require_same_origin(request: Request) -> None:
|
|
if request.headers.get("sec-fetch-site", "").lower() == "cross-site":
|
|
raise HTTPException(status_code=403, detail="Cross-site control request rejected")
|
|
origin = request.headers.get("origin")
|
|
if origin:
|
|
parsed = urlsplit(origin)
|
|
if parsed.scheme not in {"http", "https"} or parsed.netloc != request.url.netloc:
|
|
raise HTTPException(status_code=403, detail="Cross-site control request rejected")
|
|
|
|
|
|
def _actor(current_user: dict) -> tuple[UUID, str]:
|
|
return UUID(str(current_user["id"])), str(
|
|
current_user.get("username") or current_user.get("email") or "unknown"
|
|
)
|
|
|
|
|
|
def _queue(
|
|
service: MinerControlService,
|
|
miner_id: UUID,
|
|
action: str,
|
|
requested_value: str | None,
|
|
current_user: dict,
|
|
) -> MinerControlAcceptedResponse:
|
|
actor_id, username = _actor(current_user)
|
|
try:
|
|
operation = service.queue_control(
|
|
miner_id,
|
|
action,
|
|
requested_value=requested_value,
|
|
actor_user_id=actor_id,
|
|
actor_username=username,
|
|
)
|
|
except ControlledMinerNotFound as exc:
|
|
raise HTTPException(status_code=404, detail="Miner not found") from exc
|
|
except MinerControlUnavailable as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
except MinerControlConflict as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
return MinerControlAcceptedResponse(data=operation)
|
|
|
|
|
|
@router.get("/{miner_id}/controls", response_model=MinerControlResponse)
|
|
def get_miner_controls(
|
|
miner_id: UUID,
|
|
current_user: ControlViewer,
|
|
service: ControlService,
|
|
) -> MinerControlResponse:
|
|
del current_user
|
|
try:
|
|
return MinerControlResponse(data=service.get_controls(miner_id))
|
|
except ControlledMinerNotFound as exc:
|
|
raise HTTPException(status_code=404, detail="Miner not found") from exc
|
|
|
|
|
|
@router.post(
|
|
"/{miner_id}/controls/suspend",
|
|
response_model=MinerControlAcceptedResponse,
|
|
status_code=status.HTTP_202_ACCEPTED,
|
|
dependencies=[Depends(require_same_origin)],
|
|
)
|
|
def suspend_miner(
|
|
miner_id: UUID,
|
|
payload: EmptyControlRequest,
|
|
current_user: ControlUser,
|
|
service: ControlService,
|
|
) -> MinerControlAcceptedResponse:
|
|
del payload
|
|
return _queue(service, miner_id, "suspend", None, current_user)
|
|
|
|
|
|
@router.post(
|
|
"/{miner_id}/controls/resume",
|
|
response_model=MinerControlAcceptedResponse,
|
|
status_code=status.HTTP_202_ACCEPTED,
|
|
dependencies=[Depends(require_same_origin)],
|
|
)
|
|
def resume_miner(
|
|
miner_id: UUID,
|
|
payload: EmptyControlRequest,
|
|
current_user: ControlUser,
|
|
service: ControlService,
|
|
) -> MinerControlAcceptedResponse:
|
|
del payload
|
|
return _queue(service, miner_id, "resume", None, current_user)
|
|
|
|
|
|
@router.post(
|
|
"/{miner_id}/controls/performance-mode",
|
|
response_model=MinerControlAcceptedResponse,
|
|
status_code=status.HTTP_202_ACCEPTED,
|
|
dependencies=[Depends(require_same_origin)],
|
|
)
|
|
def set_performance_mode(
|
|
miner_id: UUID,
|
|
payload: PerformanceModeRequest,
|
|
current_user: ControlUser,
|
|
service: ControlService,
|
|
) -> MinerControlAcceptedResponse:
|
|
return _queue(service, miner_id, "performance_mode", payload.mode, current_user)
|