395 lines
18 KiB
Python
395 lines
18 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,
|
||
resolve_strategy,
|
||
strategy_branches_for_profile,
|
||
strategy_identity_lines,
|
||
)
|
||
from profile.strategy_catalog import load_strategy_catalogs
|
||
|
||
|
||
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.com (интерфейс)")
|
||
|
||
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.com (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.com")), "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=googlevideo.com (CDN сервера)",
|
||
"--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())
|
||
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:
|
||
invalid_lines: list[str] = []
|
||
for path in sorted(Path("src/system/strategy_catalogs/winws2").glob("*.txt")):
|
||
invalid_lines.extend(
|
||
f"{path.name}:{line_number}: {line.strip()}"
|
||
for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1)
|
||
if line.strip().startswith("--") and not line.strip().startswith("--lua-desync=")
|
||
)
|
||
|
||
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
|
||
branches = strategy_branches_for_profile(profile, entries)
|
||
custom_branches = [branch for branch in branches if branch.strategy_id == "custom"]
|
||
if custom_branches:
|
||
catalog = strategy_catalog_from_match_lines(tuple(profile.match.all_lines()))
|
||
unresolved.append(
|
||
f"{path.name}:{index}: {profile.display_name} "
|
||
f"({catalog}/{profile_list_type(profile)}; custom branches={len(custom_branches)})"
|
||
)
|
||
|
||
self.assertEqual(unresolved, [])
|
||
|
||
def test_git_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 == "git.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/git-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}"
|
||
)
|
||
|
||
version_line = next(
|
||
(
|
||
line
|
||
for line in path.read_text(encoding="utf-8").splitlines()[:5]
|
||
if line.startswith("# BuiltinVersion: ")
|
||
),
|
||
"",
|
||
)
|
||
if version_line != "# BuiltinVersion: 2.39":
|
||
offenders.append(f"{path.name}: версия набора не 2.39")
|
||
|
||
self.assertGreater(checked_presets, 0)
|
||
self.assertEqual(offenders, [])
|
||
|
||
def test_flowseal_exp_1100_keeps_payload_scopes_and_ready_branches(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.media (voice RTC)")
|
||
media_branches = strategy_branches_for_profile(media, basic_strategy_entries(media, self.catalogs))
|
||
self.assertEqual(
|
||
[(branch.payload, branch.strategy_id) for branch in media_branches],
|
||
[
|
||
("tls_client_hello", "flowseal_exp_1100_discord_tls"),
|
||
("http_req", "flowseal_exp_1100_discord_http"),
|
||
("tls_client_hello,http_req", "general_alt6_184"),
|
||
],
|
||
)
|
||
|
||
steam = next(profile for profile in preset.profiles if profile.display_name == "Steam")
|
||
steam_branches = strategy_branches_for_profile(steam, basic_strategy_entries(steam, self.catalogs))
|
||
self.assertEqual(
|
||
[(branch.out_range, branch.payload, branch.strategy_id) for branch in steam_branches],
|
||
[
|
||
("<n4", "tls_client_hello", "flowseal_exp_1100_game_tls"),
|
||
("<n4", "http_req", "flowseal_exp_1100_game_http"),
|
||
("<n4", "all", "flowseal_exp_1100_game_other"),
|
||
],
|
||
)
|
||
|
||
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 _ready_strategy_identity(engine: str, lines) -> tuple[str, ...]:
|
||
normalized = normalize_lines(lines)
|
||
if engine == "winws2":
|
||
return tuple(line for line in normalized if line.lower().startswith("--lua-desync="))
|
||
return normalized
|
||
|
||
|
||
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()
|