add: add vulnerable_ports logic

Author: jinyoungmoonDEV

src/plugin/manager/ec2/security_group_manager.py

@@ -37,7 +37,7 @@ def create_cloud_service(self, region, options, secret_data, schema):
         cloudtrail_resource_type = "AWS::EC2::SecurityGroup"
 
         # If Port Filter Option Exist
-        vulnerable_ports = self.options.get("vulnerable_ports", DEFAULT_VULNERABLE_PORTS)
+        vulnerable_ports = options.get("vulnerable_ports", DEFAULT_VULNERABLE_PORTS)
 
         # Get default VPC
         default_vpcs = self._get_default_vpc()
@@ -166,16 +166,15 @@ def create_cloud_service(self, region, options, secret_data, schema):
                     )
 
     def custom_security_group_rule_info(self, raw_rule, remote, remote_type, vulnerable_ports):
+        protocol_display = self._get_protocol_display(raw_rule.get("IpProtocol"))
         raw_rule.update(
             {
-                "protocol_display": self._get_protocol_display(
-                    raw_rule.get("IpProtocol")
-                ),
+                "protocol_display": protocol_display,
                 "port_display": self._get_port_display(raw_rule),
                 "source_display": self._get_source_display(remote),
                 "description_display": self._get_description_display(remote),
                 remote_type: remote,
-                "vulnerable_ports": self._get_vulnerable_ports(raw_rule, vulnerable_ports)
+                "vulnerable_ports": self._get_vulnerable_ports(protocol_display, raw_rule, vulnerable_ports)
             }
         )
 
@@ -295,23 +294,21 @@ def get_instance_name_from_tags(instance):
         return ""
 
     @staticmethod
-    def _get_vulnerable_ports(raw_rule, vulnerable_ports):
-        is_port_all = False
-
+    def _get_vulnerable_ports(protocol_display: str, raw_rule: dict, vulnerable_ports: str):
         try:
-            toPort = int(raw_rule.get("ToPort"))
-            fromPort = int(raw_rule.get("FromPort"))
-        except (ValueError, TypeError):
-            is_port_all = True
-            toPort, fromPort = None, None
+            if protocol_display == "ALL":
+                return [int(port.strip()) for port in vulnerable_ports.split(',')]
 
-        ports = []
-        try:
-            for port in map(str.strip, vulnerable_ports.split(',')):
-                target_port = int(port)
-                if is_port_all or (fromPort <= target_port <= toPort):
-                    ports.append(target_port)
+            to_port = raw_rule.get("ToPort")
+            from_port = raw_rule.get("FromPort")
+
+            if to_port is None or from_port is None:
+                return []
 
-            return ports
-        except Exception:
+            return [
+                int(port.strip())
+                for port in vulnerable_ports.split(',')
+                if from_port <= int(port.strip()) <= to_port
+            ]
+        except ValueError:
             raise ERROR_VULNERABLE_PORTS(vulnerable_ports)