#!/usr/bin/env python3 """Lifecycle integration test for the pool B service-account CLI. Validates plans/api.key.support.cognito.two-pool.plan.3.claude.md §6.3 beyond the super-admin coverage in test_two_pool_auth.py. Operates on a dedicated throwaway user (default: svc-lifecycle-test-1) so it never collides with the long-lived super-admin test account. What this script does (top to bottom): Step 1. service.account.tool create (provision) Step 2. InitiateAuth(, password) (baseline OK) Step 3. service.account.tool disable InitiateAuth(refresh) (expect NotAuthorized) Step 4. service.account.tool enable InitiateAuth(, password) (expect OK again) Step 5. service.account.tool rotate-password InitiateAuth(, OLD_password) (expect NotAuthorized) InitiateAuth(, NEW_password) (expect OK) Step 6. service.account.tool delete InitiateAuth(, password) (expect UserNotFound) service.account.tool show (expect non-zero exit) The script captures the initial password from the `create` output and the new password from `rotate-password` output by scanning stdout — no secret is ever read from disk or written to a file. Prerequisites: - Run via the wrapper (recommended): bash scripts/manual_tests/run_test_two_pool_lifecycle.sh The wrapper sources the repo-root .env for secrets, hardcodes the public pool B identifiers (pool ID, app client ID, region, PERMIT_IO_BASE_URL), and execs this script. - Required in .env (or shell): PERMIT_IO_API_KEY Permit.io API key (secret). Public identifiers (pool ID, app client ID, region) are hardcoded in the wrapper — not secrets. - Optional in .env (or shell): AWS_PROFILE AWS SSO profile (default: aarete). Must have cognito-idp Admin* rights on user pool us-east-2_Qom1i9qIG. SVC_LIFECYCLE_USERNAME Override the throwaway username (default: svc-lifecycle-test-1). - AWS_PROFILE must have an active session. Run aws sso login --profile "$AWS_PROFILE" if cached credentials have expired. Usage: bash scripts/manual_tests/run_test_two_pool_lifecycle.sh Exit codes: 0 — all lifecycle assertions held. 1 — any assertion failed; script attempts a best-effort cleanup of the throwaway user before exiting. 2 — required env / preconditions missing. """ from __future__ import annotations import json import os import re import subprocess import sys import urllib.error import urllib.request from pathlib import Path from typing import Any, Dict, Tuple HTTP_TIMEOUT = 15 REPO_ROOT = Path(__file__).resolve().parents[2] RUN_TOOL = REPO_ROOT / "cmd" / "auth_related" / "service.account.tool" / "run.tool.sh" def _load_dotenv() -> None: """Populate os.environ from the repo-root .env without external deps.""" here = Path(__file__).resolve() for parent in [here.parent, *here.parents]: candidate = parent / ".env" if candidate.is_file(): for raw in candidate.read_text().splitlines(): line = raw.strip() if not line or line.startswith("#"): continue if "=" not in line: continue k, v = line.split("=", 1) k = k.strip() v = v.strip().strip('"').strip("'") if k and k not in os.environ: os.environ[k] = v return def section(title: str) -> None: print() print("=" * 72) print(title) print("=" * 72) def require_env(name: str) -> str: v = os.environ.get(name, "").strip() if not v: print(f"FAIL: required env var {name} is unset. See .env.example.") sys.exit(2) return v def run_tool(*args: str, check: bool = True) -> subprocess.CompletedProcess: """Invoke run.tool.sh with the supplied subcommand. Streams stdout/err.""" cmd = ["bash", str(RUN_TOOL), *args] print(f"$ {' '.join(args)}") proc = subprocess.run(cmd, capture_output=True, text=True, check=False) if proc.stdout: sys.stdout.write(proc.stdout) if proc.stderr: sys.stderr.write(proc.stderr) if check and proc.returncode != 0: raise RuntimeError(f"run.tool.sh {args!r} failed: rc={proc.returncode}") return proc def extract_password(stdout: str) -> str: """Pull the Password line out of `create` / `rotate-password` output.""" # `create` prints: Password : # `rotate-password`: New password for : for raw in stdout.splitlines(): line = raw.strip() m = re.match(r"^Password\s*:\s*(\S+)\s*$", line) if m: return m.group(1) m = re.match(r"^New password for \S+:\s*(\S+)\s*$", line) if m: return m.group(1) raise RuntimeError("could not find password line in tool output") def cognito_call(region: str, operation: str, body: Dict[str, Any]) -> Tuple[int, Dict[str, Any]]: """POST to Cognito IDP; returns (http_status, parsed_json or error_dict).""" url = f"https://cognito-idp.{region}.amazonaws.com/" payload = json.dumps(body).encode("utf-8") req = urllib.request.Request( url, data=payload, method="POST", headers={ "Content-Type": "application/x-amz-json-1.1", "X-Amz-Target": f"AWSCognitoIdentityProviderService.{operation}", }, ) try: with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as r: return r.status, json.loads(r.read()) except urllib.error.HTTPError as e: body_text = e.read().decode("utf-8", errors="replace") try: return e.code, json.loads(body_text) except json.JSONDecodeError: return e.code, {"__type": "ParseError", "message": body_text} def initiate_user_password(region: str, client_id: str, username: str, password: str) -> Tuple[int, Dict[str, Any]]: return cognito_call(region, "InitiateAuth", { "AuthFlow": "USER_PASSWORD_AUTH", "ClientId": client_id, "AuthParameters": {"USERNAME": username, "PASSWORD": password}, }) def initiate_refresh(region: str, client_id: str, refresh_token: str) -> Tuple[int, Dict[str, Any]]: return cognito_call(region, "InitiateAuth", { "AuthFlow": "REFRESH_TOKEN_AUTH", "ClientId": client_id, "AuthParameters": {"REFRESH_TOKEN": refresh_token}, }) def expect_ok(status: int, resp: Dict[str, Any], label: str) -> Dict[str, Any]: if status != 200 or "AuthenticationResult" not in resp: raise AssertionError(f"{label}: expected 200+AuthenticationResult, got {status} {resp.get('__type')}: {resp.get('message')}") return resp["AuthenticationResult"] def expect_not_authorized(status: int, resp: Dict[str, Any], label: str) -> None: if status == 200 and "AuthenticationResult" in resp: raise AssertionError(f"{label}: expected auth failure, but got tokens back") err = resp.get("__type", "") if "NotAuthorized" not in err and "UserNotFound" not in err: raise AssertionError(f"{label}: expected NotAuthorized/UserNotFound, got {status} {err}: {resp.get('message')}") def main() -> int: _load_dotenv() # Strip LocalStack endpoint overrides before the Go subprocess # (run.tool.sh) inherits them. aws-sdk-go-v2 treats # AWS_ENDPOINT_URL as a global service-endpoint override — # devbox commonly sets it to http://localhost:4566, where # cognito-idp is unimplemented and every Admin* call returns # HTTP 501. The wrapper also unsets these, but doing it here # makes the script robust when invoked directly. for key in ("AWS_ENDPOINT_URL", "AWS_ENDPOINT_URL_COGNITO_IDP"): os.environ.pop(key, None) user_pool_id = require_env("COGNITO_SVC_USER_POOL_ID") client_id = require_env("COGNITO_SVC_APP_CLIENT_ID") require_env("PERMIT_IO_API_KEY") # Region is encoded in the pool ID prefix (us-east-2_Qom1i9qIG). # Override any stale AWS_REGION so a shell that exports us-east-1 # (devbox default) cannot redirect the SDK to the wrong region. region = user_pool_id.split("_", 1)[0] os.environ["AWS_REGION"] = region username = os.environ.get("SVC_LIFECYCLE_USERNAME", "svc-lifecycle-test-1").strip() if not username: username = "svc-lifecycle-test-1" print(f"Region : {region}") print(f"User pool id : {user_pool_id}") print(f"App client id : {client_id}") print(f"Throwaway username: {username}") # Best-effort: delete a leftover user from a prior aborted run so # `create` does not blow up with UsernameExistsException. We swallow # all errors — non-existence is expected on a clean run. section("Pre-flight cleanup (idempotent)") run_tool("delete", username, check=False) cleanup_needed = False try: section("Step 1: create throwaway user with super_admin role") proc = run_tool("create", username, "--roles", "super_admin", "--description", "lifecycle test user — safe to delete") cleanup_needed = True password = extract_password(proc.stdout) print(f" captured initial password (len={len(password)})") section("Step 2: baseline InitiateAuth must succeed") status, resp = initiate_user_password(region, client_id, username, password) result = expect_ok(status, resp, "baseline InitiateAuth") refresh_token = result["RefreshToken"] print(" baseline InitiateAuth OK; captured RefreshToken") section("Step 3: disable + refresh must be rejected") run_tool("disable", username) status, resp = initiate_refresh(region, client_id, refresh_token) expect_not_authorized(status, resp, "refresh after disable") print(" refresh correctly rejected after disable") # Access tokens minted before disable remain valid until exp; # we deliberately do not assert on that here (would require a # short-TTL test client — see plan §6.3 deferred item). section("Step 4: enable + InitiateAuth must succeed again") run_tool("enable", username) status, resp = initiate_user_password(region, client_id, username, password) expect_ok(status, resp, "InitiateAuth after enable") print(" InitiateAuth OK after re-enable") section("Step 5: rotate-password — old fails, new succeeds") old_password = password proc = run_tool("rotate-password", username) new_password = extract_password(proc.stdout) if new_password == old_password: raise AssertionError("rotate-password returned the same password") status, resp = initiate_user_password(region, client_id, username, old_password) expect_not_authorized(status, resp, "InitiateAuth with old password") print(" old password correctly rejected") status, resp = initiate_user_password(region, client_id, username, new_password) expect_ok(status, resp, "InitiateAuth with new password") password = new_password print(" new password accepted") section("Step 6: delete — auth must fail and show must error") run_tool("delete", username) cleanup_needed = False status, resp = initiate_user_password(region, client_id, username, password) expect_not_authorized(status, resp, "InitiateAuth after delete") proc = run_tool("show", username, check=False) if proc.returncode == 0: raise AssertionError("show exited 0; expected non-zero") print(" delete confirmed: auth fails, show errors as expected") section("Result") print(f"PASS — full lifecycle for {username} validated end-to-end.") return 0 except AssertionError as e: print(f"\nFAIL: {e}") return 1 except Exception as e: print(f"\nFAIL: unexpected error: {e}") return 1 finally: if cleanup_needed: section("Cleanup (script failed mid-run)") run_tool("delete", username, check=False) if __name__ == "__main__": sys.exit(main())