net-kingdom/runbooks/object-storage-sts/check_sts.py

#!/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)