aheh/MasterHttpRelayVPN
1
0
Fork 0
mirror of https://github.com/masterking32/MasterHttpRelayVPN.git synced 2026-05-17 21:24:37 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge upstream/python_testing

Browse Source
This commit is contained in:
PK3NZO
2026-04-23 15:06:20 +03:30
parent afdd3e1036 9e5c074a1d
commit 7723b91140
5 changed files with 148 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+23 -1
View File
@@ -16,6 +16,28 @@ For the latest news, releases, and project updates, follow our Telegram channel:
---
### If you like this project, please support it by starring it on GitHub (⭐). It helps the project get discovered.
---
### Optional Financial Support 💸
- TON network:
`masterking32.ton`
- EVM-compatible networks (ETH and compatible chains):
`0x517f07305D6ED781A089322B6cD93d1461bF8652`
- TRC20 network (TRON):
`TLApdY8APWkFHHoxebxGY8JhMeChiETqFH`
Every contribution and every piece of feedback is appreciated. Support directly helps ongoing development and improvement.
---
## Disclaimer
MasterHttpRelayVPN is provided for educational, testing, and research purposes only.
@@ -287,7 +309,7 @@ Install everything from [`requirements.txt`](requirements.txt). All listed packa
| `h2` | HTTP/2 multiplexing to the Apps Script relay (significantly faster) |
| `brotli` | Decompression of `Content-Encoding: br` responses |
| `zstandard` | Decompression of `Content-Encoding: zstd` responses |
| `netifaces` | Better network interface detection for LAN sharing (fallback available without it) |
### Load Balancing
README_FA.md
+22 -1
View File
@@ -14,6 +14,28 @@
---
### اگر از پروژه راضی‌اید، با دادن ستاره (⭐) در گیت‌هاب از ما حمایت کنید — این کار به دیده‌شدن پروژه کمک می‌کند.
---
### حمایت مالی (اختیاری) 💸
- شبکه TON:
`masterking32.ton`
- آدرس روی شبکه‌های EVM (ETH و سازگارها):
`0x517f07305D6ED781A089322B6cD93d1461bF8652`
- شبکه TRC20 (TRON):
`TLApdY8APWkFHHoxebxGY8JhMeChiETqFH`
از هر نوع حمایت و بازخورد شما سپاسگزاریم — کمک‌ها برای توسعه و بهبود پروژه بسیار ارزشمند است.
---
## سلب مسئولیت
پروژه MasterHttpRelayVPN فقط برای اهداف آموزشی، تست و پژوهش ارائه شده است.
@@ -235,7 +257,6 @@ json
| `h2` | ارتباط HTTP/2 با رله Apps Script (به‌طور محسوسی سریع‌تر) |
| `brotli` | پشتیبانی از فشرده‌سازی `Content-Encoding: br` |
| `zstandard` | پشتیبانی از فشرده‌سازی `Content-Encoding: zstd` |
| `netifaces` | تشخیص بهتر اینترفیس‌های شبکه برای اشتراک‌گذاری LAN (در صورت نبود آن، حالت جایگزین در دسترس است) |
### استفاده از چند Script ID
requirements.txt
+2 -2
View File
@@ -13,5 +13,5 @@ brotli>=1.1.0
# Optional: Zstandard decompression (some CDNs now serve `zstd`)
zstandard>=0.22.0
# Optional: Better network interface detection for LAN sharing
netifaces>=0.11.0
# LAN interface detection now uses only the Python standard library
# (works on Windows, Linux, macOS, Android/Termux without a C compiler).
src/lan_utils.py
+82 -79
View File
@@ -1,101 +1,103 @@
"""
LAN utilities for detecting network interfaces and IP addresses.
LAN utilities for detecting network interfaces and IPv4 addresses.
Provides functionality to enumerate local network interfaces and their
associated IP addresses for LAN proxy sharing.
Provides functionality to enumerate local IPv4 addresses for LAN proxy
sharing. IPv6 is intentionally not reported — this project only exposes
the proxy over IPv4 LANs, which is what every consumer router and
phone/desktop client actually uses.
Implementation notes
--------------------
This module relies only on the Python standard library so it works
out-of-the-box on every supported OS (Windows, Linux, macOS,
Android/Termux, *BSD) without requiring a C compiler or native build
tools (previous versions depended on ``netifaces``, which needs
"Microsoft Visual C++ 14.0 or greater" on Windows and was a frequent
install blocker for users on slow connections).
Strategy (in order):
1. "UDP connect trick" to reliably discover the primary outbound
IPv4 address on any OS.
2. ``socket.getaddrinfo(hostname, AF_INET)`` to enumerate any additional
IPv4 addresses bound to the host (covers multi-homed machines).
"""
import ipaddress
import logging
import socket
from typing import Dict, List, Optional
from typing import Dict, List, Optional, Set
log = logging.getLogger("LAN")
# ---------------------------------------------------------------------------
# Primary-IP discovery (UDP connect trick)
# ---------------------------------------------------------------------------
def _primary_ipv4() -> Optional[str]:
"""
Return the primary local IPv4 the OS would use for outbound traffic.
Uses a connected UDP socket which does *not* actually send packets —
the kernel just picks the source address from its routing table.
Works identically on Windows, Linux, macOS, and Android.
"""
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.settimeout(0.5)
# TEST-NET-1 address, port is arbitrary; no packet is sent for UDP connect().
s.connect(('192.0.2.1', 80))
return s.getsockname()[0]
except OSError:
return None
finally:
s.close()
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def get_network_interfaces() -> Dict[str, List[str]]:
"""
Get all network interfaces and their associated IP addresses.
Returns a dictionary mapping interface names to lists of IP addresses
(both IPv4 and IPv6). Only includes interfaces with valid IP addresses
that are not loopback.
Get network interfaces and their associated non-loopback IPv4 addresses.
Returns:
Dict[str, List[str]]: Interface name -> list of IP addresses
Dict[str, List[str]]: Interface label -> list of IPv4 addresses.
Labels are best-effort synthetic names such as ``"primary"``
and ``"host"``.
"""
interfaces = {}
interfaces: Dict[str, List[str]] = {}
seen_ips: Set[str] = set()
def _add(label: str, ip: Optional[str]) -> None:
if not ip or ip in seen_ips:
return
if ip.startswith('127.'):
return
seen_ips.add(ip)
interfaces.setdefault(label, []).append(ip)
# 1) Primary outbound IPv4 (most reliable, cross-platform).
_add('primary', _primary_ipv4())
# 2) Enumerate via hostname resolution (picks up multi-homed hosts).
try:
import netifaces
for iface in netifaces.interfaces():
addrs = netifaces.ifaddresses(iface)
ips = []
# IPv4 addresses
if netifaces.AF_INET in addrs:
for addr in addrs[netifaces.AF_INET]:
ip = addr.get('addr')
if ip and not ip.startswith('127.'):
ips.append(ip)
# IPv6 addresses (without scope)
if netifaces.AF_INET6 in addrs:
for addr in addrs[netifaces.AF_INET6]:
ip = addr.get('addr')
if ip and not ip.startswith('::1') and not '%' in ip:
# Remove scope if present
ips.append(ip.split('%')[0])
if ips:
interfaces[iface] = ips
except ImportError:
# Fallback to socket method for basic detection
log.debug("netifaces not available, using socket fallback")
interfaces = _get_interfaces_socket_fallback()
return interfaces
def _get_interfaces_socket_fallback() -> Dict[str, List[str]]:
"""
Fallback method to get network interfaces using socket.
This is less comprehensive than netifaces but works without extra dependencies.
"""
interfaces = {}
try:
# Get hostname and try to resolve to IPs
hostname = socket.gethostname()
try:
# Get IPv4 addresses
ipv4_info = socket.getaddrinfo(hostname, None, socket.AF_INET)
ipv4_addrs = [info[4][0] for info in ipv4_info if not info[4][0].startswith('127.')]
if ipv4_addrs:
interfaces['primary'] = list(set(ipv4_addrs)) # Remove duplicates
except socket.gaierror:
pass
except OSError:
hostname = ''
if hostname:
try:
# Get IPv6 addresses
ipv6_info = socket.getaddrinfo(hostname, None, socket.AF_INET6)
ipv6_addrs = []
for info in ipv6_info:
ip = info[4][0]
if not ip.startswith('::1') and not '%' in ip:
ipv6_addrs.append(ip.split('%')[0])
if ipv6_addrs:
interfaces['primary_ipv6'] = list(set(ipv6_addrs))
except socket.gaierror:
for info in socket.getaddrinfo(hostname, None, socket.AF_INET):
_add('host', info[4][0])
except (socket.gaierror, OSError):
pass
except Exception as e:
log.debug("Socket fallback failed: %s", e)
return interfaces
def get_lan_ips(port: int = 8085) -> List[str]:
"""
Get list of LAN-accessible proxy addresses.
Get list of LAN-accessible proxy addresses (IPv4 only).
Returns a list of IP:port combinations that can be used to access
the proxy from other devices on the local network.
@@ -107,21 +109,22 @@ def get_lan_ips(port: int = 8085) -> List[str]:
List[str]: List of "IP:port" strings for LAN access
"""
interfaces = get_network_interfaces()
lan_addresses = []
lan_addresses: List[str] = []
for iface_ips in interfaces.values():
for ip in iface_ips:
try:
# Validate IP and check if it's a private address
addr = ipaddress.ip_address(ip)
if addr.is_private or addr.is_link_local:
lan_addresses.append(f"{ip}:{port}")
except ValueError:
addr = ipaddress.IPv4Address(ip)
except (ValueError, ipaddress.AddressValueError):
continue
if addr.is_loopback or addr.is_unspecified:
continue
if addr.is_private or addr.is_link_local:
lan_addresses.append(f"{ip}:{port}")
# Remove duplicates while preserving order
seen = set()
unique_addresses = []
# Remove duplicates while preserving order.
seen: Set[str] = set()
unique_addresses: List[str] = []
for addr in lan_addresses:
if addr not in seen:
seen.add(addr)
src/proxy_server.py
+19 -12
View File
@@ -661,12 +661,17 @@ class ProxyServer:
# • port 443 → MITM + relay through Apps Script
# • port 80 → plain-HTTP relay through Apps Script
# • other → give up (non-HTTP; can't be relayed)
# We use a shorter connect timeout for IP literals (4 s) because
# when the route is DPI-dropped, waiting longer doesn't help and
# clients like Telegram speed up DC-rotation when we fail fast.
# We remember per-IP failures for a short while so subsequent
# connects skip the doomed direct attempt.
if _is_ip_literal(host):
if not self._direct_temporarily_disabled(host):
log.info("Direct tunnel → %s:%d (IP literal)", host, port)
ok = await self._do_direct_tunnel(host, port, reader, writer)
ok = await self._do_direct_tunnel(
host, port, reader, writer, timeout=4.0,
)
if ok:
return
self._remember_direct_failure(host, ttl=300)
@@ -900,7 +905,8 @@ class ProxyServer:
async def _do_direct_tunnel(self, host: str, port: int,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
connect_ip: str | None = None):
connect_ip: str | None = None,
timeout: float | None = None):
"""Pipe raw TLS bytes directly to the target server.
connect_ip overrides DNS: the TCP connection goes to that IP
@@ -910,9 +916,12 @@ class ProxyServer:
normal edge instead of being forced onto the fronting IP.
"""
target_ip = connect_ip or host
effective_timeout = (
self._tcp_connect_timeout if timeout is None else float(timeout)
)
try:
r_remote, w_remote = await self._open_tcp_connection(
target_ip, port, timeout=self._tcp_connect_timeout,
target_ip, port, timeout=effective_timeout,
)
except Exception as e:
log.error("Direct tunnel connect failed (%s via %s): %s",
@@ -1044,17 +1053,15 @@ class ProxyServer:
# • Telegram Desktop / MTProto over port 443 sends obfuscated
# non-TLS bytes — we literally cannot decrypt these, and
# since the target IP is blocked we can't direct-tunnel
# either. The only workaround is to configure Telegram as
# an HTTP proxy (not SOCKS5), so it sends hostnames our
# SNI-rewrite path can handle.
# either. Telegram will rotate to another DC on its own;
# failing fast here lets that happen sooner.
# • Client CONNECTs but never speaks TLS (some probes).
if _is_ip_literal(host) and port == 443:
log.warning(
"MITM TLS handshake failed for %s:%d (%s). "
"Likely non-TLS traffic (e.g. Telegram MTProto over "
"SOCKS5). Cannot relay raw TCP to a blocked IP — "
"use the HTTP proxy instead so hostnames are preserved.",
host, port, e,
log.info(
"Non-TLS traffic on %s:%d (likely Telegram MTProto / "
"obfuscated protocol). This DC appears blocked; the "
"client should rotate to another endpoint shortly.",
host, port,
)
elif port != 443:
log.debug(