selfprivacy-rest-api/selfprivacy_api/services/__init__.py

264 lines
7.2 KiB
Python
Raw Normal View History

"""Services module."""
import base64
import typing
from typing import List
2024-08-21 13:55:30 +00:00
from os import path, mkdir, remove
from os import makedirs
from os import listdir
from os.path import join
from pathlib import Path
from selfprivacy_api.services.bitwarden import Bitwarden
2024-06-30 19:02:07 +00:00
from selfprivacy_api.services.forgejo import Forgejo
from selfprivacy_api.services.jitsimeet import JitsiMeet
from selfprivacy_api.services.prometheus import Prometheus
2024-06-06 19:14:07 +00:00
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
import selfprivacy_api.utils.network as network_utils
from selfprivacy_api.services.test_service.icon import BITWARDEN_ICON
from selfprivacy_api.utils import USERDATA_FILE, DKIM_DIR, SECRETS_FILE
from selfprivacy_api.utils.block_devices import BlockDevices
from shutil import copyfile, copytree, rmtree
2024-08-21 13:55:30 +00:00
CONFIG_STASH_DIR = "/etc/selfprivacy/dump"
class ServiceManager(Service):
folders: List[str] = [CONFIG_STASH_DIR]
@staticmethod
def get_all_services() -> list[Service]:
return services
@staticmethod
def get_service_by_id(service_id: str) -> typing.Optional[Service]:
for service in 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()]
# 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()]
@staticmethod
def get_services_by_location(location: str) -> list[Service]:
return [service for service in services if service.get_drive() == location]
@staticmethod
def get_all_required_dns_records() -> list[ServiceDnsRecord]:
ip4 = network_utils.get_ip4()
ip6 = network_utils.get_ip6()
dns_records: list[ServiceDnsRecord] = [
ServiceDnsRecord(
type="A",
name="api",
content=ip4,
ttl=3600,
display_name="SelfPrivacy API",
),
]
if ip6 is not None:
dns_records.append(
ServiceDnsRecord(
type="AAAA",
name="api",
content=ip6,
ttl=3600,
display_name="SelfPrivacy API (IPv6)",
)
)
for service in ServiceManager.get_enabled_services():
dns_records += service.get_dns_records(ip4, ip6)
return dns_records
@staticmethod
def get_id() -> str:
"""Return service id."""
return "api"
@staticmethod
def get_display_name() -> str:
"""Return service display name."""
return "Selfprivacy API"
@staticmethod
def get_description() -> str:
"""Return service description."""
return "A proto-service for API itself. Currently manages backups of settings."
@staticmethod
def get_svg_icon() -> str:
"""Read SVG icon from file and return it as base64 encoded string."""
# return ""
return base64.b64encode(BITWARDEN_ICON.encode("utf-8")).decode("utf-8")
@staticmethod
def get_url() -> typing.Optional[str]:
"""Return service url."""
# TODO : placeholder, get actual domain here
return f"https://domain"
@staticmethod
def get_subdomain() -> typing.Optional[str]:
return None
@staticmethod
def is_always_active() -> bool:
return True
@staticmethod
def is_movable() -> bool:
return False
@staticmethod
def is_required() -> bool:
return True
@staticmethod
def is_enabled() -> bool:
return True
@staticmethod
def get_backup_description() -> str:
return "How did we get here?"
@classmethod
def get_status(cls) -> ServiceStatus:
return ServiceStatus.ACTIVE
@classmethod
def can_be_backed_up(cls) -> bool:
"""`True` if the service can be backed up."""
return True
@classmethod
def merge_settings(cls):
# 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:
locations[service.get_id()] = service.get_drive()
# Copy files
for p in [USERDATA_FILE, SECRETS_FILE, DKIM_DIR]:
cls.retrieve_stashed_path(p)
# Pop locations
for service in services:
device = BlockDevices().get_block_device(locations[service.get_id()])
if device is not None:
service.set_location(device)
@classmethod
def stop(cls):
"""
We are always active
"""
raise ValueError("tried to stop an always active service")
@classmethod
def start(cls):
"""
We are always active
"""
pass
@classmethod
def restart(cls):
"""
We are always active
"""
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
@classmethod
def get_folders(cls) -> List[str]:
return cls.folders
@classmethod
def stash_for(cls, p: str) -> str:
basename = path.basename(p)
stashed_file_location = join(cls.dump_dir(), basename)
return stashed_file_location
@classmethod
def stash_a_path(cls, p: str):
if path.isdir(p):
rmtree(cls.stash_for(p), ignore_errors=True)
copytree(p, cls.stash_for(p))
else:
copyfile(p, cls.stash_for(p))
@classmethod
def retrieve_stashed_path(cls, p: str):
"""
Takes an original path, hopefully it is stashed somewhere
"""
if path.isdir(p):
rmtree(p, ignore_errors=True)
copytree(cls.stash_for(p), p)
else:
copyfile(cls.stash_for(p), p)
@classmethod
def pre_backup(cls):
tempdir = cls.dump_dir()
2024-08-21 13:55:30 +00:00
if not path.exists(tempdir):
makedirs(tempdir)
paths = listdir(tempdir)
for file in paths:
remove(file)
for p in [USERDATA_FILE, SECRETS_FILE, DKIM_DIR]:
cls.stash_a_path(p)
@classmethod
def dump_dir(cls) -> str:
"""
A directory we dump our settings into
"""
return cls.folders[0]
@classmethod
def post_restore(cls):
cls.merge_settings()
rmtree(cls.dump_dir(), ignore_errors=True)
services: list[Service] = [
Bitwarden(),
Forgejo(),
MailServer(),
Nextcloud(),
Pleroma(),
Ocserv(),
JitsiMeet(),
Roundcube(),
ServiceManager(),
Prometheus(),
]