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