From e277edf6e795d59a913e60ffca6fb32f5c2f03f7 Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sun, 1 Nov 2026 08:57:17 +0100 Subject: [PATCH] fix(web-app-keycloak): close self-service email takeover Users could change their Keycloak email without verification, and every deploy stamped any address an inventory user held as verified. An address no Keycloak account holds (for example erpnext-api-bot@) could thus be claimed and reach apps that link SSO logins by email. The user profile now lets only admins edit the email, and email_verified.py stamps only an address that equals the inventory address. Co-Authored-By: Claude Opus 5.5 --- .../files/python/email_verified.py | 36 +++++-- .../tasks/update/06_users/email_verified.yml | 4 +- ...ak.userprofile.UserProfileProvider.json.j2 | 2 +- .../files/test_email_verified.py | 100 ++++++++++++++---- .../templates/test_realm_import_json.py | 18 ++++ 5 files changed, 130 insertions(+), 30 deletions(-) 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"]}