diff --git a/docs/reuse-surface-hub-writer-attended.md b/docs/reuse-surface-hub-writer-attended.md index 37efc52..c14450a 100644 --- a/docs/reuse-surface-hub-writer-attended.md +++ b/docs/reuse-surface-hub-writer-attended.md @@ -15,16 +15,30 @@ shape go through identity-provisioner `POST /v1/groups/ensure`; see `docs/directory-group-binding.md`. -The script leaves `~/.vault-token` in place. It calls `sys/capabilities` with -the first local token that is allowed to: `BAO_TOKEN` or `VAULT_TOKEN`, then -`~/.local/openbao/platform-admin.token`, then `~/.vault-token`. The lane token -is not printed and is revoked before the script exits. A passing receipt is -the sanitized verification record. +The script leaves `~/.vault-token` in place and reads no operator or root +token. Since NK-WP-0051-T06 it asks `sys/capabilities-self` with the lane +token itself, which needs railiance-platform CCR-2026-0045 applied (the +generated `sys/capabilities-self` grant). Until that apply, `member` fails +with the OpenBao permission error in `openbao_error`. The lane token is not +printed and is revoked before the script exits. A passing receipt is the +sanitized verification record. + +`bao login` runs with `skip_browser=true`: it prints the sign-in address and +waits, so the default browser and its Authelia session are not used. Open the +address in a private window, or set `OPS_MASON_LOGIN_BROWSER` to a Chromium or +Firefox binary to open it in a fresh temporary profile. + +The receipt follows `net-kingdom/docs/sign-in-diagnosis.md`: `uid`, +`openbao_error` (OpenBao's text, verbatim, with tokens, codes and emails +redacted) and `assurance` when known. Pass `--uid` and `--assurance` as +KeyCape's diagnosis page showed them for this request; otherwise `uid` comes +from the token display name on success, and `assurance=aal2` follows from the +role's aal2 bound claim on success. No token and no KV value is recorded. ```bash export BAO_ADDR=http://127.0.0.1:18200 -python3 scripts/verify-reuse-surface-hub-writer.py member -python3 scripts/verify-reuse-surface-hub-writer.py non-member +python3 scripts/verify-reuse-surface-hub-writer.py member --uid bernd.worsch-3 +python3 scripts/verify-reuse-surface-hub-writer.py non-member --uid signin-probe-negative --assurance aal2 ``` `member` passes when the token's policies are exactly @@ -33,8 +47,8 @@ the runtime-secrets data and metadata paths are `read`, the parent paths and `platform/data/workloads/issue-core/issue-core/issue-core-runtime` are `deny`, and revoke succeeded. -`non-member` passes when the role refuses the login. Use a private window so -an existing KeyCape session is not reused. +`non-member` passes when the role refuses the login and the refusal text is +recorded in `openbao_error`. Do not run `bao kv get` on this path to prove the lane. Do not attach the workload policy to the admin entity. diff --git a/plans/reuse-surface-hub-writer.md b/plans/reuse-surface-hub-writer.md index 13e48eb..8dbe863 100644 --- a/plans/reuse-surface-hub-writer.md +++ b/plans/reuse-surface-hub-writer.md @@ -127,6 +127,10 @@ stanza. The verifier therefore asks `sys/capabilities` with the operator token and puts the lane token only in a mode-0600 body file that it deletes. The receipt never contains the token. +Superseded by NK-WP-0051-T06 (MASON-WP-0012): railiance-platform now +generates the grant from `openbao.token_capabilities_self` (CCR-2026-0045), +and the verifier calls `sys/capabilities-self` with the lane token only. + Catalog `reuse-surface-hub-write-token` stays active. This plan does not ask ops-warden to vend the value. After both verifier modes pass, ops-warden can name the role in `auth_method`. Until then the entry's caller is unnamed diff --git a/scripts/verify-reuse-surface-hub-writer.py b/scripts/verify-reuse-surface-hub-writer.py index 313c30d..41433b1 100644 --- a/scripts/verify-reuse-surface-hub-writer.py +++ b/scripts/verify-reuse-surface-hub-writer.py @@ -1,58 +1,69 @@ #!/usr/bin/env python3 """Attended proof of the reuse-surface hub-writer caller. - ./scripts/verify-reuse-surface-hub-writer.py member - ./scripts/verify-reuse-surface-hub-writer.py non-member + ./scripts/verify-reuse-surface-hub-writer.py member [--uid UID] [--assurance aal1|aal2] + ./scripts/verify-reuse-surface-hub-writer.py non-member [--uid UID] [--assurance aal1|aal2] Logs in on auth/netkingdom role reuse-surface-hub-writer-workload-kv-read with -``bao login -token-only``, so the operator's token helper is left alone. The -lane token stays in memory. Capability queries use the first local token that -may call ``sys/capabilities`` (``BAO_TOKEN`` or ``VAULT_TOKEN``, then -``~/.local/openbao/platform-admin.token``, then ``~/.vault-token``), because -the workload policy has no sys/capabilities-self grant. The receipt records -policies, entity id, path capabilities, and pass/fail. It does not record the -token or any KV value. +``bao login -token-only ... skip_browser=true``: bao prints the sign-in address +and waits, so the default browser's Authelia session is never reused. Open the +address in a private window, or set ``OPS_MASON_LOGIN_BROWSER`` to a Chromium +or Firefox binary and the script opens it in a fresh temporary profile. The +operator's token helper is left alone and the lane token stays in memory. -Sign in with a private browser window. A reused platform-admin session is the -wrong principal for ``member``. +Capabilities come from ``sys/capabilities-self`` called with the lane token +itself (CCR-2026-0045). No operator or root token is read or used. + +The receipt follows ``net-kingdom/docs/sign-in-diagnosis.md``: ``uid``, +``openbao_error`` (verbatim, redacted of tokens, codes and emails) and +``assurance`` when known, plus policies, entity id, path capabilities and +pass/fail. ``--uid`` and ``--assurance`` are what KeyCape's diagnosis page +showed for this request. It never records a token or any KV value. """ from __future__ import annotations +import argparse import json import os -import re -import stat +import shutil import subprocess import sys import tempfile +import threading from datetime import datetime, timezone from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) +from ops_mason.attended_login import ( + TOKEN_RE, + classify_login_line, + fresh_profile_argv, + login_args, + openbao_error_text, + redact, + resolve_assurance, + resolve_uid, +) from ops_mason.reuse_surface_hub_writer import ( DATA, METADATA, + MOUNT, MUST_DENY, PARENT_DATA, PARENT_METADATA, POLICY, ROLE, + ROLE_REQUIRES_AAL2, SIBLING, assess, ) BAO_ADDR = os.environ.get("BAO_ADDR", "http://127.0.0.1:18200") -# This OpenBao issues s. service tokens. hvs/hvb/hvr are the newer Vault shapes. -TOKEN_RE = re.compile(r"\b(?:hvs|hvb|hvr|s)\.[A-Za-z0-9_-]{8,}\b") PATHS = [DATA, METADATA, PARENT_DATA, PARENT_METADATA, SIBLING] -def redact(text: str) -> str: - return TOKEN_RE.sub("[redacted-token]", text or "") - - def receipts_dir() -> Path: home = Path(os.environ.get("HOME", str(Path.home()))) path = home / ".local/state/railiance/receipts" @@ -60,48 +71,6 @@ def receipts_dir() -> Path: return path -def token_candidates() -> list[tuple[str, str]]: - found: list[tuple[str, str]] = [] - for name in ("BAO_TOKEN", "VAULT_TOKEN"): - value = os.environ.get(name, "").strip() - if value: - found.append((name, value)) - home = Path(os.environ.get("HOME", str(Path.home()))) - for label, path in ( - ("~/.local/openbao/platform-admin.token", home / ".local/openbao/platform-admin.token"), - ("~/.vault-token", home / ".vault-token"), - ): - if path.is_file(): - value = path.read_text(encoding="utf-8").strip() - if value: - found.append((label, value)) - return found - - -def write_capabilities(caller: str, subject: str, paths: list[str]) -> subprocess.CompletedProcess[str]: - body = json.dumps({"token": subject, "paths": paths}) - tmp = tempfile.NamedTemporaryFile("w", prefix="rs-caps-", delete=False) - try: - os.fchmod(tmp.fileno(), stat.S_IRUSR | stat.S_IWUSR) - tmp.write(body) - tmp.close() - return run_bao(["write", "-format=json", "sys/capabilities", f"@{tmp.name}"], caller) - finally: - try: - os.remove(tmp.name) - except OSError: - pass - - -def operator_token() -> str: - """Return a token allowed to call sys/capabilities, or an empty string.""" - for _label, token in token_candidates(): - probe = write_capabilities(token, token, ["auth/token/lookup-self"]) - if probe.returncode == 0: - return token - return "" - - def run_bao(args: list[str], token: str, stdin: str | None = None) -> subprocess.CompletedProcess[str]: env = os.environ.copy() env["BAO_ADDR"] = BAO_ADDR @@ -117,16 +86,67 @@ def run_bao(args: list[str], token: str, stdin: str | None = None) -> subprocess ) -def login() -> str: - proc = run_bao( - ["login", "-token-only", "-method=oidc", "-path=netkingdom", f"role={ROLE}"], - token="", +def open_fresh_profile(url: str, profile_dirs: list[str]) -> None: + browser = os.environ.get("OPS_MASON_LOGIN_BROWSER", "").strip() + if not browser: + return + resolved = shutil.which(browser) or browser + profile = tempfile.mkdtemp(prefix="ops-mason-login-profile-") + profile_dirs.append(profile) + try: + subprocess.Popen( + fresh_profile_argv(resolved, Path(profile), url), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + sys.stderr.write("Opened in a fresh browser profile.\n") + except OSError as exc: + sys.stderr.write(f"Could not start {browser}: {exc}. Open the address in a private window.\n") + + +def login() -> tuple[str, str]: + """Return (lane token or "", OpenBao error text or "").""" + env = os.environ.copy() + env["BAO_ADDR"] = BAO_ADDR + env.pop("BAO_TOKEN", None) + env.pop("VAULT_TOKEN", None) + proc = subprocess.Popen( + ["bao", *login_args(MOUNT, ROLE)], + text=True, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, ) - token = (proc.stdout or "").strip() + kept: list[str] = [] + profile_dirs: list[str] = [] + + def pump() -> None: + assert proc.stderr is not None + for line in proc.stderr: + kind = classify_login_line(line) + if kind == "url": + sys.stderr.write("\nOpen this address in a private browser window:\n\n") + sys.stderr.write(line.strip() + "\n\n") + open_fresh_profile(line.strip(), profile_dirs) + elif kind == "instruction": + if line.strip(): + sys.stderr.write(line.strip() + "\n") + else: + kept.append(line) + sys.stderr.write(redact(line)) + + reader = threading.Thread(target=pump, daemon=True) + reader.start() + token = (proc.stdout.read() if proc.stdout else "").strip() + proc.wait() + reader.join(timeout=5) + for profile in profile_dirs: + shutil.rmtree(profile, ignore_errors=True) if proc.returncode != 0 or TOKEN_RE.fullmatch(token) is None: - sys.stderr.write(redact(proc.stderr or "login output was not a token\n")) - return "" - return token + return "", openbao_error_text(kept) or "login output was not a token" + return token, "" def lookup(token: str) -> dict: @@ -137,15 +157,14 @@ def lookup(token: str) -> dict: return payload.get("data") or payload -def capabilities(operator: str, lane: str) -> dict[str, list[str]]: - if not operator: - raise RuntimeError( - "no token can call sys/capabilities " - "(BAO_TOKEN, VAULT_TOKEN, ~/.local/openbao/platform-admin.token, or ~/.vault-token)" - ) - proc = write_capabilities(operator, lane, PATHS) +def capabilities(lane: str) -> dict[str, list[str]]: + """The lane token's own capabilities, from sys/capabilities-self.""" + proc = run_bao( + ["write", "-format=json", "sys/capabilities-self", "paths=" + ",".join(PATHS)], + lane, + ) if proc.returncode != 0: - raise RuntimeError(redact(proc.stderr or "sys/capabilities failed")) + raise RuntimeError(redact(proc.stderr or "sys/capabilities-self failed").strip()) payload = json.loads(proc.stdout) data = payload.get("data") or payload found: dict[str, list[str]] = {} @@ -168,57 +187,30 @@ def write_receipt(name: str, lines: list[str]) -> Path: return path -def main(argv: list[str]) -> int: - mode = argv[1] if len(argv) > 1 else "" - if mode not in {"member", "non-member"}: - sys.stderr.write("usage: verify-reuse-surface-hub-writer.py member|non-member\n") - return 2 - operator = operator_token() - if not operator: - sys.stderr.write( - "no token can call sys/capabilities. Checked BAO_TOKEN, VAULT_TOKEN, " - "~/.local/openbao/platform-admin.token, and ~/.vault-token.\n" - ) - return 1 - sys.stderr.write( - f"Sign in as the {mode} identity in a private browser window.\n" - "A platform-admin session is the wrong principal for member.\n" +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(prog="verify-reuse-surface-hub-writer.py") + parser.add_argument("mode", choices=["member", "non-member"]) + parser.add_argument("--uid", default="", help="uid KeyCape showed for this request") + parser.add_argument( + "--assurance", default="", choices=["", "aal1", "aal2"], + help="assurance KeyCape showed for this request", ) - now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - lane = login() - report: dict = { - "mode": mode, - "login": "ok" if lane else "refused", - "policies": [], - "identity_policies": [], - "entity_id": "", - "capabilities": {}, - "revoked": None, - } - try: - if lane: - info = lookup(lane) - report["policies"] = list(info.get("policies") or []) - report["identity_policies"] = list(info.get("identity_policies") or []) - report["entity_id"] = str(info.get("entity_id") or "") - if mode == "member": - report["capabilities"] = capabilities(operator, lane) - except (RuntimeError, json.JSONDecodeError, OSError) as exc: - report["error"] = redact(str(exc)) - finally: - if lane: - report["revoked"] = revoke(lane) - lane = "" - failures = assess(report) - if report.get("error"): - failures = [report["error"], *failures] + return parser.parse_args(argv[1:]) + + +def receipt_lines(report: dict, now: str, failures: list[str]) -> list[str]: lines = [ - f"mode={mode}", + f"mode={report['mode']}", f"at={now}", f"addr={BAO_ADDR}", f"role={ROLE}", f"policy={POLICY}", + f"uid={report['uid']}", + f"uid_source={report['uid_source']}", + f"assurance={report['assurance']}", + f"assurance_source={report['assurance_source']}", f"login={report['login']}", + f"openbao_error={report['openbao_error']}", f"entity_id={report['entity_id']}", "policies=" + ",".join(report["policies"]), "identity_policies=" + ",".join(report["identity_policies"]), @@ -230,8 +222,59 @@ def main(argv: list[str]) -> int: lines.append("revoked=" + ("yes" if report["revoked"] else "no" if report["revoked"] is False else "n/a")) lines.append("result=" + ("pass" if not failures else "fail")) for item in failures: - lines.append("failure=" + item) - receipt = write_receipt(f"reuse-surface-hub-writer-{mode}.txt", lines) + lines.append("failure=" + redact(item)) + return [redact(line) for line in lines] + + +def main(argv: list[str]) -> int: + args = parse_args(argv) + mode = args.mode + try: + resolve_uid(args.uid, "", MOUNT) + except ValueError as exc: + sys.stderr.write(f"{exc}\n") + return 2 + sys.stderr.write( + f"Sign in as the {mode} identity. bao prints the address and waits;\n" + "open it in a private window. A platform-admin session is the wrong principal for member.\n" + ) + now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + lane, login_error = login() + report: dict = { + "mode": mode, + "login": "ok" if lane else "refused", + "openbao_error": login_error, + "policies": [], + "identity_policies": [], + "entity_id": "", + "display_name": "", + "capabilities": {}, + "revoked": None, + } + try: + if lane: + info = lookup(lane) + report["policies"] = list(info.get("policies") or []) + report["identity_policies"] = list(info.get("identity_policies") or []) + report["entity_id"] = str(info.get("entity_id") or "") + report["display_name"] = str(info.get("display_name") or "") + if mode == "member": + report["capabilities"] = capabilities(lane) + except (RuntimeError, json.JSONDecodeError, OSError) as exc: + report["error"] = redact(str(exc)) + report["openbao_error"] = report["openbao_error"] or report["error"] + finally: + if lane: + report["revoked"] = revoke(lane) + lane = "" + report["uid"], report["uid_source"] = resolve_uid(args.uid, report.pop("display_name"), MOUNT) + report["assurance"], report["assurance_source"] = resolve_assurance( + args.assurance, report["login"] == "ok", ROLE_REQUIRES_AAL2 + ) + failures = assess(report) + if report.get("error"): + failures = [report["error"], *failures] + receipt = write_receipt(f"reuse-surface-hub-writer-{mode}.txt", receipt_lines(report, now, failures)) sys.stderr.write(f"receipt: {receipt}\n") return 0 if not failures else 1 diff --git a/src/ops_mason/attended_login.py b/src/ops_mason/attended_login.py new file mode 100644 index 0000000..8551309 --- /dev/null +++ b/src/ops_mason/attended_login.py @@ -0,0 +1,139 @@ +"""Attended ``bao login`` for a named OIDC role, and its diagnosis receipt. + +NK-WP-0051-T06, contract ``net-kingdom/docs/sign-in-diagnosis.md``: + +- A named-role login must not reuse the default Authelia session. The login + runs with ``skip_browser=true``: bao prints the address and waits. If + ``OPS_MASON_LOGIN_BROWSER`` names a Chromium-family or Firefox binary, that + address opens in a fresh, temporary browser profile instead. +- The receipt records ``uid``, ``openbao_error`` (OpenBao's error text, + verbatim) and ``assurance`` when known. Never a token, an authorization + code, an email address, a one-time password, or a KV value. + +Pure helpers live here so they can be tested without OpenBao. +""" + +from __future__ import annotations + +import re +from pathlib import Path + +# This OpenBao issues s. service tokens. hvs/hvb/hvr are the newer Vault shapes. +TOKEN_RE = re.compile(r"\b(?:hvs|hvb|hvr|s)\.[A-Za-z0-9_-]{8,}\b") +CODE_RE = re.compile(r"(?i)\b(code|id_token|access_token|refresh_token)=[^&\s\"']+") +EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}") +UID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") +ASSURANCE_VALUES = ("aal1", "aal2") + +# Lines bao prints around the browser step. They are shown to the operator and +# never kept as error text. +_INSTRUCTION_MARKERS = ( + "complete the login via your oidc provider", + "launching browser to", + "open the following link", + "waiting for oidc authentication to complete", +) + + +def redact(text: str) -> str: + """Remove tokens, authorization codes and email addresses from text.""" + text = TOKEN_RE.sub("[redacted-token]", text or "") + text = CODE_RE.sub(lambda m: f"{m.group(1)}=[redacted]", text) + return EMAIL_RE.sub("[redacted-email]", text) + + +def login_args(path: str, role: str) -> list[str]: + """bao arguments for a named-role login that prints the address and waits.""" + return [ + "login", + "-token-only", + "-method=oidc", + f"-path={path}", + f"role={role}", + "skip_browser=true", + ] + + +def is_auth_url(line: str) -> bool: + stripped = line.strip() + return stripped.startswith(("https://", "http://")) and ( + "client_id=" in stripped or "/authorize" in stripped + ) + + +def classify_login_line(line: str) -> str: + """Return ``url``, ``instruction`` or ``error`` for one bao stderr line.""" + if is_auth_url(line): + return "url" + lowered = line.strip().lower() + if not lowered or any(marker in lowered for marker in _INSTRUCTION_MARKERS): + return "instruction" + return "error" + + +def openbao_error_text(lines: list[str]) -> str: + """OpenBao's error text from the kept stderr lines, verbatim but redacted. + + Lines are joined with `` | `` so the receipt stays one field per line. + """ + kept = [line.strip() for line in lines if line.strip()] + return redact(" | ".join(kept)) + + +def fresh_profile_argv(browser: str, profile_dir: Path, url: str) -> list[str]: + """Command that opens ``url`` in a new, empty browser profile.""" + name = Path(browser).name.lower() + if "firefox" in name: + return [browser, "-no-remote", "-profile", str(profile_dir), "-new-instance", url] + return [ + browser, + f"--user-data-dir={profile_dir}", + "--no-first-run", + "--no-default-browser-check", + "--new-window", + url, + ] + + +def uid_from_display_name(display_name: str, mount: str) -> str: + """The alias name OpenBao put in the token display name, if it is a uid. + + OpenBao sets ``display_name`` to ``-``. An empty + string is returned when the value is missing or looks like an email. + """ + value = (display_name or "").strip() + prefix = f"{mount}-" + if value.startswith(prefix): + value = value[len(prefix):] + if not value or "@" in value or not UID_RE.match(value): + return "" + return value + + +def resolve_uid(operator_uid: str, display_name: str, mount: str) -> tuple[str, str]: + """Return ``(uid, source)``. The operator's value is what KeyCape showed.""" + given = (operator_uid or "").strip() + if given: + if "@" in given or not UID_RE.match(given): + raise ValueError("--uid must be a directory uid, not an email or free text") + return given, "operator" + derived = uid_from_display_name(display_name, mount) + if derived: + return derived, "token-display-name" + return "", "unknown" + + +def resolve_assurance(operator_value: str, login_ok: bool, role_requires_aal2: bool) -> tuple[str, str]: + """Return ``(assurance, source)``; empty when not known. + + A successful login on a role bound to ``/assurance/level=aal2`` proves + this request carried aal2. A refusal proves nothing about assurance. + """ + given = (operator_value or "").strip() + if given: + if given not in ASSURANCE_VALUES: + raise ValueError("--assurance must be aal1 or aal2") + return given, "operator" + if login_ok and role_requires_aal2: + return "aal2", "role-bound-claim" + return "", "unknown" diff --git a/src/ops_mason/reuse_surface_hub_writer.py b/src/ops_mason/reuse_surface_hub_writer.py index 58ded33..8b21c57 100644 --- a/src/ops_mason/reuse_surface_hub_writer.py +++ b/src/ops_mason/reuse_surface_hub_writer.py @@ -8,6 +8,10 @@ from __future__ import annotations POLICY = "workload-kv-read-reuse-surface-runtime" ROLE = "reuse-surface-hub-writer-workload-kv-read" +MOUNT = "netkingdom" +# The role binds /assurance/level=aal2 (CCR-2026-0044), so a login that +# succeeds on it carried aal2 for that request. +ROLE_REQUIRES_AAL2 = True FORBIDDEN_ENTITY_ID = "4a1f663c-0621-0483-edd0-f852274f4b6b" FORBIDDEN_POLICIES = frozenset( { @@ -48,9 +52,11 @@ def assess(report: dict) -> list[str]: return [f"unknown mode {mode!r}"] login = report.get("login") if mode == "non-member": - if login == "refused": - return [] - return ["non-member login succeeded; the role accepted the wrong principal"] + if login != "refused": + return ["non-member login succeeded; the role accepted the wrong principal"] + if not str(report.get("openbao_error") or "").strip(): + return ["non-member refusal recorded no openbao_error text"] + return [] failures: list[str] = [] if login != "ok": diff --git a/tests/test_attended_login.py b/tests/test_attended_login.py new file mode 100644 index 0000000..6fe2046 --- /dev/null +++ b/tests/test_attended_login.py @@ -0,0 +1,181 @@ +"""NK-WP-0051-T06: attended named-role login and its diagnosis receipt.""" + +from __future__ import annotations + +import importlib.util +import json +import os +import stat +import sys +from pathlib import Path + +import pytest + +from ops_mason.attended_login import ( + classify_login_line, + fresh_profile_argv, + login_args, + openbao_error_text, + redact, + resolve_assurance, + resolve_uid, + uid_from_display_name, +) + +REPO = Path(__file__).resolve().parents[1] +SCRIPT = REPO / "scripts/verify-reuse-surface-hub-writer.py" +FAKE_TOKEN = "s.FAKElanetoken0123456789" +REFUSAL = ( + "Error authenticating: Error making API request. Code: 400. Errors: " + '* claim "groups" does not match any associated bound claim values' +) +AUTH_URL = ( + "https://kc.coulomb.social/authorize?client_id=openbao-admin&nonce=n1" + "&redirect_uri=http%3A%2F%2Flocalhost%3A8250%2Foidc%2Fcallback&state=st1" +) + + +def test_login_prints_address_and_waits_instead_of_launching_browser(): + args = login_args("netkingdom", "reuse-surface-hub-writer-workload-kv-read") + assert "skip_browser=true" in args + assert "-token-only" in args + assert "role=reuse-surface-hub-writer-workload-kv-read" in args + + +def test_fresh_profile_argv_never_uses_default_profile(tmp_path): + chromium = fresh_profile_argv("/usr/bin/chromium", tmp_path, AUTH_URL) + assert f"--user-data-dir={tmp_path}" in chromium and chromium[-1] == AUTH_URL + firefox = fresh_profile_argv("/usr/bin/firefox", tmp_path, AUTH_URL) + assert firefox[1:4] == ["-no-remote", "-profile", str(tmp_path)] + + +def test_stderr_classification_keeps_only_error_text(): + assert classify_login_line(" " + AUTH_URL) == "url" + assert classify_login_line("Complete the login via your OIDC provider.") == "instruction" + assert classify_login_line("Waiting for OIDC authentication to complete...") == "instruction" + assert classify_login_line(REFUSAL) == "error" + assert openbao_error_text(["", REFUSAL + "\n"]) == REFUSAL + + +def test_redact_removes_never_logged_values(): + text = f"token {FAKE_TOKEN} code=abc123&state=x mail bernd@example.org" + out = redact(text) + assert FAKE_TOKEN not in out and "abc123" not in out and "bernd@example.org" not in out + assert "state=x" in out + + +def test_uid_resolution(): + assert uid_from_display_name("netkingdom-bernd.worsch-3", "netkingdom") == "bernd.worsch-3" + assert uid_from_display_name("netkingdom-someone@example.org", "netkingdom") == "" + assert resolve_uid("signin-probe-negative", "", "netkingdom") == ("signin-probe-negative", "operator") + assert resolve_uid("", "netkingdom-bernd.worsch-3", "netkingdom") == ( + "bernd.worsch-3", "token-display-name") + assert resolve_uid("", "", "netkingdom") == ("", "unknown") + with pytest.raises(ValueError): + resolve_uid("someone@example.org", "", "netkingdom") + + +def test_assurance_resolution(): + assert resolve_assurance("", True, True) == ("aal2", "role-bound-claim") + assert resolve_assurance("", False, True) == ("", "unknown") + assert resolve_assurance("aal2", False, True) == ("aal2", "operator") + with pytest.raises(ValueError): + resolve_assurance("high", False, True) + + +FAKE_BAO = r"""#!/usr/bin/env python3 +import json, os, sys +args = sys.argv[1:] +mode = os.environ["FAKE_BAO_MODE"] +log = os.environ["FAKE_BAO_LOG"] +with open(log, "a") as fh: + fh.write(json.dumps({"args": args, "has_token": bool(os.environ.get("BAO_TOKEN"))}) + "\n") +if args[0] == "login": + sys.stderr.write("Complete the login via your OIDC provider. Open the following link in your browser:\n\n") + sys.stderr.write(" " + os.environ["FAKE_AUTH_URL"] + "\n\n") + sys.stderr.write("Waiting for OIDC authentication to complete...\n") + if mode == "member": + print(os.environ["FAKE_TOKEN"]) + sys.exit(0) + sys.stderr.write(os.environ["FAKE_REFUSAL"] + "\n") + sys.exit(2) +if args[:2] == ["token", "lookup"]: + print(json.dumps({"data": {"policies": ["workload-kv-read-reuse-surface-runtime"], + "identity_policies": [], "entity_id": "11111111-1111-1111-1111-111111111111", + "display_name": "netkingdom-bernd.worsch-3"}})) + sys.exit(0) +if args[:3] == ["write", "-format=json", "sys/capabilities-self"]: + paths = args[3].split("=", 1)[1].split(",") + data = {p: (["read"] if p.endswith("runtime-secrets") else ["deny"]) for p in paths} + print(json.dumps({"data": data})) + sys.exit(0) +if args[:3] == ["token", "revoke", "-self"]: + sys.exit(0) +sys.stderr.write("unexpected bao call\n") +sys.exit(9) +""" + + +def run_script(tmp_path, monkeypatch, mode, *extra): + bindir = tmp_path / "bin" + bindir.mkdir() + fake = bindir / "bao" + fake.write_text(FAKE_BAO) + fake.chmod(fake.stat().st_mode | stat.S_IEXEC) + home = tmp_path / "home" + home.mkdir() + log = tmp_path / "bao.log" + monkeypatch.setenv("PATH", f"{bindir}{os.pathsep}{os.environ['PATH']}") + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("BAO_TOKEN", "s.OPERATORtoken0123456789") + monkeypatch.setenv("FAKE_BAO_MODE", mode) + monkeypatch.setenv("FAKE_BAO_LOG", str(log)) + monkeypatch.setenv("FAKE_TOKEN", FAKE_TOKEN) + monkeypatch.setenv("FAKE_REFUSAL", REFUSAL) + monkeypatch.setenv("FAKE_AUTH_URL", AUTH_URL) + monkeypatch.delenv("OPS_MASON_LOGIN_BROWSER", raising=False) + spec = importlib.util.spec_from_file_location("verify_hub_writer", SCRIPT) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + code = module.main([str(SCRIPT), mode, *extra]) + receipt = home / f".local/state/railiance/receipts/reuse-surface-hub-writer-{mode}.txt" + calls = [json.loads(line) for line in log.read_text().splitlines()] + return code, receipt, calls + + +def test_member_receipt_uses_capabilities_self_and_records_diagnosis(tmp_path, monkeypatch): + code, receipt, calls = run_script(tmp_path, monkeypatch, "member") + text = receipt.read_text() + assert code == 0, text + assert stat.S_IMODE(receipt.stat().st_mode) == 0o600 + assert "uid=bernd.worsch-3\n" in text + assert "assurance=aal2\n" in text + assert "openbao_error=\n" in text + assert "result=pass" in text + assert FAKE_TOKEN not in text and "OPERATORtoken" not in text + assert "state=st1" not in text + assert not any(c["args"][:2] == ["write", "-format=json"] and "sys/capabilities" in c["args"] + for c in calls) + login = calls[0] + assert login["args"][0] == "login" and "skip_browser=true" in login["args"] + assert login["has_token"] is False + + +def test_non_member_receipt_records_verbatim_refusal(tmp_path, monkeypatch): + code, receipt, calls = run_script( + tmp_path, monkeypatch, "non-member", "--uid", "signin-probe-negative", "--assurance", "aal2") + text = receipt.read_text() + assert code == 0, text + assert "uid=signin-probe-negative\n" in text + assert "uid_source=operator\n" in text + assert "assurance=aal2\n" in text + assert f"openbao_error={REFUSAL}\n" in text + assert "login=refused" in text + assert [c["args"][0] for c in calls] == ["login"] + + +def test_email_uid_is_refused(tmp_path, monkeypatch): + spec = importlib.util.spec_from_file_location("verify_hub_writer_uid", SCRIPT) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + assert module.main([str(SCRIPT), "non-member", "--uid", "x@example.org"]) == 2 diff --git a/tests/test_reuse_surface_hub_writer.py b/tests/test_reuse_surface_hub_writer.py index 3669820..a6783a0 100644 --- a/tests/test_reuse_surface_hub_writer.py +++ b/tests/test_reuse_surface_hub_writer.py @@ -63,7 +63,11 @@ def test_missing_revoke_fails(): def test_non_member_refusal_passes_and_unexpected_login_fails(): - assert assess({"mode": "non-member", "login": "refused"}) == [] + refusal = 'claim "groups" does not match any associated bound claim values' + assert assess({"mode": "non-member", "login": "refused", "openbao_error": refusal}) == [] + assert assess({"mode": "non-member", "login": "refused"}) == [ + "non-member refusal recorded no openbao_error text" + ] assert assess({"mode": "non-member", "login": "ok"}) diff --git a/workplans/MASON-WP-0012-sign-in-diagnosis-caller-proof.md b/workplans/MASON-WP-0012-sign-in-diagnosis-caller-proof.md new file mode 100644 index 0000000..2ad106c --- /dev/null +++ b/workplans/MASON-WP-0012-sign-in-diagnosis-caller-proof.md @@ -0,0 +1,89 @@ +--- +id: MASON-WP-0012 +type: workplan +title: "Named-role login without session reuse, diagnosis receipt, and a self-only capability proof" +domain: infotech +repo: ops-mason +status: active +owner: claude +topic_slug: custodian +created: "2026-10-14" +updated: "2026-10-14" +--- + +# Named-role login without session reuse, diagnosis receipt, and a self-only capability proof + +## Goal + +Answer NetKingdom NK-WP-0051-T06 for ops-mason. Contract and the only source +of field names: `net-kingdom/docs/sign-in-diagnosis.md`. Background: +MASON-WP-0011. Paired railiance-platform work: RPF-WP-0057 (CCR-2026-0045 and +the OpenBao UI role choice). + +## Boundaries + +- `ops-mason-survey` stays read-only; this plan does not touch it. +- `workload-kv-read-reuse-surface-runtime` is never attached to entity + `4a1f663c-0621-0483-edd0-f852274f4b6b`. +- No `verbose_oidc_logging`. Receipts carry no token, authorization code, + email, OTP, or KV value. +- Repo changes and tests only; nothing applied to OpenBao or the cluster. + +## Named-role login prints the address and waits + +```task +id: MASON-WP-0012-T01 +status: done +priority: high +``` + +`scripts/verify-reuse-surface-hub-writer.py` logs in with +`skip_browser=true`, shows the address, and waits. `OPS_MASON_LOGIN_BROWSER` +opens it in a fresh temporary Chromium or Firefox profile instead. The default +Authelia session is never used. Helpers: `src/ops_mason/attended_login.py`. + +Done 2026-10-14. Tests: `tests/test_attended_login.py`. + +## Receipt records uid, openbao_error, assurance + +```task +id: MASON-WP-0012-T02 +status: done +priority: high +``` + +The receipt adds `uid`, `openbao_error` (verbatim, redacted of tokens, codes +and emails) and `assurance` when known, each with its source. `--uid` and +`--assurance` take what KeyCape's diagnosis page showed. A non-member pass now +needs the refusal text recorded. + +Done 2026-10-14. Tests: `tests/test_attended_login.py`, +`tests/test_reuse_surface_hub_writer.py`. + +## Proof uses sys/capabilities-self only + +```task +id: MASON-WP-0012-T03 +status: done +priority: high +``` + +The verifier no longer reads `BAO_TOKEN`, `VAULT_TOKEN`, +`~/.local/openbao/platform-admin.token` or `~/.vault-token` and never calls +`sys/capabilities`. It calls `sys/capabilities-self` with the lane token. + +Done 2026-10-14 in the repo. Works live only after CCR-2026-0045 is applied. + +## Re-run both proofs after CCR-2026-0045 + +```task +id: MASON-WP-0012-T04 +status: wait +depends_on: [MASON-WP-0012-T03] +priority: medium +``` + +After railiance-platform applies CCR-2026-0045 (RPF-WP-0057-T02): run +`member` (as `bernd.worsch-3`) and `non-member` (as `signin-probe-negative`, +NK-WP-0051-T05) attended, copy the sanitized receipts into `docs/evidence/`, +and report acceptance to net-kingdom for NK-WP-0051-T06.