diff --git a/identity-provisioner/provisioner.py b/identity-provisioner/provisioner.py index 6a8a280..e2d6bcd 100644 --- a/identity-provisioner/provisioner.py +++ b/identity-provisioner/provisioner.py @@ -36,7 +36,7 @@ class LLDAPProvisioner: def provision(self, payload: dict[str, Any]) -> Result: _required(payload, "user_id", "tenant", "primary_email", "idempotency_key", "correlation_id") email = str(payload["primary_email"]).strip().lower() - username = _username(email) + username = _username(email, payload.get("preferred_username")) token = self._login() users, groups = self._directory(token) existing = next((user for user in users if user.get("id") == username), None) @@ -259,7 +259,14 @@ def dispatch( raise KeyError(path) -def _username(email: str) -> str: +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"}: + raise ValueError("preferred_username is reserved") + return value local = email.partition("@")[0].lower() value = re.sub(r"[^a-z0-9._-]+", "-", local).strip("-") if not value or "@" not in email: diff --git a/identity-provisioner/tests/test_provisioner.py b/identity-provisioner/tests/test_provisioner.py index f904ff6..2957a12 100644 --- a/identity-provisioner/tests/test_provisioner.py +++ b/identity-provisioner/tests/test_provisioner.py @@ -19,6 +19,13 @@ class ProvisionerTests(unittest.TestCase): def test_username_is_stable_and_sanitized(self): self.assertEqual("bernd.worsch", _username("Bernd.Worsch@binky-hedgehog.com")) + def test_preferred_username_is_honored_and_validated(self): + self.assertEqual("chosen.name", _username("other@example.test", "Chosen.Name")) + with self.assertRaisesRegex(ValueError, "invalid"): + _username("other@example.test", "not allowed!") + with self.assertRaisesRegex(ValueError, "reserved"): + _username("other@example.test", "admin") + def test_dispatch_requires_idempotency(self): with self.assertRaisesRegex(ValueError, "idempotency_key"): dispatch(Fake(), "/v1/identities/suspend", {