dafda56b22
Во всех встроенных пресетах winws2 profile теперь называется zapret.moe и берёт список lists/zapret-moe.txt (в нём домен zapret.moe, под него подходят и git.zapret.moe, и остальные поддомены). Копии пресетов у пользователей со старым путём lists/git-zapret-moe.txt переводятся на новый файл при сохранении и разовым проходом после запуска (LEGACY_LIST_FILE_RENAMES в договоре пресета): старого файла в поставке нет, а без файла hostlist winws2 не стартует. Прежнее имя profile-а git.zapret.moe в таких копиях только узнаётся (stock_names) и показывается как zapret.moe. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
474 lines
22 KiB
Python
474 lines
22 KiB
Python
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
import unittest
|
||
|
||
from core.paths import AppPaths
|
||
from profile.match_filters import strategy_catalog_from_match_lines
|
||
from profile.parser import parse_preset_text
|
||
from profile.derived_cache import (
|
||
basic_strategy_entries,
|
||
normalize_lines,
|
||
profile_list_type,
|
||
profile_strategy_shape,
|
||
resolve_strategy,
|
||
strategy_identity_lines,
|
||
)
|
||
from profile.strategy_catalog import load_strategy_catalogs
|
||
from profile.strategy_shape import composite_identity, is_lua_desync_line, is_range_line, strategy_shape
|
||
|
||
|
||
class ProfileStrategyResolutionTests(unittest.TestCase):
|
||
@classmethod
|
||
def setUpClass(cls) -> None:
|
||
cls.paths = AppPaths(user_root=Path("src").resolve(), local_root=Path("src").resolve())
|
||
cls.catalogs = load_strategy_catalogs(cls.paths, "winws2")
|
||
cls.preset = parse_preset_text(
|
||
Path("src/presets/builtin/winws2/Default v5 (game filter).txt").read_text(encoding="utf-8"),
|
||
engine="winws2",
|
||
source_name="Default v5 (game filter).txt",
|
||
)
|
||
|
||
def test_default_v5_youtube_tcp_strategy_is_detected_by_lua_desync_lines(self) -> None:
|
||
profile = self._profile_by_name("YouTube · сайт и приложение")
|
||
|
||
self.assertIn("--out-range=-d8", profile.strategy.strategy_lines)
|
||
self.assertEqual(self._resolved_strategy_id(profile), "stock_default_v5_11")
|
||
|
||
def test_default_v5_udp_strategy_is_detected_when_payload_is_in_catalog(self) -> None:
|
||
profile = self._profile_by_name("YouTube · быстрый протокол QUIC")
|
||
|
||
self.assertIn("--out-range=-n8", profile.strategy.strategy_lines)
|
||
self.assertIn("--payload=all", profile.strategy.strategy_lines)
|
||
self.assertEqual(self._resolved_strategy_id(profile), "fake_2_n2")
|
||
|
||
def test_default_v5_discord_and_telegram_strategies_are_detected(self) -> None:
|
||
self.assertEqual(self._resolved_strategy_id(self._profile_by_name("Discord · сайт и приложение")), "stock_default_v5_12")
|
||
self.assertEqual(self._resolved_strategy_id(self._profile_by_name("Telegram")), "stock_default_v5_13")
|
||
|
||
def test_duplicate_ready_strategy_args_are_still_detected_as_ready_strategy(self) -> None:
|
||
preset = parse_preset_text(
|
||
"\n".join(
|
||
(
|
||
"--name=YouTube · видео (googlevideo.com)",
|
||
"--filter-tcp=80,443",
|
||
"--hostlist=lists/googlevideo.txt",
|
||
"",
|
||
"--lua-desync=fake:blob=fake_default_tls:tls_mod=rnd,dupsid,sni=www.google.com:repeats=8:tcp_ts=-600000",
|
||
"--lua-desync=multisplit:pos=2:seqovl=681:seqovl_pattern=tls_google:repeats=8:tcp_ts=-600000",
|
||
"",
|
||
)
|
||
),
|
||
engine="winws2",
|
||
)
|
||
|
||
self.assertEqual(self._resolved_strategy_id(preset.profiles[0]), "stock_general_fake_tls_auto_alt3_game_filter_38")
|
||
|
||
def test_voice_l7_profile_uses_voice_catalog_without_list_file(self) -> None:
|
||
for filter_l7 in ("stun", "discord", "stun,discord"):
|
||
with self.subTest(filter_l7=filter_l7):
|
||
preset = parse_preset_text(
|
||
"\n".join(
|
||
(
|
||
"--name=Голосовые звонки/чаты",
|
||
f"--filter-l7={filter_l7}",
|
||
"--payload=stun,discord_ip_discovery",
|
||
"--lua-desync=fake:blob=fake_default_udp",
|
||
"",
|
||
)
|
||
),
|
||
engine="winws2",
|
||
)
|
||
profile = preset.profiles[0]
|
||
entries = basic_strategy_entries(profile, self.catalogs)
|
||
|
||
self.assertEqual(strategy_catalog_from_match_lines(tuple(profile.match.all_lines())), "voice")
|
||
self.assertEqual(profile_list_type(profile), "voice")
|
||
self.assertIn("fake_simple", entries)
|
||
self.assertEqual(resolve_strategy(profile, entries), ("fake_simple", "Fake · простой"))
|
||
|
||
def test_tcp_filter_only_profile_uses_tcp_catalog_without_list_file(self) -> None:
|
||
preset = parse_preset_text(
|
||
"\n".join(
|
||
(
|
||
"--name=Tanki X",
|
||
"--filter-tcp=80,443,5050,8080",
|
||
"--out-range=-n8",
|
||
"--payload=all",
|
||
"--lua-desync=send:repeats=2",
|
||
"--lua-desync=syndata:blob=stun_pat",
|
||
"",
|
||
)
|
||
),
|
||
engine="winws2",
|
||
)
|
||
profile = preset.profiles[0]
|
||
entries = basic_strategy_entries(profile, self.catalogs)
|
||
|
||
self.assertEqual(strategy_catalog_from_match_lines(tuple(profile.match.all_lines())), "tcp")
|
||
self.assertEqual(profile_list_type(profile), "tcp")
|
||
self.assertIn("send_syndata", entries)
|
||
self.assertEqual(resolve_strategy(profile, entries), ("send_syndata", "Send repeats 2 + Syndata · stun"))
|
||
|
||
def test_strategy_catalogs_do_not_contain_duplicate_args(self) -> None:
|
||
for engine in ("winws1", "winws2"):
|
||
catalogs = load_strategy_catalogs(self.paths, engine)
|
||
with self.subTest(engine=engine):
|
||
duplicate_groups: list[str] = []
|
||
for catalog_name, entries in catalogs.items():
|
||
seen: dict[tuple[str, ...], str] = {}
|
||
for strategy_id, entry in entries.items():
|
||
identity = _ready_strategy_identity(engine, entry.args.splitlines(), entry.is_composite)
|
||
previous_id = seen.get(identity)
|
||
if previous_id is not None:
|
||
duplicate_groups.append(f"{catalog_name}: {previous_id} = {strategy_id}")
|
||
continue
|
||
seen[identity] = strategy_id
|
||
self.assertEqual(duplicate_groups, [])
|
||
|
||
def test_winws2_catalogs_contain_only_lua_desync_args(self) -> None:
|
||
# Обычная готовая стратегия — только --lua-desync. --payload и
|
||
# диапазоны допустимы лишь в составной стратегии как разделители
|
||
# веток, причём диапазон — только после первой --lua-desync
|
||
# (диапазон до неё — настройка profile-а).
|
||
invalid_lines: list[str] = []
|
||
for path in sorted(Path("src/system/strategy_catalogs/winws2").glob("*.txt")):
|
||
for strategy_id, lines in _raw_catalog_entries(path):
|
||
other = [line for line in lines if not is_lua_desync_line(line)]
|
||
if not other:
|
||
continue
|
||
first_lua = next((index for index, line in enumerate(lines) if is_lua_desync_line(line)), len(lines))
|
||
shape = strategy_shape(lines)
|
||
bad = [
|
||
line
|
||
for index, line in enumerate(lines)
|
||
if not is_lua_desync_line(line)
|
||
and (
|
||
not shape.composite
|
||
or not (line.lower().startswith("--payload=") or is_range_line(line))
|
||
or (is_range_line(line) and index < first_lua)
|
||
)
|
||
]
|
||
invalid_lines.extend(f"{path.name}:[{strategy_id}]: {line}" for line in bad)
|
||
|
||
self.assertEqual(invalid_lines, [])
|
||
|
||
def test_builtin_winws2_preset_strategies_are_known(self) -> None:
|
||
unresolved: list[str] = []
|
||
for path in sorted(Path("src/presets/builtin/winws2").glob("*.txt")):
|
||
if "(circular)" in path.stem.lower():
|
||
continue
|
||
preset = parse_preset_text(path.read_text(encoding="utf-8"), engine="winws2", source_name=path.name)
|
||
for index, profile in enumerate(preset.profiles, start=1):
|
||
strategy_lines = strategy_identity_lines(profile, profile.strategy.strategy_lines)
|
||
if any("circular:" in line.lower() for line in strategy_lines):
|
||
continue
|
||
if not strategy_lines:
|
||
continue
|
||
entries = basic_strategy_entries(profile, self.catalogs)
|
||
strategy_id, _strategy_name = resolve_strategy(profile, entries)
|
||
if strategy_id != "custom":
|
||
continue
|
||
catalog = strategy_catalog_from_match_lines(tuple(profile.match.all_lines()))
|
||
composite = profile_strategy_shape(profile).composite
|
||
unresolved.append(
|
||
f"{path.name}:{index}: {profile.display_name} "
|
||
f"({catalog}/{profile_list_type(profile)}; composite={composite})"
|
||
)
|
||
|
||
self.assertEqual(unresolved, [])
|
||
|
||
def test_every_builtin_preset_has_domain_and_ip_ru_exclusions_with_pass(self) -> None:
|
||
expected_profiles = {
|
||
"Исключения домены (RU сайты)": (
|
||
"--filter-tcp=80,443-65535",
|
||
"--hostlist=lists/netrogat.txt",
|
||
),
|
||
"Исключения айпи (RU сайты)": (
|
||
"--filter-tcp=80,443-65535",
|
||
"--ipset=lists/ipset-ru.txt",
|
||
"--ipset=lists/ipset-dns.txt",
|
||
"--ipset=lists/ipset-exclude.txt",
|
||
),
|
||
}
|
||
expected_strategy = (
|
||
"--out-range=-d8",
|
||
"--payload=tls_client_hello",
|
||
"--lua-desync=pass",
|
||
)
|
||
preset_paths = sorted(Path("src/presets/builtin/winws2").glob("*.txt"))
|
||
|
||
self.assertGreater(len(preset_paths), 0)
|
||
for path in preset_paths:
|
||
with self.subTest(preset=path.name):
|
||
text = path.read_text(encoding="utf-8")
|
||
preset = parse_preset_text(text, engine="winws2", source_name=path.name)
|
||
exclusion_profiles = [
|
||
profile
|
||
for profile in preset.profiles
|
||
if profile.display_name in expected_profiles
|
||
]
|
||
|
||
self.assertEqual(
|
||
[profile.display_name for profile in exclusion_profiles],
|
||
list(expected_profiles),
|
||
)
|
||
self.assertNotIn("--name=Исключения (RU сайты)", text)
|
||
# У пресета свой номер: он растёт при каждой правке именно этого файла.
|
||
self.assertGreaterEqual(_builtin_version(text), (2, 42))
|
||
|
||
for profile in exclusion_profiles:
|
||
self.assertEqual(
|
||
tuple(profile.match.all_lines()),
|
||
expected_profiles[profile.display_name],
|
||
)
|
||
self.assertEqual(
|
||
normalize_lines(profile.strategy.strategy_lines),
|
||
expected_strategy,
|
||
)
|
||
self.assertEqual(self._resolved_strategy_id(profile), "pass")
|
||
|
||
def test_zapret_moe_is_first_and_uses_alt9_in_every_builtin_preset(self) -> None:
|
||
expected_strategy = (
|
||
"--out-range=-d8",
|
||
"--lua-desync=hostfakesplit:host=ozon.ru:tcp_ts=-1000:tcp_md5:repeats=4",
|
||
)
|
||
preset_paths = sorted(Path("src/presets/builtin/winws2").glob("*.txt"))
|
||
|
||
self.assertGreater(len(preset_paths), 0)
|
||
for path in preset_paths:
|
||
with self.subTest(preset=path.name):
|
||
preset = parse_preset_text(
|
||
path.read_text(encoding="utf-8"),
|
||
engine="winws2",
|
||
source_name=path.name,
|
||
)
|
||
matches = [profile for profile in preset.profiles if profile.display_name == "zapret.moe"]
|
||
self.assertEqual(len(matches), 1)
|
||
profile = matches[0]
|
||
self.assertIs(profile, preset.profiles[0])
|
||
self.assertEqual(profile.match.filter_lines, ["--filter-tcp=443"])
|
||
self.assertEqual(profile.match.hostlist_lines, ["--hostlist=lists/zapret-moe.txt"])
|
||
self.assertEqual(normalize_lines(profile.strategy.strategy_lines), expected_strategy)
|
||
self.assertEqual(self._resolved_strategy_id(profile), "alt9")
|
||
|
||
def test_youtube_profiles_precede_ru_pass_in_every_builtin_preset(self) -> None:
|
||
# AS12389 остаётся в ipset-ru: внутри него могут находиться российские
|
||
# кэш-серверы Google/YouTube. Поэтому специальные YouTube-профили должны
|
||
# проверяться раньше широкого RU-профиля с действием pass.
|
||
offenders: list[str] = []
|
||
checked_presets = 0
|
||
|
||
for path in sorted(Path("src/presets/builtin/winws2").glob("*.txt")):
|
||
preset = parse_preset_text(
|
||
path.read_text(encoding="utf-8"),
|
||
engine="winws2",
|
||
source_name=path.name,
|
||
)
|
||
youtube_indexes = [
|
||
index
|
||
for index, profile in enumerate(preset.profiles)
|
||
if _is_youtube_profile(profile)
|
||
]
|
||
ru_pass_indexes = [
|
||
index
|
||
for index, profile in enumerate(preset.profiles)
|
||
if "--ipset=lists/ipset-ru.txt" in profile.match.all_lines()
|
||
]
|
||
if not youtube_indexes or not ru_pass_indexes:
|
||
continue
|
||
|
||
checked_presets += 1
|
||
if max(youtube_indexes) >= min(ru_pass_indexes):
|
||
offenders.append(
|
||
f"{path.name}: YouTube {youtube_indexes}, RU pass {ru_pass_indexes}"
|
||
)
|
||
|
||
if _builtin_version(path.read_text(encoding="utf-8")) < (2, 42):
|
||
offenders.append(f"{path.name}: версия пресета ниже 2.42")
|
||
|
||
self.assertGreater(checked_presets, 0)
|
||
self.assertEqual(offenders, [])
|
||
|
||
def test_flowseal_exp_1100_payload_branches_are_one_composite_strategy(self) -> None:
|
||
path = Path("src/presets/builtin/winws2/general EXP 1.10.0 (game filter).txt")
|
||
preset = parse_preset_text(path.read_text(encoding="utf-8"), engine="winws2", source_name=path.name)
|
||
|
||
media = next(profile for profile in preset.profiles if profile.display_name == "Discord · голос и видео (discord.media)")
|
||
self.assertEqual(self._resolved_strategy_id(media), "flowseal_exp_1100_discord_media")
|
||
self.assertEqual(
|
||
profile_strategy_shape(media).payload_scopes,
|
||
("tls_client_hello", "http_req", "tls_client_hello,http_req"),
|
||
)
|
||
|
||
steam = next(profile for profile in preset.profiles if profile.display_name == "Steam")
|
||
steam_shape = profile_strategy_shape(steam)
|
||
self.assertEqual(self._resolved_strategy_id(steam), "flowseal_exp_1100_game_tcp")
|
||
# --out-range=<n4 перед первой --lua-desync — настройка profile-а,
|
||
# а не часть готовой стратегии.
|
||
self.assertEqual(steam_shape.profile_range_lines, ("--out-range=<n4",))
|
||
self.assertEqual(steam_shape.payload_scopes, ("tls_client_hello", "http_req", "all"))
|
||
entry = self.catalogs["tcp"]["flowseal_exp_1100_game_tcp"]
|
||
self.assertTrue(entry.is_composite)
|
||
self.assertNotIn("--out-range=<n4", entry.args)
|
||
|
||
def test_new_199_builtin_presets_do_not_mix_payload_scoped_strategies(self) -> None:
|
||
violations: list[str] = []
|
||
for path in sorted(Path("src/presets/builtin/winws2").glob("*1.9.9*.txt")):
|
||
preset = parse_preset_text(path.read_text(encoding="utf-8"), engine="winws2", source_name=path.name)
|
||
for profile in preset.profiles:
|
||
payload_groups = _payload_groups_with_lua(profile.strategy.strategy_lines)
|
||
if payload_groups > 1:
|
||
violations.append(f"{path.name}: {profile.display_name}: {payload_groups}")
|
||
|
||
self.assertEqual(violations, [])
|
||
|
||
def test_new_199_builtin_presets_use_payload_all_only(self) -> None:
|
||
violations: list[str] = []
|
||
for path in sorted(Path("src/presets/builtin/winws2").glob("*1.9.9*.txt")):
|
||
preset = parse_preset_text(path.read_text(encoding="utf-8"), engine="winws2", source_name=path.name)
|
||
for profile in preset.profiles:
|
||
lines = normalize_lines(profile.strategy.strategy_lines)
|
||
lua_lines = [line for line in lines if line.lower().startswith("--lua-desync=")]
|
||
if lua_lines and all(line.lower() == "--lua-desync=pass" for line in lua_lines):
|
||
# Профили-исключения (pass-through) намеренно сужают no-op
|
||
# до tls_client_hello и не обязаны использовать payload=all.
|
||
continue
|
||
for line in lines:
|
||
if line.lower().startswith("--payload=") and line != "--payload=all":
|
||
violations.append(f"{path.name}: {profile.display_name}: {line}")
|
||
|
||
self.assertEqual(violations, [])
|
||
|
||
def test_new_199_builtin_presets_do_not_add_inner_payload_to_fake_instances(self) -> None:
|
||
violations: list[str] = []
|
||
for path in sorted(Path("src/presets/builtin/winws2").glob("*1.9.9*.txt")):
|
||
preset = parse_preset_text(path.read_text(encoding="utf-8"), engine="winws2", source_name=path.name)
|
||
for profile in preset.profiles:
|
||
violations.extend(
|
||
f"{path.name}: {profile.display_name}: {line}"
|
||
for line in normalize_lines(profile.strategy.strategy_lines)
|
||
if line.startswith("--lua-desync=fake:") and ":payload=" in line
|
||
)
|
||
|
||
self.assertEqual(violations, [])
|
||
|
||
def test_new_199_alt4_telegram_keeps_two_tls_fakes_and_http_fake_under_payload_all(self) -> None:
|
||
preset = parse_preset_text(
|
||
Path("src/presets/builtin/winws2/general ALT4 1.9.9 (game filter).txt").read_text(encoding="utf-8"),
|
||
engine="winws2",
|
||
source_name="general ALT4 1.9.9 (game filter).txt",
|
||
)
|
||
profile = next(profile for profile in preset.profiles if profile.display_name == "Telegram")
|
||
|
||
self.assertEqual(
|
||
normalize_lines(profile.strategy.strategy_lines),
|
||
(
|
||
"--payload=all",
|
||
"--lua-desync=fake:blob=fake_stun_as_tls:tcp_seq=1000:repeats=6",
|
||
"--lua-desync=fake:blob=tls_google:tcp_seq=1000:repeats=6",
|
||
"--lua-desync=fake:blob=fake_http_max:tcp_seq=1000:repeats=6",
|
||
"--lua-desync=multisplit:pos=2",
|
||
),
|
||
)
|
||
|
||
def test_new_199_alt12_voice_keeps_two_discord_fakes_under_payload_all(self) -> None:
|
||
preset = parse_preset_text(
|
||
Path("src/presets/builtin/winws2/general ALT12 1.9.9 (game filter).txt").read_text(encoding="utf-8"),
|
||
engine="winws2",
|
||
source_name="general ALT12 1.9.9 (game filter).txt",
|
||
)
|
||
profile = next(profile for profile in preset.profiles if profile.display_name == "Голосовые звонки/чаты")
|
||
|
||
self.assertEqual(
|
||
normalize_lines(profile.strategy.strategy_lines),
|
||
(
|
||
"--payload=all",
|
||
"--lua-desync=fake:blob=fake_stun:repeats=3",
|
||
"--lua-desync=fake:blob=fake_dbankcloud:repeats=3",
|
||
),
|
||
)
|
||
|
||
def test_catalog_entry_contains_strategy_visual_description(self) -> None:
|
||
entry = self.catalogs["tcp"]["stock_default_v5_11"]
|
||
|
||
self.assertEqual(entry.visual.technique_keys, ("fake", "multidisorder"))
|
||
self.assertEqual(entry.visual.label, "Fake + MultiDisorder")
|
||
|
||
def _resolved_strategy_id(self, profile) -> str:
|
||
catalog = strategy_catalog_from_match_lines(tuple(profile.match.all_lines()))
|
||
self.assertIn(catalog, self.catalogs)
|
||
strategy_id, _strategy_name = resolve_strategy(profile, basic_strategy_entries(profile, self.catalogs))
|
||
return strategy_id
|
||
|
||
def _profile_by_name(self, display_name: str):
|
||
return next(profile for profile in self.preset.profiles if profile.display_name == display_name)
|
||
|
||
|
||
def _builtin_version(text: str) -> tuple[int, ...]:
|
||
for line in text.splitlines()[:5]:
|
||
if line.startswith("# BuiltinVersion: "):
|
||
try:
|
||
return tuple(int(part) for part in line.removeprefix("# BuiltinVersion: ").split("."))
|
||
except ValueError:
|
||
return ()
|
||
return ()
|
||
|
||
|
||
def _ready_strategy_identity(engine: str, lines, composite: bool = False) -> tuple:
|
||
normalized = normalize_lines(lines)
|
||
if engine == "winws2":
|
||
# Составные стратегии сравниваются целиком и только между собой,
|
||
# обычные — по строкам --lua-desync.
|
||
if composite:
|
||
return ("composite", composite_identity(strategy_shape(normalized).body_lines))
|
||
return tuple(line for line in normalized if line.lower().startswith("--lua-desync="))
|
||
return normalized
|
||
|
||
|
||
def _raw_catalog_entries(path: Path) -> list[tuple[str, list[str]]]:
|
||
entries: list[tuple[str, list[str]]] = []
|
||
for raw in path.read_text(encoding="utf-8").splitlines():
|
||
line = raw.strip()
|
||
if line.startswith("[") and line.endswith("]"):
|
||
entries.append((line[1:-1].strip(), []))
|
||
continue
|
||
if entries and line.startswith("--"):
|
||
entries[-1][1].append(line)
|
||
return entries
|
||
|
||
|
||
def _is_youtube_profile(profile) -> bool:
|
||
identity = "\n".join(
|
||
(
|
||
str(profile.display_name or ""),
|
||
*profile.match.all_lines(),
|
||
)
|
||
).lower()
|
||
return any(
|
||
token in identity
|
||
for token in ("youtube", "googlevideo", "i.ytimg", "i-ytimg")
|
||
)
|
||
|
||
|
||
def _payload_groups_with_lua(lines) -> int:
|
||
payload_groups = 0
|
||
in_payload = False
|
||
has_lua = False
|
||
for line in normalize_lines(lines):
|
||
lowered = line.lower()
|
||
if lowered.startswith("--payload="):
|
||
if in_payload and has_lua:
|
||
payload_groups += 1
|
||
in_payload = True
|
||
has_lua = False
|
||
continue
|
||
if lowered.startswith("--lua-desync=") and in_payload:
|
||
has_lua = True
|
||
if in_payload and has_lua:
|
||
payload_groups += 1
|
||
return payload_groups
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|