Телеграм канал «𝆤͝𝄽͡𝔯𝔬𝔬𝔱𝔣𝔞𝔠𝔢»

Thanks from 𝘍𝘶𝘸𝘶 ʚ₍ᐢ. .ᐢ₎ɞ for share.

vmess://ewogICJ2IjogIjIiLAogICJwcyI6ICIiLAogICJhZGQiOiAiMTQxLjExLjE0OS45NyIsCiAgInBvcnQiOiA2NjYsCiAgImlkIjogInJvb3RmYWNlIiwKICAiYWlkIjogMCwKICAibmV0IjogIndzIiwKICAidHlwZSI6ICJub25lIiwKICAiaG9zdCI6ICIiLAogICJwYXRoIjogIi90Z0Byb290ZmFjZSIsCiAgInRscyI6ICJub25lIgp9

vmess://ewogICJ2IjogIjIiLAogICJwcyI6ICIiLAogICJhZGQiOiAiMjE2LjE2Ny43LjE5MSIsCiAgInBvcnQiOiA2NjYsCiAgImlkIjogInJvb3RmYWNlIiwKICAiYWlkIjogMCwKICAibmV0IjogInRjcCIsCiAgInR5cGUiOiAiaHR0cCIsCiAgImhvc3QiOiAiIiwKICAicGF0aCI6ICIvdGdAcm9vdGZhY2UiLAogICJ0bHMiOiAibm9uZSIKfQ==

درحال بررسی ویژه بر روی پروتکل های اینترنتی ام...

def main() -> int:
p = argparse.ArgumentParser(
description="Zero-dependency DNS resolver scanner (war conditions)")
p.add_argument("-i", "--input", required=True,
help="Input file with IPs or CIDRs")
p.add_argument("-o", "--output", default="good-resolvers.txt",
help="Output file (only good IPs written)")
p.add_argument("-d", "--domain", default="example.com",
help="Baseline test domain")
p.add_argument("-c", "--concurrency", type=int, default=50)
p.add_argument("-t", "--timeout", type=float, default=3.0)
p.add_argument("--burst", type=int, default=20,
help="Number of queries for QPS measurement (0 to skip)")
p.add_argument("--no-burst", action="store_true")
p.add_argument("--tcp", action="store_true",
help="Use TCP/53 instead of UDP/53")
p.add_argument("--limit", type=int,
help="Only scan first N IPs after shuffle")

g = p.add_argument_group("Tunnel domain probe (soft E2E)")
g.add_argument("--tunnel-domain",
help="Your tunnel zone, e.g. t.example.com")
g.add_argument("--tunnel-qtype", type=int, default=1,
help="1=A, 16=TXT, 28=AAAA (default 1)")
g.add_argument("--strict", action="store_true",
help="Only keep IPs that pass the tunnel probe")

args = p.parse_args()

if not os.path.isfile(args.input):
print(f"Input file not found: {args.input}", file=sys.stderr)
return 2

ips = read_ips(args.input)
random.shuffle(ips)
if args.limit:
ips = ips[:args.limit]
if not ips:
print("No valid IPs in input", file=sys.stderr)
return 2

print(f"Scanning {len(ips)} IPs | conc={args.concurrency} "
f"timeout={args.timeout}s burst={args.burst} "
f"tunnel={'yes' if args.tunnel_domain else 'no'} "
f"proto={'tcp' if args.tcp else 'udp'}")
print("-" * 78)

q: queue.Queue = queue.Queue()
for ip in ips:
q.put(ip)
results: list[dict] = []
lock = threading.Lock()

def worker():
while True:
try:
ip = q.get_nowait()
except queue.Empty:
return
try:
r = scan_ip(ip, args)
except Exception as e: # never crash a worker
r = {"ip": ip, "ok": False, "latency_ms": -1.0,
"qps": -1.0, "hijack": False, "tunnel_ok": False,
"error": type(e).name, "score": 0.0}
with lock:
results.append(r)
print(format_line(r, args))
sys.stdout.flush()

threads = [threading.Thread(target=worker, daemon=True)
for _ in range(max(1, args.concurrency))]
try:
for t in threads:
t.start()
for t in threads:
t.join()
except KeyboardInterrupt:
print("\nInterrupted, writing partial results...",
file=sys.stderr)

results.sort(key=lambda r: r["score"], reverse=True)

kept = []
with open(args.output, "w", encoding="utf-8") as f:
for r in results:
if not (r["ok"] and not r["hijack"]):
continue
if args.strict and args.tunnel_domain and not r["tunnel_ok"]:
continue
f.write(r["ip"] + "\n")
kept.append(r)

dns_ok = sum(1 for r in results if r["ok"] and not r["hijack"])
hijack = sum(1 for r in results if r["hijack"])
tunnel_ok = sum(1 for r in results if r["tunnel_ok"])

print("-" * 78)
print(f"Scanned : {len(results)}")
print(f"DNS OK : {dns_ok}")
print(f"Hijacked : {hijack}")
if args.tunnel_domain:
print(f"Tunnel probe ✓: {tunnel_ok}")
print(f"Written to : {args.output} ({len(kept)} IPs)")
return 0


if name == "main":
sys.exit(main())`

s.settimeout(timeout)
except OSError:
return 0.0
try:
for _ in range(count):
try:
qid, pkt = build_query(qname)
s.sendto(pkt, (ip, 53))
data, _ = s.recvfrom(4096)
if len(data) >= 2 and struct.unpack("!H", data[:2])[0] == qid:
ok += 1
except (OSError, ValueError):
pass
finally:
s.close()
dt = time.monotonic() - t0
return ok / dt if dt > 0 else 0.0


def is_private(s: str) -> bool:
try:
a = ipaddress.ip_address(s)
return (a.is_private or a.is_loopback or a.is_link_local
or a.is_reserved or a.is_unspecified or a.is_multicast)
except ValueError:
return True


# --------------------------------------------------------------------------- #
# Per-IP scanner
# --------------------------------------------------------------------------- #

def scan_ip(ip: str, args) -> dict:
r = {
"ip": ip, "ok": False, "latency_ms": -1.0, "qps": -1.0,
"hijack": False, "tunnel_ok": False, "error": "", "score": 0.0,
}

# 1) Baseline
base = query(ip, args.domain, timeout=args.timeout, use_tcp=args.tcp)
if base is None:
r["error"] = "no-response"
return r
r["latency_ms"] = base["latency_ms"]
if base["rcode"] != 0:
r["error"] = f"rcode={base['rcode']}"
return r

# 2) Hijack: private answer
if any(is_private(a) for a in base["ips"]):
r["hijack"] = True
r["error"] = "private-answer"
return r

# 3) NXDOMAIN trap
trap = f"x{random.getrandbits(64):016x}.invalid"
tr = query(ip, trap, timeout=args.timeout, use_tcp=args.tcp)
if tr and tr["rcode"] == 0 and tr["ips"]:
r["hijack"] = True
r["error"] = "nxdomain-hijack"
return r

r["ok"] = True

# 4) Tunnel-domain forwarding probe
if args.tunnel_domain:
probe = f"x{random.getrandbits(32):08x}.{args.tunnel_domain}"
t = query(ip, probe, qtype=args.tunnel_qtype,
timeout=args.timeout, use_tcp=args.tcp)
if t is not None and t["rcode"] in (0, 3):
r["tunnel_ok"] = True
else:
r["tunnel_ok"] = False

# 5) Burst / QPS
if not args.no_burst and args.burst > 0:
r["qps"] = burst(ip, args.domain, args.burst, args.timeout)

r["score"] = (r["qps"] * 10.0) + (1000.0 / max(r["latency_ms"], 1.0))
if r["tunnel_ok"]:
r["score"] += 500.0
return r


# --------------------------------------------------------------------------- #
# Input / output
# --------------------------------------------------------------------------- #

def read_ips(path: str) -> list[str]:
out = []
with open(path, "r", encoding="utf-8") as f:
for line in f:
s = line.strip()
if not s or s.startswith("#"):
continue
if "/" in s:
try:
net = ipaddress.ip_network(s, strict=False)
out.extend(str(h) for h in net.hosts())
except ValueError:
pass
else:
try:
ipaddress.ip_address(s)
out.append(s)
except ValueError:
pass
return out


def format_line(r: dict, args) -> str:
sym = "OK " if r["ok"] and not r["hijack"] else "BAD"
extra = ""
if args.tunnel_domain:
extra = " TUN✓" if r["tunnel_ok"] else " TUN✗"
return (f"[{sym}] {r['ip']:<16} "
f"lat={r['latency_ms']:7.1f}ms "
f"qps={r['qps']:6.1f}{extra} {r['error']}")


# --------------------------------------------------------------------------- #
# Main
# --------------------------------------------------------------------------- #`

`python
#!/usr/bin/env python3

import argparse
import ipaddress
import os
import queue
import random
import socket
import struct
import sys
import threading
import time


# --------------------------------------------------------------------------- #
# DNS packet helpers (pure struct/socket)
# --------------------------------------------------------------------------- #

def build_query(qname: str, qtype: int = 1, qid: int | None = None):
if qid is None:
qid = random.getrandbits(16)
header = struct.pack("!HHHHHH", qid, 0x0100, 1, 0, 0, 0) # RD=1
body = b""
for label in qname.rstrip(".").split("."):
if not label:
continue
if len(label) > 63:
raise ValueError("label too long")
body += bytes([len(label)]) + label.encode("ascii")
body += b"\x00" + struct.pack("!HH", qtype, 1)
return qid, header + body


def _skip_name(data: bytes, off: int) -> int:
while True:
if off >= len(data):
raise ValueError("name overflow")
length = data[off]
if length == 0:
return off + 1
if (length & 0xC0) == 0xC0:
return off + 2
off += 1 + length


def parse_answer(data: bytes, qid: int):
"""Return (rcode, [ipv4 strings]) or None."""
if len(data) < 12:
return None
rid, flags, qd, an, _ns, _ar = struct.unpack("!HHHHHH", data[:12])
if rid != qid:
return None
rcode = flags & 0x000F
off = 12
try:
for _ in range(qd):
off = _skip_name(data, off) + 4
ips = []
for _ in range(an):
off = _skip_name(data, off)
rtype, _rclass, _ttl, rdlen = struct.unpack("!HHIH", data[off:off + 10])
off += 10
if rtype == 1 and rdlen == 4:
ips.append(socket.inet_ntoa(data[off:off + 4]))
off += rdlen
return rcode, ips
except (ValueError, struct.error):
return None


def query(ip: str, qname: str, qtype: int = 1,
timeout: float = 3.0, use_tcp: bool = False):
"""Return dict(rcode, ips, latency_ms) or None on failure."""
try:
qid, pkt = build_query(qname, qtype)
except ValueError:
return None
t0 = time.monotonic()
try:
if use_tcp:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(timeout)
try:
s.connect((ip, 53))
s.sendall(struct.pack("!H", len(pkt)) + pkt)
hdr = b""
while len(hdr) < 2:
chunk = s.recv(2 - len(hdr))
if not chunk:
return None
hdr += chunk
ln = struct.unpack("!H", hdr)[0]
data = b""
while len(data) < ln:
chunk = s.recv(ln - len(data))
if not chunk:
break
data += chunk
finally:
s.close()
else:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(timeout)
try:
s.sendto(pkt, (ip, 53))
data, _ = s.recvfrom(4096)
finally:
s.close()
dt = (time.monotonic() - t0) * 1000.0
parsed = parse_answer(data, qid)
if parsed is None:
return None
rcode, ips = parsed
return {"rcode": rcode, "ips": ips, "latency_ms": dt}
except OSError:
return None


def burst(ip: str, qname: str, count: int, timeout: float) -> float:
"""Send `count` queries back-to-back, return qps."""
ok = 0
t0 = time.monotonic()
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)