Add basic support for guessing additional actions #22

tests/cloudtrail/map_to_iam_sanity_test.py

@@ -1,10 +1,13 @@
 import os
+
 import pytest
 
-from tests.test_utils_iam import all_aws_api_methods, all_known_iam_actions
+from tests.test_utils_iam import all_aws_api_methods
 from trailscraper.cloudtrail import Record
 
 # Actions that we know are supported but aren't in our documentation just yet:
+from trailscraper.iam import all_known_iam_permissions
+
 UNDOCUMENTED = {
     "elasticbeanstalk:CreatePlatformVersion",
     "elasticbeanstalk:DeletePlatformVersion",
@@ -1180,7 +1183,7 @@ def unknown_actions():
         if statement is not None:
             iam_actions_from_api_calls.add(statement.Action[0].json_repr())
 
-    known_actions = all_known_iam_actions()
+    known_actions = all_known_iam_permissions()
 
     return iam_actions_from_api_calls.difference(known_actions)
 

tests/iam/action_test.py

@@ -0,0 +1,48 @@
+import pytest
+
+from trailscraper.iam import Action
+
+
+@pytest.mark.parametrize("test_input,expected", [
+    (Action('autoscaling', 'DescribeLaunchConfigurations'), "LaunchConfiguration"),
+    (Action('autoscaling', 'CreateLaunchConfiguration'), "LaunchConfiguration"),
+    (Action('autoscaling', 'DeleteLaunchConfiguration'), "LaunchConfiguration"),
+    (Action('autoscaling', 'UpdateAutoScalingGroup'), "AutoScalingGroup"),
+])
+def test_create_base_action(test_input, expected):
+    assert test_input._base_action() == expected
+
+
+@pytest.mark.parametrize("test_input,expected", [
+    (Action('autoscaling', 'DescribeLaunchConfigurations'), [
+        Action('autoscaling', 'CreateLaunchConfiguration'),
+        Action('autoscaling', 'DeleteLaunchConfiguration'),
+    ]),
+    (Action('autoscaling', 'CreateLaunchConfiguration'), [
+        Action('autoscaling', 'DeleteLaunchConfiguration'),
+        Action('autoscaling', 'DescribeLaunchConfigurations'),
+    ]),
+    (Action('autoscaling', 'DeleteLaunchConfiguration'), [
+        Action('autoscaling', 'CreateLaunchConfiguration'),
+        Action('autoscaling', 'DescribeLaunchConfigurations'),
+    ]),
+    (Action('autoscaling', 'UpdateAutoScalingGroup'), [
+        Action('autoscaling', 'CreateAutoScalingGroup'),
+        Action('autoscaling', 'DeleteAutoScalingGroup'),
+        Action('autoscaling', 'DescribeAutoScalingGroups'),
+    ]),
+    (Action('autoscaling', 'DeleteAutoScalingGroup'), [
+        Action('autoscaling', 'CreateAutoScalingGroup'),
+        Action('autoscaling', 'UpdateAutoScalingGroup'),
+        Action('autoscaling', 'DescribeAutoScalingGroups'),
+    ]),
+])
+def test_find_create_action(test_input, expected):
+    assert test_input.matching_actions() == expected
+
+
+# TODO:
+# * Attach/Detach?
+# * list
+# * Encrypt/Decrypt/GenerateDataKey?
+# * Put

tests/iam/known_iam_actions_test.py

@@ -0,0 +1,19 @@
+from trailscraper.iam import all_known_iam_permissions, Action, known_iam_actions
+
+
+def test_all_iam_permissions():
+    permissions = all_known_iam_permissions()
+
+    assert permissions != []
+    assert "ec2:DescribeInstances" in permissions
+    assert len(permissions) == len(set(permissions)), "expected no duplicates"
+
+
+def test_known_iam_action_for_prefix():
+    actions = known_iam_actions("acm")
+    assert len(actions) == 10
+    assert Action("acm","DescribeCertificate") in actions
+
+
+def test_known_iam_action_for_prefix_does_not_fail_if_action_not_found():
+    assert known_iam_actions("something-unknown") == []

tests/iam/policy_document_test.py

@@ -1,6 +1,7 @@
 import json
+from StringIO import StringIO
 
-from trailscraper.iam import PolicyDocument, Statement, Action
+from trailscraper.iam import PolicyDocument, Statement, Action, parse_policy_document
 
 
 def test_policy_document_renders_to_json():
@@ -50,4 +51,31 @@ def test_policy_document_renders_to_json():
     ],
     "Version": "2012-10-17"
 }'''
-    assert json.loads(pd.to_json()) == json.loads(expected_json)
+    assert json.loads(pd.to_json()) == json.loads(expected_json)
+
+
+def test_json_parses_to_policy_document():
+    pd = PolicyDocument(
+        Version="2012-10-17",
+        Statement=[
+            Statement(
+                Effect="Allow",
+                Action=[
+                    Action('autoscaling', 'DescribeLaunchConfigurations'),
+                ],
+                Resource=["*"]
+            ),
+            Statement(
+                Effect="Allow",
+                Action=[
+                    Action('sts', 'AssumeRole'),
+                ],
+                Resource=[
+                    "arn:aws:iam::111111111111:role/someRole"
+                ]
+            )
+        ]
+    )
+
+    assert parse_policy_document(StringIO(pd.to_json())).to_json() == pd.to_json()
+

tests/integration/cli_guess_test.py

@@ -0,0 +1,66 @@
+from StringIO import StringIO
+
+from click.testing import CliRunner
+
+from trailscraper import cli
+from trailscraper.iam import PolicyDocument, Statement, Action, parse_policy_document
+
+
+def test_should_guess_create_statements():
+    input_policy = PolicyDocument(
+        Version="2012-10-17",
+        Statement=[
+            Statement(
+                Effect="Allow",
+                Action=[
+                    Action('autoscaling', 'DescribeLaunchConfigurations'),
+                ],
+                Resource=["*"]
+            ),
+            Statement(
+                Effect="Allow",
+                Action=[
+                    Action('sts', 'AssumeRole'),
+                ],
+                Resource=[
+                    "arn:aws:iam::111111111111:role/someRole"
+                ]
+            )
+        ]
+    )
+
+    expected_output = PolicyDocument(
+        Version="2012-10-17",
+        Statement=[
+            Statement(
+                Effect="Allow",
+                Action=[
+                    Action('autoscaling', 'DescribeLaunchConfigurations'),
+                ],
+                Resource=["*"]
+            ),
+            Statement(
+                Effect="Allow",
+                Action=[
+                    Action('autoscaling', 'CreateLaunchConfiguration'),
+                    Action('autoscaling', 'DeleteLaunchConfiguration'),
+                ],
+                Resource=["*"]
+            ),
+            Statement(
+                Effect="Allow",
+                Action=[
+                    Action('sts', 'AssumeRole'),
+                ],
+                Resource=[
+                    "arn:aws:iam::111111111111:role/someRole"
+                ]
+            )
+        ]
+    )
+
+    runner = CliRunner()
+    result = runner.invoke(cli.root_group, args=["guess"], input=StringIO(input_policy.to_json()))
+    assert result.exit_code == 0
+    assert parse_policy_document(StringIO(result.output)) == expected_output
+

tests/test_utils_iam.py

@@ -1,6 +1,5 @@
 import json
 import logging
-import os
 
 from trailscraper.boto_service_definitions import boto_service_definition_files
 
@@ -19,23 +18,10 @@ def all_aws_api_methods():
     return set(result)
 
 
-def all_known_iam_actions():
-    with open(os.path.join(os.path.dirname(__file__), 'known-iam-actions.txt')) as iam_file:
-        return set([line.rstrip('\n') for line in iam_file.readlines()])
-
-
 def test_all_aws_api_methods():
     api_methods = all_aws_api_methods()
 
     assert api_methods != []
     assert "ec2:DescribeInstances" in api_methods
     assert len(api_methods) == len(set(api_methods)), "expected no duplicates"
 
-
-def test_all_iam_permissions_known_in_cloudonaut():
-    permissions = all_known_iam_actions()
-
-    assert permissions != []
-    assert "ec2:DescribeInstances" in permissions
-    assert len(permissions) == len(set(permissions)), "expected no duplicates"
-

trailscraper/cli.py

@@ -10,6 +10,8 @@
 from trailscraper import time_utils, policy_generator
 from trailscraper.cloudtrail import load_from_dir, load_from_api, last_event_timestamp_in_dir, filter_records, \
     parse_records
+from trailscraper.guess import guess_statements
+from trailscraper.iam import parse_policy_document
 from trailscraper.s3_download import download_cloudtrail_logs
 
 
@@ -104,6 +106,15 @@ def generate():
     click.echo(policy.to_json())
 
 
+@click.command("guess")
+def guess():
+    """Extend a policy passed in through STDIN by guessing related actions"""
+    stdin = click.get_text_stream('stdin')
+    policy = parse_policy_document(stdin)
+    policy = guess_statements(policy)
+    click.echo(policy.to_json())
+
+
 @click.command("last-event-timestamp")
 @click.option('--log-dir', default="~/.trailscraper/logs", type=click.Path(),
               help='Where to put logfiles')
@@ -116,4 +127,5 @@ def last_event_timestamp(log_dir):
 root_group.add_command(download)
 root_group.add_command(select)
 root_group.add_command(generate)
+root_group.add_command(guess)
 root_group.add_command(last_event_timestamp)

trailscraper/guess.py

@@ -0,0 +1,25 @@
+"""Logic to guess related IAM statements"""
+from trailscraper.iam import PolicyDocument, Statement
+
+
+def _guess_actions(actions):
+    return [item for action in actions
+            for item in action.matching_actions()]
+
+
+def _extend_statement(statement):
+    extended_actions = _guess_actions(statement.Action)
+    if extended_actions:
+        return [statement, Statement(Action=extended_actions,
+                                     Effect=statement.Effect,
+                                     Resource=["*"])]
+
+    return [statement]
+
+
+def guess_statements(policy):
+    """Guess additional create actions"""
+    extended_statements = [item for statement in policy.Statement
+                           for item in _extend_statement(statement)]
+
+    return PolicyDocument(Version=policy.Version, Statement=extended_statements)

trailscraper/iam.py

@@ -1,5 +1,11 @@
 """Classes to deal with IAM Policies"""
 import json
+import os
+
+import re
+from toolz import pipe
+from toolz.curried import groupby as groupbyz
+from toolz.curried import map as mapz
 
 
 class BaseElement(object):
@@ -35,6 +41,25 @@ def __init__(self, prefix, action):
     def json_repr(self):
         return ':'.join([self.prefix, self.action])
 
+    def _base_action(self):
+        without_prefix = re.sub(r"(Describe)|(Create)|(Delete)|(Update)", "", self.action)
+        without_plural = re.sub(r"s$", "", without_prefix)
+
+        return without_plural
+
+    def matching_actions(self):
+        """Return a matching create action for this Action"""
+        potential_matches = [
+            Action(prefix=self.prefix, action="Create" + self._base_action()),
+            Action(prefix=self.prefix, action="Update" + self._base_action()),
+            Action(prefix=self.prefix, action="Delete" + self._base_action()),
+            Action(prefix=self.prefix, action="Describe" + self._base_action()),
+            Action(prefix=self.prefix, action="Describe" + self._base_action()+"s"),
+        ]
+        return [potential_match for potential_match in potential_matches
+                                if potential_match in known_iam_actions(self.prefix) and potential_match != self]
+
+
 
 class Statement(BaseElement):
     """Statement in an IAM Policy."""
@@ -106,3 +131,42 @@ def default(self, o):  # pylint: disable=method-hidden
         if hasattr(o, 'json_repr'):
             return o.json_repr()
         return json.JSONEncoder.default(self, o)
+
+
+def _parse_action(action):
+    parts = action.split(":")
+    return Action(parts[0], parts[1])
+
+
+def _parse_statement(statement):
+    return Statement(Action=[_parse_action(action) for action in statement['Action']],
+                     Effect=statement['Effect'],
+                     Resource=statement['Resource'])
+
+
+def _parse_statements(json_data):
+    # TODO: jsonData could also be dict, aka one statement; similar things happen in the rest of the policy # pylint: disable=fixme
+    # https://github.com/flosell/iam-policy-json-to-terraform/blob/fafc231/converter/decode.go#L12-L22
+    return [_parse_statement(statement) for statement in json_data]
+
+
+def parse_policy_document(stream):
+    """Parse a stream of JSON data to a PolicyDocument object"""
+    json_dict = json.load(stream)
+    return PolicyDocument(_parse_statements(json_dict['Statement']), Version=json_dict['Version'])
+
+
+def all_known_iam_permissions():
+    "Return a list of all known IAM actions"
+    with open(os.path.join(os.path.dirname(__file__), 'known-iam-actions.txt')) as iam_file:
+        return set([line.rstrip('\n') for line in iam_file.readlines()])
+
+
+def known_iam_actions(prefix):
+    """Return known IAM actions for a prefix, e.g. all ec2 actions"""
+    # This could be memoized for performance improvements
+    knowledge = pipe(all_known_iam_permissions(),
+                     mapz(_parse_action),
+                     groupbyz(lambda x: x.prefix))
+
+    return knowledge.get(prefix, [])