Integration of Access Analyzer's Automated Reasoning with ConsoleMe

consoleme/default_plugins/plugins/aws/aws.py

@@ -31,12 +31,23 @@
 from consoleme.lib.plugins import get_plugin_by_name
 from consoleme.lib.policies import send_communications_policy_change_request_v2
 from consoleme.lib.redis import RedisHandler
+from consoleme.lib.access_analyzer import AccessAnalyzer
+from consoleme.models import (
+    Action,
+    ExtendedRequestModel,
+)
+
+
 
 stats = get_plugin_by_name(config.get("plugins.metrics", "default_metrics"))()
 
 log = config.get_logger(__name__)
 
-
+try:
+    accessanalyzer = AccessAnalyzer()
+except Exception as e:  # noqa
+     accessanalyzer = None
+
 class Aws:
     """The AWS class handles interactions with AWS."""
 
@@ -720,9 +731,128 @@ async def send_communications_new_policy_request(
     def handle_detected_role(role):
         pass
 
-    async def should_auto_approve_policy_v2(self, extended_request, user, user_groups):
-        return {"approved": False}
+    async def should_auto_approve_policy_v2(
+        self, extended_request: ExtendedRequestModel, user: str, user_groups: str
+    ) -> bool: 
+        """Add here the logic to call the check_no_new_access code
+
+        Args:
+            extended_request (_type_): _description_
+            user (_type_): _description_
+            user_groups (_type_): _description_
+
+        Returns:
+            _type_: _description_
+        """
+
+        function = f"{__name__}.{sys._getframe().f_code.co_name}"
+        log_data = {"function": function, "user": user} 
+
+        try:
+            if not config.get("dynamic_config.policy_request_autoapprove_probes.enabled"):
+                return {"approved": False}
+
+            for change in extended_request.changes.changes:
+                account_id = extended_request.principal.principal_arn.split(":")[4]
+                log_data = {
+                    "function": function, 
+                    "requested_policy": extended_request.dict(),
+                    "user": user,
+                    "arn": extended_request.principal.principal_arn,
+                }
+                approving_probe = None
+                # We only support inline policies at this time
+                if change.change_type != "inline_policy":
+                    return {"approved": False}
+
+                # We only want to analyze attach events
+                print(f"Checking action: {change.action}") 
+                if change.action != Action.attach:
+                    return {"approved": False}
+
+                if not accessanalyzer:
+                    return {"approved": False}
+
+                for probe in config.get(
+                    "dynamic_config.policy_request_autoapprove_probes.probes"
+                ):
+                    log_data["probe"] = probe["name"]
+                    probe_name = probe["name"]
+                    log_data["requested_policy"] = change
+                    log_data["message"] = "Running probe on requested policy"
+                    probe_result = False
+                    requested_policy_text = change.policy.policy_document
+
+                    # Do not approve "Deny" policies automatically
+                    if isinstance(requested_policy_text, dict):
+                        statements = requested_policy_text.get("Statement", [])
+                        for statement in statements:
+                            if not isinstance(statement, dict):
+                                continue
+                            if statement.get("Effect") == "Deny":
+                                return {"approved": False}
+
+                    if isinstance(requested_policy_text, dict):
+                        requested_policy_text = json.dumps(requested_policy_text)
+                    accessanalyzer_result = await sync_to_async(accessanalyzer.check_no_new_access)(
+                        new_policy_document=requested_policy_text,
+                        existing_policy_document=probe["policy"].replace(
+                            "{account_id}", account_id
+                        ),
+                        policy_type="IDENTITY_POLICY"
+                    )
 
+                    comparison = accessanalyzer_result
+
+                    allow_listed = False
+                    allowed_group = False
+
+                     # Probe will fail if ARN account ID is not in the probe's account allow-list. Default allow-list is
+                    # *
+                    for account in probe.get("accounts", {}).get("allowlist", ["*"]):
+                        if account == "*" or account_id == str(account):
+                            allow_listed = True
+                            break
+
+                    if not allow_listed:
+                        comparison = "DENIED_BY_ALLOWLIST"
+
+                    # Probe will fail if ARN account ID is in the probe's account blocklist
+                    for account in probe.get("accounts", {}).get("blocklist", []):
+                        if account_id == str(account):
+                            comparison = "DENIED_BY_BLOCKLIST"
+
+                    for group in probe.get("required_user_or_group", ["*"]):
+                        for g in user_groups:
+                            if group == "*" or group == g or group == user:
+                                allowed_group = True
+                                break
+
+                    if comparison == "PASS":
+                        probe_result = True
+                        policy_result = True
+                        approving_probe = probe
+                    log_data["comparison"] = comparison
+                    log_data["probe_result"] = probe_result
+                    log.debug(log_data)
+
+                if not policy_result:
+                    # If one of the policies in the request fails to auto-approve, everything fails
+                    log_data["result"] = False
+                    log_data["message"] = "Successfully ran all probes"
+                    log.debug(log_data)
+                    stats.count(f"{function}.called", tags={"result": False})
+                    return {"approved": False}
+
+            log_data["result"] = True
+            log_data["message"] = "Successfully ran all probes"
+            log.debug(log_data)
+            stats.count(f"{function}.called", tags={"result": True})
+            print(f"Checking probe entire: {approving_probe}") 
+            return {"approved": True, "approving_probes": [approving_probe]}
+        except Exception as e:
+            print(f"An unexpected error occurred: {e}")
+            return {"approved": False}
 
 def init():
     """Initialize the AWS plugin."""

consoleme/handlers/v2/requests.py

@@ -330,10 +330,12 @@ async def post(self):
                                 f"{log_data['function']}.probe_auto_approved",
                                 tags={"user": self.user},
                             )
+
                             approving_probes = []
                             for approving_probe in should_auto_approve_request[
                                 "approving_probes"
                             ]:
+
                                 approving_probe_comment = CommentModel(
                                     id=str(uuid.uuid4()),
                                     timestamp=int(time.time()),

consoleme/lib/access_analyzer.py

@@ -0,0 +1,37 @@
+from cloudaux.aws.sts import boto3_cached_conn
+import botocore.exceptions
+
+class AccessAnalyzer:
+    """
+    The Access Analyzer class that hosts the methods related to the Access Analyzer API.
+    """
+
+    def check_no_new_access(self, new_policy_document, existing_policy_document, policy_type):
+        """
+        The check_no_new_access method compares the new policy document with the 
+        existing policy document and returns the result.
+
+        Args:
+            existing_policy_document (string): The existing base policy as source of the truth.
+            new_policy_document (string): The requested policy document to be compared.
+            policy_type (string): The policy type to be compared. Only "IDENTITY_POLICY" is supported at the moment.
+
+        Raises:
+            ce: Raises the botocore exception if any error occurs.
+
+        Returns:
+            string: Returns the result of the comparison. Either "FAIL" or "PASS".
+        """
+        try:
+            client = boto3_cached_conn("accessanalyzer")
+
+            if policy_type != "IDENTITY_POLICY":
+                raise ValueError("Only Identity Policy is supported at the moment.")
+            response = client.check_no_new_access(
+                existingPolicyDocument=existing_policy_document,
+                newPolicyDocument=new_policy_document,
+                policyType=policy_type
+            )
+            return response["result"]
+        except botocore.exceptions.ClientError as ce:
+            raise ce

example_config/example_config_development.yaml

@@ -32,3 +32,66 @@ export_to_terraform_enabled: true
 # because Weep tends to make an IPv6 request, and the browser makes an IPv4 request.
 challenge_url:
   request_ip_must_match_challenge_creation_ip: false
+
+dynamic_config:
+  policy_request_autoapprove_probes: 
+    enabled: true
+    probes:
+      - name: same_account_sqs
+        required_user_or_group:
+          - "*"
+        description: Automatically approve same-account SQS requests
+        accounts:
+          allowlist:
+            - "*"
+          blocklist:
+            - 1234567
+        policy: |-
+          { 
+            "Version": "2012-10-17",
+            "Statement": [
+              {
+                "Action": [
+                  "sqs:List*",
+                  "sqs:Get*",
+                  "sqs:Send*",
+                  "sqs:ChangeMessageVisibility*",
+                  "sqs:DeleteMessage*",
+                  "sqs:ReceiveMessage*",
+                  "sqs:StartMessageMoveTask",
+                  "sqs:ListMessageMoveTasks",
+                  "sqs:CancelMessageMoveTask"
+                ],
+                "Effect": "Allow",
+                "Resource": [
+                  "arn:aws:sqs:*:{account_id}:*"
+                ]
+              }
+            ]
+          }
+      - name: s3_probe
+        required_user_or_group:
+          - '[email protected]'
+        description: Automatically approve s3 read/write requests made by the S3 team
+        accounts:
+          allowlist:
+            - "*"
+          blocklist:
+            - 1234567
+        policy: |-
+          { 
+            "Version": "2012-10-17",
+            "Statement": [
+              {
+                "Action": [
+                  "s3:Get*",
+                  "s3:List*",
+                  "s3:PutObject*"
+                ],
+                "Effect": "Allow",
+                "Resource": [
+                  "arn:aws:s3:::*"
+                ]
+              }
+            ]
+          }

requirements-test.txt

@@ -88,14 +88,14 @@ boto==2.49.0
     #   -c requirements.txt
     #   -r requirements.txt
     #   cloudaux
-boto3==1.24.30
+boto3==1.34.67
     # via
     #   -c requirements.txt
     #   -r requirements.txt
     #   cloudaux
     #   moto
     #   parliament
-botocore==1.27.30
+botocore==1.34.67
     # via
     #   -c requirements.txt
     #   -r requirements.txt
@@ -786,7 +786,7 @@ ruamel-yaml-clib==0.2.6
     #   -c requirements.txt
     #   -r requirements.txt
     #   ruamel-yaml
-s3transfer==0.6.0
+s3transfer==0.10.1
     # via
     #   -c requirements.txt
     #   -r requirements.txt

requirements.txt

@@ -40,12 +40,12 @@ bleach==5.0.1
     # via -r requirements.in
 boto==2.49.0
     # via cloudaux
-boto3==1.24.30
+boto3==1.34.67
     # via
     #   -r requirements.in
     #   cloudaux
     #   parliament
-botocore==1.27.30
+botocore==1.34.67
     # via
     #   boto3
     #   cloudaux
@@ -378,7 +378,7 @@ ruamel-yaml==0.17.21
     #   prance
 ruamel-yaml-clib==0.2.6
     # via ruamel-yaml
-s3transfer==0.6.0
+s3transfer==0.10.1
     # via boto3
 schema==0.7.5
     # via policy-sentry