JWT, JWS and JWE Deep Dive

mkdir -p ~/ib-labs/ib-token-lab
cd ~/ib-labs/ib-token-lab
python3.11 -m venv .venv
source .venv/bin/activate
pip install jwcrypto==1.5.6

# The direct grant (Resource Owner Password Credentials) is used here purely
# for lab convenience to obtain a token without a browser round trip. It is
# deprecated for production use in OAuth 2.1; Lab 04's authorization code
# with PKCE is the correct production flow.

curl -sk -X POST \
  "https://localhost:8443/realms/northgate/protocol/openid-connect/token" \
  -d "client_id=hr-portal" \
  -d "grant_type=password" \
  -d "username=asmith" \
  -d "password=REPLACE_WITH_ASMITH_PASSWORD" \
  -d "scope=openid profile" \
  | jq -r '.access_token' > access_token.txt

cat access_token.txt

# A helper that adds Base64url padding and decodes a single segment.
# JWT uses Base64url without padding; standard base64 -d needs padding
# restored, and the URL-safe alphabet (-_ instead of +/) translated.
b64url_decode() {
  local input="$1"
  local pad=$(( 4 - ${#input} % 4 ))
  [ $pad -ne 4 ] && input="${input}$(printf '=%.0s' $(seq 1 $pad))"
  echo "$input" | tr '_-' '/+' | base64 -d
}

TOKEN="$(cat access_token.txt)"
HEADER="$(echo "$TOKEN" | cut -d. -f1)"
PAYLOAD="$(echo "$TOKEN" | cut -d. -f2)"

echo "=== HEADER ==="
b64url_decode "$HEADER" | jq .

echo "=== PAYLOAD (note: fully readable, no decryption needed) ==="
b64url_decode "$PAYLOAD" | jq .

curl -sk \
  "https://localhost:8443/realms/northgate/protocol/openid-connect/certs" \
  | jq . > jwks.json

# List the key IDs and their intended use
jq '.keys[] | {kid: .kid, use: .use, alg: .alg}' jwks.json

# verify_token.py
# Purpose: verify a Keycloak JWS access token against the realm JWKS,
# pinning the algorithm and checking core claims.

import json
import sys
import time
from jwcrypto import jwt, jwk

EXPECTED_ISS = "https://localhost:8443/realms/northgate"
# The audience your resource server expects. For a Keycloak access token this
# is often the "account" client or your API client id; adjust to match what
# you saw in the aud claim in Step 1.3.
EXPECTED_AUD = "account"

with open("access_token.txt") as f:
    token = f.read().strip()

with open("jwks.json") as f:
    jwks = jwk.JWKSet.from_json(f.read())

# jwcrypto selects the key from the set by matching the token header kid.
# We pin the algorithm to RS256 explicitly: the verifier decides the
# algorithm, never the attacker-controlled token header.
verified = jwt.JWT(
    jwt=token,
    key=jwks,
    algs=["RS256"],
)

claims = json.loads(verified.claims)

# Signature is valid at this point. Now enforce claim checks that a valid
# signature alone does not guarantee.
now = int(time.time())
errors = []
if claims.get("iss") != EXPECTED_ISS:
    errors.append(f"issuer mismatch: {claims.get('iss')}")
aud = claims.get("aud")
aud_list = aud if isinstance(aud, list) else [aud]
if EXPECTED_AUD not in aud_list:
    errors.append(f"audience mismatch: {aud}")
if claims.get("exp", 0) < now:
    errors.append("token expired")

if errors:
    print("REJECTED:", "; ".join(errors))
    sys.exit(1)

print("VERIFIED. Signature valid and claims accepted.")
print(f"Subject: {claims.get('sub')}")
print(f"Username: {claims.get('preferred_username')}")
print(f"Expires in: {claims['exp'] - now} seconds")

python3.11 verify_token.py

# Decode the payload, change a claim, re-encode, and reassemble a token
# with the ORIGINAL signature. Verification must fail.
python3.11 - <<'PYEOF'
import base64, json
def b64url(data): return base64.urlsafe_b64encode(data).rstrip(b'=').decode()
def b64url_dec(s):
    s += '=' * (-len(s) % 4)
    return base64.urlsafe_b64decode(s)

token = open("access_token.txt").read().strip()
h, p, s = token.split('.')
payload = json.loads(b64url_dec(p))
payload["preferred_username"] = "attacker"   # tamper
p_new = b64url(json.dumps(payload).encode())
open("tampered_token.txt", "w").write(f"{h}.{p_new}.{s}")
print("Tampered token written.")
PYEOF

# Now verify the tampered token by temporarily pointing the verifier at it
cp access_token.txt access_token.bak
cp tampered_token.txt access_token.txt
python3.11 verify_token.py || echo "As expected: verification failed on tampered token."
cp access_token.bak access_token.txt   # restore original

openssl genpkey -algorithm RSA -pkeyopt rsa_keygen_bits:2048 \
  -out jwe_private.pem
openssl pkey -in jwe_private.pem -pubout -out jwe_public.pem
ls -l jwe_private.pem jwe_public.pem

# make_jwe.py
# Purpose: take the signed access token (a JWS) and encrypt it into a JWE,
# producing a nested token. The signed token becomes the plaintext, so its
# claims are no longer readable without the recipient's private key.

from jwcrypto import jwk, jwe

with open("access_token.txt") as f:
    signed_token = f.read().strip()

with open("jwe_public.pem", "rb") as f:
    public_key = jwk.JWK.from_pem(f.read())

# alg RSA-OAEP-256 protects the content encryption key with the recipient's
# public RSA key; enc A256GCM encrypts the actual content with AES-256-GCM,
# an authenticated cipher. cty JWT signals that the plaintext is itself a JWT.
protected_header = {
    "alg": "RSA-OAEP-256",
    "enc": "A256GCM",
    "cty": "JWT",
}

token = jwe.JWE(
    plaintext=signed_token.encode("utf-8"),
    protected=protected_header,
)
token.add_recipient(public_key)
compact = token.serialize(compact=True)

with open("nested_token.txt", "w") as f:
    f.write(compact)

parts = compact.split(".")
print(f"JWE created with {len(parts)} parts (compact JWE has 5).")
print("First part (protected header, still Base64url readable):")
import base64, json
hdr = parts[0] + "=" * (-len(parts[0]) % 4)
print(json.dumps(json.loads(base64.urlsafe_b64decode(hdr)), indent=2))
print("\nSecond part onward is ciphertext: the claims are no longer readable.")

python3.11 make_jwe.py

# decrypt_jwe.py
# Purpose: decrypt the nested JWE with the recipient private key and recover
# the original signed JWS, which can then be verified exactly as in Phase 2.

from jwcrypto import jwk, jwe

with open("nested_token.txt") as f:
    nested = f.read().strip()

with open("jwe_private.pem", "rb") as f:
    private_key = jwk.JWK.from_pem(f.read())

token = jwe.JWE()
token.deserialize(nested, key=private_key)
recovered_jws = token.payload.decode("utf-8")

with open("recovered_token.txt", "w") as f:
    f.write(recovered_jws)

print("Decryption succeeded. Recovered inner token matches original:")
print(recovered_jws == open("access_token.txt").read().strip())

python3.11 decrypt_jwe.py