selfprivacy-rest-api/selfprivacy_api/actions/system.py

160 lines
4.3 KiB
Python

"""Actions to manage the system."""
import os
import subprocess
import pytz
from typing import Optional, List
from pydantic import BaseModel
from selfprivacy_api.utils import WriteUserData, ReadUserData
def get_timezone() -> str:
"""Get the timezone of the server"""
with ReadUserData() as user_data:
if "timezone" in user_data:
return user_data["timezone"]
return "Europe/Uzhgorod"
class InvalidTimezone(Exception):
"""Invalid timezone"""
pass
def change_timezone(timezone: str) -> None:
"""Change the timezone of the server"""
if timezone not in pytz.all_timezones:
raise InvalidTimezone(f"Invalid timezone: {timezone}")
with WriteUserData() as user_data:
user_data["timezone"] = timezone
class UserDataAutoUpgradeSettings(BaseModel):
"""Settings for auto-upgrading user data"""
enable: bool = True
allowReboot: bool = False
def get_auto_upgrade_settings() -> UserDataAutoUpgradeSettings:
"""Get the auto-upgrade settings"""
with ReadUserData() as user_data:
if "autoUpgrade" in user_data:
return UserDataAutoUpgradeSettings(**user_data["autoUpgrade"])
return UserDataAutoUpgradeSettings()
def set_auto_upgrade_settings(
enalbe: Optional[bool] = None, allowReboot: Optional[bool] = None
) -> None:
"""Set the auto-upgrade settings"""
with WriteUserData() as user_data:
if "autoUpgrade" not in user_data:
user_data["autoUpgrade"] = {}
if enalbe is not None:
user_data["autoUpgrade"]["enable"] = enalbe
if allowReboot is not None:
user_data["autoUpgrade"]["allowReboot"] = allowReboot
class ShellException(Exception):
"""Something went wrong when calling another process"""
pass
def run_blocking(cmd: List[str], new_session: bool = False) -> str:
"""Run a process, block until done, return output, complain if failed"""
process_handle = subprocess.Popen(
cmd,
shell=False,
start_new_session=new_session,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
stdout_raw, stderr_raw = process_handle.communicate()
stdout = stdout_raw.decode("utf-8")
if stderr_raw is not None:
stderr = stderr_raw.decode("utf-8")
else:
stderr = ""
output = stdout + "\n" + stderr
if process_handle.returncode != 0:
raise ShellException(
f"Shell command failed, command array: {cmd}, output: {output}"
)
return stdout
def rebuild_system() -> int:
"""Rebuild the system"""
run_blocking(["systemctl", "start", "sp-nixos-rebuild.service"], new_session=True)
return 0
def rollback_system() -> int:
"""Rollback the system"""
run_blocking(["systemctl", "start", "sp-nixos-rollback.service"], new_session=True)
return 0
def upgrade_system() -> int:
"""Upgrade the system"""
run_blocking(["systemctl", "start", "sp-nixos-upgrade.service"], new_session=True)
return 0
def reboot_system() -> None:
"""Reboot the system"""
run_blocking(["reboot"], new_session=True)
def get_system_version() -> str:
"""Get system version"""
return subprocess.check_output(["uname", "-a"]).decode("utf-8").strip()
def get_python_version() -> str:
"""Get Python version"""
return subprocess.check_output(["python", "-V"]).decode("utf-8").strip()
class SystemActionResult(BaseModel):
"""System action result"""
status: int
message: str
data: str
def pull_repository_changes() -> SystemActionResult:
"""Pull repository changes"""
git_pull_command = ["git", "pull"]
current_working_directory = os.getcwd()
os.chdir("/etc/nixos")
git_pull_process_descriptor = subprocess.Popen(
git_pull_command,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
shell=False,
)
data = git_pull_process_descriptor.communicate()[0].decode("utf-8")
os.chdir(current_working_directory)
if git_pull_process_descriptor.returncode == 0:
return SystemActionResult(
status=0,
message="Pulled repository changes",
data=data,
)
return SystemActionResult(
status=git_pull_process_descriptor.returncode,
message="Failed to pull repository changes",
data=data,
)