NK-WP-0056-T09: lane looks up the privacyIDEA token by serial; name login failures

The first founder setup run stopped in phase absent (no change made):
listing tokens by a user that does not resolve yet is an error in
privacyIDEA. Ownership is now read from the token found by serial, and
pi-admin login, privacyIDEA requests and the LLDAP admin login refuse with
their own reasons. Failed receipt kept as 00-setup-refused-absent.json.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 994874@bnt-lap001
Assistant-Session: 6017c416-d619-47eb-bd1a-f670a02d7d87

docs/evidence/2026-10-31-nk56-t09-synthetic-identity/00-setup-refused-absent.json

@@ -0,0 +1,9 @@
+{
+  "action": "setup",
+  "identity": "syn-journey",
+  "phase": "absent",
+  "reason": "Refused",
+  "source_sha256": "7e9c9e7508cecc689a72ed523fe9a1cc3eb7852ac5593bb192379dee8cd5fa44",
+  "status": "failed",
+  "values_recorded": false
+}

sso-mfa/k8s/synthetic-journey/lane.py

@@ -134,14 +134,14 @@ def lldap_user(d, token) -> dict | None:
 
 
 def pi_tokens(d) -> list[dict]:
-    value = d.pi_call(f"/token/?user={USER}&realm={REALM}")
+    """The lane's token, by serial. Listing by user is an error in privacyIDEA while the user does
+    not resolve (before setup creates it, after teardown deletes it), so ownership is read from the
+    token itself instead."""
+    value = d.pi_call(f"/token/?serial={SERIAL}")
     tokens = value.get("tokens") if isinstance(value, dict) else None
     if not isinstance(tokens, list):
         raise Refused("pi_token_list")
-    by_serial = d.pi_call(f"/token/?serial={SERIAL}")
-    if not isinstance(by_serial, dict) or not isinstance(by_serial.get("tokens"), list):
-        raise Refused("pi_token_list")
-    return tokens + [t for t in by_serial["tokens"] if t not in tokens]
+    return tokens
 
 
 def current_code(seed_b32: str, wall) -> str:
@@ -194,6 +194,7 @@ def setup(d, expected: str, receipt: dict) -> None:
         receipt["enrollment_confirmed_with_code"] = True
     tokens = pi_tokens(d)
     if (len(tokens) != 1 or tokens[0].get("serial") != SERIAL or tokens[0].get("active") is not True
+            or tokens[0].get("username") != USER or tokens[0].get("user_realm", REALM) != REALM
             or tokens[0].get("tokentype") != "totp" or tokens[0].get("rollout_state") not in ("", "enrolled", None)):
         raise Refused("pi_token_shape")
 
@@ -243,7 +244,7 @@ def teardown(d, expected: str, receipt: dict) -> None:
     receipt["phase"] = "pi_token"
     d.pi_login()
     for token in pi_tokens(d):
-        if token.get("serial") != SERIAL:
+        if token.get("serial") != SERIAL or token.get("username") not in (USER, "", None):
             raise Refused("unexpected_pi_token")  # never delete what this lane did not create
         d.pi_call(f"/token/{SERIAL}", method="DELETE")
     receipt["phase"] = "lldap_user"
@@ -317,7 +318,10 @@ class Live:
         return self._bao(["read", "-format=json", role], allow_failure=True).returncode == 0
 
     def lldap_admin_token(self):
-        token = self._split.admin_token(self.directory, None)
+        try:
+            token = self._split.admin_token(self.directory, None)
+        except Exception:
+            raise Refused("lldap_admin_login") from None
         if not token:
             raise Refused("lldap_admin_login")
         return token
@@ -333,12 +337,19 @@ class Live:
 
     def pi_login(self):
         if self.pi is None:
-            self.pi = self._realm.Client(self.pi_endpoint, getpass.getpass("privacyIDEA pi-admin password: "))
+            secret = getpass.getpass("privacyIDEA pi-admin password: ")
+            try:
+                self.pi = self._realm.Client(self.pi_endpoint, secret)
+            except Exception:
+                raise Refused("pi_admin_login") from None
 
     def pi_call(self, path, payload=None, method=None):
         self.pi_login()
         if method is None:
-            return self.pi.call(path, payload)
+            try:
+                return self.pi.call(path, payload)
+            except Exception:
+                raise Refused("pi_request:" + path.split("?")[0]) from None
         status, body = self._pi_api.request(self.pi.url + path, payload, self.pi.token, method=method)
         result = body.get("result") if isinstance(body, dict) else None
         if status != 200 or not isinstance(result, dict) or result.get("status") is not True:

sso-mfa/k8s/synthetic-journey/test_lane.py

@@ -126,10 +126,10 @@ class Fake:
             return {"rollout_state": self.tokens[payload["serial"]]["rollout_state"]}
         if path.startswith("/token/?"):
             public = [{k: v for k, v in t.items() if not k.startswith("_")} for t in self.tokens.values()]
-            if "serial=" in path:
-                serial = path.split("serial=")[1]
-                return {"tokens": [t for t in public if t["serial"] == serial]}
-            return {"tokens": [t for t in public if t["username"] == lane.USER]}
+            if "user=" in path and lane.USER not in self.users:
+                raise lane.Refused("pi_request:/token/")  # privacyIDEA: the user does not resolve
+            serial = path.split("serial=")[1]
+            return {"tokens": [t for t in public if t["serial"] == serial]}
         raise AssertionError(path)
 
     def pi_rollout_state(self, serial):
@@ -193,7 +193,7 @@ class LaneTests(unittest.TestCase):
             "attended_platform_admin_required": lambda d: setattr(d, "attended_ok", False),
             "kv_exists": lambda d: setattr(d, "kv", {"PASSWORD": "x", "TOTP_SEED": "y"}),
             "lldap_user_exists": lambda d: d.users.update({lane.USER: {"id": lane.USER, "groups": []}}),
-            "pi_token_exists": lambda d: d.tokens.update({lane.SERIAL: {"serial": lane.SERIAL, "username": lane.USER}}),
+            "pi_token_exists": lambda d: d.tokens.update({lane.SERIAL: {"serial": lane.SERIAL, "username": ""}}),
             "ccr_status": lambda d: d.ccr.update(status="proposed"),
             "ccr_approval": lambda d: d.ccr["review"]["comments"].pop(),
             "ccr_shape": lambda d: d.ccr["openbao"].update(runtime_access=True),