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 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 308701@bnt-lap001
Assistant-Session: 5548e447-444c-4dcc-bed1-05c16eebfe56
This commit is contained in:
tegwick 2026-10-28 20:11:04 +01:00
parent cdf7aa0c77
commit 2ebc4f0ae8
5 changed files with 29 additions and 2 deletions

View file

@ -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,
)

View file

@ -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()

View file

@ -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")),

View file

@ -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(

View file

@ -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"):