Merge branch 'feature/updater' into feature/store-i18n

tests/unit/python/roles/web-app-keycloak/files/__init__.py

@@ -0,0 +1,5 @@
+from pathlib import Path
+
+PROJECT_ROOT: Path = Path(__file__).resolve().parents[6]
+
+__all__ = ["PROJECT_ROOT"]

tests/unit/python/roles/web-app-keycloak/files/test_email_verified.py

@@ -0,0 +1,129 @@
+import contextlib
+import io
+import os
+import sys
+import unittest
+import urllib.parse
+from unittest import mock
+
+from . import PROJECT_ROOT
+
+sys.path.insert(0, str(PROJECT_ROOT / "roles/web-app-keycloak/files/python"))
+
+import email_verified as m  # noqa: E402
+
+
+class FakeKeycloak:
+    def __init__(self):
+        self.users = {}
+        self.puts = []
+        self.failing = set()
+
+    def add_user(self, username, verified, **extra):
+        user_id = f"u-{len(self.users) + 1}"
+        self.users[user_id] = {
+            "id": user_id,
+            "username": username,
+            "emailVerified": verified,
+            **extra,
+        }
+        return user_id
+
+    def call(self, method, path, payload=None, timeout=60):
+        route, _, query = path.partition("?")
+        if method == "GET" and route == "users":
+            wanted = urllib.parse.parse_qs(query)["username"][0]
+            if wanted in self.failing:
+                raise m.KeycloakError(f"GET {path}: HTTP 500")
+            return [
+                dict(u) for u in self.users.values() if wanted.lower() in u["username"]
+            ], ""
+        if method == "GET" and route.startswith("users/"):
+            return dict(self.users[route.split("/")[1]]), ""
+        if method == "PUT" and route.startswith("users/"):
+            user_id = route.split("/")[1]
+            self.puts.append(user_id)
+            self.users[user_id] = dict(payload)
+            return None, ""
+        raise AssertionError(f"unexpected call {method} {path}")
+
+
+class TestVerify(unittest.TestCase):
+    def test_an_unverified_user_is_updated_and_keeps_its_other_fields(self):
+        kc = FakeKeycloak()
+        user_id = kc.add_user("biber", False, email="biber@example.org")
+        self.assertEqual(m.verify(kc, "biber"), "updated")
+        self.assertIs(kc.users[user_id]["emailVerified"], True)
+        self.assertEqual(kc.users[user_id]["email"], "biber@example.org")
+
+    def test_a_verified_user_is_left_alone(self):
+        kc = FakeKeycloak()
+        kc.add_user("biber", True)
+        self.assertEqual(m.verify(kc, "biber"), "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(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")
+        self.assertEqual(kc.puts, [target])
+        self.assertIs(kc.users[other]["emailVerified"], False)
+
+
+class TestMain(unittest.TestCase):
+    def _run(self, kc, usernames):
+        out = io.StringIO()
+        env = {
+            "KC_URL": "http://kc",
+            "KC_PID": "1",
+            "KC_REALM": "r",
+            "USERNAMES": usernames,
+        }
+        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),
+            contextlib.redirect_stderr(io.StringIO()),
+        ):
+            code = m.main()
+        return code, out.getvalue().splitlines()
+
+    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")
+        self.assertEqual(code, 0)
+        self.assertEqual(lines[-1], "summary updated=1 unchanged=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.failing.add("administrator")
+        code, lines = self._run(kc, "administrator biber")
+        self.assertEqual(code, 1)
+        self.assertIn("failed administrator", lines)
+        self.assertIn("updated biber", lines)
+
+    def test_a_missing_environment_key_fails_before_any_call(self):
+        kc = FakeKeycloak()
+        with (
+            mock.patch.dict(os.environ, {}, clear=True),
+            mock.patch.object(m, "Keycloak", lambda *args: kc),
+            contextlib.redirect_stdout(io.StringIO()) as out,
+        ):
+            code = m.main()
+        self.assertEqual(code, 1)
+        self.assertTrue(out.getvalue().startswith("setup-failed"))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/python/roles/web-app-keycloak/files/test_keycloak_admin.py

@@ -0,0 +1,79 @@
+import io
+import json
+import sys
+import tempfile
+import unittest
+import urllib.error
+from pathlib import Path
+from unittest import mock
+
+from . import PROJECT_ROOT
+
+sys.path.insert(0, str(PROJECT_ROOT / "roles/web-app-keycloak/files/python"))
+
+import keycloak_admin as m  # noqa: E402
+
+
+class TestContainerCredentials(unittest.TestCase):
+    def test_credentials_come_from_the_container_environment(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            environ = Path(tmp) / "environ"
+            environ.write_bytes(
+                b"PATH=/bin\0KEYCLOAK_PERMANENT_ADMIN_USERNAME=admin\0"
+                b"KEYCLOAK_PERMANENT_ADMIN_PASSWORD=a=b\0"
+            )
+            self.assertEqual(m.container_credentials(environ), ("admin", "a=b"))
+
+
+class _Response(io.BytesIO):
+    def __init__(self, body=b"", location=""):
+        super().__init__(body)
+        self.headers = {"Location": location} if location else {}
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+
+class TestKeycloakClient(unittest.TestCase):
+    def test_one_token_serves_calls_until_it_nears_expiry(self):
+        token = json.dumps({"access_token": "t", "expires_in": 300}).encode()
+        calls = []
+
+        def urlopen(request, timeout):
+            calls.append(request.full_url)
+            if request.full_url.endswith("/token"):
+                return _Response(token)
+            return _Response(b"", "http://kc/admin/realms/r/groups/new-id")
+
+        kc = m.Keycloak("http://kc", "r", "admin", "pw")
+        with mock.patch.object(m.urllib.request, "urlopen", urlopen):
+            _, created = kc.call("POST", "groups/root/children", {"name": "x"})
+            kc.call("GET", "groups/new-id")
+        self.assertEqual(created, "new-id")
+        self.assertEqual(sum(url.endswith("/token") for url in calls), 1)
+
+    def test_an_http_error_names_the_status_but_not_the_body(self):
+        token = json.dumps({"access_token": "t", "expires_in": 300}).encode()
+
+        def urlopen(request, timeout):
+            if request.full_url.endswith("/token"):
+                return _Response(token)
+            raise urllib.error.HTTPError(
+                request.full_url, 409, "Conflict", {}, io.BytesIO(b"secret detail")
+            )
+
+        kc = m.Keycloak("http://kc", "r", "admin", "pw")
+        with (
+            mock.patch.object(m.urllib.request, "urlopen", urlopen),
+            self.assertRaises(m.KeycloakError) as raised,
+        ):
+            kc.call("POST", "groups/root/children", {"name": "x"})
+        self.assertIn("HTTP 409", str(raised.exception))
+        self.assertNotIn("secret", str(raised.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/python/roles/web-app-keycloak/files/test_rbac_reconcile.py

@@ -0,0 +1,226 @@
+import base64
+import json
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+from . import PROJECT_ROOT
+
+sys.path.insert(0, str(PROJECT_ROOT / "roles/web-app-keycloak/files/python"))
+
+import rbac_reconcile as m  # noqa: E402
+
+LDAP = "ldap-1"
+
+
+class FakeKeycloak:
+    def __init__(self):
+        self.groups = {"root": {"id": "root", "name": "roles", "attributes": {}}}
+        self.parents = {}
+        self.components = {}
+        self.syncs = []
+        self.failing_syncs = set()
+        self._next = 0
+
+    def _id(self, prefix):
+        self._next += 1
+        return f"{prefix}-{self._next}"
+
+    def add_group(self, name, parent="root", attributes=None):
+        group_id = self._id("g")
+        self.groups[group_id] = {
+            "id": group_id,
+            "name": name,
+            "attributes": attributes or {},
+        }
+        self.parents[group_id] = parent
+        return group_id
+
+    def add_mapper(self, name, config):
+        mapper_id = self._id("c")
+        self.components[mapper_id] = {
+            "id": mapper_id,
+            "name": name,
+            "parentId": LDAP,
+            "config": config,
+        }
+        return mapper_id
+
+    def children(self, group_id):
+        return [
+            dict(g)
+            for gid, g in self.groups.items()
+            if self.parents.get(gid) == group_id
+        ]
+
+    def call(self, method, path, payload=None, timeout=60):
+        route = path.split("?", 1)[0]
+        if method == "GET" and route == "components":
+            return [dict(c) for c in self.components.values()], ""
+        if method == "GET" and route == "group-by-path/roles":
+            return dict(self.groups["root"]), ""
+        if method == "POST" and route == "components":
+            mapper_id = self._id("c")
+            self.components[mapper_id] = {**payload, "id": mapper_id}
+            return None, mapper_id
+        if method == "PUT" and route.startswith("components/"):
+            self.components[route.split("/")[1]] = dict(payload)
+            return None, ""
+        if method == "DELETE" and route.startswith("components/"):
+            del self.components[route.split("/")[1]]
+            return None, ""
+        if (
+            method == "POST"
+            and route.startswith("groups/")
+            and route.endswith("/children")
+        ):
+            return None, self.add_group(payload["name"], route.split("/")[1])
+        if method == "GET" and route.startswith("groups/"):
+            return json.loads(json.dumps(self.groups[route.split("/")[1]])), ""
+        if method == "PUT" and route.startswith("groups/"):
+            self.groups[route.split("/")[1]] = dict(payload)
+            return None, ""
+        if method == "DELETE" and route.startswith("groups/"):
+            del self.groups[route.split("/")[1]]
+            return None, ""
+        if method == "POST" and route.startswith(f"user-storage/{LDAP}/mappers/"):
+            mapper_id = route.split("/")[3]
+            if mapper_id in self.failing_syncs:
+                raise m.KeycloakError("POST sync: HTTP 500")
+            self.syncs.append(mapper_id)
+            return None, ""
+        raise AssertionError(f"unexpected call {method} {path}")
+
+
+def _payload(app_id, filter_suffix=""):
+    return json.dumps(
+        {
+            "name": f"ldap-roles-{app_id}",
+            "parentId": LDAP,
+            "providerId": "group-ldap-mapper",
+            "config": {
+                "groups.path": [f"/roles/{app_id}"],
+                "groups.ldap.filter": [f"(cn={app_id}-*){filter_suffix}"],
+            },
+        }
+    )
+
+
+def _app(app_id, fingerprint="fp-1", filter_suffix=""):
+    return m.App(
+        app_id, f"ldap-roles-{app_id}", _payload(app_id, filter_suffix), fingerprint
+    )
+
+
+def _reconciler(kc, attempts=2):
+    groups = {g["name"]: g["id"] for g in kc.children("root")}
+    mappers = {c["name"]: c for c in kc.components.values()}
+    return m.Reconciler(kc, LDAP, "root", groups, mappers, attempts, 0, 5)
+
+
+class TestReconciler(unittest.TestCase):
+    def test_a_new_app_gets_its_group_mapper_sync_and_fingerprint(self):
+        kc = FakeKeycloak()
+        self.assertEqual(_reconciler(kc).run(_app("web-app-a")), "synced")
+        group = next(g for g in kc.groups.values() if g["name"] == "web-app-a")
+        mapper = next(
+            c for c in kc.components.values() if c["name"] == "ldap-roles-web-app-a"
+        )
+        self.assertEqual(mapper["config"]["groups.path"], ["/roles/web-app-a"])
+        self.assertEqual(kc.syncs, [mapper["id"]])
+        self.assertEqual(group["attributes"][m.FINGERPRINT], ["fp-1"])
+
+    def test_an_unchanged_app_is_not_synced_again(self):
+        kc = FakeKeycloak()
+        _reconciler(kc).run(_app("web-app-a"))
+        self.assertEqual(_reconciler(kc).run(_app("web-app-a")), "unchanged")
+        self.assertEqual(len(kc.syncs), 1)
+
+    def test_a_new_fingerprint_syncs_again(self):
+        kc = FakeKeycloak()
+        _reconciler(kc).run(_app("web-app-a"))
+        self.assertEqual(_reconciler(kc).run(_app("web-app-a", "fp-2")), "synced")
+        self.assertEqual(len(kc.syncs), 2)
+
+    def test_a_drifted_mapper_is_updated_and_synced(self):
+        kc = FakeKeycloak()
+        _reconciler(kc).run(_app("web-app-a"))
+        self.assertEqual(
+            _reconciler(kc).run(_app("web-app-a", filter_suffix="(drift)")), "synced"
+        )
+        mapper = next(
+            c for c in kc.components.values() if c["name"] == "ldap-roles-web-app-a"
+        )
+        self.assertEqual(
+            mapper["config"]["groups.ldap.filter"], ["(cn=web-app-a-*)(drift)"]
+        )
+
+    def test_extra_config_keycloak_adds_does_not_count_as_drift(self):
+        kc = FakeKeycloak()
+        _reconciler(kc).run(_app("web-app-a"))
+        mapper = next(
+            c for c in kc.components.values() if c["name"] == "ldap-roles-web-app-a"
+        )
+        mapper["config"]["mapped.group.attributes"] = [""]
+        self.assertEqual(_reconciler(kc).run(_app("web-app-a")), "unchanged")
+
+    def test_a_failed_sync_records_no_fingerprint(self):
+        kc = FakeKeycloak()
+        group_id = kc.add_group("web-app-a")
+        mapper_id = kc.add_mapper(
+            "ldap-roles-web-app-a",
+            {
+                "groups.path": ["/roles/web-app-a"],
+                "groups.ldap.filter": ["(cn=web-app-a-*)"],
+            },
+        )
+        kc.failing_syncs.add(mapper_id)
+        with self.assertRaises(m.StepError) as raised:
+            _reconciler(kc).run(_app("web-app-a"))
+        self.assertEqual(raised.exception.step, "sync")
+        self.assertNotIn(m.FINGERPRINT, kc.groups[group_id]["attributes"])
+
+    def test_existing_group_attributes_survive_the_fingerprint(self):
+        kc = FakeKeycloak()
+        kc.add_group("web-app-a", attributes={"LDAP_ID": ["x"]})
+        _reconciler(kc).run(_app("web-app-a"))
+        group = next(g for g in kc.groups.values() if g["name"] == "web-app-a")
+        self.assertEqual(group["attributes"]["LDAP_ID"], ["x"])
+
+
+class TestHousekeeping(unittest.TestCase):
+    def test_orphans_go_and_app_groups_stay(self):
+        kc = FakeKeycloak()
+        keep = kc.add_group("web-app-a")
+        kc.add_group("web-app-a-administrator")
+        kc.add_group("administrator")
+        kept, failures = m.purge_orphans(kc, "root", {"web-app-a"}, 2)
+        self.assertEqual(kept, {"web-app-a": keep})
+        self.assertEqual(failures, [])
+        self.assertEqual(sorted(g["name"] for g in kc.children("root")), ["web-app-a"])
+
+    def test_the_shared_mapper_is_dropped(self):
+        kc = FakeKeycloak()
+        kc.add_mapper(m.SHARED_MAPPER, {})
+        per_app = kc.add_mapper("ldap-roles-web-app-a", {})
+        mappers = m.ldap_mappers(kc, LDAP)
+        self.assertEqual(m.drop_shared_mapper(kc, mappers), [])
+        self.assertEqual(list(kc.components), [per_app])
+        self.assertNotIn(m.SHARED_MAPPER, mappers)
+
+    def test_payload_lines_are_parsed(self):
+        encoded = base64.b64encode(_payload("web-app-a").encode()).decode()
+        with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as handle:
+            handle.write(f"web-app-a ldap-roles-web-app-a {encoded} fp-9\n\n")
+        apps = m.read_payloads(handle.name)
+        Path(handle.name).unlink()
+        self.assertEqual(
+            [(a.app_id, a.mapper_name, a.fingerprint) for a in apps],
+            [("web-app-a", "ldap-roles-web-app-a", "fp-9")],
+        )
+        self.assertEqual(json.loads(apps[0].payload)["name"], "ldap-roles-web-app-a")
+
+
+if __name__ == "__main__":
+    unittest.main()