#!/usr/bin/env python3
"""Check a CAIN Action Credential in your own service (needs only `cryptography`). Pin the JWKS you downloaded from
https://cainstudio.online/api/v1/broker-keystone/.well-known/jwks.json. `seen` must be a store of used jti values shared by all your instances (a set
works for one process). Any failure raises; only a fully valid, unused credential returns its claims."""
import base64, hashlib, json, re, time
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

class Refused(Exception):
    pass

def _d(s): return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))

def _require(ok, why):
    # explicit checks, not `assert`: python -O strips asserts and would accept expired or replayed credentials
    if not ok:
        raise Refused(why)

def verify(token, jwks, *, aud, method, path, seen, body=b"", issuer="https://cainstudio.online/api/v1/broker-keystone", skew=5):
    h64, c64, s64 = token.split(".")
    header, claims = json.loads(_d(h64)), json.loads(_d(c64))
    _require(header.get("alg") == "EdDSA", "alg")
    key = next((k for k in jwks["keys"] if k["kid"] == header.get("kid")), None)
    _require(key is not None, "unknown key id")
    Ed25519PublicKey.from_public_bytes(_d(key["x"])).verify(_d(s64), (h64 + "." + c64).encode())
    act, now = claims["act"], time.time()
    _require("typ" not in claims, "not a credential")
    _require(claims["iss"] == issuer and claims["aud"] == aud, "issuer/audience")
    _require(claims["exp"] + skew >= now >= claims["nbf"] - skew, "time")
    _require(not re.search(r"%(?:2e|2f|5c|00)|\\|[\x00-\x1f\x7f]", path, re.I), "ambiguous path (encoded dot/slash, control char)")
    _require(not re.search(r";|%25|[^\x21-\x7e]", path.split("?", 1)[0]), "ambiguous path (';', double encoding, non-ASCII)")
    _require(act["method"] == method.upper() and act["path"] == path, "action")
    _require(act["body_sha256"] == hashlib.sha256(body).hexdigest(), "body")
    # one use per credential family: a parent and the children derived from it share the root's single use
    root = claims.get("root_jti") or claims["jti"]
    _require(("root:" + root) not in seen and claims["jti"] not in seen, "replayed")
    seen.add("root:" + root)
    seen.add(claims["jti"])
    return claims