coulomb/net-kingdom
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Align provisioned identities with OIDC subjects
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details

Browse source
This commit is contained in:
tegwick 2026-08-14 18:53:37 +02:00
parent cbbd66697f
commit 2a2d8ba055
3 changed files with 16 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

15
identity-provisioner/provisioner.py
View file

@ -68,9 +68,10 @@ mutation CreateUser($id: String!, $email: String!, $display: String!) {
except Exception:
pass
raise
return Result("netkingdom-lldap", username, "password_setup_required", resumed)
return Result("netkingdom-lldap", _oidc_subject(username), "password_setup_required", resumed)
def suspend(self, subject: str) -> Result:
subject = _directory_username(subject)
token = self._login()
_, groups = self._directory(token)
group_id = self._ensure_group(token, groups, "netkingdom-suspended")
@ -78,6 +79,7 @@ mutation CreateUser($id: String!, $email: String!, $display: String!) {
return Result("netkingdom-lldap", subject, "suspended", False)
def reactivate(self, subject: str) -> Result:
subject = _directory_username(subject)
token = self._login()
_, groups = self._directory(token)
group = next((item for item in groups if item.get("displayName") == "netkingdom-suspended"), None)
@ -89,6 +91,7 @@ mutation Remove($userId: String!, $groupId: Int!) {
return Result("netkingdom-lldap", subject, "active", False)
def deprovision(self, subject: str) -> Result:
subject = _directory_username(subject)
token = self._login()
if self._user(token, subject) is None:
return Result("netkingdom-lldap", subject, "deprovisioned", True)
@ -249,7 +252,7 @@ def dispatch(
if path == "/v1/identities/reconcile":
return provisioner.reconcile(payload)
_required(payload, "external_subject", "idempotency_key", "correlation_id")
subject = str(payload["external_subject"])
subject = _directory_username(str(payload["external_subject"]))
if path == "/v1/identities/suspend":
return provisioner.suspend(subject)
if path == "/v1/identities/reactivate":
@ -305,3 +308,11 @@ def _desired(payload: dict[str, Any]) -> tuple[str, str, set[str], str]:
if desired_status == "suspended":
groups.add("netkingdom-suspended")
return subject, email, groups, desired_status
def _directory_username(subject: str) -> str:
return subject[4:].split(",", 1)[0] if subject.startswith("uid=") else subject
def _oidc_subject(username: str) -> str:
return f"uid={username},ou=people,dc=netkingdom,dc=local"