Add saml_id to the account import actions

openslides_backend/action/actions/user/import_.py

@@ -39,7 +39,7 @@ def update_instance(self, instance: Dict[str, Any]) -> Dict[str, Any]:
         data = self.result.get("rows", [])
         for entry in data:
             # Revert username-info and default-password-info
-            for field in ("username", "default_password"):
+            for field in ("username", "default_password", "saml_id"):
                 if field in entry["data"]:
                     if field == "username" and "id" in entry["data"][field]:
                         entry["data"]["id"] = entry["data"][field]["id"]
@@ -48,7 +48,7 @@ def update_instance(self, instance: Dict[str, Any]) -> Dict[str, Any]:
         search_data_list = [
             {
                 field: entry["data"].get(field, "")
-                for field in ("username", "first_name", "last_name", "email")
+                for field in ("username", "first_name", "last_name", "email", "saml_id")
             }
             for entry in data
         ]
@@ -60,38 +60,64 @@ def update_instance(self, instance: Dict[str, Any]) -> Dict[str, Any]:
         self.error = False
         for payload_index, entry in enumerate(data):
             if entry["state"] == ImportState.NEW:
-                if not entry["data"].get("username"):
+                if not entry["data"].get("username") and not entry["data"].get(
+                    "saml_id"
+                ):
                     self.error = True
                     entry["state"] = ImportState.ERROR
                     entry["messages"].append(
                         "Error: Want to create user, but missing username in import data."
                     )
-                elif self.check_username_for_duplicate(
+                elif entry["data"].get(
+                    "username"
+                ) and self.check_username_for_duplicate(
                     entry["data"]["username"], payload_index
                 ):
                     self.error = True
                     entry["state"] = ImportState.ERROR
                     entry["messages"].append(
                         "Error: want to create a new user, but username already exists."
                     )
+                elif entry["data"].get("saml_id") and self.check_saml_id_for_duplicate(
+                    entry["data"]["saml_id"], payload_index
+                ):
+                    self.error = True
+                    entry["state"] = ImportState.ERROR
+                    entry["messages"].append(
+                        "Error: want to create a new user, but saml_id already exists."
+                    )
                 else:
                     create_action_payload.append(entry["data"])
             elif entry["state"] == ImportState.DONE:
                 search_data = self.get_search_data(payload_index)
-                if not entry["data"].get("username"):
+                if not entry["data"].get("username") and not entry["data"].get(
+                    "saml_id"
+                ):
                     self.error = True
                     entry["state"] = ImportState.ERROR
                     entry["messages"].append(
                         "Error: Want to update user, but missing username in import data."
                     )
-                elif not self.check_username_for_duplicate(
+                elif entry["data"].get(
+                    "username"
+                ) and not self.check_username_for_duplicate(
                     entry["data"]["username"], payload_index
                 ):
                     self.error = True
                     entry["state"] = ImportState.ERROR
                     entry["messages"].append(
                         "Error: want to update, but missing user in db."
                     )
+                elif entry["data"].get(
+                    "saml_id"
+                ) and not self.check_saml_id_for_duplicate(
+                    entry["data"]["saml_id"], payload_index
+                ):
+                    self.error = True
+                    entry["state"] = ImportState.ERROR
+                    entry["messages"].append(
+                        "Error: want to update, but missing user in db."
+                    )
                 elif search_data is None:
                     self.error = True
                     entry["state"] = ImportState.ERROR
@@ -105,7 +131,10 @@ def update_instance(self, instance: Dict[str, Any]) -> Dict[str, Any]:
                         "Error: want to update, but found search data doesn't match."
                     )
                 else:
-                    del entry["data"]["username"]
+                    for field in ("username", "saml_id"):
+                        if field in entry["data"]:
+                            del entry["data"][field]
+
                     update_action_payload.append(entry["data"])
             else:
                 self.error = True

openslides_backend/action/actions/user/json_upload.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Tuple
+from typing import Any, Dict, Optional, Tuple
 
 import fastjsonschema
 
@@ -37,6 +37,7 @@ class UserJsonUpload(DuplicateCheckMixin, UsernameMixin, JsonUploadMixin):
                             "username",
                             "gender",
                             "pronoun",
+                            "saml_id",
                         ),
                     },
                     "required": [],
@@ -58,6 +59,7 @@ class UserJsonUpload(DuplicateCheckMixin, UsernameMixin, JsonUploadMixin):
         {"property": "username", "type": "string"},
         {"property": "gender", "type": "string"},
         {"property": "pronoun", "type": "string"},
+        {"property": "saml_id", "type": "string"},
     ]
     permission = OrganizationManagementLevel.CAN_MANAGE_USERS
     skip_archived_meeting_check = True
@@ -70,7 +72,13 @@ def update_instance(self, instance: Dict[str, Any]) -> Dict[str, Any]:
             [
                 {
                     field: entry.get(field, "")
-                    for field in ("username", "first_name", "last_name", "email")
+                    for field in (
+                        "username",
+                        "saml_id",
+                        "first_name",
+                        "last_name",
+                        "email",
+                    )
                 }
                 for entry in data
             ]
@@ -108,20 +116,20 @@ def generate_entry(
             UserCreate.schema_validator(entry)
             if entry.get("username"):
                 if self.check_username_for_duplicate(entry["username"], payload_index):
-                    state = ImportState.DONE
-                    if searchdata := self.get_search_data(payload_index):
-                        entry["username"] = {
-                            "value": entry["username"],
-                            "info": "done",
-                            "id": searchdata["id"],
-                        }
-                    else:
-                        entry["username"] = {"value": entry["username"], "info": "done"}
-                        state = ImportState.ERROR
-                        messages.append(f"Duplicate in csv list index: {payload_index}")
+                    state, msg = self.handle_done_entry(
+                        "username", entry, payload_index
+                    )
+                    if msg:
+                        messages.append(msg)
                 else:
-                    state = ImportState.NEW
-                    entry["username"] = {"value": entry["username"], "info": "done"}
+                    state = self.handle_new_entry("username", entry)
+            elif entry.get("saml_id"):
+                if self.check_saml_id_for_duplicate(entry["saml_id"], payload_index):
+                    state, msg = self.handle_done_entry("saml_id", entry, payload_index)
+                    if msg:
+                        messages.append(msg)
+                else:
+                    state = self.handle_new_entry("saml_id", entry)
             else:
                 if not (entry.get("first_name") or entry.get("last_name")):
                     state = ImportState.ERROR
@@ -153,6 +161,12 @@ def generate_entry(
                         "info": ImportState.GENERATED,
                     }
             self.handle_default_password(entry, state)
+            if entry.get("saml_id", {}).get("value"):
+                if entry.get("password") or entry.get("default_password"):
+                    messages.append("Remove password or default_password from entry.")
+                entry.pop("password", None)
+                entry.pop("default_password", None)
+                entry["can_change_own_password"] = False
         except fastjsonschema.JsonSchemaException as exception:
             state = ImportState.ERROR
             messages.append(exception.message)
@@ -181,3 +195,25 @@ def _names_and_email(entry: Dict[str, Any]) -> Tuple[str, str, str]:
             entry.get("last_name", ""),
             entry.get("email", ""),
         )
+
+    def handle_done_entry(
+        self, fieldname: str, entry: Dict[str, Any], payload_index: int
+    ) -> Tuple[ImportState, Optional[str]]:
+        state = ImportState.DONE
+        message = None
+        if searchdata := self.get_search_data(payload_index):
+            entry[fieldname] = {
+                "value": entry[fieldname],
+                "info": "done",
+                "id": searchdata["id"],
+            }
+        else:
+            entry[fieldname] = {"value": entry[fieldname], "info": "done"}
+            state = ImportState.ERROR
+            message = f"Duplicate in csv list index: {payload_index}"
+        return state, message
+
+    def handle_new_entry(self, fieldname: str, entry: Dict[str, Any]) -> ImportState:
+        state = ImportState.NEW
+        entry[fieldname] = {"value": entry[fieldname], "info": "done"}
+        return state

openslides_backend/action/actions/user/update.py

@@ -51,6 +51,7 @@ class UserUpdate(
             "organization_management_level",
             "committee_management_ids",
             "is_demo_user",
+            "saml_id",
         ],
         additional_optional_fields={
             "meeting_id": optional_id_schema,

openslides_backend/action/actions/user/user_mixin.py

@@ -200,6 +200,7 @@ def init_duplicate_set(self, data: List[Any]) -> None:
             },
         )
         self.used_usernames: List[str] = []
+        self.used_saml_ids: List[str] = []
         self.used_names_and_email: List[Any] = []
 
     def check_username_for_duplicate(self, username: str, payload_index: int) -> bool:
@@ -211,6 +212,15 @@ def check_username_for_duplicate(self, username: str, payload_index: int) -> boo
             self.used_usernames.append(username)
         return result
 
+    def check_saml_id_for_duplicate(self, saml_id: str, payload_index: int) -> bool:
+        result = (
+            bool(self.users_in_double_lists[payload_index])
+            or saml_id in self.used_saml_ids
+        )
+        if saml_id not in self.used_saml_ids:
+            self.used_saml_ids.append(saml_id)
+        return result
+
     def check_name_and_email_for_duplicate(
         self, first_name: str, last_name: str, email: str, payload_index: int
     ) -> bool:

tests/system/action/user/test_import.py

@@ -151,6 +151,91 @@ def get_action_worker_data(
             }
         }
 
+    def test_import_with_saml_id(self) -> None:
+        self.set_models(
+            self.get_action_worker_data(
+                6,
+                ImportState.NEW,
+                {"saml_id": {"value": "testsaml", "info": ImportState.NEW}},
+            )
+        )
+        response = self.request("user.import", {"id": 6, "import": True})
+        self.assert_status_code(response, 200)
+        self.assert_model_exists(
+            "user/2",
+            {
+                "username": "testsaml",
+                "saml_id": "testsaml",
+            },
+        )
+
+    def test_import_saml_id_error_new_and_saml_id_exists(self) -> None:
+        """Set saml_id 'testsaml' to user 1, add the import user 1 will be
+        found and the import should result in an error."""
+        self.set_models(
+            {
+                "user/1": {"saml_id": "testsaml"},
+                **self.get_action_worker_data(
+                    6,
+                    ImportState.NEW,
+                    {"saml_id": {"value": "testsaml", "info": ImportState.NEW}},
+                ),
+            }
+        )
+        response = self.request("user.import", {"id": 6, "import": True})
+        self.assert_status_code(response, 200)
+        entry = response.json["results"][0][0]["rows"][0]
+        assert entry["state"] == ImportState.ERROR
+        assert entry["messages"] == [
+            "Error: want to create a new user, but saml_id already exists."
+        ]
+
+    def test_import_done_with_saml_id(self) -> None:
+        self.set_models(
+            {
+                "user/2": {"username": "test", "saml_id": "testsaml"},
+                **self.get_action_worker_data(
+                    6,
+                    ImportState.DONE,
+                    {
+                        "saml_id": {"value": "testsaml", "info": ImportState.DONE},
+                        "id": 2,
+                        "first_name": "Hugo",
+                    },
+                ),
+            }
+        )
+        response = self.request("user.import", {"id": 6, "import": True})
+        self.assert_status_code(response, 200)
+        self.assert_model_exists(
+            "user/2",
+            {
+                "username": "test",
+                "saml_id": "testsaml",
+                "first_name": "Hugo",
+            },
+        )
+
+    def test_import_done_error_missing_user(self) -> None:
+        self.set_models(
+            {
+                **self.get_action_worker_data(
+                    6,
+                    ImportState.DONE,
+                    {
+                        "saml_id": {"value": "testsaml", "info": ImportState.NEW},
+                        "first_name": "Hugo",
+                        "id": 2,
+                    },
+                ),
+            }
+        )
+        response = self.request("user.import", {"id": 6, "import": True})
+        self.assert_status_code(response, 200)
+        entry = response.json["results"][0][0]["rows"][0]
+        assert entry["state"] == ImportState.ERROR
+        assert entry["messages"] == ["Error: want to update, but missing user in db."]
+
     def test_import_error_at_state_new(self) -> None:
         self.set_models(
             {

tests/system/action/user/test_json_upload.py

@@ -107,6 +107,7 @@ def test_json_upload_results(self) -> None:
                 {"property": "username", "type": "string"},
                 {"property": "gender", "type": "string"},
                 {"property": "pronoun", "type": "string"},
+                {"property": "saml_id", "type": "string"},
             ],
             "rows": [
                 {
@@ -346,6 +347,34 @@ def test_json_upload_generate_default_password(self) -> None:
             == ImportState.GENERATED
         )
 
+    def test_json_upload_saml_id(self) -> None:
+        response = self.request(
+            "user.json_upload",
+            {
+                "data": [
+                    {
+                        "saml_id": "test",
+                        "password": "test2",
+                        "default_password": "test3",
+                    }
+                ],
+            },
+        )
+        self.assert_status_code(response, 200)
+        worker = self.assert_model_exists("action_worker/1")
+        assert worker["result"]["import"] == "account"
+        assert worker["result"]["rows"][0]["messages"] == [
+            "Remove password or default_password from entry."
+        ]
+        data = worker["result"]["rows"][0]["data"]
+        assert data.get("saml_id") == {
+            "value": "test",
+            "info": "done",
+        }
+        assert "password" not in data
+        assert "default_password" not in data
+        assert data.get("can_change_own_password") is False
+
     def test_json_upload_no_permission(self) -> None:
         self.base_permission_test(
             {}, "user.json_upload", {"data": [{"username": "test"}]}