Add purging of no longer allowed connections from conntrack

Previously adding a firewall rule didn't close already established connections:
QubesOS/qubes-issues#4141

qubesagent/firewall.py

@@ -23,6 +23,7 @@
 import logging
 import os
 import socket
+import ipaddress
 import subprocess
 import pwd
 import shutil
@@ -180,6 +181,86 @@ def update_handled(self, addr):
     def list_targets(self):
         return set(t.split('/')[2] for t in self.qdb.list('/qubes-firewall/'))
 
+    def conntrack_drop(self, src, con):
+        subprocess.run(['conntrack', '-D', '--src', src, '--dst', con[1],
+                        '--proto', con[0], '--dport', con[2]],
+                       stdout=subprocess.PIPE,
+                       stderr=subprocess.STDOUT)
+
+    def conntrack_get_connections(self, family, source):
+        connections = set()
+
+        with subprocess.Popen(['conntrack', '-L',
+                               '--family', f'ipv{family}', '--src', source],
+                             stdout=subprocess.PIPE,
+                             stderr=subprocess.STDOUT) as p:
+            while True:
+                line = p.stdout.readline()
+                print(line)
+                if not line:
+                    break
+
+                line_split = line.decode().split(' ')
+
+                proto = line_split[0]
+                dst = None
+                dport = None
+                for i in line_split:
+                    if i.startswith('dst='):
+                        dst = i[len('dst='):]
+                    elif i.startswith('dport='):
+                        dport = i[len('dport='):]
+                        break
+
+                if not dst or not dport:
+                    continue
+
+                connections.add((proto, dst, dport))
+
+        return connections
+
+    def is_blocked(self, rules, con, dns):
+        con_proto, con_dst, con_dport = con
+
+        family = 6 if self.is_ip6(con_dst) else 4
+        dns_servers = list(self.dns_addresses(family))
+        fullmask = '/128' if family == 6 else '/32'
+
+        for rule in rules:
+            if rule.get('proto') and rule['proto'] != con_proto:
+                continue
+
+            if rule.get('dstports'):
+                if '-' in rule['dstports']:
+                    rule_port_range = rule['dstports'].split('-')
+                    if not (rule_port_range[0] <= con_dport and \
+                            con_dport <= rule_port_range[1]):
+                        continue
+                else:
+                    if con_dport != rule['dstports']:
+                        continue
+
+            if family == 4 and rule.get('dst6') or \
+               family == 6 and rule.get('dst4'):
+                continue
+
+            if rule.get(f'dst{family}'):
+                if not ipaddress.ip_address(con_dst) in \
+                   ipaddress.ip_network(rule[f'dst{family}'], False):
+                    continue
+            elif rule.get('dsthost'):
+                if not f'{con_dst}{fullmask}' in dns[rule['dsthost']]:
+                    continue
+
+            if rule.get('specialtarget') == 'dns':
+                if int(con_dport) != 53 or not con_dst in dns_servers:
+                    continue
+
+            return rule['action'] == 'drop'
+
+        # Blocked by default
+        return True
+
     @staticmethod
     def is_ip6(addr):
         return addr.count(':') > 0
@@ -474,6 +555,12 @@ def apply_rules_family(self, source, rules, family):
             raise RuleApplyError('\'iptables -F {}\' failed: {}'.format(
                 chain, e.output))
 
+        connections = self.conntrack_get_connections(family, source)
+        for con in connections:
+            is_blocked = self.is_blocked(rules, con, dns)
+            if is_blocked:
+                self.conntrack_drop(source, con)
+
     def apply_rules(self, source, rules):
         if self.is_ip6(source):
             self.apply_rules_family(source, rules, 6)
@@ -766,6 +853,12 @@ def apply_rules_family(self, source, rules, family):
         self.run_nft(nft)
         self.update_dns_info(source, dns)
 
+        connections = self.conntrack_get_connections(family, source)
+        for con in connections:
+            is_blocked = self.is_blocked(rules, con, dns)
+            if is_blocked:
+                self.conntrack_drop(source, con)
+
     def apply_rules(self, source, rules):
         if self.is_ip6(source):
             self.apply_rules_family(source, rules, 6)

qubesagent/test_firewall.py

@@ -1,8 +1,9 @@
 import logging
 import operator
 import re
+from io import BytesIO
 from unittest import TestCase
-from unittest.mock import patch
+from unittest.mock import patch, Mock
 
 import qubesagent.firewall
 
@@ -94,6 +95,13 @@ def run_user_script(self):
     def update_connected_ips(self, family):
         self.update_connected_ips_called_with.append(family)
 
+    @staticmethod
+    def dns_addresses(family=None):
+        if family == 4:
+            return ['1.1.1.1', '2.2.2.2']
+        else:
+            return ['2001::1', '2001::2']
+
 
 class IptablesWorker(qubesagent.firewall.IptablesWorker):
     '''Override methods actually modifying system state to only log what
@@ -180,6 +188,7 @@ def assertPrepareRulesDnsRet(self, dns_ret, expected_domain, family):
             raise ValueError()
 
     def test_701_dns_info(self):
+        self.obj.conntrack_get_connections = Mock(return_value=[])
         rules = [
             {'action': 'accept', 'proto': 'tcp',
                 'dstports': '80-80', 'dsthost': 'ripe.net'},
@@ -291,6 +300,7 @@ def test_003_prepare_rules6(self):
         self.assertPrepareRulesDnsRet(ret[1], 'ripe.net', 6)
 
     def test_004_apply_rules4(self):
+        self.obj.conntrack_get_connections = Mock(return_value=[])
         rules = [{'action': 'accept'}]
         chain = 'qbs-10-137-0-1'
         self.obj.apply_rules('10.137.0.1', rules)
@@ -305,6 +315,7 @@ def test_004_apply_rules4(self):
         self.assertIsNone(self.obj.loaded_iptables[6])
 
     def test_005_apply_rules6(self):
+        self.obj.conntrack_get_connections = Mock(return_value=[])
         rules = [{'action': 'accept'}]
         chain = 'qbs-2000--a'
         self.obj.apply_rules('2000::a', rules)
@@ -519,6 +530,7 @@ def test_003_prepare_rules6(self):
         self.assertPrepareRulesDnsRet(ret[1], 'ripe.net', 6)
 
     def test_004_apply_rules4(self):
+        self.obj.conntrack_get_connections = Mock(return_value=[])
         rules = [{'action': 'accept'}]
         chain = 'qbs-10-137-0-1'
         self.obj.apply_rules('10.137.0.1', rules)
@@ -528,6 +540,7 @@ def test_004_apply_rules4(self):
              ])
 
     def test_005_apply_rules6(self):
+        self.obj.conntrack_get_connections = Mock(return_value=[])
         rules = [{'action': 'accept'}]
         chain = 'qbs-2000--a'
         self.obj.apply_rules('2000::a', rules)
@@ -726,3 +739,60 @@ def test_main(self):
         self.assertEqual(self.obj.update_connected_ips_called_with, [4, 6])
         self.assertEqual(set(self.obj.rules.keys()), self.obj.list_targets())
         # rules content were already tested
+
+    @patch('subprocess.Popen')
+    def test_conntrack_get_connections(self, mock_subprocess):
+        conntrack_stdout = (
+            b'tcp      6 431963 ESTABLISHED src=10.138.38.13 dst=1.1.1.1 '
+                b'sport=34488 dport=443 src=1.1.1.1 dst=10.138.38.13 sport=443 '
+                b'dport=34488 [ASSURED] mark=0 use=1\n'
+
+            b'udp      17 3 src=10.138.38.13 dst=10.139.1.1 sport=33295 dport=53 '
+                b'src=10.139.1.1 dst=10.138.38.13 sport=53 dport=33295 mark=0 use=1\n'
+        )
+        mock_subprocess().__enter__().stdout = BytesIO(conntrack_stdout)
+        ret = self.obj.conntrack_get_connections(4, "10.138.38.13")
+        self.assertEqual(ret, {
+            ('udp', '10.139.1.1', '53'),
+            ('tcp', '1.1.1.1', '443')
+        })
+
+    def test_is_blocked(self):
+        dns_servers_ipv4 = list(self.obj.dns_addresses(4))
+        dns_servers_ipv6 = list(self.obj.dns_addresses(6))
+        dns = {
+            "example.com": set(["1.2.3.4/32", "4.3.2.1/32"]),
+            "example2.com": set(["2001::1/128", "2001::2/128"])
+        }
+        rules = [
+            {'proto': 'tcp', 'dstports': '80-80', 'action': 'drop'},
+            {'proto': 'tcp', 'dstports': '90-92', 'action': 'drop'},
+            {'proto': 'tcp', 'dsthost': 'example.com', 'action': 'drop'},
+            {'proto': 'tcp', 'dsthost': 'example2.com', 'action': 'drop'},
+            {'dst4': '3.3.3.3/32', 'action': 'drop'},
+            {'dst4': '4.4.4.4/24', 'action': 'drop'},
+            {'dst6': '2002::3/128', 'action': 'drop'},
+            {'dst6': '2003::3/112', 'action': 'drop'},
+            {'proto': 'udp', 'specialtarget': 'dns', 'action': 'drop'},
+            {'action': 'accept'},
+        ]
+
+        self.assertTrue(self.obj.is_blocked({}, ("tcp", "1.1.1.1", "123"), dns))
+
+        self.assertFalse(self.obj.is_blocked(rules, ("tcp", "10.0.0.1", "443"), dns))
+        self.assertFalse(self.obj.is_blocked(rules, ("udp", "10.0.0.1", "80"), dns))
+        self.assertTrue(self.obj.is_blocked(rules, ("tcp", "10.0.0.1", "80"), dns))
+        self.assertTrue(self.obj.is_blocked(rules, ("tcp", "10.0.0.1", "91"), dns))
+        self.assertTrue(self.obj.is_blocked(rules, ("tcp", "1.2.3.4", "123"), dns))
+        self.assertTrue(self.obj.is_blocked(rules, ("tcp", "4.3.2.1", "123"), dns))
+        self.assertTrue(self.obj.is_blocked(rules, ("tcp", "2003::2", "123"), dns))
+        self.assertTrue(self.obj.is_blocked(rules, ("tcp", "3.3.3.3", "123"), dns))
+        self.assertTrue(self.obj.is_blocked(rules, ("tcp", "4.4.4.8", "123"), dns))
+        self.assertTrue(self.obj.is_blocked(rules, ("tcp", "2002::3", "123"), dns))
+        self.assertTrue(self.obj.is_blocked(rules, ("tcp", "2003::5", "123"), dns))
+
+        for server in dns_servers_ipv4:
+            self.assertTrue(self.obj.is_blocked(rules, ("udp", server, "53"), dns))
+
+        for server in dns_servers_ipv6:
+            self.assertTrue(self.obj.is_blocked(rules, ("udp", server, "53"), dns))