From 369b49412ef2dc942a6fea241ae0db7758f5f663 Mon Sep 17 00:00:00 2001 From: root Date: Thu, 28 May 2026 02:18:25 +0430 Subject: [PATCH] initial commit --- .gitignore | 22 + LICENSE | 21 + config.json | 12 + pyproject.toml | 43 ++ run.py | 12 + sni_spoofing/__init__.py | 6 + sni_spoofing/bypass/__init__.py | 16 + sni_spoofing/bypass/base.py | 44 ++ sni_spoofing/bypass/combined.py | 143 ++++++ sni_spoofing/bypass/fake_sni.py | 296 +++++++++++ sni_spoofing/bypass/fragment.py | 89 ++++ sni_spoofing/bypass/raw_injector.py | 425 ++++++++++++++++ sni_spoofing/cli.py | 670 +++++++++++++++++++++++++ sni_spoofing/forwarder.py | 377 ++++++++++++++ sni_spoofing/scanner/__init__.py | 21 + sni_spoofing/scanner/domain_checker.py | 430 ++++++++++++++++ sni_spoofing/tls/__init__.py | 433 ++++++++++++++++ sni_spoofing/tls/fragment.py | 161 ++++++ sni_spoofing/utils/__init__.py | 130 +++++ tests/__init__.py | 0 tests/test_tls.py | 448 +++++++++++++++++ 21 files changed, 3799 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 config.json create mode 100644 pyproject.toml create mode 100644 run.py create mode 100644 sni_spoofing/__init__.py create mode 100644 sni_spoofing/bypass/__init__.py create mode 100644 sni_spoofing/bypass/base.py create mode 100644 sni_spoofing/bypass/combined.py create mode 100644 sni_spoofing/bypass/fake_sni.py create mode 100644 sni_spoofing/bypass/fragment.py create mode 100644 sni_spoofing/bypass/raw_injector.py create mode 100644 sni_spoofing/cli.py create mode 100644 sni_spoofing/forwarder.py create mode 100644 sni_spoofing/scanner/__init__.py create mode 100644 sni_spoofing/scanner/domain_checker.py create mode 100644 sni_spoofing/tls/__init__.py create mode 100644 sni_spoofing/tls/fragment.py create mode 100644 sni_spoofing/utils/__init__.py create mode 100644 tests/__init__.py create mode 100644 tests/test_tls.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..98de7ba --- /dev/null +++ b/.gitignore @@ -0,0 +1,22 @@ +__pycache__/ +*.py[cod] +*$py.class +*.so +*.egg-info/ +dist/ +build/ +*.egg +.eggs/ +.venv/ +venv/ +env/ +.env +*.log +.DS_Store +Thumbs.db +*.swp +*.swo +*~ +.idea/ +.vscode/ +*.bak diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..a96d35b --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2024 Rainman69 + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/config.json b/config.json new file mode 100644 index 0000000..d5d92f5 --- /dev/null +++ b/config.json @@ -0,0 +1,12 @@ +{ + "LISTEN_HOST": "0.0.0.0", + "LISTEN_PORT": 40443, + "CONNECT_IP": "104.19.230.21", + "CONNECT_PORT": 443, + "FAKE_SNI": "www.hcaptcha.com", + "BYPASS_METHOD": "fragment", + "FRAGMENT_STRATEGY": "sni_split", + "FRAGMENT_DELAY": 0.1, + "USE_TTL_TRICK": false, + "FAKE_SNI_METHOD": "prefix_fake" +} diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..b185561 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,43 @@ +[build-system] +requires = ["setuptools>=68.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "snispf" +version = "2.0.0" +description = "SNISPF - Cross-platform DPI bypass tool via SNI spoofing" +readme = "README.md" +license = {text = "MIT"} +requires-python = ">=3.8" +authors = [ + {name = "Rainman69"}, +] +keywords = ["sni", "spoofing", "dpi", "bypass", "censorship", "tls", "fragment", "cli"] +classifiers = [ + "Development Status :: 4 - Beta", + "Environment :: Console", + "Intended Audience :: End Users/Desktop", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Topic :: Internet", + "Topic :: Security", + "Topic :: System :: Networking", +] + +[project.urls] +Homepage = "https://github.com/Rainman69/SNISPF" +Repository = "https://github.com/Rainman69/SNISPF" +Issues = "https://github.com/Rainman69/SNISPF/issues" + +[project.scripts] +snispf = "sni_spoofing.cli:main" + +[tool.setuptools.packages.find] +include = ["sni_spoofing*"] diff --git a/run.py b/run.py new file mode 100644 index 0000000..13baf8d --- /dev/null +++ b/run.py @@ -0,0 +1,12 @@ +#!/usr/bin/env python3 +"""Direct entry point for running without installation.""" +import sys +import os + +# Add project root to path +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from sni_spoofing.cli import main + +if __name__ == "__main__": + main() diff --git a/sni_spoofing/__init__.py b/sni_spoofing/__init__.py new file mode 100644 index 0000000..0dc41b7 --- /dev/null +++ b/sni_spoofing/__init__.py @@ -0,0 +1,6 @@ +""" +SNISPF - Cross-platform SNI spoofing and DPI bypass tool. +""" + +__version__ = "2.0.0" +__author__ = "Rainman69" diff --git a/sni_spoofing/bypass/__init__.py b/sni_spoofing/bypass/__init__.py new file mode 100644 index 0000000..cec54d3 --- /dev/null +++ b/sni_spoofing/bypass/__init__.py @@ -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", +] diff --git a/sni_spoofing/bypass/base.py b/sni_spoofing/bypass/base.py new file mode 100644 index 0000000..89ea44b --- /dev/null +++ b/sni_spoofing/bypass/base.py @@ -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}'>" diff --git a/sni_spoofing/bypass/combined.py b/sni_spoofing/bypass/combined.py new file mode 100644 index 0000000..8bc9e12 --- /dev/null +++ b/sni_spoofing/bypass/combined.py @@ -0,0 +1,143 @@ +"""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 via a SEPARATE socket with low TTL + # so it reaches DPI but expires before the server. The + # main socket stays clean for the real TLS handshake. + fake_hello = ClientHelloBuilder.build_client_hello(sni=fake_sni) + try: + remote_addr = server_sock.getpeername() + probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + probe.setblocking(False) + probe.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + for ttl in (1, 2, 3): + try: + probe.setsockopt( + socket.IPPROTO_IP, socket.IP_TTL, ttl + ) + try: + await asyncio.wait_for( + loop.sock_connect(probe, remote_addr), + timeout=0.3, + ) + await loop.sock_sendall(probe, fake_hello) + except (asyncio.TimeoutError, OSError): + pass + break + except OSError: + continue + try: + probe.close() + except OSError: + pass + except OSError: + pass + + await asyncio.sleep(0.05) + + # 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 diff --git a/sni_spoofing/bypass/fake_sni.py b/sni_spoofing/bypass/fake_sni.py new file mode 100644 index 0000000..44f16d8 --- /dev/null +++ b/sni_spoofing/bypass/fake_sni.py @@ -0,0 +1,296 @@ +"""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") + +# Delay between TLS fragments when fragmenting the real ClientHello +# after the seq_id fake injection. Matches the value used in the +# combined strategy so behaviour is consistent. +_REAL_FRAGMENT_DELAY = 0.1 + + +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 the TTL trick (sending a fake ClientHello with + low IP TTL) combined with TLS fragmentation. + + 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. Works on macOS, Android, Linux. + - "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, + use_ttl_trick: bool = False, + fragment_real: bool = True, + fragment_strategy: str = "sni_split"): + """Initialise the fake_sni strategy. + + Args: + method: Sub-method name (kept for backwards compatibility). + raw_injector: Active ``RawInjector`` instance or ``None``. + use_ttl_trick: Force the TTL trick fallback path. + fragment_real: When a raw injector is active, also fragment + the real ClientHello at the SNI boundary after the + out-of-window fake has been confirmed. This protects + against DPI that reassembles TCP and matches the SNI + on the real stream (observed with some xhttp / ws + configs that carry larger ClientHello records, e.g. + multi-value ALPN). Defaults to ``True``. + fragment_strategy: Fragmentation strategy passed through to + ``fragment_client_hello`` when ``fragment_real`` is on. + """ + self.method = method + self.raw_injector = raw_injector + self.use_ttl_trick = use_ttl_trick + self.fragment_real = fragment_real + self.fragment_strategy = fragment_strategy + + 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, use TTL trick if enabled (auto-enabled + # on macOS/Android/unprivileged Linux), otherwise fragment only. + if self.method == "ttl_trick" or self.use_ttl_trick: + return await self._ttl_trick_and_fragment( + 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. + + After the server confirms it ignored the fake, send the real + ClientHello. By default the real ClientHello is also split at + the SNI boundary; this matches the behaviour of the ``combined`` + strategy and is required for stricter DPI that reassembles the + TCP stream and matches the SNI on the real handshake (observed + with xhttp / ws configs that carry larger ClientHellos, e.g. + multi-value ALPN such as ``h3,h2,http/1.1``). + + Set ``fragment_real=False`` to restore the previous behaviour + of sending the real ClientHello as a single segment. + """ + 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. Fragmenting at the SNI boundary + # in addition to the seq_id trick covers DPI that does TCP + # reassembly on the real stream (some xhttp / ws configs). + if self.fragment_real: + try: + server_sock.setsockopt( + socket.IPPROTO_TCP, socket.TCP_NODELAY, 1 + ) + except OSError: + pass + + 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 _REAL_FRAGMENT_DELAY > 0: + await asyncio.sleep(_REAL_FRAGMENT_DELAY) + + try: + server_sock.setsockopt( + socket.IPPROTO_TCP, socket.TCP_NODELAY, 0 + ) + except OSError: + pass + else: + # Legacy path: send the real ClientHello untouched. + await loop.sock_sendall(server_sock, first_data) + + return True + + except Exception: + return False + + async def _ttl_trick_and_fragment( + self, + server_sock: socket.socket, + fake_sni: str, + first_data: bytes, + loop, + ) -> bool: + """Send fake ClientHello via a separate socket, then real data fragmented. + + The fake ClientHello is sent through a **separate** raw TCP socket + (not the proxied connection) with a very low IP TTL. This ensures + the fake reaches DPI middleboxes (typically 1-3 hops away) but + expires before the real server, so the server never sees it and + the real TLS handshake on the main socket stays clean. + + If the separate-socket approach fails (e.g. no permission), + we fall back to pure fragmentation which still works well. + + This is the default fallback on macOS, Android/Termux, and + unprivileged Linux where AF_PACKET raw sockets are not available. + """ + try: + server_sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + + # --- Send fake on a SEPARATE socket with low TTL --- + # This prevents corrupting the real TLS stream. + remote_addr = server_sock.getpeername() + fake_hello = ClientHelloBuilder.build_client_hello(sni=fake_sni) + + try: + probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + probe.setblocking(False) + probe.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + # Use a very low TTL so the packet dies before the server + for ttl in (1, 2, 3): + try: + probe.setsockopt( + socket.IPPROTO_IP, socket.IP_TTL, ttl + ) + # Non-blocking connect -- we don't care if it + # completes; we just want the SYN + fake data + # to traverse the DPI middlebox path. + try: + await asyncio.wait_for( + loop.sock_connect(probe, remote_addr), + timeout=0.3, + ) + await loop.sock_sendall(probe, fake_hello) + except (asyncio.TimeoutError, OSError): + pass + break + except OSError: + continue + try: + probe.close() + except OSError: + pass + except OSError: + # Separate socket approach failed, that's fine + pass + + await asyncio.sleep(0.05) + + # --- Send the real ClientHello fragmented on the main socket --- + 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 + + async def _ttl_trick( + self, + server_sock: socket.socket, + fake_sni: str, + first_data: bytes, + loop, + ) -> bool: + """Legacy TTL trick: send fake with low TTL then real data normally.""" + return await self._ttl_trick_and_fragment( + server_sock, fake_sni, first_data, loop + ) + + 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 diff --git a/sni_spoofing/bypass/fragment.py b/sni_spoofing/bypass/fragment.py new file mode 100644 index 0000000..697f07a --- /dev/null +++ b/sni_spoofing/bypass/fragment.py @@ -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 diff --git a/sni_spoofing/bypass/raw_injector.py b/sni_spoofing/bypass/raw_injector.py new file mode 100644 index 0000000..05d60dc --- /dev/null +++ b/sni_spoofing/bypass/raw_injector.py @@ -0,0 +1,425 @@ +"""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 + try: + self.raw_fd.bind((self.iface_name, ETH_P_ALL)) + except OSError as e: + logger.warning(f"Cannot bind raw socket to {self.iface_name}: {e}") + logger.warning("Raw injection unavailable on this platform") + self.raw_fd.close() + self.raw_fd = None + return False + + 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 diff --git a/sni_spoofing/cli.py b/sni_spoofing/cli.py new file mode 100644 index 0000000..81bfcb5 --- /dev/null +++ b/sni_spoofing/cli.py @@ -0,0 +1,670 @@ +""" +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 104.18.38.202:443 --sni cdnjs.cloudflare.com + snispf --check-domains domains.txt --output verified.txt +""" + +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 ────────────────────────────────────────────────────────────────── + +def _build_banner() -> str: + """Render the startup banner with the current package version. + + The version is read from ``sni_spoofing.__version__`` so the banner always + matches the installed package (fixes the long-standing "source still says + v1.7.0" reports from users who clone the repo at a newer release). + """ + version_line = f"SNI Spoofing + TLS Fragmentation v{__version__}" + # Inside-the-box content is 63 chars wide (between '│ ' and the trailing '│'). + return ( + "\n" + " ███████╗███╗ ██╗██╗███████╗██████╗ ███████╗\n" + " ██╔════╝████╗ ██║██║██╔════╝██╔══██╗██╔════╝\n" + " ███████╗██╔██╗ ██║██║███████╗██████╔╝█████╗\n" + " ╚════██║██║╚██╗██║██║╚════██║██╔═══╝ ██╔══╝\n" + " ███████║██║ ╚████║██║███████║██║ ██║\n" + " ╚══════╝╚═╝ ╚═══╝╚═╝╚══════╝╚═╝ ╚═╝\n" + "\n" + " ┌──────────────────────────────────────────────────────────────────┐\n" + " │ SNISPF - Cross-Platform DPI Bypass Tool │\n" + f" │ {version_line:<63}│\n" + " │ Works on Windows / macOS / Linux │\n" + " │ https://github.com/Rainman69/SNISPF │\n" + " └──────────────────────────────────────────────────────────────────┘\n" + ) + + +BANNER = _build_banner() + +# ─── Logging ───────────────────────────────────────────────────────────────── + +def setup_logging(verbose: bool = False, quiet: bool = False): + """Configure logging with deduplication guard.""" + 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", + ) + + logger = logging.getLogger("snispf") + # Prevent handler accumulation on repeated calls + if not logger.handlers: + handler = logging.StreamHandler(sys.stdout) + handler.setFormatter(formatter) + logger.addHandler(handler) + logger.setLevel(level) + + return logger + + +# ─── Config ────────────────────────────────────────────────────────────────── + +DEFAULT_CONFIG = { + "LISTEN_HOST": "0.0.0.0", + "LISTEN_PORT": 40443, + "CONNECT_IP": "104.18.38.202", + "CONNECT_PORT": 443, + "FAKE_SNI": "cdnjs.cloudflare.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": "104.18.38.202", + "CONNECT_PORT": 443, + "FAKE_SNI": "cdnjs.cloudflare.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, + use_ttl_trick=config.get("USE_TTL_TRICK", False), + # When a raw injector is active, also fragment the real + # ClientHello at the SNI boundary so that DPI which + # reassembles the TCP stream cannot match the SNI on the + # real handshake. Required for some xhttp / ws configs + # (e.g. multi-value ALPN like h3,h2,http/1.1). + fragment_real=config.get("FAKE_SNI_FRAGMENT_REAL", True), + fragment_strategy=config.get("FRAGMENT_STRATEGY", "sni_split"), + ) + 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 104.18.38.202:443 -s cdnjs.cloudflare.com\n" + " %(prog)s -l :40443 -c 104.18.38.202:443 -s www.speedtest.net -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" + "\nDomain Checker:\n" + " %(prog)s --check-domains domains.txt\n" + " %(prog)s --check-domains domains.txt --output verified.txt\n" + " %(prog)s --check-domains domains.txt --check-http\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: 104.18.38.202:443)", + ) + parser.add_argument( + "--sni", "-s", + metavar="HOSTNAME", + help="Fake SNI hostname (default: cdnjs.cloudflare.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", + ) + + # ─── Domain checker ────────────────────────────────────────────── + domain_group = parser.add_argument_group("Domain Checker Options") + domain_group.add_argument( + "--check-domains", + metavar="FILE", + help="Check domains from a file to find Cloudflare-backed ones", + ) + domain_group.add_argument( + "--check-workers", + type=int, + default=50, + metavar="N", + help="Parallel workers for domain checking (default: 50)", + ) + domain_group.add_argument( + "--check-timeout", + type=float, + default=3.0, + metavar="SECONDS", + help="Per-domain timeout for checking (default: 3.0)", + ) + domain_group.add_argument( + "--output", + metavar="FILE", + help="Export verified Cloudflare domains to a file", + ) + domain_group.add_argument( + "--check-http", + action="store_true", + help="Also verify HTTP connectivity during domain check", + ) + + # 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. Returns (host, port).""" + if not addr: + return default_host, default_port + + if addr.startswith(":"): + try: + return default_host, int(addr[1:]) + except ValueError: + print(f"Error: Invalid port in '{addr}'") + sys.exit(1) + + parts = addr.rsplit(":", 1) + if len(parts) == 2: + host = parts[0] or default_host + try: + port = int(parts[1]) + except ValueError: + print(f"Error: Invalid port in '{addr}'") + sys.exit(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(" ✓ fake_sni - TTL trick + fragmentation (auto-enabled)") + print(" ✓ combined - TTL trick + fragmentation (recommended)") + print(" ★ Recommended: combined (auto-uses TTL trick)") + if platform.system() != "Windows": + print(" ℹ Run with sudo/root for raw injection (seq_id trick)") + print(" ℹ TTL trick is auto-enabled when raw sockets are unavailable") + + print("\nDomain Checker:") + print(" ✓ Bulk Cloudflare-backed domain verifier (no privileges needed)") + print(" ★ Use --check-domains FILE to validate SNI candidates") + + +# ─── Domain Checker Command ────────────────────────────────────────────────── + +def run_domain_check(args, logger): + """Execute a bulk domain check and print results.""" + from sni_spoofing.scanner import DomainChecker + + checker = DomainChecker( + concurrency=args.check_workers, + timeout=args.check_timeout, + verify_tls=True, + verify_http=getattr(args, "check_http", False), + ) + + # Load domains from file + try: + domains = checker.load_domains_from_file(args.check_domains) + except FileNotFoundError: + print(f"Error: File not found: {args.check_domains}") + return + except Exception as e: + print(f"Error reading file: {e}") + return + + if not domains: + print("No domains found in file.") + return + + print(f"\n Checking {len(domains)} domains...\n") + + # Progress display + def progress(done, total): + pct = done * 100 // total + bar = "█" * (pct // 5) + "░" * (20 - pct // 5) + print(f"\r Checking: [{bar}] {done}/{total} ({pct}%)", end="", flush=True) + + results = checker.check_domains(domains, progress_cb=progress) + print() # newline after progress bar + + # Display results + cf_results = [r for r in results if r.is_cloudflare] + usable = [r for r in results if r.usable_as_sni] + + print(f"\n{'═' * 90}") + print(f" Domain Check Results") + print(f"{'═' * 90}") + print(f" Total domains: {len(results)}") + print(f" Behind Cloudflare: {len(cf_results)}") + print(f" Usable as SNI: {len(usable)}") + print(f"{'═' * 90}\n") + + # Show Cloudflare-backed domains + print(checker.results_table(results, cloudflare_only=True)) + + # Export if requested + if args.output: + count = checker.export_sni_list(results, args.output) + print(f"\n Exported {count} verified domains to {args.output}") + + # Also show summary for non-CF domains + non_cf = [r for r in results if not r.is_cloudflare and r.ip] + if non_cf: + print(f"\n Note: {len(non_cf)} domains are NOT behind Cloudflare") + print(f" (these will not work for SNI spoofing through Cloudflare IPs)") + print() + + +# ─── 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, "104.18.38.202", 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 + + # ── Domain checker mode ─────────────────────────────────────────── + if args.check_domains: + run_domain_check(args, logger) + return + + # ── Auto-load config.json if present and no --config given ────── + if not args.config: + for candidate in ["config.json", "snispf.json"]: + if os.path.isfile(candidate): + logger.info("Auto-loading config from %s", candidate) + user_config = load_config(candidate) + # Only apply file values for keys that were NOT + # explicitly overridden by CLI arguments. + cli_overridden = set() + if args.listen: + cli_overridden.update(["LISTEN_HOST", "LISTEN_PORT"]) + if args.connect: + cli_overridden.update(["CONNECT_IP", "CONNECT_PORT"]) + if args.sni: + cli_overridden.add("FAKE_SNI") + if args.method: + cli_overridden.add("BYPASS_METHOD") + if args.fragment_strategy: + cli_overridden.add("FRAGMENT_STRATEGY") + if args.fragment_delay is not None: + cli_overridden.add("FRAGMENT_DELAY") + if args.ttl_trick: + cli_overridden.add("USE_TTL_TRICK") + + for key, val in user_config.items(): + if key not in cli_overridden: + config[key] = val + break + + # ── Validate config ─────────────────────────────────────────────── + 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'}") + + # ── Raw injector setup ──────────────────────────────────────────── + 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. " + "Enabling TTL trick as fallback." + ) + raw_injector = None + config["USE_TTL_TRICK"] = True + else: + # No raw sockets (macOS, Android/Termux, unprivileged Linux). + # Auto-enable the TTL trick: sends a fake ClientHello with a + # low IP TTL that reaches the nearby DPI middlebox but expires + # before the real server. Works on any platform that supports + # setsockopt(IP_TTL). + config["USE_TTL_TRICK"] = True + if method == "fake_sni": + logger.info( + "Raw sockets not available (need root/CAP_NET_RAW). " + "fake_sni will use TTL trick + fragmentation." + ) + elif method == "combined": + logger.info( + "Raw sockets not available. " + "Using TTL trick + fragmentation 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() diff --git a/sni_spoofing/forwarder.py b/sni_spoofing/forwarder.py new file mode 100644 index 0000000..1261e35 --- /dev/null +++ b/sni_spoofing/forwarder.py @@ -0,0 +1,377 @@ +"""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. Connects to the configured upstream IP +4. Applies the chosen DPI bypass strategy +5. 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. +""" + +import asyncio +import logging +import socket +import sys +import time +import traceback +from typing import Optional + +# `resource` is a POSIX-only module (Linux/macOS/BSD). It does not exist on +# Windows, so we import it defensively and skip the fd-limit tweak there. +try: + import resource # type: ignore +except ImportError: # pragma: no cover -- Windows + resource = None # type: ignore + +from .bypass.base import BypassStrategy +from .tls import ClientHelloBuilder + +logger = logging.getLogger("snispf") + +# Buffer size for socket operations +BUFFER_SIZE = 65535 + +# How many consecutive failures on a single IP before triggering failover +FAILOVER_THRESHOLD = 3 + +# Rapid failure window -- if we get FAILOVER_THRESHOLD failures within +# this many seconds, the IP is considered blocked. +FAILOVER_WINDOW = 30.0 + +# Maximum concurrent connections. Keeps the process well under the +# OS file-descriptor limit and avoids "Too many open files" crashes +# that macOS users hit with the default 256 fd limit. +MAX_CONCURRENT_CONNECTIONS = 512 + + +def _raise_fd_limit(): + """Try to raise the OS file-descriptor soft limit. + + macOS defaults to 256, which is far too low for a proxy that handles + many parallel connections (each needs 2 fds: incoming + outgoing). + We attempt to raise the soft limit to the hard limit, or at least + to a reasonable value. + + No-op on Windows, where the `resource` module does not exist and + socket count is governed differently. + """ + if resource is None: + return + try: + soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE) + if soft < 4096: + target = min(hard, 65536) if hard > soft else 4096 + try: + resource.setrlimit(resource.RLIMIT_NOFILE, (target, hard)) + logger.debug("Raised fd limit from %d to %d", soft, target) + except (ValueError, OSError): + # On some systems we cannot raise beyond hard limit + try: + resource.setrlimit(resource.RLIMIT_NOFILE, (hard, hard)) + except (ValueError, OSError): + pass + except (AttributeError, OSError): + # resource module not available (unlikely) or unsupported platform + pass + + +class ConnectionTracker: + """Tracks per-IP connection failures to detect blocking.""" + + def __init__(self): + self._failures = {} # ip -> list of failure timestamps + self._successes = {} # ip -> count + + def record_failure(self, ip: str) -> int: + """Record a failure and return how many occurred within the window.""" + now = time.monotonic() + if ip not in self._failures: + self._failures[ip] = [] + self._failures[ip].append(now) + # Prune old entries + cutoff = now - FAILOVER_WINDOW + self._failures[ip] = [t for t in self._failures[ip] if t > cutoff] + return len(self._failures[ip]) + + def record_success(self, ip: str): + """Record a successful connection (resets the failure counter).""" + self._failures.pop(ip, None) + self._successes[ip] = self._successes.get(ip, 0) + 1 + + def should_failover(self, ip: str) -> bool: + count = len(self._failures.get(ip, [])) + return count >= FAILOVER_THRESHOLD + + def clear(self, ip: str): + self._failures.pop(ip, None) + + +# Module-level tracker shared across connections +_conn_tracker = ConnectionTracker() + + +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 + active_ip = connect_ip + active_sni = fake_sni + + 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"{active_ip}:{connect_port} | SNI: {client_sni} | " + f"Fake: {active_sni} | Method: {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=active_sni) + raw_injector.register_port(local_port, fake_hello) + + # Connect to target server (triggers SYN -> SYN+ACK -> ACK) + try: + await asyncio.wait_for( + loop.sock_connect(outgoing_sock, (active_ip, connect_port)), + timeout=15.0, + ) + except (asyncio.TimeoutError, ConnectionRefusedError, OSError) as exc: + fail_count = _conn_tracker.record_failure(active_ip) + logger.debug( + "[%s:%d] Connect to %s failed (%d/%d): %s", + incoming_addr[0], incoming_addr[1], active_ip, + fail_count, FAILOVER_THRESHOLD, exc, + ) + raise + + # 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=active_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) + + # NOTE: Do NOT mark success yet. We need to verify the server + # actually responds with valid data (not a block page or RST). + # Success is recorded only after the first server response + # is received in the relay loop below. + + # Bidirectional relay + done = asyncio.Event() + server_responded = False + + async def _relay(s_in, s_out, label): + nonlocal server_responded + try: + while True: + data = await loop.sock_recv(s_in, BUFFER_SIZE) + if not data: + break + await loop.sock_sendall(s_out, data) + # Record success only when we get the first + # response from the server (S->C direction). + # This proves the connection is actually working + # and the server accepted our ClientHello. + if label == "S->C" and not server_responded: + server_responded = True + _conn_tracker.record_success(active_ip) + 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) + + # If the server never responded, record a failure. + # This catches cases where DPI allows the handshake but + # blocks or RSTs actual application data. + if not server_responded: + _conn_tracker.record_failure(active_ip) + + 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. + """ + # Raise the OS file-descriptor limit before binding + _raise_fd_limit() + + # 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() + + # Semaphore limits concurrent connections to prevent fd exhaustion. + # Each proxied connection uses 2 fds (client + server), plus the + # listening socket itself. This cap prevents the "Too many open + # files" crash that happens on macOS (default fd limit 256) and + # Android/Termux when VPN clients open many connections at once. + conn_semaphore = asyncio.Semaphore(MAX_CONCURRENT_CONNECTIONS) + + 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) + + async def _guarded_handle(sock, addr): + """Wrap handle_connection with the concurrency semaphore.""" + async with conn_semaphore: + await handle_connection( + incoming_sock=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, + ) + + try: + while True: + incoming_sock, addr = await loop.sock_accept(server_sock) + incoming_sock.setblocking(False) + + loop.create_task(_guarded_handle(incoming_sock, addr)) + except asyncio.CancelledError: + pass + finally: + server_sock.close() + logger.info("Server stopped.") diff --git a/sni_spoofing/scanner/__init__.py b/sni_spoofing/scanner/__init__.py new file mode 100644 index 0000000..184fd83 --- /dev/null +++ b/sni_spoofing/scanner/__init__.py @@ -0,0 +1,21 @@ +"""Domain utilities: bulk Cloudflare-domain checker. + +This module previously hosted a Cloudflare clean-IP scanner together with +an SNI rotator and a background re-scan engine. The scanner was never +finished and frequently produced misleading results (rate-limit false +negatives, stale caches, mis-ordered latency rankings), so the entire +``--scan`` / ``--auto`` / ``--rescan`` feature surface has been removed. + +What stays is the lightweight bulk **domain checker**: given a text file +of hostnames it tells you which ones are actually fronted by Cloudflare +and therefore usable as a fake SNI value. It is fully synchronous, has +no background threads, and is safe to invoke from the CLI. +""" + +from .domain_checker import DomainChecker, DomainResult, is_cloudflare_ip + +__all__ = [ + "DomainChecker", + "DomainResult", + "is_cloudflare_ip", +] diff --git a/sni_spoofing/scanner/domain_checker.py b/sni_spoofing/scanner/domain_checker.py new file mode 100644 index 0000000..fa1e6fb --- /dev/null +++ b/sni_spoofing/scanner/domain_checker.py @@ -0,0 +1,430 @@ +"""Bulk SNI domain checker and verifier. + +Checks a list of domains to determine which ones are behind Cloudflare's +CDN and suitable for use as SNI spoof targets. Performs DNS resolution, +ASN lookup, TLS handshake, and HTTP validation to verify each domain. + +Inspired by community scanner tools that identify Cloudflare-fronted +domains for censorship bypass. The goal is to maintain a large, +verified list of domains that can be used as fake SNI values when +connecting through Cloudflare IPs. + +Usage (standalone):: + + checker = DomainChecker(concurrency=50, timeout=3.0) + results = checker.check_domains(["example.com", "test.org"]) + for r in results: + if r.is_cloudflare: + print(f"{r.domain} -> {r.ip} (CF)") + +Usage (from CLI):: + + snispf --check-domains domains.txt + snispf --check-domains domains.txt --output verified.txt +""" + +import asyncio +import concurrent.futures +import ipaddress +import logging +import socket +import ssl +import time +from dataclasses import dataclass, field +from typing import Callable, Dict, List, Optional, Set + +logger = logging.getLogger("snispf") + +# Known Cloudflare ASN numbers +CLOUDFLARE_ASNS = {13335, 209242} + +# Cloudflare's published IPv4 ranges (https://www.cloudflare.com/ips-v4). +# Kept inline here so this module has zero intra-package dependencies; the +# list is small and very stable. Update when Cloudflare publishes new ranges. +CLOUDFLARE_IPV4_RANGES = [ + "173.245.48.0/20", + "103.21.244.0/22", + "103.22.200.0/22", + "103.31.4.0/22", + "141.101.64.0/18", + "108.162.192.0/18", + "190.93.240.0/20", + "188.114.96.0/20", + "197.234.240.0/22", + "198.41.128.0/17", + "162.158.0.0/15", + "104.16.0.0/13", + "104.24.0.0/14", + "172.64.0.0/13", + "131.0.72.0/22", +] + +# Cloudflare IP networks (parsed once for fast lookups) +_CF_NETWORKS = None + + +def _get_cf_networks(): + """Lazily parse Cloudflare CIDR ranges into network objects.""" + global _CF_NETWORKS + if _CF_NETWORKS is None: + _CF_NETWORKS = [] + for cidr in CLOUDFLARE_IPV4_RANGES: + try: + _CF_NETWORKS.append(ipaddress.IPv4Network(cidr, strict=False)) + except (ipaddress.AddressValueError, ValueError): + pass + return _CF_NETWORKS + + +def is_cloudflare_ip(ip: str) -> bool: + """Check whether an IP belongs to a known Cloudflare range. + + This is the primary detection method -- it doesn't require any + external databases or network requests. The IP ranges are from + Cloudflare's official published list. + """ + try: + addr = ipaddress.IPv4Address(ip) + except (ipaddress.AddressValueError, ValueError): + return False + return any(addr in net for net in _get_cf_networks()) + + +@dataclass +class DomainResult: + """Result of checking a single domain.""" + + domain: str + ip: str = "" + is_cloudflare: bool = False + tcp_ok: bool = False + tls_ok: bool = False + http_ok: bool = False + http_status: int = 0 + tls_ms: float = 0.0 + error: str = "" + + @property + def usable_as_sni(self) -> bool: + """Domain is usable as a fake SNI for Cloudflare IP spoofing. + + Must be: + 1. Resolved to a Cloudflare IP (so TLS handshake works through CF) + 2. TCP port 443 reachable + 3. TLS handshake succeeds + """ + return self.is_cloudflare and self.tcp_ok and self.tls_ok + + def summary(self) -> str: + parts = [self.domain] + if self.ip: + parts.append(self.ip) + if self.is_cloudflare: + parts.append("CF") + parts.append("TCP:OK" if self.tcp_ok else "TCP:FAIL") + parts.append("TLS:OK" if self.tls_ok else "TLS:FAIL") + if self.http_ok: + parts.append(f"HTTP:{self.http_status}") + if self.error: + parts.append(f"ERR:{self.error}") + return " | ".join(parts) + + +class DomainChecker: + """Bulk domain checker for Cloudflare CDN detection. + + Resolves domains, checks if they're behind Cloudflare, and + verifies TLS connectivity. Results can be filtered to produce + a verified list of domains suitable for SNI spoofing. + """ + + def __init__( + self, + concurrency: int = 50, + timeout: float = 3.0, + verify_tls: bool = True, + verify_http: bool = False, + ): + """ + Args: + concurrency: Maximum parallel checks. + timeout: Per-check timeout in seconds. + verify_tls: Also perform TLS handshake (not just DNS+IP check). + verify_http: Also perform HTTP request for deeper validation. + """ + self.concurrency = concurrency + self.timeout = timeout + self.verify_tls = verify_tls + self.verify_http = verify_http + + def check_domains( + self, + domains: List[str], + progress_cb: Optional[Callable] = None, + ) -> List[DomainResult]: + """Check a list of domains in parallel. + + Returns results sorted with Cloudflare-backed domains first, + then by TLS latency. + """ + results: List[DomainResult] = [] + done_count = 0 + total = len(domains) + + logger.info( + "Checking %d domains (workers=%d, timeout=%.1fs, tls=%s, http=%s)", + total, self.concurrency, self.timeout, + self.verify_tls, self.verify_http, + ) + + t_start = time.monotonic() + + with concurrent.futures.ThreadPoolExecutor( + max_workers=self.concurrency + ) as executor: + futures = { + executor.submit(self._check_one, domain): domain + for domain in domains + } + for future in concurrent.futures.as_completed(futures): + done_count += 1 + try: + result = future.result() + results.append(result) + except Exception as exc: + domain = futures[future] + results.append(DomainResult( + domain=domain, error=str(exc) + )) + + if progress_cb: + try: + progress_cb(done_count, total) + except Exception: + pass + + elapsed = time.monotonic() - t_start + cf_count = sum(1 for r in results if r.is_cloudflare) + usable_count = sum(1 for r in results if r.usable_as_sni) + + logger.info( + "Domain check complete: %d/%d Cloudflare, %d usable (%.1fs)", + cf_count, total, usable_count, elapsed, + ) + + # Sort: Cloudflare + usable first, then by TLS latency + results.sort( + key=lambda r: ( + not r.usable_as_sni, + not r.is_cloudflare, + r.tls_ms if r.tls_ms > 0 else 9999, + ) + ) + + return results + + def _check_one(self, domain: str) -> DomainResult: + """Check a single domain.""" + result = DomainResult(domain=domain) + + # Step 1: DNS resolution + try: + ip = socket.gethostbyname(domain) + result.ip = ip + except (socket.gaierror, socket.herror, OSError): + result.error = "dns_fail" + return result + + # Step 2: Check if IP is in Cloudflare ranges + result.is_cloudflare = is_cloudflare_ip(ip) + + # Step 3: TCP connect test + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(self.timeout) + sock.connect((ip, 443)) + result.tcp_ok = True + except (socket.timeout, TimeoutError): + result.error = "tcp_timeout" + try: + sock.close() + except Exception: + pass + return result + except (ConnectionRefusedError, OSError) as exc: + result.error = f"tcp_{getattr(exc, 'errno', 'error')}" + try: + sock.close() + except Exception: + pass + return result + + # Step 4: TLS handshake (optional but recommended) + if self.verify_tls: + try: + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + ctx.set_alpn_protocols(["h2", "http/1.1"]) + ctx.minimum_version = ssl.TLSVersion.TLSv1_2 + + t0 = time.monotonic() + ssl_sock = ctx.wrap_socket(sock, server_hostname=domain) + result.tls_ms = (time.monotonic() - t0) * 1000 + result.tls_ok = True + + # Step 5: HTTP check (optional) + if self.verify_http and result.tls_ok: + self._http_check(ssl_sock, domain, result) + + try: + ssl_sock.close() + except Exception: + pass + except ssl.SSLError as exc: + result.error = f"tls_{exc.reason}" + try: + sock.close() + except Exception: + pass + except (socket.timeout, TimeoutError): + result.error = "tls_timeout" + try: + sock.close() + except Exception: + pass + except OSError: + result.error = "tls_error" + try: + sock.close() + except Exception: + pass + else: + try: + sock.close() + except Exception: + pass + + return result + + def _http_check( + self, ssl_sock: ssl.SSLSocket, domain: str, result: DomainResult + ): + """Send a lightweight HTTP request and check the status code.""" + try: + req = ( + f"GET / HTTP/1.1\r\n" + f"Host: {domain}\r\n" + f"User-Agent: Mozilla/5.0\r\n" + f"Accept: */*\r\n" + f"Connection: close\r\n\r\n" + ).encode() + ssl_sock.settimeout(self.timeout) + ssl_sock.sendall(req) + + response = b"" + while len(response) < 4096: + try: + chunk = ssl_sock.recv(4096) + if not chunk: + break + response += chunk + if b"\r\n\r\n" in response: + break + except (socket.timeout, TimeoutError): + break + + if response: + first_line = response.decode("utf-8", errors="replace").split("\r\n", 1)[0] + # Parse HTTP status code + parts = first_line.split(" ", 2) + if len(parts) >= 2: + try: + result.http_status = int(parts[1]) + if 200 <= result.http_status < 400: + result.http_ok = True + except ValueError: + pass + except Exception: + pass + + # ── Utility methods ────────────────────────────────────────────── + + @staticmethod + def load_domains_from_file(filepath: str) -> List[str]: + """Load domain list from a text file (one per line). + + Supports comments (#) and empty lines. + """ + domains = [] + with open(filepath, "r") as f: + for line in f: + line = line.strip() + # Skip empty lines and comments + if not line or line.startswith("#"): + continue + # Strip protocol prefixes if present + if line.startswith("http://"): + line = line[7:] + if line.startswith("https://"): + line = line[8:] + # Strip paths + line = line.split("/")[0] + # Strip port numbers + line = line.split(":")[0] + if line: + domains.append(line) + return domains + + @staticmethod + def results_table(results: List[DomainResult], cloudflare_only: bool = False) -> str: + """Format results as a human-readable table.""" + lines = [ + f"{'#':>4} {'Domain':<40} {'IP':<16} {'CDN':>4} " + f"{'TCP':>4} {'TLS':>4} {'TLS ms':>7} {'Status':<6}" + ] + lines.append("-" * 90) + + filtered = results + if cloudflare_only: + filtered = [r for r in results if r.is_cloudflare] + + for i, r in enumerate(filtered, 1): + cdn = "CF" if r.is_cloudflare else "-" + tcp = "OK" if r.tcp_ok else "-" + tls = "OK" if r.tls_ok else "-" + tls_ms = f"{r.tls_ms:.0f}ms" if r.tls_ms > 0 else "-" + status = "SNI" if r.usable_as_sni else ("CF" if r.is_cloudflare else "skip") + lines.append( + f"{i:>4} {r.domain:<40} {r.ip:<16} {cdn:>4} " + f"{tcp:>4} {tls:>4} {tls_ms:>7} {status:<6}" + ) + + return "\n".join(lines) + + @staticmethod + def export_sni_list( + results: List[DomainResult], + filepath: str, + usable_only: bool = True, + ) -> int: + """Export verified domains to a text file. + + Returns the number of domains written. + """ + domains = [] + for r in results: + if usable_only and not r.usable_as_sni: + continue + elif not usable_only and not r.is_cloudflare: + continue + domains.append(r.domain) + + with open(filepath, "w") as f: + f.write("# Verified Cloudflare-backed SNI domains\n") + f.write(f"# Generated by SNISPF domain checker\n") + f.write(f"# Total: {len(domains)} domains\n\n") + for d in domains: + f.write(d + "\n") + + return len(domains) diff --git a/sni_spoofing/tls/__init__.py b/sni_spoofing/tls/__init__.py new file mode 100644 index 0000000..49d3ed8 --- /dev/null +++ b/sni_spoofing/tls/__init__.py @@ -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 diff --git a/sni_spoofing/tls/fragment.py b/sni_spoofing/tls/fragment.py new file mode 100644 index 0000000..d0855cb --- /dev/null +++ b/sni_spoofing/tls/fragment.py @@ -0,0 +1,161 @@ +"""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 + """ + if not sizes or not data: + return [data] if data else [] + + 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:]) + pos = len(data) + else: + fragments.append(data[pos : pos + size]) + pos += size + + # If we consumed all specified sizes but data remains + if pos < len(data): + fragments.append(data[pos:]) + + return fragments if fragments else [data] diff --git a/sni_spoofing/utils/__init__.py b/sni_spoofing/utils/__init__.py new file mode 100644 index 0000000..1977a49 --- /dev/null +++ b/sni_spoofing/utils/__init__.py @@ -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 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_tls.py b/tests/test_tls.py new file mode 100644 index 0000000..78633fb --- /dev/null +++ b/tests/test_tls.py @@ -0,0 +1,448 @@ +"""Unit tests for TLS ClientHello builder and parser.""" + +import os +import sys +import struct +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from sni_spoofing.tls import ClientHelloBuilder +from sni_spoofing.tls.fragment import ( + fragment_client_hello, + fragment_data, + _find_sni_offset, +) + + +class TestClientHelloBuilder(unittest.TestCase): + """Test TLS ClientHello construction.""" + + def test_build_client_hello_basic(self): + """Test basic ClientHello construction.""" + hello = ClientHelloBuilder.build_client_hello(sni="example.com") + + # Should start with TLS record header + self.assertEqual(hello[0], 0x16) # Handshake + self.assertEqual(hello[1], 0x03) # TLS major version + self.assertEqual(hello[2], 0x01) # TLS 1.0 (legacy) + + # Record length should match + record_len = struct.unpack("!H", hello[3:5])[0] + self.assertEqual(record_len, len(hello) - 5) + + # Handshake type should be ClientHello + self.assertEqual(hello[5], 0x01) + + def test_build_client_hello_target_size(self): + """Test that ClientHello hits 517 bytes (matching Go template).""" + hello = ClientHelloBuilder.build_client_hello(sni="mci.ir") + self.assertEqual(len(hello), 517) + + def test_build_client_hello_contains_sni(self): + """Test that built ClientHello contains the specified SNI.""" + sni = "auth.vercel.com" + hello = ClientHelloBuilder.build_client_hello(sni=sni) + + # The SNI should be present in the packet + self.assertIn(sni.encode("ascii"), hello) + + def test_build_client_hello_different_snis(self): + """Test building with different SNI values.""" + for sni in ["google.com", "cloudflare.com", "example.org", "test.co"]: + hello = ClientHelloBuilder.build_client_hello(sni=sni) + self.assertIn(sni.encode("ascii"), hello) + self.assertEqual(hello[0], 0x16) + + def test_build_client_hello_custom_session_id(self): + """Test with custom session ID.""" + session_id = os.urandom(32) + hello = ClientHelloBuilder.build_client_hello( + sni="test.com", session_id=session_id + ) + self.assertIn(session_id, hello) + + def test_build_client_hello_custom_random(self): + """Test with custom random bytes.""" + random_bytes = os.urandom(32) + hello = ClientHelloBuilder.build_client_hello( + sni="test.com", random_bytes=random_bytes + ) + self.assertIn(random_bytes, hello) + + def test_parse_client_hello_roundtrip(self): + """Test build and parse roundtrip.""" + sni = "auth.vercel.com" + hello = ClientHelloBuilder.build_client_hello(sni=sni) + parsed = ClientHelloBuilder.parse_client_hello(hello) + + self.assertEqual(parsed.get("handshake_type"), "ClientHello") + self.assertEqual(parsed.get("sni"), sni) + self.assertEqual(parsed.get("content_type"), 0x16) + + def test_parse_client_hello_multiple(self): + """Test parsing multiple different ClientHellos.""" + for sni in ["test.com", "example.org", "cloudflare.com"]: + hello = ClientHelloBuilder.build_client_hello(sni=sni) + parsed = ClientHelloBuilder.parse_client_hello(hello) + self.assertEqual(parsed.get("sni"), sni) + + def test_build_sni_extension(self): + """Test SNI extension construction.""" + ext = ClientHelloBuilder.build_sni_extension("test.com") + + # Extension type should be 0x0000 (SNI) + ext_type = struct.unpack("!H", ext[0:2])[0] + self.assertEqual(ext_type, 0x0000) + + # Should contain the hostname + self.assertIn(b"test.com", ext) + + def test_build_key_share_extension(self): + """Test key share extension construction.""" + key = os.urandom(32) + ext = ClientHelloBuilder.build_key_share_extension(key) + + # Extension type should be 0x0033 (key_share) + ext_type = struct.unpack("!H", ext[0:2])[0] + self.assertEqual(ext_type, 0x0033) + + # Should contain the key + self.assertIn(key, ext) + + def test_build_client_response(self): + """Test client response (CCS + AppData) construction.""" + resp = ClientHelloBuilder.build_client_response() + + # Should start with Change Cipher Spec + self.assertEqual(resp[0], 0x14) # CCS content type + self.assertEqual(resp[1], 0x03) + self.assertEqual(resp[2], 0x03) + + def test_parse_empty_data(self): + """Test parsing empty or too-short data.""" + self.assertEqual(ClientHelloBuilder.parse_client_hello(b""), {}) + self.assertEqual(ClientHelloBuilder.parse_client_hello(b"\x00"), {}) + + def test_parse_non_handshake(self): + """Test parsing non-handshake data.""" + result = ClientHelloBuilder.parse_client_hello(b"\x17\x03\x03\x00\x05hello") + self.assertEqual(result.get("content_type"), 0x17) + self.assertNotIn("handshake_type", result) + + +class TestFragmentation(unittest.TestCase): + """Test TLS record fragmentation.""" + + def test_sni_split_fragments(self): + """Test SNI-split fragmentation produces exactly 2 fragments.""" + hello = ClientHelloBuilder.build_client_hello(sni="test.example.com") + fragments = fragment_client_hello(hello, "sni_split") + + self.assertEqual(len(fragments), 2) + # Reassembled should equal original + self.assertEqual(b"".join(fragments), hello) + + def test_half_split(self): + """Test half-split fragmentation.""" + hello = ClientHelloBuilder.build_client_hello(sni="test.com") + fragments = fragment_client_hello(hello, "half") + + self.assertEqual(len(fragments), 2) + self.assertEqual(b"".join(fragments), hello) + + def test_multi_split(self): + """Test multi-fragment split.""" + hello = ClientHelloBuilder.build_client_hello(sni="test.com") + fragments = fragment_client_hello(hello, "multi") + + self.assertGreater(len(fragments), 2) + self.assertEqual(b"".join(fragments), hello) + + def test_tls_record_fragment(self): + """Test TLS record-level fragmentation.""" + hello = ClientHelloBuilder.build_client_hello(sni="test.com") + fragments = fragment_client_hello(hello, "tls_record_frag") + + self.assertEqual(len(fragments), 2) + # Each fragment should be a valid TLS record + for frag in fragments: + self.assertEqual(frag[0], 0x16) # Handshake type + + def test_no_fragmentation(self): + """Test 'none' strategy returns single fragment.""" + hello = ClientHelloBuilder.build_client_hello(sni="test.com") + fragments = fragment_client_hello(hello, "none") + + self.assertEqual(len(fragments), 1) + self.assertEqual(fragments[0], hello) + + def test_find_sni_offset(self): + """Test SNI offset detection.""" + hello = ClientHelloBuilder.build_client_hello(sni="example.com") + offset, length = _find_sni_offset(hello) + + self.assertGreater(offset, 0) + self.assertEqual(length, len("example.com")) + # Verify the SNI at that offset + self.assertEqual(hello[offset:offset + length], b"example.com") + + def test_fragment_data_custom_sizes(self): + """Test custom size fragmentation.""" + data = b"A" * 100 + fragments = fragment_data(data, [10, 20, 30]) + + self.assertEqual(len(fragments[0]), 10) + self.assertEqual(len(fragments[1]), 20) + self.assertEqual(b"".join(fragments), data) + + def test_fragment_preserves_data(self): + """Test that fragmentation preserves all data.""" + for strategy in ["sni_split", "half", "multi", "tls_record_frag", "none"]: + hello = ClientHelloBuilder.build_client_hello(sni="test.example.org") + fragments = fragment_client_hello(hello, strategy) + if strategy != "tls_record_frag": + # For TLS record frag, the output is re-wrapped + reassembled = b"".join(fragments) + self.assertEqual( + len(reassembled), + len(hello), + f"Strategy '{strategy}' changed data length", + ) + + +class TestRawInjector(unittest.TestCase): + """Test raw injector frame construction.""" + + def test_build_fake_frame_checksum(self): + """Test that _build_fake_frame produces valid IP and TCP checksums.""" + try: + from sni_spoofing.bypass.raw_injector import ( + _build_fake_frame, + _ip_checksum, + _ip_hdr_len, + _tcp_checksum, + ) + except ImportError: + self.skipTest("raw_injector not importable") + + # Build a minimal Ethernet+IP+TCP template (14+20+20 = 54 bytes) + # Ethernet: dst(6) + src(6) + type(2) + eth = bytes(6) + bytes(6) + b"\x08\x00" + # IP header: version/ihl(1)+tos(1)+totlen(2)+id(2)+flags/frag(2)+ttl(1)+proto(1)+cksum(2)+src(4)+dst(4) + iph = bytearray(20) + iph[0] = 0x45 # IPv4, IHL=5 + iph[8] = 64 # TTL + iph[9] = 6 # TCP + iph[12:16] = b"\xc0\xa8\x01\x02" # src 192.168.1.2 + iph[16:20] = b"\x68\x12\x04\x82" # dst 104.18.4.130 + struct.pack_into("!H", iph, 2, 40) # total length + # TCP header: srcport(2)+dstport(2)+seq(4)+ack(4)+offset/flags(2)+window(2)+cksum(2)+urgent(2) + tcph = bytearray(20) + struct.pack_into("!H", tcph, 0, 54321) # src port + struct.pack_into("!H", tcph, 2, 443) # dst port + struct.pack_into("!I", tcph, 4, 1000) # seq + struct.pack_into("!I", tcph, 8, 2000) # ack + tcph[12] = 0x50 # data offset = 5 words + tcph[13] = 0x10 # ACK flag + + template = bytes(eth) + bytes(iph) + bytes(tcph) + + # Build the fake frame + fake_payload = ClientHelloBuilder.build_client_hello(sni="test.com") + frame = _build_fake_frame(template, 999, fake_payload) + + # Check that the frame is longer than the template + self.assertGreater(len(frame), len(template)) + + # Verify the seq number: ISN + 1 - len(fake) + tcp_off = 14 + 20 + seq = struct.unpack("!I", frame[tcp_off + 4:tcp_off + 8])[0] + expected_seq = (1000 - len(fake_payload)) & 0xFFFFFFFF + self.assertEqual(seq, expected_seq) + + # Check PSH flag is set + self.assertTrue(frame[tcp_off + 13] & 0x08) + + def test_is_raw_available(self): + """Test raw availability detection doesn't crash.""" + from sni_spoofing.bypass.raw_injector import is_raw_available + result = is_raw_available() + self.assertIsInstance(result, bool) + + +class TestDomainChecker(unittest.TestCase): + """Test the bulk Cloudflare-domain checker.""" + + def test_is_cloudflare_ip_positive(self): + """Known Cloudflare IPs should be detected.""" + from sni_spoofing.scanner import is_cloudflare_ip + # 104.16.0.0/13 belongs to Cloudflare + self.assertTrue(is_cloudflare_ip("104.16.1.1")) + self.assertTrue(is_cloudflare_ip("172.64.0.1")) + + def test_is_cloudflare_ip_negative(self): + """Non-Cloudflare IPs should be rejected.""" + from sni_spoofing.scanner import is_cloudflare_ip + self.assertFalse(is_cloudflare_ip("8.8.8.8")) + self.assertFalse(is_cloudflare_ip("1.1.1.1")) # Cloudflare DNS, not CDN + self.assertFalse(is_cloudflare_ip("not-an-ip")) + self.assertFalse(is_cloudflare_ip("")) + + def test_domain_result_usable_as_sni(self): + """DomainResult.usable_as_sni requires CF + TCP + TLS.""" + from sni_spoofing.scanner import DomainResult + r = DomainResult(domain="x.com", is_cloudflare=True, tcp_ok=True, tls_ok=True) + self.assertTrue(r.usable_as_sni) + r2 = DomainResult(domain="x.com", is_cloudflare=False, tcp_ok=True, tls_ok=True) + self.assertFalse(r2.usable_as_sni) + + +class TestUtilities(unittest.TestCase): + """Test utility functions.""" + + def test_imports(self): + """Test that all modules import correctly.""" + from sni_spoofing.bypass import ( + BypassStrategy, + CombinedBypass, + FakeSNIBypass, + FragmentBypass, + RawInjector, + is_raw_available, + ) + from sni_spoofing.forwarder import handle_connection, start_server + from sni_spoofing.utils import ( + get_default_interface_ipv4, + check_platform_capabilities, + resolve_host, + is_valid_ip, + is_valid_port, + ) + + def test_is_valid_ip(self): + """Test IP validation.""" + from sni_spoofing.utils import is_valid_ip + + self.assertTrue(is_valid_ip("127.0.0.1")) + self.assertTrue(is_valid_ip("192.168.1.1")) + self.assertTrue(is_valid_ip("0.0.0.0")) + self.assertFalse(is_valid_ip("not-an-ip")) + self.assertFalse(is_valid_ip("")) + + def test_is_valid_port(self): + """Test port validation.""" + from sni_spoofing.utils import is_valid_port + + self.assertTrue(is_valid_port(80)) + self.assertTrue(is_valid_port(443)) + self.assertTrue(is_valid_port(40443)) + self.assertTrue(is_valid_port(65535)) + self.assertFalse(is_valid_port(0)) + self.assertFalse(is_valid_port(65536)) + self.assertFalse(is_valid_port(-1)) + + def test_platform_capabilities(self): + """Test platform capabilities detection.""" + from sni_spoofing.utils import check_platform_capabilities + + caps = check_platform_capabilities() + self.assertIn("platform", caps) + self.assertIn("fragment_support", caps) + self.assertIn("tls_record_frag", caps) + self.assertIn("af_packet", caps) + self.assertIn("raw_injection", caps) + self.assertTrue(caps["fragment_support"]) + self.assertTrue(caps["tls_record_frag"]) + self.assertTrue(caps["fake_sni"]) + + def test_strategy_construction(self): + """Test bypass strategy construction.""" + from sni_spoofing.bypass import FragmentBypass, FakeSNIBypass, CombinedBypass + + frag = FragmentBypass(strategy="sni_split") + self.assertEqual(frag.name, "fragment") + + fake = FakeSNIBypass(method="prefix_fake") + self.assertEqual(fake.name, "fake_sni") + + combo = CombinedBypass() + self.assertEqual(combo.name, "combined") + + def test_strategy_with_raw_injector(self): + """Test strategy construction with raw_injector parameter.""" + from sni_spoofing.bypass import FakeSNIBypass, CombinedBypass + + fake = FakeSNIBypass(raw_injector="mock") + self.assertEqual(fake.raw_injector, "mock") + + combo = CombinedBypass(raw_injector="mock") + self.assertEqual(combo.raw_injector, "mock") + + def test_fake_sni_ttl_trick_flag(self): + """Test FakeSNIBypass accepts use_ttl_trick parameter.""" + from sni_spoofing.bypass import FakeSNIBypass + + fake = FakeSNIBypass(use_ttl_trick=True) + self.assertTrue(fake.use_ttl_trick) + self.assertIsNone(fake.raw_injector) + + def test_fake_sni_ttl_trick_default(self): + """Test FakeSNIBypass use_ttl_trick defaults to False.""" + from sni_spoofing.bypass import FakeSNIBypass + + fake = FakeSNIBypass() + self.assertFalse(fake.use_ttl_trick) + + def test_combined_ttl_trick_flag(self): + """Test CombinedBypass accepts use_ttl_trick parameter.""" + from sni_spoofing.bypass import CombinedBypass + + combo = CombinedBypass(use_ttl_trick=True) + self.assertTrue(combo.use_ttl_trick) + + def test_build_strategy_fake_sni_with_ttl(self): + """Test build_strategy passes USE_TTL_TRICK to FakeSNIBypass.""" + from sni_spoofing.cli import build_strategy + + config = {"BYPASS_METHOD": "fake_sni", "FAKE_SNI_METHOD": "prefix_fake", + "USE_TTL_TRICK": True} + strategy = build_strategy(config) + self.assertTrue(strategy.use_ttl_trick) + + def test_build_strategy_combined_with_ttl(self): + """Test build_strategy passes USE_TTL_TRICK to CombinedBypass.""" + from sni_spoofing.cli import build_strategy + + config = {"BYPASS_METHOD": "combined", "FRAGMENT_STRATEGY": "sni_split", + "USE_TTL_TRICK": True, "FRAGMENT_DELAY": 0.1} + strategy = build_strategy(config) + self.assertTrue(strategy.use_ttl_trick) + + def test_parse_host_port_no_port(self): + """Test parse_host_port with just an IP (no port).""" + from sni_spoofing.cli import parse_host_port + + host, port = parse_host_port("104.19.229.21", "0.0.0.0", 443) + self.assertEqual(host, "104.19.229.21") + self.assertEqual(port, 443) + + def test_parse_host_port_with_port(self): + """Test parse_host_port with IP:PORT format.""" + from sni_spoofing.cli import parse_host_port + + host, port = parse_host_port("104.19.229.21:8443", "0.0.0.0", 443) + self.assertEqual(host, "104.19.229.21") + self.assertEqual(port, 8443) + + def test_parse_host_port_port_only(self): + """Test parse_host_port with :PORT format.""" + from sni_spoofing.cli import parse_host_port + + host, port = parse_host_port(":40443", "0.0.0.0", 443) + self.assertEqual(host, "0.0.0.0") + self.assertEqual(port, 40443) + + +if __name__ == "__main__": + unittest.main(verbosity=2)