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

Merge branch 'master' into def/add_users_repositories

Browse source
This commit is contained in:
dettlaff 2025-01-14 18:23:59 +04:00
parent 4e1d9076e2 043d280d53
commit ffe00bce51
48 changed files with 1135 additions and 1083 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
nixos/module.nix
View file

@ -5,6 +5,19 @@ let
config-id = "default";
nixos-rebuild = "${config.system.build.nixos-rebuild}/bin/nixos-rebuild";
nix = "${config.nix.package.out}/bin/nix";
sp-fetch-remote-module = pkgs.writeShellApplication {
name = "sp-fetch-remote-module";
runtimeInputs = [ config.nix.package.out ];
text = ''
if [ "$#" -ne 1 ]; then
echo "Usage: $0 <URL>"
exit 1
fi
URL="$1"
nix eval --file /etc/sp-fetch-remote-module.nix --raw --apply "f: f { flakeURL = \"$URL\"; }"
'';
};
in
{
options.services.selfprivacy-api = {
@ -46,12 +59,12 @@ in
pkgs.util-linux
pkgs.e2fsprogs
pkgs.iproute2
pkgs.kanidm
];
after = [ "network-online.target" ];
wants = [ "network-online.target" ];
wantedBy = [ "multi-user.target" ];
serviceConfig = {
# Do not forget to edit Postgres identMap if you change the user!
User = "root";
ExecStart = "${selfprivacy-graphql-api}/bin/app.py";
Restart = "always";
@ -82,12 +95,15 @@ in
pkgs.util-linux
pkgs.e2fsprogs
pkgs.iproute2
pkgs.postgresql_16.out
sp-fetch-remote-module
pkgs.kanidm
];
after = [ "network-online.target" ];
wants = [ "network-online.target" ];
wantedBy = [ "multi-user.target" ];
serviceConfig = {
# Do not forget to edit Postgres identMap if you change the user!
User = "root";
ExecStart = "${pkgs.python312Packages.huey}/bin/huey_consumer.py selfprivacy_api.task_registry.huey";
Restart = "always";

16
selfprivacy_api/backup/__init__.py
View file

@ -246,9 +246,10 @@ 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()
service.pre_backup(job=job)
Jobs.update(job, status=JobStatus.RUNNING, status_text="Uploading backup")
snapshot = Backups.provider().backupper.start_backup(
folders,
service_name,
@ -258,12 +259,12 @@ class Backups:
Backups._on_new_snapshot_created(service_name, snapshot)
if reason == BackupReason.AUTO:
Backups._prune_auto_snaps(service)
service.post_backup()
service.post_backup(job=job)
except Exception as error:
Jobs.update(job, status=JobStatus.ERROR, error=str(error))
raise error
Jobs.update(job, status=JobStatus.FINISHED)
Jobs.update(job, status=JobStatus.FINISHED, result="Backup finished")
if reason in [BackupReason.AUTO, BackupReason.PRE_RESTORE]:
Jobs.set_expiration(job, AUTOBACKUP_JOB_EXPIRATION_SECONDS)
return Backups.sync_date_from_cache(snapshot)
@ -452,6 +453,7 @@ class Backups:
with StoppedService(service):
if not service.is_always_active():
Backups.assert_dead(service)
service.pre_restore(job=job)
if strategy == RestoreStrategy.INPLACE:
Backups._inplace_restore(service, snapshot, job)
else: # verify_before_download is our default
@ -464,7 +466,7 @@ class Backups:
service, snapshot.id, verify=True
)
service.post_restore()
service.post_restore(job=job)
Jobs.update(
job,
status=JobStatus.RUNNING,
@ -515,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,
@ -715,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())

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

@ -4,12 +4,13 @@ import subprocess
import json
import datetime
import tempfile
import logging
import os
from typing import List, Optional, TypeVar, Callable
from collections.abc import Iterable
from json.decoder import JSONDecodeError
from os.path import exists, join
from os import mkdir
from os.path import exists, join, isfile, islink, isdir
from shutil import rmtree
from selfprivacy_api.utils.waitloop import wait_until_success
@ -28,6 +29,8 @@ FILESYSTEM_TIMEOUT_SEC = 60
T = TypeVar("T", bound=Callable)
logger = logging.getLogger(__name__)
def unlocked_repo(func: T) -> T:
"""unlock repo and retry if it appears to be locked"""
@ -363,7 +366,27 @@ class ResticBackupper(AbstractBackupper):
parsed_output = ResticBackupper.parse_json_output(output)
return parsed_output["total_size"]
except ValueError as error:
raise ValueError("cannot restore a snapshot: " + output) from error
raise ValueError("Cannot restore a snapshot: " + output) from error
def _rm_all_folder_contents(self, folder: str) -> None:
"""
Remove all contents of a folder, including subfolders.
Raises:
ValueError: If it encounters an error while removing contents.
"""
try:
for filename in os.listdir(folder):
path = join(folder, filename)
try:
if isfile(path) or islink(path):
os.unlink(path)
elif isdir(path):
rmtree(path)
except Exception as error:
raise ValueError("Cannot remove folder contents: ", path) from error
except OSError as error:
raise ValueError("Cannot access folder: ", folder) from error
@unlocked_repo
def restore_from_backup(
@ -376,7 +399,7 @@ class ResticBackupper(AbstractBackupper):
Restore from backup with restic
"""
if folders is None or folders == []:
raise ValueError("cannot restore without knowing where to!")
raise ValueError("Cannot restore without knowing where to!")
with tempfile.TemporaryDirectory() as temp_dir:
if verify:
@ -394,9 +417,9 @@ class ResticBackupper(AbstractBackupper):
else: # attempting inplace restore
for folder in folders:
wait_until_success(
lambda: rmtree(folder), timeout_sec=FILESYSTEM_TIMEOUT_SEC
lambda: self._rm_all_folder_contents(folder),
timeout_sec=FILESYSTEM_TIMEOUT_SEC,
)
mkdir(folder)
self._raw_verified_restore(snapshot_id, target="/")
return

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 "3.4.0"
return "3.5.0"

40
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]]:
@ -207,7 +235,10 @@ class Service:
if not config_items:
return None
# By the "type" field convert every dict into a ConfigItem. In the future there will be more types.
return [config_item_to_graphql(config_items[item]) for item in config_items]
unsorted_config_items = [config_items[item] for item in config_items]
# Sort the items by their weight. If there is no weight, implicitly set it to 50.
config_items = sorted(unsorted_config_items, key=lambda x: x.get("weight", 50))
return [config_item_to_graphql(item) for item in config_items]
# TODO: fill this
@strawberry.field
@ -238,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,
)

7
selfprivacy_api/graphql/queries/services.py
View file

@ -16,5 +16,8 @@ from selfprivacy_api.services import ServiceManager
class Services:
@strawberry.field
def all_services(self) -> typing.List[Service]:
services = ServiceManager.get_all_services()
return [service_to_graphql_service(service) for service in services]
services = [
service_to_graphql_service(service)
for service in ServiceManager.get_all_services()
]
return sorted(services, key=lambda service: service.display_name)

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

152
selfprivacy_api/services/__init__.py
View file

@ -3,30 +3,34 @@
import logging
import base64
import typing
import subprocess
import json
from typing import List
from os import path
from os import makedirs
from os.path import join
from functools import lru_cache
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.jobs import Job, JobStatus, Jobs
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
from selfprivacy_api.utils.cached_call import get_ttl_hash
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 (
SP_MODULES_DEFENITIONS_PATH,
SP_SUGGESTED_MODULES_PATH,
TemplatedService,
)
CONFIG_STASH_DIR = "/etc/selfprivacy/dump"
KANIDM_A_RECORD = "auth"
@ -39,27 +43,33 @@ class ServiceManager(Service):
@staticmethod
def get_all_services() -> list[Service]:
return services
return get_services()
@staticmethod
def get_service_by_id(service_id: str) -> typing.Optional[Service]:
for service in services:
for service in get_services():
if service.get_id() == service_id:
return service
return None
@staticmethod
def get_enabled_services() -> list[Service]:
return [service for service in services if service.is_enabled()]
return [service for service in get_services() if service.is_enabled()]
# This one is not currently used by any code.
@staticmethod
def get_disabled_services() -> list[Service]:
return [service for service in services if not service.is_enabled()]
return [service for service in get_services() if not service.is_enabled()]
@staticmethod
def get_services_by_location(location: str) -> list[Service]:
return [service for service in services if service.get_drive() == location]
return [
service
for service in get_services(
exclude_remote=True,
)
if service.get_drive() == location
]
@staticmethod
def get_all_required_dns_records() -> list[ServiceDnsRecord]:
@ -76,7 +86,7 @@ class ServiceManager(Service):
),
]
# TODO: Reenable with 3.5.0 release when clients are ready.
# TODO: Reenable with 3.6.0 release when clients are ready.
# Do not forget about tests!
# try:
# dns_records.append(
@ -141,6 +151,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."
@ -159,7 +173,9 @@ class ServiceManager(Service):
# For now we will just copy settings EXCEPT the locations of services
# Stash locations as they are set by user right now
locations = {}
for service in services:
for service in get_services(
exclude_remote=True,
):
if service.is_movable():
locations[service.get_id()] = service.get_drive()
@ -167,8 +183,10 @@ class ServiceManager(Service):
for p in [USERDATA_FILE, SECRETS_FILE, DKIM_DIR]:
cls.retrieve_stashed_path(p)
# Pop locations
for service in services:
# Pop location
for service in get_services(
exclude_remote=True,
):
if service.is_movable():
device = BlockDevices().get_block_device(locations[service.get_id()])
if device is not None:
@ -195,11 +213,6 @@ class ServiceManager(Service):
"""
pass
@staticmethod
def get_logs():
# TODO: maybe return the logs for api itself
return ""
@classmethod
def get_drive(cls) -> str:
return BlockDevices().get_root_block_device().name
@ -234,7 +247,12 @@ class ServiceManager(Service):
copyfile(cls.stash_for(p), p)
@classmethod
def pre_backup(cls):
def pre_backup(cls, job: Job):
Jobs.update(
job,
status_text="Stashing settings",
status=JobStatus.RUNNING,
)
tempdir = cls.dump_dir()
rmtree(join(tempdir), ignore_errors=True)
makedirs(tempdir)
@ -243,7 +261,7 @@ class ServiceManager(Service):
cls.stash_a_path(p)
@classmethod
def post_backup(cls):
def post_backup(cls, job: Job):
rmtree(cls.dump_dir(), ignore_errors=True)
@classmethod
@ -254,20 +272,80 @@ class ServiceManager(Service):
return cls.folders[0]
@classmethod
def post_restore(cls):
def post_restore(cls, job: Job):
cls.merge_settings()
rmtree(cls.dump_dir(), ignore_errors=True)
services: list[Service] = [
Bitwarden(),
Forgejo(),
MailServer(),
Nextcloud(),
Pleroma(),
Ocserv(),
JitsiMeet(),
Roundcube(),
ServiceManager(),
Prometheus(),
]
# @redis_cached_call(ttl=30)
@lru_cache()
def get_templated_service(service_id: str, ttl_hash=None) -> TemplatedService:
del ttl_hash
return TemplatedService(service_id)
# @redis_cached_call(ttl=3600)
@lru_cache()
def get_remote_service(id: str, url: str, ttl_hash=None) -> TemplatedService:
del ttl_hash
response = subprocess.run(
["sp-fetch-remote-module", url],
capture_output=True,
text=True,
check=True,
)
return TemplatedService(id, response.stdout)
DUMMY_SERVICES = []
TEST_FLAGS: list[str] = []
def get_services(exclude_remote=False) -> 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] = [
MailServer(),
ServiceManager(),
Prometheus(),
]
if DUMMY_SERVICES:
hardcoded_services += DUMMY_SERVICES
service_ids = [service.get_id() for service in hardcoded_services]
templated_services: List[Service] = []
if path.exists(SP_MODULES_DEFENITIONS_PATH):
for module in listdir(SP_MODULES_DEFENITIONS_PATH):
if module in service_ids:
continue
try:
templated_services.append(
get_templated_service(module, ttl_hash=get_ttl_hash(30))
)
service_ids.append(module)
except Exception as e:
logger.error(f"Failed to load service {module}: {e}")
if not exclude_remote and path.exists(SP_SUGGESTED_MODULES_PATH):
# It is a file with a JSON array
with open(SP_SUGGESTED_MODULES_PATH) as f:
suggested_modules = json.load(f)
for module in suggested_modules:
if module in service_ids:
continue
try:
templated_services.append(
get_remote_service(
module,
f"git+https://git.selfprivacy.org/SelfPrivacy/selfprivacy-nixos-config.git?ref=flakes&dir=sp-modules/{module}",
ttl_hash=get_ttl_hash(3600),
)
)
service_ids.append(module)
except Exception as e:
logger.error(f"Failed to load service {module}: {e}")
return hardcoded_services + templated_services

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

@ -1,113 +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_logs():
return ""
@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>
"""

34
selfprivacy_api/services/config_item.py
View file

@ -14,6 +14,7 @@ class ServiceConfigItem(ABC):
description: str
widget: str
type: str
weight: int
@abstractmethod
def get_value(self, service_id):
@ -27,13 +28,14 @@ class ServiceConfigItem(ABC):
def validate_value(self, value):
return True
def as_dict(self, service_options):
def as_dict(self, service_id: str):
return {
"id": self.id,
"type": self.type,
"description": self.description,
"widget": self.widget,
"value": self.get_value(service_options),
"value": self.get_value(service_id),
"weight": self.weight,
}
@ -46,6 +48,7 @@ class StringServiceConfigItem(ServiceConfigItem):
regex: Optional[str] = None,
widget: Optional[str] = None,
allow_empty: bool = False,
weight: int = 50,
):
if widget == "subdomain" and not regex:
raise ValueError("Subdomain widget requires regex")
@ -56,6 +59,7 @@ class StringServiceConfigItem(ServiceConfigItem):
self.regex = re.compile(regex) if regex else None
self.widget = widget if widget else "text"
self.allow_empty = allow_empty
self.weight = weight
def get_value(self, service_id):
with ReadUserData() as user_data:
@ -73,15 +77,16 @@ class StringServiceConfigItem(ServiceConfigItem):
user_data["modules"][service_id] = {}
user_data["modules"][service_id][self.id] = value
def as_dict(self, service_options):
def as_dict(self, service_id):
return {
"id": self.id,
"type": self.type,
"description": self.description,
"widget": self.widget,
"value": self.get_value(service_options),
"value": self.get_value(service_id),
"default_value": self.default_value,
"regex": self.regex.pattern if self.regex else None,
"weight": self.weight,
}
def validate_value(self, value):
@ -104,12 +109,14 @@ class BoolServiceConfigItem(ServiceConfigItem):
default_value: bool,
description: str,
widget: Optional[str] = None,
weight: int = 50,
):
self.id = id
self.type = "bool"
self.default_value = default_value
self.description = description
self.widget = widget if widget else "switch"
self.weight = weight
def get_value(self, service_id):
with ReadUserData() as user_data:
@ -127,14 +134,15 @@ class BoolServiceConfigItem(ServiceConfigItem):
user_data["modules"][service_id] = {}
user_data["modules"][service_id][self.id] = value
def as_dict(self, service_options):
def as_dict(self, service_id):
return {
"id": self.id,
"type": self.type,
"description": self.description,
"widget": self.widget,
"value": self.get_value(service_options),
"value": self.get_value(service_id),
"default_value": self.default_value,
"weight": self.weight,
}
def validate_value(self, value):
@ -149,6 +157,7 @@ class EnumServiceConfigItem(ServiceConfigItem):
description: str,
options: list[str],
widget: Optional[str] = None,
weight: int = 50,
):
self.id = id
self.type = "enum"
@ -156,6 +165,7 @@ class EnumServiceConfigItem(ServiceConfigItem):
self.description = description
self.options = options
self.widget = widget if widget else "select"
self.weight = weight
def get_value(self, service_id):
with ReadUserData() as user_data:
@ -173,15 +183,16 @@ class EnumServiceConfigItem(ServiceConfigItem):
user_data["modules"][service_id] = {}
user_data["modules"][service_id][self.id] = value
def as_dict(self, service_options):
def as_dict(self, service_id):
return {
"id": self.id,
"type": self.type,
"description": self.description,
"widget": self.widget,
"value": self.get_value(service_options),
"value": self.get_value(service_id),
"default_value": self.default_value,
"options": self.options,
"weight": self.weight,
}
def validate_value(self, value):
@ -200,6 +211,7 @@ class IntServiceConfigItem(ServiceConfigItem):
widget: Optional[str] = None,
min_value: Optional[int] = None,
max_value: Optional[int] = None,
weight: int = 50,
) -> None:
self.id = id
self.type = "int"
@ -208,6 +220,7 @@ class IntServiceConfigItem(ServiceConfigItem):
self.widget = widget if widget else "number"
self.min_value = min_value
self.max_value = max_value
self.weight = weight
def get_value(self, service_id):
with ReadUserData() as user_data:
@ -225,16 +238,17 @@ class IntServiceConfigItem(ServiceConfigItem):
user_data["modules"][service_id] = {}
user_data["modules"][service_id][self.id] = value
def as_dict(self, service_options):
def as_dict(self, service_id):
return {
"id": self.id,
"type": self.type,
"description": self.description,
"widget": self.widget,
"value": self.get_value(service_options),
"value": self.get_value(service_id),
"default_value": self.default_value,
"min_value": self.min_value,
"max_value": self.max_value,
"weight": self.weight,
}
def validate_value(self, value):

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

@ -1,138 +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_logs():
return ""
@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>
"""

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

@ -1,113 +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_logs():
return ""
@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>
"""

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

@ -89,10 +89,6 @@ class MailServer(Service):
subprocess.run(["systemctl", "restart", "dovecot2.service"], check=False)
subprocess.run(["systemctl", "restart", "postfix.service"], check=False)
@staticmethod
def get_logs():
return ""
@staticmethod
def get_folders() -> List[str]:
return ["/var/vmail", "/var/sieve"]

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

@ -1,104 +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_logs():
"""Return Nextcloud logs."""
return ""
@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

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

@ -1,86 +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_logs():
return ""
@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

4
selfprivacy_api/services/owned_path.py
View file

@ -105,8 +105,8 @@ class Bind:
["umount", self.binding_path],
check=True,
)
except subprocess.CalledProcessError:
raise BindError(f"Unable to unmount folder {self.binding_path}.")
except subprocess.CalledProcessError as error:
raise BindError(f"Unable to unmount folder {self.binding_path}. {error}")
pass
def ensure_ownership(self) -> None:

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

@ -1,93 +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_logs():
return ""
@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

8
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
@ -71,10 +75,6 @@ class Prometheus(Service):
def restart():
subprocess.run(["systemctl", "restart", "prometheus.service"])
@staticmethod
def get_logs():
return ""
@staticmethod
def get_owned_folders() -> List[OwnedPath]:
return [

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

@ -1,104 +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_logs():
return ""
@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>
"""

62
selfprivacy_api/services/service.py
View file

@ -1,6 +1,7 @@
"""Abstract class for a service running on a server"""
from abc import ABC, abstractmethod
import logging
from typing import List, Optional
from os.path import exists
@ -14,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 (
@ -30,6 +36,8 @@ from selfprivacy_api.services.moving import (
DEFAULT_START_STOP_TIMEOUT = 5 * 60
logger = logging.getLogger(__name__)
class Service(ABC):
"""
@ -163,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:
@ -226,11 +265,6 @@ class Service(ABC):
cls.get_id(),
)
@staticmethod
@abstractmethod
def get_logs():
pass
@classmethod
def get_storage_usage(cls) -> int:
"""
@ -314,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]:
"""
@ -330,6 +368,9 @@ class Service(ABC):
def get_foldername(path: str) -> str:
return path.split("/")[-1]
def get_postgresql_databases(self) -> List[str]:
return []
# TODO: with better json utils, it can be one line, and not a separate function
@classmethod
def set_location(cls, volume: BlockDevice):
@ -477,13 +518,16 @@ class Service(ABC):
group=group,
)
def pre_backup(self):
def pre_backup(self, job: Job):
pass
def post_backup(self):
def post_backup(self, job: Job):
pass
def post_restore(self):
def pre_restore(self, job: Job):
pass
def post_restore(self, job: Job):
pass

514
selfprivacy_api/services/templated_service.py Normal file
View file

@ -0,0 +1,514 @@
"""A Service implementation that loads all needed data from a JSON file"""
import base64
import logging
import json
import subprocess
from typing import List, Optional
from os.path import join, exists
from os import mkdir, remove
from selfprivacy_api.utils.postgres import PostgresDumper
from selfprivacy_api.jobs import Job, JobStatus, Jobs
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
from selfprivacy_api.services.service import Service
from selfprivacy_api.utils import ReadUserData, WriteUserData, get_domain
from selfprivacy_api.services.config_item import (
ServiceConfigItem,
StringServiceConfigItem,
BoolServiceConfigItem,
EnumServiceConfigItem,
IntServiceConfigItem,
)
from selfprivacy_api.utils.block_devices import BlockDevice, BlockDevices
from selfprivacy_api.utils.systemd import get_service_status_from_several_units
SP_MODULES_DEFENITIONS_PATH = "/etc/sp-modules"
SP_SUGGESTED_MODULES_PATH = "/etc/suggested-sp-modules"
logger = logging.getLogger(__name__)
def config_item_from_json(json_data: dict) -> Optional[ServiceConfigItem]:
"""Create a ServiceConfigItem from JSON data."""
weight = json_data.get("meta", {}).get("weight", 50)
if json_data["meta"]["type"] == "enable":
return None
if json_data["meta"]["type"] == "location":
return None
if json_data["meta"]["type"] == "string":
return StringServiceConfigItem(
id=json_data["name"],
default_value=json_data["default"],
description=json_data["description"],
regex=json_data["meta"].get("regex"),
widget=json_data["meta"].get("widget"),
allow_empty=json_data["meta"].get("allowEmpty", False),
weight=weight,
)
if json_data["meta"]["type"] == "bool":
return BoolServiceConfigItem(
id=json_data["name"],
default_value=json_data["default"],
description=json_data["description"],
widget=json_data["meta"].get("widget"),
weight=weight,
)
if json_data["meta"]["type"] == "enum":
return EnumServiceConfigItem(
id=json_data["name"],
default_value=json_data["default"],
description=json_data["description"],
options=json_data["meta"]["options"],
widget=json_data["meta"].get("widget"),
weight=weight,
)
if json_data["meta"]["type"] == "int":
return IntServiceConfigItem(
id=json_data["name"],
default_value=json_data["default"],
description=json_data["description"],
widget=json_data["meta"].get("widget"),
min_value=json_data["meta"].get("minValue"),
max_value=json_data["meta"].get("maxValue"),
weight=weight,
)
raise ValueError("Unknown config item type")
class TemplatedService(Service):
"""Class representing a dynamically loaded service."""
def __init__(self, service_id: str, source_data: Optional[str] = None) -> None:
if source_data:
self.definition_data = json.loads(source_data)
else:
# Check if the service exists
if not exists(join(SP_MODULES_DEFENITIONS_PATH, service_id)):
raise FileNotFoundError(f"Service {service_id} not found")
# Load the service
with open(join(SP_MODULES_DEFENITIONS_PATH, service_id)) as file:
self.definition_data = json.load(file)
# Check if required fields are present
if "meta" not in self.definition_data:
raise ValueError("meta not found in service definition")
if "options" not in self.definition_data:
raise ValueError("options not found in service definition")
# Load the meta data
self.meta = ServiceMetaData(**self.definition_data["meta"])
# Load the options
self.options = self.definition_data["options"]
# Load the config items
self.config_items = {}
for option in self.options.values():
config_item = config_item_from_json(option)
if config_item:
self.config_items[config_item.id] = config_item
# If it is movable, check for the location option
if self.meta.is_movable and "location" not in self.options:
raise ValueError("Service is movable but does not have a location option")
# Load all subdomains via options with "subdomain" widget
self.subdomain_options: List[str] = []
for option in self.options.values():
if option.get("meta", {}).get("widget") == "subdomain":
self.subdomain_options.append(option["name"])
def get_id(self) -> str:
# Check if ID contains elements that might be a part of the path
if "/" in self.meta.id or "\\" in self.meta.id:
raise ValueError("Invalid ID")
return self.meta.id
def get_display_name(self) -> str:
return self.meta.name
def get_description(self) -> str:
return self.meta.description
def get_svg_icon(self) -> str:
return base64.b64encode(self.meta.svg_icon.encode("utf-8")).decode("utf-8")
def get_subdomain(self) -> Optional[str]:
# If there are no subdomain options, return None
if not self.subdomain_options:
return None
# If primary_subdomain is set, try to find it in the options
if (
self.meta.primary_subdomain
and self.meta.primary_subdomain in self.subdomain_options
):
option_name = self.meta.primary_subdomain
# Otherwise, use the first subdomain option
else:
option_name = self.subdomain_options[0]
# Now, read the value from the userdata
name = self.get_id()
with ReadUserData() as user_data:
if "modules" in user_data:
if name in user_data["modules"]:
if option_name in user_data["modules"][name]:
return user_data["modules"][name][option_name]
# Otherwise, return default value for the option
return self.options[option_name].get("default")
def get_subdomains(self) -> List[str]:
# Return a current subdomain for every subdomain option
subdomains = []
with ReadUserData() as user_data:
for option in self.subdomain_options:
if "modules" in user_data:
if self.get_id() in user_data["modules"]:
if option in user_data["modules"][self.get_id()]:
subdomains.append(
user_data["modules"][self.get_id()][option]
)
continue
subdomains.append(self.options[option]["default"])
return subdomains
def get_url(self) -> Optional[str]:
if not self.meta.showUrl:
return None
subdomain = self.get_subdomain()
if not subdomain:
return None
return f"https://{subdomain}.{get_domain()}"
def get_user(self) -> Optional[str]:
if not self.meta.user:
return self.get_id()
return self.meta.user
def get_group(self) -> Optional[str]:
if not self.meta.group:
return self.get_user()
return self.meta.group
def is_movable(self) -> bool:
return self.meta.is_movable
def is_required(self) -> bool:
return self.meta.is_required
def can_be_backed_up(self) -> bool:
return self.meta.can_be_backed_up
def get_backup_description(self) -> str:
return self.meta.backup_description
def is_enabled(self) -> bool:
name = self.get_id()
with ReadUserData() as user_data:
return user_data.get("modules", {}).get(name, {}).get("enable", False)
def is_installed(self) -> bool:
name = self.get_id()
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
return get_service_status_from_several_units(self.meta.systemd_services)
def _set_enable(self, enable: bool):
name = self.get_id()
with WriteUserData() as user_data:
if "modules" not in user_data:
user_data["modules"] = {}
if name not in user_data["modules"]:
user_data["modules"][name] = {}
user_data["modules"][name]["enable"] = enable
def enable(self):
"""Enable the service. Usually this means enabling systemd unit."""
name = self.get_id()
if not self.is_installed():
# First, double-check that it is a suggested module
if exists(SP_SUGGESTED_MODULES_PATH):
with open(SP_SUGGESTED_MODULES_PATH) as file:
suggested_modules = json.load(file)
if name not in suggested_modules:
raise ValueError("Service is not a suggested module")
else:
raise FileNotFoundError("Suggested modules file not found")
with FlakeServiceManager() as service_manager:
service_manager.services[name] = (
f"git+https://git.selfprivacy.org/SelfPrivacy/selfprivacy-nixos-config.git?ref=flakes&dir=sp-modules/{name}"
)
if "location" in self.options:
with WriteUserData() as user_data:
if "modules" not in user_data:
user_data["modules"] = {}
if name not in user_data["modules"]:
user_data["modules"][name] = {}
if "location" not in user_data["modules"][name]:
user_data["modules"][name]["location"] = (
BlockDevices().get_root_block_device().name
)
self._set_enable(True)
def disable(self):
"""Disable the service. Usually this means disabling systemd unit."""
self._set_enable(False)
def start(self):
"""Start the systemd units"""
for unit in self.meta.systemd_services:
subprocess.run(["systemctl", "start", unit], check=False)
def stop(self):
"""Stop the systemd units"""
for unit in self.meta.systemd_services:
subprocess.run(["systemctl", "stop", unit], check=False)
def restart(self):
"""Restart the systemd units"""
for unit in self.meta.systemd_services:
subprocess.run(["systemctl", "restart", unit], check=False)
def get_configuration(self) -> dict:
# If there are no options, return an empty dict
if not self.config_items:
return {}
return {
key: self.config_items[key].as_dict(self.get_id())
for key in self.config_items
}
def set_configuration(self, config_items):
for key, value in config_items.items():
if key not in self.config_items:
raise ValueError(f"Key {key} is not valid for {self.get_id()}")
if self.config_items[key].validate_value(value) is False:
raise ValueError(f"Value {value} is not valid for {key}")
for key, value in config_items.items():
self.config_items[key].set_value(
value,
self.get_id(),
)
def get_storage_usage(self) -> int:
"""
Calculate the real storage usage of folders occupied by service
Calculate using pathlib.
Do not follow symlinks.
"""
storage_used = 0
for folder in self.get_folders():
storage_used += get_storage_usage(folder)
return storage_used
def has_folders(self) -> int:
"""
If there are no folders on disk, moving is noop
"""
for folder in self.get_folders():
if exists(folder):
return True
return False
def get_dns_records(self, ip4: str, ip6: Optional[str]) -> List[ServiceDnsRecord]:
display_name = self.get_display_name()
subdomains = self.get_subdomains()
# Generate records for every subdomain
records: List[ServiceDnsRecord] = []
for subdomain in subdomains:
if not subdomain:
continue
records.append(
ServiceDnsRecord(
type="A",
name=subdomain,
content=ip4,
ttl=3600,
display_name=display_name,
)
)
if ip6:
records.append(
ServiceDnsRecord(
type="AAAA",
name=subdomain,
content=ip6,
ttl=3600,
display_name=display_name,
)
)
return records
def get_drive(self) -> 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 self.is_movable():
return root_device
with ReadUserData() as userdata:
if userdata.get("useBinds", False):
return (
userdata.get("modules", {})
.get(self.get_id(), {})
.get(
"location",
root_device,
)
)
else:
return root_device
def _get_db_dumps_folder(self) -> str:
# Get the drive where the service is located and append the folder name
return join("/var/lib/postgresql-dumps", self.get_id())
def get_folders(self) -> List[str]:
folders = self.meta.folders
owned_folders = self.meta.owned_folders
# Include the contents of folders list
resulting_folders = folders.copy()
for folder in owned_folders:
resulting_folders.append(folder.path)
return resulting_folders
def get_owned_folders(self) -> List[OwnedPath]:
folders = self.meta.folders
owned_folders = self.meta.owned_folders
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(self._get_db_dumps_folder())
return resulting_folders
def set_location(self, volume: BlockDevice):
"""
Only changes userdata
"""
service_id = self.get_id()
with WriteUserData() as user_data:
if "modules" not in user_data:
user_data["modules"] = {}
if service_id not in user_data["modules"]:
user_data["modules"][service_id] = {}
user_data["modules"][service_id]["location"] = volume.name
def get_postgresql_databases(self) -> List[str]:
return self.meta.postgre_databases
def owned_path(self, path: str):
"""Default folder ownership"""
service_name = self.get_display_name()
try:
owner = self.get_user()
if owner is None:
# TODO: assume root?
# (if we do not want to do assumptions, maybe not declare user optional?)
raise LookupError(f"no user for service: {service_name}")
group = self.get_group()
if group is None:
raise LookupError(f"no group for service: {service_name}")
except Exception as error:
raise LookupError(
f"when deciding a bind for folder {path} of service {service_name}, error: {str(error)}"
)
return OwnedPath(
path=path,
owner=owner,
group=group,
)
def pre_backup(self, job: Job):
if self.get_postgresql_databases():
db_dumps_folder = self._get_db_dumps_folder()
# 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():
Jobs.update(
job,
status_text=f"Creating a dump of database {db_name}",
status=JobStatus.RUNNING,
)
db_dumper = PostgresDumper(db_name)
backup_file = join(db_dumps_folder, f"{db_name}.dump")
db_dumper.backup_database(backup_file)
def _clear_db_dumps(self):
db_dumps_folder = self._get_db_dumps_folder()
for db_name in self.get_postgresql_databases():
backup_file = join(db_dumps_folder, f"{db_name}.dump")
if exists(backup_file):
remove(backup_file)
unpacked_file = backup_file.replace(".gz", "")
if exists(unpacked_file):
remove(unpacked_file)
def post_backup(self, job: Job):
if self.get_postgresql_databases():
db_dumps_folder = self._get_db_dumps_folder()
# Remove the backup files
for db_name in self.get_postgresql_databases():
backup_file = join(db_dumps_folder, f"{db_name}.dump")
if exists(backup_file):
remove(backup_file)
def pre_restore(self, job: Job):
if self.get_postgresql_databases():
# Create folder for the dumps if it does not exist
db_dumps_folder = self._get_db_dumps_folder()
if not exists(db_dumps_folder):
mkdir(db_dumps_folder)
# Remove existing dumps if they exist
self._clear_db_dumps()
def post_restore(self, job: Job):
if self.get_postgresql_databases():
# Recover the databases
db_dumps_folder = self._get_db_dumps_folder()
for db_name in self.get_postgresql_databases():
if exists(join(db_dumps_folder, f"{db_name}.dump")):
Jobs.update(
job,
status_text=f"Restoring database {db_name}",
status=JobStatus.RUNNING,
)
db_dumper = PostgresDumper(db_name)
backup_file = join(db_dumps_folder, f"{db_name}.dump")
db_dumper.restore_database(backup_file)
else:
logger.error(f"Database dump for {db_name} not found")
raise FileNotFoundError(f"Database dump for {db_name} not found")
# Remove the dumps
self._clear_db_dumps()

6
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)
@ -195,10 +195,6 @@ class DummyService(Service):
def set_configuration(cls, config_items):
return super().set_configuration(config_items)
@staticmethod
def get_logs():
return ""
@staticmethod
def get_storage_usage() -> int:
storage_usage = 0

6
selfprivacy_api/utils/cached_call.py Normal file
View file

@ -0,0 +1,6 @@
import time
def get_ttl_hash(seconds=3600):
"""Return the same value withing `seconds` time period"""
return round(time.time() / seconds)

34
selfprivacy_api/utils/postgres.py Normal file
View file

@ -0,0 +1,34 @@
import subprocess
class PostgresDumper:
"""--dbname=postgresql://postgres@%2Frun%2Fpostgresql/pleroma"""
def __init__(self, db_name: str):
self.db_name = db_name
self.user = "postgres"
self.socket_dir = r"%2Frun%2Fpostgresql"
def backup_database(self, backup_file: str):
# Create the database dump in custom format
dump_command = [
"pg_dump",
f"--dbname=postgresql://{self.user}@{self.socket_dir}/{self.db_name}",
"--format=custom",
f"--file={backup_file}",
]
subprocess.run(dump_command, check=True)
return backup_file
def restore_database(self, backup_file: str):
restore_command = [
"pg_restore",
f"--dbname=postgresql://{self.user}@{self.socket_dir}",
"--clean",
"--create",
"--exit-on-error",
backup_file,
]
subprocess.run(restore_command, check=True)

7
selfprivacy_api/utils/redis_pool.py
View file

@ -29,6 +29,7 @@ class RedisPool:
url,
decode_responses=True,
)
self._raw_pool = redis.ConnectionPool.from_url(url)
@staticmethod
def connection_url(dbnumber: int) -> str:
@ -44,6 +45,12 @@ class RedisPool:
"""
return redis.Redis(connection_pool=self._pool)
def get_raw_connection(self):
"""
Get a raw connection from the pool.
"""
return redis.Redis(connection_pool=self._raw_pool)
def get_connection_async(self) -> redis_async.Redis:
"""
Get an async connection from the pool.

2
setup.py
View file

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

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):

4
tests/test_autobackup.py
View file

@ -177,7 +177,9 @@ def test_services_to_autobackup(backups, dummy_service):
Backups.set_autobackup_period_minutes(backup_period)
services = Backups.services_to_back_up(now)
assert set(services) == set(backuppable_services())
assert set(service.get_id() for service in services) == set(
service.get_id() for service in backuppable_services()
)
assert dummy_service.get_id() in [
service.get_id() for service in backuppable_services()
]

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

2
tests/test_graphql/test_api_backup.py
View file

@ -434,7 +434,7 @@ def test_snapshots_orphaned_service(authorized_client, dummy_service, backups):
snaps = api_snapshots(authorized_client)
assert len(snaps) == 1
all_services.services.remove(dummy_service)
all_services.DUMMY_SERVICES.remove(dummy_service)
assert ServiceManager.get_service_by_id(dummy_service.get_id()) is None
snaps = api_snapshots(authorized_client)

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"]

192
tests/test_graphql/test_system.py
View file

@ -337,30 +337,30 @@ def test_graphql_get_domain(
assert is_dns_record_in_array(
dns_records, dns_record(name="api", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="cloud"))
assert is_dns_record_in_array(
dns_records, dns_record(name="cloud", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="git"))
assert is_dns_record_in_array(
dns_records, dns_record(name="git", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="meet"))
assert is_dns_record_in_array(
dns_records, dns_record(name="meet", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="password"))
assert is_dns_record_in_array(
dns_records, dns_record(name="password", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="social"))
assert is_dns_record_in_array(
dns_records, dns_record(name="social", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="vpn"))
assert is_dns_record_in_array(
dns_records, dns_record(name="vpn", record_type="AAAA")
)
# assert is_dns_record_in_array(dns_records, dns_record(name="cloud"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="cloud", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="git"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="git", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="meet"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="meet", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="password"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="password", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="social"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="social", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="vpn"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="vpn", record_type="AAAA")
# )
assert is_dns_record_in_array(
dns_records,
dns_record(
@ -490,30 +490,30 @@ def test_graphql_get_domain_no_uri_account_file(
assert is_dns_record_in_array(
dns_records, dns_record(name="api", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="cloud"))
assert is_dns_record_in_array(
dns_records, dns_record(name="cloud", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="git"))
assert is_dns_record_in_array(
dns_records, dns_record(name="git", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="meet"))
assert is_dns_record_in_array(
dns_records, dns_record(name="meet", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="password"))
assert is_dns_record_in_array(
dns_records, dns_record(name="password", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="social"))
assert is_dns_record_in_array(
dns_records, dns_record(name="social", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="vpn"))
assert is_dns_record_in_array(
dns_records, dns_record(name="vpn", record_type="AAAA")
)
# assert is_dns_record_in_array(dns_records, dns_record(name="cloud"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="cloud", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="git"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="git", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="meet"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="meet", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="password"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="password", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="social"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="social", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="vpn"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="vpn", record_type="AAAA")
# )
assert is_dns_record_in_array(
dns_records,
dns_record(
@ -590,30 +590,30 @@ def test_graphql_get_domain_not_found_account_file(
assert is_dns_record_in_array(
dns_records, dns_record(name="api", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="cloud"))
assert is_dns_record_in_array(
dns_records, dns_record(name="cloud", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="git"))
assert is_dns_record_in_array(
dns_records, dns_record(name="git", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="meet"))
assert is_dns_record_in_array(
dns_records, dns_record(name="meet", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="password"))
assert is_dns_record_in_array(
dns_records, dns_record(name="password", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="social"))
assert is_dns_record_in_array(
dns_records, dns_record(name="social", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="vpn"))
assert is_dns_record_in_array(
dns_records, dns_record(name="vpn", record_type="AAAA")
)
# assert is_dns_record_in_array(dns_records, dns_record(name="cloud"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="cloud", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="git"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="git", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="meet"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="meet", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="password"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="password", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="social"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="social", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="vpn"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="vpn", record_type="AAAA")
# )
assert is_dns_record_in_array(
dns_records,
dns_record(
@ -690,30 +690,30 @@ def test_graphql_get_domain_black_account_file(
assert is_dns_record_in_array(
dns_records, dns_record(name="api", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="cloud"))
assert is_dns_record_in_array(
dns_records, dns_record(name="cloud", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="git"))
assert is_dns_record_in_array(
dns_records, dns_record(name="git", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="meet"))
assert is_dns_record_in_array(
dns_records, dns_record(name="meet", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="password"))
assert is_dns_record_in_array(
dns_records, dns_record(name="password", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="social"))
assert is_dns_record_in_array(
dns_records, dns_record(name="social", record_type="AAAA")
)
assert is_dns_record_in_array(dns_records, dns_record(name="vpn"))
assert is_dns_record_in_array(
dns_records, dns_record(name="vpn", record_type="AAAA")
)
# assert is_dns_record_in_array(dns_records, dns_record(name="cloud"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="cloud", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="git"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="git", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="meet"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="meet", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="password"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="password", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="social"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="social", record_type="AAAA")
# )
# assert is_dns_record_in_array(dns_records, dns_record(name="vpn"))
# assert is_dns_record_in_array(
# dns_records, dns_record(name="vpn", record_type="AAAA")
# )
assert is_dns_record_in_array(
dns_records,
dns_record(

34
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):
@ -163,5 +161,5 @@ def test_mailserver_with_no_dkim_returns_no_dns(no_dkim_file):
assert MailServer().get_dns_records("157.90.247.192", "2a01:4f8:c17:7e3d::2") == []
def test_services_enabled_by_default(generic_userdata):
assert set(ServiceManager.get_enabled_services()) == set(services_module.services)
# def test_services_enabled_by_default(generic_userdata):
# assert set(ServiceManager.get_enabled_services()) == set(services_module.services)

27
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):
@ -74,21 +69,11 @@ 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_ok(mock_popen_systemctl_service_ok):
# assert MailServer.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)
# def test_systemctl_failed_service(mock_popen_systemctl_service_not_ok):
# assert MailServer.get_status() == ServiceStatus.FAILED
# call_args_asserts(mock_popen_systemctl_service_not_ok)