sshuttle/tests/client/test_firewall.py
Scott Kuhl bc065e368d Remove ttl hack & require -r option.
Previously, it was possible to run sshuttle locally without using ssh
and connecting to a remote server. In this configuration, traffic was
redirected to the sshuttle server running on the localhost. However,
the firewall needed to distinguish between traffic leaving the
sshuttle server and traffic that originated from the machine that
still needed to be routed through the sshuttle server. The TTL of the
packets leaving the sshuttle server were manipulated to indicate to
the firewall what should happen. The TTL was adjusted for all packets
leaving the sshuttle server (even if it wasn't necessary because the
server and client were running on different machines).

Changing the TTL caused trouble and some machines, and
the --ttl option was added as a workaround to change how the TTL was
set for traffic leaving sshuttle. All of this added complexity to the
code for a feature (running the server on localhost) that is likely
only used for testing and rarely used by others.

This commit updates the associated documentation, but doesn't fully
fix the ipfw method since I am unable to test that.

This change will also make sshuttle fail to work if -r is used to
specify a localhost. Pull request #610 partially addresses that issue.

For example, see: #240, #490, #660, #606.
2021-07-12 11:24:29 -04:00

142 lines
4.4 KiB
Python

import io
from socket import AF_INET, AF_INET6
from unittest.mock import Mock, patch, call
import sshuttle.firewall
def setup_daemon():
stdin = io.StringIO(u"""ROUTES
{inet},24,0,1.2.3.0,8000,9000
{inet},32,1,1.2.3.66,8080,8080
{inet6},64,0,2404:6800:4004:80c::,0,0
{inet6},128,1,2404:6800:4004:80c::101f,80,80
NSLIST
{inet},1.2.3.33
{inet6},2404:6800:4004:80c::33
PORTS 1024,1025,1026,1027
GO 1 - 0x01
HOST 1.2.3.3,existing
""".format(inet=AF_INET, inet6=AF_INET6))
stdout = Mock()
return stdin, stdout
def test_rewrite_etc_hosts(tmpdir):
orig_hosts = tmpdir.join("hosts.orig")
orig_hosts.write("1.2.3.3 existing\n")
new_hosts = tmpdir.join("hosts")
orig_hosts.copy(new_hosts)
hostmap = {
'myhost': '1.2.3.4',
'myotherhost': '1.2.3.5',
}
with patch('sshuttle.firewall.HOSTSFILE', new=str(new_hosts)):
sshuttle.firewall.rewrite_etc_hosts(hostmap, 10)
with new_hosts.open() as f:
line = f.readline()
s = line.split()
assert s == ['1.2.3.3', 'existing']
line = f.readline()
s = line.split()
assert s == ['1.2.3.4', 'myhost',
'#', 'sshuttle-firewall-10', 'AUTOCREATED']
line = f.readline()
s = line.split()
assert s == ['1.2.3.5', 'myotherhost',
'#', 'sshuttle-firewall-10', 'AUTOCREATED']
line = f.readline()
assert line == ""
with patch('sshuttle.firewall.HOSTSFILE', new=str(new_hosts)):
sshuttle.firewall.restore_etc_hosts(hostmap, 10)
assert orig_hosts.computehash() == new_hosts.computehash()
def test_subnet_weight():
subnets = [
(AF_INET, 16, 0, '192.168.0.0', 0, 0),
(AF_INET, 24, 0, '192.168.69.0', 0, 0),
(AF_INET, 32, 0, '192.168.69.70', 0, 0),
(AF_INET, 32, 1, '192.168.69.70', 0, 0),
(AF_INET, 32, 1, '192.168.69.70', 80, 80),
(AF_INET, 0, 1, '0.0.0.0', 0, 0),
(AF_INET, 0, 1, '0.0.0.0', 8000, 9000),
(AF_INET, 0, 1, '0.0.0.0', 8000, 8500),
(AF_INET, 0, 1, '0.0.0.0', 8000, 8000),
(AF_INET, 0, 1, '0.0.0.0', 400, 450)
]
subnets_sorted = [
(AF_INET, 32, 1, '192.168.69.70', 80, 80),
(AF_INET, 0, 1, '0.0.0.0', 8000, 8000),
(AF_INET, 0, 1, '0.0.0.0', 400, 450),
(AF_INET, 0, 1, '0.0.0.0', 8000, 8500),
(AF_INET, 0, 1, '0.0.0.0', 8000, 9000),
(AF_INET, 32, 1, '192.168.69.70', 0, 0),
(AF_INET, 32, 0, '192.168.69.70', 0, 0),
(AF_INET, 24, 0, '192.168.69.0', 0, 0),
(AF_INET, 16, 0, '192.168.0.0', 0, 0),
(AF_INET, 0, 1, '0.0.0.0', 0, 0)
]
assert subnets_sorted == sorted(subnets,
key=sshuttle.firewall.subnet_weight,
reverse=True)
@patch('sshuttle.firewall.rewrite_etc_hosts')
@patch('sshuttle.firewall.setup_daemon')
@patch('sshuttle.firewall.get_method')
def test_main(mock_get_method, mock_setup_daemon, mock_rewrite_etc_hosts):
stdin, stdout = setup_daemon()
mock_setup_daemon.return_value = stdin, stdout
mock_get_method("not_auto").name = "test"
mock_get_method.reset_mock()
sshuttle.firewall.main("not_auto", False)
assert mock_rewrite_etc_hosts.mock_calls == [
call({'1.2.3.3': 'existing'}, 1024),
call({}, 1024),
]
assert stdout.mock_calls == [
call.write('READY test\n'),
call.flush(),
call.write('STARTED\n'),
call.flush()
]
assert mock_setup_daemon.mock_calls == [call()]
assert mock_get_method.mock_calls == [
call('not_auto'),
call().is_supported(),
call().is_supported().__bool__(),
call().setup_firewall(
1024, 1026,
[(AF_INET6, u'2404:6800:4004:80c::33')],
AF_INET6,
[(AF_INET6, 64, False, u'2404:6800:4004:80c::', 0, 0),
(AF_INET6, 128, True, u'2404:6800:4004:80c::101f', 80, 80)],
True,
None,
'0x01'),
call().setup_firewall(
1025, 1027,
[(AF_INET, u'1.2.3.33')],
AF_INET,
[(AF_INET, 24, False, u'1.2.3.0', 8000, 9000),
(AF_INET, 32, True, u'1.2.3.66', 8080, 8080)],
True,
None,
'0x01'),
call().restore_firewall(1024, AF_INET6, True, None),
call().restore_firewall(1025, AF_INET, True, None),
]