mirror of
https://github.com/LorenEteval/Furious.git
synced 2026-10-13 01:38:33 +03:00
761 lines
29 KiB
Python
761 lines
29 KiB
Python
# Copyright (C) 2024–present Loren Eteval & contributors <loren.eteval@proton.me>
|
||
#
|
||
# This file is part of Furious.
|
||
#
|
||
# This program is free software: you can redistribute it and/or modify
|
||
# it under the terms of the GNU General Public License as published by
|
||
# the Free Software Foundation, either version 3 of the License, or
|
||
# (at your option) any later version.
|
||
#
|
||
# This program is distributed in the hope that it will be useful,
|
||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||
# GNU General Public License for more details.
|
||
#
|
||
# You should have received a copy of the GNU General Public License
|
||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||
|
||
"""Verify sing-tun's Linux helpers without changing real host networking."""
|
||
|
||
from Furious.Models.SingTUN import prepareSingTUNSettings, addSingTUNExclusions
|
||
from Furious.Service.SingTUNHost import SingTUNHostPlan
|
||
from Furious.Frozenlib.SingTUNRouting import (
|
||
SingTUNLinuxRouting,
|
||
_linuxRouteRanges,
|
||
_linuxRules,
|
||
_linuxRuleDelete,
|
||
_linuxRuleSelectors,
|
||
)
|
||
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
|
||
import importlib
|
||
import ipaddress
|
||
import json
|
||
import os
|
||
import shutil
|
||
import shlex
|
||
import subprocess
|
||
import tempfile
|
||
import threading
|
||
import unittest
|
||
|
||
|
||
def _bash():
|
||
"""Use native Bash or Git Bash, never a Windows WSL launcher."""
|
||
if os.name != 'nt':
|
||
return shutil.which('bash')
|
||
|
||
git = shutil.which('git')
|
||
|
||
if git:
|
||
candidate = Path(git).resolve().parent.parent / 'bin' / 'bash.exe'
|
||
|
||
if candidate.exists():
|
||
return str(candidate)
|
||
|
||
return None
|
||
|
||
|
||
class SingTUNLinuxHostTest(unittest.TestCase):
|
||
def setUp(self):
|
||
self.module = importlib.import_module('Furious.Frozenlib.SingTUNRouting')
|
||
routingTable = importlib.import_module('Furious.Frozenlib.SystemRoutingTable')
|
||
|
||
platform = mock.patch.object(routingTable, 'PLATFORM', 'Linux')
|
||
flatpak = mock.patch.object(
|
||
routingTable.SystemRuntime, 'flatpakID', return_value=''
|
||
)
|
||
|
||
platform.start()
|
||
flatpak.start()
|
||
|
||
self.addCleanup(platform.stop)
|
||
self.addCleanup(flatpak.stop)
|
||
|
||
self.tun = prepareSingTUNSettings(
|
||
{'tun_options': {'Name': 'utun101', 'Inet6Address': ['fd00::1/64']}}
|
||
)['tun_options']
|
||
self.host = SingTUNLinuxRouting(self.tun, 123456, 2000, threading.Event())
|
||
|
||
def tearDown(self):
|
||
if self.host._process is not None:
|
||
self.host._process.poll.return_value = 0
|
||
|
||
self.host.drain()
|
||
|
||
def testHelperCreatesUserOwnedTunAndQuotesOnlyHostArguments(self):
|
||
scripts = []
|
||
process = mock.Mock()
|
||
process.wait.return_value = 0
|
||
process.poll.return_value = None
|
||
|
||
def launch(arguments, **kwargs):
|
||
self.assertEqual(arguments[:2], ['pkexec', 'bash'])
|
||
self.assertEqual(kwargs['stdin'], subprocess.PIPE)
|
||
self.assertTrue(
|
||
Path(arguments[2]).name.startswith(tempfile.gettempprefix())
|
||
)
|
||
self.assertEqual(Path(arguments[2]).suffix, '.sh')
|
||
|
||
scripts.append(Path(arguments[2]).read_text(encoding='utf-8'))
|
||
Path(arguments[2]).parent.joinpath('ready').write_text(
|
||
'77', encoding='utf-8'
|
||
)
|
||
return process
|
||
|
||
link = {'ifname': 'utun101', 'ifindex': 77}
|
||
|
||
with mock.patch.object(
|
||
self.module.os, 'geteuid', return_value=1000, create=True
|
||
), mock.patch.object(
|
||
self.module.os, 'getuid', return_value=1000, create=True
|
||
), mock.patch.object(
|
||
self.module.subprocess, 'Popen', side_effect=launch
|
||
), mock.patch.object(
|
||
self.module, '_command', return_value=json.dumps([link])
|
||
):
|
||
self.host.prepare()
|
||
with self.assertRaisesRegex(RuntimeError, 'already active'):
|
||
self.host.execute([])
|
||
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])
|
||
self.assertIn('ip link set dev utun101 up', scripts[0])
|
||
self.assertIn('createdIndex=', scripts[0])
|
||
self.assertIn('trap ', scripts[0])
|
||
self.assertNotIn('ifalias', ''.join(scripts))
|
||
self.assertNotIn('furious-sing-tun-', ''.join(scripts))
|
||
self.assertEqual(len(scripts), 1)
|
||
self.assertIn('/sys/class/net/utun101/ifindex', scripts[0])
|
||
self.assertIn('resolvectl dns utun101 1.1.1.1', scripts[0])
|
||
self.assertNotIn('socks5', ''.join(scripts))
|
||
self.assertIs(self.host._process, process)
|
||
self.assertIsNotNone(self.host._directory)
|
||
|
||
# Forwarding stops first; releasing this same helper needs no pkexec.
|
||
with mock.patch.object(
|
||
self.module, '_command', return_value='[]'
|
||
), mock.patch.object(self.module.subprocess, 'Popen') as launch:
|
||
self.host.cleanup()
|
||
self.host.cleanup()
|
||
|
||
launch.assert_not_called()
|
||
process.stdin.close.assert_called_once()
|
||
self.assertIsNone(self.host._process)
|
||
self.assertIsNone(self.host._directory)
|
||
|
||
bash = _bash()
|
||
|
||
if bash and Path(bash).exists():
|
||
for script in scripts:
|
||
checked = subprocess.run(
|
||
[bash, '-n', '-c', script], capture_output=True, timeout=5
|
||
)
|
||
self.assertEqual(
|
||
checked.returncode, 0, checked.stderr.decode(errors='replace')
|
||
)
|
||
|
||
def testRootHelperUsesTheSharedLauncherWithoutPkexec(self):
|
||
"""An already privileged caller keeps sing-tun's direct Bash behavior."""
|
||
process = mock.Mock()
|
||
process.poll.return_value = 0
|
||
process.wait.return_value = 0
|
||
|
||
with (
|
||
mock.patch.object(self.module.os, 'geteuid', return_value=0, create=True),
|
||
mock.patch.object(
|
||
self.module.subprocess, 'Popen', return_value=process
|
||
) as launch,
|
||
mock.patch.object(
|
||
self.module.SystemRoutingTable,
|
||
'LinuxStartPrivilegedScript',
|
||
wraps=self.module.SystemRoutingTable.LinuxStartPrivilegedScript,
|
||
) as start,
|
||
):
|
||
self.host.execute([['test', 'true']])
|
||
|
||
start.assert_called_once_with(
|
||
start.call_args.args[0],
|
||
stdin=subprocess.DEVNULL,
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL,
|
||
)
|
||
self.assertEqual(launch.call_args.args[0], ['bash', start.call_args.args[0]])
|
||
process.wait.assert_called_once_with()
|
||
|
||
self.assertIsNone(self.host._process)
|
||
self.assertIsNone(self.host._directory)
|
||
|
||
def testRecoveryWaitsForAuthorizationWithoutADeadline(self):
|
||
process = mock.Mock()
|
||
process.poll.return_value = None
|
||
entered, authorized, complete = (
|
||
threading.Event(),
|
||
threading.Event(),
|
||
threading.Event(),
|
||
)
|
||
errors = []
|
||
|
||
def wait(timeout=None):
|
||
self.assertIsNone(timeout)
|
||
entered.set()
|
||
|
||
if not authorized.wait(5):
|
||
raise RuntimeError('fixture authorization was not released')
|
||
|
||
process.poll.return_value = 0
|
||
return 0
|
||
|
||
process.wait.side_effect = wait
|
||
|
||
def recover():
|
||
try:
|
||
self.host.execute([['ip', 'link', 'delete', 'dev', 'utun101']])
|
||
except Exception as ex:
|
||
# Any non-exit exceptions
|
||
|
||
errors.append(ex)
|
||
finally:
|
||
complete.set()
|
||
|
||
with mock.patch.object(
|
||
self.module.os, 'geteuid', return_value=1000, create=True
|
||
), mock.patch.object(self.module.subprocess, 'Popen', return_value=process):
|
||
worker = threading.Thread(target=recover)
|
||
worker.start()
|
||
|
||
try:
|
||
self.assertTrue(entered.wait(2))
|
||
self.assertFalse(complete.is_set())
|
||
self.assertIs(self.host._process, process)
|
||
self.assertIsNotNone(self.host._directory)
|
||
process.terminate.assert_not_called()
|
||
process.kill.assert_not_called()
|
||
finally:
|
||
authorized.set()
|
||
worker.join(5)
|
||
|
||
self.assertFalse(worker.is_alive())
|
||
self.assertEqual(errors, [])
|
||
process.wait.assert_called_once_with()
|
||
self.assertIsNone(self.host._process)
|
||
self.assertIsNone(self.host._directory)
|
||
|
||
def testDelayedAuthorizationIsUsedForSetupDnsAndDisconnect(self):
|
||
"""A slow password entry does not create extra authorization requests."""
|
||
process = mock.Mock()
|
||
process.wait.return_value = 0
|
||
process.poll.return_value = None
|
||
clock = [0.0]
|
||
script = []
|
||
|
||
def launch(arguments, **kwargs):
|
||
script.append(Path(arguments[-1]))
|
||
return process
|
||
|
||
def advance(_seconds):
|
||
clock[0] += 3600
|
||
if clock[0] >= 86400:
|
||
script[0].parent.joinpath('ready').write_text('77', encoding='utf-8')
|
||
|
||
with mock.patch.object(
|
||
self.module.os, 'geteuid', return_value=1000, create=True
|
||
), mock.patch.object(
|
||
self.module.os, 'getuid', return_value=1000, create=True
|
||
), mock.patch.object(
|
||
self.module.subprocess, 'Popen', side_effect=launch
|
||
) as launchHelper, mock.patch.object(
|
||
self.module.time, 'monotonic', side_effect=lambda: clock[0]
|
||
), mock.patch.object(
|
||
self.module.time, 'sleep', side_effect=advance
|
||
), mock.patch.object(
|
||
self.module,
|
||
'_command',
|
||
return_value=json.dumps([{'ifname': 'utun101', 'ifindex': 77}]),
|
||
):
|
||
self.host.prepare('9.9.9.9')
|
||
self.host.applyDNS('9.9.9.9')
|
||
self.host.cancel()
|
||
self.host.drain()
|
||
process.stdin.close.assert_not_called()
|
||
|
||
with mock.patch.object(self.module, '_command', return_value='[]'):
|
||
self.host.cleanup()
|
||
|
||
self.assertEqual(clock[0], 86400)
|
||
launchHelper.assert_called_once()
|
||
process.stdin.close.assert_called_once()
|
||
self.assertFalse(self.host.dnsTouched)
|
||
|
||
def testPendingAuthorizationCancellationRetainsHelperUntilItExits(self):
|
||
process = mock.Mock()
|
||
process.poll.return_value = None
|
||
process.wait.side_effect = subprocess.TimeoutExpired('pkexec', 5)
|
||
|
||
def launch(arguments, **kwargs):
|
||
self.host.cancelled.set()
|
||
return process
|
||
|
||
with mock.patch.object(
|
||
self.module.os, 'geteuid', return_value=1000, create=True
|
||
), mock.patch.object(self.module.subprocess, 'Popen', side_effect=launch):
|
||
with self.assertRaisesRegex(RuntimeError, 'preparation was cancelled'):
|
||
self.host.execute([], preparing=True, holding=True)
|
||
|
||
directory = Path(self.host._directory.name)
|
||
self.assertTrue((directory / 'cancelled').exists())
|
||
process.stdin.close.assert_called_once()
|
||
|
||
with self.assertRaisesRegex(RuntimeError, 'still running'):
|
||
self.host.drain()
|
||
|
||
self.assertIs(self.host._process, process)
|
||
self.assertTrue(directory.exists())
|
||
|
||
process.poll.return_value = 126
|
||
self.host.drain()
|
||
|
||
self.assertFalse(directory.exists())
|
||
|
||
def testLiveForwarderDnsCannotBeReportedAsRestored(self):
|
||
"""A refused child reap retains DNS and the authorized helper together."""
|
||
plan = SingTUNHostPlan({}, platform='Linux')
|
||
plan._linux = self.host
|
||
self.host._process = process = mock.Mock()
|
||
process.poll.return_value = None
|
||
self.host._directory = tempfile.TemporaryDirectory()
|
||
self.host._holding = self.host._ready = self.host.dnsTouched = True
|
||
|
||
with self.assertRaisesRegex(
|
||
RuntimeError, 'requires stopping the forwarding child'
|
||
):
|
||
while not plan.finishDNSCleanup():
|
||
self.assertTrue(plan._done.wait(2))
|
||
plan._worker.join(2)
|
||
|
||
self.assertTrue(self.host.dnsTouched)
|
||
self.assertIs(self.host._process, process)
|
||
process.stdin.close.assert_not_called()
|
||
self.host._releaseHelper()
|
||
|
||
def testScriptRetainsSuccessfulSetupAndRollsBackFailure(self):
|
||
bash = _bash()
|
||
|
||
if not bash or not Path(bash).exists():
|
||
self.skipTest('Bash is unavailable for the isolated script fixture')
|
||
|
||
realPopen = subprocess.Popen
|
||
|
||
failures = (
|
||
('mtu', 'rule', 'dns', 'cancel', 'abandon', 'none', 'eof')
|
||
if os.name != 'nt'
|
||
else ('mtu', 'rule', 'dns', 'cancel', 'none', 'eof')
|
||
)
|
||
|
||
for failure in failures:
|
||
with self.subTest(
|
||
failure=failure
|
||
), tempfile.TemporaryDirectory() as directory:
|
||
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
|
||
# root. No real ip, PolicyKit, TUN, route or DNS command is run.
|
||
with ip.open('w', encoding='utf-8', newline='\n') as file:
|
||
file.write(
|
||
'#!/usr/bin/env bash\n'
|
||
'set -eu\n'
|
||
'device="$FAKE_NET/utun101"\n'
|
||
'case "$1 $2 $3" in\n'
|
||
'"tuntap add dev")\n'
|
||
' mkdir "$device"\n'
|
||
' echo 77 > "$device/ifindex"\n'
|
||
' if [ "$FAKE_FAILURE" = cancel ]; then touch "$FAKE_CANCEL"; fi\n'
|
||
' if [ "$FAKE_FAILURE" = abandon ]; then\n'
|
||
' rm "$FAKE_SCRIPT"\n'
|
||
' rmdir "${FAKE_CANCEL%/cancelled}"\n'
|
||
' fi\n'
|
||
' ;;\n'
|
||
'"link set dev")\n'
|
||
' if [ "$FAKE_FAILURE" = mtu ] && [ "$5" = mtu ]; then exit 1; fi\n'
|
||
' ;;\n'
|
||
'"-4 rule add"|"-6 rule add")\n'
|
||
' count=$(cat "$FAKE_NET/rules" 2>/dev/null || echo 0)\n'
|
||
' if [ "$FAKE_FAILURE" = rule ] && [ "$count" = 2 ]; then exit 1; fi\n'
|
||
' echo $((count + 1)) > "$FAKE_NET/rules"\n'
|
||
' ;;\n'
|
||
'"-4 rule del"|"-6 rule del")\n'
|
||
' count=$(cat "$FAKE_NET/rules")\n'
|
||
' echo $((count - 1)) > "$FAKE_NET/rules"\n'
|
||
' ;;\n'
|
||
'"-4 addr add"|"-6 addr add"|"-4 route add"|"-6 route add") ;;\n'
|
||
'"link delete dev")\n'
|
||
' rm "$device/ifindex"\n'
|
||
' rmdir "$device"\n'
|
||
' ;;\n'
|
||
'*) exit 90 ;;\n'
|
||
'esac\n'
|
||
)
|
||
|
||
ip.chmod(0o755)
|
||
resolver = binDirectory / 'resolvectl'
|
||
resolver.write_text(
|
||
'#!/usr/bin/env bash\n'
|
||
'echo "$1" >> "$FAKE_NET/dns"\n'
|
||
'if [ "$FAKE_FAILURE" = dns ] && [ "$1" = domain ]; then exit 1; fi\n',
|
||
encoding='utf-8',
|
||
)
|
||
resolver.chmod(0o755)
|
||
|
||
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())
|
||
cancelled = script.parent / 'cancelled'
|
||
text = text.replace(str(cancelled), cancelled.as_posix())
|
||
text = text.replace(str(script.parent), script.parent.as_posix())
|
||
|
||
binPath = shlex.quote(binDirectory.as_posix())
|
||
|
||
if os.name == 'nt':
|
||
binPath = '$(cygpath -u ' + binPath + ')'
|
||
|
||
with script.open('w', encoding='utf-8', newline='\n') as file:
|
||
file.write('PATH=' + binPath + ':"$PATH"\n' + text)
|
||
|
||
environment = os.environ.copy()
|
||
environment.update(
|
||
FAKE_NET=net.as_posix(),
|
||
FAKE_FAILURE=failure,
|
||
FAKE_CANCEL=(script.parent / 'cancelled').as_posix(),
|
||
FAKE_SCRIPT=script.as_posix(),
|
||
)
|
||
|
||
return realPopen([bash, str(script)], env=environment, **kwargs)
|
||
|
||
with mock.patch.object(
|
||
self.module.os, 'geteuid', return_value=1000, create=True
|
||
), mock.patch.object(
|
||
self.module.os, 'getuid', return_value=1000, create=True
|
||
), mock.patch.object(
|
||
self.module.subprocess, 'Popen', side_effect=launch
|
||
), mock.patch.object(
|
||
self.module,
|
||
'_command',
|
||
side_effect=lambda args: json.dumps(
|
||
[{'ifname': 'utun101', 'ifindex': 77}]
|
||
if args == ['ip', '-N', '-j', 'link', 'show']
|
||
and (net / 'utun101').exists()
|
||
else []
|
||
),
|
||
):
|
||
if failure in ('none', 'eof'):
|
||
self.host.prepare()
|
||
|
||
self.assertEqual(self.host.index, 77)
|
||
self.assertTrue((net / 'utun101').exists())
|
||
self.assertIsNone(self.host._process.poll())
|
||
if failure == 'eof':
|
||
# Closing the parent pipe also covers an application
|
||
# exit without calling the normal cleanup method.
|
||
self.host._releaseHelper()
|
||
self.assertEqual(self.host._process.wait(timeout=5), 0)
|
||
self.host.cleanup()
|
||
else:
|
||
with self.assertRaisesRegex(
|
||
RuntimeError, 'host operation failed'
|
||
):
|
||
self.host.prepare()
|
||
|
||
self.assertFalse((net / 'utun101').exists())
|
||
self.assertIsNone(self.host._process)
|
||
self.assertIsNone(self.host._directory)
|
||
if (net / 'rules').exists():
|
||
self.assertEqual((net / 'rules').read_text().strip(), '0')
|
||
if (net / 'dns').exists():
|
||
self.assertEqual(
|
||
(net / 'dns').read_text().splitlines()[-1], 'revert'
|
||
)
|
||
|
||
def testCancellationBeforeAuthorizationCannotAcquireDevice(self):
|
||
scripts = []
|
||
process = mock.Mock()
|
||
process.wait.return_value = 1
|
||
process.poll.return_value = 1
|
||
|
||
def launch(arguments, **kwargs):
|
||
directory = Path(arguments[-1]).parent
|
||
self.host.cancelled.set()
|
||
self.host.cancel()
|
||
self.assertTrue((directory / 'cancelled').exists())
|
||
scripts.append(Path(arguments[-1]).read_text(encoding='utf-8'))
|
||
return process
|
||
|
||
with mock.patch.object(
|
||
self.module.os, 'geteuid', return_value=1000, create=True
|
||
), mock.patch.object(self.module.subprocess, 'Popen', side_effect=launch):
|
||
with self.assertRaisesRegex(RuntimeError, 'authorization was cancelled'):
|
||
self.host.execute(
|
||
[['ip', 'link', 'set', 'dev', 'utun101', 'up']], preparing=True
|
||
)
|
||
|
||
self.assertLess(scripts[0].index('cancelled'), scripts[0].index('ip link'))
|
||
self.assertIsNone(self.host._process)
|
||
|
||
def testCancelledWorkDoesNotLaunchAnotherAuthorization(self):
|
||
self.host.cancelled.set()
|
||
|
||
with mock.patch.object(self.module.subprocess, 'Popen') as launch:
|
||
with self.assertRaisesRegex(RuntimeError, 'preparation was cancelled'):
|
||
self.host.execute(
|
||
[['ip', 'link', 'set', 'dev', 'utun101', 'up']], preparing=True
|
||
)
|
||
|
||
launch.assert_not_called()
|
||
self.assertIsNone(self.host._directory)
|
||
|
||
def testRootHelperRunsWithoutPolicyKitAndFailedLaunchReleasesScript(self):
|
||
with mock.patch.object(
|
||
self.module.os, 'geteuid', return_value=0, create=True
|
||
), mock.patch.object(
|
||
self.module.subprocess, 'Popen', side_effect=OSError('launch')
|
||
) as launch:
|
||
with self.assertRaisesRegex(RuntimeError, 'Unable to start'):
|
||
self.host.execute([['ip', 'link', 'set', 'dev', 'utun101', 'up']])
|
||
|
||
self.assertEqual(launch.call_args.args[0][0], 'bash')
|
||
self.assertIsNone(self.host._directory)
|
||
|
||
def testRouteRangesKeepCustomPrefixesAndAllRemoteExclusions(self):
|
||
configuration = addSingTUNExclusions(
|
||
{'tun_options': self.tun}, ['192.0.2.8', '2001:db8::8']
|
||
)
|
||
ranges = _linuxRouteRanges(configuration['tun_options'])
|
||
|
||
for address in ('192.0.2.8', '2001:db8::8'):
|
||
ip = ipaddress.ip_address(address)
|
||
self.assertFalse(
|
||
any(ip.version == prefix.version and ip in prefix for prefix in ranges)
|
||
)
|
||
|
||
self.tun['Inet4RouteAddress'] = ['203.0.113.0/24']
|
||
ranges = _linuxRouteRanges(self.tun)
|
||
self.assertEqual(
|
||
[prefix for prefix in ranges if prefix.version == 4],
|
||
[ipaddress.ip_network('203.0.113.0/24')],
|
||
)
|
||
|
||
def testIPv6LoopPreventionPrecedesSourceAndDefaultRules(self):
|
||
rules = _linuxRules(self.tun, 123456, 2000)
|
||
v6 = [command for command in rules if command[1] == '-6']
|
||
goto = next(command for command in v6 if 'utun101' in command)
|
||
source = next(command for command in v6 if 'fd00::/64' in command)
|
||
default = next(command for command in v6 if command[6:] == ['table', '123456'])
|
||
|
||
self.assertEqual(goto[5], '2001')
|
||
self.assertEqual(goto[-2:], ['goto', '2010'])
|
||
self.assertEqual(source[5], '2002')
|
||
self.assertEqual(default[5], '2004')
|
||
|
||
def testReplacementInterfaceCannotReceiveDnsOrBeDeleted(self):
|
||
self.host.index = 77
|
||
link = {'ifname': 'utun101', 'ifindex': 78}
|
||
|
||
with mock.patch.object(
|
||
self.module, '_command', return_value=json.dumps([link])
|
||
), mock.patch.object(self.host, 'execute') as execute:
|
||
with self.assertRaisesRegex(RuntimeError, 'Conflicting'):
|
||
self.host.cleanup()
|
||
with self.assertRaisesRegex(RuntimeError, 'Conflicting'):
|
||
self.host.applyDNS('1.1.1.1')
|
||
execute.assert_not_called()
|
||
|
||
def testFailedPreparationCannotClaimAnUncapturedInterface(self):
|
||
link = {'ifname': 'utun101', 'ifindex': 78}
|
||
|
||
with mock.patch.object(
|
||
self.module, '_command', return_value=json.dumps([link])
|
||
), mock.patch.object(self.host, 'execute') as execute:
|
||
with self.assertRaisesRegex(RuntimeError, 'identity was not captured'):
|
||
self.host.cleanup()
|
||
|
||
execute.assert_not_called()
|
||
self.assertIsNone(self.host.index)
|
||
|
||
def testIpRouteJsonFlagsAndSeparatePrefixLengthsMatchExactRules(self):
|
||
expected = {_linuxRuleSelectors(command) for command in self.host.rules}
|
||
records = (
|
||
(
|
||
'-4',
|
||
{
|
||
'priority': 2000,
|
||
'src': 'all',
|
||
'dst': '198.18.0.0',
|
||
'dstlen': 15,
|
||
'table': 123456,
|
||
},
|
||
),
|
||
(
|
||
'-4',
|
||
{
|
||
'priority': 2002,
|
||
'src': 'all',
|
||
'not': None,
|
||
'table': 'main',
|
||
'suppress_prefixlength': 0,
|
||
'dport': 53,
|
||
'dport_mask': '0xffff',
|
||
},
|
||
),
|
||
(
|
||
'-4',
|
||
{
|
||
'priority': 2003,
|
||
'src': 'all',
|
||
'not': None,
|
||
'iif': 'lo',
|
||
'table': 123456,
|
||
},
|
||
),
|
||
('-4', {'priority': 2003, 'src': '0.0.0.0', 'iif': 'lo', 'table': 123456}),
|
||
(
|
||
'-6',
|
||
{
|
||
'priority': 2002,
|
||
'src': 'fd00::',
|
||
'srclen': 64,
|
||
'iif': 'lo',
|
||
'table': 123456,
|
||
},
|
||
),
|
||
(
|
||
'-6',
|
||
{
|
||
'priority': 2001,
|
||
'src': 'all',
|
||
'iif': 'utun101',
|
||
'goto': 2010,
|
||
'unresolved': None,
|
||
},
|
||
),
|
||
(
|
||
'-6',
|
||
{'priority': 2003, 'src': '0', 'srclen': 1, 'iif': 'lo', 'goto': 2010},
|
||
),
|
||
('-6', {'priority': 2010, 'src': 'all', 'nop': None}),
|
||
)
|
||
|
||
for flag, record in records:
|
||
with self.subTest(rule=record):
|
||
command = _linuxRuleDelete(flag, record, 123456, 2000, 'utun101')
|
||
self.assertIn(_linuxRuleSelectors(command), expected)
|
||
|
||
with self.assertRaisesRegex(RuntimeError, 'Conflicting'):
|
||
_linuxRuleDelete(
|
||
'-4',
|
||
{
|
||
'priority': 2002,
|
||
'not': None,
|
||
'dport': 53,
|
||
'dport_mask': '0xff00',
|
||
'table': 'main',
|
||
},
|
||
123456,
|
||
2000,
|
||
'utun101',
|
||
)
|
||
|
||
def testKernelSuppressionFieldMatchesExactRule(self):
|
||
"""Use iproute2's JSON spelling, which differs from its CLI spelling."""
|
||
expected = {_linuxRuleSelectors(command) for command in self.host.rules}
|
||
rule = {
|
||
'priority': 2001,
|
||
'src': 'all',
|
||
'table': 123456,
|
||
'suppress_prefixlen': 0,
|
||
'protocol': 0,
|
||
}
|
||
command = _linuxRuleDelete('-4', rule, 123456, 2000, 'utun101')
|
||
self.assertIn(_linuxRuleSelectors(command), expected)
|
||
self.assertIn('suppress_prefixlength', command)
|
||
|
||
for value in (1, -1, '0'):
|
||
with self.subTest(value=value), self.assertRaisesRegex(
|
||
RuntimeError, 'Conflicting'
|
||
):
|
||
_linuxRuleDelete(
|
||
'-4', {**rule, 'suppress_prefixlen': value}, 123456, 2000, 'utun101'
|
||
)
|
||
|
||
def testCleanupUsesExactSelectorsAndRetainsRefusedCleanupForRetry(self):
|
||
plan = SingTUNHostPlan({}, platform='Linux')
|
||
plan._linux = self.host
|
||
plan._activated = True
|
||
link = {'ifname': 'utun101', 'ifindex': 77}
|
||
self.host.index = 77
|
||
present = True
|
||
rule = {
|
||
'priority': 2001,
|
||
'src': 'all',
|
||
'table': 123456,
|
||
'suppress_prefixlen': 0,
|
||
'protocol': 0,
|
||
}
|
||
|
||
def command(arguments):
|
||
if arguments == ['ip', '-N', '-j', 'link', 'show']:
|
||
return json.dumps([link] if present else [])
|
||
if arguments == ['ip', '-4', '-N', '-j', 'rule', 'show']:
|
||
return json.dumps([rule] if present else [])
|
||
return '[]'
|
||
|
||
with mock.patch.object(
|
||
self.module, '_command', side_effect=command
|
||
), mock.patch.object(self.host, 'restoreDNS'), mock.patch.object(
|
||
self.host, 'execute', side_effect=RuntimeError('authorization denied')
|
||
):
|
||
with self.assertRaisesRegex(RuntimeError, 'authorization denied'):
|
||
plan.cleanup(False)
|
||
|
||
self.assertTrue(plan._activated)
|
||
self.assertFalse(plan._cleanupComplete)
|
||
|
||
def release(commands, **kwargs):
|
||
nonlocal present
|
||
self.assertTrue(kwargs['guard'])
|
||
self.assertEqual(commands[-1], ['ip', 'link', 'delete', 'dev', 'utun101'])
|
||
self.assertIn('suppress_prefixlength', commands[0])
|
||
self.assertIn('123456', commands[0])
|
||
self.assertNotIn('flush', str(commands))
|
||
present = False
|
||
|
||
with mock.patch.object(
|
||
self.module, '_command', side_effect=command
|
||
), mock.patch.object(self.host, 'restoreDNS'), mock.patch.object(
|
||
self.host, 'execute', side_effect=release
|
||
) as execute:
|
||
plan.cleanup(False)
|
||
plan.cleanup(True)
|
||
|
||
execute.assert_called_once()
|
||
self.assertFalse(plan._activated)
|
||
self.assertTrue(plan._cleanupComplete)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|