chatnest.ir/SNISPF
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

first init

Browse Source
This commit is contained in:
nouri_a
2026-04-15 11:11:44 +03:30
parent f41bc57b71
commit 1bad6394d1
20 changed files with 2844 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sni_spoofing/__init__.py
+6
View File
@@ -0,0 +1,6 @@
"""
SNISPF - Cross-platform SNI spoofing and DPI bypass tool.
"""
__version__ = "1.1.0"
__author__ = "Rainman69"
sni_spoofing/bypass/__init__.py
+16
View File
@@ -0,0 +1,16 @@
"""Bypass strategy implementations."""
from .base import BypassStrategy
from .fragment import FragmentBypass
from .fake_sni import FakeSNIBypass
from .combined import CombinedBypass
from .raw_injector import RawInjector, is_raw_available
__all__ = [
"BypassStrategy",
"FragmentBypass",
"FakeSNIBypass",
"CombinedBypass",
"RawInjector",
"is_raw_available",
]
sni_spoofing/bypass/base.py
+44
View File
@@ -0,0 +1,44 @@
"""Base class for bypass strategies."""
import abc
import socket
from typing import Optional
class BypassStrategy(abc.ABC):
"""Abstract base for DPI bypass strategies.
Each strategy implements a different technique for evading
Deep Packet Inspection when forwarding TCP connections.
"""
name: str = "base"
@abc.abstractmethod
async def apply(
self,
client_sock: socket.socket,
server_sock: socket.socket,
fake_sni: str,
first_data: bytes,
loop=None,
) -> bool:
"""Apply the bypass strategy to an outgoing connection.
This method is called after the TCP connection to the server
is established but before any real data is forwarded.
Args:
client_sock: The incoming client socket
server_sock: The outgoing socket to the real server
fake_sni: The fake SNI hostname to use
first_data: First data received from the client
loop: asyncio event loop
Returns:
True if strategy was applied successfully, False otherwise
"""
pass
def __repr__(self):
return f"<{self.__class__.__name__} strategy='{self.name}'>"
sni_spoofing/bypass/combined.py
+130
View File
@@ -0,0 +1,130 @@
"""Combined bypass strategy.
Combines multiple bypass techniques for maximum effectiveness.
With raw sockets (Linux + root):
1. The raw injector sends a fake ClientHello with an out-of-window
seq number during the TCP handshake (DPI parses it, server drops it)
2. Then the real ClientHello is fragmented at the SNI boundary
Both techniques hit DPI at once.
Without raw sockets (fallback):
Uses fragmentation only (with optional TTL trick for the fake).
The fake_sni prefix method is NOT used on the real TCP stream
because it corrupts the TLS handshake.
"""
import asyncio
import logging
import socket
import time
from typing import Optional
from .base import BypassStrategy
from ..tls import ClientHelloBuilder
from ..tls.fragment import fragment_client_hello, fragment_data
logger = logging.getLogger("snispf")
class CombinedBypass(BypassStrategy):
"""Combined DPI bypass using multiple techniques simultaneously.
With raw injector available:
1. Fake ClientHello injected out-of-window (by the sniffer/injector)
2. Real ClientHello fragmented at SNI boundary
3. Small inter-fragment delays
Without raw injector:
1. (Optional) TTL trick to send fake ClientHello that expires
before reaching the server
2. Real ClientHello fragmented at SNI boundary
3. Small inter-fragment delays
"""
name = "combined"
def __init__(
self,
fragment_strategy: str = "sni_split",
use_ttl_trick: bool = False,
fragment_delay: float = 0.1,
fake_first: bool = True,
raw_injector=None,
):
self.fragment_strategy = fragment_strategy
self.use_ttl_trick = use_ttl_trick
self.fragment_delay = fragment_delay
self.fake_first = fake_first
self.raw_injector = raw_injector
async def apply(
self,
client_sock: socket.socket,
server_sock: socket.socket,
fake_sni: str,
first_data: bytes,
loop=None,
) -> bool:
if loop is None:
loop = asyncio.get_running_loop()
try:
server_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
# Step 1: Handle fake ClientHello
if self.raw_injector is not None:
# Raw injector already sent the fake out-of-window during
# the TCP handshake. Wait for server confirmation.
local_port = server_sock.getsockname()[1]
confirmed = await loop.run_in_executor(
None,
self.raw_injector.wait_for_confirmation,
local_port,
2.0,
)
if not confirmed:
logger.warning(
f"port={local_port}: no confirmation that server "
f"ignored the fake packet (timeout)"
)
elif self.fake_first and self.use_ttl_trick:
# TTL trick: send fake with low TTL so it reaches DPI
# but expires before the server
fake_hello = ClientHelloBuilder.build_client_hello(sni=fake_sni)
try:
original_ttl = server_sock.getsockopt(
socket.IPPROTO_IP, socket.IP_TTL
)
server_sock.setsockopt(
socket.IPPROTO_IP, socket.IP_TTL, 3
)
await loop.sock_sendall(server_sock, fake_hello)
await asyncio.sleep(0.05)
server_sock.setsockopt(
socket.IPPROTO_IP, socket.IP_TTL, original_ttl
)
except OSError:
# TTL trick not available, skip the fake entirely
pass
await asyncio.sleep(0.001)
# NOTE: Without raw sockets or TTL trick, we do NOT send a fake
# ClientHello on the real TCP stream. It would corrupt the
# handshake because the server receives it as real data.
# Step 2: Fragment and send the real ClientHello
fragments = fragment_client_hello(first_data, self.fragment_strategy)
for i, fragment in enumerate(fragments):
await loop.sock_sendall(server_sock, fragment)
if i < len(fragments) - 1 and self.fragment_delay > 0:
await asyncio.sleep(self.fragment_delay)
server_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 0)
return True
except Exception:
return False
sni_spoofing/bypass/fake_sni.py
+190
View File
@@ -0,0 +1,190 @@
"""Fake SNI bypass strategy.
Sends a fake TLS ClientHello with an allowed SNI that DPI will parse
and whitelist, but the real server will ignore.
Two operating modes:
- With raw sockets (Linux + root): Uses the seq_id trick from the
original tool. Injects a fake ClientHello with an out-of-window
TCP sequence number. DPI parses it, server drops it.
- Without raw sockets (fallback): Sends the real ClientHello in
fragments so DPI cannot read the SNI from any single packet.
The fake_sni prefix method does NOT work without raw sockets
because sending the fake on the same TCP stream corrupts the
TLS handshake.
"""
import asyncio
import logging
import socket
from typing import Optional
from .base import BypassStrategy
from ..tls import ClientHelloBuilder
from ..tls.fragment import fragment_client_hello
logger = logging.getLogger("snispf")
class FakeSNIBypass(BypassStrategy):
"""Bypass DPI by injecting a fake TLS ClientHello with spoofed SNI.
The only reliable way to do this is with raw socket injection
(out-of-window seq trick). When raw sockets are not available,
this falls back to TLS fragmentation which hides the SNI by
splitting it across TCP segments.
Methods:
- "raw_inject" - Inject fake ClientHello with wrong seq number
via AF_PACKET. DPI sees it, server drops it. (Linux + root)
- "ttl_trick" - Send fake with low IP TTL. May reach DPI but
expire before the server. Unreliable, platform-dependent.
- "fragment_fallback" - Falls back to fragmenting the real
ClientHello. No fake is sent on the real stream.
"""
name = "fake_sni"
def __init__(self, method: str = "prefix_fake", raw_injector=None):
self.method = method
self.raw_injector = raw_injector
async def apply(
self,
client_sock: socket.socket,
server_sock: socket.socket,
fake_sni: str,
first_data: bytes,
loop=None,
) -> bool:
if loop is None:
loop = asyncio.get_running_loop()
# If we have a raw injector running, the fake was already injected
# during the TCP handshake. Just send the real data and go.
if self.raw_injector is not None:
return await self._raw_inject_send(
server_sock, first_data, loop
)
# Without raw sockets, the old "prefix_fake" method of sending
# a fake ClientHello on the same TCP stream is broken - the server
# receives both and the TLS handshake fails. Fall back to
# TTL trick if requested, otherwise just fragment the real hello.
if self.method == "ttl_trick":
return await self._ttl_trick(
server_sock, fake_sni, first_data, loop
)
else:
# Fragment fallback: split the real ClientHello so DPI can't
# read the SNI from any single packet.
return await self._fragment_fallback(
server_sock, first_data, loop
)
async def _raw_inject_send(
self,
server_sock: socket.socket,
first_data: bytes,
loop,
) -> bool:
"""With raw injection, the fake was already sent out-of-window.
Just send the real ClientHello normally."""
try:
local_port = server_sock.getsockname()[1]
# Wait for the sniffer to confirm the server ignored the fake.
confirmed = await loop.run_in_executor(
None,
self.raw_injector.wait_for_confirmation,
local_port,
2.0,
)
if not confirmed:
logger.warning(
f"port={local_port}: server did not confirm fake was "
f"ignored (timeout). Sending real data anyway."
)
# Send the real ClientHello (untouched)
await loop.sock_sendall(server_sock, first_data)
return True
except Exception:
return False
async def _ttl_trick(
self,
server_sock: socket.socket,
fake_sni: str,
first_data: bytes,
loop,
) -> bool:
"""Send fake ClientHello with low TTL, then real data normally.
The fake packet has a TTL low enough to expire before reaching
the server, but the DPI middlebox (typically 1-3 hops away)
will see it. This is unreliable depending on network topology.
"""
try:
server_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
fake_hello = ClientHelloBuilder.build_client_hello(sni=fake_sni)
# Save original TTL
original_ttl = server_sock.getsockopt(
socket.IPPROTO_IP, socket.IP_TTL
)
# Low TTL: reaches DPI (1-5 hops) but expires before server
server_sock.setsockopt(socket.IPPROTO_IP, socket.IP_TTL, 3)
try:
await loop.sock_sendall(server_sock, fake_hello)
except OSError:
pass # May get ICMP TTL exceeded
await asyncio.sleep(0.05)
# Restore normal TTL
server_sock.setsockopt(
socket.IPPROTO_IP, socket.IP_TTL, original_ttl
)
# Send real ClientHello normally
await loop.sock_sendall(server_sock, first_data)
server_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 0)
return True
except Exception:
return False
async def _fragment_fallback(
self,
server_sock: socket.socket,
first_data: bytes,
loop,
) -> bool:
"""Fallback: fragment the real ClientHello at the SNI boundary.
Without raw sockets we cannot safely send a fake ClientHello
(it would corrupt the TLS stream). Instead, fragment the real
ClientHello so DPI cannot read the full SNI from a single packet.
"""
try:
server_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
fragments = fragment_client_hello(first_data, "sni_split")
for i, fragment in enumerate(fragments):
await loop.sock_sendall(server_sock, fragment)
if i < len(fragments) - 1:
await asyncio.sleep(0.1)
server_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 0)
return True
except Exception:
return False
sni_spoofing/bypass/fragment.py
+89
View File
@@ -0,0 +1,89 @@
"""Fragment-based DPI bypass strategy.
Splits the real TLS ClientHello into fragments so that DPI systems
that only inspect the first packet or don't reassemble TCP streams
cannot read the SNI.
"""
import asyncio
import socket
import time
from typing import Optional
from .base import BypassStrategy
from ..tls.fragment import fragment_client_hello
class FragmentBypass(BypassStrategy):
"""Bypass DPI by fragmenting the TLS ClientHello.
This is the most compatible cross-platform bypass method.
It works by splitting the ClientHello into multiple TCP segments,
with the split point strategically placed in the middle of the
SNI extension value.
DPI systems that don't reassemble TCP streams will see an incomplete
SNI in the first packet and won't be able to filter it.
"""
name = "fragment"
def __init__(
self,
strategy: str = "sni_split",
fragment_delay: float = 0.1,
tcp_nodelay: bool = True,
):
"""Initialize fragment bypass.
Args:
strategy: Fragmentation strategy (sni_split, half, multi, tls_record_frag)
fragment_delay: Delay between fragments in seconds
tcp_nodelay: Enable TCP_NODELAY to send fragments immediately
"""
self.strategy = strategy
self.fragment_delay = fragment_delay
self.tcp_nodelay = tcp_nodelay
async def apply(
self,
client_sock: socket.socket,
server_sock: socket.socket,
fake_sni: str,
first_data: bytes,
loop=None,
) -> bool:
"""Apply fragmentation to the first TLS record.
The first_data from the client (usually a TLS ClientHello) is
fragmented and sent to the server in multiple TCP segments.
"""
if loop is None:
loop = asyncio.get_running_loop()
try:
# Enable TCP_NODELAY so each send() becomes its own segment
if self.tcp_nodelay:
server_sock.setsockopt(
socket.IPPROTO_TCP, socket.TCP_NODELAY, 1
)
# Fragment the ClientHello
fragments = fragment_client_hello(first_data, self.strategy)
# Send each fragment as a separate TCP segment
for i, fragment in enumerate(fragments):
await loop.sock_sendall(server_sock, fragment)
if i < len(fragments) - 1 and self.fragment_delay > 0:
await asyncio.sleep(self.fragment_delay)
# Disable TCP_NODELAY after fragments are sent (optional)
if self.tcp_nodelay:
server_sock.setsockopt(
socket.IPPROTO_TCP, socket.TCP_NODELAY, 0
)
return True
except Exception:
return False
sni_spoofing/bypass/raw_injector.py
+418
View File
@@ -0,0 +1,418 @@
"""Raw socket packet injection for out-of-window fake SNI.
Implements the seq_id trick from the Go reference:
1. Sniff the outbound SYN to record the ISN (Initial Sequence Number)
2. Sniff the outbound 3rd ACK (handshake complete)
3. Inject a fake TLS ClientHello with seq = ISN+1 - len(fake)
This puts it BEFORE the server's receive window, so the server drops it,
but DPI sees and parses the fake SNI.
4. Wait for the server to ACK with ack == ISN+1, confirming the fake was
ignored and the server still expects the real data.
Linux only. Requires CAP_NET_RAW (run as root).
"""
import logging
import os
import socket
import struct
import threading
import time
from typing import Optional, Dict
logger = logging.getLogger("snispf")
ETH_P_IP = 0x0800
ETH_P_ALL = 0x0003
IPPROTO_TCP = 6
# TCP flags
FIN = 0x01
SYN = 0x02
RST = 0x04
PSH = 0x08
ACK = 0x10
def _htons(v):
return socket.htons(v)
def _ip_hdr_len(ip_bytes):
return (ip_bytes[0] & 0x0F) * 4
def _checksum_fold(s):
while s >> 16:
s = (s & 0xFFFF) + (s >> 16)
return (~s) & 0xFFFF
def _sum16(data):
s = 0
for i in range(0, len(data) - 1, 2):
s += (data[i] << 8) | data[i + 1]
if len(data) % 2 == 1:
s += data[-1] << 8
while s >> 16:
s = (s & 0xFFFF) + (s >> 16)
return s
def _ip_checksum(iph):
return _checksum_fold(_sum16(iph))
def _tcp_checksum(iph, tcp_with_payload):
ihl = _ip_hdr_len(iph)
pseudo = bytearray(12)
pseudo[0:4] = iph[12:16] # src IP
pseudo[4:8] = iph[16:20] # dst IP
pseudo[9] = 6 # TCP protocol
struct.pack_into("!H", pseudo, 10, len(tcp_with_payload))
return _checksum_fold(_sum16(pseudo) + _sum16(tcp_with_payload))
def _build_fake_frame(template_pkt, isn, fake_payload):
"""Build the injection frame from a captured 3rd-ACK packet template.
Takes the captured Ethernet+IP+TCP headers from the 3rd handshake ACK,
appends the fake TLS ClientHello as payload, and sets:
- seq = ISN + 1 - len(fake_payload) (out of window for the server)
- PSH flag added
- Proper IP and TCP checksums recalculated
"""
ip_off = 14 # Ethernet header is 14 bytes
ihl = _ip_hdr_len(template_pkt[ip_off:])
tcp_off = ip_off + ihl
tcp_hdr_len = (template_pkt[tcp_off + 12] >> 4) * 4
# Copy headers (Ethernet + IP + TCP) and append fake payload
headers = bytearray(template_pkt[:tcp_off + tcp_hdr_len])
out = headers + fake_payload
# Update IP total length
struct.pack_into("!H", out, ip_off + 2, len(out) - ip_off)
# Increment IP ID
old_id = struct.unpack("!H", out[ip_off + 4:ip_off + 6])[0]
struct.pack_into("!H", out, ip_off + 4, (old_id + 1) & 0xFFFF)
# Recalculate IP checksum
out[ip_off + 10] = 0
out[ip_off + 11] = 0
ip_cksum = _ip_checksum(out[ip_off:ip_off + ihl])
struct.pack_into("!H", out, ip_off + 10, ip_cksum)
# Set PSH flag
out[tcp_off + 13] |= PSH
# Set out-of-window sequence number: ISN + 1 - len(fake)
seq = (isn + 1 - len(fake_payload)) & 0xFFFFFFFF
struct.pack_into("!I", out, tcp_off + 4, seq)
# Recalculate TCP checksum
out[tcp_off + 16] = 0
out[tcp_off + 17] = 0
tcp_cksum = _tcp_checksum(
out[ip_off:ip_off + ihl],
bytes(out[tcp_off:]),
)
struct.pack_into("!H", out, tcp_off + 16, tcp_cksum)
return bytes(out)
class PortState:
"""Per-connection state tracked by the sniffer."""
def __init__(self, syn_seq, fake_hello):
self.syn_seq = syn_seq
self.fake_hello = fake_hello
self.fake_sent = False
self.confirmed = threading.Event()
self.lock = threading.Lock()
class RawInjector:
"""Raw socket sniffer and injector for out-of-window fake SNI.
This is the core mechanism that makes the seq_id trick work:
- Monitors all TCP traffic between local and target IPs
- When a new outbound SYN is detected, records the ISN
- When the 3rd handshake ACK is seen, injects the fake ClientHello
- Waits for server confirmation (ACK with ack == ISN+1)
"""
def __init__(self, local_ip, remote_ip, remote_port, fake_sni_builder):
self.local_ip = socket.inet_aton(local_ip)
self.remote_ip = socket.inet_aton(remote_ip)
self.remote_port = remote_port
self.fake_sni_builder = fake_sni_builder
self.ports: Dict[int, PortState] = {}
self.ports_lock = threading.Lock()
self.raw_fd = None
self.iface_idx = None
self.iface_name = None
self.running = False
self._sniffer_thread = None
def start(self):
"""Open the raw socket and start the sniffer loop."""
try:
self.raw_fd = socket.socket(
socket.AF_PACKET,
socket.SOCK_RAW,
socket.htons(ETH_P_ALL),
)
except (PermissionError, OSError) as e:
logger.warning(f"Cannot open AF_PACKET socket: {e}")
logger.warning("Raw injection unavailable - need root/CAP_NET_RAW")
return False
# Find the interface
iface_info = self._find_interface()
if iface_info is None:
logger.warning("Cannot determine outgoing interface for raw injection")
self.raw_fd.close()
self.raw_fd = None
return False
self.iface_name, self.iface_idx = iface_info
self.raw_fd.bind((self.iface_name, ETH_P_ALL))
self.running = True
self._sniffer_thread = threading.Thread(
target=self._sniff_loop, daemon=True
)
self._sniffer_thread.start()
logger.info("Raw packet injector started")
return True
def stop(self):
"""Stop the sniffer."""
self.running = False
if self.raw_fd:
try:
self.raw_fd.close()
except Exception:
pass
def _find_interface(self):
"""Find the network interface name and index for the target IP.
Returns:
Tuple of (interface_name, interface_index) or None if not found.
"""
import fcntl
import array
try:
# Use a UDP connect to find which interface is used
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect((socket.inet_ntoa(self.remote_ip), 53))
local_addr = s.getsockname()[0]
s.close()
# Get all interfaces and find the matching one
# Using SIOCGIFCONF
max_bytes = 8096
buf = array.array("B", b"\0" * max_bytes)
ifconf = struct.pack("iL", max_bytes, buf.buffer_info()[0])
result = fcntl.ioctl(
self.raw_fd.fileno(), 0x8912, ifconf # SIOCGIFCONF
)
out_bytes = struct.unpack("iL", result)[0]
offset = 0
while offset < out_bytes:
name = buf[offset:offset + 16].tobytes().split(b"\0", 1)[0]
ip_bytes = buf[offset + 20:offset + 24].tobytes()
ip_str = socket.inet_ntoa(ip_bytes)
if ip_str == local_addr:
iface_name = name.decode("ascii", errors="replace")
# Get interface index
ifreq = struct.pack("16sI", name, 0)
result = fcntl.ioctl(
self.raw_fd.fileno(), 0x8933, ifreq # SIOCGIFINDEX
)
idx = struct.unpack("16sI", result)[1]
logger.debug(f"Using interface {iface_name} (index {idx})")
return (iface_name, idx)
offset += 40 # struct ifreq size
except Exception as e:
logger.debug(f"Interface detection error: {e}")
return None
def register_port(self, local_port, fake_hello):
"""Register a port for monitoring (called before connect)."""
with self.ports_lock:
self.ports[local_port] = PortState(0, fake_hello)
def wait_for_confirmation(self, local_port, timeout=2.0):
"""Wait for the server to confirm it ignored the fake packet.
Returns True if confirmed, False on timeout.
"""
with self.ports_lock:
ps = self.ports.get(local_port)
if ps is None:
return False
return ps.confirmed.wait(timeout=timeout)
def cleanup_port(self, local_port):
"""Clean up state for a port."""
with self.ports_lock:
self.ports.pop(local_port, None)
def _inject_frame(self, frame):
"""Inject a raw Ethernet frame."""
try:
addr = (
self.iface_name or "", # interface name
ETH_P_IP,
0, # packet type
0, # arp hardware type
frame[0:6], # destination MAC
)
self.raw_fd.sendto(frame, addr)
return True
except Exception as e:
logger.debug(f"Inject error: {e}")
# Fallback: try sendto with sockaddr_ll style
try:
sll = struct.pack(
"HH I BB 8s",
socket.htons(ETH_P_IP), # protocol
self.iface_idx, # ifindex
0, # pkttype
6, # halen
0,
frame[0:8], # addr
)
os.write(self.raw_fd.fileno(), frame)
return True
except Exception as e2:
logger.debug(f"Inject fallback error: {e2}")
return False
def _sniff_loop(self):
"""Main sniffer loop - watches TCP handshakes and injects fake packets."""
while self.running:
try:
pkt, _ = self.raw_fd.recvfrom(65536)
except (OSError, socket.error):
if not self.running:
break
continue
if len(pkt) < 14 + 20 + 20:
continue
# Check Ethernet type is IPv4
eth_type = struct.unpack("!H", pkt[12:14])[0]
if eth_type != ETH_P_IP:
continue
ip = pkt[14:]
if (ip[0] >> 4) != 4 or ip[9] != IPPROTO_TCP:
continue
ihl = _ip_hdr_len(ip)
src_ip = ip[12:16]
dst_ip = ip[16:20]
tcp = ip[ihl:]
if len(tcp) < 20:
continue
flags = tcp[13]
tcp_hdr_len = (tcp[12] >> 4) * 4
payload_len = len(tcp) - tcp_hdr_len
outbound = (src_ip == self.local_ip and dst_ip == self.remote_ip)
inbound = (src_ip == self.remote_ip and dst_ip == self.local_ip)
if outbound:
src_port = struct.unpack("!H", tcp[0:2])[0]
seq = struct.unpack("!I", tcp[4:8])[0]
# SYN (no ACK): new outbound connection
if (flags & SYN) and not (flags & ACK):
with self.ports_lock:
ps = self.ports.get(src_port)
if ps is not None:
with ps.lock:
ps.syn_seq = seq
logger.debug(
f"[sniff] SYN port={src_port} isn={seq}"
)
continue
# 3rd-handshake ACK: ACK only, no payload
if (flags & ACK) and not (flags & (SYN | FIN | RST)) and payload_len == 0:
with self.ports_lock:
ps = self.ports.get(src_port)
if ps is None:
continue
with ps.lock:
if ps.fake_sent:
continue
ps.fake_sent = True
syn_seq = ps.syn_seq
fake = ps.fake_hello
# Inject after a tiny delay (like the Go version's 1ms)
tpl_copy = bytearray(pkt)
def _do_inject(tpl=tpl_copy, isn=syn_seq, payload=fake, port=src_port):
time.sleep(0.001)
frame = _build_fake_frame(bytes(tpl), isn, payload)
if self._inject_frame(frame):
out_seq = (isn + 1 - len(payload)) & 0xFFFFFFFF
logger.debug(
f"[inject] port={port} fake seq={out_seq} "
f"(ISN={isn}, fake_len={len(payload)})"
)
else:
logger.debug(f"[inject] port={port} injection failed")
threading.Thread(target=_do_inject, daemon=True).start()
if inbound:
dst_port = struct.unpack("!H", tcp[2:4])[0]
ack_num = struct.unpack("!I", tcp[8:12])[0]
# Server's ACK confirming fake was ignored
if (flags & ACK) and not (flags & (SYN | FIN | RST)) and payload_len == 0:
with self.ports_lock:
ps = self.ports.get(dst_port)
if ps is None:
continue
with ps.lock:
if ps.fake_sent and ack_num == (ps.syn_seq + 1) & 0xFFFFFFFF:
if not ps.confirmed.is_set():
ps.confirmed.set()
logger.debug(
f"[sniff] port={dst_port} CONFIRMED "
f"server acked ISN+1={ack_num}"
)
def is_raw_available():
"""Check if raw socket injection is available on this system."""
try:
s = socket.socket(
socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)
)
s.close()
return True
except (PermissionError, OSError, AttributeError):
return False
sni_spoofing/cli.py
+495
View File
@@ -0,0 +1,495 @@
"""
SNISPF - Cross-platform SNI spoofing and DPI bypass tool.
Works on Windows, macOS, and Linux without requiring kernel drivers.
On Linux with root, enables raw packet injection for the seq_id trick.
Usage:
snispf --config config.json
snispf --listen 0.0.0.0:40443 --connect 188.114.98.0:443 --sni auth.vercel.com
"""
import argparse
import asyncio
import json
import logging
import os
import platform
import signal
import sys
from pathlib import Path
# Add parent to path for direct script execution
if __name__ == "__main__":
sys.path.insert(0, str(Path(__file__).parent.parent))
from sni_spoofing import __version__
from sni_spoofing.bypass import (
BypassStrategy,
CombinedBypass,
FakeSNIBypass,
FragmentBypass,
RawInjector,
is_raw_available,
)
from sni_spoofing.forwarder import start_server
from sni_spoofing.utils import (
check_platform_capabilities,
get_default_interface_ipv4,
is_valid_ip,
is_valid_port,
resolve_host,
)
# ─── Banner ──────────────────────────────────────────────────────────────────
BANNER = r"""
███████╗███╗ ██╗██╗███████╗██████╗ ███████╗
██╔════╝████╗ ██║██║██╔════╝██╔══██╗██╔════╝
███████╗██╔██╗ ██║██║███████╗██████╔╝█████╗
╚════██║██║╚██╗██║██║╚════██║██╔═══╝ ██╔══╝
███████║██║ ╚████║██║███████║██║ ██║
╚══════╝╚═╝ ╚═══╝╚═╝╚══════╝╚═╝ ╚═╝
┌──────────────────────────────────────────────────────────────────┐
│ SNISPF - Cross-Platform DPI Bypass Tool │
│ SNI Spoofing + TLS Fragmentation │
│ Works on Windows / macOS / Linux │
│ https://github.com/Rainman69/SNISPF │
└──────────────────────────────────────────────────────────────────┘
"""
# ─── Logging ─────────────────────────────────────────────────────────────────
def setup_logging(verbose: bool = False, quiet: bool = False):
"""Configure logging."""
if quiet:
level = logging.WARNING
elif verbose:
level = logging.DEBUG
else:
level = logging.INFO
formatter = logging.Formatter(
"%(asctime)s │ %(levelname)-7s │ %(message)s",
datefmt="%H:%M:%S",
)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(formatter)
logger = logging.getLogger("snispf")
logger.setLevel(level)
logger.addHandler(handler)
return logger
# ─── Config ──────────────────────────────────────────────────────────────────
DEFAULT_CONFIG = {
"LISTEN_HOST": "0.0.0.0",
"LISTEN_PORT": 40443,
"CONNECT_IP": "188.114.98.0",
"CONNECT_PORT": 443,
"FAKE_SNI": "auth.vercel.com",
"BYPASS_METHOD": "fragment",
"FRAGMENT_STRATEGY": "sni_split",
"FRAGMENT_DELAY": 0.1,
"USE_TTL_TRICK": False,
"FAKE_SNI_METHOD": "prefix_fake",
}
def load_config(config_path: str) -> dict:
"""Load configuration from JSON file."""
try:
with open(config_path, "r") as f:
user_config = json.load(f)
# Merge with defaults
config = DEFAULT_CONFIG.copy()
config.update(user_config)
return config
except FileNotFoundError:
print(f"Error: Config file not found: {config_path}")
sys.exit(1)
except json.JSONDecodeError as e:
print(f"Error: Invalid JSON in config file: {e}")
sys.exit(1)
def generate_config(output_path: str):
"""Generate a default configuration file."""
config = {
"LISTEN_HOST": "0.0.0.0",
"LISTEN_PORT": 40443,
"CONNECT_IP": "188.114.98.0",
"CONNECT_PORT": 443,
"FAKE_SNI": "auth.vercel.com",
"BYPASS_METHOD": "fragment",
"FRAGMENT_STRATEGY": "sni_split",
"FRAGMENT_DELAY": 0.1,
"USE_TTL_TRICK": False,
"FAKE_SNI_METHOD": "prefix_fake",
}
with open(output_path, "w") as f:
json.dump(config, f, indent=2)
print(f"Generated default config: {output_path}")
print(json.dumps(config, indent=2))
# ─── Strategy Builder ────────────────────────────────────────────────────────
def build_strategy(config: dict, raw_injector=None) -> BypassStrategy:
"""Build the appropriate bypass strategy from config.
Available methods:
- "fragment": Fragment TLS ClientHello at SNI boundary
- "fake_sni": Send fake ClientHello with spoofed SNI (needs raw sockets
for the seq_id trick; falls back to fragmentation without them)
- "combined": Both fragmentation and fake SNI (recommended)
"""
method = config.get("BYPASS_METHOD", "fragment").lower()
if method == "fragment":
return FragmentBypass(
strategy=config.get("FRAGMENT_STRATEGY", "sni_split"),
fragment_delay=config.get("FRAGMENT_DELAY", 0.1),
)
elif method == "fake_sni":
return FakeSNIBypass(
method=config.get("FAKE_SNI_METHOD", "prefix_fake"),
raw_injector=raw_injector,
)
elif method == "combined":
return CombinedBypass(
fragment_strategy=config.get("FRAGMENT_STRATEGY", "sni_split"),
use_ttl_trick=config.get("USE_TTL_TRICK", False),
fragment_delay=config.get("FRAGMENT_DELAY", 0.1),
raw_injector=raw_injector,
)
else:
print(f"Warning: Unknown bypass method '{method}', using 'fragment'")
return FragmentBypass()
# ─── CLI ─────────────────────────────────────────────────────────────────────
def parse_args():
"""Parse command line arguments."""
parser = argparse.ArgumentParser(
prog="snispf",
description=(
"SNISPF - Cross-platform DPI bypass tool.\n\n"
"This tool forwards TCP connections while applying DPI bypass\n"
"techniques (SNI spoofing, TLS fragmentation) to circumvent\n"
"internet censorship."
),
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"Examples:\n"
" %(prog)s --config config.json\n"
" %(prog)s -l 0.0.0.0:40443 -c 188.114.98.0:443 -s auth.vercel.com\n"
" %(prog)s -l :40443 -c 188.114.98.0:443 -s dl.google.com -m combined\n"
" %(prog)s --generate-config my_config.json\n"
"\nBypass Methods:\n"
" fragment - Fragment TLS ClientHello at SNI boundary (default)\n"
" fake_sni - Inject fake ClientHello (needs root for seq_id trick)\n"
" combined - Both fragmentation and fake SNI (most effective)\n"
"\nFragment Strategies (for fragment/combined methods):\n"
" sni_split - Split in middle of SNI value (default)\n"
" half - Split record in half\n"
" multi - Split into many small fragments\n"
" tls_record_frag - Use TLS-level record fragmentation\n"
"\nhttps://github.com/Rainman69/SNISPF"
),
)
# Config file
parser.add_argument(
"--config", "-C",
help="Path to JSON config file",
)
parser.add_argument(
"--generate-config",
metavar="PATH",
help="Generate a default config file and exit",
)
# Connection settings
parser.add_argument(
"--listen", "-l",
metavar="HOST:PORT",
help="Listen address (default: 0.0.0.0:40443)",
)
parser.add_argument(
"--connect", "-c",
metavar="IP:PORT",
help="Target server address (default: 188.114.98.0:443)",
)
parser.add_argument(
"--sni", "-s",
metavar="HOSTNAME",
help="Fake SNI hostname (default: auth.vercel.com)",
)
# Bypass settings
parser.add_argument(
"--method", "-m",
choices=["fragment", "fake_sni", "combined"],
help="Bypass method (default: fragment)",
)
parser.add_argument(
"--fragment-strategy",
choices=["sni_split", "half", "multi", "tls_record_frag"],
help="Fragment strategy (default: sni_split)",
)
parser.add_argument(
"--fragment-delay",
type=float,
metavar="SECONDS",
help="Delay between fragments in seconds (default: 0.1)",
)
parser.add_argument(
"--ttl-trick",
action="store_true",
help="Use IP TTL trick for fake packets (may need privileges)",
)
parser.add_argument(
"--no-raw",
action="store_true",
help="Disable raw socket injection even if available",
)
# Output settings
parser.add_argument(
"--verbose", "-v",
action="store_true",
help="Verbose output (debug logging)",
)
parser.add_argument(
"--quiet", "-q",
action="store_true",
help="Quiet output (warnings only)",
)
parser.add_argument(
"--version", "-V",
action="version",
version=f"SNISPF {__version__}",
)
parser.add_argument(
"--info",
action="store_true",
help="Show platform capabilities and exit",
)
return parser.parse_args()
def parse_host_port(addr: str, default_host: str = "0.0.0.0", default_port: int = 443) -> tuple:
"""Parse HOST:PORT string."""
if not addr:
return default_host, default_port
if addr.startswith(":"):
return default_host, int(addr[1:])
parts = addr.rsplit(":", 1)
if len(parts) == 2:
host = parts[0] or default_host
port = int(parts[1])
return host, port
else:
return parts[0], default_port
def show_platform_info():
"""Display platform capability information."""
caps = check_platform_capabilities()
# Also check raw injection availability
caps["raw_injection"] = is_raw_available()
print("\n╔══════════════════════════════════════════╗")
print("║ Platform Capabilities ║")
print("╠══════════════════════════════════════════╣")
for key, value in caps.items():
status = "✓" if value is True else ("✗" if value is False else str(value))
print(f"║ {key:<28} {status:>8} ║")
print("╚══════════════════════════════════════════╝")
print("\nRecommended bypass methods for your platform:")
if caps["raw_injection"]:
print(" ✓ Raw packet injection available (running as root)")
print(" ★ Recommended: combined (uses seq_id trick + fragmentation)")
print(" ★ Also good: fake_sni (uses seq_id trick)")
elif caps["raw_socket"]:
print(" ✓ All methods available (running with sufficient privileges)")
print(" ★ Recommended: combined --ttl-trick")
else:
print(" ✓ fragment - TLS ClientHello fragmentation")
print(" ✓ combined - Fragmentation (fake_sni needs root for seq_id)")
print(" ★ Recommended: fragment or combined")
if platform.system() != "Windows":
print(" ℹ Run with sudo/root for raw injection (seq_id trick)")
# ─── Main ────────────────────────────────────────────────────────────────────
def main():
"""Main entry point."""
args = parse_args()
# Handle special commands
if args.generate_config:
generate_config(args.generate_config)
return
if args.info:
print(BANNER)
show_platform_info()
return
# Print banner
print(BANNER)
# Setup logging
logger = setup_logging(verbose=args.verbose, quiet=args.quiet)
# Load configuration
if args.config:
config = load_config(args.config)
else:
config = DEFAULT_CONFIG.copy()
# Override with CLI arguments
if args.listen:
host, port = parse_host_port(args.listen, "0.0.0.0", 40443)
config["LISTEN_HOST"] = host
config["LISTEN_PORT"] = port
if args.connect:
host, port = parse_host_port(args.connect, "188.114.98.0", 443)
config["CONNECT_IP"] = host
config["CONNECT_PORT"] = port
if args.sni:
config["FAKE_SNI"] = args.sni
if args.method:
config["BYPASS_METHOD"] = args.method
if args.fragment_strategy:
config["FRAGMENT_STRATEGY"] = args.fragment_strategy
if args.fragment_delay is not None:
config["FRAGMENT_DELAY"] = args.fragment_delay
if args.ttl_trick:
config["USE_TTL_TRICK"] = True
# Validate configuration
if not is_valid_port(config["LISTEN_PORT"]):
print(f"Error: Invalid listen port: {config['LISTEN_PORT']}")
sys.exit(1)
if not is_valid_port(config["CONNECT_PORT"]):
print(f"Error: Invalid connect port: {config['CONNECT_PORT']}")
sys.exit(1)
# Resolve target host if needed
config["CONNECT_IP"] = resolve_host(config["CONNECT_IP"])
# Detect interface IP
interface_ip = get_default_interface_ipv4(config["CONNECT_IP"])
logger.info(f"Default interface: {interface_ip or 'auto'}")
# Try to start raw injector (Linux + root only)
raw_injector = None
use_raw = not getattr(args, 'no_raw', False)
method = config.get("BYPASS_METHOD", "fragment").lower()
if use_raw and method in ("fake_sni", "combined") and interface_ip:
if is_raw_available():
from sni_spoofing.bypass.raw_injector import RawInjector
raw_injector = RawInjector(
local_ip=interface_ip,
remote_ip=config["CONNECT_IP"],
remote_port=config["CONNECT_PORT"],
fake_sni_builder=None,
)
if not raw_injector.start():
logger.warning(
"Raw injector failed to start. "
"Falling back to fragmentation."
)
raw_injector = None
else:
if method == "fake_sni":
logger.warning(
"Raw sockets not available (need root/CAP_NET_RAW). "
"fake_sni will fall back to fragmentation."
)
elif method == "combined":
logger.info(
"Raw sockets not available. "
"Using fragmentation-only bypass."
)
# Build bypass strategy
strategy = build_strategy(config, raw_injector=raw_injector)
# Show configuration summary
logger.info(f"Platform: {platform.system()} {platform.machine()}")
logger.info(f"Python: {platform.python_version()}")
# Setup signal handlers for graceful shutdown
def signal_handler(sig, frame):
print("\n\nShutting down...")
if raw_injector:
raw_injector.stop()
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
if hasattr(signal, "SIGTERM"):
signal.signal(signal.SIGTERM, signal_handler)
# Run the server
try:
asyncio.run(
start_server(
listen_host=config["LISTEN_HOST"],
listen_port=config["LISTEN_PORT"],
connect_ip=config["CONNECT_IP"],
connect_port=config["CONNECT_PORT"],
fake_sni=config["FAKE_SNI"],
bypass_strategy=strategy,
interface_ip=interface_ip,
raw_injector=raw_injector,
)
)
except KeyboardInterrupt:
print("\nShutting down...")
except PermissionError:
print(f"\nError: Permission denied on port {config['LISTEN_PORT']}.")
if config["LISTEN_PORT"] < 1024:
print("Ports below 1024 require root/administrator privileges.")
print(f"Try: sudo {sys.argv[0]} ... or use a port >= 1024")
sys.exit(1)
except OSError as e:
if "address already in use" in str(e).lower():
print(f"\nError: Port {config['LISTEN_PORT']} is already in use.")
print("Use --listen :PORT to specify a different port.")
else:
print(f"\nError: {e}")
sys.exit(1)
finally:
if raw_injector:
raw_injector.stop()
if __name__ == "__main__":
main()
sni_spoofing/forwarder.py
+248
View File
@@ -0,0 +1,248 @@
"""Core TCP forwarder with DPI bypass.
This is the main engine that:
1. Listens for incoming TCP connections
2. Reads the first TLS ClientHello from the client
3. Applies the chosen DPI bypass strategy
4. Relays data bidirectionally between client and server
When a raw injector is available (Linux + root), it registers each
outgoing connection so the sniffer can capture the SYN/ACK handshake
and inject the fake ClientHello with an out-of-window seq number.
Uses pure userspace techniques on platforms without raw socket support.
"""
import asyncio
import logging
import socket
import sys
import traceback
from typing import Optional
from .bypass.base import BypassStrategy
from .tls import ClientHelloBuilder
logger = logging.getLogger("snispf")
# Buffer size for socket operations
BUFFER_SIZE = 65535
async def handle_connection(
incoming_sock: socket.socket,
incoming_addr: tuple,
connect_ip: str,
connect_port: int,
fake_sni: str,
bypass_strategy: BypassStrategy,
interface_ip: Optional[str] = None,
raw_injector=None,
):
"""Handle a single incoming connection.
Flow:
1. Read first data from client (should be TLS ClientHello)
2. Create outgoing socket, optionally register with raw injector
3. Connect to target server (3-way handshake happens here;
the raw injector captures SYN and injects after 3rd ACK)
4. Apply the bypass strategy (sends real data, waits for inject confirmation)
5. Relay data bidirectionally
"""
loop = asyncio.get_running_loop()
outgoing_sock = None
local_port = None
try:
# Read the first data from client (should be TLS ClientHello)
first_data = await asyncio.wait_for(
loop.sock_recv(incoming_sock, BUFFER_SIZE),
timeout=30.0,
)
if not first_data:
incoming_sock.close()
return
# Parse to see if it's a TLS ClientHello
parsed = ClientHelloBuilder.parse_client_hello(first_data)
client_sni = parsed.get("sni", "unknown")
logger.info(
f"[{incoming_addr[0]}:{incoming_addr[1]}] -> "
f"{connect_ip}:{connect_port} | SNI: {client_sni} | "
f"Bypass: {bypass_strategy.name}"
)
# Create outgoing socket
outgoing_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
outgoing_sock.setblocking(False)
# Bind to specific interface if configured
if interface_ip:
outgoing_sock.bind((interface_ip, 0))
# Set keepalive
outgoing_sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
try:
outgoing_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 60)
outgoing_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 10)
outgoing_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 5)
except (AttributeError, OSError):
pass # Not available on all platforms
# If raw injector is available, register the outgoing port
# BEFORE connecting so the sniffer can see the SYN.
if raw_injector is not None:
# We need to bind first to know the local port
if not interface_ip:
outgoing_sock.bind(("", 0))
local_port = outgoing_sock.getsockname()[1]
fake_hello = ClientHelloBuilder.build_client_hello(sni=fake_sni)
raw_injector.register_port(local_port, fake_hello)
# Connect to target server (triggers SYN -> SYN+ACK -> ACK)
await asyncio.wait_for(
loop.sock_connect(outgoing_sock, (connect_ip, connect_port)),
timeout=30.0,
)
# If we didn't know the port before, grab it now
if local_port is None and raw_injector is not None:
local_port = outgoing_sock.getsockname()[1]
# Apply DPI bypass strategy
# The strategy handles:
# - Waiting for raw injection confirmation (if available)
# - Sending the real ClientHello (fragmented or not)
success = await bypass_strategy.apply(
client_sock=incoming_sock,
server_sock=outgoing_sock,
fake_sni=fake_sni,
first_data=first_data,
loop=loop,
)
if not success:
logger.warning(
f"[{incoming_addr[0]}:{incoming_addr[1]}] "
f"Bypass strategy '{bypass_strategy.name}' failed, "
f"falling back to direct relay"
)
# Fallback: just send the data directly
await loop.sock_sendall(outgoing_sock, first_data)
# Bidirectional relay
done = asyncio.Event()
async def _relay(s_in, s_out, label):
try:
while True:
data = await loop.sock_recv(s_in, BUFFER_SIZE)
if not data:
break
await loop.sock_sendall(s_out, data)
except (ConnectionResetError, BrokenPipeError, OSError):
pass
except Exception:
logger.debug(f"Relay error ({label}): {traceback.format_exc()}")
finally:
done.set()
c2s_task = loop.create_task(_relay(incoming_sock, outgoing_sock, "C->S"))
s2c_task = loop.create_task(_relay(outgoing_sock, incoming_sock, "S->C"))
# Wait until one direction closes, then cancel the other
await done.wait()
c2s_task.cancel()
s2c_task.cancel()
await asyncio.gather(c2s_task, s2c_task, return_exceptions=True)
except asyncio.TimeoutError:
logger.debug(f"[{incoming_addr[0]}:{incoming_addr[1]}] Connection timeout")
except Exception:
logger.debug(f"Connection handler error: {traceback.format_exc()}")
finally:
try:
incoming_sock.close()
except Exception:
pass
try:
if outgoing_sock:
outgoing_sock.close()
except Exception:
pass
# Clean up raw injector port state
if raw_injector is not None and local_port is not None:
raw_injector.cleanup_port(local_port)
async def start_server(
listen_host: str,
listen_port: int,
connect_ip: str,
connect_port: int,
fake_sni: str,
bypass_strategy: BypassStrategy,
interface_ip: Optional[str] = None,
raw_injector=None,
):
"""Start the TCP forwarding server.
Creates a listening socket and handles incoming connections,
applying the DPI bypass strategy to each one.
"""
# Create listening socket
server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_sock.setblocking(False)
server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_sock.bind((listen_host, listen_port))
# Set keepalive on the listening socket
server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
try:
server_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 60)
server_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 10)
server_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 5)
except (AttributeError, OSError):
pass
server_sock.listen(128)
loop = asyncio.get_running_loop()
logger.info(f"Listening on {listen_host}:{listen_port}")
logger.info(f"Forwarding to {connect_ip}:{connect_port}")
logger.info(f"Fake SNI: {fake_sni}")
logger.info(f"Bypass strategy: {bypass_strategy.name}")
if raw_injector is not None:
logger.info("Raw packet injection: ACTIVE (seq_id trick enabled)")
else:
logger.info("Raw packet injection: not available (fragmentation only)")
logger.info(f"Interface IP: {interface_ip or 'auto'}")
logger.info("=" * 60)
logger.info("Ready! Configure your application to use:")
logger.info(f" Address: 127.0.0.1:{listen_port}")
logger.info("=" * 60)
try:
while True:
incoming_sock, addr = await loop.sock_accept(server_sock)
incoming_sock.setblocking(False)
loop.create_task(
handle_connection(
incoming_sock=incoming_sock,
incoming_addr=addr,
connect_ip=connect_ip,
connect_port=connect_port,
fake_sni=fake_sni,
bypass_strategy=bypass_strategy,
interface_ip=interface_ip,
raw_injector=raw_injector,
)
)
except asyncio.CancelledError:
pass
finally:
server_sock.close()
logger.info("Server stopped.")
sni_spoofing/tls/__init__.py
+433
View File
@@ -0,0 +1,433 @@
"""TLS ClientHello builder and parser module.
Constructs TLS 1.3 ClientHello messages with customizable SNI fields
for DPI bypass purposes.
"""
import struct
import os
from typing import Optional
class ClientHelloBuilder:
"""Builds TLS ClientHello packets with spoofed SNI.
The ClientHello is the first message in a TLS handshake. DPI systems
inspect the SNI (Server Name Indication) extension to determine the
destination hostname. By sending a ClientHello with a fake SNI to an
allowed domain, we can bypass SNI-based filtering.
"""
# Pre-built template parts from the original tool
# TLS Record Header + Handshake Header + Client Version + ...
# Cipher suites, compression methods, and most extensions are static
# Only SNI, session_id, random, and key_share are dynamic
# TLS 1.3 cipher suites that look legitimate
CIPHER_SUITES = bytes.fromhex(
"0024" # length = 36 bytes (18 cipher suites x 2)
"1302" # TLS_AES_256_GCM_SHA384
"1303" # TLS_CHACHA20_POLY1305_SHA256
"1301" # TLS_AES_128_GCM_SHA256
"c02c" # TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384
"c030" # TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384
"c02b" # TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256
"c02f" # TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256
"cca9" # TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256
"cca8" # TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256
"c024" # TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA384
"c028" # TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA384
"c023" # TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256
"c027" # TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA256
"009f" # TLS_DHE_RSA_WITH_AES_256_GCM_SHA384
"009e" # TLS_DHE_RSA_WITH_AES_128_GCM_SHA256
"006b" # TLS_DHE_RSA_WITH_AES_256_CBC_SHA256
"0067" # TLS_DHE_RSA_WITH_AES_128_CBC_SHA256
"00ff" # TLS_EMPTY_RENEGOTIATION_INFO_SCSV
)
# Supported groups extension
SUPPORTED_GROUPS = bytes.fromhex(
"000a" # extension type: supported_groups
"0016" # length
"0014" # list length
"001d" # x25519
"0017" # secp256r1
"001e" # x448
"0019" # secp521r1
"0018" # secp384r1
"0100" # ffdhe2048
"0101" # ffdhe3072
"0102" # ffdhe4096
"0103" # ffdhe6144
"0104" # ffdhe8192
)
# Signature algorithms extension
SIGNATURE_ALGORITHMS = bytes.fromhex(
"000d" # extension type: signature_algorithms
"002a" # length
"0028" # list length
"0403" # ecdsa_secp256r1_sha256
"0503" # ecdsa_secp384r1_sha384
"0603" # ecdsa_secp521r1_sha512
"0807" # ed25519
"0808" # ed448
"0809" # ...
"080a"
"080b"
"0804" # rsa_pss_rsae_sha256
"0805" # rsa_pss_rsae_sha384
"0806" # rsa_pss_rsae_sha512
"0401" # rsa_pkcs1_sha256
"0501" # rsa_pkcs1_sha384
"0601" # rsa_pkcs1_sha512
"0303" # ...
"0301"
"0302"
"0402"
"0502"
"0602"
)
# EC point formats
EC_POINT_FORMATS = bytes.fromhex(
"000b" # extension type: ec_point_formats
"0004" # length
"0300" # list length + uncompressed
"0102" # ansiX962_compressed_prime + ansiX962_compressed_char2
)
# Session ticket extension (empty)
SESSION_TICKET = bytes.fromhex(
"0023" # extension type: session_ticket
"0000" # length: 0
)
# ALPN extension (h2, http/1.1)
ALPN = bytes.fromhex(
"0010" # extension type: ALPN
"000e" # length
"000c" # protocols length
"0268" # length + 'h'
"3208" # '2' + length
"6874" # 'ht'
"7470" # 'tp'
"2f31" # '/1'
"2e31" # '.1'
)
# Encrypt then MAC
ENCRYPT_THEN_MAC = bytes.fromhex("0016" "0000")
# Extended master secret
EXTENDED_MASTER_SECRET = bytes.fromhex("0017" "0000")
# Supported versions extension (TLS 1.3, TLS 1.2)
SUPPORTED_VERSIONS = bytes.fromhex(
"002b" # extension type: supported_versions
"0005" # length: 5 bytes of data follow
"04" # supported_versions list length: 4 bytes (2 versions x 2 bytes)
"0304" # TLS 1.3
"0303" # TLS 1.2
)
# PSK key exchange modes
PSK_KEY_EXCHANGE = bytes.fromhex(
"002d" # extension type: psk_key_exchange_modes
"0002" # length
"0101" # psk_dhe_ke
)
@classmethod
def build_sni_extension(cls, sni: str) -> bytes:
"""Build the SNI (Server Name Indication) extension."""
sni_bytes = sni.encode("ascii")
sni_len = len(sni_bytes)
# Server name entry: type(1) + length(2) + name
entry = struct.pack("!BH", 0, sni_len) + sni_bytes
# Server name list: length(2) + entries
name_list = struct.pack("!H", len(entry)) + entry
# Extension: type(2) + length(2) + data
return struct.pack("!HH", 0x0000, len(name_list)) + name_list
@classmethod
def build_key_share_extension(cls, public_key: Optional[bytes] = None) -> bytes:
"""Build the key_share extension with x25519 key."""
if public_key is None:
public_key = os.urandom(32)
# Key share entry: group(2) + key_length(2) + key
entry = struct.pack("!HH", 0x001D, 32) + public_key
# Key share extension: length(2) + entries
data = struct.pack("!H", len(entry)) + entry
return struct.pack("!HH", 0x0033, len(data)) + data
@classmethod
def build_padding_extension(cls, target_length: int, current_length: int) -> bytes:
"""Build padding extension to reach target ClientHello size.
Padding is used to make the ClientHello a specific size, which helps
avoid fingerprinting and ensures consistent packet sizes.
"""
# Extension header is 4 bytes (type + length)
padding_needed = target_length - current_length - 4
if padding_needed < 0:
return b""
return struct.pack("!HH", 0x0015, padding_needed) + (b"\x00" * padding_needed)
@classmethod
def build_client_hello(
cls,
sni: str,
session_id: Optional[bytes] = None,
random_bytes: Optional[bytes] = None,
key_share: Optional[bytes] = None,
target_size: int = 517,
) -> bytes:
"""Build a complete TLS ClientHello record.
Args:
sni: The Server Name Indication to include
session_id: 32-byte session ID (random if None)
random_bytes: 32-byte client random (random if None)
key_share: 32-byte x25519 public key (random if None)
target_size: Target total size for the TLS record (default 517)
Returns:
Complete TLS record bytes ready to send
"""
if session_id is None:
session_id = os.urandom(32)
if random_bytes is None:
random_bytes = os.urandom(32)
# Client version: TLS 1.2 (0x0303) - real version in extensions
client_version = b"\x03\x03"
# Session ID
session_id_field = struct.pack("!B", len(session_id)) + session_id
# Compression methods: null only
compression = b"\x01\x00"
# Build extensions
sni_ext = cls.build_sni_extension(sni)
key_share_ext = cls.build_key_share_extension(key_share)
# Assemble extensions (order matters for fingerprint matching)
extensions = b"".join([
sni_ext,
cls.EC_POINT_FORMATS,
cls.SUPPORTED_GROUPS,
cls.SESSION_TICKET,
cls.ALPN,
cls.ENCRYPT_THEN_MAC,
cls.EXTENDED_MASTER_SECRET,
cls.SIGNATURE_ALGORITHMS,
cls.SUPPORTED_VERSIONS,
cls.PSK_KEY_EXCHANGE,
key_share_ext,
])
# Calculate size for padding
# Handshake body (without record header): version(2) + random(32) + session_id_field + cipher_suites + compression + extensions_header(2) + extensions
handshake_body_no_pad = (
client_version
+ random_bytes
+ session_id_field
+ cls.CIPHER_SUITES
+ compression
)
extensions_len_so_far = len(extensions)
# Total handshake msg = 4 (handshake header) + body + 2 (extensions length) + extensions
total_so_far = 4 + len(handshake_body_no_pad) + 2 + extensions_len_so_far
# TLS record = 5 (record header) + handshake
record_so_far = 5 + total_so_far
# Add padding to reach target size
padding_ext = cls.build_padding_extension(target_size, record_so_far)
extensions += padding_ext
# Extensions length prefix
extensions_with_len = struct.pack("!H", len(extensions)) + extensions
# Handshake body
handshake_body = handshake_body_no_pad + extensions_with_len
# Handshake message: type(1) + length(3) + body
handshake_len = len(handshake_body)
handshake = (
b"\x01" # ClientHello
+ struct.pack("!I", handshake_len)[1:] # 3-byte length
+ handshake_body
)
# TLS record: content_type(1) + version(2) + length(2) + data
record = (
b"\x16" # Handshake
+ b"\x03\x01" # TLS 1.0 (legacy for compatibility)
+ struct.pack("!H", len(handshake))
+ handshake
)
return record
@classmethod
def build_client_response(cls, random_bytes: Optional[bytes] = None) -> bytes:
"""Build a fake TLS client response (ChangeCipherSpec + ApplicationData).
This simulates the client's response after receiving ServerHello,
which is useful for making the connection look legitimate to DPI.
"""
if random_bytes is None:
random_bytes = os.urandom(32)
# Change Cipher Spec
ccs = b"\x14\x03\x03\x00\x01\x01"
# Application Data (fake encrypted payload)
app_data = (
b"\x17" # Application Data
+ b"\x03\x03" # TLS 1.2
+ struct.pack("!H", len(random_bytes))
+ random_bytes
)
return ccs + app_data
@staticmethod
def parse_client_hello(data: bytes) -> dict:
"""Parse a TLS ClientHello to extract SNI and other fields.
Args:
data: Raw TLS record bytes
Returns:
Dictionary with parsed fields
"""
result = {}
if len(data) < 5:
return result
# TLS Record header
content_type = data[0]
tls_version = struct.unpack("!H", data[1:3])[0]
record_len = struct.unpack("!H", data[3:5])[0]
result["content_type"] = content_type
result["tls_version"] = f"0x{tls_version:04x}"
if content_type != 0x16: # Not handshake
return result
pos = 5 # Skip record header
# Handshake header
if pos + 4 > len(data):
return result
hs_type = data[pos]
hs_len = struct.unpack("!I", b"\x00" + data[pos + 1 : pos + 4])[0]
pos += 4
if hs_type != 0x01: # Not ClientHello
return result
result["handshake_type"] = "ClientHello"
# Client version
client_version = struct.unpack("!H", data[pos : pos + 2])[0]
result["client_version"] = f"0x{client_version:04x}"
pos += 2
# Random (32 bytes)
result["random"] = data[pos : pos + 32].hex()
pos += 32
# Session ID
sess_len = data[pos]
pos += 1
result["session_id"] = data[pos : pos + sess_len].hex()
pos += sess_len
# Cipher suites
cs_len = struct.unpack("!H", data[pos : pos + 2])[0]
pos += 2 + cs_len
# Compression
comp_len = data[pos]
pos += 1 + comp_len
# Extensions
if pos + 2 > len(data):
return result
ext_len = struct.unpack("!H", data[pos : pos + 2])[0]
pos += 2
ext_end = pos + ext_len
while pos + 4 <= ext_end:
ext_type = struct.unpack("!H", data[pos : pos + 2])[0]
ext_data_len = struct.unpack("!H", data[pos + 2 : pos + 4])[0]
ext_data = data[pos + 4 : pos + 4 + ext_data_len]
pos += 4 + ext_data_len
if ext_type == 0x0000: # SNI
if len(ext_data) >= 5:
name_list_len = struct.unpack("!H", ext_data[0:2])[0]
name_type = ext_data[2]
name_len = struct.unpack("!H", ext_data[3:5])[0]
sni = ext_data[5 : 5 + name_len].decode("ascii", errors="replace")
result["sni"] = sni
return result
@staticmethod
def parse_server_hello(data: bytes) -> dict:
"""Parse a TLS ServerHello message."""
result = {}
if len(data) < 5:
return result
content_type = data[0]
if content_type != 0x16:
return result
pos = 5 # Skip record header
if pos + 4 > len(data):
return result
hs_type = data[pos]
pos += 4
if hs_type != 0x02: # Not ServerHello
return result
result["handshake_type"] = "ServerHello"
# Server version
server_version = struct.unpack("!H", data[pos : pos + 2])[0]
result["server_version"] = f"0x{server_version:04x}"
pos += 2
# Server random
result["random"] = data[pos : pos + 32].hex()
pos += 32
# Session ID
sess_len = data[pos]
pos += 1
result["session_id"] = data[pos : pos + sess_len].hex()
pos += sess_len
# Cipher suite
cipher = struct.unpack("!H", data[pos : pos + 2])[0]
result["cipher_suite"] = f"0x{cipher:04x}"
pos += 2
# Compression
result["compression"] = data[pos]
pos += 1
return result
sni_spoofing/tls/fragment.py
+157
View File
@@ -0,0 +1,157 @@
"""TLS record fragmentation utilities.
Implements various strategies for splitting TLS records to confuse
DPI (Deep Packet Inspection) systems that don't fully reassemble
TLS handshakes.
"""
import struct
from typing import List, Tuple
def fragment_client_hello(data: bytes, strategy: str = "sni_split") -> List[bytes]:
"""Fragment a TLS ClientHello into multiple TCP segments.
DPI systems often only inspect the first packet or fail to reassemble
fragmented TLS records. By splitting the ClientHello at strategic points
(especially around the SNI extension), we can hide the real SNI.
Args:
data: Complete TLS record bytes
strategy: Fragmentation strategy:
- "sni_split": Split right in the middle of the SNI value
- "half": Split the record in half
- "multi": Split into many small fragments
- "tls_record_frag": Use TLS-level record fragmentation
- "none": No fragmentation
Returns:
List of byte fragments to send as separate TCP segments
"""
if strategy == "none" or len(data) < 10:
return [data]
if strategy == "sni_split":
return _fragment_at_sni(data)
elif strategy == "half":
mid = len(data) // 2
return [data[:mid], data[mid:]]
elif strategy == "multi":
return _fragment_multi(data)
elif strategy == "tls_record_frag":
return _tls_record_fragment(data)
else:
return [data]
def _find_sni_offset(data: bytes) -> Tuple[int, int]:
"""Find the offset and length of the SNI value in a ClientHello.
Returns:
Tuple of (sni_value_offset, sni_value_length) or (-1, 0) if not found
"""
# Look for SNI extension type (0x0000) followed by reasonable length
pos = 0
while pos < len(data) - 10:
# Look for the SNI extension pattern: 00 00 xx xx xx xx 00 xx xx 00
if data[pos] == 0x00 and data[pos + 1] == 0x00:
try:
ext_len = struct.unpack("!H", data[pos + 2 : pos + 4])[0]
if 4 < ext_len < 256: # Reasonable SNI extension length
list_len = struct.unpack("!H", data[pos + 4 : pos + 6])[0]
name_type = data[pos + 6]
name_len = struct.unpack("!H", data[pos + 7 : pos + 9])[0]
if name_type == 0 and name_len > 0 and name_len < 256:
sni_start = pos + 9
# Verify it looks like a domain name
sni_data = data[sni_start : sni_start + name_len]
if all(0x20 <= b < 0x7F for b in sni_data):
return sni_start, name_len
except (struct.error, IndexError):
pass
pos += 1
return -1, 0
def _fragment_at_sni(data: bytes) -> List[bytes]:
"""Split the TLS record right in the middle of the SNI value."""
sni_offset, sni_len = _find_sni_offset(data)
if sni_offset < 0:
# Fallback to half split
mid = len(data) // 2
return [data[:mid], data[mid:]]
# Split in the middle of the SNI hostname
split_point = sni_offset + sni_len // 2
return [data[:split_point], data[split_point:]]
def _fragment_multi(data: bytes, chunk_size: int = 24) -> List[bytes]:
"""Split into many small fragments.
Each fragment gets sent as its own TCP segment with TCP_NODELAY.
A chunk size of 24 bytes keeps the fragment count reasonable
(about 22 fragments for a 517-byte ClientHello) while still being
small enough that no single fragment contains the entire SNI.
"""
fragments = []
for i in range(0, len(data), chunk_size):
fragments.append(data[i : i + chunk_size])
return fragments
def _tls_record_fragment(data: bytes) -> List[bytes]:
"""Use TLS-level record fragmentation.
Instead of splitting at the TCP level, we create multiple valid
TLS records that together contain the full handshake message.
This is a more sophisticated approach that some DPI systems
can't handle.
"""
if len(data) < 6 or data[0] != 0x16:
return [data]
# Extract the handshake data from the TLS record
record_version = data[1:3]
handshake_data = data[5:]
# Split the handshake data into two parts
mid = len(handshake_data) // 2
part1 = handshake_data[:mid]
part2 = handshake_data[mid:]
# Create two separate TLS records
record1 = b"\x16" + record_version + struct.pack("!H", len(part1)) + part1
record2 = b"\x16" + record_version + struct.pack("!H", len(part2)) + part2
return [record1, record2]
def fragment_data(data: bytes, sizes: List[int]) -> List[bytes]:
"""Fragment data into specified sizes.
Args:
data: Raw bytes to fragment
sizes: List of fragment sizes. Last fragment gets remaining data.
Returns:
List of byte fragments
"""
fragments = []
pos = 0
for i, size in enumerate(sizes):
if pos >= len(data):
break
if i == len(sizes) - 1:
# Last specified size: include all remaining data
fragments.append(data[pos:])
else:
fragments.append(data[pos : pos + size])
pos += size
# If we consumed all specified sizes but data remains
if pos < len(data) and len(fragments) < len(sizes):
fragments.append(data[pos:])
return fragments if fragments else [data]
sni_spoofing/utils/__init__.py
+130
View File
@@ -0,0 +1,130 @@
"""Network utility functions.
Cross-platform network interface detection and helpers.
"""
import socket
import sys
import platform
from typing import Optional
def get_default_interface_ipv4(dest: str = "8.8.8.8") -> Optional[str]:
"""Get the IPv4 address of the default network interface.
Creates a UDP socket and connects to a public address to determine
which local IP would be used for outgoing connections.
Args:
dest: Destination IP to determine route (not actually contacted)
Returns:
Local IPv4 address string, or None on failure
"""
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect((dest, 53))
addr = s.getsockname()[0]
s.close()
return addr
except OSError:
return None
def get_default_interface_ipv6(dest: str = "2001:4860:4860::8888") -> Optional[str]:
"""Get the IPv6 address of the default network interface.
Args:
dest: Destination IPv6 to determine route
Returns:
Local IPv6 address string, or None on failure
"""
try:
s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
s.connect((dest, 53))
addr = s.getsockname()[0]
s.close()
return addr
except OSError:
return None
def check_platform_capabilities() -> dict:
"""Check what DPI bypass capabilities are available on this platform.
Returns:
Dictionary of available features
"""
caps = {
"platform": platform.system(),
"python_version": sys.version,
"fragment_support": True, # Always available (userspace TCP)
"tls_record_frag": True, # Always available (application layer)
"fake_sni": True, # Always available (application layer)
"tcp_nodelay": True, # Always available
"raw_socket": False, # Platform-dependent
"ip_ttl_trick": False, # Platform-dependent
}
# Check raw socket support (needed for advanced tricks)
try:
if platform.system() != "Windows":
s = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP)
s.close()
caps["raw_socket"] = True
caps["ip_ttl_trick"] = True
else:
# Windows raw sockets are limited
caps["raw_socket"] = False
except (PermissionError, OSError):
pass
# Check AF_PACKET support (Linux only, needed for seq_id injection)
try:
if platform.system() == "Linux":
s = socket.socket(
socket.AF_PACKET, socket.SOCK_RAW, socket.htons(0x0003)
)
s.close()
caps["af_packet"] = True
caps["raw_injection"] = True
else:
caps["af_packet"] = False
caps["raw_injection"] = False
except (PermissionError, OSError, AttributeError):
caps["af_packet"] = False
caps["raw_injection"] = False
return caps
def resolve_host(host: str) -> str:
"""Resolve hostname to IP address.
Args:
host: Hostname or IP address
Returns:
IP address string
"""
try:
return socket.gethostbyname(host)
except socket.gaierror:
return host
def is_valid_ip(addr: str) -> bool:
"""Check if string is a valid IPv4 or IPv6 address."""
for family in (socket.AF_INET, socket.AF_INET6):
try:
socket.inet_pton(family, addr)
return True
except (socket.error, OSError):
continue
return False
def is_valid_port(port: int) -> bool:
"""Check if port number is valid."""
return isinstance(port, int) and 1 <= port <= 65535