diff --git a/roles/web-app-keycloak/files/python/email_verified.py b/roles/web-app-keycloak/files/python/email_verified.py index 1f3ea428d1..8352c7f1fd 100755 --- a/roles/web-app-keycloak/files/python/email_verified.py +++ b/roles/web-app-keycloak/files/python/email_verified.py @@ -5,14 +5,15 @@ Env: KC_URL: Keycloak base URL reachable from the network namespace the script runs in. KC_PID: host PID of the Keycloak container, whose environment holds the admin credentials. KC_REALM: realm holding the users. - USERNAMES: space-separated usernames to mark verified. + EMAILS_JSON: JSON object of username to the address the inventory declares for that user. Output: one `` `` line per user, then one ``summary`` line. -Exit: 0 every user is verified or absent from the realm, 1 otherwise. +Exit: 0 every user is verified, absent from the realm or holds another address, 1 otherwise. """ from __future__ import annotations +import json import os import sys import urllib.parse @@ -20,7 +21,7 @@ from pathlib import Path from keycloak_admin import Keycloak, KeycloakError, container_credentials -OUTCOMES = ("updated", "unchanged", "missing", "failed") +OUTCOMES = ("updated", "unchanged", "mismatch", "missing", "failed") def find_user(kc: Keycloak, username: str) -> dict | None: @@ -36,16 +37,20 @@ def find_user(kc: Keycloak, username: str) -> dict | None: return next((user for user in found if user["username"].lower() == wanted), None) -def verify(kc: Keycloak, username: str) -> str: - """Return ``updated``, ``unchanged`` or ``missing`` for a user. +def verify(kc: Keycloak, username: str, email: str) -> str: + """Return ``updated``, ``unchanged``, ``mismatch`` or ``missing`` for a user. Args: kc: client. username: the user whose email address gets marked verified. + email: the address the inventory declares; a user holding any other address stays as it is. """ user = find_user(kc, username) if user is None: return "missing" + held = user.get("email") + if not isinstance(held, str) or held.lower() != email.lower(): + return "mismatch" if user.get("emailVerified") is True: return "unchanged" current, _ = kc.call("GET", f"users/{user['id']}") @@ -53,6 +58,21 @@ def verify(kc: Keycloak, username: str) -> str: return "updated" +def load_emails(raw: str) -> dict[str, str]: + """Return the username to address map, refusing anything else. + + Args: + raw: the JSON text of ``EMAILS_JSON``. + """ + emails = json.loads(raw) + if not isinstance(emails, dict) or not all( + isinstance(key, str) and isinstance(value, str) and value + for key, value in emails.items() + ): + raise ValueError("EMAILS_JSON must map every username to a non-empty address") + return emails + + def main() -> int: env = os.environ try: @@ -61,15 +81,15 @@ def main() -> int: env["KC_REALM"], *container_credentials(Path("/proc") / env["KC_PID"] / "environ"), ) - usernames = env["USERNAMES"].split() + emails = load_emails(env["EMAILS_JSON"]) except (OSError, KeyError, ValueError) as error: print(f"setup-failed {error}") return 1 outcomes = [] - for username in usernames: + for username, email in emails.items(): try: - outcome = verify(kc, username) + outcome = verify(kc, username, email) except (KeycloakError, OSError, KeyError, ValueError) as error: print(f"{username}: {error}", file=sys.stderr) outcome = "failed" diff --git a/roles/web-app-keycloak/tasks/update/06_users/email_verified.yml b/roles/web-app-keycloak/tasks/update/06_users/email_verified.yml index 797a3ab3c6..9a4c578a8f 100644 --- a/roles/web-app-keycloak/tasks/update/06_users/email_verified.yml +++ b/roles/web-app-keycloak/tasks/update/06_users/email_verified.yml @@ -1,5 +1,5 @@ --- -- name: "🔧 Set emailVerified=true for users in realm '{{ KEYCLOAK_REALM }}'" +- name: "🔧 Set emailVerified=true for users holding their inventory address in realm '{{ KEYCLOAK_REALM }}'" ansible.builtin.include_tasks: file: "{{ role_path }}/tasks/utils/rest_script.yml" apply: @@ -7,7 +7,7 @@ vars: rest_script: email_verified.py rest_script_env: - USERNAMES: "{{ lookup('users_non_reserved') | dict2items | map(attribute='value.username') | join(' ') }}" + EMAILS_JSON: "{{ lookup('users_non_reserved') | dict2items | map(attribute='value') | list | items2dict(key_name='username', value_name='email') | to_json }}" rest_script_stdin: "" rest_script_ok_rcs: [0] rest_script_changed: "updated=[1-9]" diff --git a/roles/web-app-keycloak/templates/import/components/org.keycloak.userprofile.UserProfileProvider.json.j2 b/roles/web-app-keycloak/templates/import/components/org.keycloak.userprofile.UserProfileProvider.json.j2 index 195a7f733d..0e405482ed 100644 --- a/roles/web-app-keycloak/templates/import/components/org.keycloak.userprofile.UserProfileProvider.json.j2 +++ b/roles/web-app-keycloak/templates/import/components/org.keycloak.userprofile.UserProfileProvider.json.j2 @@ -41,7 +41,7 @@ "displayName": "${email}", "validations": {"email": {}, "length": {"max": 255}}, "required": {"roles": ["user"]}, - "permissions": {"view": ["admin","user"], "edit": ["admin","user"]}, + "permissions": {"view": ["admin","user"], "edit": ["admin"]}, "multivalued": false }, { diff --git a/tests/unit/python/roles/web-app-keycloak/files/test_email_verified.py b/tests/unit/python/roles/web-app-keycloak/files/test_email_verified.py index d57b40936b..6718e7236b 100644 --- a/tests/unit/python/roles/web-app-keycloak/files/test_email_verified.py +++ b/tests/unit/python/roles/web-app-keycloak/files/test_email_verified.py @@ -1,5 +1,6 @@ import contextlib import io +import json import os import sys import unittest @@ -49,41 +50,60 @@ class FakeKeycloak: class TestVerify(unittest.TestCase): - def test_an_unverified_user_is_updated_and_keeps_its_other_fields(self): + def test_an_unverified_user_holding_its_inventory_address_is_updated(self): kc = FakeKeycloak() - user_id = kc.add_user("biber", False, email="biber@example.org") - self.assertEqual(m.verify(kc, "biber"), "updated") + user_id = kc.add_user("biber", False, email="biber@example.org", firstName="B") + self.assertEqual(m.verify(kc, "biber", "biber@example.org"), "updated") self.assertIs(kc.users[user_id]["emailVerified"], True) self.assertEqual(kc.users[user_id]["email"], "biber@example.org") + self.assertEqual(kc.users[user_id]["firstName"], "B") - def test_a_verified_user_is_left_alone(self): + def test_a_user_holding_another_address_is_never_marked_verified(self): kc = FakeKeycloak() - kc.add_user("biber", True) - self.assertEqual(m.verify(kc, "biber"), "unchanged") + user_id = kc.add_user("biber", False, email="erpnext-api-bot@example.org") + self.assertEqual(m.verify(kc, "biber", "biber@example.org"), "mismatch") + self.assertIs(kc.users[user_id]["emailVerified"], False) + self.assertEqual(kc.puts, []) + + def test_a_user_without_an_address_is_never_marked_verified(self): + kc = FakeKeycloak() + kc.add_user("biber", False) + self.assertEqual(m.verify(kc, "biber", "biber@example.org"), "mismatch") + self.assertEqual(kc.puts, []) + + def test_the_addresses_compare_case_insensitively(self): + kc = FakeKeycloak() + kc.add_user("biber", False, email="biber@example.org") + self.assertEqual(m.verify(kc, "biber", "Biber@Example.org"), "updated") + + def test_a_verified_user_holding_its_inventory_address_is_left_alone(self): + kc = FakeKeycloak() + kc.add_user("biber", True, email="biber@example.org") + self.assertEqual(m.verify(kc, "biber", "biber@example.org"), "unchanged") self.assertEqual(kc.puts, []) def test_an_absent_user_is_reported_missing(self): kc = FakeKeycloak() - self.assertEqual(m.verify(kc, "biber"), "missing") + self.assertEqual(m.verify(kc, "biber", "biber@example.org"), "missing") self.assertEqual(kc.puts, []) def test_only_the_exact_username_is_touched(self): kc = FakeKeycloak() - other = kc.add_user("contact2", False) - target = kc.add_user("contact", False) - self.assertEqual(m.verify(kc, "Contact"), "updated") + other = kc.add_user("contact2", False, email="contact@example.org") + target = kc.add_user("contact", False, email="contact@example.org") + self.assertEqual(m.verify(kc, "Contact", "contact@example.org"), "updated") self.assertEqual(kc.puts, [target]) self.assertIs(kc.users[other]["emailVerified"], False) class TestMain(unittest.TestCase): - def _run(self, kc, usernames): + def _run(self, kc, emails): out = io.StringIO() env = { "KC_URL": "http://kc", "KC_PID": "1", "KC_REALM": "r", - "USERNAMES": usernames, + "EMAILS_JSON": json.dumps(emails), } with ( mock.patch.dict(os.environ, env), @@ -97,18 +117,37 @@ class TestMain(unittest.TestCase): def test_the_summary_counts_every_outcome(self): kc = FakeKeycloak() - kc.add_user("administrator", True) - kc.add_user("biber", False) - code, lines = self._run(kc, "administrator biber ghost") + kc.add_user("administrator", True, email="administrator@example.org") + kc.add_user("biber", False, email="biber@example.org") + kc.add_user("mallory", False, email="erpnext-api-bot@example.org") + code, lines = self._run( + kc, + { + "administrator": "administrator@example.org", + "biber": "biber@example.org", + "mallory": "mallory@example.org", + "ghost": "ghost@example.org", + }, + ) self.assertEqual(code, 0) - self.assertEqual(lines[-1], "summary updated=1 unchanged=1 missing=1 failed=0") + self.assertIn("mismatch mallory", lines) + self.assertEqual( + lines[-1], + "summary updated=1 unchanged=1 mismatch=1 missing=1 failed=0", + ) def test_one_failed_user_fails_the_run_but_not_the_others(self): kc = FakeKeycloak() - kc.add_user("administrator", False) - kc.add_user("biber", False) + kc.add_user("administrator", False, email="administrator@example.org") + kc.add_user("biber", False, email="biber@example.org") kc.failing.add("administrator") - code, lines = self._run(kc, "administrator biber") + code, lines = self._run( + kc, + { + "administrator": "administrator@example.org", + "biber": "biber@example.org", + }, + ) self.assertEqual(code, 1) self.assertIn("failed administrator", lines) self.assertIn("updated biber", lines) @@ -124,6 +163,29 @@ class TestMain(unittest.TestCase): self.assertEqual(code, 1) self.assertTrue(out.getvalue().startswith("setup-failed")) + def test_emails_that_are_no_username_to_address_object_fail_before_any_call( + self, + ): + kc = FakeKeycloak() + kc.add_user("biber", False, email="biber@example.org") + out = io.StringIO() + env = { + "KC_URL": "http://kc", + "KC_PID": "1", + "KC_REALM": "r", + "EMAILS_JSON": json.dumps(["biber"]), + } + with ( + mock.patch.dict(os.environ, env), + mock.patch.object(m, "Keycloak", lambda *args: kc), + mock.patch.object(m, "container_credentials", lambda path: ("a", "p")), + contextlib.redirect_stdout(out), + ): + code = m.main() + self.assertEqual(code, 1) + self.assertTrue(out.getvalue().startswith("setup-failed")) + self.assertEqual(kc.puts, []) + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/python/roles/web-app-keycloak/templates/test_realm_import_json.py b/tests/unit/python/roles/web-app-keycloak/templates/test_realm_import_json.py index 9b0d8f36a4..3ab6d57024 100644 --- a/tests/unit/python/roles/web-app-keycloak/templates/test_realm_import_json.py +++ b/tests/unit/python/roles/web-app-keycloak/templates/test_realm_import_json.py @@ -199,6 +199,24 @@ class TestRealmImportJson(unittest.TestCase): CONTEXT["IAM_USERNAME_MAX_LENGTH"], ) + def test_only_an_admin_may_change_an_email_address(self) -> None: + realm = json.loads(render([])) + provider = realm["components"]["org.keycloak.userprofile.UserProfileProvider"] + profile = json.loads(provider[0]["config"]["kc.user.profile.config"][0]) + email = next( + attribute + for attribute in profile["attributes"] + if attribute["name"] == "email" + ) + self.assertEqual( + email["permissions"]["edit"], + ["admin"], + "applications resolve their account by this address and the realm does " + "not verify a change, so a self-service edit takes over whichever account " + "the typed address names", + ) + self.assertEqual(email["permissions"]["view"], ["admin", "user"]) + def test_realm_is_valid_json_with_one_saml_app(self) -> None: realm = json.loads(render(["web-app-suitecrm"])) clients = {c["clientId"]: c for c in realm["clients"]}