Standardizing Impact Checks (#54)

* impact-improvements

* fix-exposure-impact

README.md

@@ -133,7 +133,6 @@ The following are the impact criteria that MetaHub evaluates by default:
 | ----------------------- | --------------- |
 | 🔴 effectively-public   |                 |
 | 🟠 restricted-public    |                 |
-| 🟠 unknown-public       |                 |
 | 🟠 unrestricted-private |                 |
 | 🟢 restricted           |                 |
 | 🔵 unknown              |                 |

lib/config/configuration.py

@@ -1,9 +1,28 @@
-# Security Hub Configurations
+# MetaHub Configurations
 
-# Default filters for Security Hub findings, not implemented yet
-# sh_default_filters = {"RecordState": ["ACTIVE"], "WorkflowStatus": ["NEW"]}
+# ---------------------------------- #
+# Security Hub Configurations        #
+# ---------------------------------- #
 
-# Impact Checks Configurations
+# Default filters for Security Hub
+sh_default_filters = {"RecordState": ["ACTIVE"], "WorkflowStatus": ["NEW"]}
+
+
+# ---------------------------------- #
+# Impact Configurations              #
+# ---------------------------------- #
+
+# Impact Scoring Defintion File
+path_yaml_impact = "lib/config/impact.yaml"
+
+# Severity Values for impact scoring calculation
+findings_severity_value = {
+    "CRITICAL": 4,
+    "HIGH": 3,
+    "MEDIUM": 1,
+    "LOW": 0.5,
+    "INFORMATIONAL": 0,
+}
 
 # List of AWS accounts ids that are trusted and not considered as external.
 # This is used in check untrusted_principal for policies.
@@ -48,16 +67,10 @@
     "environment": ["dev"],
 }
 
-# Severity Values for Impact Findings Scores Calculation
-findings_severity_value = {
-    "CRITICAL": 4,
-    "HIGH": 3,
-    "MEDIUM": 1,
-    "LOW": 0.5,
-    "INFORMATIONAL": 0,
-}
 
-# Output Configurations
+# ---------------------------------- #
+# Output Configurations              #
+# ---------------------------------- #
 
 # Columns
 # You can define the columns that will be displayed in the output HTML, CSV AND XLSX.
@@ -70,6 +83,3 @@
 
 # Decide if you want to output as part of the findings the whole json resource policy
 output_resource_policy = True
-
-
-path_yaml_impact = "lib/config/impact.yaml"

lib/context/resources/AwsEc2SecurityGroup.py

@@ -5,7 +5,6 @@
 
 from lib.AwsHelpers import get_boto3_client
 from lib.context.resources.Base import ContextBase
-from lib.context.resources.ContextHelpers import SGHelper
 
 
 class Metacheck(ContextBase):
@@ -31,9 +30,6 @@ def __init__(
         self.network_interfaces = self._describe_network_interfaces_interfaces()
         self.instances = self._describe_network_interfaces_instances()
         self.security_group_rules = self.describe_security_group_rules()
-        self.checked_security_group_rules = SGHelper(
-            self.logger, self.security_group_rules
-        ).check_security_group_rules()
         # Associated MetaChecks
         self.vpcs = self._describe_security_group_vpc()
 
@@ -120,15 +116,16 @@ def _describe_network_interfaces_instances(self):
         return instances
 
     def describe_security_group_rules(self):
+        security_group_rules = []
         if self.security_group:
             response = self.client.describe_security_group_rules(
                 Filters=[
                     {"Name": "group-id", "Values": [self.resource_id]},
                 ],
             )
             if response["SecurityGroupRules"]:
-                return response["SecurityGroupRules"]
-        return False
+                security_group_rules = response["SecurityGroupRules"]
+        return security_group_rules
 
     def _describe_security_group_vpc(self):
         vcps = {}
@@ -181,22 +178,6 @@ def its_referenced_by_a_security_group(self):
             return references
         return False
 
-    def is_ingress_rules_unrestricted(self):
-        is_ingress_rules_unrestricted = self.checked_security_group_rules[
-            "is_ingress_rules_unrestricted"
-        ]
-        if is_ingress_rules_unrestricted:
-            return is_ingress_rules_unrestricted
-        return False
-
-    def is_egress_rules_unrestricted(self):
-        is_egress_rules_unrestricted = self.checked_security_group_rules[
-            "is_egress_rules_unrestricted"
-        ]
-        if is_egress_rules_unrestricted:
-            return is_egress_rules_unrestricted
-        return False
-
     def default(self):
         if self.security_group:
             if self.security_group["GroupName"] == "default":
@@ -233,8 +214,7 @@ def checks(self):
             "public_ips": self.public_ips(),
             "managed_services": self.managed_services(),
             "its_referenced_by_a_security_group": self.its_referenced_by_a_security_group(),
-            "is_ingress_rules_unrestricted": self.is_ingress_rules_unrestricted(),
-            "is_egress_rules_unrestricted": self.is_egress_rules_unrestricted(),
+            "security_group_rules": self.security_group_rules,
             "public": self.public(),
             "default": self.default(),
             "attached": self.attached(),

lib/context/resources/AwsRdsDbCluster.py

@@ -87,8 +87,8 @@ def _describe_db_clusters_iam_roles(self):
 
     def endpoint(self):
         if self.rds_cluster:
-            if self.rds_cluster.get("Endpoint"):
-                return self.rds_cluster.get("Endpoint")
+            if self.rds_cluster.get("PubliclyAccessible"):
+                return self.rds_cluster.get("PubliclyAccessible")
         return False
 
     def storage_encrypted(self):

lib/context/resources/AwsRdsDbInstance.py

@@ -92,8 +92,9 @@ def endpoint(self):
         return False
 
     def public(self):
-        if self.endpoint():
-            return True
+        if self.rds_instances:
+            if self.rds_instances.get("PubliclyAccessible"):
+                return self.rds_instances.get("PubliclyAccessible")
         return False
 
     def storage_encrypted(self):

lib/context/resources/AwsSecretsManagerSecret.py

@@ -62,8 +62,8 @@ def describe_secret(self):
     def get_resource_policy(self):
         if self.secret:
             response = self.client.get_resource_policy(SecretId=self.resource_arn)
-        if response.get("ResourcePolicy"):
-            return json.loads(response["ResourcePolicy"])
+            if response.get("ResourcePolicy"):
+                return json.loads(response["ResourcePolicy"])
         return False
 
     # Context Config
@@ -84,6 +84,14 @@ def rotation_enabled(self):
                 return False
         return False
 
+    def kms_key_id(self):
+        if self.secret:
+            try:
+                return self.secret["KmsKeyId"]
+            except KeyError:
+                return False
+        return False
+
     def is_unrotated(self):
         if self.secret:
             current_date = datetime.now(timezone.utc)
@@ -107,6 +115,7 @@ def checks(self):
         checks = {
             "resource_policy": self.resource_policy,
             "name": self.name(),
+            "kms_key_id": self.kms_key_id(),
             "rotation_enabled": self.rotation_enabled(),
             "is_unrotated": self.is_unrotated(),
             "public": self.public(),