Compare IP fact rules with address facts by address instead of regex prefix

app/utility/rule_set.py

@@ -15,8 +15,10 @@ def __init__(self, rules):
     async def is_fact_allowed(self, fact):
         allowed = True
         for rule in await self._applicable_rules(fact):
-            if await self._is_ip_rule_match(rule, fact):
-                allowed = await self._rule_judgement(rule.action)
+            if await self._is_ip_rule(rule, fact):
+                # An IP rule must not fall back to regex matching: as a regex, '10.0.0.1' also matches '10.0.0.10'.
+                if await self._is_ip_rule_match(rule, fact):
+                    allowed = await self._rule_judgement(rule.action)
                 continue
             if await self._is_regex_rule_match(rule, fact):
                 allowed = await self._rule_judgement(rule.action)
@@ -70,6 +72,19 @@ async def _is_ip_address(value):
     async def _is_regex_rule_match(rule, fact):
         return re.match(rule.match, fact.value)
 
+    async def _is_ip_rule(self, rule, fact):
+        """True when the rule match is an IPv4 address or network and the fact value is an IPv4 address.
+
+        Network-valued facts keep the regex fallback: _is_ip_rule_match never matches a rule address against a
+        fact network, and deciding on that alone would let a fact covering a denied address pass a DENY rule.
+        """
+        if rule.match == '.*':
+            return False
+        is_rule_ip = await self._is_ip_address(rule.match) or await self._is_ip_network(rule.match)
+        if not is_rule_ip:
+            return False
+        return await self._is_ip_address(fact.value)
+
     async def _is_ip_rule_match(self, rule, fact):
         """ We only match string-equivalent IP addresses, string-equivalent subnets in CIDR notation, and IP address
         facts to the subnet rules where the address is a member of the subnet. Matching non-equivalent subnets is

tests/utility/test_ip_rule.py

@@ -58,3 +58,28 @@ async def test_larger_subnet(self):
     async def test_same_subnet(self):
         assert await self.rs._is_ip_rule_match(self.rule, self.fact6)
         assert (not await self.rs.is_fact_allowed(self.fact6))
+
+    async def test_address_rule_does_not_match_longer_address(self):
+        rs = RuleSet(rules=[Rule(trait='host.ip.address', action=RuleAction.DENY, match='127.0.0.1')])
+        assert not await rs.is_fact_allowed(Fact(trait='host.ip.address', value='127.0.0.1'))
+        assert await rs.is_fact_allowed(Fact(trait='host.ip.address', value='127.0.0.10'))
+        assert await rs.is_fact_allowed(Fact(trait='host.ip.address', value='127.0.0.123'))
+
+    async def test_allow_single_address_after_deny_all(self):
+        rs = RuleSet(rules=[Rule(trait='host.ip.address', action=RuleAction.DENY, match='.*'),
+                            Rule(trait='host.ip.address', action=RuleAction.ALLOW, match='10.0.0.1')])
+        assert await rs.is_fact_allowed(Fact(trait='host.ip.address', value='10.0.0.1'))
+        assert not await rs.is_fact_allowed(Fact(trait='host.ip.address', value='10.0.0.10'))
+        assert not await rs.is_fact_allowed(Fact(trait='host.ip.address', value='10.0.0.100'))
+
+    async def test_address_rule_still_denies_network_facts(self):
+        rs = RuleSet(rules=[Rule(trait='host.ip.address', action=RuleAction.DENY, match='10.0.0.1')])
+        assert not await rs.is_fact_allowed(Fact(trait='host.ip.address', value='10.0.0.1/32'))
+        rs = RuleSet(rules=[Rule(trait='host.ip.address', action=RuleAction.DENY, match='10.0.0.0')])
+        assert not await rs.is_fact_allowed(Fact(trait='host.ip.address', value='10.0.0.0/24'))
+
+    async def test_regex_rule_still_applies_to_ip_facts(self):
+        rs = RuleSet(rules=[Rule(trait='host.ip.address', action=RuleAction.DENY, match='.*'),
+                            Rule(trait='host.ip.address', action=RuleAction.ALLOW, match=r'10\.0\.0\.\d+$')])
+        assert await rs.is_fact_allowed(Fact(trait='host.ip.address', value='10.0.0.10'))
+        assert not await rs.is_fact_allowed(Fact(trait='host.ip.address', value='10.0.1.10'))