Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

access_check now prints json; shows references to the policies #703

Merged
merged 6 commits into from
Jun 19, 2020
Merged

access_check now prints json; shows references to the policies #703

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
973047e
access_check now prints json; shows references to the policies
0xdabbad00 May 13, 2020
1b40141
Ensure we use the latest parliament version
0xdabbad00 May 13, 2020
0b14955
Fix tests
0xdabbad00 May 13, 2020
0dff45e
Bump version
0xdabbad00 May 13, 2020
8a67580
Black formatting
0xdabbad00 May 13, 2020
0f9f0c3
Merge branch 'master' into track_statements
0xdabbad00 Jun 19, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
2 changes: 1 addition & 1 deletion cloudmapper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
import pkgutil
import importlib

__version__ = "2.8.2"
__version__ = "2.8.3"


def show_help(commands):
Expand Down
118 changes: 83 additions & 35 deletions commands/access_check.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,16 @@

__description__ = "[proof-of-concept] Check who has access to a resource"


def replace_principal_variables(reference, principal):
"""
Given a resource reference string (ie. the Resource string from an IAM policy) and a prinicipal, replace any variables in the resource string that are principal related.
"""
reference = reference.lower()
for tag in principal.tags:
reference = reference.replace("${aws:principaltag/"+tag["Key"].lower()+"}", tag["Value"].lower())
reference = reference.replace(
"${aws:principaltag/" + tag["Key"].lower() + "}", tag["Value"].lower()
)

reference = reference.replace("${aws:principaltype}", principal.mytype.lower())
return reference
Expand All @@ -36,7 +39,7 @@ def apply_condition_function(condition_function, left_side, right_side):
# "t2.*",
# "m3.*"
# ]}}
#
#
# or
#
# "Condition": {
Expand All @@ -48,7 +51,7 @@ def apply_condition_function(condition_function, left_side, right_side):
# ]
# }
# }
#
#
# or
#
# "Condition": {
Expand All @@ -60,27 +63,28 @@ def apply_condition_function(condition_function, left_side, right_side):
# }
# }

if condition_function == 'StringEquals':
if condition_function == "StringEquals":
return left_side == right_side
elif condition_function == 'StringNotEquals':
elif condition_function == "StringNotEquals":
return left_side != right_side
elif condition_function == 'StringEqualsIgnoreCase':
elif condition_function == "StringEqualsIgnoreCase":
return left_side.lower() == right_side.lower()
elif condition_function == 'StringNotEqualsIgnoreCase':
elif condition_function == "StringNotEqualsIgnoreCase":
return left_side.lower() != right_side.lower()
elif condition_function == 'StringLike':

elif condition_function == "StringLike":
right_side.replace("*", ".*")
matcher = re.compile("^{}$".format(right_side))
return matcher.match(left_side)
elif condition_function == 'StringNotLike':
elif condition_function == "StringNotLike":
right_side.replace("*", ".*")
matcher = re.compile("^{}$".format(right_side))
return not matcher.match(left_side)

# TODO Need to handle other operators from https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_elements_condition_operators.html
return None


def get_condition_result(condition_function, condition_values, resource_arn, principal):
"""
Given a condition_function such as: 'StringEquals'
Expand All @@ -93,9 +97,12 @@ def get_condition_result(condition_function, condition_values, resource_arn, pri
for k in condition_values:
if k.startswith("aws:PrincipalTag/"):
for tag in principal.tags:
if k == "aws:PrincipalTag/"+tag["Key"]:
results.append(apply_condition_function(condition_function, tag["Value"], condition_values[k]))

if k == "aws:PrincipalTag/" + tag["Key"]:
results.append(
apply_condition_function(
condition_function, tag["Value"], condition_values[k]
)
)

# The array results should now look something like [True, False, True],
# although more commonly is just [], [False], or [True]
Expand All @@ -114,7 +121,10 @@ def get_condition_result(condition_function, condition_values, resource_arn, pri

return None

def get_privilege_statements(policy_doc, privilege_matches, resource_arn, principal):

def get_privilege_statements(
policy_doc, privilege_matches, resource_arn, principal, policy_identifier
):
policy = parliament.policy.Policy(policy_doc)
policy.analyze()

Expand All @@ -138,10 +148,16 @@ def get_privilege_statements(policy_doc, privilege_matches, resource_arn, princi
stmts = references[reference]
condition_allowed_stmts = []
for stmt in stmts:
stmt.set_policy_identifier(policy_identifier)
allowed_by_conditions = True
for condition_function in stmt.stmt.get("Condition", {}):
condition_values = stmt.stmt["Condition"][condition_function]
condition_result = get_condition_result(condition_function, condition_values, resource_arn, principal)
condition_result = get_condition_result(
condition_function,
condition_values,
resource_arn,
principal,
)
# TODO Need to do something different for Deny, to avoid false negatives
if condition_result is not None:
if condition_result == False:
Expand Down Expand Up @@ -255,7 +271,11 @@ def access_check_command(accounts, config, args):
policy_doc = get_managed_policy(iam, policy["PolicyArn"])
privileged_statements.extend(
get_privilege_statements(
policy_doc, privilege_matches, args.resource_arn, principal
policy_doc,
privilege_matches,
args.resource_arn,
principal,
policy["PolicyArn"],
)
)

Expand All @@ -264,7 +284,11 @@ def access_check_command(accounts, config, args):
policy_doc = policy["PolicyDocument"]
privileged_statements.extend(
get_privilege_statements(
policy_doc, privilege_matches, args.resource_arn, principal
policy_doc,
privilege_matches,
args.resource_arn,
principal,
role["Arn"] + ":" + policy["PolicyName"],
)
)

Expand All @@ -282,19 +306,24 @@ def access_check_command(accounts, config, args):
if boundary is not None:
policy_doc = get_managed_policy(iam, boundary["PermissionsBoundaryArn"])
boundary_statements = get_privilege_statements(
policy_doc, privilege_matches, args.resource_arn, principal
policy_doc,
privilege_matches,
args.resource_arn,
principal,
boundary["PermissionsBoundaryArn"],
)

# Find the allowed privileges
allowed_privileges = get_allowed_privileges(
privilege_matches, privileged_statements, boundary_statements
)
for priv in allowed_privileges:
print(
"{} - {}:{}".format(
role["Arn"], priv["privilege_prefix"], priv["privilege_name"]
)
)
priv_object = {
"principal": role["Arn"],
"privilege": f"{priv['privilege_prefix']}:{priv['privilege_name']}",
"references": list(priv["references"]),
}
print(json.dumps(priv_object))

# Check the users
for user in iam["UserDetailList"]:
Expand All @@ -307,7 +336,11 @@ def access_check_command(accounts, config, args):
policy_doc = get_managed_policy(iam, policy["PolicyArn"])
privileged_statements.extend(
get_privilege_statements(
policy_doc, privilege_matches, args.resource_arn, principal
policy_doc,
privilege_matches,
args.resource_arn,
principal,
policy["PolicyArn"],
)
)

Expand All @@ -316,7 +349,11 @@ def access_check_command(accounts, config, args):
policy_doc = policy["PolicyDocument"]
privileged_statements.extend(
get_privilege_statements(
policy_doc, privilege_matches, args.resource_arn, principal
policy_doc,
privilege_matches,
args.resource_arn,
principal,
user["Arn"] + ":" + policy["PolicyName"],
)
)

Expand All @@ -333,6 +370,7 @@ def access_check_command(accounts, config, args):
privilege_matches,
args.resource_arn,
principal,
policy["PolicyArn"],
)
)

Expand All @@ -344,6 +382,7 @@ def access_check_command(accounts, config, args):
privilege_matches,
args.resource_arn,
principal,
group["Arn"] + ":" + policy["PolicyName"],
)
)

Expand All @@ -361,19 +400,24 @@ def access_check_command(accounts, config, args):
if boundary is not None:
policy_doc = get_managed_policy(iam, boundary["PermissionsBoundaryArn"])
boundary_statements = get_privilege_statements(
policy_doc, privilege_matches, args.resource_arn, principal
policy_doc,
privilege_matches,
args.resource_arn,
principal,
boundary["PermissionsBoundaryArn"],
)

# Find the allowed privileges
allowed_privileges = get_allowed_privileges(
privilege_matches, privileged_statements, boundary_statements
)
for priv in allowed_privileges:
print(
"{} - {}:{}".format(
user["Arn"], priv["privilege_prefix"], priv["privilege_name"]
)
)
priv_object = {
"principal": user["Arn"],
"privilege": f"{priv['privilege_prefix']}:{priv['privilege_name']}",
"references": list(priv["references"]),
}
print(json.dumps(priv_object))


def get_managed_policy(iam, policy_arn):
Expand Down Expand Up @@ -406,7 +450,7 @@ def is_allowed(privilege_prefix, privilege_name, statements):
is_allowed = False

if is_allowed:
return True
return stmts_for_privilege
return False


Expand All @@ -425,11 +469,15 @@ def get_allowed_privileges(
):
continue

if is_allowed(
allowed_stmts = is_allowed(
privilege["privilege_prefix"],
privilege["privilege_name"],
privileged_statements,
):
)
if allowed_stmts:
privilege["references"] = set()
for stmt in allowed_stmts:
privilege["references"].add(stmt.policy_id)
allowed_privileges.append(privilege)
return allowed_privileges

Expand Down
2 changes: 1 addition & 1 deletion requirements.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,8 @@ mccabe==0.6.1
mock==4.0.2
netaddr==0.7.19
nose==1.3.7
parliament==0.4.14
pandas==1.0.3
parliament==0.3.6
policyuniverse==1.1.0.1
pycodestyle==2.5.0
pyflakes==2.2.0
Expand Down