network_ipv6.py 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451
  1. #
  2. # The Qubes OS Project, https://www.qubes-os.org/
  3. #
  4. # Copyright (C) 2015-2020
  5. # Marek Marczykowski-Górecki <marmarek@invisiblethingslab.com>
  6. # Copyright (C) 2015 Wojtek Porczyk <woju@invisiblethingslab.com>
  7. #
  8. # This library is free software; you can redistribute it and/or
  9. # modify it under the terms of the GNU Lesser General Public
  10. # License as published by the Free Software Foundation; either
  11. # version 2.1 of the License, or (at your option) any later version.
  12. #
  13. # This library is distributed in the hope that it will be useful,
  14. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16. # Lesser General Public License for more details.
  17. #
  18. # You should have received a copy of the GNU Lesser General Public
  19. # License along with this library; if not, see <https://www.gnu.org/licenses/>.
  20. #
  21. import asyncio
  22. import subprocess
  23. import sys
  24. import time
  25. import unittest
  26. from distutils import spawn
  27. import qubes.firewall
  28. import qubes.tests
  29. import qubes.vm
  30. from qubes.tests.integ.network import VmNetworkingMixin
  31. # noinspection PyAttributeOutsideInit,PyPep8Naming
  32. class VmIPv6NetworkingMixin(VmNetworkingMixin):
  33. test_ip6 = '2000:abcd::1'
  34. ping6_cmd = 'ping6 -W 1 -n -c 1 {target}'
  35. def setUp(self):
  36. super(VmIPv6NetworkingMixin, self).setUp()
  37. self.ping6_ip = self.ping6_cmd.format(target=self.test_ip6)
  38. self.ping6_name = self.ping6_cmd.format(target=self.test_name)
  39. def tearDown(self):
  40. # collect more info on failure (ipv4 info collected in parent)
  41. if self._outcome and not self._outcome.success:
  42. for vm in (self.testnetvm, self.testvm1, getattr(self, 'proxy', None)):
  43. if vm is None:
  44. continue
  45. self._run_cmd_and_log_output(vm, 'ip -6 r')
  46. self._run_cmd_and_log_output(vm, 'ip6tables -vnL')
  47. self._run_cmd_and_log_output(vm, 'ip6tables -vnL -t nat')
  48. self._run_cmd_and_log_output(vm, 'nft list table ip6 qubes-firewall')
  49. super().tearDown()
  50. def configure_netvm(self):
  51. '''
  52. :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
  53. '''
  54. self.testnetvm.features['ipv6'] = True
  55. super(VmIPv6NetworkingMixin, self).configure_netvm()
  56. def run_netvm_cmd(cmd):
  57. try:
  58. self.loop.run_until_complete(
  59. self.testnetvm.run_for_stdio(cmd, user='root'))
  60. except subprocess.CalledProcessError as e:
  61. self.fail("Command '%s' failed: %s%s" %
  62. (cmd, e.stdout.decode(), e.stderr.decode()))
  63. run_netvm_cmd("ip addr add {}/128 dev test0".format(self.test_ip6))
  64. run_netvm_cmd(
  65. "ip6tables -I INPUT -d {} -j ACCEPT".format(self.test_ip6))
  66. # ignore failure
  67. self.run_cmd(self.testnetvm, "pkill dnsmasq")
  68. run_netvm_cmd(
  69. "dnsmasq -a {ip} -A /{name}/{ip} -A /{name}/{ip6} -i test0 -z".
  70. format(ip=self.test_ip, ip6=self.test_ip6, name=self.test_name))
  71. def test_500_ipv6_simple_networking(self):
  72. '''
  73. :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
  74. '''
  75. self.loop.run_until_complete(self.testvm1.start())
  76. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0)
  77. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_name), 0)
  78. def test_510_ipv6_simple_proxyvm(self):
  79. '''
  80. :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
  81. '''
  82. self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  83. name=self.make_vm_name('proxy'),
  84. label='red')
  85. self.proxy.provides_network = True
  86. self.proxy.netvm = self.testnetvm
  87. self.loop.run_until_complete(self.proxy.create_on_disk())
  88. self.testvm1.netvm = self.proxy
  89. self.app.save()
  90. self.loop.run_until_complete(self.testvm1.start())
  91. self.assertTrue(self.proxy.is_running())
  92. self.assertEqual(self.run_cmd(self.proxy, self.ping6_ip), 0,
  93. "Ping by IP from ProxyVM failed")
  94. self.assertEqual(self.run_cmd(self.proxy, self.ping6_name), 0,
  95. "Ping by name from ProxyVM failed")
  96. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0,
  97. "Ping by IP from AppVM failed")
  98. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_name), 0,
  99. "Ping by IP from AppVM failed")
  100. @qubes.tests.expectedFailureIfTemplate('debian-7')
  101. @unittest.skipUnless(spawn.find_executable('xdotool'),
  102. "xdotool not installed")
  103. def test_520_ipv6_simple_proxyvm_nm(self):
  104. '''
  105. :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
  106. '''
  107. self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  108. name=self.make_vm_name('proxy'),
  109. label='red')
  110. self.proxy.provides_network = True
  111. self.loop.run_until_complete(self.proxy.create_on_disk())
  112. self.proxy.netvm = self.testnetvm
  113. self.proxy.features['service.network-manager'] = True
  114. self.testvm1.netvm = self.proxy
  115. self.app.save()
  116. self.loop.run_until_complete(self.testvm1.start())
  117. self.assertTrue(self.proxy.is_running())
  118. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0,
  119. "Ping by IP failed")
  120. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_name), 0,
  121. "Ping by name failed")
  122. # reconnect to make sure that device was configured by NM
  123. self.assertEqual(
  124. self.run_cmd(self.proxy, "nmcli device disconnect eth0",
  125. user="user"),
  126. 0, "Failed to disconnect eth0 using nmcli")
  127. self.assertNotEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0,
  128. "Network should be disabled, but apparently it isn't")
  129. self.assertEqual(
  130. self.run_cmd(self.proxy,
  131. 'nmcli connection up "VM uplink eth0" ifname eth0',
  132. user="user"),
  133. 0, "Failed to connect eth0 using nmcli")
  134. self.assertEqual(self.run_cmd(self.proxy, "nm-online",
  135. user="user"), 0,
  136. "Failed to wait for NM connection")
  137. # wait for duplicate-address-detection to complete - by default it has
  138. # 1s timeout
  139. time.sleep(2)
  140. # check for nm-applet presence
  141. self.assertEqual(subprocess.call([
  142. 'xdotool', 'search', '--class', '{}:nm-applet'.format(
  143. self.proxy.name)],
  144. stdout=subprocess.DEVNULL), 0, "nm-applet window not found")
  145. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0,
  146. "Ping by IP failed (after NM reconnection")
  147. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_name), 0,
  148. "Ping by name failed (after NM reconnection)")
  149. def test_530_ipv6_firewallvm_firewall(self):
  150. '''
  151. :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
  152. '''
  153. self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  154. name=self.make_vm_name('proxy'),
  155. label='red')
  156. self.proxy.provides_network = True
  157. self.loop.run_until_complete(self.proxy.create_on_disk())
  158. self.proxy.netvm = self.testnetvm
  159. self.testvm1.netvm = self.proxy
  160. self.app.save()
  161. # block all for first
  162. self.testvm1.firewall.rules = [qubes.firewall.Rule(action='drop')]
  163. self.testvm1.firewall.save()
  164. self.loop.run_until_complete(self.testvm1.start())
  165. self.assertTrue(self.proxy.is_running())
  166. server = self.loop.run_until_complete(self.testnetvm.run(
  167. 'socat TCP6-LISTEN:1234,fork EXEC:/bin/uname'))
  168. try:
  169. self.assertEqual(self.run_cmd(self.proxy, self.ping6_ip), 0,
  170. "Ping by IP from ProxyVM failed")
  171. self.assertEqual(self.run_cmd(self.proxy, self.ping6_name), 0,
  172. "Ping by name from ProxyVM failed")
  173. self.assertNotEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0,
  174. "Ping by IP should be blocked")
  175. client6_cmd = "socat TCP:[{}]:1234 -".format(self.test_ip6)
  176. client4_cmd = "socat TCP:{}:1234 -".format(self.test_ip)
  177. self.assertNotEqual(self.run_cmd(self.testvm1, client6_cmd), 0,
  178. "TCP connection should be blocked")
  179. # block all except ICMP
  180. self.testvm1.firewall.rules = [(
  181. qubes.firewall.Rule(None, action='accept', proto='icmp')
  182. )]
  183. self.testvm1.firewall.save()
  184. # Ugly hack b/c there is no feedback when the rules are actually
  185. # applied
  186. time.sleep(3)
  187. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0,
  188. "Ping by IP failed (should be allowed now)")
  189. self.assertNotEqual(self.run_cmd(self.testvm1, self.ping6_name), 0,
  190. "Ping by name should be blocked")
  191. # all TCP still blocked
  192. self.testvm1.firewall.rules = [
  193. qubes.firewall.Rule(None, action='accept', proto='icmp'),
  194. qubes.firewall.Rule(None, action='accept', specialtarget='dns'),
  195. ]
  196. self.testvm1.firewall.save()
  197. # Ugly hack b/c there is no feedback when the rules are actually
  198. # applied
  199. time.sleep(3)
  200. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_name), 0,
  201. "Ping by name failed (should be allowed now)")
  202. self.assertNotEqual(self.run_cmd(self.testvm1, client6_cmd), 0,
  203. "TCP connection should be blocked")
  204. # block all except target
  205. self.testvm1.firewall.rules = [
  206. qubes.firewall.Rule(None, action='accept',
  207. dsthost=self.test_ip6,
  208. proto='tcp', dstports=1234),
  209. ]
  210. self.testvm1.firewall.save()
  211. # Ugly hack b/c there is no feedback when the rules are actually
  212. # applied
  213. time.sleep(3)
  214. self.assertEqual(self.run_cmd(self.testvm1, client6_cmd), 0,
  215. "TCP connection failed (should be allowed now)")
  216. # block all except target - by name
  217. self.testvm1.firewall.rules = [
  218. qubes.firewall.Rule(None, action='accept',
  219. dsthost=self.test_name,
  220. proto='tcp', dstports=1234),
  221. ]
  222. self.testvm1.firewall.save()
  223. # Ugly hack b/c there is no feedback when the rules are actually
  224. # applied
  225. time.sleep(3)
  226. self.assertEqual(self.run_cmd(self.testvm1, client6_cmd), 0,
  227. "TCP (IPv6) connection failed (should be allowed now)")
  228. self.assertEqual(self.run_cmd(self.testvm1, client4_cmd),
  229. 0,
  230. "TCP (IPv4) connection failed (should be allowed now)")
  231. # allow all except target
  232. self.testvm1.firewall.rules = [
  233. qubes.firewall.Rule(None, action='drop', dsthost=self.test_ip6,
  234. proto='tcp', dstports=1234),
  235. qubes.firewall.Rule(action='accept'),
  236. ]
  237. self.testvm1.firewall.save()
  238. # Ugly hack b/c there is no feedback when the rules are actually
  239. # applied
  240. time.sleep(3)
  241. self.assertNotEqual(self.run_cmd(self.testvm1, client6_cmd), 0,
  242. "TCP connection should be blocked")
  243. finally:
  244. server.terminate()
  245. self.loop.run_until_complete(server.wait())
  246. def test_540_ipv6_inter_vm(self):
  247. '''
  248. :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
  249. '''
  250. self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  251. name=self.make_vm_name('proxy'),
  252. label='red')
  253. self.loop.run_until_complete(self.proxy.create_on_disk())
  254. self.proxy.provides_network = True
  255. self.proxy.netvm = self.testnetvm
  256. self.testvm1.netvm = self.proxy
  257. self.testvm2 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  258. name=self.make_vm_name('vm2'),
  259. label='red')
  260. self.loop.run_until_complete(self.testvm2.create_on_disk())
  261. self.testvm2.netvm = self.proxy
  262. self.app.save()
  263. self.loop.run_until_complete(asyncio.wait([
  264. self.testvm1.start(),
  265. self.testvm2.start()]))
  266. self.assertNotEqual(self.run_cmd(self.testvm1,
  267. self.ping_cmd.format(target=self.testvm2.ip6)), 0)
  268. self.testvm2.netvm = self.testnetvm
  269. self.assertNotEqual(self.run_cmd(self.testvm1,
  270. self.ping_cmd.format(target=self.testvm2.ip6)), 0)
  271. self.assertNotEqual(self.run_cmd(self.testvm2,
  272. self.ping_cmd.format(target=self.testvm1.ip6)), 0)
  273. self.testvm1.netvm = self.testnetvm
  274. self.assertNotEqual(self.run_cmd(self.testvm1,
  275. self.ping_cmd.format(target=self.testvm2.ip6)), 0)
  276. self.assertNotEqual(self.run_cmd(self.testvm2,
  277. self.ping_cmd.format(target=self.testvm1.ip6)), 0)
  278. def test_550_ipv6_spoof_ip(self):
  279. '''Test if VM IP spoofing is blocked
  280. :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
  281. '''
  282. self.loop.run_until_complete(self.testvm1.start())
  283. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0)
  284. # add a simple rule counting packets
  285. self.assertEqual(self.run_cmd(self.testnetvm,
  286. 'ip6tables -I INPUT -i vif+ ! -s {} -p icmpv6 -j LOG'.format(
  287. self.testvm1.ip6)), 0)
  288. self.loop.run_until_complete(self.testvm1.run_for_stdio(
  289. 'ip -6 addr flush dev eth0 && '
  290. 'ip -6 addr add {}/128 dev eth0 && '
  291. 'ip -6 route add default via {} dev eth0'.format(
  292. str(self.testvm1.visible_ip6) + '1',
  293. str(self.testvm1.visible_gateway6)),
  294. user='root'))
  295. self.assertNotEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0,
  296. "Spoofed ping should be blocked")
  297. try:
  298. (output, _) = self.loop.run_until_complete(
  299. self.testnetvm.run_for_stdio('ip6tables -nxvL INPUT',
  300. user='root'))
  301. except subprocess.CalledProcessError:
  302. self.fail('ip6tables -nxvL INPUT failed')
  303. output = output.decode().splitlines()
  304. packets = output[2].lstrip().split()[0]
  305. self.assertEquals(packets, '0', 'Some packet hit the INPUT rule')
  306. def test_710_ipv6_custom_ip_simple(self):
  307. '''Custom AppVM IP
  308. :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
  309. '''
  310. self.testvm1.ip6 = '2000:aaaa:bbbb::1'
  311. self.app.save()
  312. self.loop.run_until_complete(self.testvm1.start())
  313. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0)
  314. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_name), 0)
  315. def test_711_ipv6_custom_ip_proxy(self):
  316. '''Custom ProxyVM IP
  317. :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
  318. '''
  319. self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  320. name=self.make_vm_name('proxy'),
  321. label='red')
  322. self.loop.run_until_complete(self.proxy.create_on_disk())
  323. self.proxy.provides_network = True
  324. self.proxy.netvm = self.testnetvm
  325. self.testvm1.ip6 = '2000:aaaa:bbbb::1'
  326. self.testvm1.netvm = self.proxy
  327. self.app.save()
  328. self.loop.run_until_complete(self.testvm1.start())
  329. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0)
  330. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_name), 0)
  331. def test_712_ipv6_custom_ip_firewall(self):
  332. '''Custom VM IP and firewall
  333. :type self: qubes.tests.SystemTestCase | VmIPv6NetworkingMixin
  334. '''
  335. self.testvm1.ip6 = '2000:aaaa:bbbb::1'
  336. self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
  337. name=self.make_vm_name('proxy'),
  338. label='red')
  339. self.proxy.provides_network = True
  340. self.loop.run_until_complete(self.proxy.create_on_disk())
  341. self.proxy.netvm = self.testnetvm
  342. self.testvm1.netvm = self.proxy
  343. self.app.save()
  344. # block all but ICMP and DNS
  345. self.testvm1.firewall.rules = [
  346. qubes.firewall.Rule(None, action='accept', proto='icmp'),
  347. qubes.firewall.Rule(None, action='accept', specialtarget='dns'),
  348. ]
  349. self.testvm1.firewall.save()
  350. self.loop.run_until_complete(self.testvm1.start())
  351. self.assertTrue(self.proxy.is_running())
  352. server = self.loop.run_until_complete(self.testnetvm.run(
  353. 'socat TCP6-LISTEN:1234,fork EXEC:/bin/uname'))
  354. try:
  355. self.assertEqual(self.run_cmd(self.proxy, self.ping6_ip), 0,
  356. "Ping by IP from ProxyVM failed")
  357. self.assertEqual(self.run_cmd(self.proxy, self.ping6_name), 0,
  358. "Ping by name from ProxyVM failed")
  359. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_ip), 0,
  360. "Ping by IP should be allowed")
  361. self.assertEqual(self.run_cmd(self.testvm1, self.ping6_name), 0,
  362. "Ping by name should be allowed")
  363. client_cmd = "socat TCP:[{}]:1234 -".format(self.test_ip6)
  364. self.assertNotEqual(self.run_cmd(self.testvm1, client_cmd), 0,
  365. "TCP connection should be blocked")
  366. finally:
  367. server.terminate()
  368. self.loop.run_until_complete(server.wait())
  369. def create_testcases_for_templates():
  370. yield from qubes.tests.create_testcases_for_templates('VmIPv6Networking',
  371. VmIPv6NetworkingMixin, qubes.tests.SystemTestCase,
  372. module=sys.modules[__name__])
  373. def load_tests(loader, tests, pattern):
  374. tests.addTests(loader.loadTestsFromNames(
  375. create_testcases_for_templates()))
  376. return tests
  377. qubes.tests.maybe_create_testcases_on_import(create_testcases_for_templates)