Initial commit of v1.0.0. Updated the .gitignore for the standard Python gitignore. Added Azure DevOps release pipeline for proper artifact release from the start.

PRIMAITE/acl/__init__.py

@@ -0,0 +1 @@
+# Crown Copyright (C) Dstl 2022. DEFCON 703. Shared in confidence.

PRIMAITE/acl/access_control_list.py

@@ -0,0 +1,134 @@
+# Crown Copyright (C) Dstl 2022. DEFCON 703. Shared in confidence.
+"""
+A class that implements the access control list implementation for the network
+"""
+
+from acl.acl_rule import ACLRule
+
+class AccessControlList():
+    """
+    Access Control List class
+    """
+
+    def __init__(self):
+        """
+        Init
+        """
+
+        self.acl = {}       # A dictionary of ACL Rules
+    
+    def check_address_match(self, _rule, _source_ip_address, _dest_ip_address):
+        """
+        Checks for IP address matches
+
+        Args:
+            _rule: The rule being checked
+            _source_ip_address: the source IP address to compare
+            _dest_ip_address: the destination IP address to compare
+
+        Returns:
+             True if match; False otherwise.
+        """
+
+        if ((_rule.get_source_ip() == _source_ip_address and _rule.get_dest_ip() == _dest_ip_address) or
+            (_rule.get_source_ip() == "ANY" and _rule.get_dest_ip() == _dest_ip_address) or
+            (_rule.get_source_ip() == _source_ip_address and _rule.get_dest_ip() == "ANY") or
+            (_rule.get_source_ip() == "ANY" and _rule.get_dest_ip() == "ANY")):
+            return True
+        else:
+            return False
+
+    def is_blocked(self, _source_ip_address, _dest_ip_address, _protocol, _port):
+        """
+        Checks for rules that block a protocol / port
+
+        Args:
+            _source_ip_address: the source IP address to check
+            _dest_ip_address: the destination IP address to check
+            _protocol: the protocol to check
+            _port: the port to check
+
+        Returns:
+             Indicates block if all conditions are satisfied.
+        """
+        
+        for rule_key, rule_value in self.acl.items():
+            if self.check_address_match(rule_value, _source_ip_address, _dest_ip_address):
+                if ((rule_value.get_protocol() == _protocol or rule_value.get_protocol() == "ANY") and
+                    (str(rule_value.get_port()) == str(_port) or rule_value.get_port() == "ANY")):
+                    # There's a matching rule. Get the permission
+                    if rule_value.get_permission() == "DENY":
+                        return True
+                    elif rule_value.get_permission() == "ALLOW":
+                        return False
+
+        # If there has been no rule to allow the IER through, it will return a blocked signal by default
+        return True
+
+    def add_rule(self, _permission, _source_ip, _dest_ip, _protocol, _port):
+        """
+        Adds a new rule
+
+        Args:
+            _permission: the permission value (e.g. "ALLOW" or "DENY")
+            _source_ip: the source IP address
+            _dest_ip: the destination IP address
+            _protocol: the protocol
+            _port: the port
+        """
+
+        new_rule = ACLRule(_permission, _source_ip, _dest_ip, _protocol, str(_port))
+        hash_value = hash(new_rule)
+        self.acl[hash_value] = new_rule
+
+    def remove_rule(self, _permission, _source_ip, _dest_ip, _protocol, _port):
+        """
+        Removes a rule
+
+        Args:
+            _permission: the permission value (e.g. "ALLOW" or "DENY")
+            _source_ip: the source IP address
+            _dest_ip: the destination IP address
+            _protocol: the protocol
+            _port: the port
+        """
+
+        rule = ACLRule(_permission, _source_ip, _dest_ip, _protocol, str(_port))
+        hash_value = hash(rule)
+        # There will not always be something 'popable' since the agent will be trying random things
+        try:
+            self.acl.pop(hash_value)
+        except:
+            return
+
+    def remove_all_rules(self):
+        """
+        Removes all rules
+        """
+
+        self.acl.clear()
+
+    def get_dictionary_hash(self, _permission, _source_ip, _dest_ip, _protocol, _port):
+        """
+        Produces a hash value for a rule
+
+        Args:
+            _permission: the permission value (e.g. "ALLOW" or "DENY")
+            _source_ip: the source IP address
+            _dest_ip: the destination IP address
+            _protocol: the protocol
+            _port: the port
+
+        Returns:
+             Hash value based on rule parameters.
+        """
+
+        rule = ACLRule(_permission, _source_ip, _dest_ip, _protocol, str(_port))
+        hash_value = hash(rule)
+        return hash_value
+
+
+
+
+
+

PRIMAITE/acl/acl_rule.py

@@ -0,0 +1,88 @@
+# Crown Copyright (C) Dstl 2022. DEFCON 703. Shared in confidence.
+"""
+A class that implements an access control list rule
+"""
+
+class ACLRule():
+    """
+    Access Control List Rule class
+    """
+
+    def __init__(self, _permission, _source_ip, _dest_ip, _protocol, _port):
+        """
+        Init
+
+        Args:
+            _permission: The permission (ALLOW or DENY)
+            _source_ip: The source IP address
+            _dest_ip: The destination IP address
+            _protocol: The rule protocol
+            _port: The rule port
+        """
+
+        self.permission = _permission
+        self.source_ip = _source_ip
+        self.dest_ip = _dest_ip
+        self.protocol = _protocol
+        self.port = _port
+
+    def __hash__(self):
+        """
+        Override the hash function
+
+        Returns:
+             Returns hash of core parameters.
+        """
+
+        return hash((self.permission, self.source_ip, self.dest_ip, self.protocol, self.port))
+
+    def get_permission(self):
+        """
+        Gets the permission attribute
+
+        Returns:
+             Returns permission attribute
+        """
+
+        return self.permission
+
+    def get_source_ip(self):
+        """
+        Gets the source IP address attribute
+
+        Returns:
+             Returns source IP address attribute
+        """
+
+        return self.source_ip
+
+    def get_dest_ip(self):
+        """
+        Gets the desintation IP address attribute
+
+        Returns:
+             Returns destination IP address attribute
+        """
+
+        return self.dest_ip
+
+    def get_protocol(self):
+        """
+        Gets the protocol attribute
+
+        Returns:
+             Returns protocol attribute
+        """
+        return self.protocol
+
+    def get_port(self):
+        """
+        Gets the port attribute
+
+        Returns:
+             Returns port attribute
+        """
+
+        return self.port
+        
+