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

refactor: Dynamic service loading

Browse source
This commit is contained in:
Inex Code 2024-12-24 15:46:48 +03:00
parent cb595959f8
commit 146b4cd1e7
No known key found for this signature in database
36 changed files with 292 additions and 974 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
selfprivacy_api/backup/__init__.py
View file

@ -246,7 +246,7 @@ class Backups:
try:
if service.can_be_backed_up() is False:
raise ValueError("cannot backup a non-backuppable service")
folders = service.get_folders()
folders = service.get_folders_to_back_up()
service_name = service.get_id()
service.pre_backup(job=job)
Jobs.update(job, status=JobStatus.RUNNING, status_text="Uploading backup")
@ -517,7 +517,7 @@ class Backups:
snapshot_id: str,
verify=True,
) -> None:
folders = service.get_folders()
folders = service.get_folders_to_back_up()
Backups.provider().backupper.restore_from_backup(
snapshot_id,
@ -717,7 +717,7 @@ class Backups:
Returns the amount of space available on the volume the given
service is located on.
"""
folders = service.get_folders()
folders = service.get_folders_to_back_up()
if folders == []:
raise ValueError("unallocated service", service.get_id())

5
selfprivacy_api/backup/backuppers/restic_backupper.py
View file

@ -222,11 +222,6 @@ class ResticBackupper(AbstractBackupper):
tags=tags,
)
logger.warning(
"Starting backup: " + " ".join(self._censor_command(backup_command))
)
logger.warning("Folders: " + str(folders))
try:
messages = ResticBackupper._run_backup_command(backup_command, job)

35
selfprivacy_api/graphql/common_types/service.py
View file

@ -6,6 +6,7 @@ import strawberry
from selfprivacy_api.graphql.common_types.backup import BackupReason
from selfprivacy_api.graphql.common_types.dns import DnsRecord
from selfprivacy_api.models.services import License
from selfprivacy_api.services import ServiceManager
from selfprivacy_api.services import Service as ServiceInterface
from selfprivacy_api.services import ServiceDnsRecord
@ -71,6 +72,28 @@ class ServiceStatusEnum(Enum):
OFF = "OFF"
@strawberry.enum
class SupportLevelEnum(Enum):
"""Enum representing the support level of a service."""
NORMAL = "normal"
EXPERIMENTAL = "experimental"
DEPRECATED = "deprecated"
COMMUNITY = "community"
UNKNOWN = "unknown"
@strawberry.experimental.pydantic.type(model=License)
class LicenseType:
free: strawberry.auto
full_name: strawberry.auto
redistributable: strawberry.auto
short_name: strawberry.auto
spdx_id: strawberry.auto
url: strawberry.auto
deprecated: strawberry.auto
def get_storage_usage(root: "Service") -> ServiceStorageUsage:
"""Get storage usage for a service"""
service = ServiceManager.get_service_by_id(root.id)
@ -176,10 +199,15 @@ class Service:
is_required: bool
is_enabled: bool
is_installed: bool
is_system_service: bool
can_be_backed_up: bool
backup_description: str
status: ServiceStatusEnum
url: Optional[str]
license: List[LicenseType]
homepage: Optional[str]
source_page: Optional[str]
support_level: SupportLevelEnum
@strawberry.field
def dns_records(self) -> Optional[List[DnsRecord]]:
@ -241,6 +269,13 @@ def service_to_graphql_service(service: ServiceInterface) -> Service:
backup_description=service.get_backup_description(),
status=ServiceStatusEnum(service.get_status().value),
url=service.get_url(),
is_system_service=service.is_system_service(),
license=[
LicenseType.from_pydantic(license) for license in service.get_license()
],
homepage=service.get_homepage(),
source_page=service.get_source_page(),
support_level=SupportLevelEnum(service.get_support_level().value),
)

6
selfprivacy_api/graphql/queries/backup.py
View file

@ -13,6 +13,7 @@ from selfprivacy_api.graphql.common_types.service import (
Service,
ServiceStatusEnum,
SnapshotInfo,
SupportLevelEnum,
service_to_graphql_service,
)
from selfprivacy_api.graphql.common_types.backup import AutobackupQuotas
@ -53,6 +54,11 @@ def tombstone_service(service_id: str) -> Service:
can_be_backed_up=False,
backup_description="",
is_installed=False,
homepage=None,
source_page=None,
license=[],
is_system_service=False,
support_level=SupportLevelEnum.UNKNOWN,
)

65
selfprivacy_api/jobs/migrate_to_binds.py
View file

@ -7,11 +7,8 @@ import logging
from pydantic import BaseModel
from selfprivacy_api.jobs import Job, JobStatus, Jobs
from selfprivacy_api.services.bitwarden import Bitwarden
from selfprivacy_api.services.forgejo import Forgejo
from selfprivacy_api.services import ServiceManager
from selfprivacy_api.services.mailserver import MailServer
from selfprivacy_api.services.nextcloud import Nextcloud
from selfprivacy_api.services.pleroma import Pleroma
from selfprivacy_api.utils import ReadUserData, WriteUserData
from selfprivacy_api.utils.huey import huey
from selfprivacy_api.utils.block_devices import BlockDevices
@ -105,6 +102,50 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
)
return
Jobs.update(
job=job,
status=JobStatus.RUNNING,
progress=0,
status_text="Checking if services are present.",
)
nextcloud_service = ServiceManager.get_service_by_id("nextcloud")
bitwarden_service = ServiceManager.get_service_by_id("bitwarden")
gitea_service = ServiceManager.get_service_by_id("gitea")
pleroma_service = ServiceManager.get_service_by_id("pleroma")
if not nextcloud_service:
Jobs.update(
job=job,
status=JobStatus.ERROR,
error="Nextcloud service not found.",
)
return
if not bitwarden_service:
Jobs.update(
job=job,
status=JobStatus.ERROR,
error="Bitwarden service not found.",
)
return
if not gitea_service:
Jobs.update(
job=job,
status=JobStatus.ERROR,
error="Gitea service not found.",
)
return
if not pleroma_service:
Jobs.update(
job=job,
status=JobStatus.ERROR,
error="Pleroma service not found.",
)
return
Jobs.update(
job=job,
status=JobStatus.RUNNING,
@ -172,7 +213,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
status_text="Migrating Nextcloud.",
)
Nextcloud().stop()
nextcloud_service.stop()
# If /volumes/sda1/nextcloud or /volumes/sdb/nextcloud exists, skip it.
if not pathlib.Path("/volumes/sda1/nextcloud").exists():
@ -187,7 +228,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
)
# Start Nextcloud
Nextcloud().start()
nextcloud_service.start()
# Perform migration of Bitwarden
@ -198,7 +239,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
status_text="Migrating Bitwarden.",
)
Bitwarden().stop()
bitwarden_service.stop()
if not pathlib.Path("/volumes/sda1/bitwarden").exists():
if not pathlib.Path("/volumes/sdb/bitwarden").exists():
@ -223,7 +264,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
)
# Start Bitwarden
Bitwarden().start()
bitwarden_service.start()
# Perform migration of Gitea
@ -234,7 +275,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
status_text="Migrating Gitea.",
)
Forgejo().stop()
gitea_service.stop()
if not pathlib.Path("/volumes/sda1/gitea").exists():
if not pathlib.Path("/volumes/sdb/gitea").exists():
@ -245,7 +286,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
group="gitea",
)
Forgejo().start()
gitea_service.start()
# Perform migration of Mail server
@ -287,7 +328,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
status_text="Migrating Pleroma.",
)
Pleroma().stop()
pleroma_service.stop()
if not pathlib.Path("/volumes/sda1/pleroma").exists():
if not pathlib.Path("/volumes/sdb/pleroma").exists():
@ -311,7 +352,7 @@ def migrate_to_binds(config: BindMigrationConfig, job: Job):
group="postgres",
)
Pleroma().start()
pleroma_service.start()
Jobs.update(
job=job,

74
selfprivacy_api/models/services.py
View file

@ -1,6 +1,16 @@
from enum import Enum
from typing import Optional
from pydantic import BaseModel
from typing import Optional, List
from pydantic import BaseModel, ConfigDict
from pydantic.alias_generators import to_camel
from selfprivacy_api.services.owned_path import OwnedPath
class BaseSchema(BaseModel):
model_config = ConfigDict(
alias_generator=to_camel,
populate_by_name=True,
from_attributes=True,
)
class ServiceStatus(Enum):
@ -15,6 +25,29 @@ class ServiceStatus(Enum):
OFF = "OFF"
class SupportLevel(Enum):
"""Enum representing the support level of a service."""
NORMAL = "normal"
EXPERIMENTAL = "experimental"
DEPRECATED = "deprecated"
COMMUNITY = "community"
UNKNOWN = "unknown"
@classmethod
def from_str(cls, support_level: str) -> "SupportLevel":
"""Return the SupportLevel from a string."""
if support_level == "normal":
return cls.NORMAL
if support_level == "experimental":
return cls.EXPERIMENTAL
if support_level == "deprecated":
return cls.DEPRECATED
if support_level == "community":
return cls.COMMUNITY
return cls.UNKNOWN
class ServiceDnsRecord(BaseModel):
type: str
name: str
@ -23,3 +56,40 @@ class ServiceDnsRecord(BaseModel):
ttl: int
display_name: str
priority: Optional[int] = None
class License(BaseSchema):
"""Model representing a license."""
deprecated: bool
free: bool
full_name: str
redistributable: bool
short_name: str
spdx_id: str
url: str
class ServiceMetaData(BaseSchema):
"""Model representing the meta data of a service."""
id: str
name: str
description: str = "No description found!"
svg_icon: str = ""
showUrl: bool = True
primary_subdomain: Optional[str] = None
is_movable: bool = False
is_required: bool = False
can_be_backed_up: bool = True
backup_description: str = "No backup description found!"
systemd_services: List[str]
user: Optional[str] = None
group: Optional[str] = None
folders: List[str] = []
owned_folders: List[OwnedPath] = []
postgre_databases: List[str] = []
license: List[License] = []
homepage: Optional[str] = None
source_page: Optional[str] = None
support_level: SupportLevel = SupportLevel.UNKNOWN

39
selfprivacy_api/services/__init__.py
View file

@ -6,21 +6,14 @@ import typing
import subprocess
import json
from typing import List
from os import path, remove
from os import path
from os import makedirs
from os import listdir
from os.path import join
from shutil import copyfile, copytree, rmtree
from selfprivacy_api.services.bitwarden import Bitwarden
from selfprivacy_api.services.forgejo import Forgejo
from selfprivacy_api.services.jitsimeet import JitsiMeet
from selfprivacy_api.services.prometheus import Prometheus
from selfprivacy_api.services.roundcube import Roundcube
from selfprivacy_api.services.mailserver import MailServer
from selfprivacy_api.services.nextcloud import Nextcloud
from selfprivacy_api.services.pleroma import Pleroma
from selfprivacy_api.services.ocserv import Ocserv
from selfprivacy_api.services.service import Service, ServiceDnsRecord
from selfprivacy_api.services.service import ServiceStatus
@ -28,7 +21,7 @@ from selfprivacy_api.utils.cached_call import redis_cached_call
import selfprivacy_api.utils.network as network_utils
from selfprivacy_api.services.api_icon import API_ICON
from selfprivacy_api.utils import USERDATA_FILE, DKIM_DIR, SECRETS_FILE, get_domain
from selfprivacy_api.utils import USERDATA_FILE, DKIM_DIR, SECRETS_FILE
from selfprivacy_api.utils.block_devices import BlockDevices
from selfprivacy_api.utils import read_account_uri
from selfprivacy_api.services.templated_service import (
@ -143,6 +136,10 @@ class ServiceManager(Service):
def is_enabled() -> bool:
return True
@staticmethod
def is_system_service() -> bool:
return True
@staticmethod
def get_backup_description() -> str:
return "General server settings."
@ -263,8 +260,6 @@ def get_templated_service(service_id: str) -> TemplatedService:
@redis_cached_call(ttl=3600)
def get_remote_service(id: str, url: str) -> TemplatedService:
# Get JSON from calling the sp-fetch-remote-module command with the URL
# Parse the JSON into a TemplatedService object
response = subprocess.run(
["sp-fetch-remote-module", url],
capture_output=True,
@ -274,23 +269,26 @@ def get_remote_service(id: str, url: str) -> TemplatedService:
return TemplatedService(id, response.stdout)
DUMMY_SERVICES = []
TEST_FLAGS: list[str] = []
@redis_cached_call(ttl=5)
def get_services() -> List[Service]:
if "ONLY_DUMMY_SERVICE" in TEST_FLAGS:
return DUMMY_SERVICES
if "DUMMY_SERVICE_AND_API" in TEST_FLAGS:
return DUMMY_SERVICES + [ServiceManager()]
hardcoded_services: list[Service] = [
Bitwarden(),
# Forgejo(),
MailServer(),
Nextcloud(),
# Pleroma(),
Ocserv(),
JitsiMeet(),
Roundcube(),
ServiceManager(),
Prometheus(),
]
if DUMMY_SERVICES:
hardcoded_services += DUMMY_SERVICES
service_ids = [service.get_id() for service in hardcoded_services]
# Load services from SP_MODULES_DEFENITIONS_PATH
templated_services: List[Service] = []
if path.exists(SP_MODULES_DEFENITIONS_PATH):
for module in listdir(SP_MODULES_DEFENITIONS_PATH):
@ -322,6 +320,3 @@ def get_services() -> List[Service]:
logger.error(f"Failed to load service {module}: {e}")
return hardcoded_services + templated_services
# services = get_services()

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

@ -1,109 +0,0 @@
"""Class representing Bitwarden service"""
import base64
import subprocess
from typing import List
from selfprivacy_api.utils.systemd import get_service_status
from selfprivacy_api.services.service import Service, ServiceStatus
from selfprivacy_api.services.bitwarden.icon import BITWARDEN_ICON
from selfprivacy_api.services.config_item import (
StringServiceConfigItem,
BoolServiceConfigItem,
ServiceConfigItem,
)
from selfprivacy_api.utils.regex_strings import SUBDOMAIN_REGEX
class Bitwarden(Service):
"""Class representing Bitwarden service."""
config_items: dict[str, ServiceConfigItem] = {
"subdomain": StringServiceConfigItem(
id="subdomain",
default_value="password",
description="Subdomain",
regex=SUBDOMAIN_REGEX,
widget="subdomain",
),
"signupsAllowed": BoolServiceConfigItem(
id="signupsAllowed",
default_value=True,
description="Allow new user signups",
),
"sendsAllowed": BoolServiceConfigItem(
id="sendsAllowed",
default_value=True,
description="Allow users to use Bitwarden Send",
),
"emergencyAccessAllowed": BoolServiceConfigItem(
id="emergencyAccessAllowed",
default_value=True,
description="Allow users to enable Emergency Access",
),
}
@staticmethod
def get_id() -> str:
"""Return service id."""
return "bitwarden"
@staticmethod
def get_display_name() -> str:
"""Return service display name."""
return "Bitwarden"
@staticmethod
def get_description() -> str:
"""Return service description."""
return "Bitwarden is a password manager."
@staticmethod
def get_svg_icon() -> str:
"""Read SVG icon from file and return it as base64 encoded string."""
return base64.b64encode(BITWARDEN_ICON.encode("utf-8")).decode("utf-8")
@staticmethod
def get_user() -> str:
return "vaultwarden"
@staticmethod
def is_movable() -> bool:
return True
@staticmethod
def is_required() -> bool:
return False
@staticmethod
def get_backup_description() -> str:
return "Password database, encryption certificate and attachments."
@staticmethod
def get_status() -> ServiceStatus:
"""
Return Bitwarden status from systemd.
Use command return code to determine status.
Return code 0 means service is running.
Return code 1 or 2 means service is in error stat.
Return code 3 means service is stopped.
Return code 4 means service is off.
"""
return get_service_status("vaultwarden.service")
@staticmethod
def stop():
subprocess.run(["systemctl", "stop", "vaultwarden.service"])
@staticmethod
def start():
subprocess.run(["systemctl", "start", "vaultwarden.service"])
@staticmethod
def restart():
subprocess.run(["systemctl", "restart", "vaultwarden.service"])
@staticmethod
def get_folders() -> List[str]:
return ["/var/lib/bitwarden", "/var/lib/bitwarden_rs"]

3
selfprivacy_api/services/bitwarden/bitwarden.svg
View file

@ -1,3 +0,0 @@
<svg width="24" height="24" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<path d="M5.125 2C4.2962 2 3.50134 2.32924 2.91529 2.91529C2.32924 3.50134 2 4.2962 2 5.125L2 18.875C2 19.7038 2.32924 20.4987 2.91529 21.0847C3.50134 21.6708 4.2962 22 5.125 22H18.875C19.7038 22 20.4987 21.6708 21.0847 21.0847C21.6708 20.4987 22 19.7038 22 18.875V5.125C22 4.2962 21.6708 3.50134 21.0847 2.91529C20.4987 2.32924 19.7038 2 18.875 2H5.125ZM6.25833 4.43333H17.7583C17.9317 4.43333 18.0817 4.49667 18.2083 4.62333C18.2688 4.68133 18.3168 4.7511 18.3494 4.82835C18.3819 4.9056 18.3983 4.98869 18.3975 5.0725V12.7392C18.3975 13.3117 18.2858 13.8783 18.0633 14.4408C17.8558 14.9751 17.5769 15.4789 17.2342 15.9383C16.8824 16.3987 16.4882 16.825 16.0567 17.2117C15.6008 17.6242 15.18 17.9667 14.7942 18.24C14.4075 18.5125 14.005 18.77 13.5858 19.0133C13.1667 19.2558 12.8692 19.4208 12.6925 19.5075C12.5158 19.5942 12.375 19.6608 12.2675 19.7075C12.1872 19.7472 12.0987 19.7674 12.0092 19.7667C11.919 19.7674 11.8299 19.7468 11.7492 19.7067C11.6062 19.6429 11.4645 19.5762 11.3242 19.5067C11.0218 19.3511 10.7242 19.1866 10.4317 19.0133C10.0175 18.7738 9.6143 18.5158 9.22333 18.24C8.7825 17.9225 8.36093 17.5791 7.96083 17.2117C7.52907 16.825 7.13456 16.3987 6.7825 15.9383C6.44006 15.4788 6.16141 14.9751 5.95417 14.4408C5.73555 13.9 5.62213 13.3225 5.62 12.7392V5.0725C5.62 4.89917 5.68333 4.75 5.80917 4.6225C5.86726 4.56188 5.93717 4.51382 6.01457 4.48129C6.09196 4.44875 6.17521 4.43243 6.25917 4.43333H6.25833ZM12.0083 6.35V17.7C12.8 17.2817 13.5092 16.825 14.135 16.3333C15.6992 15.1083 16.4808 13.9108 16.4808 12.7392V6.35H12.0083Z" fill="black"/>
</svg>
Side by side

Before

Width:  |  Height:  |  Size: 1.6 KiB

5
selfprivacy_api/services/bitwarden/icon.py
View file

@ -1,5 +0,0 @@
BITWARDEN_ICON = """
<svg width="24" height="24" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<path d="M5.125 2C4.2962 2 3.50134 2.32924 2.91529 2.91529C2.32924 3.50134 2 4.2962 2 5.125L2 18.875C2 19.7038 2.32924 20.4987 2.91529 21.0847C3.50134 21.6708 4.2962 22 5.125 22H18.875C19.7038 22 20.4987 21.6708 21.0847 21.0847C21.6708 20.4987 22 19.7038 22 18.875V5.125C22 4.2962 21.6708 3.50134 21.0847 2.91529C20.4987 2.32924 19.7038 2 18.875 2H5.125ZM6.25833 4.43333H17.7583C17.9317 4.43333 18.0817 4.49667 18.2083 4.62333C18.2688 4.68133 18.3168 4.7511 18.3494 4.82835C18.3819 4.9056 18.3983 4.98869 18.3975 5.0725V12.7392C18.3975 13.3117 18.2858 13.8783 18.0633 14.4408C17.8558 14.9751 17.5769 15.4789 17.2342 15.9383C16.8824 16.3987 16.4882 16.825 16.0567 17.2117C15.6008 17.6242 15.18 17.9667 14.7942 18.24C14.4075 18.5125 14.005 18.77 13.5858 19.0133C13.1667 19.2558 12.8692 19.4208 12.6925 19.5075C12.5158 19.5942 12.375 19.6608 12.2675 19.7075C12.1872 19.7472 12.0987 19.7674 12.0092 19.7667C11.919 19.7674 11.8299 19.7468 11.7492 19.7067C11.6062 19.6429 11.4645 19.5762 11.3242 19.5067C11.0218 19.3511 10.7242 19.1866 10.4317 19.0133C10.0175 18.7738 9.6143 18.5158 9.22333 18.24C8.7825 17.9225 8.36093 17.5791 7.96083 17.2117C7.52907 16.825 7.13456 16.3987 6.7825 15.9383C6.44006 15.4788 6.16141 14.9751 5.95417 14.4408C5.73555 13.9 5.62213 13.3225 5.62 12.7392V5.0725C5.62 4.89917 5.68333 4.75 5.80917 4.6225C5.86726 4.56188 5.93717 4.51382 6.01457 4.48129C6.09196 4.44875 6.17521 4.43243 6.25917 4.43333H6.25833ZM12.0083 6.35V17.7C12.8 17.2817 13.5092 16.825 14.135 16.3333C15.6992 15.1083 16.4808 13.9108 16.4808 12.7392V6.35H12.0083Z" fill="black"/>
</svg>
"""

134
selfprivacy_api/services/forgejo/__init__.py
View file

@ -1,134 +0,0 @@
"""Class representing Bitwarden service"""
import base64
import subprocess
from typing import List
from selfprivacy_api.utils import ReadUserData, WriteUserData
from selfprivacy_api.utils.systemd import get_service_status
from selfprivacy_api.services.service import Service, ServiceStatus
from selfprivacy_api.services.forgejo.icon import FORGEJO_ICON
from selfprivacy_api.services.config_item import (
StringServiceConfigItem,
BoolServiceConfigItem,
EnumServiceConfigItem,
ServiceConfigItem,
)
from selfprivacy_api.utils.regex_strings import SUBDOMAIN_REGEX
class Forgejo(Service):
"""Class representing Forgejo service.
Previously was Gitea, so some IDs are still called gitea for compatibility.
"""
config_items: dict[str, ServiceConfigItem] = {
"subdomain": StringServiceConfigItem(
id="subdomain",
default_value="git",
description="Subdomain",
regex=SUBDOMAIN_REGEX,
widget="subdomain",
),
"appName": StringServiceConfigItem(
id="appName",
default_value="SelfPrivacy git Service",
description="The name displayed in the web interface",
),
"enableLfs": BoolServiceConfigItem(
id="enableLfs",
default_value=True,
description="Enable Git LFS",
),
"forcePrivate": BoolServiceConfigItem(
id="forcePrivate",
default_value=False,
description="Force all new repositories to be private",
),
"disableRegistration": BoolServiceConfigItem(
id="disableRegistration",
default_value=False,
description="Disable registration of new users",
),
"requireSigninView": BoolServiceConfigItem(
id="requireSigninView",
default_value=False,
description="Force users to log in to view any page",
),
"defaultTheme": EnumServiceConfigItem(
id="defaultTheme",
default_value="forgejo-auto",
description="Default theme",
options=[
"forgejo-auto",
"forgejo-light",
"forgejo-dark",
"gitea-auto",
"gitea-light",
"gitea-dark",
],
),
}
@staticmethod
def get_id() -> str:
"""Return service id. For compatibility keep in gitea."""
return "gitea"
@staticmethod
def get_display_name() -> str:
"""Return service display name."""
return "Forgejo"
@staticmethod
def get_description() -> str:
"""Return service description."""
return "Forgejo is a Git forge."
@staticmethod
def get_svg_icon() -> str:
"""Read SVG icon from file and return it as base64 encoded string."""
return base64.b64encode(FORGEJO_ICON.encode("utf-8")).decode("utf-8")
@staticmethod
def is_movable() -> bool:
return True
@staticmethod
def is_required() -> bool:
return False
@staticmethod
def get_backup_description() -> str:
return "Git repositories, database and user data."
@staticmethod
def get_status() -> ServiceStatus:
"""
Return Gitea status from systemd.
Use command return code to determine status.
Return code 0 means service is running.
Return code 1 or 2 means service is in error stat.
Return code 3 means service is stopped.
Return code 4 means service is off.
"""
return get_service_status("forgejo.service")
@staticmethod
def stop():
subprocess.run(["systemctl", "stop", "forgejo.service"])
@staticmethod
def start():
subprocess.run(["systemctl", "start", "forgejo.service"])
@staticmethod
def restart():
subprocess.run(["systemctl", "restart", "forgejo.service"])
@staticmethod
def get_folders() -> List[str]:
"""The data folder is still called gitea for compatibility."""
return ["/var/lib/gitea"]

3
selfprivacy_api/services/forgejo/gitea.svg
View file

@ -1,3 +0,0 @@
<svg width="24" height="24" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<path d="M2.60007 10.5899L8.38007 4.79995L10.0701 6.49995C9.83007 7.34995 10.2201 8.27995 11.0001 8.72995V14.2699C10.4001 14.6099 10.0001 15.2599 10.0001 15.9999C10.0001 16.5304 10.2108 17.0391 10.5859 17.4142C10.9609 17.7892 11.4696 17.9999 12.0001 17.9999C12.5305 17.9999 13.0392 17.7892 13.4143 17.4142C13.7894 17.0391 14.0001 16.5304 14.0001 15.9999C14.0001 15.2599 13.6001 14.6099 13.0001 14.2699V9.40995L15.0701 11.4999C15.0001 11.6499 15.0001 11.8199 15.0001 11.9999C15.0001 12.5304 15.2108 13.0391 15.5859 13.4142C15.9609 13.7892 16.4696 13.9999 17.0001 13.9999C17.5305 13.9999 18.0392 13.7892 18.4143 13.4142C18.7894 13.0391 19.0001 12.5304 19.0001 11.9999C19.0001 11.4695 18.7894 10.9608 18.4143 10.5857C18.0392 10.2107 17.5305 9.99995 17.0001 9.99995C16.8201 9.99995 16.6501 9.99995 16.5001 10.0699L13.9301 7.49995C14.1901 6.56995 13.7101 5.54995 12.7801 5.15995C12.3501 4.99995 11.9001 4.95995 11.5001 5.06995L9.80007 3.37995L10.5901 2.59995C11.3701 1.80995 12.6301 1.80995 13.4101 2.59995L21.4001 10.5899C22.1901 11.3699 22.1901 12.6299 21.4001 13.4099L13.4101 21.3999C12.6301 22.1899 11.3701 22.1899 10.5901 21.3999L2.60007 13.4099C1.81007 12.6299 1.81007 11.3699 2.60007 10.5899Z" fill="black"/>
</svg>
Side by side

Before

Width:  |  Height:  |  Size: 1.3 KiB

5
selfprivacy_api/services/forgejo/icon.py
View file

@ -1,5 +0,0 @@
FORGEJO_ICON = """
<svg width="24" height="24" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<path d="M2.60007 10.5899L8.38007 4.79995L10.0701 6.49995C9.83007 7.34995 10.2201 8.27995 11.0001 8.72995V14.2699C10.4001 14.6099 10.0001 15.2599 10.0001 15.9999C10.0001 16.5304 10.2108 17.0391 10.5859 17.4142C10.9609 17.7892 11.4696 17.9999 12.0001 17.9999C12.5305 17.9999 13.0392 17.7892 13.4143 17.4142C13.7894 17.0391 14.0001 16.5304 14.0001 15.9999C14.0001 15.2599 13.6001 14.6099 13.0001 14.2699V9.40995L15.0701 11.4999C15.0001 11.6499 15.0001 11.8199 15.0001 11.9999C15.0001 12.5304 15.2108 13.0391 15.5859 13.4142C15.9609 13.7892 16.4696 13.9999 17.0001 13.9999C17.5305 13.9999 18.0392 13.7892 18.4143 13.4142C18.7894 13.0391 19.0001 12.5304 19.0001 11.9999C19.0001 11.4695 18.7894 10.9608 18.4143 10.5857C18.0392 10.2107 17.5305 9.99995 17.0001 9.99995C16.8201 9.99995 16.6501 9.99995 16.5001 10.0699L13.9301 7.49995C14.1901 6.56995 13.7101 5.54995 12.7801 5.15995C12.3501 4.99995 11.9001 4.95995 11.5001 5.06995L9.80007 3.37995L10.5901 2.59995C11.3701 1.80995 12.6301 1.80995 13.4101 2.59995L21.4001 10.5899C22.1901 11.3699 22.1901 12.6299 21.4001 13.4099L13.4101 21.3999C12.6301 22.1899 11.3701 22.1899 10.5901 21.3999L2.60007 13.4099C1.81007 12.6299 1.81007 11.3699 2.60007 10.5899Z" fill="black"/>
</svg>
"""

109
selfprivacy_api/services/jitsimeet/__init__.py
View file

@ -1,109 +0,0 @@
"""Class representing Jitsi Meet service"""
import base64
import subprocess
from typing import List
from selfprivacy_api.jobs import Job
from selfprivacy_api.utils.systemd import (
get_service_status_from_several_units,
)
from selfprivacy_api.services.service import Service, ServiceStatus
from selfprivacy_api.utils.block_devices import BlockDevice
from selfprivacy_api.services.jitsimeet.icon import JITSI_ICON
from selfprivacy_api.services.config_item import (
StringServiceConfigItem,
ServiceConfigItem,
)
from selfprivacy_api.utils.regex_strings import SUBDOMAIN_REGEX
class JitsiMeet(Service):
"""Class representing Jitsi service"""
config_items: dict[str, ServiceConfigItem] = {
"subdomain": StringServiceConfigItem(
id="subdomain",
default_value="meet",
description="Subdomain",
regex=SUBDOMAIN_REGEX,
widget="subdomain",
),
"appName": StringServiceConfigItem(
id="appName",
default_value="Jitsi Meet",
description="The name displayed in the web interface",
),
}
@staticmethod
def get_id() -> str:
"""Return service id."""
return "jitsi-meet"
@staticmethod
def get_display_name() -> str:
"""Return service display name."""
return "JitsiMeet"
@staticmethod
def get_description() -> str:
"""Return service description."""
return "Jitsi Meet is a free and open-source video conferencing solution."
@staticmethod
def get_svg_icon() -> str:
"""Read SVG icon from file and return it as base64 encoded string."""
return base64.b64encode(JITSI_ICON.encode("utf-8")).decode("utf-8")
@staticmethod
def is_movable() -> bool:
return False
@staticmethod
def is_required() -> bool:
return False
@staticmethod
def get_backup_description() -> str:
return "Secrets that are used to encrypt the communication."
@staticmethod
def get_status() -> ServiceStatus:
return get_service_status_from_several_units(
["prosody.service", "jitsi-videobridge2.service", "jicofo.service"]
)
@staticmethod
def stop():
subprocess.run(
["systemctl", "stop", "jitsi-videobridge2.service"],
check=False,
)
subprocess.run(["systemctl", "stop", "jicofo.service"], check=False)
subprocess.run(["systemctl", "stop", "prosody.service"], check=False)
@staticmethod
def start():
subprocess.run(["systemctl", "start", "prosody.service"], check=False)
subprocess.run(
["systemctl", "start", "jitsi-videobridge2.service"],
check=False,
)
subprocess.run(["systemctl", "start", "jicofo.service"], check=False)
@staticmethod
def restart():
subprocess.run(["systemctl", "restart", "prosody.service"], check=False)
subprocess.run(
["systemctl", "restart", "jitsi-videobridge2.service"],
check=False,
)
subprocess.run(["systemctl", "restart", "jicofo.service"], check=False)
@staticmethod
def get_folders() -> List[str]:
return ["/var/lib/jitsi-meet"]
def move_to_volume(self, volume: BlockDevice) -> Job:
raise NotImplementedError("jitsi-meet service is not movable")

5
selfprivacy_api/services/jitsimeet/icon.py
View file

@ -1,5 +0,0 @@
JITSI_ICON = """
<svg width="32" height="32" viewBox="0 0 32 32" fill="none" xmlns="http://www.w3.org/2000/svg">
<path d="M26.6665 2.66663H5.33317C3.8665 2.66663 2.67984 3.86663 2.67984 5.33329L2.6665 29.3333L7.99984 24H26.6665C28.1332 24 29.3332 22.8 29.3332 21.3333V5.33329C29.3332 3.86663 28.1332 2.66663 26.6665 2.66663ZM26.6665 21.3333H6.89317L5.33317 22.8933V5.33329H26.6665V21.3333ZM18.6665 14.1333L22.6665 17.3333V9.33329L18.6665 12.5333V9.33329H9.33317V17.3333H18.6665V14.1333Z" fill="black"/>
</svg>
"""

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

@ -1,99 +0,0 @@
"""Class representing Nextcloud service."""
import base64
import subprocess
from typing import List
from selfprivacy_api.utils.systemd import get_service_status
from selfprivacy_api.services.service import Service, ServiceStatus
from selfprivacy_api.services.nextcloud.icon import NEXTCLOUD_ICON
from selfprivacy_api.services.config_item import (
StringServiceConfigItem,
BoolServiceConfigItem,
ServiceConfigItem,
)
from selfprivacy_api.utils.regex_strings import SUBDOMAIN_REGEX
class Nextcloud(Service):
"""Class representing Nextcloud service."""
config_items: dict[str, ServiceConfigItem] = {
"subdomain": StringServiceConfigItem(
id="subdomain",
default_value="cloud",
description="Subdomain",
regex=SUBDOMAIN_REGEX,
widget="subdomain",
),
"enableImagemagick": BoolServiceConfigItem(
id="enableImagemagick",
default_value=True,
description="Enable ImageMagick",
),
}
@staticmethod
def get_id() -> str:
"""Return service id."""
return "nextcloud"
@staticmethod
def get_display_name() -> str:
"""Return service display name."""
return "Nextcloud"
@staticmethod
def get_description() -> str:
"""Return service description."""
return "Nextcloud is a cloud storage service that offers a web interface and a desktop client."
@staticmethod
def get_svg_icon() -> str:
"""Read SVG icon from file and return it as base64 encoded string."""
return base64.b64encode(NEXTCLOUD_ICON.encode("utf-8")).decode("utf-8")
@staticmethod
def is_movable() -> bool:
return True
@staticmethod
def is_required() -> bool:
return False
@staticmethod
def get_backup_description() -> str:
return "All the files and other data stored in Nextcloud."
@staticmethod
def get_status() -> ServiceStatus:
"""
Return Nextcloud status from systemd.
Use command return code to determine status.
Return code 0 means service is running.
Return code 1 or 2 means service is in error stat.
Return code 3 means service is stopped.
Return code 4 means service is off.
"""
return get_service_status("phpfpm-nextcloud.service")
@staticmethod
def stop():
"""Stop Nextcloud service."""
subprocess.Popen(["systemctl", "stop", "phpfpm-nextcloud.service"])
@staticmethod
def start():
"""Start Nextcloud service."""
subprocess.Popen(["systemctl", "start", "phpfpm-nextcloud.service"])
@staticmethod
def restart():
"""Restart Nextcloud service."""
subprocess.Popen(["systemctl", "restart", "phpfpm-nextcloud.service"])
@staticmethod
def get_folders() -> List[str]:
return ["/var/lib/nextcloud"]

12
selfprivacy_api/services/nextcloud/icon.py
View file

@ -1,12 +0,0 @@
NEXTCLOUD_ICON = """
<svg width="24" height="24" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<g clip-path="url(#clip0_51106_4974)">
<path d="M12.018 6.53699C9.518 6.53699 7.418 8.24899 6.777 10.552C6.217 9.31999 4.984 8.44699 3.552 8.44699C2.61116 8.45146 1.71014 8.82726 1.04495 9.49264C0.379754 10.158 0.00420727 11.0591 0 12C0.00420727 12.9408 0.379754 13.842 1.04495 14.5073C1.71014 15.1727 2.61116 15.5485 3.552 15.553C4.984 15.553 6.216 14.679 6.776 13.447C7.417 15.751 9.518 17.463 12.018 17.463C14.505 17.463 16.594 15.77 17.249 13.486C17.818 14.696 19.032 15.553 20.447 15.553C21.3881 15.549 22.2895 15.1734 22.955 14.508C23.6205 13.8425 23.9961 12.9411 24 12C23.9958 11.059 23.6201 10.1577 22.9547 9.49229C22.2893 8.82688 21.388 8.4512 20.447 8.44699C19.031 8.44699 17.817 9.30499 17.248 10.514C16.594 8.22999 14.505 6.53599 12.018 6.53699ZM12.018 8.62199C13.896 8.62199 15.396 10.122 15.396 12C15.396 13.878 13.896 15.378 12.018 15.378C11.5739 15.38 11.1338 15.2939 10.7231 15.1249C10.3124 14.9558 9.93931 14.707 9.62532 14.393C9.31132 14.0789 9.06267 13.7057 8.89373 13.295C8.72478 12.8842 8.63888 12.4441 8.641 12C8.641 10.122 10.141 8.62199 12.018 8.62199ZM3.552 10.532C4.374 10.532 5.019 11.177 5.019 12C5.019 12.823 4.375 13.467 3.552 13.468C3.35871 13.47 3.16696 13.4334 2.988 13.3603C2.80905 13.2872 2.64648 13.1792 2.50984 13.0424C2.3732 12.9057 2.26524 12.7431 2.19229 12.5641C2.11934 12.3851 2.08286 12.1933 2.085 12C2.085 11.177 2.729 10.533 3.552 10.533V10.532ZM20.447 10.532C21.27 10.532 21.915 11.177 21.915 12C21.915 12.823 21.27 13.468 20.447 13.468C20.2537 13.47 20.062 13.4334 19.883 13.3603C19.704 13.2872 19.5415 13.1792 19.4048 13.0424C19.2682 12.9057 19.1602 12.7431 19.0873 12.5641C19.0143 12.3851 18.9779 12.1933 18.98 12C18.98 11.177 19.624 10.533 20.447 10.533V10.532Z" fill="black"/>
</g>
<defs>
<clipPath id="clip0_51106_4974">
<rect width="24" height="24" fill="white"/>
</clipPath>
</defs>
</svg>
"""

10
selfprivacy_api/services/nextcloud/nextcloud.svg
View file

@ -1,10 +0,0 @@
<svg width="24" height="24" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<g clip-path="url(#clip0_51106_4974)">
<path d="M12.018 6.53699C9.518 6.53699 7.418 8.24899 6.777 10.552C6.217 9.31999 4.984 8.44699 3.552 8.44699C2.61116 8.45146 1.71014 8.82726 1.04495 9.49264C0.379754 10.158 0.00420727 11.0591 0 12C0.00420727 12.9408 0.379754 13.842 1.04495 14.5073C1.71014 15.1727 2.61116 15.5485 3.552 15.553C4.984 15.553 6.216 14.679 6.776 13.447C7.417 15.751 9.518 17.463 12.018 17.463C14.505 17.463 16.594 15.77 17.249 13.486C17.818 14.696 19.032 15.553 20.447 15.553C21.3881 15.549 22.2895 15.1734 22.955 14.508C23.6205 13.8425 23.9961 12.9411 24 12C23.9958 11.059 23.6201 10.1577 22.9547 9.49229C22.2893 8.82688 21.388 8.4512 20.447 8.44699C19.031 8.44699 17.817 9.30499 17.248 10.514C16.594 8.22999 14.505 6.53599 12.018 6.53699ZM12.018 8.62199C13.896 8.62199 15.396 10.122 15.396 12C15.396 13.878 13.896 15.378 12.018 15.378C11.5739 15.38 11.1338 15.2939 10.7231 15.1249C10.3124 14.9558 9.93931 14.707 9.62532 14.393C9.31132 14.0789 9.06267 13.7057 8.89373 13.295C8.72478 12.8842 8.63888 12.4441 8.641 12C8.641 10.122 10.141 8.62199 12.018 8.62199ZM3.552 10.532C4.374 10.532 5.019 11.177 5.019 12C5.019 12.823 4.375 13.467 3.552 13.468C3.35871 13.47 3.16696 13.4334 2.988 13.3603C2.80905 13.2872 2.64648 13.1792 2.50984 13.0424C2.3732 12.9057 2.26524 12.7431 2.19229 12.5641C2.11934 12.3851 2.08286 12.1933 2.085 12C2.085 11.177 2.729 10.533 3.552 10.533V10.532ZM20.447 10.532C21.27 10.532 21.915 11.177 21.915 12C21.915 12.823 21.27 13.468 20.447 13.468C20.2537 13.47 20.062 13.4334 19.883 13.3603C19.704 13.2872 19.5415 13.1792 19.4048 13.0424C19.2682 12.9057 19.1602 12.7431 19.0873 12.5641C19.0143 12.3851 18.9779 12.1933 18.98 12C18.98 11.177 19.624 10.533 20.447 10.533V10.532Z" fill="black"/>
</g>
<defs>
<clipPath id="clip0_51106_4974">
<rect width="24" height="24" fill="white"/>
</clipPath>
</defs>
</svg>
Side by side

Before

Width:  |  Height:  |  Size: 1.9 KiB

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

@ -1,82 +0,0 @@
"""Class representing ocserv service."""
import base64
import subprocess
import typing
from selfprivacy_api.jobs import Job
from selfprivacy_api.utils.systemd import get_service_status
from selfprivacy_api.services.service import Service, ServiceStatus
from selfprivacy_api.utils.block_devices import BlockDevice
from selfprivacy_api.services.ocserv.icon import OCSERV_ICON
class Ocserv(Service):
"""Class representing ocserv service."""
@staticmethod
def get_id() -> str:
return "ocserv"
@staticmethod
def get_display_name() -> str:
return "OpenConnect VPN"
@staticmethod
def get_description() -> str:
return "OpenConnect VPN to connect your devices and access the internet."
@staticmethod
def get_svg_icon() -> str:
return base64.b64encode(OCSERV_ICON.encode("utf-8")).decode("utf-8")
@classmethod
def get_url(cls) -> typing.Optional[str]:
"""Return service url."""
return None
@staticmethod
def is_movable() -> bool:
return False
@staticmethod
def is_required() -> bool:
return False
@staticmethod
def can_be_backed_up() -> bool:
return False
@staticmethod
def get_backup_description() -> str:
return "Nothing to backup."
@staticmethod
def get_status() -> ServiceStatus:
return get_service_status("ocserv.service")
@staticmethod
def stop():
subprocess.run(["systemctl", "stop", "ocserv.service"], check=False)
@staticmethod
def start():
subprocess.run(["systemctl", "start", "ocserv.service"], check=False)
@staticmethod
def restart():
subprocess.run(["systemctl", "restart", "ocserv.service"], check=False)
@classmethod
def get_configuration(cls):
return {}
@classmethod
def set_configuration(cls, config_items):
return super().set_configuration(config_items)
@staticmethod
def get_folders() -> typing.List[str]:
return []
def move_to_volume(self, volume: BlockDevice) -> Job:
raise NotImplementedError("ocserv service is not movable")

5
selfprivacy_api/services/ocserv/icon.py
View file

@ -1,5 +0,0 @@
OCSERV_ICON = """
<svg width="24" height="24" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<path d="M12 1L3 5V11C3 16.55 6.84 21.74 12 23C17.16 21.74 21 16.55 21 11V5L12 1ZM12 11.99H19C18.47 16.11 15.72 19.78 12 20.93V12H5V6.3L12 3.19V11.99Z" fill="black"/>
</svg>
"""

3
selfprivacy_api/services/ocserv/ocserv.svg
View file

@ -1,3 +0,0 @@
<svg width="24" height="24" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<path d="M12 1L3 5V11C3 16.55 6.84 21.74 12 23C17.16 21.74 21 16.55 21 11V5L12 1ZM12 11.99H19C18.47 16.11 15.72 19.78 12 20.93V12H5V6.3L12 3.19V11.99Z" fill="black"/>
</svg>
Side by side

Before

Width:  |  Height:  |  Size: 270 B

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

@ -1,89 +0,0 @@
"""Class representing Nextcloud service."""
import base64
import subprocess
from typing import List
from selfprivacy_api.services.owned_path import OwnedPath
from selfprivacy_api.utils.systemd import get_service_status
from selfprivacy_api.services.service import Service, ServiceStatus
from selfprivacy_api.services.pleroma.icon import PLEROMA_ICON
class Pleroma(Service):
"""Class representing Pleroma service."""
@staticmethod
def get_id() -> str:
return "pleroma"
@staticmethod
def get_display_name() -> str:
return "Pleroma"
@staticmethod
def get_description() -> str:
return "Pleroma is a microblogging service that offers a web interface and a desktop client."
@staticmethod
def get_svg_icon() -> str:
return base64.b64encode(PLEROMA_ICON.encode("utf-8")).decode("utf-8")
@staticmethod
def is_movable() -> bool:
return True
@staticmethod
def is_required() -> bool:
return False
@staticmethod
def get_backup_description() -> str:
return "Your Pleroma accounts, posts and media."
@staticmethod
def get_status() -> ServiceStatus:
return get_service_status("pleroma.service")
@staticmethod
def stop():
subprocess.run(["systemctl", "stop", "pleroma.service"])
subprocess.run(["systemctl", "stop", "postgresql.service"])
@staticmethod
def start():
subprocess.run(["systemctl", "start", "pleroma.service"])
subprocess.run(["systemctl", "start", "postgresql.service"])
@staticmethod
def restart():
subprocess.run(["systemctl", "restart", "pleroma.service"])
subprocess.run(["systemctl", "restart", "postgresql.service"])
@classmethod
def get_configuration(cls):
return {}
@classmethod
def set_configuration(cls, config_items):
return super().set_configuration(config_items)
@staticmethod
def get_owned_folders() -> List[OwnedPath]:
"""
Get a list of occupied directories with ownership info
Pleroma has folders that are owned by different users
"""
return [
OwnedPath(
path="/var/lib/pleroma",
owner="pleroma",
group="pleroma",
),
OwnedPath(
path="/var/lib/postgresql",
owner="postgres",
group="postgres",
),
]

12
selfprivacy_api/services/pleroma/icon.py
View file

@ -1,12 +0,0 @@
PLEROMA_ICON = """
<svg width="24" height="24" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<g clip-path="url(#clip0_51106_4998)">
<path d="M6.35999 1.07076e-06C6.11451 -0.000261753 5.87139 0.0478616 5.64452 0.14162C5.41766 0.235378 5.21149 0.372932 5.03782 0.546418C4.86415 0.719904 4.72638 0.925919 4.63237 1.15269C4.53837 1.37945 4.48999 1.62252 4.48999 1.868V24H10.454V1.07076e-06H6.35999ZM13.473 1.07076e-06V12H17.641C18.1364 12 18.6115 11.8032 18.9619 11.4529C19.3122 11.1026 19.509 10.6274 19.509 10.132V1.07076e-06H13.473ZM13.473 18.036V24H17.641C18.1364 24 18.6115 23.8032 18.9619 23.4529C19.3122 23.1026 19.509 22.6274 19.509 22.132V18.036H13.473Z" fill="black"/>
</g>
<defs>
<clipPath id="clip0_51106_4998">
<rect width="24" height="24" fill="white"/>
</clipPath>
</defs>
</svg>
"""

10
selfprivacy_api/services/pleroma/pleroma.svg
View file

@ -1,10 +0,0 @@
<svg width="24" height="24" viewBox="0 0 24 24" fill="none" xmlns="http://www.w3.org/2000/svg">
<g clip-path="url(#clip0_51106_4998)">
<path d="M6.35999 1.07076e-06C6.11451 -0.000261753 5.87139 0.0478616 5.64452 0.14162C5.41766 0.235378 5.21149 0.372932 5.03782 0.546418C4.86415 0.719904 4.72638 0.925919 4.63237 1.15269C4.53837 1.37945 4.48999 1.62252 4.48999 1.868V24H10.454V1.07076e-06H6.35999ZM13.473 1.07076e-06V12H17.641C18.1364 12 18.6115 11.8032 18.9619 11.4529C19.3122 11.1026 19.509 10.6274 19.509 10.132V1.07076e-06H13.473ZM13.473 18.036V24H17.641C18.1364 24 18.6115 23.8032 18.9619 23.4529C19.3122 23.1026 19.509 22.6274 19.509 22.132V18.036H13.473Z" fill="black"/>
</g>
<defs>
<clipPath id="clip0_51106_4998">
<rect width="24" height="24" fill="white"/>
</clipPath>
</defs>
</svg>
Side by side

Before

Width:  |  Height:  |  Size: 794 B

4
selfprivacy_api/services/prometheus/__init__.py
View file

@ -47,6 +47,10 @@ class Prometheus(Service):
def is_required() -> bool:
return True
@staticmethod
def is_system_service() -> bool:
return True
@staticmethod
def can_be_backed_up() -> bool:
return False

100
selfprivacy_api/services/roundcube/__init__.py
View file

@ -1,100 +0,0 @@
"""Class representing Roundcube service"""
import base64
import subprocess
from typing import List
from selfprivacy_api.jobs import Job
from selfprivacy_api.utils.systemd import (
get_service_status_from_several_units,
)
from selfprivacy_api.services.service import Service, ServiceStatus
from selfprivacy_api.utils.block_devices import BlockDevice
from selfprivacy_api.services.roundcube.icon import ROUNDCUBE_ICON
from selfprivacy_api.services.config_item import (
StringServiceConfigItem,
ServiceConfigItem,
)
from selfprivacy_api.utils.regex_strings import SUBDOMAIN_REGEX
class Roundcube(Service):
"""Class representing roundcube service"""
config_items: dict[str, ServiceConfigItem] = {
"subdomain": StringServiceConfigItem(
id="subdomain",
default_value="roundcube",
description="Subdomain",
regex=SUBDOMAIN_REGEX,
widget="subdomain",
),
}
@staticmethod
def get_id() -> str:
"""Return service id."""
return "roundcube"
@staticmethod
def get_display_name() -> str:
"""Return service display name."""
return "Roundcube"
@staticmethod
def get_description() -> str:
"""Return service description."""
return "Roundcube is an open source webmail software."
@staticmethod
def get_svg_icon() -> str:
"""Read SVG icon from file and return it as base64 encoded string."""
return base64.b64encode(ROUNDCUBE_ICON.encode("utf-8")).decode("utf-8")
@staticmethod
def is_movable() -> bool:
return False
@staticmethod
def is_required() -> bool:
return False
@staticmethod
def can_be_backed_up() -> bool:
return False
@staticmethod
def get_backup_description() -> str:
return "Nothing to backup."
@staticmethod
def get_status() -> ServiceStatus:
return get_service_status_from_several_units(["phpfpm-roundcube.service"])
@staticmethod
def stop():
subprocess.run(
["systemctl", "stop", "phpfpm-roundcube.service"],
check=False,
)
@staticmethod
def start():
subprocess.run(
["systemctl", "start", "phpfpm-roundcube.service"],
check=False,
)
@staticmethod
def restart():
subprocess.run(
["systemctl", "restart", "phpfpm-roundcube.service"],
check=False,
)
@staticmethod
def get_folders() -> List[str]:
return []
def move_to_volume(self, volume: BlockDevice) -> Job:
raise NotImplementedError("roundcube service is not movable")

7
selfprivacy_api/services/roundcube/icon.py
View file

@ -1,7 +0,0 @@
ROUNDCUBE_ICON = """
<svg fill="none" version="1.1" viewBox="0 0 24 24" xmlns="http://www.w3.org/2000/svg">
<g transform="translate(29.07 -.3244)">
<path d="m-17.02 2.705c-4.01 2e-7 -7.283 3.273-7.283 7.283 0 0.00524-1.1e-5 0.01038 0 0.01562l-1.85 1.068v5.613l9.105 5.26 9.104-5.26v-5.613l-1.797-1.037c1.008e-4 -0.01573 0.00195-0.03112 0.00195-0.04688-1e-7 -4.01-3.271-7.283-7.281-7.283zm0 2.012c2.923 1e-7 5.27 2.349 5.27 5.271 0 2.923-2.347 5.27-5.27 5.27-2.923-1e-6 -5.271-2.347-5.271-5.27 0-2.923 2.349-5.271 5.271-5.271z" fill="#000" fill-rule="evenodd" stroke-linejoin="bevel"/>
</g>
</svg>
"""

42
selfprivacy_api/services/service.py
View file

@ -15,7 +15,12 @@ from selfprivacy_api.utils.block_devices import BlockDevice, BlockDevices
from selfprivacy_api.jobs import Job, Jobs, JobStatus, report_progress
from selfprivacy_api.jobs.upgrade_system import rebuild_system
from selfprivacy_api.models.services import ServiceStatus, ServiceDnsRecord
from selfprivacy_api.models.services import (
License,
ServiceStatus,
ServiceDnsRecord,
SupportLevel,
)
from selfprivacy_api.services.generic_size_counter import get_storage_usage
from selfprivacy_api.services.owned_path import OwnedPath, Bind
from selfprivacy_api.services.moving import (
@ -166,6 +171,37 @@ class Service(ABC):
with ReadUserData() as user_data:
return user_data.get("modules", {}).get(name, {}) != {}
def is_system_service(self) -> bool:
"""
`True` if the service is a system service and should be hidden from the user.
`False` if it is not a system service.
"""
return False
def get_license(self) -> List[License]:
"""
The licenses of the service.
"""
return []
def get_homepage(self) -> Optional[str]:
"""
The homepage of the service.
"""
return None
def get_source_page(self) -> Optional[str]:
"""
The source page of the service.
"""
return None
def get_support_level(self) -> SupportLevel:
"""
The support level of the service.
"""
return SupportLevel.NORMAL
@staticmethod
@abstractmethod
def get_status() -> ServiceStatus:
@ -312,6 +348,10 @@ class Service(ABC):
)
return [owned_folder.path for owned_folder in cls.get_owned_folders()]
@classmethod
def get_folders_to_back_up(cls) -> List[str]:
return cls.get_folders()
@classmethod
def get_owned_folders(cls) -> List[OwnedPath]:
"""

109
selfprivacy_api/services/templated_service.py
View file

@ -1,7 +1,6 @@
"""A Service implementation that loads all needed data from a JSON file"""
import base64
from enum import Enum
import logging
import json
import subprocess
@ -9,12 +8,15 @@ from typing import List, Optional
from os.path import join, exists
from os import mkdir, remove
from pydantic import BaseModel, ConfigDict
from pydantic.alias_generators import to_camel
from selfprivacy_api.backup.postgres import PostgresDumper
from selfprivacy_api.utils.postgres import PostgresDumper
from selfprivacy_api.jobs import Job, JobStatus, Jobs
from selfprivacy_api.models.services import ServiceDnsRecord, ServiceStatus
from selfprivacy_api.models.services import (
License,
ServiceDnsRecord,
ServiceMetaData,
ServiceStatus,
SupportLevel,
)
from selfprivacy_api.services.flake_service_manager import FlakeServiceManager
from selfprivacy_api.services.generic_size_counter import get_storage_usage
from selfprivacy_api.services.owned_path import OwnedPath
@ -36,26 +38,6 @@ SP_SUGGESTED_MODULES_PATH = "/etc/suggested-sp-modules"
logger = logging.getLogger(__name__)
class SupportLevel(Enum):
"""Enum representing the support level of a service."""
NORMAL = "normal"
EXPERIMENTAL = "experimental"
DEPRECATED = "deprecated"
UNKNOWN = "unknown"
@classmethod
def from_str(cls, support_level: str) -> "SupportLevel":
"""Return the SupportLevel from a string."""
if support_level == "normal":
return cls.NORMAL
if support_level == "experimental":
return cls.EXPERIMENTAL
if support_level == "deprecated":
return cls.DEPRECATED
return cls.UNKNOWN
def config_item_from_json(json_data: dict) -> Optional[ServiceConfigItem]:
"""Create a ServiceConfigItem from JSON data."""
weight = json_data.get("meta", {}).get("weight", 50)
@ -103,51 +85,6 @@ def config_item_from_json(json_data: dict) -> Optional[ServiceConfigItem]:
raise ValueError("Unknown config item type")
class BaseSchema(BaseModel):
model_config = ConfigDict(
alias_generator=to_camel,
populate_by_name=True,
from_attributes=True,
)
class License(BaseSchema):
"""Model representing a license."""
deprecated: bool
free: bool
full_name: str
redistributable: bool
short_name: str
spdx_id: str
url: str
class ServiceMetaData(BaseSchema):
"""Model representing the meta data of a service."""
id: str
name: str
description: str = "No description found!"
svg_icon: str = ""
showUrl: bool = True
primary_subdomain: Optional[str] = None
is_movable: bool = False
is_required: bool = False
can_be_backed_up: bool = True
backup_description: str = "No backup description found!"
systemd_services: List[str]
user: Optional[str] = None
group: Optional[str] = None
folders: List[str] = []
owned_folders: List[OwnedPath] = []
postgre_databases: List[str] = []
license: List[License] = []
homepage: Optional[str] = None
source_page: Optional[str] = None
support_level: SupportLevel = SupportLevel.UNKNOWN
class TemplatedService(Service):
"""Class representing a dynamically loaded service."""
@ -279,6 +216,18 @@ class TemplatedService(Service):
with FlakeServiceManager() as service_manager:
return name in service_manager.services
def get_license(self) -> List[License]:
return self.meta.license
def get_homepage(self) -> Optional[str]:
return self.meta.homepage
def get_source_page(self) -> Optional[str]:
return self.meta.source_page
def get_support_level(self) -> SupportLevel:
return self.meta.support_level
def get_status(self) -> ServiceStatus:
if not self.meta.systemd_services:
return ServiceStatus.INACTIVE
@ -443,8 +392,6 @@ class TemplatedService(Service):
resulting_folders = folders.copy()
for folder in owned_folders:
resulting_folders.append(folder.path)
if self.get_postgresql_databases():
resulting_folders.append(self._get_db_dumps_folder())
return resulting_folders
def get_owned_folders(self) -> List[OwnedPath]:
@ -453,14 +400,12 @@ class TemplatedService(Service):
resulting_folders = owned_folders.copy()
for folder in folders:
resulting_folders.append(self.owned_path(folder))
return resulting_folders
def get_folders_to_back_up(self) -> List[str]:
resulting_folders = self.meta.folders.copy()
if self.get_postgresql_databases():
resulting_folders.append(
OwnedPath(
path=self._get_db_dumps_folder(),
owner="selfprivacy-api",
group="selfprivacy-api",
)
)
resulting_folders.append(self._get_db_dumps_folder())
return resulting_folders
def set_location(self, volume: BlockDevice):
@ -504,16 +449,13 @@ class TemplatedService(Service):
)
def pre_backup(self, job: Job):
logger.warning("Pre backup")
if self.get_postgresql_databases():
db_dumps_folder = self._get_db_dumps_folder()
logger.warning("Pre backup: postgresql databases")
# Create folder for the dumps if it does not exist
if not exists(db_dumps_folder):
mkdir(db_dumps_folder)
# Dump the databases
for db_name in self.get_postgresql_databases():
logger.warning(f"Pre backup: db_name: {db_name}")
if job is not None:
Jobs.update(
job,
@ -522,7 +464,6 @@ class TemplatedService(Service):
)
db_dumper = PostgresDumper(db_name)
backup_file = join(db_dumps_folder, f"{db_name}.dump")
logger.warning(f"Pre backup: backup_file: {backup_file}")
db_dumper.backup_database(backup_file)
def _clear_db_dumps(self):

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

@ -95,7 +95,7 @@ class DummyService(Service):
def get_status(cls) -> ServiceStatus:
filepath = cls.status_file()
if filepath in [None, ""]:
raise ValueError(f"We do not have a path for our test dummy status file!")
raise ValueError("We do not have a path for our test dummy status file!")
if not path.exists(filepath):
raise FileNotFoundError(filepath)

0
selfprivacy_api/backup/postgres.py → selfprivacy_api/utils/postgres.py
View file

6
tests/conftest.py
View file

@ -258,7 +258,7 @@ def dummy_service(
ensure_user_exists(user)
# register our service
services.services.append(service)
services.DUMMY_SERVICES.append(service)
huey.immediate = True
assert huey.immediate is True
@ -269,8 +269,8 @@ def dummy_service(
# Cleanup because apparently it matters wrt tasks
# Some tests may remove it from the list intentionally, this is fine
if service in services.services:
services.services.remove(service)
if service in services.DUMMY_SERVICES:
services.DUMMY_SERVICES.remove(service)
def prepare_nixos_rebuild_calls(fp, unit_name):

3
tests/test_backup.py
View file

@ -250,6 +250,7 @@ def test_error_censoring_encryptionkey(dummy_service, backups):
Backups.back_up(dummy_service)
job = get_backup_fail(dummy_service)
assert job is not None
assert_job_errored(job)
job_text = all_job_text(job)
@ -287,6 +288,7 @@ def test_error_censoring_loginkey(dummy_service, backups, fp):
Backups.back_up(dummy_service)
job = get_backup_fail(dummy_service)
assert job is not None
assert_job_errored(job)
job_text = all_job_text(job)
@ -705,6 +707,7 @@ def test_provider_storage(backups):
Storage.store_provider(test_provider)
restored_provider_model = Storage.load_provider()
assert restored_provider_model is not None
assert restored_provider_model.kind == "BACKBLAZE"
assert restored_provider_model.login == test_login
assert restored_provider_model.key == test_key

23
tests/test_graphql/test_services.py
View file

@ -88,20 +88,26 @@ def dummy_service_with_binds(dummy_service, mock_lsblk_devices, volume_folders):
@pytest.fixture()
def only_dummy_service(dummy_service) -> Generator[DummyService, None, None]:
# because queries to services that are not really there error out
back_copy = service_module.services.copy()
service_module.services.clear()
service_module.services.append(dummy_service)
service_module.TEST_FLAGS.clear()
service_module.TEST_FLAGS.append("ONLY_DUMMY_SERVICE")
service_module.DUMMY_SERVICES.clear()
service_module.DUMMY_SERVICES.append(dummy_service)
yield dummy_service
service_module.services.clear()
service_module.services.extend(back_copy)
service_module.TEST_FLAGS.clear()
service_module.DUMMY_SERVICES.clear()
@pytest.fixture
def only_dummy_service_and_api(
only_dummy_service, generic_userdata, dkim_file
generic_userdata, dkim_file, dummy_service
) -> Generator[DummyService, None, None]:
service_module.services.append(ServiceManager())
return only_dummy_service
service_module.TEST_FLAGS.clear()
service_module.TEST_FLAGS.append("DUMMY_SERVICE_AND_API")
service_module.DUMMY_SERVICES.clear()
service_module.DUMMY_SERVICES.append(dummy_service)
yield dummy_service
service_module.TEST_FLAGS.clear()
service_module.DUMMY_SERVICES.clear()
@pytest.fixture()
@ -727,6 +733,7 @@ def test_graphql_move_service(
def test_mailservice_cannot_enable_disable(authorized_client):
mailservice = ServiceManager.get_service_by_id("simple-nixos-mailserver")
assert mailservice is not None
mutation_response = api_enable(authorized_client, mailservice)
data = get_data(mutation_response)["services"]["enableService"]

30
tests/test_services.py
View file

@ -10,8 +10,6 @@ from selfprivacy_api.utils.waitloop import wait_until_true
import selfprivacy_api.services as services_module
from selfprivacy_api.services.bitwarden import Bitwarden
from selfprivacy_api.services.pleroma import Pleroma
from selfprivacy_api.services.mailserver import MailServer
from selfprivacy_api.services.owned_path import OwnedPath
@ -63,22 +61,22 @@ def test_delayed_start_stop(raw_dummy_service):
assert dummy.get_status() == ServiceStatus.ACTIVE
def test_owned_folders_from_not_owned():
assert Bitwarden.get_owned_folders() == [
OwnedPath(
path=folder,
group="vaultwarden",
owner="vaultwarden",
)
for folder in Bitwarden.get_folders()
]
# def test_owned_folders_from_not_owned():
# assert Bitwarden.get_owned_folders() == [
# OwnedPath(
# path=folder,
# group="vaultwarden",
# owner="vaultwarden",
# )
# for folder in Bitwarden.get_folders()
# ]
def test_paths_from_owned_paths():
assert len(Pleroma.get_folders()) == 2
assert Pleroma.get_folders() == [
ownedpath.path for ownedpath in Pleroma.get_owned_folders()
]
# def test_paths_from_owned_paths():
# assert len(Pleroma.get_folders()) == 2
# assert Pleroma.get_folders() == [
# ownedpath.path for ownedpath in Pleroma.get_owned_folders()
# ]
def test_enabling_disabling_reads_json(dummy_service: DummyService):

15
tests/test_services_systemctl.py
View file

@ -1,12 +1,7 @@
import pytest
from selfprivacy_api.services.service import ServiceStatus
from selfprivacy_api.services.bitwarden import Bitwarden
from selfprivacy_api.services.forgejo import Forgejo
from selfprivacy_api.services.mailserver import MailServer
from selfprivacy_api.services.nextcloud import Nextcloud
from selfprivacy_api.services.ocserv import Ocserv
from selfprivacy_api.services.pleroma import Pleroma
def expected_status_call(service_name: str):
@ -76,19 +71,9 @@ def mock_popen_systemctl_service_not_ok(mocker):
def test_systemctl_ok(mock_popen_systemctl_service_ok):
assert MailServer.get_status() == ServiceStatus.ACTIVE
assert Bitwarden.get_status() == ServiceStatus.ACTIVE
assert Forgejo.get_status() == ServiceStatus.ACTIVE
assert Nextcloud.get_status() == ServiceStatus.ACTIVE
assert Ocserv.get_status() == ServiceStatus.ACTIVE
assert Pleroma.get_status() == ServiceStatus.ACTIVE
call_args_asserts(mock_popen_systemctl_service_ok)
def test_systemctl_failed_service(mock_popen_systemctl_service_not_ok):
assert MailServer.get_status() == ServiceStatus.FAILED
assert Bitwarden.get_status() == ServiceStatus.FAILED
assert Forgejo.get_status() == ServiceStatus.FAILED
assert Nextcloud.get_status() == ServiceStatus.FAILED
assert Ocserv.get_status() == ServiceStatus.FAILED
assert Pleroma.get_status() == ServiceStatus.FAILED
call_args_asserts(mock_popen_systemctl_service_not_ok)