mirror of
https://github.com/shadowsocks/ShadowsocksX-NG.git
synced 2026-11-08 10:34:51 +00:00
378 lines
13 KiB
Python
Executable File
378 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Convert the official Base64 GFWList AutoProxy list into an NG2 rule snapshot.
|
|
|
|
Domain anchors become suffix matches. URL prefixes with an empty or root path and a
|
|
literal domain host become exact matches, discarding the scheme and port.
|
|
Other URL conditions are reported as losses rather than expanded to domains.
|
|
`@@` exceptions shadowed by a broader proxy rule are kept out of the ACL and
|
|
reported item by item; the broader proxy rules stay. Unknown syntax, corrupt
|
|
input, or abnormal shrinkage block the update and leave the previous snapshot.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import binascii
|
|
import datetime as dt
|
|
import hashlib
|
|
import ipaddress
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
CONVERTER_VERSION = "2.0.0"
|
|
SCHEMA_VERSION = 2
|
|
|
|
MINIMUM_RULE_COUNT = 100
|
|
MAXIMUM_RULE_COUNT = 50_000
|
|
MINIMUM_SCALE_RATIO = 0.5
|
|
MAXIMUM_SCALE_RATIO = 2.0
|
|
|
|
DOMAIN_HOST_RE = re.compile(r"^[A-Za-z0-9]([A-Za-z0-9.-]*[A-Za-z0-9])?$")
|
|
|
|
|
|
class UnknownSyntax(Exception):
|
|
def __init__(self, line: str) -> None:
|
|
super().__init__(f"unknown AutoProxy syntax: {line}")
|
|
self.line = line
|
|
|
|
|
|
def looks_like_ip(text: str) -> bool:
|
|
try:
|
|
ipaddress.ip_address(text)
|
|
return True
|
|
except ValueError:
|
|
pass
|
|
parts = text.split(".")
|
|
return len(parts) == 4 and all(p.isdigit() for p in parts)
|
|
|
|
|
|
def is_valid_domain_host(host: str) -> bool:
|
|
return bool(host) and "." in host and DOMAIN_HOST_RE.fullmatch(host) is not None
|
|
|
|
|
|
def parse_line(raw_line: str) -> dict:
|
|
"""Classify one AutoProxy line. Raises UnknownSyntax on unrecognized forms."""
|
|
line = raw_line.strip()
|
|
if not line:
|
|
return {"kind": "blank"}
|
|
if line.startswith("!"):
|
|
return {"kind": "comment"}
|
|
if line.startswith("[") and line.endswith("]"):
|
|
return {"kind": "header"}
|
|
|
|
is_exception = line.startswith("@@")
|
|
pattern = line[2:] if is_exception else line
|
|
|
|
if pattern.startswith("||"):
|
|
return parse_domain_pattern(pattern[2:], line, is_exception)
|
|
if pattern.startswith("|"):
|
|
return parse_url_prefix(pattern[1:], line, is_exception)
|
|
if len(pattern) >= 3 and pattern.startswith("/") and pattern.endswith("/"):
|
|
return {"kind": "regexp", "original": line, "isException": is_exception}
|
|
if "*" in pattern:
|
|
return {"kind": "wildcard", "original": line, "isException": is_exception}
|
|
if looks_like_ip(pattern):
|
|
return {"kind": "ipLiteral", "original": line, "isException": is_exception}
|
|
if "." in pattern and DOMAIN_HOST_RE.fullmatch(pattern):
|
|
return {"kind": "plainText", "original": line, "isException": is_exception}
|
|
raise UnknownSyntax(line)
|
|
|
|
|
|
def parse_url_prefix(url: str, original: str, is_exception: bool) -> dict:
|
|
skipped = {"kind": "urlPrefix", "original": original, "isException": is_exception}
|
|
# urlsplit tolerates whitespace and bare hosts; neither is a literal URL rule.
|
|
if any(char.isspace() for char in url) or not re.match(
|
|
r"^[A-Za-z][A-Za-z0-9+.-]*://", url
|
|
):
|
|
return skipped
|
|
try:
|
|
parsed = urlsplit(url)
|
|
host = parsed.hostname or ""
|
|
_ = parsed.port # Validate numeric syntax and range before dropping it.
|
|
except ValueError:
|
|
return skipped
|
|
if parsed.path not in {"", "/"} or "?" in url or "#" in url or parsed.username is not None:
|
|
return skipped
|
|
if (
|
|
parsed.netloc.endswith(":")
|
|
or not is_valid_domain_host(host)
|
|
or looks_like_ip(host)
|
|
):
|
|
return skipped
|
|
labels = host.split(".")
|
|
if (
|
|
len(host) > 253
|
|
or not all(
|
|
re.fullmatch(r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?", label)
|
|
for label in labels
|
|
)
|
|
or not any(char.isalpha() for char in labels[-1])
|
|
):
|
|
return skipped
|
|
kind = "exactDomainException" if is_exception else "exactDomainProxy"
|
|
return {"kind": kind, "host": host.lower(), "original": original}
|
|
|
|
|
|
def parse_domain_pattern(host_part: str, original: str, is_exception: bool) -> dict:
|
|
if "/" in host_part:
|
|
return {"kind": "urlPath", "original": original, "isException": is_exception}
|
|
if "*" in host_part:
|
|
return {"kind": "wildcard", "original": original, "isException": is_exception}
|
|
if "$" in host_part:
|
|
return {"kind": "filterOption", "original": original, "isException": is_exception}
|
|
|
|
host = host_part
|
|
if host.endswith("^"):
|
|
host = host[:-1]
|
|
if "^" in host:
|
|
raise UnknownSyntax(original)
|
|
if looks_like_ip(host):
|
|
return {"kind": "ipLiteral", "original": original, "isException": is_exception}
|
|
if "." not in host:
|
|
return {
|
|
"kind": "singleLabelPrefix",
|
|
"original": original,
|
|
"isException": is_exception,
|
|
}
|
|
if not is_valid_domain_host(host):
|
|
raise UnknownSyntax(original)
|
|
kind = "domainException" if is_exception else "domainProxy"
|
|
return {"kind": kind, "host": host.lower(), "original": original}
|
|
|
|
|
|
def domain_suffix_covers(broad: str, narrow: str) -> bool:
|
|
broad_host = broad.lower()
|
|
narrow_host = narrow.lower()
|
|
return narrow_host == broad_host or narrow_host.endswith("." + broad_host)
|
|
|
|
|
|
def split_shadowed(
|
|
proxy_rules: list[dict], exception_rules: list[dict]
|
|
) -> tuple[list[dict], list[dict], list[tuple[dict, dict]]]:
|
|
"""Return (kept_proxy, kept_exceptions, shadowed_pairs).
|
|
|
|
A suffix proxy covers exact or suffix exceptions on that domain or subdomains.
|
|
An exact proxy only covers an exact exception on the same domain.
|
|
sslocal matches domain proxy_list before bypass_list, so a covered exception
|
|
would be an ineffective ACL entry.
|
|
"""
|
|
kept_exceptions: list[dict] = []
|
|
shadowed: list[tuple[dict, dict]] = []
|
|
for exception in exception_rules:
|
|
blocker = None
|
|
for proxy in proxy_rules:
|
|
broad, narrow = proxy["match"], exception["match"]
|
|
if (
|
|
broad["kind"] == "domainSuffix"
|
|
and domain_suffix_covers(broad["value"], narrow["value"])
|
|
) or (broad == narrow):
|
|
blocker = proxy
|
|
break
|
|
if blocker is None:
|
|
kept_exceptions.append(exception)
|
|
else:
|
|
shadowed.append((exception, blocker))
|
|
return proxy_rules, kept_exceptions, shadowed
|
|
|
|
|
|
def make_rule(action: str, match: dict) -> dict:
|
|
return {
|
|
"action": action,
|
|
"match": match,
|
|
}
|
|
|
|
|
|
def convert_document(document: str) -> tuple[list[dict], dict]:
|
|
"""Return (rules, lossReport)."""
|
|
skipped: dict[str, int] = {}
|
|
proxy_rules: list[dict] = []
|
|
exception_rules: list[dict] = []
|
|
saw_code_line = False
|
|
originals: dict[tuple[str, str, str], str] = {}
|
|
|
|
def rule_key(rule: dict) -> tuple[str, str, str]:
|
|
return (rule["action"], rule["match"]["kind"], rule["match"]["value"])
|
|
|
|
def bump(category: str) -> None:
|
|
skipped[category] = skipped.get(category, 0) + 1
|
|
|
|
for raw_line in document.splitlines():
|
|
entry = parse_line(raw_line)
|
|
kind = entry["kind"]
|
|
if kind == "blank":
|
|
bump("blank")
|
|
elif kind == "comment":
|
|
bump("comment")
|
|
elif kind == "header":
|
|
bump("header")
|
|
elif kind in {"domainProxy", "domainException", "exactDomainProxy", "exactDomainException"}:
|
|
saw_code_line = True
|
|
action = "direct" if kind.endswith("Exception") else "proxy"
|
|
match_kind = "domainExact" if kind.startswith("exact") else "domainSuffix"
|
|
rule = make_rule(
|
|
action, {"kind": match_kind, "value": entry["host"]}
|
|
)
|
|
originals.setdefault(rule_key(rule), entry["original"])
|
|
(exception_rules if action == "direct" else proxy_rules).append(rule)
|
|
elif kind in {
|
|
"urlPrefix",
|
|
"urlPath",
|
|
"filterOption",
|
|
"wildcard",
|
|
"regexp",
|
|
"plainText",
|
|
"singleLabelPrefix",
|
|
"ipLiteral",
|
|
}:
|
|
saw_code_line = True
|
|
bump(kind)
|
|
print(f"unexpressible-{kind}: {entry['original']}")
|
|
else:
|
|
raise UnknownSyntax(entry.get("original", raw_line))
|
|
|
|
if not saw_code_line:
|
|
raise SystemExit("empty input")
|
|
|
|
kept_proxy, kept_exceptions, shadowed = split_shadowed(proxy_rules, exception_rules)
|
|
if shadowed:
|
|
for exception, blocker in shadowed:
|
|
print(
|
|
"shadowed-exception: "
|
|
f"{originals[rule_key(exception)]} "
|
|
f"shadowed-by {originals[rule_key(blocker)]}"
|
|
)
|
|
|
|
rules = kept_proxy + kept_exceptions
|
|
# Deduplicate by (action, match.kind, match.value), keeping first.
|
|
seen: set[tuple[str, str, str]] = set()
|
|
deduped: list[dict] = []
|
|
for rule in rules:
|
|
key = (rule["action"], rule["match"]["kind"], rule["match"]["value"])
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
deduped.append(rule)
|
|
rules = deduped
|
|
|
|
report = {
|
|
"convertedCount": len(rules),
|
|
"absorbedCount": len(shadowed),
|
|
"skipped": skipped,
|
|
"rejected": {},
|
|
}
|
|
return rules, report
|
|
|
|
|
|
def build_snapshot(
|
|
document: str,
|
|
source: dict,
|
|
upstream_reference: str,
|
|
license_name: str,
|
|
attribution: str,
|
|
previous_rule_count: int | None,
|
|
) -> dict:
|
|
rules, report = convert_document(document)
|
|
rule_count = len(rules)
|
|
if rule_count < MINIMUM_RULE_COUNT:
|
|
raise SystemExit(f"abnormal rule count {rule_count} (min {MINIMUM_RULE_COUNT})")
|
|
if rule_count > MAXIMUM_RULE_COUNT:
|
|
raise SystemExit(f"abnormal rule count {rule_count} (max {MAXIMUM_RULE_COUNT})")
|
|
if previous_rule_count is not None and previous_rule_count > 0:
|
|
lower = int(previous_rule_count * MINIMUM_SCALE_RATIO)
|
|
upper = int(previous_rule_count * MAXIMUM_SCALE_RATIO)
|
|
if not (lower <= rule_count <= upper):
|
|
raise SystemExit(
|
|
f"abnormal scale change: {rule_count} vs previous {previous_rule_count} "
|
|
f"(allowed {lower}..{upper})"
|
|
)
|
|
|
|
return {
|
|
"schemaVersion": SCHEMA_VERSION,
|
|
"metadata": {
|
|
"source": source,
|
|
"upstreamReference": upstream_reference,
|
|
"inputDigest": hashlib.sha256(document.encode("utf-8")).hexdigest(),
|
|
"fetchedAt": dt.datetime.now(dt.timezone.utc)
|
|
.replace(microsecond=0)
|
|
.isoformat()
|
|
.replace("+00:00", "Z"),
|
|
"converterVersion": CONVERTER_VERSION,
|
|
"license": license_name,
|
|
"attribution": attribution,
|
|
},
|
|
"rules": rules,
|
|
"lossReport": report,
|
|
}
|
|
|
|
|
|
def decode_autoproxy(text: str) -> str:
|
|
compact = "".join(text.split())
|
|
try:
|
|
raw = base64.b64decode(compact, validate=False)
|
|
except (binascii.Error, ValueError) as exc:
|
|
raise SystemExit(f"corrupt Base64 input: {exc}") from exc
|
|
try:
|
|
return raw.decode("utf-8")
|
|
except UnicodeDecodeError as exc:
|
|
raise SystemExit(f"corrupt AutoProxy payload: {exc}") from exc
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("input", type=Path, help="path to official Base64 gfwlist.txt")
|
|
parser.add_argument("--upstream", required=True, help="pinned upstream commit")
|
|
parser.add_argument(
|
|
"--upstream-reference",
|
|
required=True,
|
|
help="upstream repo/product identifier including the pinned commit or URL",
|
|
)
|
|
parser.add_argument("--license", required=True)
|
|
parser.add_argument("--attribution", required=True)
|
|
parser.add_argument("--out", type=Path, required=True, help="output snapshot.json")
|
|
parser.add_argument(
|
|
"--previous-count",
|
|
type=int,
|
|
default=None,
|
|
help="previous snapshot rule count for scale-change detection",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
base64_text = args.input.read_text(encoding="utf-8")
|
|
document = decode_autoproxy(base64_text)
|
|
source = {
|
|
"kind": "gfwlist",
|
|
"upstreamVersion": args.upstream,
|
|
"label": "GFWList",
|
|
}
|
|
try:
|
|
snapshot = build_snapshot(
|
|
document,
|
|
source,
|
|
args.upstream_reference,
|
|
args.license,
|
|
args.attribution,
|
|
args.previous_count,
|
|
)
|
|
except UnknownSyntax as exc:
|
|
raise SystemExit(str(exc)) from exc
|
|
|
|
args.out.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = args.out.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(snapshot, separators=(",", ":"), sort_keys=True) + "\n", encoding="utf-8")
|
|
tmp.replace(args.out)
|
|
loss = snapshot["lossReport"]
|
|
print(
|
|
f"wrote {args.out}: {loss['convertedCount']} rules, "
|
|
f"skipped={loss['skipped']}, "
|
|
f"shadowed={loss['absorbedCount']}"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|