Class RuleSet

app/utility/rule_set.py:11–101

class RuleSet:
    def __init__(self, rules):
        self.rules = 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)
                continue
            if await self._is_regex_rule_match(rule, fact):
                allowed = await self._rule_judgement(rule.action)
        return allowed

    async def _applicable_rules(self, fact):
        applicable_rules = []
        for rule in self.rules:
            if rule.trait == fact.trait:
                applicable_rules.append(rule)
        return applicable_rules

    async def apply_rules(self, facts):
        if await self._has_rules():
            valid_facts = []
            for fact in facts:
                if await self.is_fact_allowed(fact):
                    valid_facts.append(fact)
            return [valid_facts]
        else:
            return [facts]

    async def _has_rules(self):
        return len(self.rules)

    @staticmethod
    async def _rule_judgement(action):
        if action.value == RuleAction.DENY.value:
            return False
        return True

    @staticmethod
    async def _is_ip_network(value):
        try:
            ipaddress.IPv4Network(value)
            return True
        except (ValueError, ipaddress.AddressValueError):
            pass
        return False

    @staticmethod
    async def _is_ip_address(value):
        try:
            ipaddress.IPv4Address(value)
            return True
        except (ValueError, ipaddress.AddressValueError):
            pass
        return False