ghndrx/prowler
1
0
Fork 0
mirror of https://github.com/ghndrx/prowler.git synced 2026-02-10 06:45:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(sts): Use the right region to validate credentials (#2349)

Browse Source 
Co-authored-by: Sergio Garcia <sergargar1@gmail.com>
This commit is contained in:
Pepe Fagoaga
2023-05-18 15:51:57 +02:00
committed by GitHub
parent 0bd26b19d7
commit e84f5f184e
18 changed files with 1089 additions and 345 deletions
Download Patch File Download Diff File Expand all files Collapse all files

296
tests/providers/aws/lib/arn/arn_test.py
View File

@@ -1,30 +1,255 @@
import sure # noqa
from pytest import raises
from prowler.providers.aws.lib.arn.arn import arn_parsing, is_valid_arn
from prowler.providers.aws.lib.arn.arn import is_valid_arn, parse_iam_credentials_arn
from prowler.providers.aws.lib.arn.error import (
RoleArnParsingEmptyResource,
RoleArnParsingFailedMissingFields,
RoleArnParsingIAMRegionNotEmpty,
RoleArnParsingInvalidAccountID,
RoleArnParsingInvalidResourceType,
RoleArnParsingPartitionEmpty,
RoleArnParsingServiceNotIAMnorSTS,
)
from prowler.providers.aws.lib.arn.models import ARN
ACCOUNT_ID = "123456789012"
RESOURCE_TYPE = "role"
RESOURCE_TYPE_ROLE = "role"
RESOUCE_TYPE_USER = "user"
IAM_ROLE = "test-role"
IAM_SERVICE = "iam"
COMMERCIAL_PARTITION = "aws"
CHINA_PARTITION = "aws-cn"
GOVCLOUD_PARTITION = "aws-us-gov"
class Test_ARN_Parsing:
def test_arn_parsing(self):
def test_ARN_model(self):
# https://gist.github.com/cmawhorter/80bf94f12bf7516d50a7d61ed28859d3
test_cases = [
"arn:aws:elasticbeanstalk:us-east-1:123456789012:environment/My App/MyEnvironment",
"arn:aws:iam::123456789012:user/David",
"arn:aws:rds:eu-west-1:123456789012:db:mysql-db",
"arn:aws:s3:::my_corporate_bucket/exampleobject.png",
"arn:aws:artifact:::report-package/Certifications and Attestations/SOC/*",
"arn:aws:artifact:::report-package/Certifications and Attestations/ISO/*",
"arn:aws:artifact:::report-package/Certifications and Attestations/PCI/*",
# "arn:aws:autoscaling:us-east-1:123456789012:scalingPolicy:c7a27f55-d35e-4153-b044-8ca9155fc467:autoScalingGroupName/my-test-asg1:policyName/my-scaleout-policy",
"arn:aws:acm:us-east-1:123456789012:certificate/12345678-1234-1234-1234-123456789012",
"arn:aws:cloudformation:us-east-1:123456789012:stack/MyProductionStack/abc9dbf0-43c2-11e3-a6e8-50fa526be49c",
"arn:aws:cloudformation:us-east-1:123456789012:changeSet/MyProductionChangeSet/abc9dbf0-43c2-11e3-a6e8-50fa526be49c",
"arn:aws:cloudsearch:us-east-1:123456789012:domain/imdb-movies",
"arn:aws:cloudtrail:us-east-1:123456789012:trail/mytrailname",
"arn:aws:events:us-east-1:*:*",
"arn:aws:events:us-east-1:account-id:*",
"arn:aws:events:us-east-1:account-id:rule/rule_name",
"arn:aws:logs:us-east-1:*:*",
"arn:aws:logs:us-east-1:account-id:*",
"arn:aws:logs:us-east-1:account-id:log-group:log_group_name",
"arn:aws:logs:us-east-1:account-id:log-group:log_group_name:*",
"arn:aws:logs:us-east-1:account-id:log-group:log_group_name_prefix*",
"arn:aws:logs:us-east-1:account-id:log-group:log_group_name:log-stream:log_stream_name",
"arn:aws:logs:us-east-1:account-id:log-group:log_group_name:log-stream:log_stream_name_prefix*",
"arn:aws:logs:us-east-1:account-id:log-group:log_group_name_prefix*:log-stream:log_stream_name_prefix*",
"arn:aws:codebuild:us-east-1:123456789012:project/my-demo-project",
"arn:aws:codebuild:us-east-1:123456789012:build/my-demo-project:7b7416ae-89b4-46cc-8236-61129df660ad",
"arn:aws:codecommit:us-east-1:123456789012:MyDemoRepo",
"arn:aws:codedeploy:us-east-1:123456789012:application:WordPress_App",
"arn:aws:codedeploy:us-east-1:123456789012:instance/AssetTag*",
"arn:aws:config:us-east-1:123456789012:config-rule/MyConfigRule",
"arn:aws:codepipeline:us-east-1:123456789012:MyDemoPipeline",
"arn:aws:directconnect:us-east-1:123456789012:dxcon/dxcon-fgase048",
"arn:aws:directconnect:us-east-1:123456789012:dxvif/dxvif-fgrb110x",
"arn:aws:dynamodb:us-east-1:123456789012:table/books_table",
"arn:aws:ecr:us-east-1:123456789012:repository/my-repository",
"arn:aws:ecs:us-east-1:123456789012:cluster/my-cluster",
"arn:aws:ecs:us-east-1:123456789012:container-instance/403125b0-555c-4473-86b5-65982db28a6d",
"arn:aws:ecs:us-east-1:123456789012:task-definition/hello_world:8",
"arn:aws:ecs:us-east-1:123456789012:service/sample-webapp",
"arn:aws:ecs:us-east-1:123456789012:task/1abf0f6d-a411-4033-b8eb-a4eed3ad252a",
"arn:aws:ecs:us-east-1:123456789012:container/476e7c41-17f2-4c17-9d14-412566202c8a",
"arn:aws:ec2:us-east-1:123456789012:dedicated-host/h-12345678",
"arn:aws:ec2:us-east-1::image/ami-1a2b3c4d",
"arn:aws:ec2:us-east-1:123456789012:instance/*",
"arn:aws:ec2:us-east-1:123456789012:volume/*",
"arn:aws:ec2:us-east-1:123456789012:volume/vol-1a2b3c4d",
"arn:aws:elasticbeanstalk:us-east-1:123456789012:application/My App",
"arn:aws:elasticbeanstalk:us-east-1:123456789012:applicationversion/My App/My Version",
"arn:aws:elasticbeanstalk:us-east-1:123456789012:environment/My App/MyEnvironment",
"arn:aws:elasticbeanstalk:us-east-1::solutionstack/32bit Amazon Linux running Tomcat 7",
"arn:aws:elasticbeanstalk:us-east-1:123456789012:configurationtemplate/My App/My Template",
"arn:aws:elasticfilesystem:us-east-1:123456789012:file-system-id/fs12345678",
"arn:aws:elasticloadbalancing:us-east-1:123456789012:loadbalancer/app/my-load-balancer/50dc6c495c0c9188",
"arn:aws:elasticloadbalancing:us-east-1:123456789012:listener/app/my-load-balancer/50dc6c495c0c9188/f2f7dc8efc522ab2",
"arn:aws:elasticloadbalancing:us-east-1:123456789012:listener-rule/app/my-load-balancer/50dc6c495c0c9188/f2f7dc8efc522ab2/9683b2d02a6cabee",
"arn:aws:elasticloadbalancing:us-east-1:123456789012:targetgroup/my-targets/73e2d6bc24d8a067",
"arn:aws:elasticloadbalancing:us-east-1:123456789012:loadbalancer/my-load-balancer",
"arn:aws:elastictranscoder:us-east-1:123456789012:preset/*",
"arn:aws:elasticache:us-west-2:123456789012:cluster:myCluster",
"arn:aws:elasticache:us-west-2:123456789012:snapshot:mySnapshot",
"arn:aws:es:us-east-1:123456789012:domain/streaming-logs",
"arn:aws:glacier:us-east-1:123456789012:vaults/examplevault",
"arn:aws:glacier:us-east-1:123456789012:vaults/example*",
"arn:aws:glacier:us-east-1:123456789012:vaults/*",
"arn:aws:health:us-east-1::event/AWS_EC2_EXAMPLE_ID",
"arn:aws:health:us-east-1:123456789012:entity/AVh5GGT7ul1arKr1sE1K",
"arn:aws:iam::123456789012:root",
"arn:aws:iam::123456789012:user/Bob",
"arn:aws:iam::123456789012:user/division_abc/subdivision_xyz/Bob",
"arn:aws:iam::123456789012:group/Developers",
"arn:aws:iam::123456789012:group/division_abc/subdivision_xyz/product_A/Developers",
"arn:aws:iam::123456789012:role/S3Access",
"arn:aws:iam::123456789012:role/application_abc/component_xyz/S3Access",
"arn:aws:iam::123456789012:policy/UsersManageOwnCredentials",
"arn:aws:iam::123456789012:policy/division_abc/subdivision_xyz/UsersManageOwnCredentials",
"arn:aws:iam::123456789012:instance-profile/Webserver",
"arn:aws:sts::123456789012:federated-user/Bob",
"arn:aws:sts::123456789012:assumed-role/Accounting-Role/Mary",
"arn:aws:iam::123456789012:mfa/BobJonesMFA",
"arn:aws:iam::123456789012:server-certificate/ProdServerCert",
"arn:aws:iam::123456789012:server-certificate/division_abc/subdivision_xyz/ProdServerCert",
"arn:aws:iam::123456789012:saml-provider/ADFSProvider",
"arn:aws:iam::123456789012:oidc-provider/GoogleProvider",
"arn:aws:iot:your-region:123456789012:cert/123a456b789c123d456e789f123a456b789c123d456e789f123a456b789c123c456d7",
"arn:aws:iot::123456789012:policy/MyIoTPolicy",
"arn:aws:iot:your-region:123456789012:rule/MyIoTRule",
"arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012",
"arn:aws:kms:us-east-1:123456789012:alias/example-alias",
"arn:aws:firehose:us-east-1:123456789012:deliverystream/example-stream-name",
"arn:aws:kinesis:us-east-1:123456789012:stream/example-stream-name",
"arn:aws:lambda:us-east-1:123456789012:function:ProcessKinesisRecords",
"arn:aws:lambda:us-east-1:123456789012:function:ProcessKinesisRecords:your alias",
"arn:aws:lambda:us-east-1:123456789012:function:ProcessKinesisRecords:1.0",
"arn:aws:lambda:us-east-1:123456789012:event-source-mappings:kinesis-stream-arn",
"arn:aws:machinelearning:us-east-1:123456789012:datasource/my-datasource-1",
"arn:aws:machinelearning:us-east-1:123456789012:mlmodel/my-mlmodel",
"arn:aws:machinelearning:us-east-1:123456789012:batchprediction/my-batchprediction",
"arn:aws:machinelearning:us-east-1:123456789012:evaluation/my-evaluation",
"arn:aws:polly:us-east-1:123456789012:lexicon/myLexicon",
"arn:aws:redshift:us-east-1:123456789012:cluster:my-cluster",
"arn:aws:redshift:us-east-1:123456789012:my-cluster/my-dbuser-name",
"arn:aws:redshift:us-east-1:123456789012:parametergroup:my-parameter-group",
"arn:aws:redshift:us-east-1:123456789012:securitygroup:my-public-group",
"arn:aws:redshift:us-east-1:123456789012:snapshot:my-cluster/my-snapshot20130807",
"arn:aws:redshift:us-east-1:123456789012:subnetgroup:my-subnet-10 ",
"arn:aws:rds:us-east-1:123456789012:db:mysql-db-instance1",
"arn:aws:rds:us-east-1:123456789012:snapshot:my-snapshot2",
"arn:aws:rds:us-east-1:123456789012:cluster:my-cluster1",
"arn:aws:rds:us-east-1:123456789012:cluster-snapshot:cluster1-snapshot7",
"arn:aws:rds:us-east-1:123456789012:og:mysql-option-group1",
"arn:aws:rds:us-east-1:123456789012:pg:mysql-repl-pg1",
"arn:aws:rds:us-east-1:123456789012:cluster-pg:aurora-pg3",
"arn:aws:rds:us-east-1:123456789012:secgrp:dev-secgrp2",
"arn:aws:rds:us-east-1:123456789012:subgrp:prod-subgrp1",
"arn:aws:rds:us-east-1:123456789012:es:monitor-events2",
"arn:aws:route53:::hostedzone/Z148QEXAMPLE8V",
"arn:aws:route53:::change/C2RDJ5EXAMPLE2",
"arn:aws:route53:::change/*",
"arn:aws:ssm:us-east-1:123456789012:document/highAvailabilityServerSetup",
"arn:aws:sns:*:123456789012:my_corporate_topic",
"arn:aws:sns:us-east-1:123456789012:my_corporate_topic:02034b43-fefa-4e07-a5eb-3be56f8c54ce",
"arn:aws:sqs:us-east-1:123456789012:queue1",
"arn:aws:s3:::my_corporate_bucket",
"arn:aws:s3:::my_corporate_bucket/exampleobject.png",
"arn:aws:s3:::my_corporate_bucket/*",
"arn:aws:s3:::my_corporate_bucket/Development/*",
"arn:aws:swf:us-east-1:123456789012:/domain/department1",
"arn:aws:swf:*:123456789012:/domain/*",
"arn:aws:states:us-east-1:123456789012:activity:HelloActivity",
"arn:aws:states:us-east-1:123456789012:stateMachine:HelloStateMachine",
"arn:aws:states:us-east-1:123456789012:execution:HelloStateMachine:HelloStateMachineExecution",
"arn:aws:storagegateway:us-east-1:123456789012:gateway/sgw-12A3456B",
"arn:aws:storagegateway:us-east-1:123456789012:gateway/sgw-12A3456B/volume/vol-1122AABB",
"arn:aws:storagegateway:us-east-1:123456789012:tape/AMZNC8A26D",
"arn:aws:storagegateway:us-east-1:123456789012:gateway/sgw-12A3456B/target/iqn.1997-05.com.amazon:vol-1122AABB",
"arn:aws:storagegateway:us-east-1:123456789012:gateway/sgw-12A3456B/device/AMZN_SGW-FF22CCDD_TAPEDRIVE_00010",
"arn:aws:trustedadvisor:*:123456789012:checks/fault_tolerance/BueAdJ7NrP",
"arn:aws:waf::123456789012:rule/41b5b052-1e4a-426b-8149-3595be6342c2",
"arn:aws:waf::123456789012:webacl/3bffd3ed-fa2e-445e-869f-a6a7cf153fd3",
"arn:aws:waf::123456789012:ipset/3f74bd8c-f046-4970-a1a7-41aa52e05480",
"arn:aws:waf::123456789012:bytematchset/d131bc0b-57be-4536-af1d-4894fd28acc4",
"arn:aws:waf::123456789012:sqlinjectionset/2be79d6f-2f41-4c9b-8192-d719676873f0",
"arn:aws:waf::123456789012:changetoken/03ba2197-fc98-4ac0-a67d-5b839762b16b",
"arn:aws:iam::123456789012:user/Development/product_1234/*",
"arn:aws:s3:::my_corporate_bucket/*",
"arn:aws:s3:::my_corporate_bucket/Development/*",
]
# For now we are only testing that the ARN library does not raise any exception with the above list of ARNs.
for arn in test_cases:
_ = ARN(arn)
def test_iam_credentials_arn_parsing(self):
test_cases = [
{
"input_arn": f"arn:aws:iam::{ACCOUNT_ID}:{RESOURCE_TYPE}/{IAM_ROLE}",
"input_arn": f"arn:aws:{IAM_SERVICE}::{ACCOUNT_ID}:{RESOURCE_TYPE_ROLE}/{IAM_ROLE}",
"expected": {
"partition": "aws",
"service": "iam",
"partition": COMMERCIAL_PARTITION,
"service": IAM_SERVICE,
"region": None,
"account_id": ACCOUNT_ID,
"resource_type": RESOURCE_TYPE,
"resource_type": RESOURCE_TYPE_ROLE,
"resource": IAM_ROLE,
},
}
},
{
"input_arn": f"arn:aws:{IAM_SERVICE}::{ACCOUNT_ID}:{RESOUCE_TYPE_USER}/{IAM_ROLE}",
"expected": {
"partition": COMMERCIAL_PARTITION,
"service": IAM_SERVICE,
"region": None,
"account_id": ACCOUNT_ID,
"resource_type": RESOUCE_TYPE_USER,
"resource": IAM_ROLE,
},
},
{
"input_arn": f"arn:{CHINA_PARTITION}:{IAM_SERVICE}::{ACCOUNT_ID}:{RESOURCE_TYPE_ROLE}/{IAM_ROLE}",
"expected": {
"partition": CHINA_PARTITION,
"service": IAM_SERVICE,
"region": None,
"account_id": ACCOUNT_ID,
"resource_type": RESOURCE_TYPE_ROLE,
"resource": IAM_ROLE,
},
},
{
"input_arn": f"arn:{CHINA_PARTITION}:{IAM_SERVICE}::{ACCOUNT_ID}:{RESOUCE_TYPE_USER}/{IAM_ROLE}",
"expected": {
"partition": CHINA_PARTITION,
"service": IAM_SERVICE,
"region": None,
"account_id": ACCOUNT_ID,
"resource_type": RESOUCE_TYPE_USER,
"resource": IAM_ROLE,
},
},
{
"input_arn": f"arn:{GOVCLOUD_PARTITION}:{IAM_SERVICE}::{ACCOUNT_ID}:{RESOURCE_TYPE_ROLE}/{IAM_ROLE}",
"expected": {
"partition": GOVCLOUD_PARTITION,
"service": IAM_SERVICE,
"region": None,
"account_id": ACCOUNT_ID,
"resource_type": RESOURCE_TYPE_ROLE,
"resource": IAM_ROLE,
},
},
{
"input_arn": f"arn:{GOVCLOUD_PARTITION}:{IAM_SERVICE}::{ACCOUNT_ID}:{RESOUCE_TYPE_USER}/{IAM_ROLE}",
"expected": {
"partition": GOVCLOUD_PARTITION,
"service": IAM_SERVICE,
"region": None,
"account_id": ACCOUNT_ID,
"resource_type": RESOUCE_TYPE_USER,
"resource": IAM_ROLE,
},
},
]
for test in test_cases:
input_arn = test["input_arn"]
parsed_arn = arn_parsing(input_arn)
parsed_arn = parse_iam_credentials_arn(input_arn)
parsed_arn.partition.should.equal(test["expected"]["partition"])
parsed_arn.service.should.equal(test["expected"]["service"])
parsed_arn.region.should.equal(test["expected"]["region"])
@@ -32,6 +257,59 @@ class Test_ARN_Parsing:
parsed_arn.resource_type.should.equal(test["expected"]["resource_type"])
parsed_arn.resource.should.equal(test["expected"]["resource"])
def test_iam_credentials_arn_parsing_raising_RoleArnParsingFailedMissingFields(
self,
):
input_arn = ""
with raises(RoleArnParsingFailedMissingFields) as error:
parse_iam_credentials_arn(input_arn)
assert error._excinfo[0] == RoleArnParsingFailedMissingFields
def test_iam_credentials_arn_parsing_raising_RoleArnParsingIAMRegionNotEmpty(self):
input_arn = "arn:aws:iam:eu-west-1:111111111111:user/prowler"
with raises(RoleArnParsingIAMRegionNotEmpty) as error:
parse_iam_credentials_arn(input_arn)
assert error._excinfo[0] == RoleArnParsingIAMRegionNotEmpty
def test_iam_credentials_arn_parsing_raising_RoleArnParsingPartitionEmpty(self):
input_arn = "arn::iam::111111111111:user/prowler"
with raises(RoleArnParsingPartitionEmpty) as error:
parse_iam_credentials_arn(input_arn)
assert error._excinfo[0] == RoleArnParsingPartitionEmpty
def test_iam_credentials_arn_parsing_raising_RoleArnParsingServiceNotIAM(self):
input_arn = "arn:aws:s3::111111111111:user/prowler"
with raises(RoleArnParsingServiceNotIAMnorSTS) as error:
parse_iam_credentials_arn(input_arn)
assert error._excinfo[0] == RoleArnParsingServiceNotIAMnorSTS
def test_iam_credentials_arn_parsing_raising_RoleArnParsingInvalidAccountID(self):
input_arn = "arn:aws:iam::AWS_ACCOUNT_ID:user/prowler"
with raises(RoleArnParsingInvalidAccountID) as error:
parse_iam_credentials_arn(input_arn)
assert error._excinfo[0] == RoleArnParsingInvalidAccountID
def test_iam_credentials_arn_parsing_raising_RoleArnParsingInvalidResourceType(
self,
):
input_arn = "arn:aws:iam::111111111111:account/prowler"
with raises(RoleArnParsingInvalidResourceType) as error:
parse_iam_credentials_arn(input_arn)
assert error._excinfo[0] == RoleArnParsingInvalidResourceType
def test_iam_credentials_arn_parsing_raising_RoleArnParsingEmptyResource(self):
input_arn = "arn:aws:iam::111111111111:role/"
with raises(RoleArnParsingEmptyResource) as error:
parse_iam_credentials_arn(input_arn)
assert error._excinfo[0] == RoleArnParsingEmptyResource
def test_is_valid_arn(self):
assert is_valid_arn("arn:aws:iam::012345678910:user/test")
assert is_valid_arn("arn:aws-cn:ec2:us-east-1:123456789012:vpc/vpc-12345678")

312
tests/providers/aws/lib/credentials/credentials_test.py Normal file
View File

@@ -0,0 +1,312 @@
import re
import boto3
import botocore
from mock import patch
from moto import mock_iam, mock_sts
from prowler.providers.aws.lib.arn.arn import parse_iam_credentials_arn
from prowler.providers.aws.lib.credentials.credentials import validate_aws_credentials
AWS_ACCOUNT_NUMBER = "123456789012"
# Mocking GetCallerIdentity for China and GovCloud
make_api_call = botocore.client.BaseClient._make_api_call
def mock_get_caller_identity_china(self, operation_name, kwarg):
if operation_name == "GetCallerIdentity":
return {
"UserId": "XXXXXXXXXXXXXXXXXXXXX",
"Account": AWS_ACCOUNT_NUMBER,
"Arn": f"arn:aws-cn:iam::{AWS_ACCOUNT_NUMBER}:user/test-user",
}
return make_api_call(self, operation_name, kwarg)
def mock_get_caller_identity_gov_cloud(self, operation_name, kwarg):
if operation_name == "GetCallerIdentity":
return {
"UserId": "XXXXXXXXXXXXXXXXXXXXX",
"Account": AWS_ACCOUNT_NUMBER,
"Arn": f"arn:aws-us-gov:iam::{AWS_ACCOUNT_NUMBER}:user/test-user",
}
return make_api_call(self, operation_name, kwarg)
class Test_AWS_Credentials:
@mock_sts
@mock_iam
def test_validate_credentials_commercial_partition_with_regions(self):
# AWS Region for AWS COMMERCIAL
aws_region = "eu-west-1"
aws_partition = "aws"
# Create a mock IAM user
iam_client = boto3.client("iam", region_name=aws_region)
iam_user = iam_client.create_user(UserName="test-user")["User"]
# Create a mock IAM access keys
access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[
"AccessKey"
]
access_key_id = access_key["AccessKeyId"]
secret_access_key = access_key["SecretAccessKey"]
# Create AWS session to validate
session = boto3.session.Session(
aws_access_key_id=access_key_id,
aws_secret_access_key=secret_access_key,
region_name=aws_region,
)
get_caller_identity = validate_aws_credentials(session, [aws_region])
assert get_caller_identity["region"] == aws_region
caller_identity_arn = parse_iam_credentials_arn(get_caller_identity["Arn"])
assert caller_identity_arn.partition == aws_partition
assert caller_identity_arn.region is None
assert caller_identity_arn.resource == "test-user"
assert caller_identity_arn.resource_type == "user"
assert re.match("[0-9a-zA-Z]{20}", get_caller_identity["UserId"])
assert get_caller_identity["Account"] == AWS_ACCOUNT_NUMBER
@mock_sts
@mock_iam
def test_validate_credentials_commercial_partition_with_regions_none_and_profile_region_so_profile_region(
self,
):
# AWS Region for AWS COMMERCIAL
aws_region = "eu-west-1"
aws_partition = "aws"
# Create a mock IAM user
iam_client = boto3.client("iam", region_name=aws_region)
iam_user = iam_client.create_user(UserName="test-user")["User"]
# Create a mock IAM access keys
access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[
"AccessKey"
]
access_key_id = access_key["AccessKeyId"]
secret_access_key = access_key["SecretAccessKey"]
# Create AWS session to validate
session = boto3.session.Session(
aws_access_key_id=access_key_id,
aws_secret_access_key=secret_access_key,
region_name=aws_region,
)
get_caller_identity = validate_aws_credentials(session, None)
assert get_caller_identity["region"] == aws_region
caller_identity_arn = parse_iam_credentials_arn(get_caller_identity["Arn"])
assert caller_identity_arn.partition == aws_partition
assert caller_identity_arn.region is None
assert caller_identity_arn.resource == "test-user"
assert caller_identity_arn.resource_type == "user"
assert re.match("[0-9a-zA-Z]{20}", get_caller_identity["UserId"])
assert get_caller_identity["Account"] == AWS_ACCOUNT_NUMBER
@mock_sts
@mock_iam
def test_validate_credentials_commercial_partition_with_0_regions_and_profile_region_so_profile_region(
self,
):
# AWS Region for AWS COMMERCIAL
aws_region = "eu-west-1"
aws_partition = "aws"
# Create a mock IAM user
iam_client = boto3.client("iam", region_name=aws_region)
iam_user = iam_client.create_user(UserName="test-user")["User"]
# Create a mock IAM access keys
access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[
"AccessKey"
]
access_key_id = access_key["AccessKeyId"]
secret_access_key = access_key["SecretAccessKey"]
# Create AWS session to validate
session = boto3.session.Session(
aws_access_key_id=access_key_id,
aws_secret_access_key=secret_access_key,
region_name=aws_region,
)
get_caller_identity = validate_aws_credentials(session, [])
assert get_caller_identity["region"] == aws_region
caller_identity_arn = parse_iam_credentials_arn(get_caller_identity["Arn"])
assert caller_identity_arn.partition == aws_partition
assert caller_identity_arn.region is None
assert caller_identity_arn.resource == "test-user"
assert caller_identity_arn.resource_type == "user"
assert re.match("[0-9a-zA-Z]{20}", get_caller_identity["UserId"])
assert get_caller_identity["Account"] == AWS_ACCOUNT_NUMBER
@mock_sts
@mock_iam
def test_validate_credentials_commercial_partition_without_regions_and_profile_region_so_us_east_1(
self,
):
# AWS Region for AWS COMMERCIAL
aws_region = "eu-west-1"
aws_partition = "aws"
# Create a mock IAM user
iam_client = boto3.client("iam", region_name=aws_region)
iam_user = iam_client.create_user(UserName="test-user")["User"]
# Create a mock IAM access keys
access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[
"AccessKey"
]
access_key_id = access_key["AccessKeyId"]
secret_access_key = access_key["SecretAccessKey"]
# Create AWS session to validate
session = boto3.session.Session(
aws_access_key_id=access_key_id,
aws_secret_access_key=secret_access_key,
region_name=None,
)
get_caller_identity = validate_aws_credentials(session, [])
assert get_caller_identity["region"] == "us-east-1"
caller_identity_arn = parse_iam_credentials_arn(get_caller_identity["Arn"])
assert caller_identity_arn.partition == aws_partition
assert caller_identity_arn.region is None
assert caller_identity_arn.resource == "test-user"
assert caller_identity_arn.resource_type == "user"
assert re.match("[0-9a-zA-Z]{20}", get_caller_identity["UserId"])
assert get_caller_identity["Account"] == AWS_ACCOUNT_NUMBER
@mock_sts
@mock_iam
def test_validate_credentials_china_partition_without_regions_and_profile_region_so_us_east_1(
self,
):
# AWS Region for AWS COMMERCIAL
aws_region = "eu-west-1"
aws_partition = "aws"
# Create a mock IAM user
iam_client = boto3.client("iam", region_name=aws_region)
iam_user = iam_client.create_user(UserName="test-user")["User"]
# Create a mock IAM access keys
access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[
"AccessKey"
]
access_key_id = access_key["AccessKeyId"]
secret_access_key = access_key["SecretAccessKey"]
# Create AWS session to validate
session = boto3.session.Session(
aws_access_key_id=access_key_id,
aws_secret_access_key=secret_access_key,
region_name=None,
)
get_caller_identity = validate_aws_credentials(session, [])
assert get_caller_identity["region"] == "us-east-1"
caller_identity_arn = parse_iam_credentials_arn(get_caller_identity["Arn"])
assert caller_identity_arn.partition == aws_partition
assert caller_identity_arn.region is None
assert caller_identity_arn.resource == "test-user"
assert caller_identity_arn.resource_type == "user"
assert re.match("[0-9a-zA-Z]{20}", get_caller_identity["UserId"])
assert get_caller_identity["Account"] == AWS_ACCOUNT_NUMBER
@mock_sts
@mock_iam
@patch(
"botocore.client.BaseClient._make_api_call", new=mock_get_caller_identity_china
)
def test_validate_credentials_china_partition(self):
# AWS Region for AWS CHINA
aws_region = "cn-north-1"
aws_partition = "aws-cn"
# Create a mock IAM user
iam_client = boto3.client("iam", region_name=aws_region)
iam_user = iam_client.create_user(UserName="test-user")["User"]
# Create a mock IAM access keys
access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[
"AccessKey"
]
access_key_id = access_key["AccessKeyId"]
secret_access_key = access_key["SecretAccessKey"]
# Create AWS session to validate
session = boto3.session.Session(
aws_access_key_id=access_key_id,
aws_secret_access_key=secret_access_key,
region_name=aws_region,
)
get_caller_identity = validate_aws_credentials(session, [aws_region])
# To use GovCloud or China it is either required:
# - Set the AWS profile region with a valid partition region
# - Use the -f/--region with a valid partition region
assert get_caller_identity["region"] == aws_region
caller_identity_arn = parse_iam_credentials_arn(get_caller_identity["Arn"])
assert caller_identity_arn.partition == aws_partition
assert caller_identity_arn.region is None
assert caller_identity_arn.resource == "test-user"
assert caller_identity_arn.resource_type == "user"
assert re.match("[0-9a-zA-Z]{20}", get_caller_identity["UserId"])
assert get_caller_identity["Account"] == AWS_ACCOUNT_NUMBER
@mock_sts
@mock_iam
@patch(
"botocore.client.BaseClient._make_api_call",
new=mock_get_caller_identity_gov_cloud,
)
def test_validate_credentials_gov_cloud_partition(self):
# AWS Region for US GOV CLOUD
aws_region = "us-gov-east-1"
aws_partition = "aws-us-gov"
# Create a mock IAM user
iam_client = boto3.client("iam", region_name=aws_region)
iam_user = iam_client.create_user(UserName="test-user")["User"]
# Create a mock IAM access keys
access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[
"AccessKey"
]
access_key_id = access_key["AccessKeyId"]
secret_access_key = access_key["SecretAccessKey"]
# Create AWS session to validate
session = boto3.session.Session(
aws_access_key_id=access_key_id,
aws_secret_access_key=secret_access_key,
region_name=aws_region,
)
get_caller_identity = validate_aws_credentials(session, [aws_region])
# To use GovCloud or China it is either required:
# - Set the AWS profile region with a valid partition region
# - Use the -f/--region with a valid partition region
assert get_caller_identity["region"] == aws_region
caller_identity_arn = parse_iam_credentials_arn(get_caller_identity["Arn"])
assert caller_identity_arn.partition == aws_partition
assert caller_identity_arn.region is None
assert caller_identity_arn.resource == "test-user"
assert caller_identity_arn.resource_type == "user"
assert re.match("[0-9a-zA-Z]{20}", get_caller_identity["UserId"])
assert get_caller_identity["Account"] == AWS_ACCOUNT_NUMBER

61
tests/providers/aws/lib/organizations/organizations_test.py Normal file
View File

@@ -0,0 +1,61 @@
import json
import boto3
import sure # noqa
from moto import mock_iam, mock_organizations, mock_sts
from prowler.providers.aws.lib.organizations.organizations import (
get_organizations_metadata,
)
AWS_ACCOUNT_NUMBER = "123456789012"
class Test_AWS_Organizations:
@mock_organizations
@mock_sts
@mock_iam
def test_organizations(self):
client = boto3.client("organizations", region_name="us-east-1")
iam_client = boto3.client("iam", region_name="us-east-1")
sts_client = boto3.client("sts", region_name="us-east-1")
mockname = "mock-account"
mockdomain = "moto-example.org"
mockemail = "@".join([mockname, mockdomain])
org_id = client.create_organization(FeatureSet="ALL")["Organization"]["Id"]
account_id = client.create_account(AccountName=mockname, Email=mockemail)[
"CreateAccountStatus"
]["AccountId"]
client.tag_resource(
ResourceId=account_id, Tags=[{"Key": "key", "Value": "value"}]
)
trust_policy_document = {
"Version": "2012-10-17",
"Statement": {
"Effect": "Allow",
"Principal": {"AWS": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"},
"Action": "sts:AssumeRole",
},
}
iam_role_arn = iam_client.role_arn = iam_client.create_role(
RoleName="test-role",
AssumeRolePolicyDocument=json.dumps(trust_policy_document),
)["Role"]["Arn"]
session_name = "new-session"
assumed_role = sts_client.assume_role(
RoleArn=iam_role_arn, RoleSessionName=session_name
)
org = get_organizations_metadata(account_id, assumed_role)
org.account_details_email.should.equal(mockemail)
org.account_details_name.should.equal(mockname)
org.account_details_arn.should.equal(
f"arn:aws:organizations::{AWS_ACCOUNT_NUMBER}:account/{org_id}/{account_id}"
)
org.account_details_org.should.equal(org_id)
org.account_details_tags.should.equal("key:value,")

290
tests/providers/common/audit_info_test.py
View File

@@ -1,15 +1,9 @@
import json
import boto3
import botocore
import sure # noqa
from boto3 import session
from mock import patch
from moto import (
mock_ec2,
mock_iam,
mock_organizations,
mock_resourcegroupstaggingapi,
mock_sts,
)
from moto import mock_ec2, mock_resourcegroupstaggingapi
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
from prowler.providers.azure.azure_provider import Azure_Provider
@@ -26,24 +20,8 @@ from prowler.providers.gcp.gcp_provider import GCP_Provider
from prowler.providers.gcp.lib.audit_info.models import GCP_Audit_Info
EXAMPLE_AMI_ID = "ami-12c6146b"
ACCOUNT_ID = 123456789012
mock_current_audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=None,
audited_account="123456789012",
audited_identity_arn="arn:aws:iam::123456789012:user/test",
audited_user_id="test",
audited_partition="aws",
profile="default",
profile_region="eu-west-1",
credentials=None,
assumed_role_info=None,
audited_regions=["eu-west-2", "eu-west-1"],
organizations_metadata=None,
audit_resources=None,
audit_metadata=None,
)
AWS_ACCOUNT_NUMBER = "123456789012"
mock_azure_audit_info = Azure_Audit_Info(
credentials=None,
@@ -54,6 +32,31 @@ mock_azure_audit_info = Azure_Audit_Info(
mock_set_audit_info = Audit_Info()
# Mocking GetCallerIdentity for China and GovCloud
make_api_call = botocore.client.BaseClient._make_api_call
def mock_get_caller_identity_china(self, operation_name, kwarg):
if operation_name == "GetCallerIdentity":
return {
"UserId": "XXXXXXXXXXXXXXXXXXXXX",
"Account": AWS_ACCOUNT_NUMBER,
"Arn": f"arn:aws-cn:iam::{AWS_ACCOUNT_NUMBER}:user/test-user",
}
return make_api_call(self, operation_name, kwarg)
def mock_get_caller_identity_gov_cloud(self, operation_name, kwarg):
if operation_name == "GetCallerIdentity":
return {
"UserId": "XXXXXXXXXXXXXXXXXXXXX",
"Account": AWS_ACCOUNT_NUMBER,
"Arn": f"arn:aws-us-gov:iam::{AWS_ACCOUNT_NUMBER}:user/test-user",
}
return make_api_call(self, operation_name, kwarg)
def mock_validate_credentials(*_):
caller_identity = {
@@ -81,118 +84,60 @@ def mock_set_gcp_credentials(*_):
class Test_Set_Audit_Info:
@patch(
"prowler.providers.common.audit_info.current_audit_info",
new=mock_current_audit_info,
)
@mock_sts
@mock_iam
def test_validate_credentials(self):
# Create a mock IAM user
iam_client = boto3.client("iam", region_name="us-east-1")
iam_user = iam_client.create_user(UserName="test-user")["User"]
# Create a mock IAM access keys
access_key = iam_client.create_access_key(UserName=iam_user["UserName"])[
"AccessKey"
]
access_key_id = access_key["AccessKeyId"]
secret_access_key = access_key["SecretAccessKey"]
# Create AWS session to validate
session = boto3.session.Session(
aws_access_key_id=access_key_id,
aws_secret_access_key=secret_access_key,
region_name="us-east-1",
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
),
audited_account=AWS_ACCOUNT_NUMBER,
audited_user_id=None,
audited_partition="aws",
audited_identity_arn="arn:aws:iam::123456789012:user/test",
profile=None,
profile_region="eu-west-1",
credentials=None,
assumed_role_info=None,
audited_regions=["eu-west-2", "eu-west-1"],
organizations_metadata=None,
audit_resources=None,
)
audit_info = Audit_Info()
get_caller_identity = audit_info.validate_credentials(session)
get_caller_identity["Arn"].should.equal(iam_user["Arn"])
get_caller_identity["UserId"].should.equal(iam_user["UserId"])
# assert get_caller_identity["UserId"] == str(ACCOUNT_ID)
return audit_info
@patch(
"prowler.providers.common.audit_info.current_audit_info",
new=mock_current_audit_info,
"prowler.providers.common.audit_info.validate_aws_credentials",
new=mock_validate_credentials,
)
@mock_organizations
@mock_sts
@mock_iam
def test_organizations(self):
client = boto3.client("organizations", region_name="us-east-1")
iam_client = boto3.client("iam", region_name="us-east-1")
sts_client = boto3.client("sts", region_name="us-east-1")
mockname = "mock-account"
mockdomain = "moto-example.org"
mockemail = "@".join([mockname, mockdomain])
org_id = client.create_organization(FeatureSet="ALL")["Organization"]["Id"]
account_id = client.create_account(AccountName=mockname, Email=mockemail)[
"CreateAccountStatus"
]["AccountId"]
client.tag_resource(
ResourceId=account_id, Tags=[{"Key": "key", "Value": "value"}]
)
trust_policy_document = {
"Version": "2012-10-17",
"Statement": {
"Effect": "Allow",
"Principal": {
"AWS": "arn:aws:iam::{account_id}:root".format(
account_id=ACCOUNT_ID
)
},
"Action": "sts:AssumeRole",
},
}
iam_role_arn = iam_client.role_arn = iam_client.create_role(
RoleName="test-role",
AssumeRolePolicyDocument=json.dumps(trust_policy_document),
)["Role"]["Arn"]
session_name = "new-session"
assumed_role = sts_client.assume_role(
RoleArn=iam_role_arn, RoleSessionName=session_name
)
audit_info = Audit_Info()
org = audit_info.get_organizations_metadata(account_id, assumed_role)
org.account_details_email.should.equal(mockemail)
org.account_details_name.should.equal(mockname)
org.account_details_arn.should.equal(
"arn:aws:organizations::{0}:account/{1}/{2}".format(
ACCOUNT_ID, org_id, account_id
)
)
org.account_details_org.should.equal(org_id)
org.account_details_tags.should.equal("key:value,")
@patch(
"prowler.providers.common.audit_info.current_audit_info",
new=mock_current_audit_info,
"prowler.providers.common.audit_info.print_aws_credentials",
new=mock_print_audit_credentials,
)
@patch.object(Audit_Info, "validate_credentials", new=mock_validate_credentials)
@patch.object(Audit_Info, "print_aws_credentials", new=mock_print_audit_credentials)
def test_set_audit_info_aws(self):
provider = "aws"
arguments = {
"profile": None,
"role": None,
"session_duration": None,
"external_id": None,
"regions": None,
"organizations_role": None,
"subscriptions": None,
"az_cli_auth": None,
"sp_env_auth": None,
"browser_auth": None,
"managed_entity_auth": None,
}
with patch(
"prowler.providers.common.audit_info.current_audit_info",
new=self.set_mocked_audit_info(),
):
provider = "aws"
arguments = {
"profile": None,
"role": None,
"session_duration": None,
"external_id": None,
"regions": None,
"organizations_role": None,
"subscriptions": None,
"az_cli_auth": None,
"sp_env_auth": None,
"browser_auth": None,
"managed_entity_auth": None,
}
audit_info = set_provider_audit_info(provider, arguments)
assert isinstance(audit_info, AWS_Audit_Info)
audit_info = set_provider_audit_info(provider, arguments)
assert isinstance(audit_info, AWS_Audit_Info)
@patch(
"prowler.providers.common.audit_info.azure_audit_info",
@@ -242,45 +187,48 @@ class Test_Set_Audit_Info:
@mock_resourcegroupstaggingapi
@mock_ec2
def test_get_tagged_resources(self):
client = boto3.client("ec2", region_name="eu-central-1")
instances = client.run_instances(
ImageId=EXAMPLE_AMI_ID,
MinCount=1,
MaxCount=1,
InstanceType="t2.micro",
TagSpecifications=[
{
"ResourceType": "instance",
"Tags": [
{"Key": "MY_TAG1", "Value": "MY_VALUE1"},
{"Key": "MY_TAG2", "Value": "MY_VALUE2"},
],
},
{
"ResourceType": "instance",
"Tags": [{"Key": "ami", "Value": "test"}],
},
],
)
instance_id = instances["Instances"][0]["InstanceId"]
image_id = client.create_image(Name="testami", InstanceId=instance_id)[
"ImageId"
]
client.create_tags(Resources=[image_id], Tags=[{"Key": "ami", "Value": "test"}])
with patch(
"prowler.providers.common.audit_info.current_audit_info",
new=self.set_mocked_audit_info(),
) as mock_audit_info:
client = boto3.client("ec2", region_name="eu-central-1")
instances = client.run_instances(
ImageId=EXAMPLE_AMI_ID,
MinCount=1,
MaxCount=1,
InstanceType="t2.micro",
TagSpecifications=[
{
"ResourceType": "instance",
"Tags": [
{"Key": "MY_TAG1", "Value": "MY_VALUE1"},
{"Key": "MY_TAG2", "Value": "MY_VALUE2"},
],
},
{
"ResourceType": "instance",
"Tags": [{"Key": "ami", "Value": "test"}],
},
],
)
instance_id = instances["Instances"][0]["InstanceId"]
image_id = client.create_image(Name="testami", InstanceId=instance_id)[
"ImageId"
]
client.create_tags(
Resources=[image_id], Tags=[{"Key": "ami", "Value": "test"}]
)
mock_current_audit_info.audited_regions = ["eu-central-1"]
mock_current_audit_info.audit_session = boto3.session.Session()
assert len(get_tagged_resources(["ami=test"], mock_current_audit_info)) == 2
assert image_id in str(
get_tagged_resources(["ami=test"], mock_current_audit_info)
)
assert instance_id in str(
get_tagged_resources(["ami=test"], mock_current_audit_info)
)
assert (
len(get_tagged_resources(["MY_TAG1=MY_VALUE1"], mock_current_audit_info))
== 1
)
assert instance_id in str(
get_tagged_resources(["MY_TAG1=MY_VALUE1"], mock_current_audit_info)
)
mock_audit_info.audited_regions = ["eu-central-1"]
mock_audit_info.audit_session = boto3.session.Session()
assert len(get_tagged_resources(["ami=test"], mock_audit_info)) == 2
assert image_id in str(get_tagged_resources(["ami=test"], mock_audit_info))
assert instance_id in str(
get_tagged_resources(["ami=test"], mock_audit_info)
)
assert (
len(get_tagged_resources(["MY_TAG1=MY_VALUE1"], mock_audit_info)) == 1
)
assert instance_id in str(
get_tagged_resources(["MY_TAG1=MY_VALUE1"], mock_audit_info)
)