extern/sshuttle
1
1
Fork 0
mirror of https://github.com/sshuttle/sshuttle.git synced 2024-11-07 16:44:23 +01:00
Code Issues Packages Projects Releases Wiki Activity

Fix/pep8 (#277)

Browse Source 
* re-organized imports according to pep8
* fixed all remaining pep8 issues
* moved common config into setup.cfg, additionally test `tests`
* removed --select=X -- the errors selected where by default not in
  flake8's --ignore list so effectively had no effect
* update .travis.yml to reflect changes in tox.ini
* make travis just use tox in order to avoid code duplaction
* replace py.test with pytest
* fixed .travis.yml
* try different pypy toxenv
* hopefully fixed testenv for pypy
* added pypy basepython, removed unused python2.6
* install dev package before testing (fixes missing coverage)
* fixed empty exception pass blocks with noqa
* Added dummy log message on empty try-except-pass blocks to make dodacy happy :(
* Replaced Exception with BaseException
This commit is contained in:
Bastian Venthur 2019-02-10 23:59:13 +01:00 committed by Brian May
parent 752a953101
commit 3bfb975ed9
25 changed files with 148 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
.travis.yml
View File

@ -1,19 +1,20 @@
language: python
python:
- 2.7
- 3.4
- 3.5
- 3.6
- pypy
matrix:
include:
- python: 2.7
env: TOXENV=py27
- python: 3.4
env: TOXENV=py34
- python: 3.5
env: TOXENV=py35
- python: 3.6
env: TOXENV=py36
- python: pypy
env: TOXENV=pypy
install:
- travis_retry pip install -q -r requirements-tests.txt
before_script:
# stop the build if there are Python syntax errors or undefined names.
- flake8 sshuttle --count --select=E901,E999,F821,F822,F823 --show-source --statistics
# exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide.
- flake8 sshuttle --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
- pip install tox
script:
- PYTHONPATH=. py.test
- tox

5
setup.cfg
View File

@ -8,5 +8,10 @@ universal = 1
sign=true
identity=0x1784577F811F6EAC
[flake8]
count=true
show-source=true
statistics=true
[tool:pytest]
addopts = --cov=sshuttle --cov-branch --cov-report=term-missing

7
sshuttle/assembler.py
View File

@ -2,6 +2,10 @@ import sys
import zlib
import imp
import sshuttle.helpers
import sshuttle.cmdline_options as options
from sshuttle.server import main
verbosity = verbosity # noqa: F821 must be a previously defined global
z = zlib.decompressobj()
while 1:
@ -30,10 +34,7 @@ while 1:
sys.stderr.flush()
sys.stdout.flush()
import sshuttle.helpers
sshuttle.helpers.verbose = verbosity
import sshuttle.cmdline_options as options
from sshuttle.server import main
main(options.latency_control, options.auto_hosts, options.to_nameserver,
options.auto_nets)

23
sshuttle/client.py
View File

@ -3,13 +3,14 @@ import re
import signal
import time
import subprocess as ssubprocess
import sshuttle.helpers as helpers
import os
import sys
import platform
import sshuttle.helpers as helpers
import sshuttle.ssnet as ssnet
import sshuttle.ssh as ssh
import sshuttle.ssyslog as ssyslog
import sys
import platform
from sshuttle.ssnet import SockWrapper, Handler, Proxy, Mux, MuxWrapper
from sshuttle.helpers import log, debug1, debug2, debug3, Fatal, islocal, \
resolvconf_nameservers
@ -268,11 +269,13 @@ class FirewallClient:
self.pfile.write(b'ROUTES\n')
for (family, ip, width, fport, lport) \
in self.subnets_include + self.auto_nets:
self.pfile.write(b'%d,%d,0,%s,%d,%d\n'
% (family, width, ip.encode("ASCII"), fport, lport))
self.pfile.write(b'%d,%d,0,%s,%d,%d\n' % (family, width,
ip.encode("ASCII"),
fport, lport))
for (family, ip, width, fport, lport) in self.subnets_exclude:
self.pfile.write(b'%d,%d,1,%s,%d,%d\n'
% (family, width, ip.encode("ASCII"), fport, lport))
self.pfile.write(b'%d,%d,1,%s,%d,%d\n' % (family, width,
ip.encode("ASCII"),
fport, lport))
self.pfile.write(b'NSLIST\n')
for (family, ip) in self.nslist:
@ -495,7 +498,8 @@ def _main(tcp_listener, udp_listener, fw, ssh_cmd, remotename,
def onroutes(routestr):
if auto_nets:
for line in routestr.strip().split(b'\n'):
if not line: continue
if not line:
continue
(family, ip, width) = line.split(b',', 2)
family = int(family)
width = int(width)
@ -707,7 +711,8 @@ def main(listenip_v6, listenip_v4,
ports = range(12300, 9000, -1)
for port in ports:
debug2(' %d' % port)
if port in used_ports: continue
if port in used_ports:
continue
dns_listener = MultiListener(socket.SOCK_DGRAM)

33
sshuttle/firewall.py
View File

@ -1,12 +1,13 @@
import errno
import socket
import signal
import sshuttle.ssyslog as ssyslog
import sshuttle.sdnotify as sdnotify
import sys
import os
import platform
import traceback
import sshuttle.ssyslog as ssyslog
import sshuttle.sdnotify as sdnotify
from sshuttle.helpers import debug1, debug2, Fatal
from sshuttle.methods import get_auto_method, get_method
@ -132,7 +133,7 @@ def main(method_name, syslog):
try:
(family, width, exclude, ip, fport, lport) = \
line.strip().split(',', 5)
except:
except BaseException:
raise Fatal('firewall: expected route or NSLIST but got %r' % line)
subnets.append((
int(family),
@ -154,7 +155,7 @@ def main(method_name, syslog):
break
try:
(family, ip) = line.strip().split(',', 1)
except:
except BaseException:
raise Fatal('firewall: expected nslist or PORTS but got %r' % line)
nslist.append((int(family), ip))
debug2('firewall manager: Got partial nslist: %r\n' % nslist)
@ -220,7 +221,7 @@ def main(method_name, syslog):
stdout.write('STARTED\n')
sdnotify.send(sdnotify.ready(),
sdnotify.status('Connected'))
sdnotify.status('Connected'))
try:
stdout.flush()
@ -248,43 +249,43 @@ def main(method_name, syslog):
try:
sdnotify.send(sdnotify.stop())
debug1('firewall manager: undoing changes.\n')
except:
pass
except BaseException:
debug2('An error occurred, ignoring it.')
try:
if subnets_v6 or nslist_v6:
debug2('firewall manager: undoing IPv6 changes.\n')
method.restore_firewall(port_v6, socket.AF_INET6, udp, user)
except:
except BaseException:
try:
debug1("firewall manager: "
"Error trying to undo IPv6 firewall.\n")
for line in traceback.format_exc().splitlines():
debug1("---> %s\n" % line)
except:
pass
except BaseException:
debug2('An error occurred, ignoring it.')
try:
if subnets_v4 or nslist_v4:
debug2('firewall manager: undoing IPv4 changes.\n')
method.restore_firewall(port_v4, socket.AF_INET, udp, user)
except:
except BaseException:
try:
debug1("firewall manager: "
"Error trying to undo IPv4 firewall.\n")
for line in traceback.format_exc().splitlines():
debug1("firewall manager: ---> %s\n" % line)
except:
pass
except BaseException:
debug2('An error occurred, ignoring it.')
try:
debug2('firewall manager: undoing /etc/hosts changes.\n')
restore_etc_hosts(port_v6 or port_v4)
except:
except BaseException:
try:
debug1("firewall manager: "
"Error trying to undo /etc/hosts changes.\n")
for line in traceback.format_exc().splitlines():
debug1("firewall manager: ---> %s\n" % line)
except:
pass
except BaseException:
debug2('An error occurred, ignoring it.')

16
sshuttle/helpers.py
View File

@ -56,22 +56,22 @@ class Fatal(Exception):
def resolvconf_nameservers():
l = []
lines = []
for line in open('/etc/resolv.conf'):
words = line.lower().split()
if len(words) >= 2 and words[0] == 'nameserver':
l.append(family_ip_tuple(words[1]))
return l
lines.append(family_ip_tuple(words[1]))
return lines
def resolvconf_random_nameserver():
l = resolvconf_nameservers()
if l:
if len(l) > 1:
lines = resolvconf_nameservers()
if lines:
if len(lines) > 1:
# don't import this unless we really need it
import random
random.shuffle(l)
return l[0]
random.shuffle(lines)
return lines[0]
else:
return (socket.AF_INET, '127.0.0.1')

2
sshuttle/hostwatch.py
View File

@ -44,7 +44,7 @@ def write_host_cache():
finally:
try:
os.unlink(tmpname)
except:
except BaseException:
pass

1
sshuttle/linux.py
View File

@ -2,6 +2,7 @@ import re
import os
import socket
import subprocess as ssubprocess
from sshuttle.helpers import log, debug1, Fatal, family_to_string

27
sshuttle/methods/ipfw.py
View File

@ -29,8 +29,8 @@ IPV6_RECVDSTADDR = 74
if recvmsg == "python":
def recv_udp(listener, bufsize):
debug3('Accept UDP python using recvmsg.\n')
data, ancdata, _, srcip = \
listener.recvmsg(4096, socket.CMSG_SPACE(4))
data, ancdata, _, srcip = listener.recvmsg(4096,
socket.CMSG_SPACE(4))
dstip = None
for cmsg_level, cmsg_type, cmsg_data in ancdata:
if cmsg_level == socket.SOL_IP and cmsg_type == IP_RECVDSTADDR:
@ -42,8 +42,8 @@ if recvmsg == "python":
elif recvmsg == "socket_ext":
def recv_udp(listener, bufsize):
debug3('Accept UDP using socket_ext recvmsg.\n')
srcip, data, adata, _ = \
listener.recvmsg((bufsize,), socket.CMSG_SPACE(4))
srcip, data, adata, _ = listener.recvmsg((bufsize,),
socket.CMSG_SPACE(4))
dstip = None
for a in adata:
if a.cmsg_level == socket.SOL_IP and a.cmsg_type == IP_RECVDSTADDR:
@ -134,6 +134,7 @@ def sysctl_set(name, val, permanent=False):
_changedctls.append(name)
return True
def ipfw(*args):
argv = ['ipfw', '-q'] + list(args)
debug1('>> %s\n' % ' '.join(argv))
@ -147,12 +148,13 @@ def ipfw_noexit(*args):
debug1('>> %s\n' % ' '.join(argv))
ssubprocess.call(argv)
class Method(BaseMethod):
def get_supported_features(self):
result = super(Method, self).get_supported_features()
result.ipv6 = False
result.udp = False #NOTE: Almost there, kernel patch needed
result.udp = False # NOTE: Almost there, kernel patch needed
result.dns = True
return result
@ -175,21 +177,21 @@ class Method(BaseMethod):
"couldn't determine source IP address\n" % (dstip,))
return
#debug3('Sending SRC: %r DST: %r\n' % (srcip, dstip))
# debug3('Sending SRC: %r DST: %r\n' % (srcip, dstip))
sender = socket.socket(sock.family, socket.SOCK_DGRAM)
sender.setsockopt(socket.SOL_IP, IP_BINDANY, 1)
sender.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sender.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
sender.setsockopt(socket.SOL_IP, socket.IP_TTL, 42)
sender.bind(srcip)
sender.sendto(data,dstip)
sender.sendto(data, dstip)
sender.close()
def setup_udp_listener(self, udp_listener):
if udp_listener.v4 is not None:
udp_listener.v4.setsockopt(socket.SOL_IP, IP_RECVDSTADDR, 1)
#if udp_listener.v6 is not None:
# udp_listener.v6.setsockopt(SOL_IPV6, IPV6_RECVDSTADDR, 1)
# if udp_listener.v6 is not None:
# udp_listener.v6.setsockopt(SOL_IPV6, IPV6_RECVDSTADDR, 1)
def setup_firewall(self, port, dnsport, nslist, family, subnets, udp,
user):
@ -199,7 +201,7 @@ class Method(BaseMethod):
'Address family "%s" unsupported by ipfw method_name'
% family_to_string(family))
#XXX: Any risk from this?
# XXX: Any risk from this?
ipfw_noexit('delete', '1')
while _changedctls:
@ -238,8 +240,9 @@ class Method(BaseMethod):
if subnets:
# create new subnet entries
for _, swidth, sexclude, snet \
in sorted(subnets, key=lambda s: s[1], reverse=True):
for _, swidth, sexclude, snet in sorted(subnets,
key=lambda s: s[1],
reverse=True):
if sexclude:
ipfw('table', '125', 'add', '%s/%s' % (snet, swidth))
else:

25
sshuttle/methods/pf.py
View File

@ -35,11 +35,11 @@ class Generic(object):
class pf_addr(Structure):
class _pfa(Union):
_fields_ = [("v4", c_uint32), # struct in_addr
("v6", c_uint32 * 4), # struct in6_addr
("addr8", c_uint8 * 16),
("addr16", c_uint16 * 8),
("addr32", c_uint32 * 4)]
_fields_ = [("v4", c_uint32), # struct in_addr
("v6", c_uint32 * 4), # struct in6_addr
("addr8", c_uint8 * 16),
("addr16", c_uint16 * 8),
("addr32", c_uint32 * 4)]
_fields_ = [("pfa", _pfa)]
_anonymous_ = ("pfa",)
@ -120,16 +120,18 @@ class Generic(object):
pr = self.pfioc_rule()
memmove(addressof(pr) + self.ANCHOR_CALL_OFFSET, name,
min(self.MAXPATHLEN, len(name))) # anchor_call = name
min(self.MAXPATHLEN, len(name))) # anchor_call = name
memmove(addressof(pr) + self.RULE_ACTION_OFFSET,
struct.pack('I', kind), 4) # rule.action = kind
struct.pack('I', kind), 4) # rule.action = kind
memmove(addressof(pr) + self.ACTION_OFFSET, struct.pack(
'I', self.PF_CHANGE_GET_TICKET), 4) # action = PF_CHANGE_GET_TICKET
memmove(addressof(pr) + self.ACTION_OFFSET,
struct.pack('I', self.PF_CHANGE_GET_TICKET),
4) # action = PF_CHANGE_GET_TICKET
ioctl(pf_get_dev(), pf.DIOCCHANGERULE, pr)
memmove(addressof(pr) + self.ACTION_OFFSET, struct.pack(
'I', self.PF_CHANGE_ADD_TAIL), 4) # action = PF_CHANGE_ADD_TAIL
memmove(addressof(pr) + self.ACTION_OFFSET,
struct.pack('I', self.PF_CHANGE_ADD_TAIL),
4) # action = PF_CHANGE_ADD_TAIL
ioctl(pf_get_dev(), pf.DIOCCHANGERULE, pr)
@staticmethod
@ -151,7 +153,6 @@ class Generic(object):
return b'skip' in pfctl('-s Interfaces -i lo -v')[0]
class FreeBsd(Generic):
RULE_ACTION_OFFSET = 2968

1
sshuttle/methods/tproxy.py
View File

@ -169,7 +169,6 @@ class Method(BaseMethod):
return proto + ('--dport', '%d:%d' % (fport, lport)) \
if fport else proto
mark_chain = 'sshuttle-m-%s' % port
tproxy_chain = 'sshuttle-t-%s' % port
divert_chain = 'sshuttle-d-%s' % port

1
sshuttle/options.py
View File

@ -1,6 +1,7 @@
import re
import socket
from argparse import ArgumentParser, Action, ArgumentTypeError as Fatal
from sshuttle import __version__

6
sshuttle/sdnotify.py
View File

@ -1,7 +1,9 @@
import socket
import os
from sshuttle.helpers import debug1
def _notify(message):
addr = os.environ.get("NOTIFY_SOCKET", None)
@ -27,14 +29,18 @@ def _notify(message):
debug1("Error notifying systemd: %s\n" % e)
return False
def send(*messages):
return _notify(b'\n'.join(messages))
def ready():
return b"READY=1"
def stop():
return b"STOPPING=1"
def status(message):
return b"STATUS=%s" % message.encode('utf8')

1
sshuttle/server.py
View File

@ -289,7 +289,6 @@ def main(latency_control, auto_hosts, to_nameserver, auto_nets):
helpers.logprefix = 'server: '
debug1('latency control setting = %r\n' % latency_control)
# synchronization header
sys.stdout.write('\0\0SSHUTTLE0001')
sys.stdout.flush()

9
sshuttle/ssh.py
View File

@ -6,6 +6,7 @@ import zlib
import imp
import subprocess as ssubprocess
import shlex
import sshuttle.helpers as helpers
from sshuttle.helpers import debug2
@ -76,10 +77,10 @@ def connect(ssh_cmd, rhostport, python, stderr, options):
else:
rhost = rhostport
else: # IPv4
l = (rhostport or '').rsplit(':', 1)
rhost = l[0]
if len(l) > 1:
portl = ['-p', str(int(l[1]))]
host_port = (rhostport or '').rsplit(':', 1)
rhost = host_port[0]
if len(host_port) > 1:
portl = ['-p', str(int(host_port[1]))]
if rhost == '-':
rhost = None

1
sshuttle/ssnet.py
View File

@ -4,6 +4,7 @@ import socket
import errno
import select
import os
from sshuttle.helpers import b, binary_type, log, debug1, debug2, debug3, Fatal
MAX_CHANNEL = 65535

7
tests/client/test_firewall.py
View File

@ -1,7 +1,7 @@
from mock import Mock, patch, call
import io
from socket import AF_INET, AF_INET6
from mock import Mock, patch, call
import sshuttle.firewall
@ -85,8 +85,9 @@ def test_subnet_weight():
(AF_INET, 0, 1, '0.0.0.0', 0, 0)
]
assert subnets_sorted == \
sorted(subnets, key=sshuttle.firewall.subnet_weight, reverse=True)
assert subnets_sorted == sorted(subnets,
key=sshuttle.firewall.subnet_weight,
reverse=True)
@patch('sshuttle.firewall.rewrite_etc_hosts')

2
tests/client/test_helpers.py
View File

@ -1,10 +1,10 @@
from mock import patch, call
import sys
import io
import socket
from socket import AF_INET, AF_INET6
import errno
from mock import patch, call
import sshuttle.helpers

8
tests/client/test_methods_nat.py
View File

@ -1,9 +1,9 @@
import pytest
from mock import Mock, patch, call
import socket
from socket import AF_INET, AF_INET6
import struct
import pytest
from mock import Mock, patch, call
from sshuttle.helpers import Fatal
from sshuttle.methods import get_method
@ -124,7 +124,7 @@ def test_setup_firewall(mock_ipt_chain_exists, mock_ipt_ttl, mock_ipt):
]
assert mock_ipt_ttl.mock_calls == [
call(AF_INET, 'nat', '-A', 'sshuttle-1025', '-j', 'REDIRECT',
'--dest', u'1.2.3.0/24', '-p', 'tcp', '--dport', '8000:9000',
'--dest', u'1.2.3.0/24', '-p', 'tcp', '--dport', '8000:9000',
'--to-ports', '1025'),
call(AF_INET, 'nat', '-A', 'sshuttle-1025', '-j', 'REDIRECT',
'--dest', u'1.2.3.33/32', '-p', 'udp',
@ -140,7 +140,7 @@ def test_setup_firewall(mock_ipt_chain_exists, mock_ipt_ttl, mock_ipt):
call(AF_INET, 'nat', '-I', 'OUTPUT', '1', '-j', 'sshuttle-1025'),
call(AF_INET, 'nat', '-I', 'PREROUTING', '1', '-j', 'sshuttle-1025'),
call(AF_INET, 'nat', '-A', 'sshuttle-1025', '-j', 'RETURN',
'--dest', u'1.2.3.66/32', '-p', 'tcp', '--dport', '8080:8080')
'--dest', u'1.2.3.66/32', '-p', 'tcp', '--dport', '8080:8080')
]
mock_ipt_chain_exists.reset_mock()
mock_ipt_ttl.reset_mock()

4
tests/client/test_methods_pf.py
View File

@ -1,8 +1,8 @@
import pytest
from mock import Mock, patch, call, ANY
import socket
from socket import AF_INET, AF_INET6
import pytest
from mock import Mock, patch, call, ANY
from sshuttle.methods import get_method
from sshuttle.helpers import Fatal
from sshuttle.methods.pf import FreeBsd, Darwin, OpenBsd

4
tests/client/test_methods_tproxy.py
View File

@ -132,7 +132,7 @@ def test_setup_firewall(mock_ipt_chain_exists, mock_ipt_ttl, mock_ipt):
call(AF_INET6, 'mangle', '-F', 'sshuttle-t-1024'),
call(AF_INET6, 'mangle', '-I', 'OUTPUT', '1', '-j', 'sshuttle-m-1024'),
call(AF_INET6, 'mangle', '-I', 'PREROUTING', '1', '-j',
'sshuttle-t-1024'),
'sshuttle-t-1024'),
call(AF_INET6, 'mangle', '-A', 'sshuttle-d-1024', '-j', 'MARK',
'--set-mark', '1'),
call(AF_INET6, 'mangle', '-A', 'sshuttle-d-1024', '-j', 'ACCEPT'),
@ -232,7 +232,7 @@ def test_setup_firewall(mock_ipt_chain_exists, mock_ipt_ttl, mock_ipt):
call(AF_INET, 'mangle', '-F', 'sshuttle-t-1025'),
call(AF_INET, 'mangle', '-I', 'OUTPUT', '1', '-j', 'sshuttle-m-1025'),
call(AF_INET, 'mangle', '-I', 'PREROUTING', '1', '-j',
'sshuttle-t-1025'),
'sshuttle-t-1025'),
call(AF_INET, 'mangle', '-A', 'sshuttle-d-1025',
'-j', 'MARK', '--set-mark', '1'),
call(AF_INET, 'mangle', '-A', 'sshuttle-d-1025', '-j', 'ACCEPT'),

27
tests/client/test_options.py
View File

@ -1,8 +1,10 @@
import socket
import pytest
import sshuttle.options
from argparse import ArgumentTypeError as Fatal
import pytest
import sshuttle.options
_ip4_reprs = {
'0.0.0.0': '0.0.0.0',
'255.255.255.255': '255.255.255.255',
@ -25,6 +27,7 @@ _ip6_reprs = {
_ip6_swidths = (48, 64, 96, 115, 128)
def test_parse_subnetport_ip4():
for ip_repr, ip in _ip4_reprs.items():
assert sshuttle.options.parse_subnetport(ip_repr) \
@ -41,7 +44,7 @@ def test_parse_subnetport_ip4_with_mask():
'/'.join((ip_repr, str(swidth)))
) == (socket.AF_INET, ip, swidth, 0, 0)
assert sshuttle.options.parse_subnetport('0/0') \
== (socket.AF_INET, '0.0.0.0', 0, 0, 0)
== (socket.AF_INET, '0.0.0.0', 0, 0, 0)
with pytest.raises(Fatal) as excinfo:
sshuttle.options.parse_subnetport('10.0.0.0/33')
assert str(excinfo.value) == 'width 33 is not between 0 and 32'
@ -50,17 +53,17 @@ def test_parse_subnetport_ip4_with_mask():
def test_parse_subnetport_ip4_with_port():
for ip_repr, ip in _ip4_reprs.items():
assert sshuttle.options.parse_subnetport(':'.join((ip_repr, '80'))) \
== (socket.AF_INET, ip, 32, 80, 80)
== (socket.AF_INET, ip, 32, 80, 80)
assert sshuttle.options.parse_subnetport(':'.join((ip_repr, '80-90')))\
== (socket.AF_INET, ip, 32, 80, 90)
== (socket.AF_INET, ip, 32, 80, 90)
def test_parse_subnetport_ip4_with_mask_and_port():
for ip_repr, ip in _ip4_reprs.items():
assert sshuttle.options.parse_subnetport(ip_repr + '/32:80') \
== (socket.AF_INET, ip, 32, 80, 80)
== (socket.AF_INET, ip, 32, 80, 80)
assert sshuttle.options.parse_subnetport(ip_repr + '/16:80-90') \
== (socket.AF_INET, ip, 16, 80, 90)
== (socket.AF_INET, ip, 16, 80, 90)
def test_parse_subnetport_ip6():
@ -76,7 +79,7 @@ def test_parse_subnetport_ip6_with_mask():
'/'.join((ip_repr, str(swidth)))
) == (socket.AF_INET6, ip, swidth, 0, 0)
assert sshuttle.options.parse_subnetport('::/0') \
== (socket.AF_INET6, '::', 0, 0, 0)
== (socket.AF_INET6, '::', 0, 0, 0)
with pytest.raises(Fatal) as excinfo:
sshuttle.options.parse_subnetport('fc00::/129')
assert str(excinfo.value) == 'width 129 is not between 0 and 128'
@ -85,14 +88,14 @@ def test_parse_subnetport_ip6_with_mask():
def test_parse_subnetport_ip6_with_port():
for ip_repr, ip in _ip6_reprs.items():
assert sshuttle.options.parse_subnetport('[' + ip_repr + ']:80') \
== (socket.AF_INET6, ip, 128, 80, 80)
== (socket.AF_INET6, ip, 128, 80, 80)
assert sshuttle.options.parse_subnetport('[' + ip_repr + ']:80-90') \
== (socket.AF_INET6, ip, 128, 80, 90)
== (socket.AF_INET6, ip, 128, 80, 90)
def test_parse_subnetport_ip6_with_mask_and_port():
for ip_repr, ip in _ip6_reprs.items():
assert sshuttle.options.parse_subnetport('[' + ip_repr + '/128]:80') \
== (socket.AF_INET6, ip, 128, 80, 80)
== (socket.AF_INET6, ip, 128, 80, 80)
assert sshuttle.options.parse_subnetport('[' + ip_repr + '/16]:80-90')\
== (socket.AF_INET6, ip, 16, 80, 90)
== (socket.AF_INET6, ip, 16, 80, 90)

3
tests/client/test_sdnotify.py
View File

@ -1,6 +1,7 @@
from mock import Mock, patch, call
import socket
from mock import Mock, patch, call
import sshuttle.sdnotify

6
tests/server/test_server.py
View File

@ -1,8 +1,10 @@
import io
import socket
import sshuttle.server
from mock import patch, Mock
import sshuttle.server
def test__ipmatch():
assert sshuttle.server._ipmatch("1.2.3.4") is not None
@ -45,7 +47,7 @@ Destination Gateway Genmask Flags MSS Window irtt Iface
def test_listroutes_iproute(mock_popen, mock_which):
mock_pobj = Mock()
mock_pobj.stdout = io.BytesIO(b"""
default via 192.168.1.1 dev wlan0 proto static
default via 192.168.1.1 dev wlan0 proto static
192.168.1.0/24 dev wlan0 proto kernel scope link src 192.168.1.1
""")
mock_pobj.wait.return_value = 0

12
tox.ini
View File

@ -5,17 +5,21 @@ envlist =
py34,
py35,
py36,
pypy,
[testenv]
basepython =
py26: python2.6
py27: python2.7
py34: python3.4
py35: python3.5
py36: python3.6
pypy: pypy
commands =
flake8 sshuttle --count --select=E901,E999,F821,F822,F823 --show-source --statistics
flake8 sshuttle --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
py.test
pip install -e .
# actual flake8 test
flake8 sshuttle tests
# flake8 complexity warnings
flake8 sshuttle tests --exit-zero --max-complexity=10
pytest
deps =
-rrequirements-tests.txt