a5d980a5e9
实现独立加密设备身份、网关 agent、持久生命周期任务和 Surge/FlClash 专属订阅凭证。 完善 Shadowrocket 规则转换与 Clash AI DNS，增加 Surge FFSB 适配、规则快照、运维工具及验收文档。
86 lines
4.3 KiB
Python
86 lines
4.3 KiB
Python
"""Offline tests; never opens a WG interface or contacts a gateway."""
|
|
import base64
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location('wg_agent', Path(__file__).with_name('agent.py'))
|
|
agent = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(agent)
|
|
key = lambda n: base64.b64encode(bytes([n]) * 32).decode()
|
|
|
|
|
|
class AgentTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.task = dict(id='task-1', publicKey=key(2), oldPublicKey=None, presharedKey=key(3),
|
|
address='10.203.0.2', desired='active', action='create', version=1)
|
|
self.base = '[Interface]\nPrivateKey = gateway-secret\nAddress = 10.203.0.1/24\nPostUp = preserve firewall\n\n[Peer]\nPublicKey = ' + key(8) + '\nAllowedIPs = 10.203.0.8/32\n'
|
|
|
|
def test_merge_preserves_unmanaged_peers_and_interface(self):
|
|
merged = agent.merge_config(self.base, self.task)
|
|
self.assertTrue(merged.startswith(self.base))
|
|
self.assertEqual(agent.merge_config(merged, self.task), merged)
|
|
task = {**self.task, 'desired': 'paused', 'action': 'pause'}
|
|
removed = agent.merge_config(merged, task)
|
|
self.assertNotIn(key(2), removed)
|
|
self.assertIn(key(8), removed)
|
|
self.assertIn('PostUp = preserve firewall', removed)
|
|
|
|
def test_conflicts_and_outside_network_are_rejected(self):
|
|
for address in ['8.8.8.8', '10.203.0.1', '10.203.0.255']:
|
|
with self.assertRaises(ValueError):
|
|
agent.merge_config(self.base, {**self.task, 'address': address})
|
|
with self.assertRaises(ValueError):
|
|
agent.merge_config(self.base, {**self.task, 'address': '10.203.0.8'})
|
|
with self.assertRaises(ValueError):
|
|
agent.merge_config(self.base.replace('PostUp = preserve firewall', 'SaveConfig = true'), self.task)
|
|
|
|
def test_rotate_removes_old_identity(self):
|
|
old = agent.merge_config(self.base, self.task)
|
|
rotated = agent.merge_config(old, {**self.task, 'publicKey': key(4), 'oldPublicKey': key(2), 'action': 'rotate'})
|
|
self.assertNotIn(key(2), rotated)
|
|
self.assertIn(key(4), rotated)
|
|
self.assertIn(key(8), rotated)
|
|
|
|
def test_adoption_is_read_only_and_requires_matching_psk(self):
|
|
config = agent.merge_config(self.base, self.task)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
file = Path(directory) / 'wg-njcq.conf'
|
|
file.write_text(config)
|
|
with patch.object(agent, 'CONFIG', file), patch.object(agent, 'runtime_peers', return_value=agent.peers(config)), patch.object(agent, 'command') as cmd:
|
|
self.assertEqual(agent.apply_task({**self.task, 'action': 'adopt'}), (True, True))
|
|
self.assertEqual(agent.apply_task({**self.task, 'action': 'adopt', 'presharedKey': key(9)}), (False, False))
|
|
cmd.assert_not_called()
|
|
self.assertEqual(file.read_text(), config)
|
|
|
|
def test_pause_persists_and_removes_runtime_peer_before_confirming(self):
|
|
config = agent.merge_config(self.base, self.task)
|
|
runtime = agent.peers(config)
|
|
calls = []
|
|
def command(*args):
|
|
calls.append(args)
|
|
if args[:2] == ('wg', 'showconf'):
|
|
return config
|
|
if args[-1] == 'remove':
|
|
runtime.pop(args[-2], None)
|
|
return ''
|
|
raise AssertionError(args)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
file = Path(directory) / 'wg-njcq.conf'
|
|
file.write_text(config)
|
|
with patch.object(agent, 'CONFIG', file), patch.object(agent, 'runtime_peers', side_effect=lambda: runtime), patch.object(agent, 'command', side_effect=command):
|
|
task = {**self.task, 'desired': 'paused', 'action': 'pause'}
|
|
self.assertEqual(agent.apply_task(task), (True, True))
|
|
self.assertEqual(agent.apply_task(task), (True, True))
|
|
self.assertNotIn(key(2), file.read_text())
|
|
self.assertIn(key(8), file.read_text())
|
|
self.assertNotIn(key(2), runtime)
|
|
self.assertEqual(len(list((file.parent / 'proxy-station-backups').glob('*.conf'))), 1)
|
|
self.assertEqual(file.stat().st_mode & 0o777, 0o600)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|