Skip to content

Commit

Permalink
tests/integ/network: add type annotations
Browse files Browse the repository at this point in the history 
Make PyCharm understand what mixin those objects are for.
  • Loading branch information
@marmarek
marmarek committed Oct 14, 2018
1 parent fc3b286 commit 3756888
Showing 1 changed file with 126 additions and 16 deletions.
142 changes: 126 additions & 16 deletions qubes/tests/integ/network.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -50,13 +50,19 @@ class VmNetworkingMixin(object):
template = None

def run_cmd(self, vm, cmd, user="root"):
'''
:type self: qubes.tests.SystemTestCase | VmNetworkingMixin
'''
try:
self.loop.run_until_complete(vm.run_for_stdio(cmd, user=user))
except subprocess.CalledProcessError as e:
return e.returncode
return 0

def check_nc_version(self, vm):
'''
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
if self.run_cmd(vm, 'nc -h >/dev/null 2>&1') != 0:
self.skipTest('nc not installed')
if self.run_cmd(vm, 'nc -h 2>&1|grep -q nmap.org') == 0:
Expand All @@ -65,6 +71,9 @@ def check_nc_version(self, vm):
return NcVersion.Trad

def setUp(self):
'''
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
super(VmNetworkingMixin, self).setUp()
if self.template.startswith('whonix-'):
self.skipTest("Test not supported here - Whonix uses its own "
Expand All @@ -86,6 +95,9 @@ def setUp(self):


def configure_netvm(self):
'''
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
def run_netvm_cmd(cmd):
if self.run_cmd(self.testnetvm, cmd) != 0:
self.fail("Command '%s' failed" % cmd)
Expand Down Expand Up @@ -113,12 +125,18 @@ def run_netvm_cmd(cmd):


def test_000_simple_networking(self):
'''
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.loop.run_until_complete(self.testvm1.start())
self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0)
self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0)


def test_010_simple_proxyvm(self):
'''
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('proxy'),
label='red')
Expand All @@ -144,6 +162,9 @@ def test_010_simple_proxyvm(self):
@unittest.skipUnless(spawn.find_executable('xdotool'),
"xdotool not installed")
def test_020_simple_proxyvm_nm(self):
'''
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('proxy'),
label='red')
Expand Down Expand Up @@ -189,6 +210,9 @@ def test_020_simple_proxyvm_nm(self):


def test_030_firewallvm_firewall(self):
'''
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('proxy'),
label='red')
Expand Down Expand Up @@ -290,6 +314,9 @@ def test_030_firewallvm_firewall(self):


def test_040_inter_vm(self):
'''
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('proxy'),
label='red')
Expand Down Expand Up @@ -327,7 +354,10 @@ def test_040_inter_vm(self):
self.ping_cmd.format(target=self.testvm1.ip)), 0)

def test_050_spoof_ip(self):
"""Test if VM IP spoofing is blocked"""
'''Test if VM IP spoofing is blocked
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.loop.run_until_complete(self.testvm1.start())

self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0)
Expand All @@ -353,7 +383,10 @@ def test_050_spoof_ip(self):
self.assertEquals(packets, '0', 'Some packet hit the INPUT rule')

def test_100_late_xldevd_startup(self):
"""Regression test for #1990"""
'''Regression test for #1990
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
# Simulater late xl devd startup
cmd = "systemctl stop xendriverdomain"
if self.run_cmd(self.testnetvm, cmd) != 0:
Expand All @@ -367,7 +400,10 @@ def test_100_late_xldevd_startup(self):
self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0)

def test_200_fake_ip_simple(self):
'''Test hiding VM real IP'''
'''Test hiding VM real IP
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.testvm1.features['net.fake-ip'] = '192.168.1.128'
self.testvm1.features['net.fake-gateway'] = '192.168.1.1'
self.testvm1.features['net.fake-netmask'] = '255.255.255.0'
Expand Down Expand Up @@ -398,7 +434,10 @@ def test_200_fake_ip_simple(self):
self.assertNotIn(str(self.testvm1.netvm.ip), output)

def test_201_fake_ip_without_gw(self):
'''Test hiding VM real IP'''
'''Test hiding VM real IP
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.testvm1.features['net.fake-ip'] = '192.168.1.128'
self.app.save()
self.loop.run_until_complete(self.testvm1.start())
Expand All @@ -417,7 +456,10 @@ def test_201_fake_ip_without_gw(self):
self.assertNotIn(str(self.testvm1.ip), output)

def test_202_fake_ip_firewall(self):
'''Test hiding VM real IP, firewall'''
'''Test hiding VM real IP, firewall
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.testvm1.features['net.fake-ip'] = '192.168.1.128'
self.testvm1.features['net.fake-gateway'] = '192.168.1.1'
self.testvm1.features['net.fake-netmask'] = '255.255.255.0'
Expand Down Expand Up @@ -468,7 +510,10 @@ def test_202_fake_ip_firewall(self):
self.loop.run_until_complete(nc.wait())

def test_203_fake_ip_inter_vm_allow(self):
'''Access VM with "fake IP" from other VM (when firewall allows)'''
'''Access VM with "fake IP" from other VM (when firewall allows)
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('proxy'),
label='red')
Expand Down Expand Up @@ -521,7 +566,10 @@ def test_203_fake_ip_inter_vm_allow(self):
'Packets didn\'t managed to the VM')

def test_204_fake_ip_proxy(self):
'''Test hiding VM real IP'''
'''Test hiding VM real IP
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('proxy'),
label='red')
Expand Down Expand Up @@ -582,15 +630,21 @@ def test_204_fake_ip_proxy(self):
self.assertNotIn(str(self.proxy.ip), output)

def test_210_custom_ip_simple(self):
'''Custom AppVM IP'''
'''Custom AppVM IP
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.testvm1.ip = '192.168.1.1'
self.app.save()
self.loop.run_until_complete(self.testvm1.start())
self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0)
self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0)

def test_211_custom_ip_proxy(self):
'''Custom ProxyVM IP'''
'''Custom ProxyVM IP
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('proxy'),
label='red')
Expand All @@ -607,7 +661,10 @@ def test_211_custom_ip_proxy(self):
self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0)

def test_212_custom_ip_firewall(self):
'''Custom VM IP and firewall'''
'''Custom VM IP and firewall
:type self: qubes.tests.SystemTestCase | VMNetworkingMixin
'''
self.testvm1.ip = '192.168.1.1'

self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
Expand Down Expand Up @@ -666,6 +723,9 @@ def setUp(self):
self.ping6_name = self.ping6_cmd.format(target=self.test_name)

def configure_netvm(self):
'''
:type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
'''
self.testnetvm.features['ipv6'] = True
super(VmIPv6NetworkingMixin, self).configure_netvm()

Expand All @@ -683,12 +743,18 @@ def run_netvm_cmd(cmd):
format(ip=self.test_ip, ip6=self.test_ip6, name=self.test_name))

def test_500_ipv6_simple_networking(self):
'''
:type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
'''
self.loop.run_until_complete(self.testvm1.start())
self.assertEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0)
self.assertEqual(self.run_cmd(self.testvm1, self.ping6_name), 0)


def test_510_ipv6_simple_proxyvm(self):
'''
:type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
'''
self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('proxy'),
label='red')
Expand All @@ -714,6 +780,9 @@ def test_510_ipv6_simple_proxyvm(self):
@unittest.skipUnless(spawn.find_executable('xdotool'),
"xdotool not installed")
def test_520_ipv6_simple_proxyvm_nm(self):
'''
:type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
'''
self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('proxy'),
label='red')
Expand Down Expand Up @@ -764,6 +833,9 @@ def test_520_ipv6_simple_proxyvm_nm(self):


def test_530_ipv6_firewallvm_firewall(self):
'''
:type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
'''
self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('proxy'),
label='red')
Expand Down Expand Up @@ -881,6 +953,9 @@ def test_530_ipv6_firewallvm_firewall(self):


def test_540_ipv6_inter_vm(self):
'''
:type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
'''
self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('proxy'),
label='red')
Expand Down Expand Up @@ -920,7 +995,10 @@ def test_540_ipv6_inter_vm(self):


def test_550_ipv6_spoof_ip(self):
"""Test if VM IP spoofing is blocked"""
'''Test if VM IP spoofing is blocked
:type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
'''
self.loop.run_until_complete(self.testvm1.start())

self.assertEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0)
Expand Down Expand Up @@ -949,15 +1027,21 @@ def test_550_ipv6_spoof_ip(self):
self.assertEquals(packets, '0', 'Some packet hit the INPUT rule')

def test_710_ipv6_custom_ip_simple(self):
'''Custom AppVM IP'''
'''Custom AppVM IP
:type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
'''
self.testvm1.ip6 = '2000:aaaa:bbbb::1'
self.app.save()
self.loop.run_until_complete(self.testvm1.start())
self.assertEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0)
self.assertEqual(self.run_cmd(self.testvm1, self.ping6_name), 0)

def test_711_ipv6_custom_ip_proxy(self):
'''Custom ProxyVM IP'''
'''Custom ProxyVM IP
:type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
'''
self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
name=self.make_vm_name('proxy'),
label='red')
Expand All @@ -974,7 +1058,10 @@ def test_711_ipv6_custom_ip_proxy(self):
self.assertEqual(self.run_cmd(self.testvm1, self.ping6_name), 0)

def test_712_ipv6_custom_ip_firewall(self):
'''Custom VM IP and firewall'''
'''Custom VM IP and firewall
:type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
'''
self.testvm1.ip6 = '2000:aaaa:bbbb::1'

self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
Expand Down Expand Up @@ -1099,13 +1186,20 @@ class VmUpdatesMixin(object):
)

def run_cmd(self, vm, cmd, user="root"):
'''
:type self: qubes.tests.SystemTestCase | VmUpdatesMixin
'''
try:
self.loop.run_until_complete(vm.run_for_stdio(cmd))
except subprocess.CalledProcessError as e:
return e.returncode
return 0

def setUp(self):
'''
:type self: qubes.tests.SystemTestCase | VmUpdatesMixin
'''
if not self.template.count('debian') and \
not self.template.count('fedora'):
self.skipTest("Template {} not supported by this test".format(
Expand Down Expand Up @@ -1142,6 +1236,9 @@ def setUp(self):
self.loop.run_until_complete(self.testvm1.create_on_disk())

def test_000_simple_update(self):
'''
:type self: qubes.tests.SystemTestCase | VmUpdatesMixin
'''
self.app.save()
# reload the VM to have all the properties properly set (especially
# default netvm)
Expand All @@ -1155,6 +1252,9 @@ def test_000_simple_update(self):
'{}: {}\n{}'.format(self.update_cmd, stdout, stderr))

def create_repo_apt(self):
'''
:type self: qubes.tests.SystemTestCase | VmUpdatesMixin
'''
pkg_file_name = "test-pkg_1.0-1_amd64.deb"
self.loop.run_until_complete(self.netvm_repo.run_for_stdio('''
mkdir /tmp/apt-repo \
Expand Down Expand Up @@ -1209,6 +1309,9 @@ def create_repo_apt(self):
'''))

def create_repo_yum(self):
'''
:type self: qubes.tests.SystemTestCase | VmUpdatesMixin
'''
pkg_file_name = "test-pkg-1.0-1.fc21.x86_64.rpm"
self.loop.run_until_complete(self.netvm_repo.run_for_stdio('''
mkdir /tmp/yum-repo \
Expand All @@ -1221,6 +1324,9 @@ def create_repo_yum(self):
'createrepo /tmp/yum-repo'))

def create_repo_and_serve(self):
'''
:type self: qubes.tests.SystemTestCase | VmUpdatesMixin
'''
if self.template.count("debian") or self.template.count("whonix"):
self.create_repo_apt()
self.loop.run_until_complete(self.netvm_repo.run(
Expand All @@ -1242,6 +1348,8 @@ def configure_test_repo(self):
The critical part is to use "localhost" - this will work only when
accessed through update proxy and this is exactly what we want to
test here.
:type self: qubes.tests.SystemTestCase | VmUpdatesMixin
"""

if self.template.count("debian") or self.template.count("whonix"):
Expand All @@ -1266,9 +1374,11 @@ def configure_test_repo(self):
self.template))

def test_010_update_via_proxy(self):
"""
'''
Test both whether updates proxy works and whether is actually used by the VM
"""
:type self: qubes.tests.SystemTestCase | VmUpdatesMixin
'''
if self.template.count("minimal"):
self.skipTest("Template {} not supported by this test".format(
self.template))
Expand Down

0 comments on commit 3756888

Please sign in to comment.