Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Args:
account_number (string): The current account number Repokid is being run against
Returns:
None
"""
conn = config['connection_iam']
conn['account_number'] = account_number
roles = Roles([Role(role_data) for role_data in list_roles(**conn)])
active_roles = []
LOGGER.info('Updating role data for account {}'.format(account_number))
for role in tqdm(roles):
role.account = account_number
current_policies = get_role_inline_policies(role.as_dict(), **conn) or {}
active_roles.append(role.role_id)
roledata.update_role_data(dynamo_table, account_number, role, current_policies)
LOGGER.info('Finding inactive accounts')
roledata.find_and_mark_inactive(dynamo_table, account_number, active_roles)
LOGGER.info('Filtering roles')
plugins = FilterPlugins()
# Blacklist needs to know the current account
config['filter_config']['BlacklistFilter']['current_account'] = account_number
for plugin_path in config.get('active_filters'):
plugin_name = plugin_path.split(':')[1]
plugins.load_plugin(plugin_path, config=config['filter_config'].get(plugin_name, None))
rows.append(
[
index,
policies_version["Source"],
policies_version["Discovered"],
len(policy_permissions),
roledata._get_services_in_permissions(policy_permissions),
]
)
print(tabulate(rows, headers=headers))
return
conn = config["connection_iam"]
conn["account_number"] = account_number
current_policies = get_role_inline_policies(role.as_dict(), **conn)
if selection:
pp = pprint.PrettyPrinter()
print("Will restore the following policies:")
pp.pprint(role.policies[int(selection)]["Policy"])
print("Current policies:")
pp.pprint(current_policies)
current_permissions, _ = roledata._get_permissions_in_policy(
role.policies[-1]["Policy"]
)
selected_permissions, _ = roledata._get_permissions_in_policy(
role.policies[int(selection)]["Policy"]
)
errors.append(error)
if repoed_policies:
LOGGER.info('Replacing Policies With: \n{}'.format(json.dumps(repoed_policies, indent=2, sort_keys=True)))
for policy_name, policy in repoed_policies.items():
try:
ca.call('iam.client.put_role_policy', RoleName=role.role_name, PolicyName=policy_name,
PolicyDocument=json.dumps(policy, indent=2, sort_keys=True))
except botocore.exceptions.ClientError as e:
error = 'Exception calling PutRolePolicy on {role}/{policy}\n{e}\n'.format(
role=role.role_name, policy=policy_name, e=str(e))
LOGGER.error(error)
errors.append(error)
current_policies = get_role_inline_policies(role.as_dict(), **conn) or {}
roledata.add_new_policy_version(dynamo_table, role, current_policies, 'Repo')
# regardless of whether we're successful we want to unschedule the repo
set_role_data(dynamo_table, role.role_id, {'RepoScheduled': 0})
repokid.hooks.call_hooks(hooks, 'AFTER_REPO', {'role': role})
if not errors:
# repos will stay scheduled until they are successful
set_role_data(dynamo_table, role.role_id, {'Repoed': datetime.datetime.utcnow().isoformat()})
_update_repoed_description(role.role_name, **conn)
_update_role_data(role, dynamo_table, account_number, config, conn, hooks, source='Repo', add_no_repo=False)
LOGGER.info('Successfully repoed role: {}'.format(role.role_name))
return errors
conn = config["connection_iam"]
conn["account_number"] = account_number
for name in deleted_policy_names:
error = _delete_policy(name, role, account_number, conn)
if error:
LOGGER.error(error)
errors.append(error)
if repoed_policies:
error = _replace_policies(repoed_policies, role, account_number, conn)
if error:
LOGGER.error(error)
errors.append(error)
current_policies = get_role_inline_policies(role.as_dict(), **conn) or {}
roledata.add_new_policy_version(dynamo_table, role, current_policies, "Repo")
# regardless of whether we're successful we want to unschedule the repo
set_role_data(
dynamo_table, role.role_id, {"RepoScheduled": 0, "ScheduledPerms": []}
)
repokid.hooks.call_hooks(hooks, "AFTER_REPO", {"role": role})
if not errors:
# repos will stay scheduled until they are successful
set_role_data(
dynamo_table,
role.role_id,
{"Repoed": datetime.datetime.utcnow().isoformat()},
)
Does not update:
- Filters
- Active/inactive roles
Args:
role (Role)
dynamo_table
account_number
conn (dict)
source: repo, rollback, etc
add_no_repo: if set to True newly discovered permissions will be added to no repo list
Returns:
None
"""
current_policies = get_role_inline_policies(role.as_dict(), **conn) or {}
roledata.update_role_data(
dynamo_table,
account_number,
role,
current_policies,
source=source,
add_no_repo=add_no_repo,
)
aardvark_data = _get_aardvark_data(config["aardvark_api_location"], arn=role.arn)
if not aardvark_data:
return
batch_processing = config.get("query_role_data_in_batch", False)
batch_size = config.get("batch_processing_size", 100)
Does not update:
- Filters
- Active/inactive roles
Args:
role (Role)
dynamo_table
account_number
conn (dict)
source: repo, rollback, etc
add_no_repo: if set to True newly discovered permissions will be added to no repo list
Returns:
None
"""
current_policies = get_role_inline_policies(role.as_dict(), **conn) or {}
roledata.update_role_data(dynamo_table, account_number, role, current_policies, source=source,
add_no_repo=add_no_repo)
aardvark_data = _get_aardvark_data(config['aardvark_api_location'], arn=role.arn)
if not aardvark_data:
return
role.aa_data = aardvark_data[role.arn]
roledata._calculate_repo_scores([role], config['filter_config']['AgeFilter']['minimum_age'], hooks)
set_role_data(dynamo_table, role.role_id, {'AAData': role.aa_data,
'TotalPermissions': role.total_permissions,
'RepoablePermissions': role.repoable_permissions,
'RepoableServices': role.repoable_services})
roledata.update_stats(dynamo_table, [role], source=source)
def get_inline_policies(role, **conn):
return get_role_inline_policies(role, **conn)