fix(oauth2): enforce titles to be unique

CMK-29825

Change-Id: I3b93d1c2b6d5f830dabab6868682a584152cbaef

Author: loocars

cmk/gui/watolib/configuration_entity/_oauth2_connections.py

@@ -8,6 +8,7 @@
 from marshmallow import ValidationError
 
 from cmk.gui.form_specs import (
+    FormSpecValidationError,
     get_visitor,
     process_validation_messages,
     RawDiskData,
@@ -25,9 +26,31 @@
     update_reference,
 )
 from cmk.gui.watolib.passwords import load_passwords
+from cmk.shared_typing import vue_formspec_components as shared_type_defs
 from cmk.utils.oauth2_connection import OAuth2Connection, OAuth2ConnectorType
 
 
+def _validate_unique_title(title: str, exclude_ident: str | None = None) -> None:
+    """Raise FormSpecValidationError if title is already used by another connection."""
+    for ident, entry in load_oauth2_connections().items():
+        if ident == exclude_ident:
+            continue
+        if entry["title"] == title:
+            raise FormSpecValidationError(
+                [
+                    shared_type_defs.ValidationMessage(
+                        location=["title"],
+                        message=_(
+                            "The title must be unique. The title '%s' is already used by "
+                            "the OAuth2 connection with ID %s."
+                        )
+                        % (title, ident),
+                        replacement_value=title,
+                    )
+                ]
+            )
+
+
 def update_oauth2_connection_and_passwords_from_slidein_schema(
     data: RawFrontendData,
     connector_type: OAuth2ConnectorType,
@@ -45,6 +68,8 @@ def update_oauth2_connection_and_passwords_from_slidein_schema(
     disk_data = visitor.to_disk(data)
     assert isinstance(disk_data, dict)
 
+    _validate_unique_title(disk_data["title"], exclude_ident=disk_data["ident"])
+
     owned_by = None
     match disk_data.get("editable_by"):
         case ("administrators", None):
@@ -106,6 +131,8 @@ def save_oauth2_connection_and_passwords_from_slidein_schema(
             field_name="data",
         )
 
+    _validate_unique_title(disk_data["title"])
+
     owned_by = None
     match disk_data.get("editable_by"):
         case ("administrators", None):

packages/cmk-frontend-vue/src/mode-oauth2-connection/ModeCreateOAuth2ConnectionApp.vue

@@ -7,7 +7,7 @@ conditions defined in the file COPYING, which is part of this source code packag
 <script setup lang="ts">
 import { type Oauth2ConnectionConfig } from 'cmk-shared-typing/typescript/mode_oauth2_connection'
 import type { FormSpec } from 'cmk-shared-typing/typescript/vue_formspec_components'
-import { provide, ref } from 'vue'
+import { computed, provide, ref } from 'vue'
 
 import usei18n from '@/lib/i18n'
 import type { TranslatedString } from '@/lib/i18nString'
@@ -62,18 +62,21 @@ immediateWatch(
 )
 const dataRef = ref<OAuth2FormData>(props.form_spec.data)
 provide(submitKey, submit)
+
+const initialData = props.form_spec.data
+const formSpecRef = computed(() => ({
+  id: props.form_spec.id,
+  spec: props.form_spec.spec,
+  validation: validationRef.value,
+  data: initialData
+}))
 </script>
 
 <template>
   <CreateOAuth2Connection
     v-model:data="dataRef"
     :config="config"
-    :form-spec="{
-      id: form_spec.id,
-      spec: form_spec.spec,
-      validation: validationRef,
-      data: dataRef
-    }"
+    :form-spec="formSpecRef"
     :authority-mapping="authority_mapping"
     :api="api"
     :connector-type="connector_type"

tests/unit/cmk/gui/openapi/configuration_entity/test_openapi_configuration_entity_oauth2_connection.py

@@ -18,6 +18,10 @@
 MY_CLIENT_ID = str(uuid.uuid4())
 MY_TENANT_ID = str(uuid.uuid4())
 
+SECOND_OAUTH2_CONNECTION_UUID = str(uuid.uuid4())
+SECOND_CLIENT_ID = str(uuid.uuid4())
+SECOND_TENANT_ID = str(uuid.uuid4())
+
 OAUTH2_CONNECTION_CONTENT = {
     MY_OAUTH2_CONNECTION_UUID: OAuth2Connection(
         title="My OAuth2 connection",
@@ -31,6 +35,20 @@
     ),
 }
 
+TWO_OAUTH2_CONNECTIONS_CONTENT = {
+    **OAUTH2_CONNECTION_CONTENT,
+    SECOND_OAUTH2_CONNECTION_UUID: OAuth2Connection(
+        title="Second OAuth2 connection",
+        client_secret=("cmk_postprocessed", "stored_password", ("second_client_secret", "")),
+        access_token=("cmk_postprocessed", "stored_password", ("second_access_token", "")),
+        refresh_token=("cmk_postprocessed", "stored_password", ("second_refresh_token", "")),
+        client_id=SECOND_CLIENT_ID,
+        tenant_id=SECOND_TENANT_ID,
+        authority="global",
+        connector_type="microsoft_entra_id",
+    ),
+}
+
 
 @pytest.fixture
 def mock_update_passwords_merged_file(monkeypatch: pytest.MonkeyPatch) -> Iterable[None]:
@@ -231,3 +249,117 @@ def test_create_non_existing_oauth2_connection_without_permissions(
     # THEN
     assert resp.status_code == 401
     assert resp.json["title"] == "Unauthorized"
+
+
+@pytest.mark.usefixtures("mock_update_passwords_merged_file")
+def test_create_oauth2_connection_with_duplicate_title(
+    clients: ClientRegistry, with_admin: tuple[str, str]
+) -> None:
+    # GIVEN
+    for ident, details in OAUTH2_CONNECTION_CONTENT.items():
+        save_oauth2_connection(ident, details, user_id=None, pprint_value=False, use_git=False)
+    clients.ConfigurationEntity.set_credentials(with_admin[0], with_admin[1])
+    my_new_uuid = str(uuid.uuid4())
+
+    # WHEN
+    resp = clients.ConfigurationEntity.create_configuration_entity(
+        {
+            "entity_type": ConfigEntityType.oauth2_connection.value,
+            "entity_type_specifier": "microsoft_entra_id",
+            "data": {
+                "ident": my_new_uuid,
+                "title": "My OAuth2 connection",
+                "editable_by": ("administrators", None),
+                "shared_with": [],
+                "client_secret": ("explicit_password", "", "my_client_secret", False),
+                "access_token": ("explicit_password", "", "my_access_token", False),
+                "refresh_token": ("explicit_password", "", "my_refresh_token", False),
+                "client_id": MY_CLIENT_ID,
+                "tenant_id": MY_TENANT_ID,
+                "authority": option_id("global"),
+            },
+        },
+        expect_ok=False,
+    )
+
+    # THEN
+    assert resp.status_code == 422, resp.json
+    validation_errors = resp.json["ext"]["validation_errors"]
+    assert len(validation_errors) == 1
+    assert validation_errors[0]["location"] == ["title"]
+    assert "The title must be unique" in validation_errors[0]["message"]
+
+
+@pytest.mark.usefixtures("mock_update_passwords_merged_file")
+def test_update_oauth2_connection_keeping_same_title(
+    clients: ClientRegistry, with_admin: tuple[str, str]
+) -> None:
+    # GIVEN
+    for ident, details in OAUTH2_CONNECTION_CONTENT.items():
+        save_oauth2_connection(ident, details, user_id=None, pprint_value=False, use_git=False)
+    clients.ConfigurationEntity.set_credentials(with_admin[0], with_admin[1])
+
+    # WHEN
+    resp = clients.ConfigurationEntity.update_configuration_entity(
+        {
+            "entity_id": MY_OAUTH2_CONNECTION_UUID,
+            "entity_type": ConfigEntityType.oauth2_connection.value,
+            "entity_type_specifier": "microsoft_entra_id",
+            "data": {
+                "ident": MY_OAUTH2_CONNECTION_UUID,
+                "title": "My OAuth2 connection",
+                "editable_by": ("administrators", None),
+                "shared_with": [],
+                "client_secret": ("explicit_password", "", "my_client_secret", False),
+                "access_token": ("explicit_password", "", "my_access_token", False),
+                "refresh_token": ("explicit_password", "", "my_refresh_token", False),
+                "client_id": MY_CLIENT_ID,
+                "tenant_id": MY_TENANT_ID,
+                "authority": option_id("global"),
+            },
+        },
+        expect_ok=True,
+    )
+
+    # THEN
+    assert resp.status_code == 200, resp.json
+    assert resp.json["id"] == MY_OAUTH2_CONNECTION_UUID, resp.json
+
+
+@pytest.mark.usefixtures("mock_update_passwords_merged_file")
+def test_update_oauth2_connection_to_existing_title(
+    clients: ClientRegistry, with_admin: tuple[str, str]
+) -> None:
+    # GIVEN
+    for ident, details in TWO_OAUTH2_CONNECTIONS_CONTENT.items():
+        save_oauth2_connection(ident, details, user_id=None, pprint_value=False, use_git=False)
+    clients.ConfigurationEntity.set_credentials(with_admin[0], with_admin[1])
+
+    # WHEN
+    resp = clients.ConfigurationEntity.update_configuration_entity(
+        {
+            "entity_id": MY_OAUTH2_CONNECTION_UUID,
+            "entity_type": ConfigEntityType.oauth2_connection.value,
+            "entity_type_specifier": "microsoft_entra_id",
+            "data": {
+                "ident": MY_OAUTH2_CONNECTION_UUID,
+                "title": "Second OAuth2 connection",
+                "editable_by": ("administrators", None),
+                "shared_with": [],
+                "client_secret": ("explicit_password", "", "my_client_secret", False),
+                "access_token": ("explicit_password", "", "my_access_token", False),
+                "refresh_token": ("explicit_password", "", "my_refresh_token", False),
+                "client_id": MY_CLIENT_ID,
+                "tenant_id": MY_TENANT_ID,
+                "authority": option_id("global"),
+            },
+        },
+        expect_ok=False,
+    )
+
+    # THEN
+    assert resp.status_code == 422, resp.json
+    validation_errors = resp.json["ext"]["validation_errors"]
+    assert len(validation_errors) == 1
+    assert validation_errors[0]["location"] == ["title"]
+    assert "The title must be unique" in validation_errors[0]["message"]