community-plugins/ruh-vpn/backend/routing/rules.py

"""Helpers for translating user routing rules into sing-box rule entries."""

from __future__ import annotations

import ipaddress
import re
from typing import Any

from backend.models.server import RoutingRule


# Country / region routing presets. Each one adds a pair of rule_set entries
# (domains + IPs) and a single route rule that sends matches to the proxy.
# Tags must be unique across active presets so sing-box doesn't reject the
# config — the keys below were picked to avoid collisions.
PRESETS: dict[str, dict[str, Any]] = {
    "ru": {
        "key": "ru",
        "name": "Russia",
        "flag": "🇷🇺",
        "description": "Re-filter list — sites and IPs blocked in Russia",
        "rule_sets": [
            {
                "tag": "refilter_domains",
                "type": "remote",
                "format": "binary",
                "url": "https://github.com/1andrevich/Re-filter-lists/releases/latest/download/ruleset-domain-refilter_domains.srs",
                "download_detour": "direct",
            },
            {
                "tag": "refilter_ipsum",
                "type": "remote",
                "format": "binary",
                "url": "https://github.com/1andrevich/Re-filter-lists/releases/latest/download/ruleset-ip-refilter_ipsum.srs",
                "download_detour": "direct",
            },
        ],
    },
    # There is no "blocked in China" list: the GFW blocks foreign services, so
    # the standard bypass is the inverse — proxy everything geolocated OUTSIDE
    # China and let domestic traffic go direct. sing-geosite publishes its .srs
    # files on the `rule-set` branch, not as release assets.
    "cn": {
        "key": "cn",
        "name": "China",
        "flag": "🇨🇳",
        "description": "GFW bypass — foreign (non-Chinese) sites via VPN",
        "rule_sets": [
            {
                "tag": "geosite_noncn",
                "type": "remote",
                "format": "binary",
                "url": "https://raw.githubusercontent.com/SagerNet/sing-geosite/rule-set/geosite-geolocation-!cn.srs",
                "download_detour": "direct",
            },
        ],
    },
    # geosite-sanctioned covers sites unavailable from Iran (state blocks and
    # foreign sanctions). geosite-ir would be the opposite — Iranian domestic
    # sites, which need no proxy. Same .srs-on-branch layout as sing-geosite.
    "ir": {
        "key": "ir",
        "name": "Iran",
        "flag": "🇮🇷",
        "description": "Sites unavailable from Iran (blocks and sanctions) via VPN",
        "rule_sets": [
            {
                "tag": "geosite_sanctioned",
                "type": "remote",
                "format": "binary",
                "url": "https://raw.githubusercontent.com/Chocolate4U/Iran-sing-box-rules/rule-set/geosite-sanctioned.srs",
                "download_detour": "direct",
            },
        ],
    },
}


def preset_rule_sets(active: list[str]) -> list[dict]:
    """Return rule_set entries for the given active preset keys, deduped by tag."""
    seen: set[str] = set()
    out: list[dict] = []
    for key in active:
        preset = PRESETS.get(key)
        if not preset:
            continue
        for rs in preset["rule_sets"]:
            if rs["tag"] in seen:
                continue
            seen.add(rs["tag"])
            out.append(dict(rs))
    return out


def preset_route_rules(active: list[str]) -> list[dict]:
    """One route.rules entry per active preset routing its tags to 'proxy'."""
    out: list[dict] = []
    for key in active:
        preset = PRESETS.get(key)
        if not preset:
            continue
        tags = [rs["tag"] for rs in preset["rule_sets"]]
        if tags:
            out.append({"rule_set": tags, "outbound": "proxy"})
    return out


def preset_domain_tags(active: list[str]) -> list[str]:
    """Tags of rule_sets that match domains (used for proxy-DNS rule).

    Heuristic: any tag containing 'domain' or 'site' is treated as domain-only.
    IPs don't help the DNS layer, so we skip them here.
    """
    tags: list[str] = []
    for key in active:
        preset = PRESETS.get(key)
        if not preset:
            continue
        for rs in preset["rule_sets"]:
            t = rs["tag"]
            tl = t.lower()
            if "domain" in tl or "site" in tl:
                tags.append(t)
    return tags

# A hostname label: 1–63 chars, alphanumerics + hyphens (not at edges).
# `*` is allowed as the leftmost label so wildcards like *.example.com work.
_LABEL_RE = re.compile(r"^(?:\*|[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?)$")


def normalize_pattern(pattern: str) -> str:
    """Clean a user-supplied pattern.

    - Strips http:// and https:// prefixes (extracts hostname).
    - Strips any path / query / fragment from a URL-like input.
    - Strips trailing slashes and surrounding whitespace.
    Domain and CIDR forms pass through unchanged (case-folded for domains).
    """
    p = (pattern or "").strip()
    if not p:
        return ""
    low = p.lower()
    if low.startswith("http://"):
        p = p[len("http://"):]
    elif low.startswith("https://"):
        p = p[len("https://"):]
    # Cut anything after the host: path, query, fragment.
    for sep in ("/", "?", "#"):
        # Don't cut the slash in CIDRs (digits on the right of '/').
        if sep == "/" and "/" in p:
            host, _, tail = p.partition("/")
            if tail and tail[0].isdigit() and host and (host[0].isdigit() or ":" in host):
                # Looks like a CIDR — keep as-is.
                continue
            p = host
        elif sep in p:
            p = p.split(sep, 1)[0]
    # Strip credentials and port (e.g. user:pass@host:443).
    if "@" in p:
        p = p.split("@", 1)[1]
    # Port: only strip when it's not part of an IPv6 literal.
    if p.count(":") == 1 and not p.startswith("["):
        p = p.split(":", 1)[0]
    return p.rstrip(".").lower()


def validate_pattern(pattern: str) -> str:
    """Validate and return the normalized pattern.

    Raises ValueError when the pattern is not a recognized
    domain / wildcard domain / CIDR form.
    """
    p = normalize_pattern(pattern)
    if not p:
        raise ValueError("pattern is empty")

    # CIDR
    if "/" in p and not p.startswith("*"):
        try:
            ipaddress.ip_network(p, strict=False)
        except ValueError as exc:
            raise ValueError(f"invalid CIDR: {p!r} ({exc})") from exc
        return p

    # Bare IP without prefix is not allowed here (CIDR only)
    try:
        ipaddress.ip_address(p)
        raise ValueError(f"{p!r} is a bare IP; use CIDR (e.g. {p}/32)")
    except ValueError:
        pass

    # Wildcard or domain
    labels = p.split(".")
    if not labels or any(not _LABEL_RE.match(label) for label in labels):
        raise ValueError(f"invalid domain pattern: {p!r}")
    # `*` may only appear as the leftmost label.
    if any(label == "*" for label in labels[1:]):
        raise ValueError(f"wildcard '*' only allowed as leftmost label: {p!r}")
    return p


def classify_pattern(pattern: str) -> str:
    """Return one of: 'cidr', 'wildcard', 'keyword', 'domain'."""
    p = pattern.strip()
    if not p:
        return "domain"
    if "/" in p and not p.startswith("*"):
        return "cidr"
    if p.startswith("*."):
        return "wildcard"
    if p.startswith("*") or p.endswith("*"):
        return "keyword"
    return "domain"


def rules_to_singbox(rules: list[RoutingRule]) -> list[dict]:
    """Convert a list of user rules into sing-box route.rules entries.

    Skips disabled rules and rules with no pattern.
    Result preserves input order (first match wins in sing-box).
    """
    out: list[dict] = []
    for r in rules:
        sr = r.to_singbox_rule()
        if sr is not None:
            out.append(sr)
    return out