SelfPrivacy/selfprivacy-rest-api
1
0
Fork 0
mirror of https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git synced 2024-09-16 13:57:18 +00:00
Code Issues Projects Releases Wiki Activity

feat: Service configuration (#127)

Browse source 
Reviewed-on: https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api/pulls/127
This commit is contained in:
Inex Code 2024-07-26 18:33:04 +03:00
parent 40b8eb06d0
commit 9b93107b36
20 changed files with 819 additions and 178 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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),

80
selfprivacy_api/graphql/mutations/services_mutations.py
View file

@ -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<str, Union[str, int, bool]>.
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."""

11
selfprivacy_api/graphql/schema.py
View file

@ -37,6 +37,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
@ -184,4 +190,9 @@ schema = strawberry.Schema(
query=Query,
mutation=Mutation,
subscription=Subscription,
types=[
StringConfigItem,
BoolConfigItem,
EnumConfigItem,
],
)

9
selfprivacy_api/migrations/add_roundcube.py
View file

@ -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",
}

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

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

245
selfprivacy_api/services/config_item.py Normal file
View file

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

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

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

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

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

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

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

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

@ -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."""

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

@ -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

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

@ -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

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

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

56
selfprivacy_api/services/service.py
View file

@ -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

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

@ -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

26
selfprivacy_api/utils/__init__.py
View file

@ -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())

21
selfprivacy_api/utils/default_subdomains.py Normal file
View file

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

1
selfprivacy_api/utils/regex_strings.py Normal file
View file

@ -0,0 +1 @@
SUBDOMAIN_REGEX = r"^[A-Za-z0-9][A-Za-z0-9\-]{0,61}[A-Za-z0-9]$"

157
tests/test_config_item.py Normal file
View file

@ -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

2
tests/test_graphql/test_services.py
View file

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