selfprivacy-rest-api/selfprivacy_api/graphql/mutations/system_mutations.py

212 lines
6.7 KiB
Python

"""System management mutations"""
# pylint: disable=too-few-public-methods
import typing
import strawberry
from selfprivacy_api.graphql import IsAuthenticated
from selfprivacy_api.graphql.common_types.jobs import job_to_api_job
from selfprivacy_api.graphql.mutations.mutation_interface import (
GenericJobMutationReturn,
GenericMutationReturn,
MutationReturnInterface,
GenericJobMutationReturn,
)
import selfprivacy_api.actions.system as system_actions
from selfprivacy_api.graphql.common_types.jobs import job_to_api_job
from selfprivacy_api.jobs.nix_collect_garbage import start_nix_collect_garbage
import selfprivacy_api.actions.ssh as ssh_actions
@strawberry.type
class TimezoneMutationReturn(MutationReturnInterface):
"""Return type of the timezone mutation, contains timezone"""
timezone: typing.Optional[str]
@strawberry.type
class AutoUpgradeSettingsMutationReturn(MutationReturnInterface):
"""Return type autoUpgrade Settings"""
enableAutoUpgrade: bool
allowReboot: bool
@strawberry.type
class SSHSettingsMutationReturn(MutationReturnInterface):
"""A return type for after changing SSH settings"""
enable: bool
password_authentication: bool
@strawberry.input
class SSHSettingsInput:
"""Input type for SSH settings"""
enable: bool
password_authentication: bool
@strawberry.input
class AutoUpgradeSettingsInput:
"""Input type for auto upgrade settings"""
enableAutoUpgrade: typing.Optional[bool] = None
allowReboot: typing.Optional[bool] = None
@strawberry.type
class SystemMutations:
"""Mutations related to system settings"""
@strawberry.mutation(permission_classes=[IsAuthenticated])
def change_timezone(self, timezone: str) -> TimezoneMutationReturn:
"""Change the timezone of the server. Timezone is a tzdatabase name."""
try:
system_actions.change_timezone(timezone)
except system_actions.InvalidTimezone as e:
return TimezoneMutationReturn(
success=False,
message=str(e),
code=400,
timezone=None,
)
return TimezoneMutationReturn(
success=True,
message="Timezone changed",
code=200,
timezone=timezone,
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def change_auto_upgrade_settings(
self, settings: AutoUpgradeSettingsInput
) -> AutoUpgradeSettingsMutationReturn:
"""Change auto upgrade settings of the server."""
system_actions.set_auto_upgrade_settings(
settings.enableAutoUpgrade, settings.allowReboot
)
new_settings = system_actions.get_auto_upgrade_settings()
return AutoUpgradeSettingsMutationReturn(
success=True,
message="Auto-upgrade settings changed",
code=200,
enableAutoUpgrade=new_settings.enable,
allowReboot=new_settings.allowReboot,
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def change_ssh_settings(
self, settings: SSHSettingsInput
) -> SSHSettingsMutationReturn:
"""Change ssh settings of the server."""
ssh_actions.set_ssh_settings(
enable=settings.enable,
password_authentication=settings.password_authentication,
)
new_settings = ssh_actions.get_ssh_settings()
return SSHSettingsMutationReturn(
success=True,
message="SSH settings changed",
code=200,
enable=new_settings.enable,
password_authentication=new_settings.passwordAuthentication,
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def run_system_rebuild(self) -> GenericJobMutationReturn:
try:
job = system_actions.rebuild_system()
return GenericJobMutationReturn(
success=True,
message="Starting system rebuild",
code=200,
job=job_to_api_job(job),
)
except system_actions.ShellException as e:
return GenericJobMutationReturn(
success=False,
message=str(e),
code=500,
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def run_system_rollback(self) -> GenericMutationReturn:
system_actions.rollback_system()
try:
return GenericMutationReturn(
success=True,
message="Starting system rollback",
code=200,
)
except system_actions.ShellException as e:
return GenericMutationReturn(
success=False,
message=str(e),
code=500,
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def run_system_upgrade(self) -> GenericJobMutationReturn:
try:
job = system_actions.upgrade_system()
return GenericJobMutationReturn(
success=True,
message="Starting system upgrade",
code=200,
job=job_to_api_job(job),
)
except system_actions.ShellException as e:
return GenericJobMutationReturn(
success=False,
message=str(e),
code=500,
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def reboot_system(self) -> GenericMutationReturn:
system_actions.reboot_system()
try:
return GenericMutationReturn(
success=True,
message="System reboot has started",
code=200,
)
except system_actions.ShellException as e:
return GenericMutationReturn(
success=False,
message=str(e),
code=500,
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def pull_repository_changes(self) -> GenericMutationReturn:
result = system_actions.pull_repository_changes()
if result.status == 0:
return GenericMutationReturn(
success=True,
message="Repository changes pulled",
code=200,
)
return GenericMutationReturn(
success=False,
message=f"Failed to pull repository changes:\n{result.data}",
code=500,
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def nix_collect_garbage(self) -> GenericJobMutationReturn:
job = start_nix_collect_garbage()
return GenericJobMutationReturn(
success=True,
code=200,
message="Garbage collector started...",
job=job_to_api_job(job),
)