diff --git a/selfprivacy_api/graphql/common_types/service.py b/selfprivacy_api/graphql/common_types/service.py index 275c14c..fb6c972 100644 --- a/selfprivacy_api/graphql/common_types/service.py +++ b/selfprivacy_api/graphql/common_types/service.py @@ -103,6 +103,69 @@ def service_dns_to_graphql(record: ServiceDnsRecord) -> DnsRecord: ) +@strawberry.interface +class ConfigItem: + field_id: str + description: str + widget: str + type: str + + +@strawberry.type +class StringConfigItem(ConfigItem): + value: str + default_value: str + regex: Optional[str] + + +@strawberry.type +class BoolConfigItem(ConfigItem): + value: bool + default_value: bool + + +@strawberry.type +class EnumConfigItem(ConfigItem): + value: str + default_value: str + options: list[str] + + +def config_item_to_graphql(item: dict) -> ConfigItem: + item_type = item.get("type") + if item_type == "string": + return StringConfigItem( + field_id=item["id"], + description=item["description"], + widget=item["widget"], + type=item_type, + value=item["value"], + default_value=item["default_value"], + regex=item.get("regex"), + ) + elif item_type == "bool": + return BoolConfigItem( + field_id=item["id"], + description=item["description"], + widget=item["widget"], + type=item_type, + value=item["value"], + default_value=item["default_value"], + ) + elif item_type == "enum": + return EnumConfigItem( + field_id=item["id"], + description=item["description"], + widget=item["widget"], + type=item_type, + value=item["value"], + default_value=item["default_value"], + options=item["options"], + ) + else: + raise ValueError(f"Unknown config item type {item_type}") + + @strawberry.type class Service: id: str @@ -112,6 +175,7 @@ class Service: is_movable: bool is_required: bool is_enabled: bool + is_installed: bool can_be_backed_up: bool backup_description: str status: ServiceStatusEnum @@ -132,6 +196,19 @@ class Service: """Get storage usage for a service""" return get_storage_usage(self) + @strawberry.field + def configuration(self) -> Optional[List[ConfigItem]]: + """Get service configuration""" + service = get_service_by_id(self.id) + if service is None: + return None + config_items = service.get_configuration() + # If it is an empty dict, return none + 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] + # TODO: fill this @strawberry.field def backup_snapshots(self) -> Optional[List["SnapshotInfo"]]: @@ -156,6 +233,7 @@ def service_to_graphql_service(service: ServiceInterface) -> Service: is_movable=service.is_movable(), is_required=service.is_required(), is_enabled=service.is_enabled(), + is_installed=service.is_installed(), can_be_backed_up=service.can_be_backed_up(), backup_description=service.get_backup_description(), status=ServiceStatusEnum(service.get_status().value), diff --git a/selfprivacy_api/graphql/mutations/services_mutations.py b/selfprivacy_api/graphql/mutations/services_mutations.py index be0cb77..34f9539 100644 --- a/selfprivacy_api/graphql/mutations/services_mutations.py +++ b/selfprivacy_api/graphql/mutations/services_mutations.py @@ -33,6 +33,51 @@ class ServiceMutationReturn(GenericMutationReturn): service: typing.Optional[Service] = None +@strawberry.input +class SetServiceConfigurationInput: + """Set service configuration input type. + The values might be of different types: str or bool. + """ + + service_id: str + configuration: strawberry.scalars.JSON + """Yes, it is a JSON scalar, which is supposed to be a Map. + I can't define it as a proper type because GraphQL doesn't support unions in input types. + There is a @oneOf directive, but it doesn't fit this usecase. + + Other option would have been doing something like this: + ```python + @strawberry.type + class StringConfigurationInputField: + fieldId: str + value: str + + @strawberry.type + class BoolConfigurationInputField: + fieldId: str + value: bool + + // ... + + @strawberry.input + class SetServiceConfigurationInput: + service_id: str + stringFields: List[StringConfigurationInputField] + boolFields: List[BoolConfigurationInputField] + enumFields: List[EnumConfigurationInputField] + intFields: List[IntConfigurationInputField] + ``` + + But it would be very painful to maintain and will break compatibility with + every change. + + Be careful when parsing it. Probably it will be wise to add a parser/validator + later when we get a new Pydantic integration in Strawberry. + + -- Inex, 26.07.2024 + """ + + @strawberry.input class MoveServiceInput: """Move service input type.""" @@ -157,6 +202,41 @@ class ServicesMutations: service=service_to_graphql_service(service), ) + @strawberry.mutation(permission_classes=[IsAuthenticated]) + def set_service_configuration( + self, input: SetServiceConfigurationInput + ) -> ServiceMutationReturn: + """Set the new configuration values""" + service = get_service_by_id(input.service_id) + if service is None: + return ServiceMutationReturn( + success=False, + message=f"Service does not exist: {input.service_id}", + code=404, + ) + try: + service.set_configuration(input.configuration) + return ServiceMutationReturn( + success=True, + message="Service configuration updated.", + code=200, + service=service_to_graphql_service(service), + ) + except ValueError as e: + return ServiceMutationReturn( + success=False, + message=e.args[0], + code=400, + service=service_to_graphql_service(service), + ) + except Exception as e: + return ServiceMutationReturn( + success=False, + message=pretty_error(e), + code=400, + service=service_to_graphql_service(service), + ) + @strawberry.mutation(permission_classes=[IsAuthenticated]) def move_service(self, input: MoveServiceInput) -> ServiceJobMutationReturn: """Move service.""" diff --git a/selfprivacy_api/graphql/schema.py b/selfprivacy_api/graphql/schema.py index b5e6765..761bd24 100644 --- a/selfprivacy_api/graphql/schema.py +++ b/selfprivacy_api/graphql/schema.py @@ -38,6 +38,12 @@ from selfprivacy_api.graphql.subscriptions.jobs import ( ) from selfprivacy_api.graphql.subscriptions.logs import log_stream +from selfprivacy_api.graphql.common_types.service import ( + StringConfigItem, + BoolConfigItem, + EnumConfigItem, +) + from selfprivacy_api.graphql.mutations.users_mutations import UsersMutations from selfprivacy_api.graphql.queries.users import Users from selfprivacy_api.jobs.test import test_job @@ -190,4 +196,9 @@ schema = strawberry.Schema( query=Query, mutation=Mutation, subscription=Subscription, + types=[ + StringConfigItem, + BoolConfigItem, + EnumConfigItem, + ], ) diff --git a/selfprivacy_api/migrations/add_roundcube.py b/selfprivacy_api/migrations/add_roundcube.py index 3c422c2..68505ea 100644 --- a/selfprivacy_api/migrations/add_roundcube.py +++ b/selfprivacy_api/migrations/add_roundcube.py @@ -17,9 +17,6 @@ class AddRoundcube(Migration): with FlakeServiceManager() as manager: if "roundcube" not in manager.services: return True - with ReadUserData() as data: - if "roundcube" not in data["modules"]: - return True return False def migrate(self) -> None: @@ -28,9 +25,3 @@ class AddRoundcube(Migration): manager.services[ "roundcube" ] = "git+https://git.selfprivacy.org/SelfPrivacy/selfprivacy-nixos-config.git?ref=flakes&dir=sp-modules/roundcube" - with WriteUserData() as data: - if "roundcube" not in data["modules"]: - data["modules"]["roundcube"] = { - "enable": False, - "subdomain": "roundcube", - } diff --git a/selfprivacy_api/services/bitwarden/__init__.py b/selfprivacy_api/services/bitwarden/__init__.py index 56ee6e5..4614c15 100644 --- a/selfprivacy_api/services/bitwarden/__init__.py +++ b/selfprivacy_api/services/bitwarden/__init__.py @@ -1,18 +1,47 @@ """Class representing Bitwarden service""" import base64 import subprocess -from typing import Optional, List - -from selfprivacy_api.utils import get_domain +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.""" @@ -37,16 +66,6 @@ class Bitwarden(Service): def get_user() -> str: return "vaultwarden" - @classmethod - def get_url(cls) -> Optional[str]: - """Return service url.""" - domain = get_domain() - return f"https://password.{domain}" - - @classmethod - def get_subdomain(cls) -> Optional[str]: - return "password" - @staticmethod def is_movable() -> bool: return True @@ -84,14 +103,6 @@ class Bitwarden(Service): def restart(): subprocess.run(["systemctl", "restart", "vaultwarden.service"]) - @staticmethod - def get_configuration(): - return {} - - @staticmethod - def set_configuration(config_items): - return super().set_configuration(config_items) - @staticmethod def get_logs(): return "" diff --git a/selfprivacy_api/services/config_item.py b/selfprivacy_api/services/config_item.py new file mode 100644 index 0000000..eba557a --- /dev/null +++ b/selfprivacy_api/services/config_item.py @@ -0,0 +1,245 @@ +from abc import ABC, abstractmethod +import re +from typing import Optional + +from selfprivacy_api.utils import ( + ReadUserData, + WriteUserData, + check_if_subdomain_is_taken, +) + + +class ServiceConfigItem(ABC): + id: str + description: str + widget: str + type: str + + @abstractmethod + def get_value(self, service_id): + pass + + @abstractmethod + def set_value(self, value, service_id): + pass + + @abstractmethod + def validate_value(self, value): + return True + + def as_dict(self, service_options): + return { + "id": self.id, + "type": self.type, + "description": self.description, + "widget": self.widget, + "value": self.get_value(service_options), + } + + +class StringServiceConfigItem(ServiceConfigItem): + def __init__( + self, + id: str, + default_value: str, + description: str, + regex: Optional[str] = None, + widget: Optional[str] = None, + allow_empty: bool = False, + ): + if widget == "subdomain" and not regex: + raise ValueError("Subdomain widget requires regex") + self.id = id + self.type = "string" + self.default_value = default_value + self.description = description + self.regex = re.compile(regex) if regex else None + self.widget = widget if widget else "text" + self.allow_empty = allow_empty + + def get_value(self, service_id): + with ReadUserData() as user_data: + if "modules" in user_data and service_id in user_data["modules"]: + return user_data["modules"][service_id].get(self.id, self.default_value) + return self.default_value + + def set_value(self, value, service_id): + if not self.validate_value(value): + raise ValueError(f"Value {value} is not valid") + 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][self.id] = value + + def as_dict(self, service_options): + return { + "id": self.id, + "type": self.type, + "description": self.description, + "widget": self.widget, + "value": self.get_value(service_options), + "default_value": self.default_value, + "regex": self.regex.pattern if self.regex else None, + } + + def validate_value(self, value): + if not isinstance(value, str): + return False + if not self.allow_empty and not value: + return False + if self.regex and not self.regex.match(value): + return False + if self.widget == "subdomain": + if check_if_subdomain_is_taken(value): + return False + return True + + +class BoolServiceConfigItem(ServiceConfigItem): + def __init__( + self, + id: str, + default_value: bool, + description: str, + widget: Optional[str] = None, + ): + self.id = id + self.type = "bool" + self.default_value = default_value + self.description = description + self.widget = widget if widget else "switch" + + def get_value(self, service_id): + with ReadUserData() as user_data: + if "modules" in user_data and service_id in user_data["modules"]: + return user_data["modules"][service_id].get(self.id, self.default_value) + return self.default_value + + def set_value(self, value, service_id): + if not self.validate_value(value): + raise ValueError(f"Value {value} is not a boolean") + 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][self.id] = value + + def as_dict(self, service_options): + return { + "id": self.id, + "type": self.type, + "description": self.description, + "widget": self.widget, + "value": self.get_value(service_options), + "default_value": self.default_value, + } + + def validate_value(self, value): + return isinstance(value, bool) + + +class EnumServiceConfigItem(ServiceConfigItem): + def __init__( + self, + id: str, + default_value: str, + description: str, + options: list[str], + widget: Optional[str] = None, + ): + self.id = id + self.type = "enum" + self.default_value = default_value + self.description = description + self.options = options + self.widget = widget if widget else "select" + + def get_value(self, service_id): + with ReadUserData() as user_data: + if "modules" in user_data and service_id in user_data["modules"]: + return user_data["modules"][service_id].get(self.id, self.default_value) + return self.default_value + + def set_value(self, value, service_id): + if not self.validate_value(value): + raise ValueError(f"Value {value} is not in options") + 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][self.id] = value + + def as_dict(self, service_options): + return { + "id": self.id, + "type": self.type, + "description": self.description, + "widget": self.widget, + "value": self.get_value(service_options), + "default_value": self.default_value, + "options": self.options, + } + + def validate_value(self, value): + if not isinstance(value, str): + return False + return value in self.options + + +# TODO: unused for now +class IntServiceConfigItem(ServiceConfigItem): + def __init__( + self, + id: str, + default_value: int, + description: str, + widget: Optional[str] = None, + min_value: Optional[int] = None, + max_value: Optional[int] = None, + ) -> None: + self.id = id + self.type = "int" + self.default_value = default_value + self.description = description + self.widget = widget if widget else "number" + self.min_value = min_value + self.max_value = max_value + + def get_value(self, service_id): + with ReadUserData() as user_data: + if "modules" in user_data and service_id in user_data["modules"]: + return user_data["modules"][service_id].get(self.id, self.default_value) + return self.default_value + + def set_value(self, value, service_id): + if not self.validate_value(value): + raise ValueError(f"Value {value} is not valid") + 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][self.id] = value + + def as_dict(self, service_options): + return { + "id": self.id, + "type": self.type, + "description": self.description, + "widget": self.widget, + "value": self.get_value(service_options), + "default_value": self.default_value, + "min_value": self.min_value, + "max_value": self.max_value, + } + + def validate_value(self, value): + if not isinstance(value, int): + return False + return (self.min_value is None or value >= self.min_value) and ( + self.max_value is None or value <= self.max_value + ) diff --git a/selfprivacy_api/services/forgejo/__init__.py b/selfprivacy_api/services/forgejo/__init__.py index 06cf614..a5b07b0 100644 --- a/selfprivacy_api/services/forgejo/__init__.py +++ b/selfprivacy_api/services/forgejo/__init__.py @@ -1,13 +1,20 @@ """Class representing Bitwarden service""" import base64 import subprocess -from typing import Optional, List +from typing import List -from selfprivacy_api.utils import get_domain +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): @@ -16,6 +23,54 @@ class 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", + "auto", + "gitea", + "arc-green", + ], + ), + } + @staticmethod def get_id() -> str: """Return service id. For compatibility keep in gitea.""" @@ -36,16 +91,6 @@ class Forgejo(Service): """Read SVG icon from file and return it as base64 encoded string.""" return base64.b64encode(FORGEJO_ICON.encode("utf-8")).decode("utf-8") - @classmethod - def get_url(cls) -> Optional[str]: - """Return service url.""" - domain = get_domain() - return f"https://git.{domain}" - - @classmethod - def get_subdomain(cls) -> Optional[str]: - return "git" - @staticmethod def is_movable() -> bool: return True @@ -82,14 +127,6 @@ class Forgejo(Service): def restart(): subprocess.run(["systemctl", "restart", "forgejo.service"]) - @staticmethod - def get_configuration(): - return {} - - @staticmethod - def set_configuration(config_items): - return super().set_configuration(config_items) - @staticmethod def get_logs(): return "" diff --git a/selfprivacy_api/services/jitsimeet/__init__.py b/selfprivacy_api/services/jitsimeet/__init__.py index 27a497a..0bf0370 100644 --- a/selfprivacy_api/services/jitsimeet/__init__.py +++ b/selfprivacy_api/services/jitsimeet/__init__.py @@ -1,21 +1,40 @@ """Class representing Jitsi Meet service""" import base64 import subprocess -from typing import Optional, List +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 import get_domain 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.""" @@ -36,16 +55,6 @@ class JitsiMeet(Service): """Read SVG icon from file and return it as base64 encoded string.""" return base64.b64encode(JITSI_ICON.encode("utf-8")).decode("utf-8") - @classmethod - def get_url(cls) -> Optional[str]: - """Return service url.""" - domain = get_domain() - return f"https://meet.{domain}" - - @classmethod - def get_subdomain(cls) -> Optional[str]: - return "meet" - @staticmethod def is_movable() -> bool: return False @@ -88,14 +97,6 @@ class JitsiMeet(Service): ) subprocess.run(["systemctl", "restart", "jicofo.service"], check=False) - @staticmethod - def get_configuration(): - return {} - - @staticmethod - def set_configuration(config_items): - return super().set_configuration(config_items) - @staticmethod def get_logs(): return "" diff --git a/selfprivacy_api/services/mailserver/__init__.py b/selfprivacy_api/services/mailserver/__init__.py index aba302d..d5dd481 100644 --- a/selfprivacy_api/services/mailserver/__init__.py +++ b/selfprivacy_api/services/mailserver/__init__.py @@ -89,14 +89,6 @@ class MailServer(Service): subprocess.run(["systemctl", "restart", "dovecot2.service"], check=False) subprocess.run(["systemctl", "restart", "postfix.service"], check=False) - @staticmethod - def get_configuration(): - return {} - - @staticmethod - def set_configuration(config_items): - return super().set_configuration(config_items) - @staticmethod def get_logs(): return "" diff --git a/selfprivacy_api/services/nextcloud/__init__.py b/selfprivacy_api/services/nextcloud/__init__.py index 275b11d..4cd0156 100644 --- a/selfprivacy_api/services/nextcloud/__init__.py +++ b/selfprivacy_api/services/nextcloud/__init__.py @@ -1,19 +1,32 @@ """Class representing Nextcloud service.""" import base64 import subprocess -from typing import Optional, List - -from selfprivacy_api.utils import get_domain +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, + 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", + ), + } + @staticmethod def get_id() -> str: """Return service id.""" @@ -34,16 +47,6 @@ class Nextcloud(Service): """Read SVG icon from file and return it as base64 encoded string.""" return base64.b64encode(NEXTCLOUD_ICON.encode("utf-8")).decode("utf-8") - @classmethod - def get_url(cls) -> Optional[str]: - """Return service url.""" - domain = get_domain() - return f"https://cloud.{domain}" - - @classmethod - def get_subdomain(cls) -> Optional[str]: - return "cloud" - @staticmethod def is_movable() -> bool: return True @@ -84,15 +87,6 @@ class Nextcloud(Service): """Restart Nextcloud service.""" subprocess.Popen(["systemctl", "restart", "phpfpm-nextcloud.service"]) - @staticmethod - def get_configuration() -> dict: - """Return Nextcloud configuration.""" - return {} - - @staticmethod - def set_configuration(config_items): - return super().set_configuration(config_items) - @staticmethod def get_logs(): """Return Nextcloud logs.""" diff --git a/selfprivacy_api/services/ocserv/__init__.py b/selfprivacy_api/services/ocserv/__init__.py index f600772..818b1d8 100644 --- a/selfprivacy_api/services/ocserv/__init__.py +++ b/selfprivacy_api/services/ocserv/__init__.py @@ -33,10 +33,6 @@ class Ocserv(Service): """Return service url.""" return None - @classmethod - def get_subdomain(cls) -> typing.Optional[str]: - return "vpn" - @staticmethod def is_movable() -> bool: return False @@ -69,12 +65,12 @@ class Ocserv(Service): def restart(): subprocess.run(["systemctl", "restart", "ocserv.service"], check=False) - @staticmethod - def get_configuration(): + @classmethod + def get_configuration(cls): return {} - @staticmethod - def set_configuration(config_items): + @classmethod + def set_configuration(cls, config_items): return super().set_configuration(config_items) @staticmethod diff --git a/selfprivacy_api/services/pleroma/__init__.py b/selfprivacy_api/services/pleroma/__init__.py index 64edd96..e300844 100644 --- a/selfprivacy_api/services/pleroma/__init__.py +++ b/selfprivacy_api/services/pleroma/__init__.py @@ -1,9 +1,7 @@ """Class representing Nextcloud service.""" import base64 import subprocess -from typing import Optional, List - -from selfprivacy_api.utils import get_domain +from typing import List from selfprivacy_api.services.owned_path import OwnedPath from selfprivacy_api.utils.systemd import get_service_status @@ -31,16 +29,6 @@ class Pleroma(Service): def get_svg_icon() -> str: return base64.b64encode(PLEROMA_ICON.encode("utf-8")).decode("utf-8") - @classmethod - def get_url(cls) -> Optional[str]: - """Return service url.""" - domain = get_domain() - return f"https://social.{domain}" - - @classmethod - def get_subdomain(cls) -> Optional[str]: - return "social" - @staticmethod def is_movable() -> bool: return True @@ -72,12 +60,12 @@ class Pleroma(Service): subprocess.run(["systemctl", "restart", "pleroma.service"]) subprocess.run(["systemctl", "restart", "postgresql.service"]) - @staticmethod - def get_configuration(config_items): + @classmethod + def get_configuration(cls): return {} - @staticmethod - def set_configuration(config_items): + @classmethod + def set_configuration(cls, config_items): return super().set_configuration(config_items) @staticmethod diff --git a/selfprivacy_api/services/roundcube/__init__.py b/selfprivacy_api/services/roundcube/__init__.py index 22604f5..8349ccd 100644 --- a/selfprivacy_api/services/roundcube/__init__.py +++ b/selfprivacy_api/services/roundcube/__init__.py @@ -2,21 +2,35 @@ import base64 import subprocess -from typing import List, Optional +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 import ReadUserData, get_domain 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.""" @@ -37,21 +51,6 @@ class Roundcube(Service): """Read SVG icon from file and return it as base64 encoded string.""" return base64.b64encode(ROUNDCUBE_ICON.encode("utf-8")).decode("utf-8") - @classmethod - def get_url(cls) -> Optional[str]: - """Return service url.""" - domain = get_domain() - subdomain = cls.get_subdomain() - return f"https://{subdomain}.{domain}" - - @classmethod - def get_subdomain(cls) -> Optional[str]: - with ReadUserData() as data: - if "roundcube" in data["modules"]: - return data["modules"]["roundcube"]["subdomain"] - - return "roundcube" - @staticmethod def is_movable() -> bool: return False @@ -93,14 +92,6 @@ class Roundcube(Service): check=False, ) - @staticmethod - def get_configuration(): - return {} - - @staticmethod - def set_configuration(config_items): - return super().set_configuration(config_items) - @staticmethod def get_logs(): return "" diff --git a/selfprivacy_api/services/service.py b/selfprivacy_api/services/service.py index 6e3decf..17b6f4f 100644 --- a/selfprivacy_api/services/service.py +++ b/selfprivacy_api/services/service.py @@ -3,7 +3,9 @@ from abc import ABC, abstractmethod from typing import List, Optional from selfprivacy_api import utils -from selfprivacy_api.utils import ReadUserData, WriteUserData +from selfprivacy_api.services.config_item import ServiceConfigItem +from selfprivacy_api.utils.default_subdomains import DEFAULT_SUBDOMAINS +from selfprivacy_api.utils import ReadUserData, WriteUserData, get_domain from selfprivacy_api.utils.waitloop import wait_until_true from selfprivacy_api.utils.block_devices import BlockDevice, BlockDevices @@ -33,6 +35,8 @@ class Service(ABC): can be installed, configured and used by a user. """ + config_items: dict[str, "ServiceConfigItem"] = {} + @staticmethod @abstractmethod def get_id() -> str: @@ -66,20 +70,27 @@ class Service(ABC): pass @classmethod - @abstractmethod def get_url(cls) -> Optional[str]: """ The url of the service if it is accessible from the internet browser. """ - pass + domain = get_domain() + subdomain = cls.get_subdomain() + return f"https://{subdomain}.{domain}" @classmethod - @abstractmethod def get_subdomain(cls) -> Optional[str]: """ The assigned primary subdomain for this service. """ - pass + name = cls.get_id() + with ReadUserData() as user_data: + if "modules" in user_data: + if name in user_data["modules"]: + if "subdomain" in user_data["modules"][name]: + return user_data["modules"][name]["subdomain"] + + return DEFAULT_SUBDOMAINS.get(name) @classmethod def get_user(cls) -> Optional[str]: @@ -135,6 +146,16 @@ class Service(ABC): with ReadUserData() as user_data: return user_data.get("modules", {}).get(name, {}).get("enable", False) + @classmethod + def is_installed(cls) -> bool: + """ + `True` if the service is installed. + `False` if there is no module data in user data + """ + name = cls.get_id() + with ReadUserData() as user_data: + return user_data.get("modules", {}).get(name, {}) != {} + @staticmethod @abstractmethod def get_status() -> ServiceStatus: @@ -179,15 +200,24 @@ class Service(ABC): """Restart the service. Usually this means restarting systemd unit.""" pass - @staticmethod - @abstractmethod - def get_configuration(): - pass + @classmethod + def get_configuration(cls): + return { + key: cls.config_items[key].as_dict(cls.get_id()) for key in cls.config_items + } - @staticmethod - @abstractmethod - def set_configuration(config_items): - pass + @classmethod + def set_configuration(cls, config_items): + for key, value in config_items.items(): + if key not in cls.config_items: + raise ValueError(f"Key {key} is not valid for {cls.get_id()}") + if cls.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(): + cls.config_items[key].set_value( + value, + cls.get_id(), + ) @staticmethod @abstractmethod diff --git a/selfprivacy_api/services/test_service/__init__.py b/selfprivacy_api/services/test_service/__init__.py index de3c493..1a2fd9d 100644 --- a/selfprivacy_api/services/test_service/__init__.py +++ b/selfprivacy_api/services/test_service/__init__.py @@ -1,6 +1,5 @@ """Class representing Bitwarden service""" import base64 -import typing import subprocess from typing import List @@ -57,16 +56,6 @@ class DummyService(Service): # return "" return base64.b64encode(BITWARDEN_ICON.encode("utf-8")).decode("utf-8") - @classmethod - def get_url(cls) -> typing.Optional[str]: - """Return service url.""" - domain = "test.com" - return f"https://password.{domain}" - - @classmethod - def get_subdomain(cls) -> typing.Optional[str]: - return "password" - @classmethod def is_movable(cls) -> bool: return cls.movable @@ -163,12 +152,12 @@ class DummyService(Service): cls.set_status(ServiceStatus.RELOADING) # is a correct one? cls.change_status_with_async_delay(ServiceStatus.ACTIVE, cls.startstop_delay) - @staticmethod - def get_configuration(): + @classmethod + def get_configuration(cls): return {} - @staticmethod - def set_configuration(config_items): + @classmethod + def set_configuration(cls, config_items): return super().set_configuration(config_items) @staticmethod diff --git a/selfprivacy_api/utils/__init__.py b/selfprivacy_api/utils/__init__.py index 779bdf6..53568ce 100644 --- a/selfprivacy_api/utils/__init__.py +++ b/selfprivacy_api/utils/__init__.py @@ -8,6 +8,11 @@ import subprocess import portalocker import typing +from selfprivacy_api.utils.default_subdomains import ( + DEFAULT_SUBDOMAINS, + RESERVED_SUBDOMAINS, +) + USERDATA_FILE = "/etc/nixos/userdata.json" SECRETS_FILE = "/etc/selfprivacy/secrets.json" @@ -133,6 +138,20 @@ def is_username_forbidden(username): return False +def check_if_subdomain_is_taken(subdomain: str) -> bool: + """Check if subdomain is already taken or reserved""" + if subdomain in RESERVED_SUBDOMAINS: + return True + with ReadUserData() as data: + for module in data["modules"]: + if ( + data["modules"][module].get("subdomain", DEFAULT_SUBDOMAINS[module]) + == subdomain + ): + return True + return False + + def parse_date(date_str: str) -> datetime.datetime: """Parse date string which can be in one of these formats: - %Y-%m-%dT%H:%M:%S.%fZ @@ -199,3 +218,10 @@ def hash_password(password): hashed_password = hashed_password.decode("ascii") hashed_password = hashed_password.rstrip() return hashed_password + + +def write_to_log(message): + with open("/etc/selfprivacy/log", "a") as log: + log.write(f"{datetime.datetime.now()} {message}\n") + log.flush() + os.fsync(log.fileno()) diff --git a/selfprivacy_api/utils/default_subdomains.py b/selfprivacy_api/utils/default_subdomains.py new file mode 100644 index 0000000..f6665fc --- /dev/null +++ b/selfprivacy_api/utils/default_subdomains.py @@ -0,0 +1,21 @@ +DEFAULT_SUBDOMAINS = { + "bitwarden": "password", + "gitea": "git", + "jitsi-meet": "meet", + "simple-nixos-mailserver": "", + "nextcloud": "cloud", + "ocserv": "vpn", + "pleroma": "social", + "roundcube": "roundcube", + "testservice": "test", +} + +RESERVED_SUBDOMAINS = [ + "admin", + "administrator", + "api", + "auth", + "user", + "users", + "ntfy", +] diff --git a/selfprivacy_api/utils/regex_strings.py b/selfprivacy_api/utils/regex_strings.py new file mode 100644 index 0000000..718e2ab --- /dev/null +++ b/selfprivacy_api/utils/regex_strings.py @@ -0,0 +1 @@ +SUBDOMAIN_REGEX = r"^[A-Za-z0-9][A-Za-z0-9\-]{0,61}[A-Za-z0-9]$" diff --git a/tests/test_config_item.py b/tests/test_config_item.py new file mode 100644 index 0000000..724c29e --- /dev/null +++ b/tests/test_config_item.py @@ -0,0 +1,157 @@ +import pytest +from selfprivacy_api.services.config_item import ( + StringServiceConfigItem, + BoolServiceConfigItem, + EnumServiceConfigItem, +) +from selfprivacy_api.utils.regex_strings import SUBDOMAIN_REGEX + + +@pytest.fixture +def service_options(): + return {} + + +def test_string_service_config_item(service_options): + item = StringServiceConfigItem( + id="test_string", + default_value="1337", + description="Test digits string", + regex=r"^\d+$", + widget="text", + allow_empty=False, + ) + assert item.get_value(service_options) == "1337" + item.set_value("123", service_options) + assert item.get_value(service_options) == "123" + with pytest.raises(ValueError): + item.set_value("abc", service_options) + assert item.validate_value("123") is True + assert item.validate_value("abc") is False + assert item.validate_value("123abc") is False + assert item.validate_value("") is False + assert item.validate_value(None) is False + assert item.validate_value(123) is False + assert item.validate_value("123.0") is False + assert item.validate_value(True) is False + + +def test_string_service_config_item_allows_empty(service_options): + item = StringServiceConfigItem( + id="test_string", + default_value="1337", + description="Test digits string", + widget="text", + allow_empty=True, + ) + assert item.get_value(service_options) == "1337" + item.set_value("", service_options) + assert item.get_value(service_options) == "" + assert item.validate_value("") is True + assert item.validate_value(None) is False + assert item.validate_value(123) is False + assert item.validate_value("123") is True + assert item.validate_value("abc") is True + assert item.validate_value("123abc") is True + assert item.validate_value("123.0") is True + assert item.validate_value(True) is False + + +def test_string_service_config_item_not_allows_empty(service_options): + item = StringServiceConfigItem( + id="test_string", + default_value="1337", + description="Test digits string", + widget="text", + ) + assert item.get_value(service_options) == "1337" + with pytest.raises(ValueError): + item.set_value("", service_options) + assert item.get_value(service_options) == "1337" + assert item.validate_value("") is False + assert item.validate_value(None) is False + assert item.validate_value(123) is False + assert item.validate_value("123") is True + assert item.validate_value("abc") is True + assert item.validate_value("123abc") is True + assert item.validate_value("123.0") is True + assert item.validate_value(True) is False + + +def test_bool_service_config_item(service_options): + item = BoolServiceConfigItem( + id="test_bool", + default_value=True, + description="Test bool", + widget="switch", + ) + assert item.get_value(service_options) is True + item.set_value(False, service_options) + assert item.get_value(service_options) is False + assert item.validate_value(True) is True + assert item.validate_value(False) is True + assert item.validate_value("True") is False + assert item.validate_value("False") is False + assert item.validate_value(1) is False + assert item.validate_value(0) is False + assert item.validate_value("1") is False + + +def test_enum_service_config_item(service_options): + item = EnumServiceConfigItem( + id="test_enum", + default_value="option1", + description="Test enum", + options=["option1", "option2", "option3"], + widget="select", + ) + assert item.get_value(service_options) == "option1" + item.set_value("option2", service_options) + assert item.get_value(service_options) == "option2" + with pytest.raises(ValueError): + item.set_value("option4", service_options) + assert item.validate_value("option1") is True + assert item.validate_value("option4") is False + assert item.validate_value("option2") is True + assert item.validate_value(1) is False + assert item.validate_value("1") is False + assert item.validate_value(True) is False + + +def test_string_service_config_item_subdomain(service_options, dummy_service): + item = StringServiceConfigItem( + id="test_subdomain", + default_value="example", + description="Test subdomain string", + widget="subdomain", + allow_empty=False, + regex=SUBDOMAIN_REGEX, + ) + assert item.get_value(service_options) == "example" + item.set_value("subdomain", service_options) + assert item.get_value(service_options) == "subdomain" + with pytest.raises(ValueError): + item.set_value( + "invalid-subdomain-because-it-is-very-very-very-very-very-very-long", + service_options, + ) + assert item.validate_value("subdomain") is True + assert ( + item.validate_value( + "invalid-subdomain-because-it-is-very-very-very-very-very-very-long" + ) + is False + ) + assert item.validate_value("api") is False + assert item.validate_value("auth") is False + assert item.validate_value("user") is False + assert item.validate_value("users") is False + assert item.validate_value("ntfy") is False + assert item.validate_value("") is False + assert item.validate_value(None) is False + assert item.validate_value(123) is False + assert item.validate_value("123") is True + assert item.validate_value("abc") is True + assert item.validate_value("123abc") is True + assert item.validate_value("123.0") is False + assert item.validate_value(True) is False diff --git a/tests/test_graphql/test_services.py b/tests/test_graphql/test_services.py index b7faf3d..b349f53 100644 --- a/tests/test_graphql/test_services.py +++ b/tests/test_graphql/test_services.py @@ -188,6 +188,7 @@ allServices { id status isEnabled + url } """ @@ -347,6 +348,7 @@ def test_get_services(authorized_client, only_dummy_service): assert api_dummy_service["id"] == "testservice" assert api_dummy_service["status"] == ServiceStatus.ACTIVE.value assert api_dummy_service["isEnabled"] is True + assert api_dummy_service["url"] == "https://test.test-domain.tld" def test_enable_return_value(authorized_client, only_dummy_service):