feat: auth migrations

selfprivacy_api/migrations/__init__.py

@@ -18,6 +18,7 @@ from selfprivacy_api.migrations.check_for_system_rebuild_jobs import (
 )
 from selfprivacy_api.migrations.add_roundcube import AddRoundcube
 from selfprivacy_api.migrations.add_monitoring import AddMonitoring
+from selfprivacy_api.migrations.add_auth import AddAuth
 
 logger = logging.getLogger(__name__)
 
@@ -26,6 +27,7 @@ migrations = [
     CheckForSystemRebuildJobs(),
     AddMonitoring(),
     AddRoundcube(),
+    AddAuth(),
 ]
 
 

selfprivacy_api/migrations/add_auth.py

@@ -0,0 +1,35 @@
+from selfprivacy_api.migrations.migration import Migration
+
+from selfprivacy_api.services.flake_service_manager import FlakeServiceManager
+from selfprivacy_api.utils import ReadUserData, WriteUserData
+
+
+class AddAuth(Migration):
+    """Adds auth (kanidm) service if it is not present."""
+
+    def get_migration_name(self) -> str:
+        return "add_auth"
+
+    def get_migration_description(self) -> str:
+        return "Adds the auth (Kanidm) if it is not present."
+
+    def is_migration_needed(self) -> bool:
+        with FlakeServiceManager() as manager:
+            if "auth" not in manager.services:
+                return True
+        with ReadUserData() as data:
+            if "auth" not in data["modules"]:
+                return True
+        return False
+
+    def migrate(self) -> None:
+        with FlakeServiceManager() as manager:
+            if "auth" not in manager.services:
+                manager.services["monitoring"] = (
+                    "git+https://git.selfprivacy.org/SelfPrivacy/selfprivacy-nixos-config.git?ref=sso&rev=f795bc977f03de64c10a62528bfa04a88f2611ca&dir=sp-modules/auth"
+                )
+        with WriteUserData() as data:
+            if "monitoring" not in data["modules"]:
+                data["modules"]["monitoring"] = {
+                    "enable": False,
+                }

selfprivacy_api/migrations/migrate_users_to_kanidm.py

@@ -0,0 +1,48 @@
+from selfprivacy_api.migrations.migration import Migration
+
+from selfprivacy_api.models.user import UserDataUserOrigin
+from selfprivacy_api.repositories.users import ACTIVE_USERS_PROVIDER
+from selfprivacy_api.repositories.users.kanidm_user_repository import (
+    ADMIN_GROUPS,
+    KanidmUserRepository,
+)
+from selfprivacy_api.repositories.users.json_user_repository import JsonUserRepository
+
+from selfprivacy_api.actions.users import create_user
+
+
+class MigrateUsersToKanidm(Migration):
+    """Migrate users to kanidm."""
+
+    def __init__(self):
+        self.users_to_migrate = None
+
+    def get_migration_name(self) -> str:
+        return "migrate_users_to_kanidm"
+
+    def get_migration_description(self) -> str:
+        return "Migrate users to kanidm."
+
+    def is_migration_needed(self) -> bool:
+        if isinstance(ACTIVE_USERS_PROVIDER, JsonUserRepository):
+            return False
+
+        json_repo_users = JsonUserRepository.get_users(exclude_root=True)
+        kanidm_repo_users = KanidmUserRepository.get_users(exclude_root=True)
+
+        self.users_to_migrate = [
+            user for user in json_repo_users if user not in kanidm_repo_users
+        ]
+
+        return bool(self.users_to_migrate)
+
+    def migrate(self) -> None:
+        for user in self.users_to_migrate:  # type: ignore
+
+            if user.user_type == UserDataUserOrigin.PRIMARY:
+                create_user(
+                    username=user.username,
+                    directmemberof=ADMIN_GROUPS,
+                )
+
+            create_user(username=user.username)

selfprivacy_api/repositories/users/kanidm_user_repository.py

@@ -35,7 +35,7 @@ redis = RedisPool().get_connection()
 
 KANIDM_URL = "https://127.0.0.1:3013"
 ADMIN_GROUPS = ["sp.admins"]
-DEFAULT_GROUPS = [f"idm_all_persons@{DOMAIN}", f"idm_all_accounts@{DOMAIN}"] 
+DEFAULT_GROUPS = [f"idm_all_persons@{DOMAIN}", f"idm_all_accounts@{DOMAIN}"]
 
 logger = logging.getLogger(__name__)
 
@@ -512,8 +512,14 @@ class KanidmUserRepository(AbstractUserRepository):
 
         attrs = user_data["attrs"]  # type: ignore
 
-        directmemberof = [item for item in attrs.get("directmemberof", []) if item not in DEFAULT_GROUPS]
-        memberof = [item for item in attrs.get("memberof", []) if item not in DEFAULT_GROUPS]
+        directmemberof = [
+            item
+            for item in attrs.get("directmemberof", [])
+            if item not in DEFAULT_GROUPS
+        ]
+        memberof = [
+            item for item in attrs.get("memberof", []) if item not in DEFAULT_GROUPS
+        ]
 
         return UserDataUser(
             username=attrs["name"][0],