christiangeorgelucas/paseto-tools

import json

from gen.messages_pb2 import CheckExpirationInput, ExpirationResult
from gen.axiom_context import AxiomContext

from nodes.paseto_util import parse_registered_time


def check_expiration(ax: AxiomContext, input: CheckExpirationInput) -> ExpirationResult:
    """Check ONLY whether a claims JSON object is expired, per its "exp"
    claim (an ISO 8601 string, PASETO's registered-claim time convention),
    against a caller-supplied `now`. A claims object with no exp claim is
    never reported expired.
    """
    try:
        claims = json.loads(input.claims_json)
    except (json.JSONDecodeError, TypeError) as err:
        return ExpirationResult(error=f"claims_json is not valid JSON: {err}")
    if not isinstance(claims, dict):
        return ExpirationResult(error="claims_json must decode to a JSON object")

    try:
        now = parse_registered_time(input.now)
    except ValueError as err:
        return ExpirationResult(error=f"now: {err}")

    if "exp" not in claims:
        return ExpirationResult(exp_present=False, expired=False)

    try:
        exp_dt = parse_registered_time(str(claims["exp"]))
    except ValueError as err:
        return ExpirationResult(exp_present=True, error=f"invalid exp: {err}")

    delta = exp_dt - now
    seconds_remaining = int(delta.total_seconds())
    return ExpirationResult(
        exp_present=True,
        expired=seconds_remaining < 0,
        exp=str(claims["exp"]),
        seconds_remaining=seconds_remaining,
    )

import json

from gen.messages_pb2 import TokenInput, UnverifiedClaims
from gen.axiom_context import AxiomContext

from nodes.paseto_util import (
    parse_token_header,
    base64url_decode,
    try_json_loads,
    bytes_to_claims_json,
    PUBLIC_SIG_SIZE,
)


def decode_unverified_public(ax: AxiomContext, input: TokenInput) -> UnverifiedClaims:
    """Decode a v#.public token's message and footer WITHOUT checking the
    signature, by splitting the message from the trailing signature using
    the version's fixed, public signature length. UNVERIFIED.

    A "local" token's payload is genuinely encrypted ciphertext - there is
    nothing meaningful to decode without the decryption key, so this returns
    a structured error for it rather than garbage bytes.
    """
    try:
        h = parse_token_header(input.token)
    except ValueError as err:
        return UnverifiedClaims(well_formed=False, error=str(err))

    if h.purpose != "public":
        return UnverifiedClaims(
            version=h.version,
            well_formed=True,
            error=(
                'this is a "local" (encrypted) token - its payload is ciphertext, not '
                "signed plaintext, so there is nothing to decode without the decryption "
                "key; use VerifyLocal instead"
            ),
        )

    try:
        payload_and_sig = base64url_decode(h.parts[2])
    except ValueError as err:
        return UnverifiedClaims(version=h.version, well_formed=False, error=f"payload segment is not valid base64url: {err}")

    sig_size = PUBLIC_SIG_SIZE[h.version]
    if len(payload_and_sig) <= sig_size:
        return UnverifiedClaims(
            version=h.version,
            well_formed=False,
            error=f"payload too short to contain a v{h.version} signature ({sig_size} bytes)",
        )
    message_bytes = payload_and_sig[: len(payload_and_sig) - sig_size]

    footer_json = ""
    if len(h.parts) == 4 and h.parts[3]:
        try:
            footer_bytes = base64url_decode(h.parts[3])
            footer_text = footer_bytes.decode("utf-8")
        except (ValueError, UnicodeDecodeError) as err:
            return UnverifiedClaims(version=h.version, well_formed=False, error=f"footer segment is invalid: {err}")
        parsed_json, is_json = try_json_loads(footer_text)
        # Always surface the footer as JSON text: the parsed object when it
        # is JSON, else the raw text as a JSON string literal.
        footer_json = parsed_json if is_json else json.dumps(footer_text)

    return UnverifiedClaims(
        version=h.version,
        claims_json=bytes_to_claims_json(message_bytes),
        footer_json=footer_json,
        well_formed=True,
        warning="UNVERIFIED - signature not checked; call VerifyPublic to authenticate",
    )

from gen.messages_pb2 import TokenInput, TokenInfo
from gen.axiom_context import AxiomContext

from nodes.paseto_util import parse_token_header


def identify_token(ax: AxiomContext, input: TokenInput) -> TokenInfo:
    """Identify a PASETO token's version and purpose, and check basic
    structural well-formedness, WITHOUT touching any cryptography.
    """
    try:
        h = parse_token_header(input.token)
    except ValueError as err:
        return TokenInfo(well_formed=False, error=str(err))

    return TokenInfo(
        version=h.version,
        purpose=h.purpose,
        header=h.header,
        part_count=len(h.parts),
        well_formed=True,
        footer_present=len(h.parts) == 4 and bool(h.parts[3]),
    )

from gen.messages_pb2 import TokenInput, FooterInfo
from gen.axiom_context import AxiomContext

from nodes.paseto_util import parse_token_header, base64url_decode, try_json_loads


def inspect_footer(ax: AxiomContext, input: TokenInput) -> FooterInfo:
    """Decode ONLY the footer of a PASETO token, WITHOUT needing any key -
    the footer is authenticated but never encrypted, so it is always
    inspectable.
    """
    try:
        h = parse_token_header(input.token)
    except ValueError as err:
        return FooterInfo(present=False, error=str(err))

    if len(h.parts) != 4 or not h.parts[3]:
        return FooterInfo(present=False)

    try:
        footer_bytes = base64url_decode(h.parts[3])
    except ValueError as err:
        return FooterInfo(present=True, error=f"footer segment is not valid base64url: {err}")

    try:
        footer_raw = footer_bytes.decode("utf-8")
    except UnicodeDecodeError as err:
        return FooterInfo(present=True, error=f"footer is not valid UTF-8: {err}")

    footer_json, is_json = try_json_loads(footer_raw)
    return FooterInfo(
        present=True,
        footer_raw=footer_raw,
        footer_json=footer_json,
        is_json=is_json,
    )

import base64

from gen.messages_pb2 import ParsePaserkInput, ParsePaserkResult
from gen.axiom_context import AxiomContext

from nodes.paseto_util import base64url_decode, PASERK_TYPES


def parse_paserk(ax: AxiomContext, input: ParsePaserkInput) -> ParsePaserkResult:
    """Structurally parse a PASERK-formatted key string to identify its
    PASETO version and key type, and confirm it decodes without error.
    Secret-carrying key types never have their key bytes echoed back.
    """
    paserk = input.paserk or ""
    parts = paserk.split(".")
    if len(parts) < 3:
        return ParsePaserkResult(well_formed=False, error=f"expected at least 3 dot-separated segments, found {len(parts)}")

    v_raw, key_type = parts[0], parts[1]
    if not v_raw.startswith("k") or not v_raw[1:].isdigit():
        return ParsePaserkResult(well_formed=False, error=f'malformed version segment "{v_raw}" - expected "k1".."k4"')
    version = int(v_raw[1:])
    if version not in (1, 2, 3, 4):
        return ParsePaserkResult(well_formed=False, error=f"unsupported PASERK version {version} - expected 1-4")

    info = PASERK_TYPES.get(key_type)
    if info is None:
        return ParsePaserkResult(
            version=version,
            well_formed=False,
            error=f'unrecognized PASERK key type "{key_type}" - expected one of {", ".join(sorted(PASERK_TYPES))}',
        )

    data_segments = parts[2:]
    if info["wrapped"]:
        # e.g. "k4.local-wrap.pie.<data>" - at least an algorithm prefix and a data blob.
        if len(data_segments) < 2 or not all(data_segments):
            return ParsePaserkResult(
                version=version,
                key_type=key_type,
                is_secret_material=info["secret"],
                well_formed=False,
                error=f'"{key_type}" requires a "<prefix>.<data>" tail',
            )
        # The wrapping/encryption payload itself is opaque without the wrapping
        # key/password, so only structural presence is checked here.
        return ParsePaserkResult(version=version, key_type=key_type, is_secret_material=info["secret"], well_formed=True)

    data = ".".join(data_segments)
    if not data:
        return ParsePaserkResult(
            version=version, key_type=key_type, is_secret_material=info["secret"], well_formed=False, error="empty data segment"
        )
    try:
        raw = base64url_decode(data)
    except ValueError as err:
        return ParsePaserkResult(
            version=version,
            key_type=key_type,
            is_secret_material=info["secret"],
            well_formed=False,
            error=f"data segment is not valid base64url: {err}",
        )

    public_material = base64.b64encode(raw).decode("ascii") if key_type == "public" else ""
    return ParsePaserkResult(
        version=version,
        key_type=key_type,
        is_secret_material=info["secret"],
        well_formed=True,
        public_key_material_b64=public_material,
    )

from pyseto import Key
from pyseto.exceptions import PysetoError

from gen.messages_pb2 import PaserkIdInput, PaserkIdResult
from gen.axiom_context import AxiomContext

from nodes.paseto_util import decode_key_material, VALID_VERSIONS, VALID_PURPOSES


def paserk_id(ax: AxiomContext, input: PaserkIdInput) -> PaserkIdResult:
    """Compute the PASERK ID(s) of a caller-supplied PASETO key - a
    deterministic SHA-384 (v1/v3) or BLAKE2b (v2/v4) hash-based fingerprint.
    For a "public"-purpose secret key, also returns the peer ID of its
    corresponding public key.
    """
    if input.version not in VALID_VERSIONS:
        return PaserkIdResult(error=f"version must be 1-4, got {input.version}")
    if input.purpose not in VALID_PURPOSES:
        return PaserkIdResult(error=f'purpose must be "local" or "public", got "{input.purpose}"')

    if input.purpose == "local":
        try:
            key_bytes = decode_key_material(input.key_material, input.key_encoding)
        except ValueError as err:
            return PaserkIdResult(error=str(err))
        key_arg = key_bytes
    else:
        pem = (input.key_material or "").strip()
        if not pem.startswith("-----BEGIN"):
            return PaserkIdResult(error='for purpose="public", key_material must be a PEM-encoded public or secret key')
        key_arg = pem

    try:
        key = Key.new(input.version, input.purpose, key_arg)
    except (ValueError, PysetoError) as err:
        return PaserkIdResult(error=f"invalid key material for v{input.version}.{input.purpose}: {err}")

    try:
        pid = key.to_paserk_id()
    except (ValueError, PysetoError, NotImplementedError) as err:
        return PaserkIdResult(error=f"could not compute PASERK ID: {err}")

    peer_id = ""
    if input.purpose == "public":
        try:
            peer_id = key.to_peer_paserk_id()
        except (ValueError, PysetoError, NotImplementedError):
            peer_id = ""

    return PaserkIdResult(paserk_id=pid, peer_paserk_id=peer_id)

import json
from datetime import timedelta

from gen.messages_pb2 import ValidateClaimsInput, ClaimsValidationResult, ClaimCheck
from gen.axiom_context import AxiomContext

from nodes.paseto_util import parse_registered_time


def validate_claims(ax: AxiomContext, input: ValidateClaimsInput) -> ClaimsValidationResult:
    """Validate the registered claims (exp/nbf/iat/aud/iss/sub) of a claims
    JSON object against a caller-supplied `now` (ISO 8601 - PASETO's own
    registered-claim time format, not JWT's unix-second NumericDate). Checks
    claim CONTENT only - combine with VerifyLocal/VerifyPublic to authenticate
    the token first.
    """
    if input.clock_skew_seconds < 0:
        return ClaimsValidationResult(error="clock_skew_seconds must be >= 0")

    try:
        claims = json.loads(input.claims_json)
    except (json.JSONDecodeError, TypeError) as err:
        return ClaimsValidationResult(error=f"claims_json is not valid JSON: {err}")
    if not isinstance(claims, dict):
        return ClaimsValidationResult(error="claims_json must decode to a JSON object")

    try:
        now = parse_registered_time(input.now)
    except ValueError as err:
        return ClaimsValidationResult(error=f"now: {err}")

    leeway = timedelta(seconds=input.clock_skew_seconds)

    def time_check(name: str) -> ClaimCheck:
        if name not in claims:
            return ClaimCheck(present=False, valid=True)
        try:
            t = parse_registered_time(str(claims[name]))
        except ValueError as err:
            return ClaimCheck(present=True, valid=False, detail=f"invalid {name}: {err}")
        if name == "exp":
            ok = now <= t + leeway
            detail = "" if ok else f"expired at {claims[name]}"
        elif name == "nbf":
            ok = now >= t - leeway
            detail = "" if ok else f"not valid until {claims[name]}"
        else:  # iat - informational only, no forward/backward rule
            ok = True
            detail = ""
        return ClaimCheck(present=True, valid=ok, detail=detail)

    def string_check(name: str, expected: str) -> ClaimCheck:
        present = name in claims
        if not expected:
            return ClaimCheck(present=present, valid=True)
        if not present:
            return ClaimCheck(present=False, valid=False, detail=f"{name} claim absent, expected {expected!r}")
        actual = claims[name]
        ok = actual == expected
        detail = "" if ok else f"{name} mismatch: got {actual!r}, expected {expected!r}"
        return ClaimCheck(present=True, valid=ok, detail=detail)

    exp = time_check("exp")
    nbf = time_check("nbf")
    iat = time_check("iat")
    aud = string_check("aud", input.expected_audience)
    iss = string_check("iss", input.expected_issuer)
    sub = ClaimCheck(present="sub" in claims, valid=True)

    all_valid = exp.valid and nbf.valid and iat.valid and aud.valid and iss.valid

    return ClaimsValidationResult(exp=exp, nbf=nbf, iat=iat, aud=aud, iss=iss, sub=sub, all_valid=all_valid)

import pyseto
from pyseto import Key
from pyseto.exceptions import PysetoError

from gen.messages_pb2 import VerifyLocalInput, VerifyResult
from gen.axiom_context import AxiomContext

from nodes.paseto_util import parse_token_header, decode_key_material, bytes_to_claims_json


def verify_local(ax: AxiomContext, input: VerifyLocalInput) -> VerifyResult:
    """Cryptographically decrypt and authenticate a v1-v4 "local" (symmetric)
    PASETO token against a caller-supplied key. Never checks claim content or
    the system clock - use ValidateClaims/CheckExpiration on the result.
    """
    try:
        h = parse_token_header(input.token)
    except ValueError as err:
        return VerifyResult(error=str(err))

    if h.purpose != "local":
        return VerifyResult(version=h.version, error='this is a "public" token; use VerifyPublic instead')

    implicit_assertion = (input.implicit_assertion or "").encode("utf-8")
    if h.version in (1, 2) and implicit_assertion:
        return VerifyResult(
            version=h.version,
            error=f"v{h.version} tokens do not support an implicit assertion (v3/v4 only); "
            "the platform would silently ignore it, so this is rejected explicitly rather than "
            "misleading the caller into thinking it was checked",
        )

    try:
        key_bytes = decode_key_material(input.key_material, input.key_encoding)
    except ValueError as err:
        return VerifyResult(version=h.version, error=str(err))

    try:
        key = Key.new(h.version, "local", key_bytes)
    except (ValueError, PysetoError) as err:
        return VerifyResult(version=h.version, error=f"invalid key material for v{h.version}.local: {err}")

    try:
        token = pyseto.decode(key, input.token, implicit_assertion=implicit_assertion, deserializer=None)
    except (ValueError, PysetoError) as err:
        return VerifyResult(version=h.version, valid=False, reason=str(err))
    except Exception as err:  # noqa: BLE001 - never leak a raw traceback for attacker-controlled input
        return VerifyResult(version=h.version, valid=False, reason=f"decryption/authentication failed: {err}")

    footer_json = ""
    if token.footer:
        footer_bytes = token.footer if isinstance(token.footer, bytes) else str(token.footer).encode("utf-8")
        footer_json = bytes_to_claims_json(footer_bytes)

    payload_bytes = token.payload if isinstance(token.payload, bytes) else str(token.payload).encode("utf-8")
    return VerifyResult(
        valid=True,
        version=h.version,
        claims_json=bytes_to_claims_json(payload_bytes),
        footer_json=footer_json,
        reason="verified",
    )

import pyseto
from pyseto import Key
from pyseto.exceptions import PysetoError

from gen.messages_pb2 import VerifyPublicInput, VerifyResult
from gen.axiom_context import AxiomContext

from nodes.paseto_util import parse_token_header, bytes_to_claims_json


def verify_public(ax: AxiomContext, input: VerifyPublicInput) -> VerifyResult:
    """Cryptographically verify the signature of a v1-v4 "public" (asymmetric)
    PASETO token against a caller-supplied PEM public key. Never checks claim
    content or the system clock - use ValidateClaims/CheckExpiration on the result.
    """
    try:
        h = parse_token_header(input.token)
    except ValueError as err:
        return VerifyResult(error=str(err))

    if h.purpose != "public":
        return VerifyResult(version=h.version, error='this is a "local" token; use VerifyLocal instead')

    implicit_assertion = (input.implicit_assertion or "").encode("utf-8")
    if h.version in (1, 2) and implicit_assertion:
        return VerifyResult(
            version=h.version,
            error=f"v{h.version} tokens do not support an implicit assertion (v3/v4 only); "
            "the platform would silently ignore it, so this is rejected explicitly rather than "
            "misleading the caller into thinking it was checked",
        )

    pem = (input.public_key_pem or "").strip()
    if not pem.startswith("-----BEGIN"):
        return VerifyResult(version=h.version, error="public_key_pem must be a PEM-encoded key (missing -----BEGIN header)")

    try:
        key = Key.new(h.version, "public", pem)
    except (ValueError, PysetoError) as err:
        return VerifyResult(version=h.version, error=f"invalid public key material for v{h.version}.public: {err}")

    try:
        token = pyseto.decode(key, input.token, implicit_assertion=implicit_assertion, deserializer=None)
    except (ValueError, PysetoError) as err:
        return VerifyResult(version=h.version, valid=False, reason=str(err))
    except Exception as err:  # noqa: BLE001 - never leak a raw traceback for attacker-controlled input
        return VerifyResult(version=h.version, valid=False, reason=f"signature verification failed: {err}")

    footer_json = ""
    if token.footer:
        footer_bytes = token.footer if isinstance(token.footer, bytes) else str(token.footer).encode("utf-8")
        footer_json = bytes_to_claims_json(footer_bytes)

    payload_bytes = token.payload if isinstance(token.payload, bytes) else str(token.payload).encode("utf-8")
    return VerifyResult(
        valid=True,
        version=h.version,
        claims_json=bytes_to_claims_json(payload_bytes),
        footer_json=footer_json,
        reason="verified",
    )