From 2ebc4f0ae8530e8bf35f7e075361fcde306375a3 Mon Sep 17 00:00:00 2001 From: tegwick Date: Wed, 28 Oct 2026 20:11:04 +0100 Subject: [PATCH] NK-WP-0054-T05: identity-provisioner logs in as a configurable LLDAP user; svc-*/-admin names reserved LLDAP_ADMIN_USERNAME (default admin) lets the provisioner use svc-provisioner after the bind split; preferred usernames starting svc- or ending -admin are refused like admin. Co-Authored-By: Claude Opus 5.5 Assistant: claude-code Assistant-Model: opus Assistant-Process: 308701@bnt-lap001 Assistant-Session: 5548e447-444c-4dcc-bed1-05c16eebfe56 --- identity-provisioner/password_setup.py | 3 +++ identity-provisioner/provisioner.py | 12 ++++++++++-- identity-provisioner/server.py | 2 ++ .../tests/test_directory_contract.py | 11 +++++++++++ identity-provisioner/tests/test_provisioner.py | 3 +++ 5 files changed, 29 insertions(+), 2 deletions(-) diff --git a/identity-provisioner/password_setup.py b/identity-provisioner/password_setup.py index 6104aa7..1f6e66b 100644 --- a/identity-provisioner/password_setup.py +++ b/identity-provisioner/password_setup.py @@ -115,12 +115,14 @@ class LLDAPPasswordSetter: *, base_url: str, admin_password: str, + admin_username: str = "admin", helper: str = "/app/lldap_set_password", opener: Callable = urlopen, runner: Callable = subprocess.run, ) -> None: self.base_url = base_url.rstrip("/") self.admin_password = admin_password + self.admin_username = admin_username self.helper = helper self.opener = opener self.runner = runner @@ -129,6 +131,7 @@ class LLDAPPasswordSetter: token = directory_login( base_url=self.base_url, admin_password=self.admin_password, + admin_username=self.admin_username, opener=self.opener, timeout=10, ) diff --git a/identity-provisioner/provisioner.py b/identity-provisioner/provisioner.py index 28a3e5a..a746da5 100644 --- a/identity-provisioner/provisioner.py +++ b/identity-provisioner/provisioner.py @@ -61,6 +61,7 @@ class LLDAPProvisioner: *, base_url: str, admin_password: str, + admin_username: str = "admin", opener: Callable = urlopen, auth_rejected_cooldown: float = 30.0, clock: Callable[[], float] = time.monotonic, @@ -69,6 +70,7 @@ class LLDAPProvisioner: self.group_provenance = group_provenance self.base_url = base_url.rstrip("/") self.admin_password = admin_password + self.admin_username = admin_username self.opener = opener self.auth_rejected_cooldown = auth_rejected_cooldown self.clock = clock @@ -239,6 +241,7 @@ mutation Remove($userId: String!, $groupId: Int!) { return directory_login( base_url=self.base_url, admin_password=self.admin_password, + admin_username=self.admin_username, opener=self.opener, timeout=timeout, ) @@ -504,12 +507,13 @@ def directory_login( *, base_url: str, admin_password: str, + admin_username: str = "admin", opener: Callable = urlopen, timeout: float = 10, ) -> str: request = Request( base_url.rstrip("/") + "/auth/simple/login", - data=json.dumps({"username": "admin", "password": admin_password}).encode(), + data=json.dumps({"username": admin_username, "password": admin_password}).encode(), headers={"Content-Type": "application/json"}, method="POST", ) @@ -618,12 +622,16 @@ def dispatch( raise KeyError(path) +# Human, service-account and service-admin names (Service Identity standard naming): never self-chosen. +RESERVED_USERNAMES = {"admin", "administrator", "platform-root", "root", "system"} + + def _username(email: str, preferred: object = None) -> str: if preferred is not None: value = str(preferred).strip().lower() if not re.fullmatch(r"[a-z][a-z0-9._-]{2,31}", value): raise ValueError("preferred_username is invalid") - if value in {"admin", "administrator", "platform-root", "root", "system"}: + if value in RESERVED_USERNAMES or value.startswith("svc-") or value.endswith("-admin"): raise ValueError("preferred_username is reserved") return value local = email.partition("@")[0].lower() diff --git a/identity-provisioner/server.py b/identity-provisioner/server.py index 8d43db8..422ead5 100644 --- a/identity-provisioner/server.py +++ b/identity-provisioner/server.py @@ -126,6 +126,7 @@ def main(): Handler.provisioner = LLDAPProvisioner( base_url=os.environ["LLDAP_URL"], admin_password=os.environ["LLDAP_ADMIN_PASSWORD"], + admin_username=os.environ.get("LLDAP_ADMIN_USERNAME", "admin"), group_provenance=configured_group_provenance(), ) Handler.service_token = os.environ["PROVISIONER_SERVICE_TOKEN"].strip() @@ -137,6 +138,7 @@ def main(): setter=LLDAPPasswordSetter( base_url=os.environ["LLDAP_URL"], admin_password=os.environ["LLDAP_ADMIN_PASSWORD"], + admin_username=os.environ.get("LLDAP_ADMIN_USERNAME", "admin"), ), tenant_returns=json.loads(os.environ.get("PASSWORD_SETUP_TENANT_RETURNS", "{}")), ttl_seconds=int(os.environ.get("PASSWORD_SETUP_TTL_SECONDS", "900")), diff --git a/identity-provisioner/tests/test_directory_contract.py b/identity-provisioner/tests/test_directory_contract.py index 00491ca..e7c9dca 100644 --- a/identity-provisioner/tests/test_directory_contract.py +++ b/identity-provisioner/tests/test_directory_contract.py @@ -279,6 +279,17 @@ class DirectoryContractTests(unittest.TestCase): class DirectoryLoginMappingTests(unittest.TestCase): + def test_login_uses_the_configured_username_and_defaults_to_admin(self): + seen = [] + + def opener(request, timeout=10): + seen.append(json.loads(request.data)["username"]) + return io.BytesIO(json.dumps({"token": "t"}).encode()) + + directory_login(base_url="http://directory", admin_password=SECRET, opener=opener) + directory_login(base_url="http://directory", admin_password=SECRET, admin_username="svc-provisioner", opener=opener) + self.assertEqual(["admin", "svc-provisioner"], seen) + def test_http_401_becomes_auth_rejected_without_body_or_cause(self): def opener(request, timeout=10): raise HTTPError( diff --git a/identity-provisioner/tests/test_provisioner.py b/identity-provisioner/tests/test_provisioner.py index fef0caa..3644d5d 100644 --- a/identity-provisioner/tests/test_provisioner.py +++ b/identity-provisioner/tests/test_provisioner.py @@ -44,6 +44,9 @@ class ProvisionerTests(unittest.TestCase): _username("other@example.test", "not allowed!") with self.assertRaisesRegex(ValueError, "reserved"): _username("other@example.test", "admin") + for name in ("svc-authelia", "ld-admin", "pi-admin"): + with self.assertRaisesRegex(ValueError, "reserved"): + _username("other@example.test", name) def test_dispatch_requires_idempotency(self): with self.assertRaisesRegex(ValueError, "idempotency_key"):