Draft the S3 temporary-credentials tutorial as a native pack (NK-WP-0009-T02)

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

Assistant: claude-code
Assistant-Model: sonnet
Assistant-Process: 6969@bnt-lap001
Assistant-Session: e8ae73f2-8385-4c9c-b714-94c3c52cdbc0

runbooks/object-storage-sts/check_sts.py

@@ -0,0 +1,148 @@
+#!/usr/bin/env python3
+"""Offline conformance checks for the object-storage STS credential-vending pattern.
+
+Encodes docs/object-storage-sts-credential-vending.md: canonical actions, TTL
+policy, request/response/deny shapes, and the credential_process output. It never
+talks to a service and never accepts a real credential: values that look like a
+long-lived root key are refused.
+
+Usage:
+  check_sts.py request FILE
+  check_sts.py response FILE REQUEST_FILE
+  check_sts.py deny FILE
+  check_sts.py credproc            # reads credential_process JSON on stdin
+"""
+import json
+import re
+import sys
+from datetime import datetime, timezone
+
+ACTIONS = {
+    "s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:ListBucket",
+    "s3:GetObjectAttributes", "s3:AbortMultipartUpload",
+    "s3:CreateMultipartUpload", "s3:UploadPart", "s3:CompleteMultipartUpload",
+}
+MAX_TTL = 3600
+REASONS = {
+    "tenant_scope_missing", "tenant_mismatch", "prefix_not_registered_for_tenant",
+    "token_invalid", "audience_mismatch", "policy_unavailable", "action_not_permitted",
+}
+FORBIDDEN_KEYS = {"root_access_key", "root_secret", "parent_credentials", "password"}
+
+
+def fail(msg):
+    print(f"FAIL: {msg}")
+    sys.exit(1)
+
+
+def load(path):
+    try:
+        with open(path) as f:
+            return json.load(f)
+    except (OSError, ValueError) as e:
+        fail(f"cannot read {path}: {e}")
+
+
+def need(obj, keys, where):
+    for k in keys:
+        if k not in obj:
+            fail(f"{where}: missing {k}")
+
+
+def walk_keys(obj):
+    if isinstance(obj, dict):
+        for k, v in obj.items():
+            yield k
+            yield from walk_keys(v)
+    elif isinstance(obj, list):
+        for v in obj:
+            yield from walk_keys(v)
+
+
+def parse_time(value, where):
+    try:
+        return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except ValueError:
+        fail(f"{where}: expiration is not an ISO-8601 time")
+
+
+def check_request(req):
+    need(req, ["protected_system_id", "tenant_id", "bucket", "prefix", "actions", "ttl_seconds"], "request")
+    if not re.fullmatch(r"object-storage:[a-z0-9-]+", req["protected_system_id"]):
+        fail("protected_system_id must look like object-storage:<name>")
+    if not re.fullmatch(r"tenant:[a-z0-9:-]+", req["tenant_id"]):
+        fail("tenant_id must look like tenant:<name>")
+    if not req["prefix"].endswith("/"):
+        fail("prefix must end with / (grant boundary is a prefix, not an object)")
+    bad = [a for a in req["actions"] if a not in ACTIONS]
+    if not req["actions"] or bad:
+        fail(f"actions must be non-empty canonical S3 actions; unknown: {bad}")
+    ttl = req["ttl_seconds"]
+    if not isinstance(ttl, int) or not 0 < ttl <= MAX_TTL:
+        fail(f"ttl_seconds must be 1..{MAX_TTL} (longer needs explicit policy)")
+    if set(walk_keys(req)) & FORBIDDEN_KEYS:
+        fail("request carries a forbidden credential field")
+
+
+def check_response(resp, req):
+    need(resp, ["credentials", "scope", "lease", "decision"], "response")
+    need(resp["credentials"], ["access_key_id", "secret_access_key", "session_token", "expiration"], "credentials")
+    if set(walk_keys(resp)) & FORBIDDEN_KEYS:
+        fail("response carries a forbidden credential field")
+    if not resp["credentials"]["session_token"]:
+        fail("credentials must be temporary: session_token is empty")
+    if parse_time(resp["credentials"]["expiration"], "credentials") <= datetime.now(timezone.utc):
+        fail("credentials are already expired")
+    scope = resp["scope"]
+    for key in ("protected_system_id", "tenant_id", "bucket", "prefix"):
+        if scope.get(key) != req[key]:
+            fail(f"scope.{key} differs from the request")
+    if not set(scope.get("actions", [])) <= set(req["actions"]):
+        fail("scope grants actions the request did not ask for")
+    if resp["lease"].get("ttl_seconds", MAX_TTL + 1) > MAX_TTL:
+        fail("lease ttl exceeds the one-hour normal maximum")
+    need(resp["decision"], ["decision_id", "policy_package", "audit_correlation_id"], "decision")
+
+
+def check_deny(deny):
+    need(deny, ["error", "reason_code", "decision_id", "audit_correlation_id"], "deny")
+    if deny["error"] != "credential_denied":
+        fail("deny error must be credential_denied")
+    if deny["reason_code"] not in REASONS:
+        fail(f"reason_code {deny['reason_code']!r} is not a stable documented code")
+    if "credentials" in deny:
+        fail("a deny must not carry credentials")
+
+
+def check_credproc(doc):
+    need(doc, ["Version", "AccessKeyId", "SecretAccessKey", "SessionToken", "Expiration"], "credential_process")
+    if doc["Version"] != 1:
+        fail("credential_process Version must be 1")
+    if not doc["SessionToken"]:
+        fail("SessionToken is empty: not a temporary credential")
+    if parse_time(doc["Expiration"], "credential_process") <= datetime.now(timezone.utc):
+        fail("credential_process output is already expired")
+
+
+def main(argv):
+    if len(argv) < 2:
+        fail(__doc__.strip().splitlines()[0])
+    mode = argv[1]
+    if mode == "request" and len(argv) == 3:
+        check_request(load(argv[2]))
+    elif mode == "response" and len(argv) == 4:
+        check_response(load(argv[2]), load(argv[3]))
+    elif mode == "deny" and len(argv) == 3:
+        check_deny(load(argv[2]))
+    elif mode == "credproc" and len(argv) == 2:
+        try:
+            check_credproc(json.load(sys.stdin))
+        except ValueError as e:
+            fail(f"stdin is not JSON: {e}")
+    else:
+        fail("bad arguments; see the module docstring")
+    print("OK")
+
+
+if __name__ == "__main__":
+    main(sys.argv)

runbooks/object-storage-sts/pack.yaml

@@ -0,0 +1,349 @@
+spec: runbook-pack/v0.2
+id: nk.object-storage-sts
+title: Vend temporary S3 credentials from a NetKingdom identity token
+owner: net-kingdom
+outcome: You can state who owns each hop from identity token to S3 credential, have built and checked a scoped request, a normalized response, a stable deny and a refreshing consumer setup offline, and know whether a live credential-vending service exists to try it against.
+engine: native
+framing:
+  - Today the pattern is an architecture baseline (NK-WP-0007); no credential-vending service is deployed. The offline stages teach and check the contract; they do not prove a live service.
+  - Never put a real access key, secret key, session token or IAM token in a file, chat, State Hub or a log. Every sample here is fabricated and short-lived by construction.
+  - Consumers never fall back to root or long-lived credentials. If refresh fails, stop writes before expiration.
+  - local-identity tokens are for development only; production issuers are key-cape or Keycloak.
+actors:
+  - id: learner
+    role: Learner
+    description: Works through the contract offline, and later against a live service with their own identity.
+    privileges: [read-docs, run-local-checks]
+datapoints:
+  - id: workdir
+    label: Scratch directory
+    type: path
+    default: /tmp/nk-sts-tutorial
+    help: Holds only fabricated sample files.
+    dimensions: {actor: learner, kind: location, subsystem: workstation}
+    ask_in: request
+  - id: tenant
+    label: Tenant
+    type: string
+    default: tenant:coulomb
+    pattern: "tenant:[a-z0-9:-]+"
+    dimensions: {actor: learner, kind: identity, subsystem: flex-auth}
+    ask_in: request
+  - id: bucket
+    label: Bucket
+    type: string
+    default: artifact-store-prod
+    pattern: "[a-z0-9-]+"
+    dimensions: {kind: target, subsystem: object-storage}
+    ask_in: request
+  - id: prefix
+    label: Prefix
+    type: string
+    default: tenant/coulomb/packages/
+    pattern: "[a-z0-9/_-]+/"
+    help: The preferred grant boundary; must end with a slash.
+    dimensions: {kind: target, subsystem: object-storage}
+    ask_in: request
+  - id: ttl_seconds
+    label: Requested TTL in seconds
+    type: string
+    default: "1800"
+    pattern: "[0-9]+"
+    help: Workload default 1800; interactive 900; normal maximum 3600.
+    dimensions: {kind: policy, subsystem: flex-auth}
+    ask_in: request
+  - id: service_state
+    label: Is a credential-vending service deployed?
+    type: enum
+    values: [deployed, not-deployed]
+    default: not-deployed
+    help: Decides which way the live stage is recommended. As of this pack, none is deployed.
+    dimensions: {kind: state, subsystem: credential-vending}
+    ask_in: live
+  - id: endpoint
+    label: Vending endpoint
+    type: string
+    default: ""
+    help: Only when a service is deployed; a private or tunnelled URL.
+    dimensions: {kind: target, subsystem: credential-vending}
+    ask_in: live
+stages:
+  - id: understand
+    title: Know the hops and their owners
+    goal: You can name the owner of each hop and where authority is decided.
+    steps:
+      - id: read-baseline
+        title: Read the architecture baseline
+        actor: learner
+        owner: net-kingdom
+        description: "Read docs/object-storage-sts-credential-vending.md: the core flow, trust boundaries, flex-auth vocabulary, TTL policy and failure modes."
+        idempotent: idempotent
+        verify:
+          done_when: You can trace token -> vending service -> flex-auth decision -> backend exchange -> temporary credentials -> consumer
+          expect: manual
+      - id: ownership
+        title: State who owns what
+        actor: learner
+        owner: net-kingdom
+        requires: [read-baseline]
+        description: "IAM Profile and claims: NetKingdom (issuer key-cape or Keycloak). Decision and vocabulary: flex-auth. Parent credentials, broker config and leases: OpenBao via railiance-platform. Backend STS: railiance-platform. Refresh behaviour: the consumer, for example artifact-store."
+        idempotent: idempotent
+        verify:
+          done_when: You can say why OpenBao is not the authorization decision point
+          expect: manual
+    gates:
+      - id: hops-understood
+        title: The hops and owners are understood
+        goal: clear-ownership
+        blocking: true
+        downstream_risk: Credentials get requested from, or blamed on, the wrong owner.
+        checks:
+          - done_when: Each hop has a named owner
+            expect: manual
+  - id: request
+    title: Build and check a scoped request
+    goal: A request file that names one tenant, one prefix, canonical actions and a bounded TTL passes the contract check.
+    requires: ["gate:hops-understood"]
+    steps:
+      - id: compose-request
+        title: Write a sample request
+        actor: learner
+        owner: net-kingdom
+        uses: [workdir, tenant, bucket, prefix, ttl_seconds]
+        command: |
+          mkdir -p {{workdir}} && chmod 700 {{workdir}}
+          python3 - <<'PY'
+          import json
+          json.dump({
+              "protected_system_id": "object-storage:{{bucket}}",
+              "tenant_id": "{{tenant}}",
+              "bucket": "{{bucket}}",
+              "prefix": "{{prefix}}",
+              "actions": ["s3:GetObject", "s3:PutObject", "s3:ListBucket"],
+              "ttl_seconds": int("{{ttl_seconds}}"),
+              "purpose": "tutorial package upload",
+              "correlation_id": "01TUTORIAL0000000000000001",
+          }, open("{{workdir}}/request.json", "w"), indent=2)
+          PY
+        idempotent: idempotent
+        idempotent_note: Overwrites the same sample file.
+        rollback: Delete the scratch directory.
+        verify:
+          done_when: The request file exists
+          command: test -s {{workdir}}/request.json
+          expect: exit-0
+      - id: check-request
+        title: Check the request against the contract
+        actor: learner
+        owner: net-kingdom
+        requires: [compose-request]
+        uses: [workdir]
+        command: python3 runbooks/object-storage-sts/check_sts.py request {{workdir}}/request.json
+        idempotent: idempotent
+        verify:
+          done_when: The check prints OK
+          command: python3 runbooks/object-storage-sts/check_sts.py request {{workdir}}/request.json
+          expect: output-contains
+          contains: "OK"
+      - id: refuse-bad-request
+        title: See an over-long TTL and an unknown action refused
+        actor: learner
+        owner: net-kingdom
+        requires: [check-request]
+        uses: [workdir]
+        description: The check must refuse a two-hour TTL and a made-up action, the way flex-auth policy would.
+        command: |
+          python3 - <<'PY'
+          import json
+          r = json.load(open("{{workdir}}/request.json"))
+          r["ttl_seconds"] = 7200
+          json.dump(r, open("{{workdir}}/bad-ttl.json", "w"))
+          r["ttl_seconds"] = 900
+          r["actions"] = ["s3:*"]
+          json.dump(r, open("{{workdir}}/bad-action.json", "w"))
+          PY
+          ! python3 runbooks/object-storage-sts/check_sts.py request {{workdir}}/bad-ttl.json && ! python3 runbooks/object-storage-sts/check_sts.py request {{workdir}}/bad-action.json
+        idempotent: idempotent
+        verify:
+          done_when: Both bad requests are refused
+          command: "! python3 runbooks/object-storage-sts/check_sts.py request {{workdir}}/bad-ttl.json && ! python3 runbooks/object-storage-sts/check_sts.py request {{workdir}}/bad-action.json"
+          expect: exit-0
+    gates:
+      - id: request-conforms
+        title: The request conforms and bad ones are refused
+        goal: least-privilege
+        blocking: true
+        downstream_risk: A vague request (wildcard actions, long TTL, object-wide scope) would be over-granted.
+        checks:
+          - done_when: The good request passes
+            command: python3 runbooks/object-storage-sts/check_sts.py request {{workdir}}/request.json
+            expect: exit-0
+          - done_when: The over-long TTL is refused
+            command: "! python3 runbooks/object-storage-sts/check_sts.py request {{workdir}}/bad-ttl.json"
+            expect: exit-0
+  - id: response
+    title: Read a normalized response and a deny
+    goal: You can tell a well-formed temporary grant from a deny, and see that a deny carries no credentials.
+    requires: ["gate:request-conforms"]
+    steps:
+      - id: compose-response
+        title: Write a fabricated allow response
+        actor: learner
+        owner: net-kingdom
+        uses: [workdir, tenant, bucket, prefix]
+        description: Every value is fabricated; the expiration is thirty minutes from now.
+        command: |
+          python3 - <<'PY'
+          import json
+          from datetime import datetime, timedelta, timezone
+          exp = (datetime.now(timezone.utc) + timedelta(minutes=30)).strftime("%Y-%m-%dT%H:%M:%SZ")
+          json.dump({
+              "credentials": {"access_key_id": "TUTORIALKEYID", "secret_access_key": "fabricated", "session_token": "fabricated-session", "expiration": exp},
+              "scope": {"protected_system_id": "object-storage:{{bucket}}", "tenant_id": "{{tenant}}", "bucket": "{{bucket}}", "prefix": "{{prefix}}", "actions": ["s3:GetObject", "s3:ListBucket"]},
+              "lease": {"ttl_seconds": 1800, "renewable": False, "backend": "tutorial", "openbao_lease_id": None},
+              "decision": {"decision_id": "dec_tutorial_1", "policy_package": "object-storage-{{bucket}}@tutorial", "obligations": ["checksum-required"], "audit_correlation_id": "01TUTORIAL0000000000000001"},
+          }, open("{{workdir}}/response.json", "w"), indent=2)
+          json.dump({"error": "credential_denied", "reason_code": "prefix_not_registered_for_tenant", "decision_id": "dec_tutorial_2", "audit_correlation_id": "01TUTORIAL0000000000000002"}, open("{{workdir}}/deny.json", "w"), indent=2)
+          PY
+        idempotent: idempotent
+        verify:
+          done_when: Both sample files exist
+          command: test -s {{workdir}}/response.json && test -s {{workdir}}/deny.json
+          expect: exit-0
+      - id: check-response
+        title: Check the response stays inside the request
+        actor: learner
+        owner: flex-auth
+        requires: [compose-response]
+        uses: [workdir]
+        description: The granted actions may be fewer than requested, never more; scope must match; the credential must be temporary and unexpired; the lease must not exceed one hour.
+        command: python3 runbooks/object-storage-sts/check_sts.py response {{workdir}}/response.json {{workdir}}/request.json
+        idempotent: idempotent
+        verify:
+          done_when: The check prints OK
+          command: python3 runbooks/object-storage-sts/check_sts.py response {{workdir}}/response.json {{workdir}}/request.json
+          expect: output-contains
+          contains: "OK"
+      - id: check-deny
+        title: Check the deny has a stable reason and no credentials
+        actor: learner
+        owner: flex-auth
+        requires: [compose-response]
+        uses: [workdir]
+        command: python3 runbooks/object-storage-sts/check_sts.py deny {{workdir}}/deny.json
+        idempotent: idempotent
+        verify:
+          done_when: The check prints OK
+          command: python3 runbooks/object-storage-sts/check_sts.py deny {{workdir}}/deny.json
+          expect: output-contains
+          contains: "OK"
+    gates:
+      - id: contract-shapes
+        title: Allow and deny shapes conform
+        goal: auditable-decisions
+        blocking: true
+        downstream_risk: A consumer that cannot tell a deny from a grant, or an over-broad grant from a scoped one, will misuse credentials.
+        checks:
+          - done_when: The allow response is inside the request scope
+            command: python3 runbooks/object-storage-sts/check_sts.py response {{workdir}}/response.json {{workdir}}/request.json
+            expect: exit-0
+          - done_when: The deny carries a stable reason code and no credentials
+            command: python3 runbooks/object-storage-sts/check_sts.py deny {{workdir}}/deny.json
+            expect: exit-0
+  - id: consume
+    title: Consume it through credential_process
+    goal: An SDK-style consumer obtains fresh temporary credentials and never holds a long-lived key.
+    requires: ["gate:contract-shapes"]
+    steps:
+      - id: credproc-script
+        title: Write a stand-in credential_process
+        actor: learner
+        owner: net-kingdom
+        uses: [workdir]
+        description: "A real one calls the vending service with the caller's token. This stand-in prints fabricated credentials with a fresh expiry so you can see the shape the AWS SDKs expect."
+        command: |
+          cat > {{workdir}}/credproc.sh <<'SH'
+          #!/bin/sh
+          exp=$(date -u -d '+30 minutes' +%Y-%m-%dT%H:%M:%SZ)
+          printf '{"Version":1,"AccessKeyId":"TUTORIALKEYID","SecretAccessKey":"fabricated","SessionToken":"fabricated-session","Expiration":"%s"}\n' "$exp"
+          SH
+          chmod 700 {{workdir}}/credproc.sh
+        idempotent: idempotent
+        verify:
+          done_when: The script exists and is executable
+          command: test -x {{workdir}}/credproc.sh
+          expect: exit-0
+      - id: check-credproc
+        title: Check the credential_process output
+        actor: learner
+        owner: artifact-store
+        requires: [credproc-script]
+        uses: [workdir]
+        command: "{{workdir}}/credproc.sh | python3 runbooks/object-storage-sts/check_sts.py credproc"
+        idempotent: idempotent
+        verify:
+          done_when: The check prints OK
+          command: "{{workdir}}/credproc.sh | python3 runbooks/object-storage-sts/check_sts.py credproc"
+          expect: output-contains
+          contains: "OK"
+      - id: sdk-config
+        title: See how an SDK profile points at it
+        actor: learner
+        owner: artifact-store
+        requires: [check-credproc]
+        uses: [workdir]
+        description: "An AWS-style profile uses credential_process; the SDK re-runs it before Expiration. Nothing is written to your real ~/.aws."
+        command: printf '[profile tutorial]\ncredential_process = %s/credproc.sh\n' "{{workdir}}"
+        idempotent: idempotent
+        verify:
+          done_when: You can say what the consumer does when the process fails (stop writes before expiry, retry with backoff, never fall back to root)
+          expect: manual
+    gates:
+      - id: consumer-refreshes
+        title: The consumer path yields temporary credentials
+        goal: short-lived-credentials
+        blocking: true
+        downstream_risk: A consumer wired to a static key defeats the pattern.
+        checks:
+          - done_when: credential_process output is well-formed, temporary and unexpired
+            command: "{{workdir}}/credproc.sh | python3 runbooks/object-storage-sts/check_sts.py credproc"
+            expect: exit-0
+  - id: live
+    title: Try it against a live service, if one exists
+    goal: The live status is recorded honestly; if a service is deployed, a real request is denied or granted with an audit id.
+    requires: ["gate:consumer-refreshes"]
+    steps:
+      - id: live-or-not
+        title: Choose by whether a service is deployed
+        actor: learner
+        owner: railiance-platform
+        uses: [service_state, endpoint, workdir]
+        description: As of this pack no credential-vending service is deployed, so the expected answer is "not deployed". Do not stand one up from this repo; deployment belongs to railiance-platform. If deployed, enter your IAM token only at the hidden prompt; expect a grant or a deny, each with an audit correlation id. If not deployed, record that the offline checks passed and the live path is unproven.
+        idempotent: idempotent
+        auto_confirm: manual-only
+        one_of:
+          - id: call-service
+            title: Call the deployed service with your own token
+            recommended_when: {service_state: [deployed]}
+            command: |
+              printf "IAM token: " >&2; read -rs IAM_TOKEN; printf "\n" >&2
+              curl -s -X POST {{endpoint}}/v1/object-storage/credentials -H "Authorization: Bearer $IAM_TOKEN" -H 'Content-Type: application/json' --data @{{workdir}}/request.json | python3 -c 'import json,sys; d=json.load(sys.stdin); print(d.get("error") or "granted", d.get("reason_code",""), (d.get("decision") or d).get("audit_correlation_id"))'
+              unset IAM_TOKEN
+            verify:
+              done_when: The response is a grant or a deny with an audit correlation id, and you did not see a secret
+              expect: manual
+          - id: not-deployed
+            title: Record that no service is deployed yet
+            recommended_when: {service_state: [not-deployed]}
+            verify:
+              done_when: The live status is recorded as not deployed
+              expect: manual
+    gates:
+      - id: live-status-recorded
+        title: The live status is on record
+        goal: honest-evidence
+        blocking: true
+        downstream_risk: The offline pass gets read as proof of a running service.
+        checks:
+          - done_when: Either a real request was answered with an audit id, or "not deployed" is recorded
+            expect: manual

workplans/NK-WP-0009-netkingdom-security-pattern-tutorials.md

@@ -81,7 +81,7 @@ cross-repo ownership notes.
 
 ```task
 id: NK-WP-0009-T02
-status: todo
+status: progress
 priority: high
 state_hub_task_id: "d13760fd-2527-5a39-acb1-11c8091d65a1"
 ```
@@ -91,6 +91,12 @@ NetKingdom identity token", covering key-cape/Keycloak identity,
 flex-auth authorization, object-store STS exchange, and SDK consumer
 configuration.
 
+**Drafted 2026-09-29 as a native pack**: `runbooks/object-storage-sts/` (`nk.object-storage-sts`, validates,
+`unexercised`) with an offline validator, `check_sts.py`, encoding the request, response, deny and `credential_process`
+contract from `docs/object-storage-sts-credential-vending.md`. The offline checks were run and pass, including the
+refusals (TTL over one hour, wildcard action). No credential-vending service is deployed (the baseline is NK-WP-0007),
+so the live stage records "not deployed" honestly. The task closes on a run receipt from the UI, like T03-T05.
+
 ## Document the existing OpenBao operating path
 
 ```task