Improve TUN routing readability

Signed-off-by: Loren Eteval <loren.eteval@proton.me>
This commit is contained in:
Loren Eteval
2026-10-09 20:48:16 +08:00
parent 246f4e2ae5
commit ef0b20153b
4 changed files with 20 additions and 0 deletions
+8
View File
@@ -79,6 +79,7 @@ def _linuxRuleDelete(flag, rule, table, start, device):
'dport_end',
'dport_mask',
}
priority = int(rule['priority'])
target = str(rule.get('table', ''))
@@ -345,6 +346,7 @@ class SingTUNLinuxRouting:
),
[],
)
routes = sum(
(
json.loads(
@@ -354,6 +356,7 @@ class SingTUNLinuxRouting:
),
[],
)
tables = {str(item.get('table')) for item in rules + routes}
priorities = {int(item['priority']) for item in rules}
@@ -449,6 +452,7 @@ class SingTUNLinuxRouting:
name = self.tun['Name']
index = shlex.quote('/sys/class/net/' + name + '/ifindex')
lines.append(f'createdIndex=$(cat {index})')
# Roll back device creation if a later setup command fails.
rollback = (
'code=$?; if [ "$code" -ne 0 ] '
@@ -479,6 +483,7 @@ class SingTUNLinuxRouting:
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
code = self._process.wait(timeout=self.AuthorizationTimeout)
except subprocess.TimeoutExpired:
# Do not drop a root process that the unprivileged owner cannot kill.
@@ -547,6 +552,7 @@ class SingTUNLinuxRouting:
)
commands.extend(self.rules)
self.execute(commands, preparing=True)
link = self.link()
@@ -625,6 +631,7 @@ class SingTUNLinuxRouting:
deletion = _linuxRuleDelete(
flag, rule, self.table, self.start, self.tun['Name']
)
if _linuxRuleSelectors(deletion) not in expected:
raise RuntimeError(
'Conflicting Linux policy rule; cleanup ownership retained'
@@ -643,6 +650,7 @@ class SingTUNLinuxRouting:
for flag in ('-4', '-6'):
rules = json.loads(_command(['ip', flag, '-N', '-j', 'rule', 'show']))
routes = [
route
for route in json.loads(
+1
View File
@@ -261,6 +261,7 @@ class SingTUNHostPlan:
self._linux = SingTUNLinuxRouting(tun, table, start, self._cancel)
self._activated = True
self._linux.prepare()
# Linux host changes belong to the helper; the spawned binding only
+3
View File
@@ -1276,6 +1276,7 @@ class LinuxTun2socksScriptTest(TestCase):
manager, attempt, None, None
)
)
tun.stop()
attempt.rollback.assert_not_called()
@@ -1285,9 +1286,11 @@ class LinuxTun2socksScriptTest(TestCase):
if not existing:
self.assertIn(shlex.quote(interface), scripts[0])
bypass = next(
line for line in scripts[0].splitlines() if 'via ' in line
)
self.assertEqual(
shlex.split(bypass),
[
+8
View File
@@ -85,6 +85,7 @@ class SingTUNLinuxHostTest(unittest.TestCase):
Path(arguments[2]).name.startswith(tempfile.gettempprefix())
)
self.assertEqual(Path(arguments[2]).suffix, '.sh')
scripts.append(Path(arguments[2]).read_text(encoding='utf-8'))
return process
@@ -103,6 +104,7 @@ class SingTUNLinuxHostTest(unittest.TestCase):
self.host.applyDNS('1.1.1.1')
self.assertEqual(self.host.index, 77)
self.assertIn('ip tuntap add dev utun101 mode tun user 1000', scripts[0])
self.assertIn('ip link set dev utun101 mtu 1500', scripts[0])
self.assertIn('ip -6 addr add fd00::1/64 dev utun101', scripts[0])
@@ -142,6 +144,7 @@ class SingTUNLinuxHostTest(unittest.TestCase):
)
directory = Path(self.host._directory.name)
self.assertIs(self.host._process, process)
self.assertEqual(len(list(directory.glob('*.sh'))), 1)
self.assertTrue((directory / 'cancelled').exists())
@@ -180,8 +183,10 @@ class SingTUNLinuxHostTest(unittest.TestCase):
root = Path(directory).resolve()
net = root / 'net'
binDirectory = root / 'bin'
net.mkdir()
binDirectory.mkdir()
ip = binDirectory / 'ip'
# This fixture only creates/removes files under its temporary
@@ -217,6 +222,7 @@ class SingTUNLinuxHostTest(unittest.TestCase):
def launch(arguments, **kwargs):
self.assertEqual(arguments[:2], ['pkexec', 'bash'])
script = Path(arguments[-1])
text = script.read_text(encoding='utf-8')
text = text.replace('/sys/class/net', net.as_posix())
@@ -239,6 +245,7 @@ class SingTUNLinuxHostTest(unittest.TestCase):
FAKE_CANCEL=(script.parent / 'cancelled').as_posix(),
FAKE_SCRIPT=script.as_posix(),
)
return realPopen([bash, str(script)], env=environment, **kwargs)
with mock.patch.object(
@@ -254,6 +261,7 @@ class SingTUNLinuxHostTest(unittest.TestCase):
):
if failure == 'none':
self.host.prepare()
self.assertEqual(self.host.index, 77)
else:
with self.assertRaisesRegex(