使用 Moto 将 s3 存储桶模拟为 IAM 用户

Mock s3 bucket as an IAM user using Moto

模拟场景:

我正在尝试以附加了 s3 拒绝策略的 iam 用户身份访问 s3 存储桶。所以访问s3 bucket会通过Access Denied错误。但是我可以看到桶里的东西..

下面是我的代码:

@pytest.fixture()
def s3():
    with moto.mock_s3():
        yield boto3.client(
            "s3",
            region_name="us-east-1",
            aws_access_key_id="testing",
            aws_secret_access_key="testing",
            aws_session_token="testing",
        )
@pytest.fixture
def bucket_name(s3):
    bucket_name = "test_bucket"
    s3.create_bucket(Bucket=bucket_name)
    s3.put_object(Bucket=bucket_name, Key="a/b/c/abc.txt")

    return bucket_name

@pytest.fixture()
def iam():
    with moto.mock_iam():
        yield boto3.client(
            "iam",
            region_name="us-east-1",
            aws_access_key_id="testing",
            aws_secret_access_key="testing",
            aws_session_token="testing",
        )
#
#
@pytest.fixture()
def iam_user(iam, s3, bucket_name):
    user_name = "test-user"
    policy_name = "policy1"
    iam.create_user(UserName=user_name)
    policy_document = {
        "Version": "2012-10-17",
        "Statement": {"Effect": "Deny", "Action": "s3:ListBucket", "Resource": "*"}
    policy_arn = iam.create_policy(PolicyName=policy_name, PolicyDocument=json.dumps(policy_document))["Policy"][
        "Arn"
    ]
    iam.attach_user_policy(UserName=user_name, PolicyArn=policy_arn)
    access_key = iam.create_access_key(UserName=user_name)

    client = boto3.client(
        "s3",
        region_name="us-east-1",
        aws_access_key_id=access_key["AccessKey"]["AccessKeyId"],
        aws_secret_access_key=access_key["AccessKey"]["SecretAccessKey"],
    )

    print(client.list_objects(Bucket=bucket_name))
def test_check(iam, iam_user):
        print("DOne")

回应

{'ResponseMetadata': {'HTTPStatusCode': 200, 'HTTPHeaders': {}, 'RetryAttempts': 0}, 'IsTruncated': False, 'Contents': [{'Key': 'a/b/c/abc.txt', 'LastModified': datetime.datetime(2021, 3, 25, 20, 31, 15, tzinfo=tzutc()), 'ETag': '"abcdefghikkd"', 'Size': 0, 'StorageClass': 'STANDARD', 'Owner': {'DisplayName': 'webfile', 'ID': 'abcdefgh'}}], 'Name': 'test_bucket', 'MaxKeys': 1000}

感谢任何帮助。 谢谢

默认情况下,moto 将允许任何操作。不过可以打开基本策略验证 - 请参阅 this section on the README

启用验证是使用 set_initial_no_auth_action_count-装饰器完成的,这实质上意味着:不验证初始 x 操作(以允许用户设置所有 IAM actions/policies),但验证之后的一切。

像这样重写示例使我成功失败:

import boto3
import json
import moto
import pytest

from moto.core import set_initial_no_auth_action_count

bucket_name = "test_bucket"


@pytest.fixture()
def s3():
    with moto.mock_s3():
        yield boto3.client(
            "s3",
            region_name="us-east-1",
            aws_access_key_id="testing",
            aws_secret_access_key="testing",
            aws_session_token="testing",
        )


@pytest.fixture
def bucket(s3):
    s3.create_bucket(Bucket=bucket_name)
    s3.put_object(Bucket=bucket_name, Key="a/b/c/abc.txt")


@pytest.fixture()
def iam():
    with moto.mock_iam():
        yield boto3.client(
            "iam",
            region_name="us-east-1",
            aws_access_key_id="testing",
            aws_secret_access_key="testing",
            aws_session_token="testing",
        )
#
#
@pytest.fixture()
def iam_user(iam, s3):
    user_name = "test-user"
    policy_name = "policy1"
    iam.create_user(UserName=user_name)
    policy_document = {
        "Version": "2012-10-17",
        "Statement": {"Effect": "Deny", "Action": "s3:ListBucket", "Resource": "*"}
    }
    policy_arn = iam.create_policy(PolicyName=policy_name, PolicyDocument=json.dumps(policy_document))["Policy"][
        "Arn"
    ]
    iam.attach_user_policy(UserName=user_name, PolicyArn=policy_arn)
    access_key = iam.create_access_key(UserName=user_name)

    yield access_key


@set_initial_no_auth_action_count(0)
def test_check(iam, iam_user, bucket):
        access_key = iam_user
        print(access_key)
        client = boto3.client(
            "s3",
            region_name="us-east-1",
            aws_access_key_id=access_key["AccessKey"]["AccessKeyId"],
            aws_secret_access_key=access_key["AccessKey"]["SecretAccessKey"],
        )

        print(client.list_objects(Bucket=bucket_name))

请注意,'no_auth_action_count' 设置为 0。首先执行固定装置,没有任何 IAM 验证。之后,装饰器仅应用于测试方法。由于我们要验证函数内的每个语句,因此计数设置为 0。