"""Counterfactuals and boundaries, run against the shipped jwt.py.

Every number in the commentary that is not in demo.py comes from here.
Keep this runnable: a counterfactual whose script was deleted is a claim
nobody can re-check.

    python3 sweep.py
"""

import hmac
import json
import statistics
import time

import jwt

NOW = 1_700_000_000
HONEST = {"sub": "alice", "role": "user", "exp": NOW + 3600}
GREEDY = {"sub": "alice", "role": "admin", "exp": NOW + 86400 * 365}

honest = jwt.encode(HONEST, "RS256", jwt.PRIVATE_KEY)
f_none = jwt.encode(GREEDY, "none", None)
f_conf = jwt.encode(GREEDY, "HS256", jwt.PUBLIC_KEY)


def r(x):
    return "ADMIN" if x and x["role"] == "admin" else ("ok" if x else "reject")


def head(title):
    print()
    print(title)
    print("-" * len(title))


head("C1  Is the verification key even an input? (alg:none)")
for label, key in (("real public key", jwt.PUBLIC_KEY),
                   ("empty bytes", b""),
                   ("garbage", b"not a key at all"),
                   ("the HS256 secret", jwt.SECRET)):
    print(f"  verify_naive(alg:none, key={label:<18}) -> "
          f"{r(jwt.verify_naive(f_none, key, NOW))}")


head("C2  The blocklist and its folklore bypass: does 'nOnE' work?")


def verify_normalising(token, key, now=0):
    """v1, but with a parser that is liberal in what it accepts:
    canonicalise `alg` before dispatch, and blocklist the exact string.
    """
    header, payload, sig, si = jwt.split(token)
    if header["alg"] == "none":
        return None
    alg = header["alg"].strip().upper()
    if alg == "NONE":
        ok = sig == b""
    elif alg == "HS256":
        ok = hmac.compare_digest(jwt.hs256(si, key), sig)
    elif alg == "RS256":
        ok = jwt.rs256_check(si, sig, key)
    else:
        return None
    return payload if ok else None


for variant in ("none", "nOnE", "NONE", " none", "none "):
    h = jwt.b64e(jwt._json({"alg": variant, "typ": "JWT"}))
    tok = f"{h}.{jwt.b64e(jwt._json(GREEDY))}."
    print(f"  alg={variant!r:<9} v1 as shipped -> "
          f"{r(jwt.verify_blocklist(tok, jwt.PUBLIC_KEY, NOW)):<7}"
          f" v1 + normalising parser -> {r(verify_normalising(tok, jwt.PUBLIC_KEY, NOW))}")


head("C3  Key confusion: how exact must the attacker's key bytes be?")
base = jwt.PUBLIC_KEY
for label, kb in (
        ("exact bytes", base),
        ("trailing newline", base + b"\n"),
        ("one digit off", base.replace(b"3351", b"3352", 1)),
        ("keys reordered", json.dumps({"kty": "RSA", "e": jwt.E, "n": jwt.N},
                                      separators=(",", ":")).encode()),
        ("spaces after colons", json.dumps({"e": jwt.E, "kty": "RSA", "n": jwt.N},
                                           sort_keys=True).encode()),
        ("modulus alone, decimal", str(jwt.N).encode())):
    tok = jwt.encode(GREEDY, "HS256", kb)
    print(f"  attacker uses {label:<24} -> v2 allowlist "
          f"{r(jwt.verify_allowlist(tok, base, NOW))}")


head("C4  Boundary: an HS256-only deployment (the verification key is secret)")
print(f"  honest HS256 token, v0            -> "
      f"{r(jwt.verify_naive(jwt.encode(HONEST, 'HS256', jwt.SECRET), jwt.SECRET, NOW))}")
for label, guess in (("attacker guesses ''", b""),
                     ("attacker guesses 'secret'", b"secret"),
                     ("attacker tries PUBLIC_KEY", jwt.PUBLIC_KEY),
                     ("attacker knows the secret", jwt.SECRET)):
    tok = jwt.encode(GREEDY, "HS256", guess)
    print(f"  {label:<33} -> v2 allowlist "
          f"{r(jwt.verify_allowlist(tok, jwt.SECRET, NOW))}")
print(f"  but alg:none still walks in       -> v0 "
      f"{r(jwt.verify_naive(f_none, jwt.SECRET, NOW))}"
      f"  v1 {r(jwt.verify_blocklist(f_none, jwt.SECRET, NOW))}")


head("C5  Is 'the server chooses' enough, or must the alg match the key?")
print(f"  v3 strict, server says RS256, key=PUBLIC -> "
      f"{r(jwt.verify_strict(f_conf, jwt.PUBLIC_KEY, 'RS256', NOW))}")
print(f"  v3 strict, server says HS256, key=PUBLIC -> "
      f"{r(jwt.verify_strict(f_conf, jwt.PUBLIC_KEY, 'HS256', NOW))}")
print(f"  v3 strict, server says HS256, key=SECRET -> "
      f"{r(jwt.verify_strict(f_conf, jwt.SECRET, 'HS256', NOW))}")


head("C6  Under every verifier, is the header data or instruction?")
_h, p, s = f_conf.split(".")
for label, header in (("edit typ to JWS", {"alg": "HS256", "typ": "JWS"}),
                      ("relabel alg to RS256", {"alg": "RS256", "typ": "JWT"})):
    tok = f"{jwt.b64e(jwt._json(header))}.{p}.{s}"
    print(f"  {label:<24} v0 {r(jwt.verify_naive(tok, jwt.PUBLIC_KEY, NOW)):<8}"
          f"v2 {r(jwt.verify_allowlist(tok, jwt.PUBLIC_KEY, NOW))}")


head("C7  The expiry check: `if 'exp' in payload` vs `payload.get('exp', 0)`")
forever = jwt.encode({"sub": "alice", "role": "admin"}, "HS256", jwt.SECRET)
expired = jwt.encode({"sub": "alice", "role": "admin", "exp": NOW - 1},
                     "HS256", jwt.SECRET)
LATER = NOW + 86400 * 3650
print(f"  signed, no `exp` claim, v0 (`in`)   -> {r(jwt.verify_naive(forever, jwt.SECRET, LATER))}")
print(f"  signed, no `exp` claim, v3 (`.get`) -> "
      f"{r(jwt.verify_strict(forever, jwt.SECRET, 'HS256', LATER))}")
print(f"  signed, exp in the past, v0         -> {r(jwt.verify_naive(expired, jwt.SECRET, NOW))}")
print(f"  ten years after issue, v0           -> {r(jwt.verify_naive(forever, jwt.SECRET, LATER))}")


head("C8  `==` vs hmac.compare_digest: does the timing attack reproduce?")
REPS, TRIALS = 4000, 9
si = b"a" * 64
good = jwt.hs256(si, jwt.SECRET)


def bench(op, cand):
    samples = []
    for _ in range(TRIALS):
        t0 = time.perf_counter_ns()
        for _ in range(REPS):
            op(good, cand)
        samples.append((time.perf_counter_ns() - t0) / REPS)
    return statistics.median(samples)


eq = lambda a, b: a == b            # noqa: E731 -- the whole point is the operator
far = b"\x00" * 32                              # 0 leading bytes match
near = bytes([good[0]]) + b"\x00" * 31          # 1
allbut = good[:31] + bytes([good[31] ^ 1])      # 31
full = bytes(bytearray(good))                   # 32: equal, NOT the same object
print(f"  good is full -> {good is full}, good == full -> {good == full}")
print(f"  {'candidate':<26}{'==  ns/op':>12}{'compare_digest':>17}")
rows = []
for label, cand in (("0 leading bytes match", far),
                    ("1 leading byte matches", near),
                    ("31 leading bytes match", allbut),
                    ("32 = full match", full)):
    a, b = bench(eq, cand), bench(hmac.compare_digest, cand)
    rows.append(a)
    print(f"  {label:<26}{a:>12.2f}{b:>17.2f}")
print(f"  ({REPS} reps x {TRIALS} trials, median of trials)")
print(f"  spread across prefix lengths, `==`: {max(rows) - min(rows):.2f} ns/op"
      f" ({(max(rows) - min(rows)) / min(rows) * 100:.1f}%)")

# A single boolean off a timing run is not a finding, so repeat it. If `==`
# really leaked a per-byte signal, every round would come out monotone.
ROUNDS = 10
REPS, TRIALS = 2000, 5
mono = 0
for _ in range(ROUNDS):
    r4 = [bench(eq, c) for c in (far, near, allbut, full)]
    mono += (r4 == sorted(r4))
print(f"  monotone in prefix length, over {ROUNDS} fresh rounds: {mono}/{ROUNDS}")


head("C9  What each forgery costs, in secret bits")
print(f"  honest RS256 signature needs D            : {jwt.D.bit_length()} bits of secret")
print(f"  alg:none forgery needs                    : 0 bits of secret")
print(f"  key confusion needs PUBLIC_KEY, {len(jwt.PUBLIC_KEY) * 8} bits")
print(f"    of material published on purpose        : 0 bits of secret")
print(f"  HS256-only forgery needs SECRET           : {len(jwt.SECRET) * 8} bits of secret")
print(f"  bytes of the accepted key-confusion token written by the attacker: "
      f"{len(f_conf)} of {len(f_conf)}")


head("C10 base64 is not encryption")
print(f"  peek(honest RS256 token), no key at all:")
print(f"    header  {jwt.peek(honest)[0]}")
print(f"    payload {jwt.peek(honest)[1]}")
print(f"  bytes of that token that are ciphertext: 0 of {len(honest)}")
