#!/usr/bin/env python
"""
Script to perform different tests against a OAUTH 2.x Authorization Server or OpenID Provider
in order to identity interesting elements that can be part of its attack surface.

Each function named "test_*" is dedicated to a test.

The approach decided was to not mandatory count on the information from the "/.well-known/openid-configuration" file and been able
to spot elements even if the descriptor file is not reacheable.

Implement is empirical, the script was created first based on KEYCLOAK behavior and then is enhanced when I face other STS during my work/spare time.
Support adding for the new STS is made in the most "clean" way in which I try to isolate specific STS processing in specific utility functions
in order to not pollute test functions

The value "CANNOT_EXISTS_CANARY" is used as a "canary" to detect if the STS expose or not a discrepancy factor that can be leveraged
for different enumeration operations:
    See https://www.fortinet.com/resources/cyberglossary/what-is-canary-in-cybersecurity
    See https://cwe.mitre.org/data/definitions/204.html
    See https://cwe.mitre.org/data/definitions/203.html

Tests are based on the following research:
    https://github.com/ExcelliumSA/OAuth2OIDC-Study

The tests list is here:
    https://github.com/ExcelliumSA/OAuth2OIDC-Study/blob/main/OAauth2_OIDC_Security_Validations.md

Dependencies:
    pip3 install requests termcolor colorama tabulate pyJWT pycryptodome

Sources used:
    https://connect2id.com/learn
    https://auth0.com/docs/flows
    https://auth0.com/docs/protocols/protocol-oauth2
    https://www.keycloak.org/docs-api/9.0/javadocs/constant-values.html
    https://github.com/keycloak/keycloak-documentation/blob/master/securing_apps/topics/oidc/javascript-adapter.adoc
    https://datatracker.ietf.org/doc/html/rfc7636#section-4.1
    https://datatracker.ietf.org/doc/html/rfc7636#section-4.2
    https://openid.net/specs/openid-connect-core-1_0.html#rfc.section.6.2
    https://portswigger.net/research/hidden-oauth-attack-vectors
    https://docs.wso2.com/display/IS570/OpenID+Connect+Discovery
    https://openid.net/specs/openid-connect-discovery-1_0.html#IssuerDiscovery
    https://openid.net/specs/openid-connect-core-1_0.html
    https://auth0.com/docs/flows/hybrid-flow
    https://openid.net/specs/openid-connect-core-1_0.html#ClaimsParameter
    https://www.scottbrady91.com/OpenID-Connect/OpenID-Connect-Flows

Script was tested against the following OAUTH/OIDC systems:
    KEYCLOAK: https://www.keycloak.org/getting-started/getting-started-docker
    WSO2 API MANAGER: https://wso2.com/integration/install/docker/community/get-started/

"""
import requests
import re
import argparse
import colorama
import jwt
import uuid
import statistics
import os.path
from termcolor import colored
from tabulate import tabulate
from urllib.parse import urlparse, unquote
from cryptography.hazmat.primitives import serialization

# Config
# Disable TLS warning when validation is disabled when requests is used
requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning)

# Constants
PROXIES = {}
TIMEOUT = 5
TLS_CLIENT_CERT = None
COMMON_SCOPES = ["address", "email", "openid", "phone", "profile", "offline_access", "microprofile-jwt", "role_list", "roles", "web-origins", "aisp", "pisp", "extended_transaction_history", "CANNOT_EXISTS_CANARY"]
COMMON_CLIENTID = ["account", "account-console", "admin-cli", "broker", "realm-management", "security-admin-console", "CANNOT_EXISTS_CANARY"]
COMMON_REALMS = ["master", "CANNOT_EXISTS_CANARY"]
COMMON_ISSUER_RESOURCES = ["admin", "CANNOT_EXISTS_CANARY"]
COMMON_AUDIENCES = ["api", "internal", "admin", "CANNOT_EXISTS_CANARY"]
OIDC_METADATA_ENDPOINT_RELATIVE_PATH = ".well-known/openid-configuration"
OIDC_PROVIDER_ISSUER_LOCATION_ENDPOINT_RELATIVE_PATH = ".well-known/webfinger?resource=[RESSOURCE]&rel=http%3A%2F%2Fopenid.net%2Fspecs%2Fconnect%2F1.0%2Fissuer"
COMMON_RESPONSE_MODES = ["query", "fragment", "form_post", "web_message", "CANNOT_EXISTS_CANARY"]


def render_results(data, column_headers=["Key", "Value"]):
    """
    Print result depending on data structure.
    """
    if isinstance(data, list) and len(data) > 0:
        print(", ".join(data))
    elif isinstance(data, dict) and len(data) > 0:
        table_rows = []
        for key, value in data.items():
            if isinstance(value, list):
                table_rows.append([key, ", ".join(value)])
            else:
                table_rows.append([key, value])
        print(tabulate(table_rows, headers=column_headers, tablefmt="github"))
    elif (isinstance(data, list) or isinstance(data, dict)) and len(data) == 0:
        print("No data found.")
    elif data is not None:
        print(data)


def get_requests_session():
    """
    Preconfigure a 'Requests' session to speed-up HTTP interactions.
    """
    session = requests.Session()
    session.verify = False
    session.proxies.update(PROXIES)
    session.cert = TLS_CLIENT_CERT
    session.headers.update({"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:89.0) Gecko/20100101 Firefox/89.0"})
    return session


def get_page_type(http_response):
    """
    Determine the type of response depending on the OAUTH/OIDC systems specificities and case meet during assessments.
    """
    page_type = "NA"
    # KEYCLOAK
    if http_response.status_code == 200 and ("sign in" in http_response.text.lower() or "log in" in http_response.text.lower()):
        page_type = "LOGIN"
    elif http_response.status_code == 400 and "invalid parameter: redirect_uri" in http_response.text.lower():
        page_type = "INVALID_REDIRECT_URI"
    elif http_response.status_code == 400 and ("invalid request" in http_response.text.lower() or "invalid_request" in http_response.text.lower()):
        page_type = "SERVER_SIDE_ERROR"
    elif http_response.status_code == 200 and http_response.headers["Content-Type"] == "application/json" and "realm" in http_response.json() and "token-service" in http_response.json():
        page_type = "REALM_DESCRIPTOR"
    # WSO2 API MANAGER
    elif http_response.status_code == 302 and "registered+callback+does+not+match+with+the+provided+url" in http_response.headers["Location"].lower():
        page_type = "INVALID_REDIRECT_URI"
    elif http_response.status_code == 302 and "/authenticationendpoint/login.do" in http_response.headers["Location"].lower():
        page_type = "LOGIN"
    elif http_response.status_code == 302 and "invalid_request" in http_response.headers["Location"].lower():
        page_type = "SERVER_SIDE_ERROR"
    elif http_response.status_code == 200 and "application/jrd+json" in http_response.headers["Content-Type"] and "subject" in http_response.json():
        page_type = "OPENID_ISSUER_LOCATION_CONFIGURATION"
    # SPECIFIC TO LUXEMBURG APP
    elif http_response.status_code == 302 and "luxtrust" in http_response.headers["Location"].lower():
        page_type = "LOGIN"
    # GENERIC
    elif http_response.status_code == 200 and http_response.headers["Content-Type"] == "application/json" and "grant_types_supported" in http_response.json():
        page_type = "OPENID_CONFIGURATION"
    elif http_response.status_code == 302 and "error_description=" in http_response.headers["Location"].lower():
        page_type = "SERVER_SIDE_ERROR"
    elif http_response.status_code == 403:
        page_type = "ACCESS_DENIED"
    return page_type


def get_system_software(base_url):
    """
    Try ty identify the software operating the OAUTH/OIDC systems depending on the OAUTH/OIDC softwares specificities.
    """
    url_components = urlparse(base_url)
    software_name = "NA"
    with get_requests_session() as session:
        # KEYCLOAK: Check the presence of the JS library
        resp = session.get(url=f"{url_components.scheme}://{url_components.netloc}/auth/js/keycloak.js", allow_redirects=False, timeout=TIMEOUT)
        if resp.status_code == 200 and "text/javascript" in resp.headers["Content-Type"]:
            software_name = "Keycloak"
        # WSO2 API MANAGER: Check the version endpoint
        if software_name == "NA":
            session.cookies.clear()
            resp = session.get(url=f"{url_components.scheme}://{url_components.netloc}/services/Version", allow_redirects=False, timeout=TIMEOUT)
            if resp.status_code == 200 and "wso2 api manager" in resp.text.lower():
                software_name = "WSO2 API Manager"
            else:
                resp = session.get(url=f"{url_components.scheme}://{url_components.netloc}/", allow_redirects=True, timeout=TIMEOUT)
                if resp.status_code == 200 and "wso2 apim" in resp.text.lower():
                    software_name = "WSO2 API Manager"
    return software_name


def adapt_dictionaries_for_sts(base_url, words_list, client_id):
    """
    Handle specific STS cases for words list and others embedded dictionaries based on the OAUTH/OIDC softwares specificities.
    The goal is to prevent to pollute test functions with STS specific code...
    """
    adapted_words_list = []
    adapted_words_list.extend(words_list)
    if get_system_software(base_url) == "WSO2 API Manager":
        COMMON_REALMS.clear()
        adapted_words_list.clear()
        adapted_words_list.append(client_id)
    return adapted_words_list


def get_protected_app_back_link(http_response):
    """
    Extract the URL of the protected application depending on the OAUTH/OIDC systems specificities.
    """
    back_link = "NA"
    # KEYCLOAK
    if http_response.status_code == 400 and "backToApplication" in http_response.text:
        # Example: <a id="backToApplication" href="http://localhost:9500/App.html">Back to Application</a>
        back_link = re.findall(r'<a\sid="backToApplication"\shref="(.*)"\s?>', http_response.text)[0]
    # WSO2 API MANAGER
    # Not possible because the error page do not contains any reference to the protected app url/name/...
    return back_link


def get_realms_base_url(base_url):
    """
    Determine the base url for REALMS depending on the OAUTH/OIDC systems specificities.
    """
    base_url_realms = base_url
    # KEYCLOAK
    marker = "/realms/"
    if marker in base_url_realms:
        base_url_realms = base_url_realms[0:base_url_realms.index(marker) + (len(marker)-1)]
    elif get_system_software(base_url) == "WSO2 API Manager":
        # WSO2 API MANAGER:
        # There is a specific case for this one, at least as I have understood reading the doc and using it.
        # The "REALM" is represented by the current URL and there is not reference to the REALM in the URL
        # like it is the case for KEYCLOAK
        # See https://docs.wso2.com/display/IS570/OpenID+Connect+Discovery
        url_components = urlparse(base_url)
        base_url_realms = f"{url_components.scheme}://{url_components.netloc}/oauth2/oidcdiscovery"

    return base_url_realms


def get_realm_oidc_metadata_edp_url(base_url_realms, realm_name):
    """
    Create the full url to the REALM OIDC metadata endpoint depending on the OAUTH/OIDC systems specificities.
    The goal is to try to prevent to pollute test functions with STS specific cases...
    """
    # See https://openid.net/specs/openid-connect-discovery-1_0.html#ProviderConfig
    edp = f"{base_url_realms}/{realm_name}/{OIDC_METADATA_ENDPOINT_RELATIVE_PATH}"
    if get_system_software(base_url) == "WSO2 API Manager":
        # WSO2 API MANAGER:
        # There is a specific case for this one, at least as I have understood reading the doc and using it.
        # The "REALM" is represented by the current URL and there is not reference to the REALM in the URL
        # like it is the case for KEYCLOAK
        # See https://docs.wso2.com/display/IS570/OpenID+Connect+Discovery
        edp = f"{base_url_realms}/{OIDC_METADATA_ENDPOINT_RELATIVE_PATH}"

    return edp


def get_realm_oidc_provider_issuer_location_edp_url(base_url_realms, realm_name, issuer_resource):
    """
    Create the full url to the REALM OIDC provider issuer location endpoint depending on the OAUTH/OIDC systems specificities.
    The goal is to try to prevent to pollute test functions with STS specific cases...
    """
    # See https://openid.net/specs/openid-connect-discovery-1_0.html#IssuerDiscovery
    part = OIDC_PROVIDER_ISSUER_LOCATION_ENDPOINT_RELATIVE_PATH.replace("[RESSOURCE]", issuer_resource)
    edp = f"{base_url_realms}/{realm_name}/{part}"
    if get_system_software(base_url) == "WSO2 API Manager":
        # WSO2 API MANAGER:
        # There is a specific case for this one, at least as I have understood reading the doc and using it.
        # The "REALM" is represented by the current URL and there is not reference to the REALM in the URL
        # like it is the case for KEYCLOAK
        # See https://docs.wso2.com/display/IS570/OpenID+Connect+Discovery
        url_components = urlparse(base_url)
        edp = f"{url_components.scheme}://{url_components.netloc}/{part}"

    return edp


def disable_existing_parameters(base_url, parameter_list_to_disable=["client_id", "response_type", "code_challenge", "code_challenge_method", "scope"]):
    """
    Disable any existing parameter present in the base url present into the provided list. Prevent collisions with parameters added by test functions.
    """
    for param_name in parameter_list_to_disable:
        base_url = base_url.replace(f"{param_name}=", f"xxx_{param_name}=")
    return base_url


def test_enabled_flows(base_url):
    """
    Identify all enabled OAUTH flows that can be initialized without having Client Credentials.
    """
    # Prevent parameter collisions
    work_url = disable_existing_parameters(base_url, ["response_type", "code_challenge", "code_challenge_method", "nonce"])
    results = {}
    # Define the url parameters for each type of flow
    flows = {}
    flows["IMPLICIT"] = "response_type=token"
    flows["HYBRID_WITHOUT_NONCE"] = "response_type=code%20id_token"
    flows["HYBRID_WITH_NONCE"] = f"response_type=code%20id_token&nonce={str(uuid.uuid4())}"
    flows["AUTHORIZATION_CODE"] = "response_type=code"
    flows["AUTHORIZATION_CODE_WITH_PKCE_WITH_S256_CHALLENGE_HASH"] = "response_type=code&code_challenge=-sUEoAV-txYvhniiuJ4-gwNCtsiD2XiIPvLQYm-sUsE&code_challenge_method=S256"
    flows["AUTHORIZATION_CODE_WITH_PKCE_WITH_PLAIN_CHALLENGE"] = f"response_type=code&code_challenge={'x'*43}&code_challenge_method=plain"
    # Test flows enabling state
    with get_requests_session() as session:
        for flow in flows:
            session.cookies.clear()
            try:
                resp = session.get(url=f"{work_url}&{flows[flow]}", allow_redirects=False, timeout=TIMEOUT)
            except:
                results[flow] = False
                continue
            # If flow is disabled then an HTTP 30x is received with error in the redirection url.
            # Otherwise it should return a login page because we do not provide any existing session cookie
            results[flow] = (get_page_type(resp) == "LOGIN")
    return results


def test_allowed_scopes(base_url, scope_list_source):
    """
    Identify all allowed scopes for the current client ID.
    """
    # Prevent parameter collisions
    work_url = disable_existing_parameters(base_url, ["response_type", "scope", "code_challenge", "code_challenge_method"])
    results = []
    # Add common scopes to the provided list
    scopes = list(set(scope_list_source + COMMON_SCOPES))
    # Use the Authorization Code flow because it is commonly enabled
    with get_requests_session() as session:
        for scope in scopes:
            print(f"\rTesting scope: {scope:<50}", end="", flush=True)
            session.cookies.clear()
            try:
                resp = session.get(url=f"{work_url}&response_type=code&scope={scope}", allow_redirects=False, timeout=TIMEOUT)
            except:
                continue
            # If scope is not allowed then an HTTP 30x is received with error in the redirection url.
            # Otherwise it should return a login page because we do not provide any existing session cookie.
            if get_page_type(resp) == "LOGIN":
                results.append(scope)
    print(f"\r{' '*50}", end="", flush=True)
    print("\r", end="", flush=True)
    # If the canary is in the list of allowed element so it indicates that the STS has no discrepancy factor
    if "CANNOT_EXISTS_CANARY" in results:
        print("\rSTS has no discrepancy factor! ", end="", flush=True)
        results.clear()
    results.sort()
    return results


def test_existing_clients(base_url, clientid_list_source):
    """
    Identify all configured clients based on their client ID.
    """
    # Prevent parameter collisions
    work_url = disable_existing_parameters(base_url, ["client_id", "response_type", "redirect_uri", "scope", "code_challenge", "code_challenge_method"])
    results = []
    # Add common client id to the provided list
    client_ids = list(set(clientid_list_source + COMMON_CLIENTID))
    # Use the Authorization Code flow because it is commonly enabled
    with get_requests_session() as session:
        for client_id in client_ids:
            print(f"\rTesting ID: {client_id:<50}", end="", flush=True)
            session.cookies.clear()
            resp = session.get(url=f"{work_url}&client_id={client_id}&response_type=code&scope=openid&redirect_uri=http://explicit-invalid.com", allow_redirects=False, timeout=TIMEOUT)
            # If the client do not exist then an HTTP 400 is returned with the text "Client not found".
            # If the client exist then an HTTP 400 is returned with the text "Invalid parameter: redirect_uri" because the redirect_uri specified is explictly invalid for the client ID.
            if get_page_type(resp) == "INVALID_REDIRECT_URI":
                results.append(client_id)
    print(f"\r{' '*50}", end="", flush=True)
    print("\r", end="", flush=True)
    # If the canary is in the list of allowed element so it indicates that the STS has no discrepancy factor
    if "CANNOT_EXISTS_CANARY" in results:
        print("\rSTS has no discrepancy factor! ", end="", flush=True)
        results.clear()
    results.sort()
    return results


def test_allowed_localhost_redirection(base_url):
    """
    Identify if a localhost redirection is allowed for the current client ID.
    """
    # Prevent parameter collisions
    work_url = disable_existing_parameters(base_url, ["response_type", "redirect_uri", "scope", "code_challenge", "code_challenge_method"])
    results = []
    # Use the Authorization Code flow because it is commonly enabled
    with get_requests_session() as session:
        for port in range(1, 65536):
            for protocol in ["http", "https"]:
                for host in ["localhost", "127.0.0.1"]:
                    redirection_url = f"{protocol}://{host}:{port}"
                    print(f"\rTesting redirection url: {redirection_url:<50}", end="", flush=True)
                    session.cookies.clear()
                    try:
                        resp = session.get(url=f"{work_url}&response_type=code&scope=openid&redirect_uri={redirection_url}", allow_redirects=False, timeout=TIMEOUT)
                    except:
                        continue
                    # If the redirection is accepted then the login page should be presented with an HTTP 200 because we do not provide any existing session cookie.
                    if get_page_type(resp) == "LOGIN":
                        results.append(redirection_url)
    print(f"\r{' '*50}", end="", flush=True)
    print("\r", end="", flush=True)
    results.sort()
    return results


def test_map_client_to_protected_app(base_url, clientid_list_source):
    """
    Identify the protected application associated to each client of the provided list by abusing the "redirect_uri" parameter and its associated error page.
    """
    # Prevent parameter collisions
    work_url = disable_existing_parameters(base_url, ["client_id", "response_type", "redirect_uri", "scope", "code_challenge", "code_challenge_method"])
    results = {}
    client_ids = clientid_list_source
    # Use the Authorization Code flow because it is commonly enabled
    with get_requests_session() as session:
        for client_id in client_ids:
            session.cookies.clear()
            try:
                # Exceptionally allow following of redirections here in order to reach and get the STS error page content
                resp = session.get(url=f"{work_url}&client_id={client_id}&response_type=code&scope=openid&redirect_uri=http://explicit-invalid.com", allow_redirects=True, timeout=TIMEOUT)
            except:
                continue
            results[client_id] = get_protected_app_back_link(resp)
    return results


def test_existing_realms(base_url, realms_list_source):
    """
    Identify all configured realms via their associated metadata endpoint.
    """
    # Add common realms to the provided list
    realms = list(set(realms_list_source + COMMON_REALMS))
    # Get base url for realms
    base_url_realms = get_realms_base_url(base_url)
    # Try to enumerate realms
    results = {}
    with get_requests_session() as session:
        for realm in realms:
            session.cookies.clear()
            print(f"\rTesting value: {realm:<50}", end="", flush=True)
            metadata_edp_url = get_realm_oidc_metadata_edp_url(base_url_realms, realm)
            # I meet case in which some realms are protected by dedicated client cert so I handle error cases
            try:
                resp = session.get(url=metadata_edp_url, allow_redirects=False, timeout=TIMEOUT)
            except:
                continue
            if get_page_type(resp) == "OPENID_CONFIGURATION":
                results[realm] = metadata_edp_url
            else:
                # Sometime the access to the OpenID Connect metadata endpoint is blocked but the realm exists.
                # So I try a request on the realm base url without any parameter so that should return the realm descriptor.
                try:
                    realm_base = f"{base_url_realms}/{realm}"
                    resp = session.get(url=realm_base, allow_redirects=False, timeout=TIMEOUT)
                    if get_page_type(resp) == "REALM_DESCRIPTOR":
                        results[realm] = None
                except:
                    continue
    print(f"\r{' '*50}", end="", flush=True)
    print("\r", end="", flush=True)
    # If the canary is in the list of allowed element so it indicates that the STS has no discrepancy factor
    if "CANNOT_EXISTS_CANARY" in results:
        print("\rSTS has no discrepancy factor! ", end="", flush=True)
        results.clear()
    return results


def test_realms_client_self_registration_endpoint(realms_collection_infos):
    """
    Identify the client self-registration endpoint for a collection of realms.
    Take as input, the results of the function test_existing_realms().
    """
    results = {}
    with get_requests_session() as session:
        for realm in realms_collection_infos:
            if realms_collection_infos[realm] is not None:
                json_metadata = session.get(url=realms_collection_infos[realm], allow_redirects=False, timeout=TIMEOUT).json()
                if "registration_endpoint" in json_metadata:
                    results[realm] = json_metadata["registration_endpoint"]
    return results


def test_none_signing_algorithm_support(metadata_json_url):
    """
    Find, in the OpenID configuration settings, ones for which the support of the NONE signature algorithm is enabled.
    """
    results = []
    if metadata_json_url is not None:
        targeted_setting_keys = ["id_token_signing_alg_values_supported", "userinfo_signing_alg_values_supported", "request_object_signing_alg_values_supported",
                                 "token_endpoint_auth_signing_alg_values_supported", "introspection_endpoint_auth_signing_alg_values_supported", "revocation_endpoint_auth_signing_alg_values_supported"]
        with get_requests_session() as session:
            json_data = session.get(url=metadata_json_url, allow_redirects=False, timeout=TIMEOUT).json()
        for setting_key in targeted_setting_keys:
            if setting_key in json_data:
                for algorithm in json_data[setting_key]:
                    if algorithm.lower() == "none":
                        results.append(setting_key)
                        break
        results.sort()
    return results


def test_hmac_signing_algorithm_support_with_public_key_export(metadata_json_url):
    """
    Find, in the OpenID configuration settings, ones for which the support of the HMAC (HS[0-9]{3,4}) signature algorithm is enabled.
    If yes, then, export all public key as PEM format in order to test exposure to "Key Confusion Attack" via the JWT tool from TICARPI.
    See https://github.com/ticarpi/jwt_tool and https://github.com/ticarpi/jwt_tool/wiki/Known-Exploits-and-Attacks#cve-2016-5431---key-confusion-attack
    """
    results = {"SettingsKeys": [], "PublicKeysExportFiles": []}
    if metadata_json_url is None:
        return results
    targeted_setting_keys = ["id_token_signing_alg_values_supported", "userinfo_signing_alg_values_supported", "request_object_signing_alg_values_supported",
                             "token_endpoint_auth_signing_alg_values_supported", "introspection_endpoint_auth_signing_alg_values_supported", "revocation_endpoint_auth_signing_alg_values_supported"]
    session = get_requests_session()
    json_data = session.get(url=metadata_json_url, allow_redirects=False, timeout=TIMEOUT).json()
    # Get the list of settings supporting HMAC signature algorithm
    for setting_key in targeted_setting_keys:
        if setting_key in json_data:
            algorithms = "/".join(json_data[setting_key])
            if len(re.findall(r'HS[0-9]{3,4}', algorithms)) > 0:
                results["SettingsKeys"].append(setting_key)
    # Export public keys used to sign if settings were founds
    if(len(results["SettingsKeys"]) > 0):
        jwks_uri = json_data["jwks_uri"]
        json_data_keys = session.get(url=jwks_uri, allow_redirects=False, timeout=TIMEOUT).json()
        for key in json_data_keys["keys"]:
            public_key = None
            # See https://github.com/jpadilla/pyjwt/blob/master/jwt/algorithms.py#L69
            if key["use"] == "sig" and key["alg"].startswith("RS"):
                public_key = jwt.algorithms.RSAAlgorithm.from_jwk(key)
            elif key["use"] == "sig" and key["alg"].startswith("ES"):
                public_key = jwt.algorithms.ECAlgorithm.from_jwk(key)
            elif key["use"] == "sig" and key["alg"].startswith("PS"):
                public_key = jwt.algorithms.RSAPSSAlgorithm.from_jwk(key)
            elif key["use"] == "sig" and key["alg"].startswith("EdDSA"):
                public_key = jwt.algorithms.Ed25519Algorithm.from_jwk(key)
            if public_key is not None:
                public_key_pem = public_key.public_bytes(encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo).decode("utf-8")
                public_key_pem_file = key["kid"] + ".pem"
                with open(public_key_pem_file, mode="w", encoding="utf-8") as f:
                    f.write(public_key_pem)
                results["PublicKeysExportFiles"].append(public_key_pem_file)
    results["SettingsKeys"].sort()
    results["PublicKeysExportFiles"].sort()
    return results


def test_enabled_response_modes(base_url, responsemodes_list_source):
    """
    Identify all response modes enabled.
    """
    results = []
    # Prevent parameter collisions
    work_url = disable_existing_parameters(base_url, ["response_type", "scope", "response_mode", "code_challenge", "code_challenge_method"])
    # Add common response modes to the provided list
    resp_modes = list(set(responsemodes_list_source + COMMON_RESPONSE_MODES))
    # Use the Authorization Code flow because it is commonly enabled
    with get_requests_session() as session:
        for resp_mode in resp_modes:
            print(f"\rTesting mode: {resp_mode:<50}", end="", flush=True)
            session.cookies.clear()
            try:
                resp = session.get(url=f"{work_url}&response_type=code&scope=openid&response_mode={resp_mode}", allow_redirects=False, timeout=TIMEOUT)
            except:
                continue
            # If the response mode is not supported then an HTTP 302 is returned with the text "Invalid+parameter%3A+response_mode" in the redirection URL
            # Otherwise the login page is returned in a HTTP 200
            if get_page_type(resp) == "LOGIN":
                results.append(resp_mode)
    print(f"\r{' '*50}", end="", flush=True)
    print("\r", end="", flush=True)
    # If the canary is in the list of allowed element so it indicates that the STS has no discrepancy factor
    if "CANNOT_EXISTS_CANARY" in results:
        print("\rSTS has no discrepancy factor! ", end="", flush=True)
        results.clear()
    results.sort()
    return results


def test_weak_pkce_code_verifier_rejection_for_plain_challenge(base_url):
    """
    Verify if weak code verifier value is rejected by the STS when PKCE with Plain Challenge Method is allowed.
    Result (return) is the weakest code challenge value accepted.
    Note that when Plain Challenge Method is allowed then "Code Verifier" == "Code Challenge".
    According to RFC7636, Code Verifier follow this pattern: [A-Za-z0-9\-\._~]{43,128}
    See https://datatracker.ietf.org/doc/html/rfc7636#section-4.1
    See https://datatracker.ietf.org/doc/html/rfc7636#section-4.2
    """
    results = "NA"
    # Prevent parameter collisions
    work_url = disable_existing_parameters(base_url, ["response_type", "scope", "code_challenge", "code_challenge_method"])
    # Use the Authorization Code with PKCE
    # Use [0]{1,129} as weak test value
    with get_requests_session() as session:
        for length in range(1, 129):
            code_challenge_test_value = "0" * length
            print(f"\rTesting value: {code_challenge_test_value:<65}", end="", flush=True)
            session.cookies.clear()
            try:
                resp = session.get(url=f"{work_url}&response_type=code&scope=openid&code_challenge={code_challenge_test_value}&code_challenge_method=plain", allow_redirects=False, timeout=TIMEOUT)
            except:
                continue
            # If the code challenge submitted is rejected then an HTTP 302 is returned with the text "Invalid+parameter%3A+code_challenge" in the redirection URL
            # Otherwise the login page is returned in a HTTP 200
            if get_page_type(resp) == "LOGIN":
                results = f"'0' * {length}"
                break
    print(f"\r{' '*65}", end="", flush=True)
    print("\r", end="", flush=True)
    return results


def test_allowed_localhost_request_uri(base_url, list_local_request_uri_hit_min_delay):
    """
    Identify if a localhost request uri is allowed for the current client ID during a flow init.
    See https://openid.net/specs/openid-connect-core-1_0.html#rfc.section.6.2
    """
    # Prevent parameter collisions
    work_url = disable_existing_parameters(base_url, ["response_type", "scope", "code_challenge", "code_challenge_method", "request_uri"])
    results = []
    response_times = {}
    # Use the Authorization Code flow because it is commonly enabled
    with get_requests_session() as session:
        for port in range(1, 65536):
            for protocol in ["http", "https"]:
                for host in ["localhost", "127.0.0.1"]:
                    request_url = f"{protocol}://{host}:{port}"
                    print(f"\rTesting request url: {request_url:<50}", end="", flush=True)
                    session.cookies.clear()
                    try:
                        resp = session.get(url=f"{work_url}&response_type=code&scope=openid&request_uri={request_url}", allow_redirects=False, timeout=TIMEOUT)
                    except:
                        continue
                    # If the request url is accepted then the login page should be presented with an HTTP 200 because we do not provide any existing session cookie.
                    # Leverage also a detection via the response time because if the hit against the specified URL is allowed but do no return a valid JWT content then a parsing error
                    # is raised by the server and an generic error page is returned (no disclosure of the kind of error meet).
                    if get_page_type(resp) == "LOGIN":
                        results.append(f"{request_url} (identified via the http response type)")
                    elif get_page_type(resp) == "SERVER_SIDE_ERROR":
                        response_times[request_url] = resp.elapsed.microseconds
    print(f"\r{' '*50}", end="", flush=True)
    print("\r", end="", flush=True)
    # Identify potential hidden hit the adjustement response time
    if len(response_times) > 0:
        response_times_collection = list(response_times.values())
        print(colored("[i]", info_color, attrs=["bold"]) + f" Average response time was {int(statistics.mean(response_times_collection)):<6} microseconds.")
        print(colored("[i]", info_color, attrs=["bold"]) + f" Median  response time was {int(statistics.median(response_times_collection)):<6} microseconds.")
        print(colored("[i]", info_color, attrs=["bold"]) + f" Max     response time was {max(response_times_collection):<6} microseconds.")
        # Take all requests for which the response time is superior to the limit defined to identify effective hits by the STS against forged URL
        for response_time in response_times:
            if response_times[response_time] > list_local_request_uri_hit_min_delay:
                results.append(f"{response_time} (identified via the http response time of {response_times[response_time]} microseconds - False positive possible)")
    results.sort()
    return results


def test_oidc_provider_issuer_discovery_support(base_url, realm, usernames_list):
    """
    Identify if the STS support the "OpenID Provider Issuer Discovery" feature and if this one can be abused to perform a end user enumeration operation against the provided REALM.
    """
    # Build a list issuer resources based on the list of username provided
    # See https://openid.net/specs/openid-connect-discovery-1_0.html#Examples
    url_components = urlparse(base_url)
    issuer_resources = []
    usernames_full_list = list(set(usernames_list + COMMON_ISSUER_RESOURCES))
    for username in usernames_full_list:
        # Add format: acct:[USERNAME]@[HOST]
        issuer_resources.append(f"acct%3A{username}%40{url_components.netloc}")
        # Add format : HTTP(S)://[HOST]/[USERNAME]
        issuer_resources.append(f"{url_components.scheme}%3A%2F%2F{url_components.netloc}%2F{username}")
    # Try to enumerate realms
    results = []
    with get_requests_session() as session:
        for issuer_resource in issuer_resources:
            session.cookies.clear()
            print(f"\rTesting value: {issuer_resource:<80}", end="", flush=True)
            oidc_provider_issuer_discovery_edp_url = get_realm_oidc_provider_issuer_location_edp_url(base_url, realm, issuer_resource)
            try:
                resp = session.get(url=oidc_provider_issuer_discovery_edp_url, allow_redirects=False, timeout=TIMEOUT)
            except:
                continue
            if get_page_type(resp) == "OPENID_ISSUER_LOCATION_CONFIGURATION":
                results.append(unquote(issuer_resource))
    print(f"\r{' '*80}", end="", flush=True)
    print("\r", end="", flush=True)
    # If the canary is in the list of allowed element so it indicates that the STS has no discrepancy factor
    if f"acct:CANNOT_EXISTS_CANARY@{url_components.netloc}" in results:
        print("\rSTS has no discrepancy factor! ", end="", flush=True)
        results.clear()
    return results


def test_weak_nonce_rejection_for_hybrid_flow(base_url):
    """
    Verify if weak nonce value is rejected by the STS when Hybrid flow is allowed.
    Result (return) is the weakest nonce value accepted.
    NONCE from the Auth0 doc is a cryptographically random string that your app adds to the initial request and Auth0 includes inside the ID Token.
    NONCE from OIDC spec doc is a parameter value needed to be included per-session state and be unguessable to attackers.
    See https://auth0.com/docs/flows/call-api-hybrid-flow#example-authorization-url
    See https://openid.net/specs/openid-connect-core-1_0.html#IDToken
    """
    results = "NA"
    # Prevent parameter collisions
    work_url = disable_existing_parameters(base_url, ["response_type", "scope", "nonce"])
    # Use the hybrid flow
    # Use [0]{1,10} as weak test value because spec do not fix a format for the NONCE
    with get_requests_session() as session:
        for length in range(1, 11):
            nonce = "0" * length
            print(f"\rTesting value: {nonce:<65}", end="", flush=True)
            session.cookies.clear()
            try:
                resp = session.get(url=f"{work_url}&response_type=code%20id_token&nonce={nonce}&scope=openid", allow_redirects=False, timeout=TIMEOUT)
            except:
                continue
            # If the NONCE submitted is rejected then an HTTP 302 is returned with the text "Invalid+parameter%3A+nonce" in the redirection URL
            # Otherwise the login page is returned
            if get_page_type(resp) == "LOGIN":
                results = f"'0' * {length}"
                break
    print(f"\r{' '*65}", end="", flush=True)
    print("\r", end="", flush=True)
    return results


def test_claims_parameter_support(base_url):
    """
    Verify if the STS supports the parameter "claims" during the init of a flow.
    Result (return) the support status as a human readable string by type of claims tested: standard claims / custom claims
    See https://openid.net/specs/openid-connect-core-1_0.html#ClaimsParameter
    See https://openid.net/specs/openid-connect-core-1_0.html#StandardClaims
    See https://twitter.com/ApiDiary/status/1426218707756457995
    """
    results = {"SUPPORT_FOR_STANDARD_CLAIMS": "Not supported", "SUPPORT_FOR_CUSTOM_CLAIMS": "Not supported"}
    param_values = {}
    # {"userinfo":{"given_name":{"essential":true}},"id_token":{"auth_time":{"essential":true}}}
    param_values["SUPPORT_FOR_STANDARD_CLAIMS"] = "%7B%22userinfo%22:%7B%22given_name%22:%7B%22essential%22:true%7D%7D,%22id_token%22:%7B%22auth_time%22:%7B%22essential%22:true%7D%7D%7D"
    # {"id_token":{"custom_claim":{"essential": true,"value":"fixed"}},"userinfo":{"custom_claim":{"essential": true,"value":"fixed"}}}
    param_values["SUPPORT_FOR_CUSTOM_CLAIMS"] = "%7B%22id_token%22:%7B%22custom_claim%22:%7B%22essential%22:%20true,%22value%22:%22fixed%22%7D%7D,%22userinfo%22:%7B%22custom_claim%22:%7B%22essential%22:%20true,%22value%22:%22fixed%22%7D%7D%7D"
    # Prevent parameter collisions
    work_url = disable_existing_parameters(base_url, ["response_type", "scope", "claims"])
    # Use the Authorization Code flow because it is commonly enabled
    with get_requests_session() as session:
        for param_value in param_values:
            session.cookies.clear()
            resp = session.get(url=f"{work_url}&response_type=code&scope=openid&claims={param_values[param_value]}", allow_redirects=False, timeout=TIMEOUT)
            # If the claims parameter submitted is accepted then the login page is returned to continue the flow
            if get_page_type(resp) == "LOGIN":
                results[param_value] = "Seems supported"
    return results


def test_enabled_audiences(base_url, audiences_list_source):
    """
    Identify all audiences enabled.
    """
    results = []
    # Prevent parameter collisions
    work_url = disable_existing_parameters(base_url, ["response_type", "scope", "audience", "code_challenge", "code_challenge_method"])
    # Add common audiences to the provided list
    audiences = list(set(audiences_list_source + COMMON_AUDIENCES))
    # Use the Authorization Code flow because it is commonly enabled
    with get_requests_session() as session:
        for audience in audiences:
            print(f"\rTesting audience: {audience:<50}", end="", flush=True)
            session.cookies.clear()
            try:
                resp = session.get(url=f"{work_url}&response_type=code&scope=openid&audience={audience}", allow_redirects=False, timeout=TIMEOUT)
            except:
                continue
            # If the audience is accepted then the login page is returned to continue the flow
            if get_page_type(resp) == "LOGIN":
                results.append(audience)
    print(f"\r{' '*50}", end="", flush=True)
    print("\r", end="", flush=True)
    # If the canary is in the list of allowed element so it indicates that the STS has no discrepancy factor
    if "CANNOT_EXISTS_CANARY" in results:
        print("\rSTS has no discrepancy factor! ", end="", flush=True)
        results.clear()
    results.sort()
    return results


if __name__ == "__main__":
    colorama.init()
    title_color = "yellow"
    info_color = "cyan"
    parser = argparse.ArgumentParser()
    required_params = parser.add_argument_group("required arguments")
    required_params.add_argument("--client-id", action="store", dest="client_id", help="Identifier of the target Client.", required=True)
    required_params.add_argument("--valid-redirect-uri", action="store", dest="valid_redirect_uri", help="Valid redirection uri for the target Client.", required=True)
    required_params.add_argument("--sts-endpoint", action="store", dest="sts_base_url", help="OAUTH/OIDC endpoint to use for init flows (ex: http://localhost:8080/auth/realms/demo/protocol/openid-connect/auth).", required=True)
    parser.add_argument("--proxy", action="store", dest="http_proxy", default=None, help="HTTP proxy to use for all HTTP call to differents services (ex: http://88.198.50.103:9080). Default to None", required=False)
    parser.add_argument("--words-file", action="store", dest="words_dictionary", default="/tools/sec-lists/Discovery/Web-Content/raft-small-words.txt", help="Path to a text file containing words used for all enumeration operations. Default to the file 'raft-small-words.txt' from the local SecLists GitHub repo clone.", required=False)
    parser.add_argument("--usernames-file", action="store", dest="usernames_dictionary", default="/tools/sec-lists/Usernames/Names/names.txt", help="Path to a text file containing usernames used for end-users enumeration operations. Default to the file 'names.txt' from the local SecLists GitHub repo clone.", required=False)
    parser.add_argument("--list-local-redirect", action="store_true", dest="list_local_redirect", default=False, help="List allowed localhost redirection on the STS for the specified Client ID. Default to False", required=False)
    parser.add_argument("--list-local-request-uri", action="store_true", dest="list_local_request_uri", default=False, help="List allowed localhost request uri on the STS for the specified Client ID during flow init. Default to False", required=False)
    parser.add_argument("--list-local-request-uri-hit-min-delay", action="store", dest="list_local_request_uri_hit_min_delay", default=10000, help="For the allowed localhost request uri test, minimum delay in microseconds indicating a hit by the STS against a local forged url. Default to 10000", required=False)
    parser.add_argument("--tls-client-cert-pem-file", action="store", dest="tls_client_cert_pem_file", default=None, help="TLS client certificate file to use for all HTTP requests (PEM format). Default to None", required=False)
    parser.add_argument("--tls-client-cert-key-pem-file", action="store", dest="tls_client_cert_key_pem_file", default=None, help="TLS client certificate private key file to use for all HTTP requests (PEM format). Default to None", required=False)
    parser.add_argument("--request-timeout", action="store", dest="request_timeout", type=int, default=5, help="Timeout, in seconds, applied to all HTTP requests. Default to 5 seconds", required=False)
    args = parser.parse_args()
    words_file = args.words_dictionary
    usernames_file = args.usernames_dictionary
    if args.http_proxy is not None:
        PROXIES["http"] = args.http_proxy
        PROXIES["https"] = args.http_proxy
    else:
        PROXIES.clear()
    if args.tls_client_cert_pem_file is not None and args.tls_client_cert_key_pem_file is not None:
        TLS_CLIENT_CERT = (args.tls_client_cert_pem_file, args.tls_client_cert_key_pem_file)
    else:
        TLS_CLIENT_CERT = None
    TIMEOUT = args.request_timeout
    print(colored("[i]", info_color, attrs=["bold"]) + " STS refer to the target 'OAuth Authorization Server' or 'OpenID Provider' instance.")
    print(colored(f"[+] Compose the base url based on parameters provided:", title_color, attrs=["bold"]))
    work_url = args.sts_base_url.split("?")[0]
    client_id = args.client_id
    redirect_uri = args.valid_redirect_uri
    state = str(uuid.uuid4())
    list_local_request_uri_hit_min_delay = int(args.list_local_request_uri_hit_min_delay)
    base_url = f"{work_url}?client_id={client_id}&redirect_uri={redirect_uri}&state={state}"
    print(base_url)
    print(colored(f"[+] Identify the software operating the STS...", title_color, attrs=["bold"]))
    results = get_system_software(base_url)
    print(results)
    print(colored(f"[+] Load the list of words from file '{words_file}'...", title_color, attrs=["bold"]))
    with open(words_file, "r") as f:
        words = f.read().splitlines()
    words.append(client_id)
    words.append(client_id.lower())
    words.append(client_id.upper())
    words.append(client_id.capitalize())
    print(f"{len(words)} words loaded in memory.")
    print(colored(f"[+] Load the list of usernames from file '{usernames_file}'...", title_color, attrs=["bold"]))
    if os.path.isfile(usernames_file):
        with open(usernames_file, "r") as f:
            usernames = f.read().splitlines()
            print(f"{len(usernames)} usernames loaded in memory.")
    else:
        print("File not found so default to the words list loaded.")
        usernames = []
        usernames.extend(words)
    print(colored(f"[+] List enabled flows on the STS for the specified Client ID...", title_color, attrs=["bold"]))
    flows_enabled_status = test_enabled_flows(base_url)
    render_results(flows_enabled_status, column_headers=["Flow", "Enabled"])
    print(colored(f"[+] Enumerate allowed Scopes on the STS for the specified Client ID...", title_color, attrs=["bold"]))
    results = test_allowed_scopes(base_url, words)
    render_results(results)
    print(colored(f"[+] Enumerate defined Clients on the STS...", title_color, attrs=["bold"]))
    client_id_list = test_existing_clients(base_url, words)
    render_results(client_id_list)
    print(colored(f"[+] Enumerate protected apps by the STS...", title_color, attrs=["bold"]))
    results = test_map_client_to_protected_app(base_url, client_id_list)
    render_results(results, column_headers=["Client ID", "App URL"])
    print(colored(f"[+] Enumerate defined Realms on the STS via different methods...", title_color, attrs=["bold"]))
    adapted_words_list = adapt_dictionaries_for_sts(base_url, words, client_id)
    realms = test_existing_realms(base_url, adapted_words_list)
    render_results(realms, column_headers=["Realm ID", "Metadata endpoint URL"])
    if len(realms) > 0:
        print(colored(f"[+] Enumerate client self-registration endpoint for defined Realms on the STS...", title_color, attrs=["bold"]))
        results = test_realms_client_self_registration_endpoint(realms)
        render_results(results, column_headers=["Realm ID", "Client self-registration endpoint URL"])
        print(colored(f"[+] Enumerate defined Realms on the STS for which the NONE signature algorithm is supported...", title_color, attrs=["bold"]))
        results = {}
        for realm_name, realm_metadata_endpoint in realms.items():
            settings = test_none_signing_algorithm_support(realm_metadata_endpoint)
            if len(settings) > 0:
                results[realm_name] = settings
        render_results(results, column_headers=["Realm ID", "Setting keys"])
        print(colored(f"[+] Enumerate defined Realms on the STS for which the HMAC signature algorithm is supported and export public keys to PEM files [KID].pem...", title_color, attrs=["bold"]))
        print(colored("[i]", info_color, attrs=["bold"]) + " Key-Confusion attack via the TICARPI JWT tool: python3 jwt_tool.py 'SOURCE_JWT_TOKEN' -X k -pk [KID].pem")
        print("    Use the key file with the KID (Key ID) with the value of the header 'kid' from 'SOURCE_JWT_TOKEN'.")
        results = {}
        at_least_one_result_found = False
        for realm_name, realm_metadata_endpoint in realms.items():
            settings = test_hmac_signing_algorithm_support_with_public_key_export(realm_metadata_endpoint)
            if len(settings["SettingsKeys"]) > 0:
                at_least_one_result_found = True
                print(colored(f"Realm '{realm_name}'", info_color, attrs=["bold"]))
                print(f"  Setting keys: ", end="", flush=True)
                render_results(settings["SettingsKeys"])
                print(f"  Key files: ", end="", flush=True)
                render_results(settings["PublicKeysExportFiles"])
        if not at_least_one_result_found:
            print("HMAC algorithm not supported.")
    print(colored(f"[+] Enumerate enabled Response Modes on the STS for the specified Client ID...", title_color, attrs=["bold"]))
    results = test_enabled_response_modes(base_url, words)
    render_results(results)
    if flows_enabled_status["AUTHORIZATION_CODE_WITH_PKCE_WITH_PLAIN_CHALLENGE"]:
        print(colored(f"[+] Identify the weakest Code Challenge value accepted by the STS for the specified Client ID in a context of Authorization Code with PKCE using PLAIN code challenge method...", title_color, attrs=["bold"]))
        results = test_weak_pkce_code_verifier_rejection_for_plain_challenge(base_url)
        render_results(results)
    if flows_enabled_status["HYBRID_WITH_NONCE"]:
        print(colored(f"[+] Identify the weakest Nonce value accepted by the STS for the specified Client ID in a context of Hybrid flow...", title_color, attrs=["bold"]))
        results = test_weak_nonce_rejection_for_hybrid_flow(base_url)
        render_results(results)
    if args.list_local_redirect:
        print(colored(f"[+] List allowed localhost redirection on the STS for the specified Client ID...", title_color, attrs=["bold"]))
        results = test_allowed_localhost_redirection(base_url)
        render_results(results)
    if args.list_local_request_uri:
        print(colored(f"[+] List allowed localhost request uri on the STS for the specified Client ID during a flow init, min delay set to {list_local_request_uri_hit_min_delay} microseconds...", title_color, attrs=["bold"]))
        results = test_allowed_localhost_request_uri(base_url, list_local_request_uri_hit_min_delay)
        render_results(results)
    print(colored(f"[+] Check if the STS support the 'OpenID Provider Issuer Discovery' feature and, if yes, try to abuse it for a end-users enumeration operations...", title_color, attrs=["bold"]))
    for realm in realms:
        print(colored(f"Realm '{realm}'", info_color, attrs=["bold"]))
        results = test_oidc_provider_issuer_discovery_support(base_url, realm, usernames)
        render_results(results)
    print(colored(f"[+] Identify if the STS for the specified Client ID supports the parameter 'claims'...", title_color, attrs=["bold"]))
    print(colored("[i]", info_color, attrs=["bold"]) + " Manual validation to confirm the support is needed because STS can simply skip the claims specified in the ID Token delivered or UserInfo response.")
    results = test_claims_parameter_support(base_url)
    render_results(results, column_headers=["Type of claims", "Support status"])
    print(colored(f"[+] Enumerate enabled Audience on the STS for the specified Client ID...", title_color, attrs=["bold"]))
    results = test_enabled_audiences(base_url, words)
    render_results(results)
