From 09a7d8431ab77b4d607ab5cda8a4019a5a08df3d Mon Sep 17 00:00:00 2001 From: "m.azadi.ir" Date: Wed, 20 May 2026 02:04:06 +0330 Subject: [PATCH] =?UTF-8?q?=D8=A7=D8=B6=D8=A7=D9=81=D9=87=20=D8=B4=D8=AF?= =?UTF-8?q?=D9=86=20geoip=20,=20geosite=20=D8=AC=D9=87=D8=AA=20=D8=B3?= =?UTF-8?q?=D9=87=D9=84=D9=88=D8=AA=20=D9=81=DB=8C=D9=84=D8=AA=D8=B1=DB=8C?= =?UTF-8?q?=D9=86=DA=AF=20=D8=B3=D8=A7=DB=8C=D8=AA=20=D9=87=D8=A7=DB=8C=20?= =?UTF-8?q?=D8=A7=DB=8C=D8=B1=D8=A7=D9=86=DB=8C=20=D9=88=20=DB=8C=D8=A7=20?= =?UTF-8?q?=D8=A7=D8=AA=D8=B5=D8=A7=D9=84=20=D8=AF=D8=A7=DB=8C=D8=B1=DA=A9?= =?UTF-8?q?=D8=AA=20=D8=A8=D9=87=20=D8=B3=D8=B1=D9=88=DB=8C=D8=B3=20=D9=87?= =?UTF-8?q?=D8=A7=DB=8C=DB=8C=20=D9=85=D8=A7=D9=86=D9=86=D8=AF=20deepseek?= =?UTF-8?q?=20=D8=A8=D8=A7=20=D8=A7=D8=B3=D8=AA=D9=81=D8=A7=D8=AF=D9=87=20?= =?UTF-8?q?=D8=A7=D8=B2=20=D8=AA=DA=AF=20=D9=87=D8=A7=DB=8C=20=D8=A7=D8=B3?= =?UTF-8?q?=D8=AA=D8=A7=D9=86=D8=AF=D8=A7=D8=B1=D8=AF=20v2dat=20=3D>=20xra?= =?UTF-8?q?y=20=D8=AC=D9=87=D8=AA=20=D9=BE=D8=B1=D8=AF=D8=A7=D8=B2=D8=B4?= =?UTF-8?q?=20=D8=A7=D8=AA=D8=B5=D8=A7=D9=84=20=D8=A8=D9=87=20=D8=B3=D8=A7?= =?UTF-8?q?=DB=8C=D8=AA=20=D9=BE=DB=8C=D8=B4=20=D8=A7=D8=B2=20=D8=A7=D8=B1?= =?UTF-8?q?=D8=B3=D8=A7=D9=84=20=D8=A8=D9=87=20GAS?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit نمونه تغییرات در کانفیگ "collector.github.com", "api.github.com", "www.github.com", "regex:^github\\.com$", "geosite:ir", "geoip:ir", "geosite:deepseek" توضیحات تگ ها geoip با استفاده از این کانفیگ ها می توان علاوه بر اتصال مستقیم به برخی سایتها یک حوزه IP مانند ایران را مستقیم و بدون نیاز به پروکسی باز کرد geosite می توان با استفاده از این تگ سایت و زیرمجموعه های دیگر آن را به لیست دایرکت یا مسدودی اضافه کرد. به عنوان مثال با تگه دیپ‌سیک تمامی وبسایتها مرتبط با وبسایت و پروژه دیپ‌سیک مستقیم باز خواهد شد regex با استفاده از رجکس می توان علاوه بر محدود سازی محدوده یا زیر دامنه می توان به صورت مطلق یک دامنه را صرفا با ریشه باز کرد و یا بست. به عنوان مثال دامنه های https://github.com https://www.github.com مستقیم وصل بشند و باقی زیر دامنه ها از GAS عبور کند. در صورتی که فایلهای geoip , geosite در ریشه اصلی برنامه موجود نباشد، برنامه سایر تگ ها را بررسی می کند و پیش فرض های تعریف شده خودش رو نیز لحاظ می کند. جهت دریافت فایلهای geosite ,geoip می توان از مسیر های زیر اقدام به دانلود نمود: Iran Xray Chocolate4U: https://github.com/Chocolate4U/Iran-v2ray-rules/releases/latest/download/{0}.dat Russia Xray runetfreedom: https://github.com/runetfreedom/russia-v2ray-rules-dat/releases/latest/download/{0}.dat پس از دانلود فایلهای geosite و geoip و با قرارگرفتن آنها در کنار پروژه اصلی، برای اولین فراخوانی و اولین اتصال به یک دامنه، دیتای فایلهای مربوطه استراخ می گردد پس از استخراج اولیه که حدود 2 تا 7 ثانیه زمان خواهد برد، پس از آن با الگوریتم بهینه شده موجود در کمترین زمان ممکته فیلترینگ دامنه اتفاق می افد --- .gitignore | 1 + config.example.json | 9 +- src/geo/ReadMe.md | 35 ++++ src/geo/__init__.py | 4 + src/geo/geoip_parser.py | 152 +++++++++++++++++ src/geo/geoparser/__init__.py | 7 + src/geo/geoparser/models.py | 80 +++++++++ src/geo/geoparser/parser.py | 306 ++++++++++++++++++++++++++++++++++ src/geo/geoparser/storage.py | 34 ++++ src/geo/geosite_parser.py | 267 +++++++++++++++++++++++++++++ src/geo/rule_engine.py | 262 +++++++++++++++++++++++++++++ src/proxy/proxy_support.py | 13 +- 12 files changed, 1163 insertions(+), 7 deletions(-) create mode 100644 src/geo/ReadMe.md create mode 100644 src/geo/__init__.py create mode 100644 src/geo/geoip_parser.py create mode 100644 src/geo/geoparser/__init__.py create mode 100644 src/geo/geoparser/models.py create mode 100644 src/geo/geoparser/parser.py create mode 100644 src/geo/geoparser/storage.py create mode 100644 src/geo/geosite_parser.py create mode 100644 src/geo/rule_engine.py diff --git a/.gitignore b/.gitignore index 471da414..26f85cde 100644 --- a/.gitignore +++ b/.gitignore @@ -25,6 +25,7 @@ env/ # IDE .vscode/ +.vs/ .idea/ *.code-workspace *.swp diff --git a/config.example.json b/config.example.json index 05f172ec..b88bd0fe 100644 --- a/config.example.json +++ b/config.example.json @@ -29,8 +29,13 @@ ".doubleclick.net" ], "direct_hosts": [ - "rubika.ir", - "soft98.ir" + "collector.github.com", + "api.github.com", + "www.github.com", + "regex:^github\\.com$", + "geosite:ir", + "geoip:ir", + "geosite:deepseek" ], "youtube_via_relay": false, "hosts": { diff --git a/src/geo/ReadMe.md b/src/geo/ReadMe.md new file mode 100644 index 00000000..87eeee57 --- /dev/null +++ b/src/geo/ReadMe.md @@ -0,0 +1,35 @@ +اضافه شدن geoip , geosite جهت سهلوت فیلترینگ سایت های ایرانی و یا اتصال دایرکت به سرویس هایی مانند deepseek با استفاده از تگ های استاندارد +v2dat => xray +جهت پردازش اتصال به سایت پیش از ارسال به GAS + +نمونه تغییرات در کانفیگ + "collector.github.com", + "api.github.com", + "www.github.com", + "regex:^github\\.com$", + "geosite:ir", + "geoip:ir", + "geosite:deepseek" +توضیحات تگ ها +geoip +با استفاده از این کانفیگ ها می توان علاوه بر اتصال مستقیم به برخی سایتها یک حوزه IP مانند ایران را مستقیم و بدون نیاز به پروکسی باز کرد +geosite +می توان با استفاده از این تگ سایت و زیرمجموعه های دیگر آن را به لیست دایرکت یا مسدودی اضافه کرد. به عنوان مثال با تگه دیپ‌سیک تمامی وبسایتها مرتبط با وبسایت و پروژه دیپ‌سیک مستقیم باز خواهد شد +regex +با استفاده از رجکس می توان علاوه بر محدود سازی محدوده یا زیر دامنه می توان به صورت مطلق یک دامنه را صرفا با ریشه باز کرد و یا بست. به عنوان مثال دامنه های +https://github.com +https://www.github.com +مستقیم وصل بشند و باقی زیر دامنه ها از GAS عبور کند. + +در صورتی که فایلهای geoip , geosite در ریشه اصلی برنامه موجود نباشد، برنامه سایر تگ ها را بررسی می کند و پیش فرض های تعریف شده خودش رو نیز لحاظ می کند. + +جهت دریافت فایلهای geosite ,geoip می توان از مسیر های زیر اقدام به دانلود نمود: + +Iran Xray Chocolate4U: +https://github.com/Chocolate4U/Iran-v2ray-rules/releases/latest/download/{0}.dat + +Russia Xray runetfreedom: +https://github.com/runetfreedom/russia-v2ray-rules-dat/releases/latest/download/{0}.dat + +پس از دانلود فایلهای geosite و geoip و با قرارگرفتن آنها در کنار پروژه اصلی، برای اولین فراخوانی و اولین اتصال به یک دامنه، دیتای فایلهای مربوطه استراخ می گردد +پس از استخراج اولیه که حدود 2 تا 7 ثانیه زمان خواهد برد، پس از آن با الگوریتم بهینه شده موجود در کمترین زمان ممکته فیلترینگ دامنه اتفاق می افد \ No newline at end of file diff --git a/src/geo/__init__.py b/src/geo/__init__.py new file mode 100644 index 00000000..208b9183 --- /dev/null +++ b/src/geo/__init__.py @@ -0,0 +1,4 @@ +from .geosite_parser import * +from .geoip_parser import * +from .rule_engine import * + diff --git a/src/geo/geoip_parser.py b/src/geo/geoip_parser.py new file mode 100644 index 00000000..7862c1b2 --- /dev/null +++ b/src/geo/geoip_parser.py @@ -0,0 +1,152 @@ +#!/usr/bin/env python3 +""" +GeoIP Manager - Using dat-editor parser +""" + +import os +import sys +import logging +from typing import List, Optional + +log = logging.getLogger("GeoParser") + +# Add current directory to path to ensure geoparser is found +_current_dir = os.path.dirname(os.path.abspath(__file__)) +if _current_dir not in sys.path: + sys.path.insert(0, _current_dir) + +# Now import from geoparser +try: + from src.geo.geoparser.parser import load_geoip + from src.geo.geoparser.models import GeoIPList, GeoIP, CIDR + from src.geo.geoparser.storage import load as load_dat_file + GEOPARSER_AVAILABLE = True +except ImportError as e: + log.info(f"Warning: Could not import geoparser: {e}") + log.info("Falling back to built-in parser...") + GEOPARSER_AVAILABLE = False + + +class GeoIPManager: + """Manager for geoip operations""" + + def __init__(self, dat_path: str = "geoip.dat"): + self.dat_path = dat_path + self.geoip_list = None + self.is_loaded = False + + def load(self) -> bool: + if (self.is_loaded): + return True + + """Load geoip.dat file""" + if not os.path.exists(self.dat_path): + log.info(f"geoip.dat not found at {self.dat_path}") + return False + + if not GEOPARSER_AVAILABLE: + log.info("Geoparser not available, using fallback mode") + return False + + try: + # Use storage.load() which auto-detects file type + from geoparser.storage import load as load_dat + ftype, data = load_dat(self.dat_path) + + if ftype == "geoip": + self.geoip_list = data + self.is_loaded = True + + total_cidrs = sum(len(entry.cidrs) for entry in self.geoip_list.entries) + log.info(f"Loaded {len(self.geoip_list.entries)} country records from geoip.dat") + log.info(f"Total IP ranges: {total_cidrs}") + + return True + else: + log.error(f"File type mismatch: expected geoip, got {ftype}") + return False + + except Exception as e: + log.error(f"Error loading geoip.dat: {e}") + return False + + def get_ip_ranges_by_country(self, country_code: str) -> List[str]: + """Get IP ranges for a country code""" + if not self.is_loaded: + if not self.load(): + return self._get_builtin_fallback(country_code) + + if not self.geoip_list: + return self._get_builtin_fallback(country_code) + + country_upper = country_code.upper() + + # Search in loaded data + for entry in self.geoip_list.entries: + if entry.code.upper() == country_upper: + return [cidr.to_string() for cidr in entry.cidrs] + + # Try partial match + for entry in self.geoip_list.entries: + if country_upper in entry.code.upper() or entry.code.upper() in country_upper: + return [cidr.to_string() for cidr in entry.cidrs] + + return self._get_builtin_fallback(country_code) + + def get_all_countries(self) -> List[str]: + """Get all available country codes""" + if not self.is_loaded: + self.load() + + if self.geoip_list: + return sorted([entry.code for entry in self.geoip_list.entries]) + return [] + + def _get_builtin_fallback(self, country_code: str) -> List[str]: + """Fallback built-in IP ranges""" + builtin_ranges = { + 'US': ['8.8.8.0/24', '8.8.4.0/24', '4.4.4.0/24', '13.32.0.0/15'], + 'CN': ['1.0.0.0/8', '14.0.0.0/8', '27.0.0.0/8', '36.0.0.0/8', '39.0.0.0/8'], + 'RU': ['5.0.0.0/8', '31.0.0.0/8', '46.0.0.0/8', '77.0.0.0/8', '80.0.0.0/8'], + 'DE': ['2.200.0.0/13', '5.144.128.0/19', '13.224.0.0/14'], + 'GB': ['2.96.0.0/11', '2.120.0.0/13', '5.64.0.0/13'], + 'JP': ['1.0.16.0/20', '1.0.64.0/18', '1.1.0.0/16', '1.21.0.0/16'], + 'KR': ['1.11.0.0/16', '1.96.0.0/13', '1.208.0.0/12', '3.32.0.0/14'], + 'FR': ['2.0.0.0/12', '2.16.0.0/13', '2.24.0.0/14', '5.48.0.0/13'], + } + + country_upper = country_code.upper() + if country_upper in builtin_ranges: + log.info(f"Using built-in IP ranges for '{country_upper}'") + return builtin_ranges[country_upper] + + return [] + + +# Quick test +if __name__ == "__main__": + manager = GeoIPManager("geoip.dat") + + print("="*50) + print("Testing GeoIP Manager") + print("="*50) + + if os.path.exists("geoip.dat"): + manager.load() + countries = manager.get_all_countries() + print(f"\nCountries found: {len(countries)}") + if countries: + print(f"Sample: {', '.join(countries[:2000])}") + else: + print("geoip.dat not found, using built-in fallback") + + # Test extraction + print("\n" + "="*50) + print("IP Range Extraction Test:") + print("="*50) + + for test_code in ['IR']: + ranges = manager.get_ip_ranges_by_country(test_code) + print(f"\n{test_code}: {len(ranges)} IP ranges") + if ranges: + print(f" Sample: {ranges[:30000]}") diff --git a/src/geo/geoparser/__init__.py b/src/geo/geoparser/__init__.py new file mode 100644 index 00000000..9d83cdb6 --- /dev/null +++ b/src/geo/geoparser/__init__.py @@ -0,0 +1,7 @@ +#!/usr/bin/env python3 +""" +Geoparser module - for reading geosite.dat and geoip.dat files +""" +from .models import * +from .parser import * +from .storage import * \ No newline at end of file diff --git a/src/geo/geoparser/models.py b/src/geo/geoparser/models.py new file mode 100644 index 00000000..e45cb666 --- /dev/null +++ b/src/geo/geoparser/models.py @@ -0,0 +1,80 @@ +""" +Data models for geosite.dat and geoip.dat entries. +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import List, Optional + + +DOMAIN_TYPE_TO_NUM = { + "plain": 0, # substring match + "regex": 1, + "root": 2, # root domain match + "full": 3, # exact match +} +NUM_TO_DOMAIN_TYPE = {v: k for k, v in DOMAIN_TYPE_TO_NUM.items()} + + +@dataclass +class Attribute: + key: str + bool_value: Optional[bool] = None + int_value: Optional[int] = None + + def render(self) -> str: + if self.int_value is not None: + return f"{self.key}={self.int_value}" + if self.bool_value in (True, None): + return self.key + if self.bool_value is False: + return f"{self.key}=false" + return self.key + + +@dataclass +class Domain: + value: str + dtype: str = "plain" + attributes: List[Attribute] = field(default_factory=list) + + def render(self) -> str: + attr = "" if not self.attributes else " @" + ",".join(a.render() for a in self.attributes) + return f"{self.dtype}: {self.value}{attr}" + + +@dataclass +class CIDR: + ip: bytes + prefix: int + + def to_string(self) -> str: + import ipaddress + + addr = ipaddress.ip_address(self.ip) + return f"{addr}/{self.prefix}" + + +@dataclass +class GeoSite: + code: str + country_code: str = "" + domains: List[Domain] = field(default_factory=list) + + +@dataclass +class GeoIP: + code: str + country_code: str = "" + cidrs: List[CIDR] = field(default_factory=list) + inverse_match: bool = False + + +@dataclass +class GeoSiteList: + entries: List[GeoSite] = field(default_factory=list) + + +@dataclass +class GeoIPList: + entries: List[GeoIP] = field(default_factory=list) diff --git a/src/geo/geoparser/parser.py b/src/geo/geoparser/parser.py new file mode 100644 index 00000000..048df2a7 --- /dev/null +++ b/src/geo/geoparser/parser.py @@ -0,0 +1,306 @@ +""" +Lightweight protobuf reader/writer specialized for v2ray geosite.dat / geoip.dat. +Keeps dependencies to the stdlib for offline usage. +""" +from __future__ import annotations + +from typing import List, Optional, Tuple + +from .models import ( + Attribute, + CIDR, + DOMAIN_TYPE_TO_NUM, + Domain, + GeoIP, + GeoIPList, + GeoSite, + GeoSiteList, + NUM_TO_DOMAIN_TYPE, +) + + +############################################################################### +# Varint helpers +############################################################################### +def _read_varint(buf: bytes, idx: int) -> Tuple[int, int]: + shift = 0 + result = 0 + while True: + if idx >= len(buf): + raise ValueError("Unexpected EOF while reading varint") + b = buf[idx] + idx += 1 + result |= (b & 0x7F) << shift + if not (b & 0x80): + break + shift += 7 + return result, idx + + +def _write_varint(value: int) -> bytes: + out = bytearray() + v = value + while True: + to_write = v & 0x7F + v >>= 7 + if v: + out.append(to_write | 0x80) + else: + out.append(to_write) + break + return bytes(out) + + +def _skip_value(buf: bytes, idx: int, wire_type: int) -> int: + if wire_type == 0: + _, idx = _read_varint(buf, idx) + return idx + if wire_type == 1: + return idx + 8 + if wire_type == 2: + length, idx = _read_varint(buf, idx) + return idx + length + if wire_type == 5: + return idx + 4 + raise ValueError(f"Unsupported wire type: {wire_type}") + + +def _read_length_delimited(buf: bytes, idx: int) -> Tuple[bytes, int]: + length, idx = _read_varint(buf, idx) + end = idx + length + return buf[idx:end], end + + +############################################################################### +# Parsers +############################################################################### +def _parse_attribute(buf: bytes, start: int, end: int) -> Attribute: + idx = start + key = "" + bool_value = None + int_value = None + while idx < end: + tag, idx = _read_varint(buf, idx) + field_no, wire = tag >> 3, tag & 0x7 + if field_no == 1 and wire == 2: + raw, idx = _read_length_delimited(buf, idx) + key = raw.decode() + elif field_no == 2 and wire == 0: + val, idx = _read_varint(buf, idx) + bool_value = bool(val) + elif field_no == 3 and wire == 0: + val, idx = _read_varint(buf, idx) + int_value = val + else: + idx = _skip_value(buf, idx, wire) + return Attribute(key=key, bool_value=bool_value, int_value=int_value) + + +def _parse_domain(buf: bytes, start: int, end: int) -> Domain: + idx = start + dtype_num = 0 + value = "" + attrs: List[Attribute] = [] + while idx < end: + tag, idx = _read_varint(buf, idx) + field_no, wire = tag >> 3, tag & 0x7 + if field_no == 1 and wire == 0: + dtype_num, idx = _read_varint(buf, idx) + elif field_no == 2 and wire == 2: + raw, idx = _read_length_delimited(buf, idx) + value = raw.decode() + elif field_no == 3 and wire == 2: + raw, idx = _read_length_delimited(buf, idx) + attrs.append(_parse_attribute(raw, 0, len(raw))) + else: + idx = _skip_value(buf, idx, wire) + return Domain(value=value, dtype=NUM_TO_DOMAIN_TYPE.get(dtype_num, "plain"), attributes=attrs) + + +def _parse_cidr(buf: bytes, start: int, end: int) -> CIDR: + idx = start + ip = b"" + prefix = 0 + while idx < end: + tag, idx = _read_varint(buf, idx) + field_no, wire = tag >> 3, tag & 0x7 + if field_no == 1 and wire == 2: + ip, idx = _read_length_delimited(buf, idx) + elif field_no == 2 and wire == 0: + prefix, idx = _read_varint(buf, idx) + else: + idx = _skip_value(buf, idx, wire) + return CIDR(ip=ip, prefix=prefix) + + +def _parse_geosite(buf: bytes, start: int, end: int) -> GeoSite: + idx = start + code = "" + country_code = "" + domains: List[Domain] = [] + while idx < end: + tag, idx = _read_varint(buf, idx) + field_no, wire = tag >> 3, tag & 0x7 + if field_no == 1 and wire == 2: + raw, idx = _read_length_delimited(buf, idx) + country_code = raw.decode() + elif field_no == 2 and wire == 2: + raw, idx = _read_length_delimited(buf, idx) + domains.append(_parse_domain(raw, 0, len(raw))) + elif field_no == 4 and wire == 2: + raw, idx = _read_length_delimited(buf, idx) + code = raw.decode() + else: + idx = _skip_value(buf, idx, wire) + return GeoSite(code=code or country_code, country_code=country_code, domains=domains) + + +def _parse_geoip(buf: bytes, start: int, end: int) -> GeoIP: + idx = start + code = "" + country_code = "" + cidrs: List[CIDR] = [] + inverse_match = False + while idx < end: + tag, idx = _read_varint(buf, idx) + field_no, wire = tag >> 3, tag & 0x7 + if field_no == 1 and wire == 2: + raw, idx = _read_length_delimited(buf, idx) + country_code = raw.decode() + elif field_no == 2 and wire == 2: + raw, idx = _read_length_delimited(buf, idx) + cidrs.append(_parse_cidr(raw, 0, len(raw))) + elif field_no == 3 and wire == 0: + val, idx = _read_varint(buf, idx) + inverse_match = bool(val) + elif field_no == 5 and wire == 2: + raw, idx = _read_length_delimited(buf, idx) + code = raw.decode() + else: + idx = _skip_value(buf, idx, wire) + return GeoIP(code=code or country_code, country_code=country_code, cidrs=cidrs, inverse_match=inverse_match) + + +############################################################################### +# Serialization +############################################################################### +def _encode_field(tag: int, payload: bytes) -> bytes: + return _write_varint(tag) + payload + + +def _ld(payload: bytes) -> bytes: + return _write_varint(len(payload)) + payload + + +def _serialize_attribute(attr: Attribute) -> bytes: + out = bytearray() + out += _encode_field((1 << 3) | 2, _ld(attr.key.encode())) + if attr.bool_value is not None: + out += _encode_field((2 << 3) | 0, _write_varint(1 if attr.bool_value else 0)) + if attr.int_value is not None: + out += _encode_field((3 << 3) | 0, _write_varint(attr.int_value)) + return bytes(out) + + +def _serialize_domain(dom: Domain) -> bytes: + out = bytearray() + out += _encode_field((1 << 3) | 0, _write_varint(DOMAIN_TYPE_TO_NUM.get(dom.dtype, 0))) + out += _encode_field((2 << 3) | 2, _ld(dom.value.encode())) + for attr in dom.attributes: + payload = _serialize_attribute(attr) + out += _encode_field((3 << 3) | 2, _ld(payload)) + return bytes(out) + + +def _serialize_cidr(cidr: CIDR) -> bytes: + out = bytearray() + out += _encode_field((1 << 3) | 2, _ld(cidr.ip)) + out += _encode_field((2 << 3) | 0, _write_varint(cidr.prefix)) + return bytes(out) + + +def _serialize_geosite(site: GeoSite) -> bytes: + out = bytearray() + if site.country_code: + out += _encode_field((1 << 3) | 2, _ld(site.country_code.encode())) + for dom in site.domains: + payload = _serialize_domain(dom) + out += _encode_field((2 << 3) | 2, _ld(payload)) + if site.code: + out += _encode_field((4 << 3) | 2, _ld(site.code.encode())) + return bytes(out) + + +def _serialize_geoip(entry: GeoIP) -> bytes: + out = bytearray() + if entry.country_code: + out += _encode_field((1 << 3) | 2, _ld(entry.country_code.encode())) + for cidr in entry.cidrs: + payload = _serialize_cidr(cidr) + out += _encode_field((2 << 3) | 2, _ld(payload)) + out += _encode_field((3 << 3) | 0, _write_varint(1 if entry.inverse_match else 0)) + if entry.code: + out += _encode_field((5 << 3) | 2, _ld(entry.code.encode())) + return bytes(out) + + +############################################################################### +# Public API +############################################################################### +def load_geosite(path: str) -> GeoSiteList: + data = _safe_read(path) + idx = 0 + entries: List[GeoSite] = [] + while idx < len(data): + tag, idx = _read_varint(data, idx) + field_no, wire = tag >> 3, tag & 0x7 + if field_no != 1 or wire != 2: + idx = _skip_value(data, idx, wire) + continue + raw, idx = _read_length_delimited(data, idx) + entries.append(_parse_geosite(raw, 0, len(raw))) + return GeoSiteList(entries=entries) + + +def load_geoip(path: str) -> GeoIPList: + data = _safe_read(path) + idx = 0 + entries: List[GeoIP] = [] + while idx < len(data): + tag, idx = _read_varint(data, idx) + field_no, wire = tag >> 3, tag & 0x7 + if field_no != 1 or wire != 2: + idx = _skip_value(data, idx, wire) + continue + raw, idx = _read_length_delimited(data, idx) + entries.append(_parse_geoip(raw, 0, len(raw))) + return GeoIPList(entries=entries) + + +def save_geosite(gs: GeoSiteList, path: str) -> None: + out = bytearray() + for entry in gs.entries: + payload = _serialize_geosite(entry) + out += _encode_field((1 << 3) | 2, _ld(payload)) + _write_all(path, bytes(out)) + + +def save_geoip(gi: GeoIPList, path: str) -> None: + out = bytearray() + for entry in gi.entries: + payload = _serialize_geoip(entry) + out += _encode_field((1 << 3) | 2, _ld(payload)) + _write_all(path, bytes(out)) + + +def _safe_read(path: str) -> bytes: + try: + with open(path, "rb") as f: + return f.read() + except FileNotFoundError: + return b"" + + +def _write_all(path: str, data: bytes) -> None: + with open(path, "wb") as f: + f.write(data) diff --git a/src/geo/geoparser/storage.py b/src/geo/geoparser/storage.py new file mode 100644 index 00000000..89ac4a1a --- /dev/null +++ b/src/geo/geoparser/storage.py @@ -0,0 +1,34 @@ +""" +Storage façade: picks correct parser based on file type, handles inference. +""" +from __future__ import annotations + +import os +from typing import Literal, Tuple + +from .models import GeoIPList, GeoSiteList +from . import parser + + +FileType = Literal["geosite", "geoip"] + + +def infer_type(path: str, explicit: FileType | None = None) -> FileType: + if explicit: + return explicit + name = os.path.basename(path).lower() + return "geoip" if "geoip" in name else "geosite" + + +def load(path: str, explicit: FileType | None = None) -> Tuple[FileType, GeoSiteList | GeoIPList]: + ftype = infer_type(path, explicit) + if ftype == "geosite": + return ftype, parser.load_geosite(path) + return ftype, parser.load_geoip(path) + + +def save(path: str, data: GeoSiteList | GeoIPList, ftype: FileType) -> None: + if ftype == "geosite": + parser.save_geosite(data, path) + else: + parser.save_geoip(data, path) diff --git a/src/geo/geosite_parser.py b/src/geo/geosite_parser.py new file mode 100644 index 00000000..7ac47994 --- /dev/null +++ b/src/geo/geosite_parser.py @@ -0,0 +1,267 @@ +#!/usr/bin/env python3 +""" +geosite_parser.py - Geosite.dat parser using dat-editor parser +Lazy loading with caching - loads once, caches results +""" + +import os +import sys +import time +import logging +from typing import List, Dict, Set, Optional + +log = logging.getLogger("GeoParser") + +# Add parent directory to path for geoparser imports +_current_dir = os.path.dirname(os.path.abspath(__file__)) +_geoparser_path = os.path.join(_current_dir, 'geoparser') +if _geoparser_path not in sys.path: + sys.path.insert(0, _geoparser_path) + +# Import from geoparser modules - مسیر درست +from src.geo.geoparser.models import GeoSiteList, GeoSite, Domain +from src.geo.geoparser.parser import load_geosite +from src.geo.geoparser.storage import load as load_dat_file + + +class GeositeParser: + """Parse geosite.dat using dat-editor parser""" + + def __init__(self, filepath: str = "geosite.dat"): + self.filepath = filepath + self.domains_by_tag: Dict[str, Set[str]] = {} + self.all_domains: Set[str] = set() + self.is_loaded = False + self.geosite_list: Optional[GeoSiteList] = None + self.load_time = 0 + + def load(self, silent: bool = False) -> bool: + """Load and extract domains from geosite.dat""" + if self.is_loaded: + return True + + if not os.path.exists(self.filepath): + if not silent: + log.error(f"geosite.dat not found at {self.filepath}") + return False + + start_time = time.time() + + try: + ftype, data = load_dat_file(self.filepath) + + if ftype == "geosite" and isinstance(data, GeoSiteList): + self.geosite_list = data + else: + self.geosite_list = load_geosite(self.filepath) + + self._parse_geosite_data() + + self.is_loaded = True + self.load_time = time.time() - start_time + + if not silent: + log.info(f"Loaded geosite.dat in {self.load_time:.2f}s") + log.info(f" - {len(self.geosite_list.entries)} site entries") + log.info(f" - {len(self.all_domains)} total domains") + log.info(f" - {len(self.domains_by_tag)} tagged categories") + + return True + + except Exception as e: + if not silent: + log.error(f"Error loading geosite.dat: {e}") + return False + + def _parse_geosite_data(self): + """Parse GeoSiteList into domains_by_tag dictionary""" + if not self.geosite_list: + return + + for entry in self.geosite_list.entries: + tag = entry.code or entry.country_code + if not tag: + continue + + if tag not in self.domains_by_tag: + self.domains_by_tag[tag] = set() + + for domain in entry.domains: + domain_str = domain.value + self.domains_by_tag[tag].add(domain_str) + self.all_domains.add(domain_str) + + def get_domains_by_tag(self, tag: str) -> List[str]: + """Get domains for a specific tag""" + if not self.is_loaded: + return [] + + tag_lower = tag.lower() + + for t in self.domains_by_tag: + if t.lower() == tag_lower: + return sorted(self.domains_by_tag[t]) + + for t in self.domains_by_tag: + if tag_lower in t.lower(): + return sorted(self.domains_by_tag[t]) + + return [] + + def search_domains(self, keyword: str) -> List[str]: + """Search domains containing keyword""" + if not self.is_loaded: + return [] + + keyword_lower = keyword.lower() + results = [] + + for domain in self.all_domains: + if keyword_lower in domain.lower(): + results.append(domain) + + return sorted(results)[:10000] + + def get_all_tags(self) -> List[str]: + """Get all available tags""" + if not self.is_loaded: + return [] + return sorted(self.domains_by_tag.keys()) + + def get_stats(self) -> dict: + """Get statistics""" + if not self.is_loaded: + return {'total_domains': 0, 'total_tags': 0, 'tags': []} + + return { + 'total_domains': len(self.all_domains), + 'total_tags': len(self.domains_by_tag), + 'tags': list(self.domains_by_tag.keys())[:2000], + 'load_time': self.load_time + } + + +class GeositeManager: + """Manager for geosite operations with fallback to built-in lists""" + + def __init__(self, dat_path: str = "geosite.dat"): + self.dat_path = dat_path + self.parser: Optional[GeositeParser] = None + self._load_attempted = False + self._cache: Dict[str, List[str]] = {} + + self.builtin_domains = { + 'google': ['google.com', 'gstatic.com', 'googleapis.com', 'youtube.com'], + 'facebook': ['facebook.com', 'fbcdn.net', 'instagram.com'], + 'twitter': ['twitter.com', 'twimg.com', 'x.com'], + 'github': ['github.com', 'github.io', 'githubusercontent.com'], + 'cloudflare': ['cloudflare.com', 'cfassets.com'], + 'microsoft': ['microsoft.com', 'windows.com', 'office.com'], + 'apple': ['apple.com', 'icloud.com'], + 'netflix': ['netflix.com', 'nflxext.com'], + 'amazon': ['amazon.com', 'amazonaws.com'], + 'telegram': ['telegram.org', 't.me'], + 'whatsapp': ['whatsapp.com'], + 'instagram': ['instagram.com'], + 'reddit': ['reddit.com'], + 'youtube': ['youtube.com', 'youtu.be'], + 'cn': ['baidu.com', 'qq.com', 'taobao.com'], + 'ir': ['aparat.com', 'digikala.com', 'divar.ir'], + } + + def _ensure_parser(self, silent: bool = False) -> bool: + """Initialize parser only when needed""" + if self.parser is not None and self.parser.is_loaded: + return True + + if self._load_attempted: + return False + + self._load_attempted = True + self.parser = GeositeParser(self.dat_path) + + if self.parser.load(silent=silent): + return True + else: + self.parser = None + return False + + def extract_domains_by_geosite(self, tag: str, silent: bool = False) -> List[str]: + """Extract domains from geosite.dat for a given tag""" + tag_lower = tag.lower() + + if tag_lower in self._cache: + return self._cache[tag_lower] + + if self._ensure_parser(silent=silent): + domains = self.parser.get_domains_by_tag(tag) + if domains: + self._cache[tag_lower] = domains + return domains + + domains = self.parser.search_domains(tag) + if domains: + self._cache[tag_lower] = domains + return domains + + if tag_lower in self.builtin_domains: + domains = self.builtin_domains[tag_lower] + self._cache[tag_lower] = domains + return domains + + for key, domains in self.builtin_domains.items(): + if tag_lower in key or key in tag_lower: + self._cache[tag_lower] = domains + return domains + + return [] + + def list_available_tags(self) -> List[str]: + """List all available tags""" + tags = set(self.builtin_domains.keys()) + + if self.parser and self.parser.is_loaded: + tags.update(self.parser.get_all_tags()) + + return sorted(tags) + + def search_tag(self, keyword: str, silent: bool = False) -> List[str]: + """Search tags containing keyword""" + return self.extract_domains_by_geosite(keyword, silent=silent) + + def is_available(self) -> bool: + """Check if geosite.dat is available""" + return os.path.exists(self.dat_path) + + def reload(self, silent: bool = False) -> bool: + """Force reload geosite.dat""" + self.parser = None + self._load_attempted = False + self._cache.clear() + return self._ensure_parser(silent=silent) + + def clear_cache(self): + """Clear resolved tags cache""" + self._cache.clear() + + +# Test +if __name__ == "__main__": + manager = GeositeManager("geosite.dat") + manager.load() + + print("="*60) + print("GEOSITE MANAGER TEST") + print("="*60) + + tags = manager.list_available_tags() + print(f"\ntags: {len(tags)} tags") + if tags: + print(f" Sample: {tags[:3000]}") + + for tag in tags: + domains = manager.extract_domains_by_geosite(tag) + print(f"\n{tag}: {len(domains)} domains") + if domains: + print(f" domains: {domains}") + diff --git a/src/geo/rule_engine.py b/src/geo/rule_engine.py new file mode 100644 index 00000000..16dde22b --- /dev/null +++ b/src/geo/rule_engine.py @@ -0,0 +1,262 @@ +#!/usr/bin/env python3 +""" +rule_engine.py - Complete rule matching with geosite, geoip, regex support +""" + +import re +import socket +import ipaddress +from typing import Set, Tuple, Optional, List, Dict +from geosite_parser import GeositeManager +from geoip_parser import GeoIPManager +import logging + +log = logging.getLogger("GeoParser") + +_GEOSITE_MANAGER = GeositeManager("geosite.dat") +_GEOIP_MANAGER = GeoIPManager("geoip.dat") + + +class GeoIPChecker: + """Simple geoip checker - IPv4 only""" + + def __init__(self, geoip_manager): + self.geoip_manager = geoip_manager + self._cidr_cache: Dict[str, List[Tuple[int, int]]] = {} + + def _ip_to_int(self, ip: str) -> Optional[int]: + """Convert IPv4 to integer - IPv6 not supported""" + if ':' in ip: + return None + + parts = ip.split('.') + if len(parts) != 4: + return None + + try: + return (int(parts[0]) << 24) + (int(parts[1]) << 16) + (int(parts[2]) << 8) + int(parts[3]) + except: + return None + + def _cidr_to_range(self, cidr: str) -> Tuple[Optional[int], Optional[int]]: + """Convert CIDR to (start_int, end_int) - IPv4 only""" + if ':' in cidr: + return (None, None) + + if '/' in cidr: + try: + network = ipaddress.ip_network(cidr, strict=False) + if ':' in str(network.network_address): + return (None, None) + + start = self._ip_to_int(str(network.network_address)) + end = self._ip_to_int(str(network.broadcast_address)) + if start is None or end is None: + return (None, None) + return (start, end) + except: + return (None, None) + else: + ip_int = self._ip_to_int(cidr) + if ip_int is None: + return (None, None) + return (ip_int, ip_int) + + def _get_country_ranges(self, country_code: str) -> List[Tuple[int, int]]: + """Get IP ranges for country as integer ranges - IPv4 only""" + country_upper = country_code.upper() + + if country_upper in self._cidr_cache: + return self._cidr_cache[country_upper] + + ranges = [] + cidrs = self.geoip_manager.get_ip_ranges_by_country(country_code) + + for cidr in cidrs: + if ':' not in cidr: + start, end = self._cidr_to_range(cidr) + if start is not None: + ranges.append((start, end)) + + ranges.sort(key=lambda x: x[0]) + self._cidr_cache[country_upper] = ranges + return ranges + + def _get_ip(self, host: str) -> Optional[str]: + """Get IP from host (domain or IP)""" + try: + ipaddress.ip_address(host) + return host + except: + pass + + try: + return socket.gethostbyname(host) + except: + return None + + def check(self, host: str, country_code: str) -> bool: + """Check if host IP belongs to country - IPv4 only""" + ip = self._get_ip(host) + if ip is None or ':' in ip: + return False + + ip_int = self._ip_to_int(ip) + if ip_int is None: + return False + + ranges = self._get_country_ranges(country_code) + + left, right = 0, len(ranges) - 1 + while left <= right: + mid = (left + right) // 2 + start, end = ranges[mid] + + if ip_int < start: + right = mid - 1 + elif ip_int > end: + left = mid + 1 + else: + return True + + return False + + +class RuleEngine: + """Main rule matching engine""" + + def __init__(self, geosite_manager=None, geoip_manager=None): + self.geosite_manager = geosite_manager + self.geoip_checker = GeoIPChecker(geoip_manager) if geoip_manager else None + self._geosite_cache: Dict[str, Tuple[Set[str], Tuple[str, ...]]] = {} + + def _parse_rule_type(self, rule: str) -> Tuple[str, str]: + """Parse rule and return (type, value)""" + if rule.startswith("geosite:"): + return ("geosite", rule[8:]) + elif rule.startswith("geoip:"): + return ("geoip", rule[6:]) + elif rule.startswith("regex:"): + return ("regex", rule[6:]) + else: + return ("exact", rule) + + def _geosite_to_rules(self, tag: str) -> Tuple[Set[str], Tuple[str, ...]]: + """Convert geosite tag to (exact_domains, suffixes)""" + if tag in self._geosite_cache: + return self._geosite_cache[tag] + + exact = set() + suffixes = set() + + if self.geosite_manager: + domains = self.geosite_manager.extract_domains_by_geosite(tag, silent=True) + + for domain in domains: + domain = domain.lower().rstrip('.') + if domain.startswith('.'): + suffixes.add(domain[1:]) + else: + exact.add(domain) + + result = (exact, tuple(suffixes)) + self._geosite_cache[tag] = result + return result + + def _match_geosite(self, host: str, tag: str) -> bool: + """Check if host matches a geosite tag""" + exact, suffixes = self._geosite_to_rules(tag) + normalized = host.lower().rstrip('.') + + if normalized in exact: + return True + + for suffix in suffixes: + if normalized == suffix or normalized.endswith('.' + suffix): + return True + + return False + + def _match_regex(self, host: str, pattern: str) -> bool: + """Check if host matches regex pattern""" + try: + return bool(re.search(pattern, host, re.IGNORECASE)) + except re.error: + return False + + def match_host_with_rules(self, host: str, rules: tuple[set[str], tuple[str, ...]]) -> bool: + """Match host with rules in order: exact -> regex -> geosite -> geoip""" + + normalized = host.lower().rstrip('.') + + # چک کردن legacy rules اول + exact_legacy, suffixes_legacy = rules + if normalized in exact_legacy: + return True + if any(normalized.endswith(suffix) for suffix in suffixes_legacy): + return True + + # جدا کردن رول‌ها از exact_legacy + exact_rules = [] + regex_rules = [] + geosite_rules = [] + geoip_rules = [] + + for rule in exact_legacy: + rule_type, value = self._parse_rule_type(rule) + if rule_type == "exact": + exact_rules.append(value) + elif rule_type == "regex": + regex_rules.append(value) + elif rule_type == "geosite": + geosite_rules.append(value) + elif rule_type == "geoip": + geoip_rules.append(value) + + # مرحله 1: exact + for value in exact_rules: + if normalized == value.lower().rstrip('.'): + return True + + # مرحله 2: regex + for pattern in regex_rules: + if self._match_regex(host, pattern): + return True + + # مرحله 3: geosite + for tag in geosite_rules: + if self.geosite_manager and self._match_geosite(host, tag): + return True + + # مرحله 4: geoip + for country in geoip_rules: + if self.geoip_checker and self.geoip_checker.check(host, country): + return True + + return False + + +# Enhanced version with geosite/geoip/regex support +def match_host_with_rules(host: str, rules: tuple[set[str], tuple[str, ...]]) -> bool: + try: + _GEOSITE_MANAGER._ensure_parser(silent=False) + _GEOIP_MANAGER.load() + except: + pass + engine = RuleEngine(_GEOSITE_MANAGER, _GEOIP_MANAGER) + return engine.match_host_with_rules(host, rules) + + +if __name__ == "__main__": + print("Rule Engine Test") + print("="*50) + + exact = {'google.com', 'youtube.com', 'geosite:github', 'geoip:ir'} + suffixes = ('.google.com', '.youtube.com') + rules = (exact, suffixes) + + test_hosts = ["google.com", "github.com", "8.8.8.8", "tabnak.ir"] + + for host in test_hosts: + result = match_host_with_rules(host, rules) + print(f"{host}: {result}") \ No newline at end of file diff --git a/src/proxy/proxy_support.py b/src/proxy/proxy_support.py index d32e6a9b..f9ef4b9c 100644 --- a/src/proxy/proxy_support.py +++ b/src/proxy/proxy_support.py @@ -17,6 +17,7 @@ CACHE_TTL_STATIC_MED, STATIC_EXTS, ) +from src.geo.rule_engine import match_host_with_rules __all__ = [ "is_ip_literal", @@ -90,11 +91,8 @@ def load_host_rules(raw) -> tuple[set[str], tuple[str, ...]]: def host_matches_rules(host: str, rules: tuple[set[str], tuple[str, ...]]) -> bool: - exact, suffixes = rules - normalized = host.lower().rstrip(".") - if normalized in exact: - return True - return any(normalized.endswith(suffix) for suffix in suffixes) + return match_host_with_rules(host, rules) + def header_value(headers: dict | None, name: str) -> str: @@ -310,3 +308,8 @@ def inject_cors_headers(response: bytes, origin: str) -> bytes: "Vary: Origin", ] return ("\r\n".join(lines) + "\r\n\r\n").encode() + body + + + + + \ No newline at end of file