327 lines
12 KiB
Python
327 lines
12 KiB
Python
|
import boto3
|
||
|
|
import pytest
|
||
|
|
from botocore.exceptions import ClientError
|
||
|
|
|
||
|
|
from moto import mock_aws
|
||
|
|
|
||
|
|
|
||
|
|
@mock_aws
|
||
|
|
def test_filter_with_primary_region():
|
||
|
|
conn = boto3.client("secretsmanager", region_name="us-east-1")
|
||
|
|
|
||
|
|
conn.create_secret(Name="simple", SecretString="secret")
|
||
|
|
conn.create_secret(
|
||
|
|
Name="dup", SecretString="s", AddReplicaRegions=[{"Region": "us-east-2"}]
|
||
|
|
)
|
||
|
|
|
||
|
|
secrets = conn.list_secrets(
|
||
|
|
Filters=[{"Key": "primary-region", "Values": ["us-east-1"]}]
|
||
|
|
)["SecretList"]
|
||
|
|
assert len(secrets) == 1
|
||
|
|
|
||
|
|
secrets = conn.list_secrets(
|
||
|
|
Filters=[{"Key": "primary-region", "Values": ["us-east-2"]}]
|
||
|
|
)["SecretList"]
|
||
|
|
assert len(secrets) == 0
|
||
|
|
|
||
|
|
secrets = conn.list_secrets()["SecretList"]
|
||
|
|
assert len(secrets) == 2
|
||
|
|
|
||
|
|
# If our entry point is us-east-2, but filter in us-east-1
|
||
|
|
# We can see the replicas in us-east-2 that have Primary Secrets in us-east-1
|
||
|
|
use2 = boto3.client("secretsmanager", region_name="us-east-2")
|
||
|
|
secrets = use2.list_secrets(
|
||
|
|
Filters=[{"Key": "primary-region", "Values": ["us-east-1"]}]
|
||
|
|
)["SecretList"]
|
||
|
|
assert len(secrets) == 1
|
||
|
|
assert secrets[0]["PrimaryRegion"] == "us-east-1"
|
||
|
|
|
||
|
|
|
||
|
|
@mock_aws
|
||
|
|
def test_create_secret_that_duplicates():
|
||
|
|
use1 = boto3.client("secretsmanager", region_name="us-east-1")
|
||
|
|
use2 = boto3.client("secretsmanager", region_name="us-east-2")
|
||
|
|
|
||
|
|
resp = use1.create_secret(
|
||
|
|
Name="dup", SecretString="s", AddReplicaRegions=[{"Region": "us-east-2"}]
|
||
|
|
)
|
||
|
|
primary_arn = resp["ARN"]
|
||
|
|
assert resp["ReplicationStatus"][0]["Region"] == "us-east-2"
|
||
|
|
assert resp["ReplicationStatus"][0]["Status"] == "InSync"
|
||
|
|
|
||
|
|
primary_secret = use1.describe_secret(SecretId=primary_arn)
|
||
|
|
assert primary_secret["ARN"] == primary_arn
|
||
|
|
|
||
|
|
replica_arn = primary_arn.replace("us-east-1", "us-east-2")
|
||
|
|
replica_secret = use2.describe_secret(SecretId=replica_arn)
|
||
|
|
assert replica_secret["ARN"] == replica_arn
|
||
|
|
assert replica_secret["Name"] == "dup"
|
||
|
|
assert replica_secret["VersionIdsToStages"] == primary_secret["VersionIdsToStages"]
|
||
|
|
|
||
|
|
replica_value = use2.get_secret_value(SecretId=replica_arn)
|
||
|
|
assert replica_value["Name"] == "dup"
|
||
|
|
assert replica_value["SecretString"] == "s"
|
||
|
|
|
||
|
|
|
||
|
|
@mock_aws
|
||
|
|
def test_create_secret_that_duplicates__in_same_region():
|
||
|
|
conn = boto3.client("secretsmanager", region_name="us-east-1")
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
conn.create_secret(
|
||
|
|
Name="dup_same_region",
|
||
|
|
SecretString="s",
|
||
|
|
AddReplicaRegions=[{"Region": "us-east-1"}],
|
||
|
|
)
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == "Invalid replica region."
|
||
|
|
|
||
|
|
|
||
|
|
@mock_aws
|
||
|
|
def test_replicate_secret():
|
||
|
|
us1 = boto3.client("secretsmanager", region_name="us-east-1")
|
||
|
|
us2 = boto3.client("secretsmanager", region_name="us-east-2")
|
||
|
|
|
||
|
|
resp = us1.create_secret(Name="dup", SecretString="s")
|
||
|
|
primary_arn = resp["ARN"]
|
||
|
|
|
||
|
|
resp = us1.replicate_secret_to_regions(
|
||
|
|
SecretId=primary_arn, AddReplicaRegions=[{"Region": "us-east-2"}]
|
||
|
|
)
|
||
|
|
assert resp["ReplicationStatus"][0]["Region"] == "us-east-2"
|
||
|
|
assert resp["ReplicationStatus"][0]["Status"] == "InSync"
|
||
|
|
|
||
|
|
replica_arn = primary_arn.replace("us-east-1", "us-east-2")
|
||
|
|
replica_secret = us2.describe_secret(SecretId=replica_arn)
|
||
|
|
assert replica_secret["ARN"] == replica_arn
|
||
|
|
|
||
|
|
|
||
|
|
@mock_aws
|
||
|
|
def test_replicate_secret__in_same_region():
|
||
|
|
conn = boto3.client("secretsmanager", region_name="us-east-1")
|
||
|
|
|
||
|
|
resp = conn.create_secret(Name="dup", SecretString="s")
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
conn.replicate_secret_to_regions(
|
||
|
|
SecretId=resp["ARN"], AddReplicaRegions=[{"Region": "us-east-1"}]
|
||
|
|
)
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == "Invalid replica region."
|
||
|
|
|
||
|
|
|
||
|
|
@mock_aws
|
||
|
|
def test_replicate_secret__existing_secret():
|
||
|
|
use1 = boto3.client("secretsmanager", region_name="us-east-1")
|
||
|
|
use2 = boto3.client("secretsmanager", region_name="us-east-2")
|
||
|
|
|
||
|
|
secret1 = use1.create_secret(Name="dup", SecretString="s1")["ARN"]
|
||
|
|
secret2 = use2.create_secret(Name="dup", SecretString="s2")["ARN"]
|
||
|
|
|
||
|
|
resp = use1.replicate_secret_to_regions(
|
||
|
|
SecretId=secret1, AddReplicaRegions=[{"Region": "us-east-2"}]
|
||
|
|
)
|
||
|
|
assert resp["ReplicationStatus"][0]["Region"] == "us-east-2"
|
||
|
|
assert resp["ReplicationStatus"][0]["Status"] == "Failed"
|
||
|
|
assert (
|
||
|
|
resp["ReplicationStatus"][0]["StatusMessage"]
|
||
|
|
== "Replication failed: Secret name simple already exists in region us-east-2."
|
||
|
|
)
|
||
|
|
|
||
|
|
# us-east-2 still only has one secret
|
||
|
|
assert len(use2.list_secrets()["SecretList"]) == 1
|
||
|
|
|
||
|
|
# Both secrets still have original values
|
||
|
|
assert use1.get_secret_value(SecretId=secret1)["SecretString"] == "s1"
|
||
|
|
assert use2.get_secret_value(SecretId=secret2)["SecretString"] == "s2"
|
||
|
|
|
||
|
|
|
||
|
|
@mock_aws
|
||
|
|
def test_replicate_secret__overwrite_existing_secret():
|
||
|
|
use1 = boto3.client("secretsmanager", region_name="us-east-1")
|
||
|
|
use2 = boto3.client("secretsmanager", region_name="us-east-2")
|
||
|
|
|
||
|
|
secret1 = use1.create_secret(Name="dup", SecretString="s1")["ARN"]
|
||
|
|
secret2 = use2.create_secret(Name="dup", SecretString="s2")["ARN"]
|
||
|
|
|
||
|
|
resp = use1.replicate_secret_to_regions(
|
||
|
|
SecretId=secret1,
|
||
|
|
AddReplicaRegions=[{"Region": "us-east-2"}],
|
||
|
|
ForceOverwriteReplicaSecret=True,
|
||
|
|
)
|
||
|
|
assert resp["ReplicationStatus"][0]["Status"] == "InSync"
|
||
|
|
|
||
|
|
# us-east-2 still only has one secret
|
||
|
|
assert len(use2.list_secrets()["SecretList"]) == 1
|
||
|
|
|
||
|
|
# Original Secret was yeeted
|
||
|
|
with pytest.raises(ClientError):
|
||
|
|
assert use2.get_secret_value(SecretId=secret2)["SecretString"] == "s1"
|
||
|
|
|
||
|
|
# Use ListSecrets to get the new ARN
|
||
|
|
# And verify it returns a copy of the us-east-1 secret
|
||
|
|
secret2 = use2.list_secrets()["SecretList"][0]["ARN"]
|
||
|
|
assert use2.get_secret_value(SecretId=secret2)["SecretString"] == "s1"
|
||
|
|
|
||
|
|
|
||
|
|
@mock_aws
|
||
|
|
def test_remove_regions_from_replication():
|
||
|
|
use1 = boto3.client("secretsmanager", region_name="us-east-1")
|
||
|
|
use2 = boto3.client("secretsmanager", region_name="us-east-2")
|
||
|
|
usw2 = boto3.client("secretsmanager", region_name="us-west-2")
|
||
|
|
|
||
|
|
arn1 = use1.create_secret(Name="dup", SecretString="s1")["ARN"]
|
||
|
|
arn2 = arn1.replace("us-east-1", "us-east-2")
|
||
|
|
arn3 = arn1.replace("us-east-1", "us-west-2")
|
||
|
|
|
||
|
|
# Replicate to multiple regions
|
||
|
|
use1.replicate_secret_to_regions(
|
||
|
|
SecretId=arn1,
|
||
|
|
AddReplicaRegions=[{"Region": "us-east-2"}, {"Region": "us-west-2"}],
|
||
|
|
)
|
||
|
|
|
||
|
|
# Replications exist
|
||
|
|
use2.describe_secret(SecretId=arn2)
|
||
|
|
usw2.describe_secret(SecretId=arn3)
|
||
|
|
|
||
|
|
# Primary Secret knows about replicas
|
||
|
|
replications = use1.describe_secret(SecretId=arn1)["ReplicationStatus"]
|
||
|
|
assert set([rep["Region"] for rep in replications]) == {"us-east-2", "us-west-2"}
|
||
|
|
|
||
|
|
# Remove us-east-2 replication
|
||
|
|
use1.remove_regions_from_replication(
|
||
|
|
SecretId=arn1, RemoveReplicaRegions=["us-east-2"]
|
||
|
|
)
|
||
|
|
with pytest.raises(ClientError):
|
||
|
|
use2.describe_secret(SecretId=arn2)
|
||
|
|
|
||
|
|
# us-west still exists
|
||
|
|
usw2.describe_secret(SecretId=arn3)
|
||
|
|
|
||
|
|
# Primary Secret no longer knows about us-east-2
|
||
|
|
replications = use1.describe_secret(SecretId=arn1)["ReplicationStatus"]
|
||
|
|
assert set([rep["Region"] for rep in replications]) == {"us-west-2"}
|
||
|
|
|
||
|
|
# Removing region is idempotent - invoking it again does not result in any errors
|
||
|
|
use1.remove_regions_from_replication(
|
||
|
|
SecretId=arn1, RemoveReplicaRegions=["us-east-2"]
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@mock_aws
|
||
|
|
def test_update_replica():
|
||
|
|
use1 = boto3.client("secretsmanager", region_name="us-east-1")
|
||
|
|
use2 = boto3.client("secretsmanager", region_name="us-east-2")
|
||
|
|
|
||
|
|
arn1 = use1.create_secret(
|
||
|
|
Name="dup", SecretString="s1", AddReplicaRegions=[{"Region": "us-east-2"}]
|
||
|
|
)["ARN"]
|
||
|
|
arn2 = arn1.replace("us-east-1", "us-east-2")
|
||
|
|
|
||
|
|
OP_NOT_PERMITTED = "Operation not permitted on a replica secret. Call must be made in primary secret's region."
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use2.update_secret(SecretId=arn2, SecretString="asdf")
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == OP_NOT_PERMITTED
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use2.put_secret_value(SecretId=arn2, SecretString="asdf")
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == OP_NOT_PERMITTED
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use2.tag_resource(SecretId=arn2, Tags=[{"Key": "k", "Value": "v"}])
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == OP_NOT_PERMITTED
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use2.untag_resource(SecretId=arn2, TagKeys=["k1"])
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == OP_NOT_PERMITTED
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use2.put_resource_policy(SecretId=arn2, ResourcePolicy="k1")
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == OP_NOT_PERMITTED
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use2.get_resource_policy(SecretId=arn2)
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == OP_NOT_PERMITTED
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use2.delete_resource_policy(SecretId=arn2)
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == OP_NOT_PERMITTED
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use2.delete_secret(SecretId=arn2)
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == OP_NOT_PERMITTED
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use2.cancel_rotate_secret(SecretId=arn2)
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == OP_NOT_PERMITTED
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use2.rotate_secret(SecretId=arn2)
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == OP_NOT_PERMITTED
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use2.restore_secret(SecretId=arn2)
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == OP_NOT_PERMITTED
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use2.replicate_secret_to_regions(SecretId=arn2, AddReplicaRegions=[{}])
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == OP_NOT_PERMITTED
|
||
|
|
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use2.remove_regions_from_replication(SecretId=arn2, RemoveReplicaRegions=["a"])
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert err["Message"] == OP_NOT_PERMITTED
|
||
|
|
|
||
|
|
|
||
|
|
@mock_aws
|
||
|
|
def test_delete_while_duplication_exist():
|
||
|
|
use1 = boto3.client("secretsmanager", region_name="us-east-1")
|
||
|
|
|
||
|
|
region = "us-east-2"
|
||
|
|
arn = use1.create_secret(
|
||
|
|
Name="dup", SecretString="s1", AddReplicaRegions=[{"Region": region}]
|
||
|
|
)["ARN"]
|
||
|
|
|
||
|
|
# unable to delete
|
||
|
|
with pytest.raises(ClientError) as exc:
|
||
|
|
use1.delete_secret(SecretId=arn)
|
||
|
|
err = exc.value.response["Error"]
|
||
|
|
assert err["Code"] == "InvalidParameterException"
|
||
|
|
assert (
|
||
|
|
err["Message"]
|
||
|
|
== f"You can't delete secret {arn} that still has replica regions [{region}]"
|
||
|
|
)
|
||
|
|
|
||
|
|
# Remove us-east-2 replication
|
||
|
|
use1.remove_regions_from_replication(SecretId=arn, RemoveReplicaRegions=[region])
|
||
|
|
|
||
|
|
# Now we can delete
|
||
|
|
use1.delete_secret(SecretId=arn)
|