Jay Brown
451da3d26d
Support for service accounts with static credentials * feature complete * tests and docs * lint fix
304 lines
12 KiB
Python
Executable File
304 lines
12 KiB
Python
Executable File
#!/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 <user> (provision)
|
|
Step 2. InitiateAuth(<user>, password) (baseline OK)
|
|
Step 3. service.account.tool disable <user>
|
|
InitiateAuth(refresh) (expect NotAuthorized)
|
|
Step 4. service.account.tool enable <user>
|
|
InitiateAuth(<user>, password) (expect OK again)
|
|
Step 5. service.account.tool rotate-password <user>
|
|
InitiateAuth(<user>, OLD_password) (expect NotAuthorized)
|
|
InitiateAuth(<user>, NEW_password) (expect OK)
|
|
Step 6. service.account.tool delete <user>
|
|
InitiateAuth(<user>, password) (expect UserNotFound)
|
|
service.account.tool show <user> (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 : <pw>
|
|
# `rotate-password`: New password for <u>: <pw>
|
|
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 <deleted-user> 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())
|