loop-uh
92d61d57d9
Clash API sing-box отдаёт chains от листа к корню (common.Reverse в trafficcontrol/tracker.go): у гибрида с xray-сайдкаром и у пула первым идёт тег узла (__app_hybrid_relay_a), а proxy — последним. Сборщик смотрел только chains[0], поэтому в TUN каждое соединение через VLESS/ VMess/Trojan показывалось как «напрямую», хотя шло через сервер. Колонка «маршрут» теперь использует ту же классификацию по всей цепочке, что и спрос на туннель (is_proxied_chain в proxy_demand). Проверено e2e на win10: план v0.7.7 (sing-box TUN + xray-сайдкар) и VLESS-сервер. curl ушёл через сервер (сервер получил google/example/ cloudflare), Clash API вернул chains ['__app_hybrid_relay_a', 'proxy'], старый сборщик показал curl.exe route=direct.
78 lines
3.2 KiB
Python
78 lines
3.2 KiB
Python
"""Сбор трафика по процессам: состояние сессии ограничено живыми соединениями."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import json
|
||
import unittest
|
||
from unittest import mock
|
||
|
||
from xray_fluent.platform.windows import process_traffic_collector as collector
|
||
|
||
|
||
def _conn(cid: str, exe: str, up: int, down: int) -> dict:
|
||
return {
|
||
"id": cid, "upload": up, "download": down, "chains": ["proxy"],
|
||
"metadata": {"processPath": f"C:/Apps/{exe}", "host": "example.com"},
|
||
}
|
||
|
||
|
||
class _Response(io.BytesIO):
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *exc):
|
||
return False
|
||
|
||
|
||
def _poll(connections: list[dict]):
|
||
payload = json.dumps({"connections": connections}).encode()
|
||
with mock.patch.object(collector.urllib.request, "urlopen", return_value=_Response(payload)):
|
||
return {s.exe.lower(): s for s in collector.collect_process_stats()}
|
||
|
||
|
||
class BoundedCollectorTests(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
collector.reset_connection_tracking()
|
||
|
||
def tearDown(self) -> None:
|
||
collector.reset_connection_tracking()
|
||
|
||
def test_closed_connections_leave_state_but_keep_bytes_and_count(self) -> None:
|
||
_poll([_conn("a", "chrome.exe", 10, 100), _conn("b", "chrome.exe", 5, 50)])
|
||
self.assertEqual(len(collector._conn_owner), 2)
|
||
stats = _poll([_conn("b", "chrome.exe", 5, 60)])
|
||
# «a» закрылось: его байты перешли в закрытые, а запись владельца удалена.
|
||
self.assertEqual(set(collector._conn_owner), {"b"})
|
||
self.assertEqual(stats["chrome.exe"].download, 160)
|
||
stats = _poll([])
|
||
self.assertEqual(collector._conn_owner, {})
|
||
self.assertEqual(stats, {})
|
||
stats = _poll([_conn("c", "chrome.exe", 1, 1)])
|
||
self.assertEqual(stats["chrome.exe"].download, 161)
|
||
|
||
def test_state_does_not_grow_with_connection_churn(self) -> None:
|
||
for batch in range(200):
|
||
_poll([_conn(f"{batch}-{i}", "app.exe", 1, 1) for i in range(20)])
|
||
self.assertLessEqual(len(collector._conn_owner), 20)
|
||
self.assertLessEqual(len(collector._conn_bytes), 20)
|
||
self.assertEqual(collector._conn_total["app.exe"], 200 * 20)
|
||
|
||
|
||
def test_route_follows_whole_leaf_first_chain(self) -> None:
|
||
# Форма цепочек снята с живого sing-box TUN (гибрид с xray-сайдкаром):
|
||
# Clash API идёт от листа к корню, поэтому ``proxy`` стоит последним.
|
||
tunneled = _conn("a", "curl.exe", 10, 600)
|
||
tunneled["chains"] = ["__app_hybrid_relay_a", "proxy"]
|
||
direct = _conn("b", "yandexmusic.exe", 5, 50)
|
||
direct["chains"] = ["direct"]
|
||
stats = _poll([tunneled, direct])
|
||
self.assertEqual(stats["curl.exe"].route, "proxy")
|
||
self.assertEqual(stats["curl.exe"].proxy_bytes, 610)
|
||
self.assertEqual(stats["curl.exe"].direct_bytes, 0)
|
||
self.assertEqual(stats["yandexmusic.exe"].route, "direct")
|
||
self.assertEqual(stats["yandexmusic.exe"].direct_bytes, 55)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|