network.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346
  1. #!/usr/bin/python
  2. # vim: fileencoding=utf-8
  3. #
  4. # The Qubes OS Project, https://www.qubes-os.org/
  5. #
  6. # Copyright (C) 2015
  7. # Marek Marczykowski-Górecki <marmarek@invisiblethingslab.com>
  8. # Copyright (C) 2015 Wojtek Porczyk <woju@invisiblethingslab.com>
  9. #
  10. # This program is free software; you can redistribute it and/or modify
  11. # it under the terms of the GNU General Public License as published by
  12. # the Free Software Foundation; either version 2 of the License, or
  13. # (at your option) any later version.
  14. #
  15. # This program is distributed in the hope that it will be useful,
  16. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  17. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  18. # GNU General Public License for more details.
  19. #
  20. # You should have received a copy of the GNU General Public License along
  21. # with this program; if not, write to the Free Software Foundation, Inc.,
  22. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  23. #
  24. from distutils import spawn
  25. import multiprocessing
  26. import os
  27. import subprocess
  28. import unittest
  29. import time
  30. from qubes.qubes import QubesVmCollection, defaults
  31. import qubes.tests
  32. class NcVersion:
  33. Trad = 1
  34. Nmap = 2
  35. class VmNetworkingMixin(qubes.tests.SystemTestsMixin):
  36. test_ip = '192.168.123.45'
  37. test_name = 'test.example.com'
  38. ping_cmd = 'ping -W 1 -n -c 1 {target}'
  39. ping_ip = ping_cmd.format(target=test_ip)
  40. ping_name = ping_cmd.format(target=test_name)
  41. def run_cmd(self, vm, cmd, user="root"):
  42. p = vm.run(cmd, user=user, passio_popen=True, ignore_stderr=True)
  43. p.stdin.close()
  44. p.stdout.read()
  45. return p.wait()
  46. def setUp(self):
  47. super(VmNetworkingMixin, self).setUp()
  48. self.testnetvm = self.qc.add_new_vm("QubesNetVm",
  49. name=self.make_vm_name('netvm1'),
  50. template=self.qc.get_vm_by_name(self.template))
  51. self.testnetvm.create_on_disk(verbose=False)
  52. self.testvm1 = self.qc.add_new_vm("QubesAppVm",
  53. name=self.make_vm_name('vm2'),
  54. template=self.qc.get_vm_by_name(self.template))
  55. self.testvm1.create_on_disk(verbose=False)
  56. self.testvm1.netvm = self.testnetvm
  57. self.qc.save()
  58. self.configure_netvm()
  59. def configure_netvm(self):
  60. def run_netvm_cmd(cmd):
  61. if self.run_cmd(self.testnetvm, cmd) != 0:
  62. self.fail("Command '%s' failed" % cmd)
  63. if not self.testnetvm.is_running():
  64. self.testnetvm.start()
  65. # Ensure that dnsmasq is installed:
  66. p = self.testnetvm.run("dnsmasq --version", user="root",
  67. passio_popen=True)
  68. if p.wait() != 0:
  69. self.skipTest("dnsmasq not installed")
  70. run_netvm_cmd("ip link add test0 type dummy")
  71. run_netvm_cmd("ip link set test0 up")
  72. run_netvm_cmd("ip addr add {}/24 dev test0".format(self.test_ip))
  73. run_netvm_cmd("iptables -I INPUT -d {} -j ACCEPT".format(self.test_ip))
  74. run_netvm_cmd("dnsmasq -a {ip} -A /{name}/{ip} -i test0 -z".format(
  75. ip=self.test_ip, name=self.test_name))
  76. run_netvm_cmd("echo nameserver {} > /etc/resolv.conf".format(
  77. self.test_ip))
  78. run_netvm_cmd("/usr/lib/qubes/qubes-setup-dnat-to-ns")
  79. def test_000_simple_networking(self):
  80. self.qc.unlock_db()
  81. self.testvm1.start()
  82. self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0)
  83. self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0)
  84. def test_010_simple_proxyvm(self):
  85. self.proxy = self.qc.add_new_vm("QubesProxyVm",
  86. name=self.make_vm_name('proxy'),
  87. template=self.qc.get_vm_by_name(self.template))
  88. self.proxy.create_on_disk(verbose=False)
  89. self.proxy.netvm = self.testnetvm
  90. self.testvm1.netvm = self.proxy
  91. self.qc.save()
  92. self.qc.unlock_db()
  93. self.testvm1.start()
  94. self.assertTrue(self.proxy.is_running())
  95. self.assertEqual(self.run_cmd(self.proxy, self.ping_ip), 0,
  96. "Ping by IP from ProxyVM failed")
  97. self.assertEqual(self.run_cmd(self.proxy, self.ping_name), 0,
  98. "Ping by name from ProxyVM failed")
  99. self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0,
  100. "Ping by IP from AppVM failed")
  101. self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0,
  102. "Ping by IP from AppVM failed")
  103. @unittest.skipUnless(spawn.find_executable('xdotool'),
  104. "xdotool not installed")
  105. def test_020_simple_proxyvm_nm(self):
  106. self.proxy = self.qc.add_new_vm("QubesProxyVm",
  107. name=self.make_vm_name('proxy'),
  108. template=self.qc.get_vm_by_name(self.template))
  109. self.proxy.create_on_disk(verbose=False)
  110. self.proxy.netvm = self.testnetvm
  111. self.proxy.services['network-manager'] = True
  112. self.testvm1.netvm = self.proxy
  113. self.qc.save()
  114. self.qc.unlock_db()
  115. self.testvm1.start()
  116. self.assertTrue(self.proxy.is_running())
  117. self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0,
  118. "Ping by IP failed")
  119. self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0,
  120. "Ping by name failed")
  121. # reconnect to make sure that device was configured by NM
  122. self.assertEqual(
  123. self.run_cmd(self.proxy, "nmcli device disconnect eth0",
  124. user="user"),
  125. 0, "Failed to disconnect eth0 using nmcli")
  126. self.assertNotEqual(self.run_cmd(self.testvm1, self.ping_ip), 0,
  127. "Network should be disabled, but apparently it isn't")
  128. self.assertEqual(
  129. self.run_cmd(self.proxy,
  130. 'nmcli connection up "VM uplink eth0" ifname eth0',
  131. user="user"),
  132. 0, "Failed to connect eth0 using nmcli")
  133. self.assertEqual(self.run_cmd(self.proxy, "nm-online", user="user"), 0,
  134. "Failed to wait for NM connection")
  135. # check for nm-applet presence
  136. self.assertEqual(subprocess.call([
  137. 'xdotool', 'search', '--all', '--name',
  138. '--class', '^(NetworkManager Applet|{})$'.format(self.proxy.name)],
  139. stdout=open('/dev/null', 'w')), 0, "nm-applet window not found")
  140. self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0,
  141. "Ping by IP failed (after NM reconnection")
  142. self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0,
  143. "Ping by name failed (after NM reconnection)")
  144. def test_030_firewallvm_firewall(self):
  145. self.proxy = self.qc.add_new_vm("QubesProxyVm",
  146. name=self.make_vm_name('proxy'),
  147. template=self.qc.get_vm_by_name(self.template))
  148. self.proxy.create_on_disk(verbose=False)
  149. self.proxy.netvm = self.testnetvm
  150. self.testvm1.netvm = self.proxy
  151. self.qc.save()
  152. self.qc.unlock_db()
  153. if self.run_cmd(self.testnetvm, 'nc -h 2>&1|grep -q nmap.org') == 0:
  154. nc_version = NcVersion.Nmap
  155. else:
  156. nc_version = NcVersion.Trad
  157. # block all for first
  158. self.testvm1.write_firewall_conf({
  159. 'allow': False,
  160. 'allowDns': False,
  161. 'allowIcmp': False,
  162. })
  163. self.testvm1.start()
  164. self.assertTrue(self.proxy.is_running())
  165. if nc_version == NcVersion.Nmap:
  166. self.testnetvm.run("nc -l --send-only -e /bin/hostname -k 1234")
  167. else:
  168. self.testnetvm.run("while nc -l -e /bin/hostname -p 1234; do "
  169. "true; done")
  170. self.assertEqual(self.run_cmd(self.proxy, self.ping_ip), 0,
  171. "Ping by IP from ProxyVM failed")
  172. self.assertEqual(self.run_cmd(self.proxy, self.ping_name), 0,
  173. "Ping by name from ProxyVM failed")
  174. self.assertNotEqual(self.run_cmd(self.testvm1, self.ping_ip), 0,
  175. "Ping by IP should be blocked")
  176. if nc_version == NcVersion.Nmap:
  177. nc_cmd = "nc -w 1 --recv-only {} 1234".format(self.test_ip)
  178. else:
  179. nc_cmd = "nc -w 1 {} 1234".format(self.test_ip)
  180. self.assertNotEqual(self.run_cmd(self.testvm1, nc_cmd), 0,
  181. "TCP connection should be blocked")
  182. # block all except ICMP
  183. self.testvm1.write_firewall_conf({
  184. 'allow': False,
  185. 'allowDns': False,
  186. 'allowIcmp': True,
  187. })
  188. self.proxy.write_iptables_qubesdb_entry()
  189. # Ugly hack b/c there is no feedback when the rules are actually applied
  190. time.sleep(3)
  191. self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0,
  192. "Ping by IP failed (should be allowed now)")
  193. self.assertNotEqual(self.run_cmd(self.testvm1, self.ping_name), 0,
  194. "Ping by name should be blocked")
  195. # all TCP still blocked
  196. self.testvm1.write_firewall_conf({
  197. 'allow': False,
  198. 'allowDns': True,
  199. 'allowIcmp': True,
  200. })
  201. self.proxy.write_iptables_qubesdb_entry()
  202. # Ugly hack b/c there is no feedback when the rules are actually applied
  203. time.sleep(3)
  204. self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0,
  205. "Ping by name failed (should be allowed now)")
  206. self.assertNotEqual(self.run_cmd(self.testvm1, nc_cmd), 0,
  207. "TCP connection should be blocked")
  208. # block all except target
  209. self.testvm1.write_firewall_conf({
  210. 'allow': False,
  211. 'allowDns': True,
  212. 'allowIcmp': True,
  213. 'rules': [{'address': self.test_ip,
  214. 'netmask': 32,
  215. 'proto': 'tcp',
  216. 'portBegin': 1234
  217. }] })
  218. self.proxy.write_iptables_qubesdb_entry()
  219. # Ugly hack b/c there is no feedback when the rules are actually applied
  220. time.sleep(3)
  221. self.assertEqual(self.run_cmd(self.testvm1, nc_cmd), 0,
  222. "TCP connection failed (should be allowed now)")
  223. # allow all except target
  224. self.testvm1.write_firewall_conf({
  225. 'allow': True,
  226. 'allowDns': True,
  227. 'allowIcmp': True,
  228. 'rules': [{'address': self.test_ip,
  229. 'netmask': 32,
  230. 'proto': 'tcp',
  231. 'portBegin': 1234
  232. }]
  233. })
  234. self.proxy.write_iptables_qubesdb_entry()
  235. # Ugly hack b/c there is no feedback when the rules are actually applied
  236. time.sleep(3)
  237. self.assertNotEqual(self.run_cmd(self.testvm1, nc_cmd), 0,
  238. "TCP connection should be blocked")
  239. def test_040_inter_vm(self):
  240. self.proxy = self.qc.add_new_vm("QubesProxyVm",
  241. name=self.make_vm_name('proxy'),
  242. template=self.qc.get_vm_by_name(self.template))
  243. self.proxy.create_on_disk(verbose=False)
  244. self.proxy.netvm = self.testnetvm
  245. self.testvm1.netvm = self.proxy
  246. self.testvm2 = self.qc.add_new_vm("QubesAppVm",
  247. name=self.make_vm_name('vm3'),
  248. template=self.qc.get_vm_by_name(self.template))
  249. self.testvm2.create_on_disk(verbose=False)
  250. self.testvm2.netvm = self.proxy
  251. self.qc.save()
  252. self.qc.unlock_db()
  253. self.testvm1.start()
  254. self.testvm2.start()
  255. self.assertNotEqual(self.run_cmd(self.testvm1,
  256. self.ping_cmd.format(target=self.testvm2.ip)), 0)
  257. self.testvm2.netvm = self.testnetvm
  258. self.assertNotEqual(self.run_cmd(self.testvm1,
  259. self.ping_cmd.format(target=self.testvm2.ip)), 0)
  260. self.assertNotEqual(self.run_cmd(self.testvm2,
  261. self.ping_cmd.format(target=self.testvm1.ip)), 0)
  262. self.testvm1.netvm = self.testnetvm
  263. self.assertNotEqual(self.run_cmd(self.testvm1,
  264. self.ping_cmd.format(target=self.testvm2.ip)), 0)
  265. self.assertNotEqual(self.run_cmd(self.testvm2,
  266. self.ping_cmd.format(target=self.testvm1.ip)), 0)
  267. def test_050_spoof_ip(self):
  268. """Test if VM IP spoofing is blocked"""
  269. self.qc.unlock_db()
  270. self.testvm1.start()
  271. self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0)
  272. self.testvm1.run("ip addr flush dev eth0", user="root")
  273. self.testvm1.run("ip addr add 10.137.1.128/24 dev eth0", user="root")
  274. self.testvm1.run("ip route add dev eth0", user="root")
  275. self.assertNotEqual(self.run_cmd(self.testvm1, self.ping_ip), 0,
  276. "Spoofed ping should be blocked")
  277. def load_tests(loader, tests, pattern):
  278. try:
  279. qc = qubes.qubes.QubesVmCollection()
  280. qc.lock_db_for_reading()
  281. qc.load()
  282. qc.unlock_db()
  283. templates = [vm.name for vm in qc.values() if
  284. isinstance(vm, qubes.qubes.QubesTemplateVm)]
  285. except OSError:
  286. templates = []
  287. for template in templates:
  288. tests.addTests(loader.loadTestsFromTestCase(
  289. type(
  290. 'VmNetworking_' + template,
  291. (VmNetworkingMixin, qubes.tests.QubesTestCase),
  292. {'template': template})))
  293. return tests