SelfPrivacy/selfprivacy-rest-api
1
0
Fork 0
mirror of https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git synced 2025-01-11 10:29:30 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request 'Volume management fixes' (#45) from fix/do-volumes into master

Browse source 
Reviewed-on: https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api/pulls/45
Reviewed-by: houkime <houkime@protonmail.com>
This commit is contained in:
Inex Code 2023-08-02 15:16:12 +03:00
parent c5b7ca7565 f4263b0288
commit a1267946fc
15 changed files with 92 additions and 91 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
selfprivacy_api/dependencies.py
View file

@ -27,4 +27,4 @@ async def get_token_header(
def get_api_version() -> str:
"""Get API version"""
return "2.2.0"
return "2.2.1"

2
selfprivacy_api/graphql/queries/storage.py
View file

@ -23,7 +23,7 @@ class Storage:
else str(volume.size),
free_space=str(volume.fsavail),
used_space=str(volume.fsused),
root=volume.name == "sda1",
root=volume.is_root(),
name=volume.name,
model=volume.model,
serial=volume.serial,

11
selfprivacy_api/services/bitwarden/__init__.py
View file

@ -3,13 +3,12 @@ import base64
import subprocess
import typing
from selfprivacy_api.jobs import Job, JobStatus, Jobs
from selfprivacy_api.jobs import Job, Jobs
from selfprivacy_api.services.generic_service_mover import FolderMoveNames, move_service
from selfprivacy_api.services.generic_status_getter import get_service_status
from selfprivacy_api.services.service import Service, ServiceDnsRecord, ServiceStatus
from selfprivacy_api.utils import ReadUserData, WriteUserData, get_domain
from selfprivacy_api.utils.block_devices import BlockDevice
from selfprivacy_api.utils.huey import huey
import selfprivacy_api.utils.network as network_utils
from selfprivacy_api.services.bitwarden.icon import BITWARDEN_ICON
@ -121,14 +120,6 @@ class Bitwarden(Service):
def get_folders() -> typing.List[str]:
return ["/var/lib/bitwarden", "/var/lib/bitwarden_rs"]
@staticmethod
def get_drive() -> str:
with ReadUserData() as user_data:
if user_data.get("useBinds", False):
return user_data.get("bitwarden", {}).get("location", "sda1")
else:
return "sda1"
@staticmethod
def get_dns_records() -> typing.List[ServiceDnsRecord]:
"""Return list of DNS records for Bitwarden service."""

2
selfprivacy_api/services/generic_service_mover.py
View file

@ -83,7 +83,7 @@ def move_service(
)
return
# Make sure the volume is mounted
if volume.name != "sda1" and f"/volumes/{volume.name}" not in volume.mountpoints:
if not volume.is_root() and f"/volumes/{volume.name}" not in volume.mountpoints:
Jobs.update(
job=job,
status=JobStatus.ERROR,

9
selfprivacy_api/services/gitea/__init__.py
View file

@ -9,7 +9,6 @@ from selfprivacy_api.services.generic_status_getter import get_service_status
from selfprivacy_api.services.service import Service, ServiceDnsRecord, ServiceStatus
from selfprivacy_api.utils import ReadUserData, WriteUserData, get_domain
from selfprivacy_api.utils.block_devices import BlockDevice
from selfprivacy_api.utils.huey import huey
import selfprivacy_api.utils.network as network_utils
from selfprivacy_api.services.gitea.icon import GITEA_ICON
@ -116,14 +115,6 @@ class Gitea(Service):
def get_folders() -> typing.List[str]:
return ["/var/lib/gitea"]
@staticmethod
def get_drive() -> str:
with ReadUserData() as user_data:
if user_data.get("useBinds", False):
return user_data.get("gitea", {}).get("location", "sda1")
else:
return "sda1"
@staticmethod
def get_dns_records() -> typing.List[ServiceDnsRecord]:
return [

30
selfprivacy_api/services/jitsi/__init__.py
View file

@ -3,16 +3,13 @@ import base64
import subprocess
import typing
from selfprivacy_api.jobs import Job, Jobs
from selfprivacy_api.services.generic_service_mover import FolderMoveNames, move_service
from selfprivacy_api.jobs import Job
from selfprivacy_api.services.generic_status_getter import (
get_service_status,
get_service_status_from_several_units,
)
from selfprivacy_api.services.service import Service, ServiceDnsRecord, ServiceStatus
from selfprivacy_api.utils import ReadUserData, WriteUserData, get_domain
from selfprivacy_api.utils.block_devices import BlockDevice
from selfprivacy_api.utils.huey import huey
import selfprivacy_api.utils.network as network_utils
from selfprivacy_api.services.jitsi.icon import JITSI_ICON
@ -87,18 +84,27 @@ class Jitsi(Service):
@staticmethod
def stop():
subprocess.run(["systemctl", "stop", "jitsi-videobridge.service"])
subprocess.run(["systemctl", "stop", "jicofo.service"])
subprocess.run(
["systemctl", "stop", "jitsi-videobridge.service"],
check=False,
)
subprocess.run(["systemctl", "stop", "jicofo.service"], check=False)
@staticmethod
def start():
subprocess.run(["systemctl", "start", "jitsi-videobridge.service"])
subprocess.run(["systemctl", "start", "jicofo.service"])
subprocess.run(
["systemctl", "start", "jitsi-videobridge.service"],
check=False,
)
subprocess.run(["systemctl", "start", "jicofo.service"], check=False)
@staticmethod
def restart():
subprocess.run(["systemctl", "restart", "jitsi-videobridge.service"])
subprocess.run(["systemctl", "restart", "jicofo.service"])
subprocess.run(
["systemctl", "restart", "jitsi-videobridge.service"],
check=False,
)
subprocess.run(["systemctl", "restart", "jicofo.service"], check=False)
@staticmethod
def get_configuration():
@ -116,10 +122,6 @@ class Jitsi(Service):
def get_folders() -> typing.List[str]:
return ["/var/lib/jitsi-meet"]
@staticmethod
def get_drive() -> str:
return "sda1"
@staticmethod
def get_dns_records() -> typing.List[ServiceDnsRecord]:
ip4 = network_utils.get_ip4()

34
selfprivacy_api/services/mailserver/__init__.py
View file

@ -4,16 +4,14 @@ import base64
import subprocess
import typing
from selfprivacy_api.jobs import Job, JobStatus, Jobs
from selfprivacy_api.jobs import Job, Jobs
from selfprivacy_api.services.generic_service_mover import FolderMoveNames, move_service
from selfprivacy_api.services.generic_status_getter import (
get_service_status,
get_service_status_from_several_units,
)
from selfprivacy_api.services.service import Service, ServiceDnsRecord, ServiceStatus
import selfprivacy_api.utils as utils
from selfprivacy_api import utils
from selfprivacy_api.utils.block_devices import BlockDevice
from selfprivacy_api.utils.huey import huey
import selfprivacy_api.utils.network as network_utils
from selfprivacy_api.services.mailserver.icon import MAILSERVER_ICON
@ -23,7 +21,7 @@ class MailServer(Service):
@staticmethod
def get_id() -> str:
return "mailserver"
return "email"
@staticmethod
def get_display_name() -> str:
@ -78,18 +76,18 @@ class MailServer(Service):
@staticmethod
def stop():
subprocess.run(["systemctl", "stop", "dovecot2.service"])
subprocess.run(["systemctl", "stop", "postfix.service"])
subprocess.run(["systemctl", "stop", "dovecot2.service"], check=False)
subprocess.run(["systemctl", "stop", "postfix.service"], check=False)
@staticmethod
def start():
subprocess.run(["systemctl", "start", "dovecot2.service"])
subprocess.run(["systemctl", "start", "postfix.service"])
subprocess.run(["systemctl", "start", "dovecot2.service"], check=False)
subprocess.run(["systemctl", "start", "postfix.service"], check=False)
@staticmethod
def restart():
subprocess.run(["systemctl", "restart", "dovecot2.service"])
subprocess.run(["systemctl", "restart", "postfix.service"])
subprocess.run(["systemctl", "restart", "dovecot2.service"], check=False)
subprocess.run(["systemctl", "restart", "postfix.service"], check=False)
@staticmethod
def get_configuration():
@ -107,14 +105,6 @@ class MailServer(Service):
def get_folders() -> typing.List[str]:
return ["/var/vmail", "/var/sieve"]
@staticmethod
def get_drive() -> str:
with utils.ReadUserData() as user_data:
if user_data.get("useBinds", False):
return user_data.get("mailserver", {}).get("location", "sda1")
else:
return "sda1"
@staticmethod
def get_dns_records() -> typing.List[ServiceDnsRecord]:
domain = utils.get_domain()
@ -142,7 +132,7 @@ class MailServer(Service):
type="MX", name=domain, content=domain, ttl=3600, priority=10
),
ServiceDnsRecord(
type="TXT", name="_dmarc", content=f"v=DMARC1; p=none", ttl=18000
type="TXT", name="_dmarc", content="v=DMARC1; p=none", ttl=18000
),
ServiceDnsRecord(
type="TXT",
@ -157,7 +147,7 @@ class MailServer(Service):
def move_to_volume(self, volume: BlockDevice) -> Job:
job = Jobs.add(
type_id="services.mailserver.move",
type_id="services.email.move",
name="Move Mail Server",
description=f"Moving mailserver data to {volume.name}",
)
@ -167,7 +157,7 @@ class MailServer(Service):
volume,
job,
FolderMoveNames.default_foldermoves(self),
"mailserver",
"email",
)
return job

9
selfprivacy_api/services/nextcloud/__init__.py
View file

@ -120,15 +120,6 @@ class Nextcloud(Service):
def get_folders() -> typing.List[str]:
return ["/var/lib/nextcloud"]
@staticmethod
def get_drive() -> str:
"""Get the name of disk where Nextcloud is installed."""
with ReadUserData() as user_data:
if user_data.get("useBinds", False):
return user_data.get("nextcloud", {}).get("location", "sda1")
else:
return "sda1"
@staticmethod
def get_dns_records() -> typing.List[ServiceDnsRecord]:
return [

13
selfprivacy_api/services/ocserv/__init__.py
View file

@ -2,8 +2,7 @@
import base64
import subprocess
import typing
from selfprivacy_api.jobs import Job, Jobs
from selfprivacy_api.services.generic_service_mover import FolderMoveNames, move_service
from selfprivacy_api.jobs import Job
from selfprivacy_api.services.generic_status_getter import get_service_status
from selfprivacy_api.services.service import Service, ServiceDnsRecord, ServiceStatus
from selfprivacy_api.utils import ReadUserData, WriteUserData
@ -77,15 +76,15 @@ class Ocserv(Service):
@staticmethod
def stop():
subprocess.run(["systemctl", "stop", "ocserv.service"])
subprocess.run(["systemctl", "stop", "ocserv.service"], check=False)
@staticmethod
def start():
subprocess.run(["systemctl", "start", "ocserv.service"])
subprocess.run(["systemctl", "start", "ocserv.service"], check=False)
@staticmethod
def restart():
subprocess.run(["systemctl", "restart", "ocserv.service"])
subprocess.run(["systemctl", "restart", "ocserv.service"], check=False)
@staticmethod
def get_configuration():
@ -99,10 +98,6 @@ class Ocserv(Service):
def get_logs():
return ""
@staticmethod
def get_drive() -> str:
return "sda1"
@staticmethod
def get_dns_records() -> typing.List[ServiceDnsRecord]:
return [

8
selfprivacy_api/services/pleroma/__init__.py
View file

@ -119,14 +119,6 @@ class Pleroma(Service):
),
]
@staticmethod
def get_drive() -> str:
with ReadUserData() as user_data:
if user_data.get("useBinds", False):
return user_data.get("pleroma", {}).get("location", "sda1")
else:
return "sda1"
@staticmethod
def get_dns_records() -> typing.List[ServiceDnsRecord]:
return [

24
selfprivacy_api/services/service.py
View file

@ -6,10 +6,11 @@ import typing
from pydantic import BaseModel
from selfprivacy_api.jobs import Job
from selfprivacy_api.utils.block_devices import BlockDevice
from selfprivacy_api.utils.block_devices import BlockDevice, BlockDevices
from selfprivacy_api.services.generic_size_counter import get_storage_usage
from selfprivacy_api.services.owned_path import OwnedPath
from selfprivacy_api import utils
from selfprivacy_api.utils.waitloop import wait_until_true
DEFAULT_START_STOP_TIMEOUT = 10 * 60
@ -197,10 +198,23 @@ class Service(ABC):
def get_dns_records() -> typing.List[ServiceDnsRecord]:
pass
@staticmethod
@abstractmethod
def get_drive() -> str:
pass
@classmethod
def get_drive(cls) -> str:
"""
Get the name of the drive/volume where the service is located.
Example values are `sda1`, `vda`, `sdb`.
"""
root_device: str = BlockDevices().get_root_block_device().name
if not cls.is_movable():
return root_device
with utils.ReadUserData() as userdata:
if userdata.get("useBinds", False):
return userdata.get(cls.get_id(), {}).get(
"location",
root_device,
)
else:
return root_device
@classmethod
def get_folders(cls) -> typing.List[str]:

1
selfprivacy_api/services/test_service/__init__.py
View file

@ -10,7 +10,6 @@ from os import path
from selfprivacy_api.jobs import Job
from selfprivacy_api.services.service import Service, ServiceDnsRecord, ServiceStatus
from selfprivacy_api.utils import ReadUserData, get_domain
from selfprivacy_api.utils.block_devices import BlockDevice
import selfprivacy_api.utils.network as network_utils

18
selfprivacy_api/utils/block_devices.py
View file

@ -71,6 +71,12 @@ class BlockDevice:
def __hash__(self):
return hash(self.name)
def is_root(self) -> bool:
"""
Return True if the block device is the root device.
"""
return "/" in self.mountpoints
def stats(self) -> typing.Dict[str, typing.Any]:
"""
Update current data and return a dictionary of stats.
@ -175,6 +181,9 @@ class BlockDevices(metaclass=SingletonMetaclass):
# Ignore devices with type "rom"
if device["type"] == "rom":
continue
# Ignore iso9660 devices
if device["fstype"] == "iso9660":
continue
if device["fstype"] is None:
if "children" in device:
for child in device["children"]:
@ -218,3 +227,12 @@ class BlockDevices(metaclass=SingletonMetaclass):
if mountpoint in block_device.mountpoints:
block_devices.append(block_device)
return block_devices
def get_root_block_device(self) -> BlockDevice:
"""
Return the root block device.
"""
for block_device in self.block_devices:
if "/" in block_device.mountpoints:
return block_device
raise RuntimeError("No root block device found")

2
setup.py
View file

@ -2,7 +2,7 @@ from setuptools import setup, find_packages
setup(
name="selfprivacy_api",
version="2.2.0",
version="2.2.1",
packages=find_packages(),
scripts=[
"selfprivacy_api/app.py",

18
tests/test_block_device_utils.py
View file

@ -488,3 +488,21 @@ def test_get_block_devices_by_mountpoint(lsblk_full_mock, authorized_client):
def test_get_block_devices_by_mountpoint_no_match(lsblk_full_mock, authorized_client):
block_devices = BlockDevices().get_block_devices_by_mountpoint("/foo")
assert len(block_devices) == 0
def test_get_root_block_device(lsblk_full_mock, authorized_client):
block_device = BlockDevices().get_root_block_device()
assert block_device is not None
assert block_device.name == "sda1"
assert block_device.path == "/dev/sda1"
assert block_device.fsavail == "4605702144"
assert block_device.fssize == "19814920192"
assert block_device.fstype == "ext4"
assert block_device.fsused == "14353719296"
assert block_device.mountpoints == ["/nix/store", "/"]
assert block_device.label is None
assert block_device.uuid == "ec80c004-baec-4a2c-851d-0e1807135511"
assert block_device.size == "20210236928"
assert block_device.model is None
assert block_device.serial is None
assert block_device.type == "part"