get_actions_matching_condition_crud_and_arn is now available. Part of…

… the scaffoldign needed for #21
kmcquade · Jan 16, 2020 · 187131f · 187131f
1 parent d234757
commit 187131f
Show file tree

Hide file tree

Showing 5 changed files with 128 additions and 4 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,7 +1,14 @@
 # Changelog
-## 2020-01-09
+## 2020-01-15
+### Added
+* `get_actions_matching_condition_key_crud_and_arn` is available. This provides some scaffolding for #21
 
+## 2020-01-10
+### Added
+* `get_all_actions_with_access_level` now supports "all" as a valid input, so you can easily request all IAM actions that are at a certain access level, regardless of service.
+* 0.6.8 Version bump so I can use it
 
+## 2020-01-09
 ### Changed
 * 0.6.7 release to avoid issues with breaking changes
 * Template:

diff --git a/examples/library-usage/querying/actions/get_actions_matching_condition_crud_and_arn.py b/examples/library-usage/querying/actions/get_actions_matching_condition_crud_and_arn.py
@@ -0,0 +1,22 @@
+#!/usr/bin/env python
+from policy_sentry.shared.database import connect_db
+from policy_sentry.querying.actions import get_actions_matching_condition_crud_and_arn
+import json
+
+if __name__ == '__main__':
+    db_session = connect_db('bundled')
+    results = get_actions_matching_condition_crud_and_arn(
+        db_session,
+        "ram:ResourceArn",
+        "Permissions management",
+        "arn:${Partition}:ram:${Region}:${Account}:resource-share/${ResourcePath}"
+    )
+    print(json.dumps(output, indent=4))
+
+"""
+Output:
+
+[
+    'ram:createresourceshare'
+]
+"""
diff --git a/policy_sentry/querying/actions.py b/policy_sentry/querying/actions.py
@@ -4,7 +4,9 @@
 """
 from sqlalchemy import and_
 from policy_sentry.shared.database import ActionTable
+from policy_sentry.querying.all import get_all_service_prefixes
 from policy_sentry.util.actions import get_full_action_name
+from policy_sentry.util.arns import get_service_from_arn
 from policy_sentry.util.access_levels import transform_access_level_text
 
 
@@ -96,6 +98,12 @@ def get_actions_with_access_level(db_session, service, access_level):
     :return: A list of actions
     """
     actions_list = []
+    all_services = get_all_service_prefixes(db_session)
+    if service == "all":
+        for serv in all_services:
+            output = get_actions_with_access_level(
+                db_session, serv, access_level)
+            actions_list.extend(output)
     rows = db_session.query(ActionTable).filter(and_(
         ActionTable.service.like(service),
         ActionTable.access_level.ilike(access_level)
@@ -161,6 +169,40 @@ def get_actions_matching_condition_key(db_session, service, condition_key):
     return actions_list
 
 
+def get_actions_matching_condition_crud_and_arn(db_session, condition_key, access_level, raw_arn):
+    """
+    Get a list of IAM Actions matching a condition key, CRUD level, and raw ARN format.
+
+    :param db_session: SQL Alchemy database session
+    :param condition_key: A condition key, like aws:TagKeys
+    :param access_level: Access level that matches the database value. "Read", "Write", "List", "Tagging", or "Permissions management"
+    :param raw_arn: The raw ARN format in the database, like arn:${Partition}:s3:::${BucketName}
+    :return: List of IAM Actions
+    """
+    actions_list = []
+    looking_for = '%{0}%'.format(condition_key)
+    if raw_arn == "*":
+        rows = db_session.query(ActionTable).filter(and_(
+            # ActionTable.service.ilike(service),
+            ActionTable.access_level.ilike(access_level),
+            ActionTable.resource_arn_format.is_(raw_arn),
+            ActionTable.condition_keys.ilike(looking_for),
+        ))
+    else:
+        service = get_service_from_arn(raw_arn)
+        rows = db_session.query(ActionTable).filter(and_(
+            ActionTable.service.ilike(service),
+            ActionTable.access_level.ilike(access_level),
+            ActionTable.resource_arn_format.ilike(raw_arn),
+            ActionTable.condition_keys.ilike(looking_for),
+        ))
+
+    for row in rows:
+        action = get_full_action_name(row.service, row.name)
+        actions_list.append(action)
+    return actions_list
+
+
 def remove_actions_not_matching_access_level(db_session, actions_list, access_level):
     """
     Given a list of actions, return a list of actions that match an access level

diff --git a/policy_sentry/querying/conditions.py b/policy_sentry/querying/conditions.py
@@ -42,3 +42,4 @@ def get_condition_key_details(db_session, service, condition_key_name):
         'condition_value_type': result.condition_value_type
     }
     return output
+
diff --git a/test/test_query.py b/test/test_query.py
@@ -3,7 +3,8 @@
 from policy_sentry.shared.database import connect_db
 from policy_sentry.querying.actions import get_actions_for_service, get_action_data, \
     get_actions_with_access_level, get_actions_with_arn_type_and_access_level, \
-    get_actions_matching_condition_key, get_actions_that_support_wildcard_arns_only
+    get_actions_matching_condition_key, get_actions_that_support_wildcard_arns_only, \
+    get_actions_matching_condition_crud_and_arn
 from policy_sentry.querying.arns import get_raw_arns_for_service, get_arn_type_details, \
     get_arn_types_for_service
 from policy_sentry.querying.conditions import get_condition_key_details, get_condition_keys_for_service
@@ -147,6 +148,18 @@ def test_get_actions_with_access_level(self):
         self.maxDiff = None
         self.assertListEqual(desired_output, output)
 
+    def test_get_all_actions_with_access_level(self):
+        """test_get_all_actions_with_access_level: Get all actions with a given access level"""
+        # list1 elements should be in result
+        list1 = get_actions_with_access_level(db_session, "all", "Permissions management")
+        list2 = ["s3:deleteaccesspointpolicy", "s3:putbucketacl"]
+        self.maxDiff = None
+        print(list1)
+        # decision = list1 contains all elements of list2 using all()
+        decision = all(elem in list1 for elem in list2)
+        print(f"decision is {decision}")
+        self.assertTrue(decision)
+
     def test_get_actions_with_arn_type_and_access_level(self):
         """test_get_actions_with_arn_type_and_access_level: Tests a function that gets a list of
         actions in a service under different access levels, specific to an ARN format."""
@@ -174,13 +187,52 @@ def test_get_actions_matching_condition_key(self):
         self.maxDiff = None
         self.assertListEqual(desired_output, output)
 
+    def test_get_actions_matching_condition_crud_and_arn(self):
+        """test_get_actions_matching_condition_crud_and_arn: Get a list of IAM Actions matching condition key,
+        CRUD level, and raw ARN"""
+        results = get_actions_matching_condition_crud_and_arn(
+            db_session,
+            "ram:ResourceArn",
+            "Permissions management",
+            "arn:${Partition}:ram:${Region}:${Account}:resource-share/${ResourcePath}"
+        )
+        desired_results = ['ram:createresourceshare']
+        self.assertListEqual(desired_results, results)
+
+    def test_get_actions_matching_condition_crud_and_wildcard_arn(self):
+        """test_get_actions_matching_condition_crud_and_wildcard_arn: Get a list of IAM Actions matching condition key
+        , CRUD level, and raw ARN. Raw ARN equals * in this case"""
+        desired_results = [
+            'swf:pollforactivitytask',
+            'swf:pollfordecisiontask',
+            'swf:respondactivitytaskcompleted',
+            'swf:startworkflowexecution'
+        ]
+        results = get_actions_matching_condition_crud_and_arn(db_session, "swf:taskList.name", "Write", "*")
+        self.assertListEqual(desired_results, results)
+
+        # This one leverages a condition key that is partway through a string in the database
+        # - luckily, SQLAlchemy's ilike function allows us to find it anyway because it's a substring
+        # kms:CallerAccount,kms:EncryptionAlgorithm,kms:EncryptionContextKeys,kms:ViaService
+        desired_results = [
+            'kms:decrypt',
+            'kms:encrypt',
+            'kms:generatedatakey',
+            'kms:generatedatakeypair',
+            'kms:generatedatakeypairwithoutplaintext',
+            'kms:generatedatakeywithoutplaintext',
+            'kms:reencryptfrom',
+            'kms:reencryptto'
+        ]
+        results = get_actions_matching_condition_crud_and_arn(db_session, "kms:EncryptionAlgorithm", "Write", "*")
+        self.assertListEqual(desired_results, results)
+
     # Nuking this test... as AWS adds on more condition keys, this becomes impossible to maintain as a single test.
     # def test_get_actions_matching_condition_key(self):
     #     """test_get_actions_matching_condition_key: Tests a function that creates a list of all IAM
     #     actions that support the supplied condition key."""
     #     # condition_key = "aws:RequestTag"
-    #     desired_list = [
-    #         'appstream:associatefleet', 'appstream:batchassociateuserstack', 'appstream:batchdisassociateuserstack', 'appstream:copyimage', 'appstream:createimagebuilderstreamingurl', 'appstream:createstreamingurl', 'appstream:deletefleet', 'appstream:deleteimage', 'appstream:deleteimagebuilder', 'appstream:deleteimagepermissions', 'appstream:deletestack', 'appstream:disassociatefleet', 'appstream:startfleet', 'appstream:startimagebuilder', 'appstream:stopfleet', 'appstream:stopimagebuilder', 'appstream:tagresource', 'appstream:updatefleet', 'appstream:updateimagepermissions', 'appstream:updatestack', 'appsync:deletegraphqlapi', 'appsync:getgraphqlapi', 'appsync:listtagsforresource', 'appsync:tagresource', 'appsync:updategraphqlapi', 'codecommit:tagresource', 'cognito-identity:createidentitypool', 'cognito-identity:listtagsforresource', 'cognito-identity:tagresource', 'cognito-identity:untagresource', 'cognito-idp:createuserpool', 'cognito-idp:listtagsforresource', 'cognito-idp:tagresource', 'cognito-idp:untagresource', 'cognito-idp:updateuserpool', 'dms:describereplicationinstancetasklogs', 'mobiletargeting:createapp', 'mobiletargeting:createcampaign', 'mobiletargeting:createsegment', 'mobiletargeting:deletecampaign', 'mobiletargeting:deletesegment', 'mobiletargeting:getapp', 'mobiletargeting:getapps', 'mobiletargeting:getcampaign', 'mobiletargeting:getcampaignversion', 'mobiletargeting:getcampaignversions', 'mobiletargeting:getcampaigns', 'mobiletargeting:getsegment', 'mobiletargeting:getsegmentversion', 'mobiletargeting:getsegmentversions', 'mobiletargeting:getsegments', 'mobiletargeting:listtagsforresource', 'mobiletargeting:tagresource', 'mobiletargeting:untagresource', 'mobiletargeting:updatecampaign', 'mobiletargeting:updatesegment']
+    #     desired_list = []
     #     stuff = "aws:ResourceTag/${TagKey}"
     #     output = get_actions_matching_condition_key(db_session, service=None, condition_key=stuff)
     #     self.maxDiff = None