thomas
1f110866f5
CI / backend (pull_request) Canceled after 0s
CI / shell (pull_request) Canceled after 0s
CI / frontend (pull_request) Canceled after 0s
CI / arm64-smoke (pull_request) Canceled after 0s
CI / backend (push) Canceled after 0s
CI / shell (push) Canceled after 0s
CI / frontend (push) Canceled after 0s
CI / arm64-smoke (push) Canceled after 0s
97 lines
2.9 KiB
Python
97 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
from dataclasses import dataclass
|
|
|
|
from app.core.config import get_settings
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class SystemCommandResult:
|
|
accepted: bool
|
|
action: str
|
|
status: str
|
|
detail: str
|
|
|
|
def as_dict(self) -> dict[str, object]:
|
|
return {
|
|
"accepted": self.accepted,
|
|
"action": self.action,
|
|
"status": self.status,
|
|
"detail": self.detail,
|
|
}
|
|
|
|
|
|
class SystemControlService:
|
|
def __init__(self) -> None:
|
|
settings = get_settings()
|
|
self._enabled = settings.system_control_enabled
|
|
self._platform = os.name
|
|
self._helper = settings.system_control_helper
|
|
|
|
async def restart_service(self) -> dict[str, object]:
|
|
return (await self._run_helper(
|
|
action="restart-service",
|
|
helper_action="restart-service",
|
|
)).as_dict()
|
|
|
|
async def reboot_host(self) -> dict[str, object]:
|
|
return (await self._run_helper(
|
|
action="reboot-host",
|
|
helper_action="reboot-host",
|
|
)).as_dict()
|
|
|
|
async def _run_helper(
|
|
self,
|
|
*,
|
|
action: str,
|
|
helper_action: str,
|
|
) -> SystemCommandResult:
|
|
if not self._enabled:
|
|
return SystemCommandResult(
|
|
accepted=False,
|
|
action=action,
|
|
status="disabled",
|
|
detail="Systemkontrol er deaktiveret i serverkonfigurationen.",
|
|
)
|
|
if self._platform != "posix":
|
|
return SystemCommandResult(
|
|
accepted=False,
|
|
action=action,
|
|
status="unsupported-platform",
|
|
detail="Systemkontrol understoettes kun paa Linux/systemd-installationer.",
|
|
)
|
|
if not os.path.exists(self._helper):
|
|
return SystemCommandResult(
|
|
accepted=False,
|
|
action=action,
|
|
status="helper-missing",
|
|
detail="Systemkontrol-helperen findes ikke paa maskinen.",
|
|
)
|
|
|
|
process = await asyncio.create_subprocess_exec(
|
|
"sudo",
|
|
"-n",
|
|
self._helper,
|
|
helper_action,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
stdout, stderr = await process.communicate()
|
|
if process.returncode != 0:
|
|
error_output = (stderr or stdout).decode("utf-8", errors="ignore").strip() or "Ukendt fejl"
|
|
return SystemCommandResult(
|
|
accepted=False,
|
|
action=action,
|
|
status="command-failed",
|
|
detail=f"Kunne ikke planlaegge handlingen: {error_output}",
|
|
)
|
|
|
|
return SystemCommandResult(
|
|
accepted=True,
|
|
action=action,
|
|
status="scheduled",
|
|
detail="Handling planlagt. Serveren kan blive utilgaengelig et kort oeje.",
|
|
)
|