mirror
/
s3-credentials
kopia lustrzana https://github.com/simonw/s3-credentials
1
0
Forkuj 0
Kod Zgłoszenia Packages Projekty Wydania Wiki Aktywność

Updated read_only policy to better match read_write, refs #11 

Also fixed issue with custom --policy where the username and policy name
still contained read-write - they now contain custom instead. Closes #14

Found this while expanding the tests for the create command.
pull/16/head
Simon Willison 2021-11-03 12:35:34 -07:00
rodzic 0531b026d0
commit 86861a3649
3 zmienionych plików z 43 dodań i 36 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

4
s3_credentials/cli.py
Wyświetl plik

@ -173,7 +173,7 @@ def create(
if not username: if not username:
# Default username is "s3.read-write.bucket1,bucket2" # Default username is "s3.read-write.bucket1,bucket2"
username = "s3.{permission}.{buckets}".format( username = "s3.{permission}.{buckets}".format(
permission=permission, buckets=",".join(buckets) permission="custom" if policy else permission, buckets=",".join(buckets)
) )
if not user_exists(iam, username): if not user_exists(iam, username):
kwargs = {"UserName": username} kwargs = {"UserName": username}
@ -202,7 +202,7 @@ def create(
# Add inline policies to the user so they can access the buckets # Add inline policies to the user so they can access the buckets
for bucket in buckets: for bucket in buckets:
policy_name = "s3.{permission}.{bucket}".format( policy_name = "s3.{permission}.{bucket}".format(
permission=permission, permission="custom" if policy else permission,
bucket=bucket, bucket=bucket,
) )
if policy: if policy:

14
s3_credentials/policies.py
Wyświetl plik

@ -23,12 +23,14 @@ def read_only(bucket):
"Statement": [ "Statement": [
{ {
"Effect": "Allow", "Effect": "Allow",
"Action": ["s3:GetObject*", "s3:ListBucket"], "Action": ["s3:ListBucket"],
"Resource": [ "Resource": ["arn:aws:s3:::{}".format(bucket)],
"arn:aws:s3:::{}".format(bucket), },
"arn:aws:s3:::{}/*".format(bucket), {
], "Effect": "Allow",
} "Action": "s3:GetObject*",
"Resource": ["arn:aws:s3:::{}/*".format(bucket)],
},
], ],
} }

59
tests/test_s3_credentials.py
Wyświetl plik

@ -63,19 +63,25 @@ def test_list_buckets(mocker, option, expected):
CUSTOM_POLICY = '{"custom": "policy", "bucket": "$!BUCKET_NAME!$"}' CUSTOM_POLICY = '{"custom": "policy", "bucket": "$!BUCKET_NAME!$"}'
DEFAULT_POLICY = '{"Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Action": ["s3:ListBucket"], "Resource": ["arn:aws:s3:::pytest-bucket-simonw-1"]}, {"Effect": "Allow", "Action": "s3:*Object", "Resource": ["arn:aws:s3:::pytest-bucket-simonw-1/*"]}]}' READ_WRITE_POLICY = '{"Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Action": ["s3:ListBucket"], "Resource": ["arn:aws:s3:::pytest-bucket-simonw-1"]}, {"Effect": "Allow", "Action": "s3:*Object", "Resource": ["arn:aws:s3:::pytest-bucket-simonw-1/*"]}]}'
READ_ONLY_POLICY = '{"Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Action": ["s3:ListBucket"], "Resource": ["arn:aws:s3:::pytest-bucket-simonw-1"]}, {"Effect": "Allow", "Action": "s3:GetObject*", "Resource": ["arn:aws:s3:::pytest-bucket-simonw-1/*"]}]}'
WRITE_ONLY_POLICY = '{"Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Action": ["s3:PutObject"], "Resource": ["arn:aws:s3:::pytest-bucket-simonw-1/*"]}]}'
@pytest.mark.parametrize( @pytest.mark.parametrize(
"custom_policy,policy_strategy", "options,use_policy_stdin,expected_policy,expected_name_fragment",
( (
(False, None), ([], False, READ_WRITE_POLICY, "read-write"),
(True, "filepath"), (["--read-only"], False, READ_ONLY_POLICY, "read-only"),
(True, "stdin"), (["--write-only"], False, WRITE_ONLY_POLICY, "write-only"),
(True, "string"), (["--policy", "POLICYFILEPATH"], False, CUSTOM_POLICY, "custom"),
(["--policy", "-"], True, CUSTOM_POLICY, "custom"),
(["--policy", CUSTOM_POLICY], False, CUSTOM_POLICY, "custom"),
), ),
) )
def test_create(mocker, tmpdir, custom_policy, policy_strategy): def test_create(
mocker, tmpdir, options, use_policy_stdin, expected_policy, expected_name_fragment
):
boto3 = mocker.patch("boto3.client") boto3 = mocker.patch("boto3.client")
boto3.return_value = Mock() boto3.return_value = Mock()
boto3.return_value.create_access_key.return_value = { boto3.return_value.create_access_key.return_value = {
@ -84,40 +90,39 @@ def test_create(mocker, tmpdir, custom_policy, policy_strategy):
"SecretAccessKey": "secret", "SecretAccessKey": "secret",
} }
} }
expected_policy = DEFAULT_POLICY
runner = CliRunner() runner = CliRunner()
with runner.isolated_filesystem(): with runner.isolated_filesystem():
args = ["create", "pytest-bucket-simonw-1", "-c"]
kwargs = {}
if policy_strategy:
expected_policy = CUSTOM_POLICY.replace(
"$!BUCKET_NAME!$", "pytest-bucket-simonw-1"
)
if policy_strategy == "filepath":
filepath = str(tmpdir / "policy.json") filepath = str(tmpdir / "policy.json")
open(filepath, "w").write(CUSTOM_POLICY) open(filepath, "w").write(CUSTOM_POLICY)
args.extend(["--policy", filepath]) fixed_options = [
elif policy_strategy == "stdin": filepath if option == "POLICYFILEPATH" else option for option in options
]
args = ["create", "pytest-bucket-simonw-1", "-c"] + fixed_options
kwargs = {}
if use_policy_stdin:
kwargs["input"] = CUSTOM_POLICY kwargs["input"] = CUSTOM_POLICY
args.extend(["--policy", "-"])
elif policy_strategy == "string":
args.extend(["--policy", CUSTOM_POLICY])
result = runner.invoke(cli, args, **kwargs) result = runner.invoke(cli, args, **kwargs)
assert result.exit_code == 0 assert result.exit_code == 0
assert result.output == ( assert result.output == (
"Attached policy s3.read-write.pytest-bucket-simonw-1 to user s3.read-write.pytest-bucket-simonw-1\n" "Attached policy s3.NAME_FRAGMENT.pytest-bucket-simonw-1 to user s3.NAME_FRAGMENT.pytest-bucket-simonw-1\n"
"Created access key for user: s3.read-write.pytest-bucket-simonw-1\n" "Created access key for user: s3.NAME_FRAGMENT.pytest-bucket-simonw-1\n"
'{\n "AccessKeyId": "access",\n "SecretAccessKey": "secret"\n}\n' '{\n "AccessKeyId": "access",\n "SecretAccessKey": "secret"\n}\n'
) ).replace("NAME_FRAGMENT", expected_name_fragment)
assert [str(c) for c in boto3.mock_calls] == [ assert [str(c) for c in boto3.mock_calls] == [
"call('s3')", "call('s3')",
"call('iam')", "call('iam')",
"call().head_bucket(Bucket='pytest-bucket-simonw-1')", "call().head_bucket(Bucket='pytest-bucket-simonw-1')",
"call().get_user(UserName='s3.read-write.pytest-bucket-simonw-1')", "call().get_user(UserName='s3.{}.pytest-bucket-simonw-1')".format(
"call().put_user_policy(PolicyDocument='{}', PolicyName='s3.read-write.pytest-bucket-simonw-1', UserName='s3.read-write.pytest-bucket-simonw-1')".format( expected_name_fragment
expected_policy ),
"call().put_user_policy(PolicyDocument='{}', PolicyName='s3.{}.pytest-bucket-simonw-1', UserName='s3.{}.pytest-bucket-simonw-1')".format(
expected_policy.replace("$!BUCKET_NAME!$", "pytest-bucket-simonw-1"),
expected_name_fragment,
expected_name_fragment,
),
"call().create_access_key(UserName='s3.{}.pytest-bucket-simonw-1')".format(
expected_name_fragment
), ),
"call().create_access_key(UserName='s3.read-write.pytest-bucket-simonw-1')",
] ]