Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from .resources.iam_role import IAMRole
from .resources.s3_bucket_policy import S3BucketPolicy
from .resources.security_group import SecurityGroup
from .resources.security_group_egress import SecurityGroupEgress
from .resources.security_group_ingress import SecurityGroupIngress
from .resources.sqs_queue_policy import SQSQueuePolicy
from .resources.sns_topic_policy import SNSTopicPolicy
from .resources.kms_key import KMSKey
_RESOURCE_MAP = {
"AWS::EC2::SecurityGroup": SecurityGroup,
"AWS::EC2::SecurityGroupEgress": SecurityGroupEgress,
"AWS::EC2::SecurityGroupIngress": SecurityGroupIngress,
"AWS::IAM::Group": IAMGroup,
"AWS::IAM::ManagedPolicy": IAMManagedPolicy,
"AWS::IAM::Policy": IAMPolicy,
"AWS::IAM::Role": IAMRole,
"AWS::IAM::User": IAMUser,
"AWS::KMS::Key": KMSKey,
"AWS::S3::BucketPolicy": S3BucketPolicy,
"AWS::SNS::TopicPolicy": SNSTopicPolicy,
"AWS::SQS::QueuePolicy": SQSQueuePolicy,
}
_DEFAULT_RESOURCE = Resource
def create_resource(logical_id: str, value: Dict[str, Any]) -> Resource:
resource = _RESOURCE_MAP.get(value.get("Type"), _DEFAULT_RESOURCE)
return resource(logical_id, value)
def invoke(self, cfmodel: CFModel, extras: Optional[Dict] = None) -> Result:
result = Result()
for logical_id, resource in cfmodel.Resources.items():
if isinstance(resource, IAMManagedPolicy) and resource.Properties.Users:
self.add_failure_to_result(result, self.REASON.format(logical_id), resource_ids={logical_id})
return result
def invoke(self, cfmodel):
for logical_id, resource in cfmodel.Resources.items():
if isinstance(resource, IAMManagedPolicy) and resource.Properties.PolicyDocument.allowed_actions_with(
REGEX_WILDCARD_POLICY_ACTION
):
self.add_failure(type(self).__name__, self.REASON.format(logical_id), resource_ids={logical_id})
from pycfmodel.constants import REGEX_AWS_MANAGED_ARN
from pycfmodel.model.resources.iam_managed_policy import IAMManagedPolicy
client = boto3.client("iam")
destination_folder = Path(__file__).parent
for response in client.get_paginator("list_policies").paginate(Scope="AWS"):
for policy in response["Policies"]:
policy_version_response = client.get_policy_version(
PolicyArn=policy["Arn"], VersionId=policy["DefaultVersionId"]
)
print(f"Dumping {policy['Arn']}")
destination_file = destination_folder / (REGEX_AWS_MANAGED_ARN.match(policy["Arn"]).group(1) + ".json")
destination_file.parent.mkdir(parents=True, exist_ok=True)
destination_file.write_text(
IAMManagedPolicy(
**{
"Arn": policy["Arn"],
"Properties": {
"ManagedPolicyName": policy["PolicyName"],
"Path": policy_version_response["PolicyVersion"].get("Path"),
"PolicyDocument": policy_version_response["PolicyVersion"]["Document"],
},
from pycfmodel.model.resources.iam_group import IAMGroup
from pycfmodel.model.resources.iam_managed_policy import IAMManagedPolicy
from pycfmodel.model.resources.iam_policy import IAMPolicy
from pycfmodel.model.resources.iam_role import IAMRole
from pycfmodel.model.resources.iam_user import IAMUser
from pycfmodel.model.resources.kms_key import KMSKey
from pycfmodel.model.resources.s3_bucket_policy import S3BucketPolicy
from pycfmodel.model.resources.security_group import SecurityGroup
from pycfmodel.model.resources.security_group_egress import SecurityGroupEgress
from pycfmodel.model.resources.security_group_ingress import SecurityGroupIngress
from pycfmodel.model.resources.sns_topic_policy import SNSTopicPolicy
from pycfmodel.model.resources.sqs_queue_policy import SQSQueuePolicy
ResourceModels = Union[
IAMGroup,
IAMManagedPolicy,
IAMPolicy,
IAMRole,
IAMUser,
KMSKey,
S3BucketPolicy,
SecurityGroup,
SecurityGroupEgress,
SecurityGroupIngress,
SNSTopicPolicy,
SQSQueuePolicy,
]
self.add_failure_to_result(
result, self.REASON.format(logical_id, "AssumeRolePolicy"), resource_ids={logical_id},
)
# check other policies of the IAM role.
if resource.Properties.Policies:
for policy in resource.Properties.Policies:
if policy.PolicyDocument.allowed_actions_with(REGEX_WILDCARD_POLICY_ACTION):
self.add_failure_to_result(
result,
self.REASON.format(logical_id, f"{policy.PolicyName} policy"),
resource_ids={logical_id},
)
# check AWS::IAM::ManagedPolicy.
elif isinstance(resource, IAMManagedPolicy) and resource.Properties.PolicyDocument.allowed_actions_with(
REGEX_WILDCARD_POLICY_ACTION
):
self.add_failure_to_result(
result, self.REASON.format(logical_id, "AWS::IAM::ManagedPolicy"), resource_ids={logical_id},
)
return result