loop-uh
610e389bb5
Some checks failed
Windows project source guards / test (push) Has been cancelled
Remnawave и совместимые панели сообщают об отозванной подписке, лимите устройств или неподдерживаемом клиенте обычным HTTP 200 с узлами 0.0.0.0:1, а текст раскладывают по их именам. Если все узлы такие, импорт теперь падает с «Провайдер сообщает: …», и рабочий список серверов не затирается. Информационный узел рядом с настоящими серверами по-прежнему импортируется.
945 lines
42 KiB
Python
945 lines
42 KiB
Python
from __future__ import annotations
|
||
|
||
import base64
|
||
from datetime import datetime, timedelta, timezone
|
||
import json
|
||
from pathlib import Path
|
||
import sys
|
||
import tempfile
|
||
from types import SimpleNamespace
|
||
import unittest
|
||
from urllib.parse import quote
|
||
from unittest.mock import patch
|
||
import urllib.error
|
||
import urllib.request
|
||
import zipfile
|
||
|
||
from PyQt6.QtGui import QImage
|
||
|
||
from xray_fluent.application.subscription_service import (
|
||
apply_not_modified,
|
||
hide_subscription_node,
|
||
mark_subscription_failure,
|
||
reconcile_subscription,
|
||
remove_subscription,
|
||
subscription_due,
|
||
)
|
||
from xray_fluent.constants import (
|
||
STATE_SCHEMA_VERSION,
|
||
SUBSCRIPTION_PARSER_REVISION,
|
||
)
|
||
from xray_fluent.diagnostics.export import export_diagnostics
|
||
from xray_fluent.profiles.models import (
|
||
AppState,
|
||
Node,
|
||
Subscription,
|
||
normalize_subscription_warnings,
|
||
)
|
||
from xray_fluent.importer.qr_utils import QrDecodeError, decode_subscription_qr
|
||
from xray_fluent.profiles.storage import StateStorage
|
||
from xray_fluent.importer.link_parser import LinkParseError, parse_single
|
||
from xray_fluent.importer.subscription_http import (
|
||
SubscriptionFetchError,
|
||
SubscriptionFetchResult,
|
||
_SafeRedirectHandler,
|
||
_device_locale,
|
||
_short_locale,
|
||
_windows_locale_code,
|
||
default_subscription_user_agent,
|
||
describe_http_failure,
|
||
fetch_subscription,
|
||
is_panel_compatible_hwid,
|
||
mask_subscription_url,
|
||
resolve_subscription_source,
|
||
sanitize_fetch_error,
|
||
subscription_request_headers,
|
||
)
|
||
from xray_fluent.importer.subscription_parser import (
|
||
SubscriptionParseError,
|
||
parse_subscription_payload,
|
||
validate_filter_patterns,
|
||
)
|
||
|
||
|
||
VLESS_A = "vless://11111111-1111-1111-1111-111111111111@a.example:443?security=tls#Same"
|
||
VLESS_B = "vless://22222222-2222-2222-2222-222222222222@b.example:443?security=tls#Same"
|
||
|
||
|
||
class SubscriptionParserTests(unittest.TestCase):
|
||
def test_mixed_hysteria_subscription_keeps_all_gecko_nodes(self) -> None:
|
||
salamander = [
|
||
f"hy2://auth-{index}@s{index}.example:443/?obfs=salamander&"
|
||
f"obfs-password=cover-{index}#Salamander-{index}"
|
||
for index in range(38)
|
||
]
|
||
gecko = [
|
||
f"hy2://auth-g{index}@g{index}.example:443/?obfs=gecko&"
|
||
f"obfs-password=cover-g{index}#Gecko-{index}"
|
||
for index in range(4)
|
||
]
|
||
|
||
parsed = parse_subscription_payload("\n".join(salamander + gecko))
|
||
|
||
self.assertEqual(len(parsed.nodes), 42)
|
||
self.assertEqual(parsed.skipped, 0)
|
||
self.assertEqual(parsed.warnings, [])
|
||
self.assertEqual(
|
||
sum((node.outbound.get("obfs") or {}).get("type") == "gecko" for node in parsed.nodes),
|
||
4,
|
||
)
|
||
|
||
def test_all_supported_link_protocols_and_wireguard_config(self) -> None:
|
||
vmess_json = json.dumps(
|
||
{
|
||
"v": "2",
|
||
"ps": "VMess",
|
||
"add": "vmess.example",
|
||
"port": "443",
|
||
"id": "33333333-3333-3333-3333-333333333333",
|
||
"aid": "0",
|
||
"net": "tcp",
|
||
"tls": "tls",
|
||
},
|
||
separators=(",", ":"),
|
||
)
|
||
vmess = "vmess://" + base64.b64encode(vmess_json.encode()).decode()
|
||
links = [
|
||
VLESS_A,
|
||
vmess,
|
||
"trojan://secret@trojan.example:443#Trojan",
|
||
"ss://YWVzLTI1Ni1nY206c2VjcmV0@ss.example:8388#SS",
|
||
"hysteria://hy.example:8443/?auth=secret&upmbps=50&downmbps=100#HY",
|
||
"hy2://secret@hy2.example:443/?insecure=1&pinSHA256=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa#HY2",
|
||
"tuic://44444444-4444-4444-4444-444444444444:secret@tuic.example:443#TUIC",
|
||
"socks://user:pass@socks.example:1080#SOCKS",
|
||
"http://user:pass@http.example:8080#HTTP",
|
||
]
|
||
parsed = parse_subscription_payload("\n".join(links))
|
||
self.assertEqual(
|
||
{node.scheme for node in parsed.nodes},
|
||
{"vless", "vmess", "trojan", "ss", "hysteria", "hysteria2", "tuic", "socks", "http"},
|
||
)
|
||
|
||
wireguard = """# WireGuard
|
||
[Interface]
|
||
PrivateKey = AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=
|
||
Address = 10.0.0.2/32
|
||
|
||
[Peer]
|
||
PublicKey = AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=
|
||
Endpoint = wg.example:51820
|
||
AllowedIPs = 0.0.0.0/0
|
||
"""
|
||
wg_parsed = parse_subscription_payload(wireguard)
|
||
self.assertEqual(wg_parsed.nodes[0].scheme, "wireguard")
|
||
awg_parsed = parse_subscription_payload(wireguard.replace("Address =", "Jc = 4\nAddress ="))
|
||
self.assertEqual(awg_parsed.nodes[0].scheme, "awg")
|
||
|
||
def test_raw_bom_crlf_metadata_partial_errors_and_duplicate_names(self) -> None:
|
||
payload = (
|
||
"\ufeff# profile-title: Demo Provider\r\n"
|
||
"# profile-update-interval: 6\r\n"
|
||
"# subscription-userinfo: upload=10; download=20; totl=100; expire=200\r\n"
|
||
"# profile-web-page-url: https://cabinet.example/user?token=secret\r\n"
|
||
f"{VLESS_A}\r\nnot-a-link\r\n{VLESS_B}\r\n"
|
||
)
|
||
|
||
parsed = parse_subscription_payload(payload, source_url="https://sub.example/list")
|
||
|
||
self.assertEqual([node.provider_name for node in parsed.nodes], ["Same", "Same"])
|
||
self.assertEqual(len({node.source_key for node in parsed.nodes}), 2)
|
||
self.assertEqual(parsed.metadata.title, "Demo Provider")
|
||
self.assertEqual(parsed.metadata.provider_interval_hours, 6)
|
||
self.assertEqual(parsed.metadata.info.used, 30)
|
||
self.assertEqual(parsed.metadata.info.total, 100)
|
||
self.assertEqual(parsed.skipped, 1)
|
||
self.assertTrue(parsed.warnings)
|
||
|
||
def test_provider_stub_nodes_become_the_provider_message(self) -> None:
|
||
stub = "vless://00000000-0000-0000-0000-000000000000@0.0.0.0:1?encryption=none&type=tcp&security=none#"
|
||
body = "\n".join(
|
||
stub + quote(name)
|
||
for name in ("❌ Подписка сломана", "получите новую", "Добавить устройство")
|
||
)
|
||
encoded = base64.b64encode(body.encode()).decode()
|
||
with self.assertRaisesRegex(
|
||
SubscriptionParseError,
|
||
"Провайдер сообщает: ❌ Подписка сломана получите новую Добавить устройство",
|
||
):
|
||
parse_subscription_payload(encoded)
|
||
|
||
# Информационный узел рядом с настоящими серверами — не заглушка.
|
||
mixed = parse_subscription_payload(f"{VLESS_A}\n{stub}{quote('Трафик: 10 ГБ')}")
|
||
self.assertEqual(len(mixed.nodes), 2)
|
||
|
||
def test_once_encoded_urlsafe_base64_and_header_priority(self) -> None:
|
||
encoded = base64.urlsafe_b64encode(VLESS_A.encode()).decode().rstrip("=")
|
||
parsed = parse_subscription_payload(
|
||
encoded,
|
||
headers={"Profile-Title": "Header title"},
|
||
source_url="https://example.com/file.txt#fragment-title",
|
||
)
|
||
self.assertEqual(len(parsed.nodes), 1)
|
||
self.assertEqual(parsed.metadata.title, "Header title")
|
||
|
||
def test_filters_are_case_insensitive_and_hidden_keys_are_stable(self) -> None:
|
||
both = parse_subscription_payload(f"{VLESS_A}\n{VLESS_B}")
|
||
hidden = both.nodes[0].source_key
|
||
parsed = parse_subscription_payload(
|
||
f"{VLESS_A}\n{VLESS_B}",
|
||
include_pattern="same",
|
||
hidden_source_keys={hidden},
|
||
)
|
||
self.assertEqual([node.server for node in parsed.nodes], ["b.example"])
|
||
with self.assertRaises(SubscriptionParseError):
|
||
validate_filter_patterns("(", "")
|
||
|
||
def test_exact_duplicates_are_removed_but_same_names_are_allowed(self) -> None:
|
||
parsed = parse_subscription_payload(f"{VLESS_A}\n{VLESS_A}\n{VLESS_B}")
|
||
self.assertEqual(len(parsed.nodes), 2)
|
||
self.assertEqual(parsed.skipped, 1)
|
||
|
||
renamed_same_outbound = VLESS_A.rsplit("#", 1)[0] + "#Another name"
|
||
aliases = parse_subscription_payload(f"{VLESS_A}\n{renamed_same_outbound}")
|
||
self.assertEqual(len(aliases.nodes), 2)
|
||
self.assertEqual(aliases.nodes[0].source_key, aliases.nodes[1].source_key)
|
||
|
||
def test_hysteria2_uri_fingerprint_prevents_distinct_pins_from_collapsing(self) -> None:
|
||
first = "hy2://secret@h.example:443/?pinSHA256=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa#Same"
|
||
second = "hy2://secret@h.example:443/?pinSHA256=bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb#Same"
|
||
|
||
parsed = parse_subscription_payload(f"{first}\n{second}\n{first}")
|
||
|
||
self.assertEqual(len(parsed.nodes), 2)
|
||
self.assertEqual(parsed.skipped, 1)
|
||
self.assertEqual(parsed.nodes[0].source_key, parsed.nodes[1].source_key)
|
||
self.assertNotEqual(
|
||
parsed.nodes[0].source_fingerprint,
|
||
parsed.nodes[1].source_fingerprint,
|
||
)
|
||
|
||
def test_full_json_extracts_only_independent_supported_proxies(self) -> None:
|
||
document = {
|
||
"dns": {"servers": ["1.1.1.1"]},
|
||
"routing": {"rules": [{"type": "field"}]},
|
||
"inbounds": [{"protocol": "socks"}],
|
||
"outbounds": [
|
||
{"protocol": "freedom", "tag": "direct"},
|
||
{
|
||
"protocol": "trojan",
|
||
"tag": "Xray proxy",
|
||
"settings": {"servers": [{"address": "x.example", "port": 443, "password": "p"}]},
|
||
},
|
||
{
|
||
"type": "hysteria2",
|
||
"tag": "Dependent",
|
||
"server": "d.example",
|
||
"server_port": 443,
|
||
"password": "p",
|
||
"detour": "other",
|
||
},
|
||
{"type": "selector", "tag": "select", "outbounds": ["proxy"]},
|
||
{"type": "mystery", "tag": "unknown", "server": "u.example", "server_port": 1},
|
||
],
|
||
"endpoints": [
|
||
{
|
||
"type": "wireguard",
|
||
"tag": "WG",
|
||
"private_key": "private",
|
||
"peers": [{"address": "wg.example", "port": 51820, "public_key": "public"}],
|
||
}
|
||
],
|
||
}
|
||
|
||
parsed = parse_subscription_payload(json.dumps(document))
|
||
|
||
self.assertEqual({node.provider_name for node in parsed.nodes}, {"Xray proxy", "WG"})
|
||
self.assertGreaterEqual(parsed.skipped, 3)
|
||
self.assertNotIn("dns", {node.scheme for node in parsed.nodes})
|
||
self.assertTrue(any("неподдерживаемый" in warning for warning in parsed.warnings))
|
||
|
||
def test_empty_or_over_limit_payload_is_rejected(self) -> None:
|
||
with self.assertRaises(SubscriptionParseError):
|
||
parse_subscription_payload("bad data only")
|
||
with self.assertRaises(SubscriptionParseError):
|
||
parse_subscription_payload(f"{VLESS_A}\n{VLESS_B}", max_nodes=1)
|
||
|
||
def test_xray_wireguard_endpoint_is_validated_and_extracted(self) -> None:
|
||
outbound = {
|
||
"protocol": "wireguard",
|
||
"tag": "Xray WG",
|
||
"settings": {
|
||
"secretKey": "private",
|
||
"peers": [
|
||
{
|
||
"endpoint": "[2001:db8::1]:51820",
|
||
"publicKey": "public",
|
||
}
|
||
],
|
||
},
|
||
}
|
||
parsed = parse_subscription_payload(json.dumps(outbound))
|
||
self.assertEqual((parsed.nodes[0].server, parsed.nodes[0].port), ("2001:db8::1", 51820))
|
||
|
||
outbound["settings"]["peers"][0].pop("publicKey")
|
||
with self.assertRaises(SubscriptionParseError):
|
||
parse_subscription_payload(json.dumps(outbound))
|
||
|
||
|
||
class SubscriptionReconcileTests(unittest.TestCase):
|
||
def _fetch(self, **headers: str) -> SubscriptionFetchResult:
|
||
return SubscriptionFetchResult(headers=headers)
|
||
|
||
def test_reconcile_preserves_id_local_fields_history_and_selection(self) -> None:
|
||
subscription = Subscription(id="sub", name="Provider", url="https://sub.example/a")
|
||
first = parse_subscription_payload(VLESS_A)
|
||
state = AppState(subscriptions=[subscription])
|
||
reconcile_subscription(state, subscription, first, self._fetch())
|
||
old = state.nodes[0]
|
||
old.id = "stable-id"
|
||
old.name = "My local name"
|
||
old.group = "My group"
|
||
old.tags = ["favorite"]
|
||
old.ping_ms = 25
|
||
old.speed_mbps = 10.5
|
||
old.ping_history = [("now", 25)]
|
||
state.selected_node_id = old.id
|
||
|
||
changed_transport = VLESS_A.replace("security=tls", "security=tls&type=ws&path=%2Fws")
|
||
second = parse_subscription_payload(changed_transport)
|
||
outcome = reconcile_subscription(state, subscription, second, self._fetch())
|
||
current = state.nodes[0]
|
||
|
||
self.assertEqual(current.id, "stable-id")
|
||
self.assertEqual(current.name, "My local name")
|
||
self.assertEqual(current.group, "My group")
|
||
self.assertEqual(current.tags, ["favorite"])
|
||
self.assertEqual(current.ping_history, [("now", 25)])
|
||
self.assertEqual(state.selected_node_id, "stable-id")
|
||
self.assertTrue(outcome.selected_node_changed)
|
||
self.assertTrue(outcome.result.reconnect_required)
|
||
|
||
def test_hysteria2_uri_only_change_reconnects_but_fragment_change_does_not(self) -> None:
|
||
subscription = Subscription(id="sub", name="Provider")
|
||
original = "hy2://secret@h.example:443/?pinSHA256=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa#Old"
|
||
state = AppState(subscriptions=[subscription])
|
||
reconcile_subscription(
|
||
state,
|
||
subscription,
|
||
parse_subscription_payload(original),
|
||
self._fetch(),
|
||
)
|
||
stable_id = state.nodes[0].id
|
||
state.selected_node_id = stable_id
|
||
|
||
fragment_only = reconcile_subscription(
|
||
state,
|
||
subscription,
|
||
parse_subscription_payload(original.rsplit("#", 1)[0] + "#New"),
|
||
self._fetch(),
|
||
)
|
||
self.assertEqual(state.nodes[0].id, stable_id)
|
||
self.assertFalse(fragment_only.result.reconnect_required)
|
||
|
||
pin_changed = reconcile_subscription(
|
||
state,
|
||
subscription,
|
||
parse_subscription_payload(
|
||
"hy2://secret@h.example:443/?pinSHA256=bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb#New"
|
||
),
|
||
self._fetch(),
|
||
)
|
||
self.assertEqual(state.nodes[0].id, stable_id)
|
||
self.assertTrue(pin_changed.selected_node_changed)
|
||
self.assertTrue(pin_changed.result.reconnect_required)
|
||
|
||
def test_failed_parse_is_atomic_and_removed_selection_prefers_same_subscription(self) -> None:
|
||
subscription = Subscription(id="sub", name="Provider", url="https://sub.example/a")
|
||
parsed = parse_subscription_payload(f"{VLESS_A}\n{VLESS_B}")
|
||
state = AppState(subscriptions=[subscription])
|
||
reconcile_subscription(state, subscription, parsed, self._fetch())
|
||
state.selected_node_id = state.nodes[1].id
|
||
snapshot = state.to_dict()
|
||
|
||
with self.assertRaises(SubscriptionParseError):
|
||
parse_subscription_payload("totally broken")
|
||
self.assertEqual(state.to_dict(), snapshot)
|
||
|
||
outcome = reconcile_subscription(
|
||
state,
|
||
subscription,
|
||
parse_subscription_payload(VLESS_A),
|
||
self._fetch(),
|
||
)
|
||
self.assertTrue(outcome.selected_node_removed)
|
||
self.assertEqual(state.selected_node_id, state.nodes[0].id)
|
||
|
||
def test_hide_and_remove_with_keep_local(self) -> None:
|
||
subscription = Subscription(id="sub", name="Provider", url="https://sub.example/a")
|
||
state = AppState(subscriptions=[subscription])
|
||
reconcile_subscription(state, subscription, parse_subscription_payload(VLESS_A), self._fetch())
|
||
node_id = state.nodes[0].id
|
||
state.selected_node_id = node_id
|
||
|
||
self.assertEqual(hide_subscription_node(state, node_id), "sub")
|
||
self.assertEqual(state.nodes, [])
|
||
self.assertEqual(len(subscription.hidden_source_keys), 1)
|
||
|
||
subscription.hidden_source_keys.clear()
|
||
reconcile_subscription(state, subscription, parse_subscription_payload(VLESS_A), self._fetch())
|
||
self.assertTrue(remove_subscription(state, "sub", keep_nodes=True))
|
||
self.assertEqual(state.subscriptions, [])
|
||
self.assertIsNone(state.nodes[0].subscription_id)
|
||
self.assertEqual(state.nodes[0].source_key, "")
|
||
self.assertEqual(state.nodes[0].source_fingerprint, "")
|
||
|
||
def test_hide_hysteria_uri_variant_keeps_other_pin_visible(self) -> None:
|
||
first = "hy2://secret@h.example:443/?pinSHA256=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa#Same"
|
||
second = "hysteria2://secret@h.example:443/?pinSHA256=bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb#Same"
|
||
subscription = Subscription(id="sub", name="Provider", url="https://sub.example/a")
|
||
state = AppState(subscriptions=[subscription])
|
||
reconcile_subscription(
|
||
state,
|
||
subscription,
|
||
parse_subscription_payload(f"{first}\n{second}"),
|
||
self._fetch(),
|
||
)
|
||
|
||
self.assertEqual(hide_subscription_node(state, state.nodes[0].id), "sub")
|
||
|
||
self.assertEqual(len(state.nodes), 1)
|
||
self.assertEqual(len(subscription.hidden_source_keys), 1)
|
||
self.assertIn(":", subscription.hidden_source_keys[0])
|
||
refreshed = parse_subscription_payload(
|
||
f"{first}\n{second}",
|
||
hidden_source_keys=set(subscription.hidden_source_keys),
|
||
)
|
||
self.assertEqual(len(refreshed.nodes), 1)
|
||
self.assertIn("bbbb", refreshed.nodes[0].link)
|
||
|
||
def test_metadata_redirect_policy_and_cache_headers(self) -> None:
|
||
subscription = Subscription(id="sub", url="https://same.example/old", etag="old")
|
||
state = AppState(subscriptions=[subscription])
|
||
parsed = parse_subscription_payload(
|
||
VLESS_A,
|
||
headers={"moved-permanently-to": "https://same.example/new"},
|
||
)
|
||
reconcile_subscription(state, subscription, parsed, self._fetch(etag="new"))
|
||
self.assertEqual(subscription.url, "https://same.example/new")
|
||
self.assertEqual(subscription.etag, "")
|
||
|
||
parsed = parse_subscription_payload(
|
||
VLESS_A,
|
||
headers={"moved-permanently-to": "https://other.example/new"},
|
||
)
|
||
reconcile_subscription(state, subscription, parsed, self._fetch(etag="current"))
|
||
self.assertEqual(subscription.url, "https://same.example/new")
|
||
self.assertEqual(subscription.pending_url, "https://other.example/new")
|
||
self.assertEqual(subscription.etag, "current")
|
||
|
||
def test_legacy_snapshot_recovers_gecko_and_stamps_parser_revision(self) -> None:
|
||
old_links = [
|
||
f"hy2://auth-{index}@s{index}.example:443/?obfs=salamander&"
|
||
f"obfs-password=cover-{index}#Server-{index}"
|
||
for index in range(38)
|
||
]
|
||
gecko_links = [
|
||
f"hy2://auth-g{index}@g{index}.example:443/?obfs=gecko&"
|
||
f"obfs-password=cover-g{index}#Gecko-{index}"
|
||
for index in range(4)
|
||
]
|
||
subscription = Subscription(id="sub", parser_revision=0, etag='"old"')
|
||
state = AppState(subscriptions=[subscription])
|
||
first = parse_subscription_payload("\n".join(old_links))
|
||
reconcile_subscription(state, subscription, first, self._fetch(etag='"old"'))
|
||
stable_ids = {node.provider_name: node.id for node in state.nodes}
|
||
subscription.parser_revision = 0
|
||
|
||
outcome = reconcile_subscription(
|
||
state,
|
||
subscription,
|
||
parse_subscription_payload("\n".join(old_links + gecko_links)),
|
||
self._fetch(etag='"new"'),
|
||
)
|
||
|
||
self.assertEqual(len(state.nodes), 42)
|
||
self.assertEqual(outcome.result.added, 4)
|
||
self.assertEqual(outcome.result.removed, 0)
|
||
self.assertEqual(subscription.parser_revision, SUBSCRIPTION_PARSER_REVISION)
|
||
self.assertEqual(subscription.etag, '"new"')
|
||
self.assertEqual(
|
||
{node.provider_name: node.id for node in state.nodes if node.provider_name in stable_ids},
|
||
stable_ids,
|
||
)
|
||
|
||
def test_persistent_parse_diagnostics_follow_applied_snapshot(self) -> None:
|
||
subscription = Subscription(id="sub")
|
||
state = AppState(subscriptions=[subscription])
|
||
partial = parse_subscription_payload(f"{VLESS_A}\nnot-a-link")
|
||
|
||
reconcile_subscription(state, subscription, partial, self._fetch(etag='"partial"'))
|
||
|
||
self.assertEqual(subscription.skipped_count, 1)
|
||
self.assertTrue(subscription.warnings)
|
||
unchanged = apply_not_modified(subscription, self._fetch(etag='"partial"'))
|
||
self.assertEqual(unchanged.skipped, 1)
|
||
self.assertEqual(unchanged.warnings, subscription.warnings)
|
||
|
||
reconcile_subscription(
|
||
state,
|
||
subscription,
|
||
parse_subscription_payload(VLESS_A),
|
||
self._fetch(etag='"clean"'),
|
||
)
|
||
self.assertEqual(subscription.skipped_count, 0)
|
||
self.assertEqual(subscription.warnings, [])
|
||
|
||
|
||
class SubscriptionSchedulingAndHttpTests(unittest.TestCase):
|
||
def test_open_client_deep_links_are_unwrapped_and_encrypted_happ_is_rejected(self) -> None:
|
||
target = "https://sub.example/list?token=secret"
|
||
encoded = base64.urlsafe_b64encode(target.encode()).decode().rstrip("=")
|
||
cases = (
|
||
(target, (target, None)),
|
||
(f"happ://add/{target}", (target, "happ")),
|
||
(f"incy://import/{target}", (target, "incy")),
|
||
(f"incy://import/{encoded}", (target, "incy")),
|
||
(f"v2raytun://import/{target}", (target, "v2raytun")),
|
||
)
|
||
for source, expected in cases:
|
||
with self.subTest(source=source):
|
||
self.assertEqual(resolve_subscription_source(source), expected)
|
||
launcher_cases = (
|
||
(f"https://nn.example/incy/{target}", (target, "incy")),
|
||
(f"https://nn.example/incy.html/{target}#x", (f"{target}#x", "incy")),
|
||
("https://nn.example/happ/https:/sub.example/t", ("https://sub.example/t", "happ")),
|
||
(
|
||
"https://nn.example/v2raytun/https%3A%2F%2Fsub.example%2Ft",
|
||
("https://sub.example/t", "v2raytun"),
|
||
),
|
||
# Не запускалка: обычный путь и прокси-обёртки чужих сервисов не трогаем.
|
||
("https://nn.example/incy/list", ("https://nn.example/incy/list", None)),
|
||
(
|
||
f"https://proxy.example/fetch/{target}",
|
||
(f"https://proxy.example/fetch/{target}", None),
|
||
),
|
||
)
|
||
for source, expected in launcher_cases:
|
||
with self.subTest(source=source):
|
||
self.assertEqual(resolve_subscription_source(source), expected)
|
||
# happ://crypt* теперь расшифровывается (см. tests/test_happ_crypt.py);
|
||
# непригодная нагрузка обязана давать доменную ошибку, а не трассировку.
|
||
with self.assertRaises(SubscriptionFetchError):
|
||
resolve_subscription_source("happ://crypt3/opaque-payload")
|
||
|
||
# Зашифрованные deep links других клиентов по-прежнему не поддерживаются.
|
||
with self.assertRaisesRegex(SubscriptionFetchError, "не расшифровываются"):
|
||
resolve_subscription_source("v2raytun://crypt/opaque-payload")
|
||
|
||
def test_client_profiles_send_only_explicit_stable_hwid_headers(self) -> None:
|
||
default_headers = subscription_request_headers(Subscription())
|
||
self.assertEqual(default_headers["User-Agent"], default_subscription_user_agent())
|
||
self.assertNotIn("X-HWID", default_headers)
|
||
|
||
subscription = Subscription(
|
||
client_profile="incy",
|
||
send_hwid=True,
|
||
hwid="DEVICE-ID-123",
|
||
)
|
||
headers = subscription_request_headers(subscription)
|
||
self.assertEqual(headers["User-Agent"], "INCY/1.0/Windows")
|
||
self.assertEqual(headers["X-HWID"], "DEVICE-ID-123")
|
||
self.assertEqual(headers["X-Device-ID"], "DEVICE-ID-123")
|
||
self.assertEqual(headers["X-Client"], "INCY")
|
||
self.assertEqual(headers["X-Device-OS"], "Windows")
|
||
self.assertNotIn("X-Real-IP", headers)
|
||
self.assertNotIn("X-Forwarded-For", headers)
|
||
|
||
# Набор Happ снят с реального клиента: лишний заголовок выдаёт подделку
|
||
# панели не хуже отсутствующего, поэтому Accept и X-App-Version не шлём.
|
||
happ = subscription_request_headers(
|
||
Subscription(client_profile="happ", send_hwid=True, hwid="DEVICE-ID-123")
|
||
)
|
||
self.assertEqual(happ["User-Agent"], "Happ/3.13.0")
|
||
self.assertNotIn("X-App-Version", happ)
|
||
self.assertNotIn("Accept", happ)
|
||
self.assertEqual(happ["X-Device-Locale"], _short_locale(_device_locale()))
|
||
self.assertNotIn("-", happ["X-Device-Locale"])
|
||
|
||
with self.assertRaises(SubscriptionFetchError):
|
||
subscription_request_headers(Subscription(user_agent="bad\r\nInjected: yes"))
|
||
with self.assertRaises(SubscriptionFetchError):
|
||
subscription_request_headers(
|
||
Subscription(send_hwid=True, hwid="bad\nvalue")
|
||
)
|
||
|
||
|
||
def test_windows_locale_name_becomes_a_language_code(self) -> None:
|
||
# На Windows locale.getlocale() отдаёт Russian_Russia, которое ни один
|
||
# клиент в заголовок не пишет.
|
||
self.assertEqual(_windows_locale_code("Russian_Russia"), "ru_RU")
|
||
self.assertEqual(_short_locale("ru-RU"), "ru")
|
||
|
||
def test_forbidden_is_explained_and_separates_the_site_protection(self) -> None:
|
||
panel = describe_http_failure(403, {})
|
||
self.assertIn("профиль клиента", panel)
|
||
protected = describe_http_failure(403, {"CF-Ray": "abc"})
|
||
self.assertIn("прокси", protected)
|
||
|
||
def test_panel_compatible_hwid_follows_the_happ_format(self) -> None:
|
||
self.assertTrue(is_panel_compatible_hwid("aaaaaaaaa-537f-4c45-a479-ee0b6cf035f7"))
|
||
self.assertTrue(is_panel_compatible_hwid("aBc123XyZ0"))
|
||
self.assertFalse(is_panel_compatible_hwid("short"))
|
||
self.assertFalse(is_panel_compatible_hwid("has spaces inside"))
|
||
|
||
|
||
def test_vendor_specific_parameters_do_not_block_the_import(self) -> None:
|
||
# Панели добавляют собственные параметры быстрее, чем их можно перечислить:
|
||
# fm и vcn у 3x-ui появились после spx.
|
||
node = parse_single(
|
||
"vless://11111111-1111-1111-1111-111111111111@a.example:443"
|
||
"?security=tls&sni=a.example&spx=%2F&fm=%7B%22a%22%3A1%7D&vcn=1#Vendor"
|
||
)
|
||
self.assertEqual(node.outbound["protocol"], "vless")
|
||
|
||
def test_unknown_security_mode_is_refused(self) -> None:
|
||
with self.assertRaises(LinkParseError):
|
||
parse_single(
|
||
"vless://11111111-1111-1111-1111-111111111111@a.example:443?security=xtls#Bad"
|
||
)
|
||
|
||
|
||
def test_certificate_pin_next_to_insecure_keeps_the_server(self) -> None:
|
||
# URI целиком получает официальное ядро, поэтому пин не конвертируется
|
||
# и не зависит от значения insecure.
|
||
node = parse_single(
|
||
"hysteria2://secret@a.example:8443"
|
||
"?insecure=1&pinSHA256=0ed6040ed6040ed6040ed6040ed6040ed6040ed6040ed6040ed6040ed6040ed6&obfs=salamander&obfs-password=pw&sni=a.example#Pin"
|
||
)
|
||
self.assertTrue(node.outbound["tls"]["insecure"])
|
||
|
||
pinned = parse_single(
|
||
"hysteria2://secret@a.example:8443?pinSHA256=0ed6040ed6040ed6040ed6040ed6040ed6040ed6040ed6040ed6040ed6040ed6#Pin"
|
||
)
|
||
self.assertIn("pinSHA256=", pinned.link)
|
||
|
||
def test_interval_override_and_backoff(self) -> None:
|
||
now = datetime.now(timezone.utc)
|
||
subscription = Subscription(
|
||
parser_revision=SUBSCRIPTION_PARSER_REVISION,
|
||
update_interval_hours=2,
|
||
provider_interval_hours=8,
|
||
last_success_at=(now - timedelta(hours=3)).isoformat(),
|
||
)
|
||
self.assertTrue(subscription_due(subscription, now))
|
||
result = mark_subscription_failure(subscription, "failed https://host/path?token=secret")
|
||
self.assertFalse(result.success)
|
||
self.assertNotIn("token=secret", result.message)
|
||
self.assertFalse(subscription_due(subscription, now))
|
||
mark_subscription_failure(subscription, "again")
|
||
self.assertEqual(subscription.failure_count, 2)
|
||
|
||
stale = Subscription(
|
||
parser_revision=0,
|
||
last_success_at=now.isoformat(),
|
||
)
|
||
self.assertTrue(subscription_due(stale, now))
|
||
stale.auto_update = False
|
||
self.assertFalse(subscription_due(stale, now))
|
||
|
||
def test_auto_fetch_retries_once_through_active_proxy(self) -> None:
|
||
subscription = Subscription(url="https://sub.example/path?token=secret")
|
||
calls: list[tuple[bool, int | None]] = []
|
||
|
||
def fake_once(_subscription, **kwargs):
|
||
calls.append((kwargs["via_proxy"], kwargs["proxy_port"]))
|
||
if not kwargs["via_proxy"]:
|
||
raise OSError("direct failed")
|
||
return SubscriptionFetchResult(data=b"ok", via_proxy=True)
|
||
|
||
with patch("xray_fluent.importer.subscription_http._fetch_once", side_effect=fake_once):
|
||
result = fetch_subscription(subscription, mode="auto", proxy_port=1391)
|
||
self.assertTrue(result.via_proxy)
|
||
self.assertEqual(calls, [(False, None), (True, 1391)])
|
||
|
||
def test_force_refresh_preserves_fallback_and_omits_validators(self) -> None:
|
||
subscription = Subscription(
|
||
url="https://sub.example/list",
|
||
etag='"old"',
|
||
last_modified="Mon, 01 Jan 2024 00:00:00 GMT",
|
||
parser_revision=SUBSCRIPTION_PARSER_REVISION,
|
||
)
|
||
attempts: list[bool] = []
|
||
|
||
def fake_once(_subscription, **kwargs):
|
||
attempts.append(bool(kwargs["force_refresh"]))
|
||
if not kwargs["via_proxy"]:
|
||
raise OSError("direct failed")
|
||
return SubscriptionFetchResult(data=b"ok", via_proxy=True)
|
||
|
||
with patch("xray_fluent.importer.subscription_http._fetch_once", side_effect=fake_once):
|
||
result = fetch_subscription(
|
||
subscription,
|
||
mode="auto",
|
||
proxy_port=1391,
|
||
force_refresh=True,
|
||
)
|
||
self.assertTrue(result.via_proxy)
|
||
self.assertEqual(attempts, [True, True])
|
||
|
||
def test_proxy_mode_requires_port_and_secrets_are_masked(self) -> None:
|
||
subscription = Subscription(url="https://user:pass@sub.example/very-long-token-value?token=secret")
|
||
with self.assertRaises(SubscriptionFetchError):
|
||
fetch_subscription(subscription, mode="proxy", proxy_port=None)
|
||
masked = mask_subscription_url(subscription.url)
|
||
self.assertEqual(masked, "********")
|
||
self.assertNotIn("token=secret", masked)
|
||
self.assertNotIn("very-long-token-value", masked)
|
||
self.assertNotIn("pass", masked)
|
||
self.assertNotIn("pass", sanitize_fetch_error(RuntimeError(subscription.url)))
|
||
|
||
def test_cache_headers_user_agent_304_size_limit_and_redirect_scheme(self) -> None:
|
||
subscription = Subscription(
|
||
url="https://sub.example/list",
|
||
etag='"abc"',
|
||
last_modified="Mon, 01 Jan 2024 00:00:00 GMT",
|
||
parser_revision=SUBSCRIPTION_PARSER_REVISION,
|
||
)
|
||
|
||
class Response:
|
||
status = 200
|
||
headers = {"ETag": '"next"'}
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *_args):
|
||
return False
|
||
|
||
def geturl(self):
|
||
return subscription.url
|
||
|
||
def read(self, _limit):
|
||
return b"payload"
|
||
|
||
opener = SimpleNamespace(open=lambda request, timeout: Response())
|
||
with patch("xray_fluent.importer.subscription_http.build_opener", return_value=opener):
|
||
result = fetch_subscription(subscription, mode="direct")
|
||
self.assertEqual(result.headers["etag"], '"next"')
|
||
self.assertEqual(default_subscription_user_agent().split("/", 1)[0], "ZapretKVN")
|
||
|
||
captured = []
|
||
|
||
def open_and_capture(request, timeout):
|
||
captured.append((request, timeout))
|
||
return Response()
|
||
|
||
opener.open = open_and_capture
|
||
with patch("xray_fluent.importer.subscription_http.build_opener", return_value=opener):
|
||
fetch_subscription(subscription, mode="direct")
|
||
request, timeout = captured[0]
|
||
self.assertEqual(request.get_header("If-none-match"), '"abc"')
|
||
self.assertEqual(request.get_header("If-modified-since"), "Mon, 01 Jan 2024 00:00:00 GMT")
|
||
self.assertEqual(request.get_header("User-agent"), default_subscription_user_agent())
|
||
self.assertEqual(timeout, 15)
|
||
|
||
class OversizedResponse(Response):
|
||
def read(self, limit):
|
||
return b"x" * limit
|
||
|
||
opener.open = lambda _request, timeout: OversizedResponse()
|
||
with patch("xray_fluent.importer.subscription_http.build_opener", return_value=opener):
|
||
with self.assertRaisesRegex(SubscriptionFetchError, "превышает"):
|
||
fetch_subscription(subscription, mode="direct", max_bytes=8)
|
||
|
||
http_error = urllib.error.HTTPError(
|
||
subscription.url,
|
||
304,
|
||
"Not Modified",
|
||
{"ETag": '"abc"'},
|
||
None,
|
||
)
|
||
opener.open = lambda _request, timeout: (_ for _ in ()).throw(http_error)
|
||
with patch("xray_fluent.importer.subscription_http.build_opener", return_value=opener):
|
||
not_modified = fetch_subscription(subscription, mode="direct")
|
||
self.assertTrue(not_modified.not_modified)
|
||
|
||
stale = Subscription(
|
||
url=subscription.url,
|
||
etag='"abc"',
|
||
last_modified=subscription.last_modified,
|
||
parser_revision=0,
|
||
)
|
||
captured.clear()
|
||
opener.open = open_and_capture
|
||
with patch("xray_fluent.importer.subscription_http.build_opener", return_value=opener):
|
||
fetch_subscription(stale, mode="direct")
|
||
stale_request = captured[0][0]
|
||
self.assertIsNone(stale_request.get_header("If-none-match"))
|
||
self.assertIsNone(stale_request.get_header("If-modified-since"))
|
||
opener.open = lambda _request, timeout: (_ for _ in ()).throw(http_error)
|
||
with patch("xray_fluent.importer.subscription_http.build_opener", return_value=opener):
|
||
with self.assertRaisesRegex(SubscriptionFetchError, "304 без условного"):
|
||
fetch_subscription(stale, mode="direct")
|
||
|
||
# Отказ по HWID под видом HTTP 200 с узлами-заглушками не импортируется.
|
||
for header, text in (
|
||
("x-hwid-not-supported", "HWID"),
|
||
("x-hwid-max-devices-reached", "лимит устройств"),
|
||
):
|
||
class RefusalResponse(Response):
|
||
headers = {header: "true", "x-hwid-limit": "true"}
|
||
|
||
opener.open = lambda _request, timeout: RefusalResponse()
|
||
with self.subTest(header=header), patch(
|
||
"xray_fluent.importer.subscription_http.build_opener", return_value=opener
|
||
):
|
||
with self.assertRaisesRegex(SubscriptionFetchError, text):
|
||
fetch_subscription(subscription, mode="auto", proxy_port=10808)
|
||
|
||
handler = _SafeRedirectHandler()
|
||
with self.assertRaises(SubscriptionFetchError):
|
||
handler.redirect_request(
|
||
urllib.request.Request(subscription.url),
|
||
None,
|
||
302,
|
||
"Found",
|
||
{},
|
||
"file:///tmp/secret",
|
||
)
|
||
|
||
|
||
class SubscriptionStateAndQrTests(unittest.TestCase):
|
||
def test_v1_migration_and_plain_encrypted_backup_roundtrip(self) -> None:
|
||
legacy = AppState.from_dict(
|
||
{
|
||
"schema_version": 1,
|
||
"nodes": [{"id": "old", "name": "Local", "server": "example.com", "port": 443}],
|
||
}
|
||
)
|
||
self.assertEqual(legacy.schema_version, 1)
|
||
self.assertIsNone(legacy.nodes[0].subscription_id)
|
||
|
||
subscription = Subscription(
|
||
id="sub",
|
||
name="Private",
|
||
url="https://sub.example/?token=secret",
|
||
client_profile="happ",
|
||
send_hwid=True,
|
||
hwid="OWN-DEVICE-ID",
|
||
parser_revision=SUBSCRIPTION_PARSER_REVISION,
|
||
skipped_count=4,
|
||
warnings=[
|
||
"hysteria2://auth@example/?obfs-password=secret",
|
||
"PrivateKey=private-value",
|
||
],
|
||
)
|
||
state = AppState(schema_version=STATE_SCHEMA_VERSION, subscriptions=[subscription])
|
||
state_data = state.to_dict()
|
||
restored = AppState.from_dict(state_data)
|
||
self.assertEqual(restored.subscription_device_id, state.subscription_device_id)
|
||
self.assertEqual(restored.subscriptions[0].client_profile, "happ")
|
||
self.assertTrue(restored.subscriptions[0].send_hwid)
|
||
self.assertEqual(restored.subscriptions[0].hwid, "OWN-DEVICE-ID")
|
||
self.assertEqual(restored.subscriptions[0].parser_revision, SUBSCRIPTION_PARSER_REVISION)
|
||
self.assertEqual(restored.subscriptions[0].skipped_count, 4)
|
||
self.assertNotIn("secret", " ".join(restored.subscriptions[0].warnings))
|
||
self.assertNotIn("private-value", " ".join(restored.subscriptions[0].warnings))
|
||
legacy_subscription = Subscription.from_dict({"url": "https://example.com"})
|
||
self.assertEqual(legacy_subscription.parser_revision, 0)
|
||
self.assertEqual(legacy_subscription.skipped_count, 0)
|
||
self.assertEqual(legacy_subscription.warnings, [])
|
||
self.assertEqual(
|
||
normalize_subscription_warnings("not-a-list"),
|
||
[],
|
||
)
|
||
with tempfile.TemporaryDirectory() as temp_dir:
|
||
state_path = Path(temp_dir) / "state.enc"
|
||
plain_path = Path(temp_dir) / "state-plain.enc"
|
||
backup_path = Path(temp_dir) / "backup.enc"
|
||
plain_storage = StateStorage(plain_path)
|
||
plain_storage.save(state)
|
||
self.assertEqual(plain_storage.load().subscriptions[0].url, subscription.url)
|
||
storage = StateStorage(state_path)
|
||
storage.passphrase = "strong passphrase"
|
||
storage.save(state)
|
||
self.assertNotIn("token=secret", state_path.read_text(encoding="utf-8"))
|
||
loaded = storage.load()
|
||
self.assertEqual(loaded.subscriptions[0].url, subscription.url)
|
||
storage.export_backup(backup_path, "backup passphrase")
|
||
imported = storage.import_backup(backup_path, "backup passphrase")
|
||
self.assertEqual(imported.subscriptions[0].url, subscription.url)
|
||
|
||
def test_diagnostics_redact_subscription_urls_node_credentials_and_log_urls(self) -> None:
|
||
state = AppState(
|
||
subscription_device_id="INSTALL-DEVICE-SECRET",
|
||
subscriptions=[
|
||
Subscription(
|
||
url="https://sub.example/?token=top-secret",
|
||
send_hwid=True,
|
||
hwid="SUBSCRIPTION-HWID-SECRET",
|
||
warnings=[
|
||
"hy2://auth-secret@example/?obfs-password=cover-secret",
|
||
"PrivateKey=warning-private-key",
|
||
],
|
||
)
|
||
],
|
||
nodes=[
|
||
Node(
|
||
link="vless://private-link",
|
||
outbound={"type": "wireguard", "private_key": "private-key"},
|
||
)
|
||
],
|
||
)
|
||
with tempfile.TemporaryDirectory() as temp_dir:
|
||
archive_path = export_diagnostics(
|
||
Path(temp_dir) / "diagnostics.zip",
|
||
state,
|
||
["failed https://sub.example/path?token=log-secret"],
|
||
)
|
||
with zipfile.ZipFile(archive_path) as archive:
|
||
state_text = archive.read("state_redacted.json").decode()
|
||
logs_text = archive.read("recent_logs.txt").decode()
|
||
self.assertNotIn("top-secret", state_text)
|
||
self.assertNotIn("private-link", state_text)
|
||
self.assertNotIn("private-key", state_text)
|
||
self.assertNotIn("INSTALL-DEVICE-SECRET", state_text)
|
||
self.assertNotIn("SUBSCRIPTION-HWID-SECRET", state_text)
|
||
self.assertNotIn("auth-secret", state_text)
|
||
self.assertNotIn("cover-secret", state_text)
|
||
self.assertNotIn("warning-private-key", state_text)
|
||
self.assertNotIn("log-secret", logs_text)
|
||
|
||
def test_qimage_is_passed_directly_and_only_http_is_accepted(self) -> None:
|
||
image = QImage(2, 2, QImage.Format.Format_RGB32)
|
||
received = []
|
||
|
||
def read_barcode(value, **_kwargs):
|
||
received.append(value)
|
||
return SimpleNamespace(text="https://sub.example/list")
|
||
|
||
fake_module = SimpleNamespace(
|
||
read_barcode=read_barcode,
|
||
BarcodeFormat=SimpleNamespace(QRCode=object()),
|
||
)
|
||
with patch.dict(sys.modules, {"zxingcpp": fake_module}):
|
||
self.assertEqual(decode_subscription_qr(image), "https://sub.example/list")
|
||
self.assertIs(received[0], image)
|
||
|
||
fake_module.read_barcode = lambda *_args, **_kwargs: SimpleNamespace(
|
||
text="happ://add/https://sub.example/list"
|
||
)
|
||
with patch.dict(sys.modules, {"zxingcpp": fake_module}):
|
||
self.assertEqual(
|
||
decode_subscription_qr(image),
|
||
"happ://add/https://sub.example/list",
|
||
)
|
||
|
||
fake_module.read_barcode = lambda *_args, **_kwargs: SimpleNamespace(text="vless://secret")
|
||
with patch.dict(sys.modules, {"zxingcpp": fake_module}):
|
||
with self.assertRaises(QrDecodeError):
|
||
decode_subscription_qr(image)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|