Java 类com.amazonaws.services.kms.AWSKMSClient 实例源码

/**
 * @deprecated use {@link AmazonS3EncryptionClientBuilder#withEncryptionMaterials(EncryptionMaterialsProvider)} and
 *                 {@link AmazonS3EncryptionClientBuilder#withCredentials(AWSCredentialsProvider)} and
 *                 {@link AmazonS3EncryptionClientBuilder#withCryptoConfiguration(CryptoConfiguration)} and
 *                 {@link AmazonS3EncryptionClientBuilder#withClientConfiguration(ClientConfiguration)} and
 *                 {@link AmazonS3EncryptionClientBuilder#withMetricsCollector(RequestMetricCollector)} and
 *                 {@link AmazonS3EncryptionClientBuilder#withKmsClient(AWSKMS)}
 */
@Deprecated
public AmazonS3EncryptionClient(AWSKMSClient kms,
        AWSCredentialsProvider credentialsProvider,
        EncryptionMaterialsProvider kekMaterialsProvider,
        ClientConfiguration clientConfig,
        CryptoConfiguration cryptoConfig,
        RequestMetricCollector requestMetricCollector) {
    super(credentialsProvider, clientConfig, requestMetricCollector);
    assertParameterNotNull(kekMaterialsProvider,
            "EncryptionMaterialsProvider parameter must not be null.");
    assertParameterNotNull(cryptoConfig,
            "CryptoConfiguration parameter must not be null.");
    this.isKMSClientInternal = kms == null;
    this.kms = isKMSClientInternal 
        ? newAWSKMSClient(credentialsProvider, clientConfig, cryptoConfig, 
                requestMetricCollector)
        : kms;
    this.crypto = new CryptoModuleDispatcher(this.kms, new S3DirectImpl(),
            credentialsProvider, kekMaterialsProvider, cryptoConfig);
}

@Test
public void test_getKmsKeyState_happy() {
    String awsRegion = "aws region";

    String kmsKeyId = "kms key id";
    String state = "state";
    AWSKMSClient kmsClient = mock(AWSKMSClient.class);
    when(kmsClientFactory.getClient(awsRegion)).thenReturn(kmsClient);
    when(kmsClient.describeKey(anyObject())).thenReturn(
            new DescribeKeyResult()
                    .withKeyMetadata(
                            new KeyMetadata()
                                    .withKeyState(state)));

    String result = kmsService.getKmsKeyState(kmsKeyId, awsRegion);

    assertEquals(state, result);
}

@Test
public void test_validateKmsKeyIsUsable_returns_true_when_state_is_pending_deletion() {
    String keyId = "key id";
    String awsRegion = "aws region";

    AWSKMSClient kmsClient = mock(AWSKMSClient.class);
    when(kmsClientFactory.getClient(awsRegion)).thenReturn(kmsClient);
    when(kmsClient.describeKey(anyObject())).thenReturn(
            new DescribeKeyResult()
                    .withKeyMetadata(
                            new KeyMetadata()
                                    .withKeyState(KeyState.PendingDeletion)));

    boolean result = kmsService.kmsKeyIsDisabledOrScheduledForDeletion(keyId, awsRegion);

    assertTrue(result);
}

@Test
public void test_validateKmsKeyIsUsable_return_true_when_state_is_disabled() {
    String keyId = "key id";
    String awsRegion = "aws region";

    AWSKMSClient kmsClient = mock(AWSKMSClient.class);
    when(kmsClientFactory.getClient(awsRegion)).thenReturn(kmsClient);
    when(kmsClient.describeKey(anyObject())).thenReturn(
            new DescribeKeyResult()
                    .withKeyMetadata(
                            new KeyMetadata()
                                    .withKeyState(KeyState.Disabled)));

    boolean result = kmsService.kmsKeyIsDisabledOrScheduledForDeletion(keyId, awsRegion);

    assertTrue(result);
}

@Test
public void test_validateKmsKeyIsUsable_returns_false_when_state_is_not_deletion_or_disabled() {
    String keyId = "key id";
    String awsRegion = "aws region";

    AWSKMSClient kmsClient = mock(AWSKMSClient.class);
    when(kmsClientFactory.getClient(awsRegion)).thenReturn(kmsClient);
    when(kmsClient.describeKey(anyObject())).thenReturn(
            new DescribeKeyResult()
                    .withKeyMetadata(
                            new KeyMetadata()
                                    .withKeyState(KeyState.Enabled)));

    boolean result = kmsService.kmsKeyIsDisabledOrScheduledForDeletion(keyId, awsRegion);

    assertFalse(result);
}

@Test(expected = ApiException.class)
public void test_validateKmsKeyIsUsable_deletes_kms_key_when_not_usable() {

    String id = "id";
    String awsKmsKeyArn = "aws kms key arn";
    String iamPrincipalArn = "arn";
    String awsRegion = "aws region";

    AwsIamRoleKmsKeyRecord kmsKey = mock(AwsIamRoleKmsKeyRecord.class);
    when(kmsKey.getId()).thenReturn(id);
    when(kmsKey.getAwsKmsKeyId()).thenReturn(awsKmsKeyArn);
    when(kmsKey.getAwsRegion()).thenReturn(awsRegion);

    AWSKMSClient kmsClient = mock(AWSKMSClient.class);
    when(kmsClientFactory.getClient(awsRegion)).thenReturn(kmsClient);
    when(kmsClient.describeKey(anyObject())).thenReturn(
            new DescribeKeyResult()
                    .withKeyMetadata(
                            new KeyMetadata()
                                    .withKeyState(KeyState.PendingDeletion)));

    kmsService.validateKmsKeyIsUsable(kmsKey, iamPrincipalArn);
}

@Test
public void test_validateKmsKeyIsUsable_does_not_delete_kms_key_when_usable() {

    String id = "id";
    String awsKmsKeyArn = "aws kms key arn";
    String iamPrincipalArn = "arn";
    String awsRegion = "aws region";

    AwsIamRoleKmsKeyRecord kmsKey = mock(AwsIamRoleKmsKeyRecord.class);
    when(kmsKey.getId()).thenReturn(id);
    when(kmsKey.getAwsKmsKeyId()).thenReturn(awsKmsKeyArn);
    when(kmsKey.getAwsRegion()).thenReturn(awsRegion);

    AWSKMSClient kmsClient = mock(AWSKMSClient.class);
    when(kmsClientFactory.getClient(awsRegion)).thenReturn(kmsClient);
    when(kmsClient.describeKey(anyObject())).thenReturn(
            new DescribeKeyResult()
                    .withKeyMetadata(
                            new KeyMetadata()
                                    .withKeyState(KeyState.Enabled)));

    kmsService.validateKmsKeyIsUsable(kmsKey, iamPrincipalArn);

    verify(awsIamRoleDao, never()).deleteKmsKeyById(id);
}

private AWSKMSClientBuilder cloneClientBuilder(final AWSKMSClientBuilder builder) {
    // We need to copy all arguments out of the builder in case it's mutated later on.
    // Unfortunately AWSKMSClientBuilder doesn't support .clone() so we'll have to do it by hand.

    if (builder.getEndpoint() != null) {
        // We won't be able to set the region later if a custom endpoint is set.
        throw new IllegalArgumentException("Setting endpoint configuration is not compatible with passing a " +
                                           "builder to the KmsMasterKeyProvider. Use withCustomClientFactory" +
                                           " instead.");
    }

    final AWSKMSClientBuilder newBuilder = AWSKMSClient.builder();
    newBuilder.setClientConfiguration(builder.getClientConfiguration());
    newBuilder.setCredentials(builder.getCredentials());
    newBuilder.setEndpointConfiguration(builder.getEndpoint());
    newBuilder.setMetricsCollector(builder.getMetricsCollector());
    if (builder.getRequestHandlers() != null) {
        newBuilder.setRequestHandlers(builder.getRequestHandlers().toArray(new RequestHandler2[0]));
    }
    return newBuilder;
}

@Override
public KeyProvider build() {
    if ( null == key || 0 == key.length ) {
        return new KeyProviderImpl(null);
    } else if ( 16 == key.length ) {
        return new KeyProviderImpl(new SecretKeySpec(key, "AES"));
    }
    AWSKMS kms = _amazonWebServiceClients.withEndpoint(
        new AWSKMSClient(
            _credProviderFactory.create(credProvider),
            _clientConfigurations.withProxy(new ClientConfiguration(), proxy)),
        endpoint);
    key = kms.decrypt(new DecryptRequest()
                      .withCiphertextBlob(ByteBuffer.wrap(key)))
        .getPlaintext().array();
    if ( 16 != key.length ) {
        LOG.warn("Expected decrypted key to be exactly 16 bytes, got "+key.length+" bytes. Please "+
                 "verify the key was not base64 encoded before encrypting with KMS");
        return new KeyProviderImpl(null);
    }
    return new KeyProviderImpl(new SecretKeySpec(key, "AES"));
}

@Before
public void setup() throws Exception {
    kmsClient = mock(AWSKMSClient.class);
    urlResolver = mock(UrlResolver.class);
    lambdaClient = mock(AWSLambdaClient.class);

    mockWebServer = new MockWebServer();
    mockWebServer.start();
    vaultUrl = "http://localhost:" + mockWebServer.getPort();

    when(urlResolver.resolve()).thenReturn(vaultUrl);


    mockStatic(Regions.class);

    when(Regions.getCurrentRegion()).thenReturn(RegionUtils.getRegion("us-west-2"));
    whenNew(AWSLambdaClient.class).withNoArguments().thenReturn(lambdaClient);
    whenNew(AWSKMSClient.class).withAnyArguments().thenReturn(kmsClient);
}

@Override
public void validate(String name, Configured configuration,
    PluginExceptionConditionAccumulator accumulator, LocalizationContext localizationContext) {

  AmazonEC2Client ec2Client = provider.getClient();
  AWSKMSClient kmsClient = provider.getKmsClient();

  checkImage(ec2Client, configuration, accumulator, localizationContext);
  Map<String, String> vpcSubnetMap = checkSubnetId(ec2Client, configuration, accumulator, localizationContext);
  Map<String, Set<String>> vpcSecurityGroupMap = checkSecurityGroupIds(ec2Client, configuration, accumulator, localizationContext);
  checkVpc(vpcSubnetMap, vpcSecurityGroupMap, accumulator, localizationContext);
  checkAvailabilityZone(ec2Client, configuration, accumulator, localizationContext);
  checkPlacementGroup(ec2Client, configuration, accumulator, localizationContext);
  checkTenancy(configuration, accumulator, localizationContext);
  checkIamProfileName(configuration, accumulator, localizationContext);
  checkRootVolumeSize(configuration, accumulator, localizationContext);
  checkRootVolumeType(configuration, accumulator, localizationContext);
  checkEbsVolumes(kmsClient, configuration, accumulator, localizationContext);
  checkKeyName(ec2Client, configuration, accumulator, localizationContext);
  checkSpotParameters(configuration, accumulator, localizationContext);
}

@Override
public String decrypt(AwsParamsDto awsParamsDto, String base64ciphertextBlob)
{
    // Construct a new AWS KMS service client using the specified client configuration.
    // A credentials provider chain will be used that searches for credentials in this order:
    // - Environment Variables - AWS_ACCESS_KEY_ID and AWS_SECRET_KEY
    // - Java System Properties - aws.accessKeyId and aws.secretKey
    // - Instance Profile Credentials - delivered through the Amazon EC2 metadata service
    AWSKMSClient awsKmsClient = new AWSKMSClient(awsHelper.getClientConfiguration(awsParamsDto));

    // Decode the base64 encoded ciphertext.
    ByteBuffer ciphertextBlob = ByteBuffer.wrap(Base64.decodeBase64(base64ciphertextBlob));

    // Create the decrypt request.
    DecryptRequest decryptRequest = new DecryptRequest().withCiphertextBlob(ciphertextBlob);

    // Call AWS KMS decrypt service method.
    DecryptResult decryptResult = kmsOperations.decrypt(awsKmsClient, decryptRequest);

    // Get decrypted plaintext data.
    ByteBuffer plainText = decryptResult.getPlaintext();

    // Return the plain text as a string.
    return new String(plainText.array(), StandardCharsets.UTF_8);
}

@Override
public DecryptResult decrypt(AWSKMSClient awsKmsClient, DecryptRequest decryptRequest)
{
    // Check the cipher text.
    if (decryptRequest.getCiphertextBlob().equals(ByteBuffer.wrap(Base64.decodeBase64(MOCK_CIPHER_TEXT_INVALID))))
    {
        throw new InvalidCiphertextException("(Service: AWSKMS; Status Code: 400; Error Code: InvalidCiphertextException; Request ID: NONE)");
    }

    DecryptResult decryptResult = new DecryptResult();

    // Convert the test plain text to byte buffer and set the plain text return value.
    decryptResult.setPlaintext(ByteBuffer.wrap(MOCK_PLAIN_TEXT.getBytes()));

    return decryptResult;
}

@BeforeMethod
public void setUp() {
    mockCredentials = mock(AWSCredentialsProvider.class);
    mockClient = mock(AmazonIdentityManagementClient.class);
    ClientConfiguration mockConfig = mock(ClientConfiguration.class);
    IAMPolicyManager policyManager = new IAMPolicyManager(mockClient, mockCredentials, mockConfig);

    // The mockito spy acts like original object but mocks out the getAccount() method. As the getAccount() calls
    // directly rather than via a client that we can pass in we need to mock this out using a spy.
    partiallyMockedPolicyManager = spy(policyManager);
    doReturn(ACCOUNT).when(partiallyMockedPolicyManager).getAccount();

    // Set up KMSEncryptor for testing the policy creation methods. This gets a bit complicated but we need to
    // mock all the AWS dependencies from the KMSManager before using it to create the KMSEncryptor. The getAliasArn
    // needs to be mocked out with a spy to stop the call to getAccount.
    mockKMSClient = mock(AWSKMSClient.class);
    KMSManager kmsManager = new KMSManager(mockKMSClient, mockCredentials, mockConfig, group);
    KMSManager partiallyMockedKMSManager = spy(kmsManager);
    doReturn(KMS_ALIAS_ARN).when(partiallyMockedKMSManager).getAliasArn();
    kmsEncryptor = new KMSEncryptor(partiallyMockedKMSManager, mockCredentials, mockConfig, group, mock(AwsCrypto.class), EncryptionStrength.AES_256);

    // Set up store for testing the policy creation methods. Mock out the getArn method with a spy to stop the
    // call to getAccount().
    mockDynamoDBClient = mock(AmazonDynamoDBClient.class);
    DynamoDB store = new DynamoDB(mockDynamoDBClient, mockCredentials, mockConfig, group, new ReentrantReadWriteLock());
    partiallyMockedStore = spy(store);
    doReturn(DYNAMODB_ARN).when(partiallyMockedStore).getArn();
}

@BeforeMethod
public void setUp() throws Exception {
    AWSCredentialsProvider mockCredentials = mock(AWSCredentialsProvider.class);
    this.mockKMSClient = mock(AWSKMSClient.class);
    this.group = new SecretsGroupIdentifier(TEST_REGION, TEST_GROUP);
    ClientConfiguration mockConfig = mock(ClientConfiguration.class);

    KMSManager manager = new KMSManager(mockKMSClient, mockCredentials, mockConfig, group);
    this.kmsManager = spy(manager);
    doReturn(KMS_ALIAS_ARN).when(kmsManager).getAliasArn();
}

/**
 * Creates and returns a new instance of AWS KMS client in the case when
 * an explicit AWS KMS client is not specified.
 */
private AWSKMSClient newAWSKMSClient(
        AWSCredentialsProvider credentialsProvider,
        ClientConfiguration clientConfig,
        CryptoConfiguration cryptoConfig,
        RequestMetricCollector requestMetricCollector
) {
    final AWSKMSClient kmsClient = new AWSKMSClient(
        credentialsProvider, clientConfig, requestMetricCollector);
    final Region kmsRegion = cryptoConfig.getAwsKmsRegion();
    if (kmsRegion != null)
        kmsClient.setRegion(kmsRegion);
    return kmsClient;
}

public AWSKMSClient build() {
    ClientConfiguration config = new ClientConfiguration();
    if (!Util.fixNull(host).trim().isEmpty()) {
        config.setProxyHost(this.host);
        config.setProxyPort(this.port);
    }
    AWSKMSClient client = new AWSKMSClient(new DefaultAWSCredentialsProviderChain(), config);
    if (!Util.fixNull(region).trim().isEmpty()) {
        client.setRegion(Region.getRegion(Regions.fromName(region)));
    }
    return client;
}

@Test
public void regionAndProxyShouldBeReflectedInClient(){
    AwsKmsClientBuilder clientBuilder = new AwsKmsClientBuilder();
    clientBuilder.region("eu-west-1").proxyHost("host").proxyPort(8080);
    AWSKMSClient amazonKmsClient = clientBuilder.build();
    {
        ClientConfiguration configuration = (ClientConfiguration) Whitebox.getInternalState(amazonKmsClient,"clientConfiguration");
        assertThat(configuration.getProxyHost()).isEqualTo("host");
        assertThat(configuration.getProxyPort()).isEqualTo(8080);
    }
}

@Test
public void regionSetButHostEmpty(){
    AwsKmsClientBuilder clientBuilder = new AwsKmsClientBuilder();
    clientBuilder.region("eu-west-1").proxyPort(8080);
    AWSKMSClient amazonKmsClient = clientBuilder.build();
    {
        ClientConfiguration configuration = (ClientConfiguration)Whitebox.getInternalState(amazonKmsClient,"clientConfiguration");
        assertThat(configuration.getProxyHost()).isNull();
        assertThat(configuration.getProxyPort()).isEqualTo(-1);
    }
}

@Test
public void regionSetButHostFilledWithSpaces() {
    AwsKmsClientBuilder clientBuilder = new AwsKmsClientBuilder();
    clientBuilder.region("eu-west-1").proxyHost("    ").proxyPort(8080);
    AWSKMSClient amazonS3Client = clientBuilder.build();
    {
        ClientConfiguration configuration = (ClientConfiguration)Whitebox.getInternalState(amazonS3Client,"clientConfiguration");
        assertThat(configuration.getProxyHost()).isNull();
        assertThat(configuration.getProxyPort()).isEqualTo(-1);
    }
}

@Test
public void regionNotSet(){
    AwsKmsClientBuilder clientBuilder = new AwsKmsClientBuilder();
    AWSKMSClient amazonKmsClient = clientBuilder.build();

    URI endpoint =(URI)Whitebox.getInternalState(amazonKmsClient,"endpoint");
    assertThat(endpoint.toString()).contains("us-east-1");
}

@Test
public void regionSet(){
    AwsKmsClientBuilder clientBuilder = new AwsKmsClientBuilder();
    AWSKMSClient amazonKmsClient = clientBuilder.region("eu-west-1").build();

    URI endpoint =(URI)Whitebox.getInternalState(amazonKmsClient,"endpoint");
    assertThat(endpoint.toString()).contains("eu-west-1");
}

@Test
public void regionWasEmptyString(){
    AwsKmsClientBuilder clientBuilder = new AwsKmsClientBuilder();
    AWSKMSClient amazonKmsClient = clientBuilder.region("   ").build();

    URI endpoint =(URI)Whitebox.getInternalState(amazonKmsClient,"endpoint");
    assertThat(endpoint.toString()).contains("us-east-1");
}

@Test
public void passwordUsesTheS3Bucket() throws Exception {
    S3Object mockS3Object = mock(S3Object.class);
    AmazonS3Client mockClient = mock(AmazonS3Client.class);
    when(mockClientBuilder.build()).thenReturn(mockClient);
    when(mockClient.getObject(any(GetObjectRequest.class))).thenReturn(mockS3Object);
    AWSKMSClient mockKmsClient = mock(AWSKMSClient.class);
    when(mockKmsClientBuilder.build()).thenReturn(mockKmsClient);

    S3ObjectInputStream mockS3ObjectInputStream = mock(S3ObjectInputStream.class);
    when(mockS3Object.getObjectContent()).thenReturn(mockS3ObjectInputStream);
    when(mockS3ObjectInputStream.read(new byte[anyInt()], anyInt(), anyByte()))
            .thenAnswer(new WriteBufferAnswer("encryptedPassword".getBytes()))
            .thenReturn(-1);

    DecryptResult result = new DecryptResult();
    CharsetEncoder charsetEncoder = Charset.forName("UTF-8").newEncoder();
    result.setPlaintext(charsetEncoder.encode(CharBuffer.wrap("password")));
    when(mockKmsClient.decrypt(any(DecryptRequest.class))).thenReturn(result);

    Secret secret = test.getPassword();

    // have we got the expected password
    assertThat(secret.getPlainText()).isEqualTo("password");

    // have we used the bucket
    ArgumentCaptor<GetObjectRequest> capturedObjectRequest = ArgumentCaptor.forClass(GetObjectRequest.class);
    verify(mockClient).getObject(capturedObjectRequest.capture());
    assertThat(capturedObjectRequest.getValue().getBucketName()).isEqualTo("bucketUri");
    assertThat(capturedObjectRequest.getValue().getS3ObjectId().getKey()).isEqualTo("/bucketPath");

    // have we used kms to decrypt
    ArgumentCaptor<DecryptRequest> capturedDecryptRequest = ArgumentCaptor.forClass(DecryptRequest.class);
    verify(mockKmsClient).decrypt(capturedDecryptRequest.capture());
    assertThat(capturedDecryptRequest.getValue().getEncryptionContext()).containsEntry("someEncryptContextKey", "kmsEncryptContextValue");
    ByteBuffer ciphertextBlob = capturedDecryptRequest.getValue().getCiphertextBlob();
    assertThat(new String(Charset.forName("UTF-8").decode(ciphertextBlob).array())).isEqualTo("encryptedPassword");

}

@Test
public void closesIfIOExceptionWhileReading() throws Exception {
    AmazonS3Client mockClient = mock(AmazonS3Client.class);
    when(mockClientBuilder.build()).thenReturn(mockClient);

    AWSKMSClient mockKmsClient = mock(AWSKMSClient.class);
    when(mockKmsClientBuilder.build()).thenReturn(mockKmsClient);

    S3Object mockS3Object = mock(S3Object.class);
    when(mockClient.getObject(any(GetObjectRequest.class))).thenReturn(mockS3Object);
    S3ObjectInputStream mockS3ObjectInputStream = mock(S3ObjectInputStream.class);
    when(mockS3Object.getObjectContent()).thenReturn(mockS3ObjectInputStream);
    when(mockS3ObjectInputStream.read(new byte[anyInt()], anyInt(), anyByte()))
            .thenAnswer(new WriteBufferAnswer("encryptedPassword".getBytes()))
            .thenThrow(new IOException("something went wrong"))
            .thenReturn(-1);

    DecryptResult result = new DecryptResult();
    CharsetEncoder charsetEncoder = Charset.forName("UTF-8").newEncoder();
    result.setPlaintext(charsetEncoder.encode(CharBuffer.wrap("password")));
    when(mockKmsClient.decrypt(any(DecryptRequest.class))).thenReturn(result);

    Secret secret = null;
    try {
        secret = test.getPassword();
        TestCase.fail("should have thrown exception");
    } catch (AwsBucketReadingException e) {
        assertThat(e.getCause()).isInstanceOf(IOException.class);
    }

    // have we used the bucket
    ArgumentCaptor<GetObjectRequest> capturedObjectRequest = ArgumentCaptor.forClass(GetObjectRequest.class);
    verify(mockClient).getObject(capturedObjectRequest.capture());
    assertThat(capturedObjectRequest.getValue().getBucketName()).isEqualTo("bucketUri");
    assertThat(capturedObjectRequest.getValue().getS3ObjectId().getKey()).isEqualTo("/bucketPath");

    // and we have closed it even if there was an exception
    verify(mockS3Object).close();
}

/**
 * Returns a KMS client for the given region.  Clients are cached by region.
 *
 * @param region Region to configure a client for
 * @return AWS KMS client
 */
public AWSKMSClient getClient(Region region) {
    AWSKMSClient client = kmsClientMap.get(region);

    if (client == null) {
        final AWSKMSClient newClient = new AWSKMSClient();
        newClient.setRegion(region);
        kmsClientMap.put(region, newClient);
        client = newClient;
    }

    return client;
}

/**
 * Returns a KMS client for the given region name.  Clients are cached by region.
 *
 * @param regionName Region to configure a client for
 * @return AWS KMS client
 */
public AWSKMSClient getClient(String regionName) {
    try {
        final Region region = Region.getRegion(Regions.fromName(regionName));
        return getClient(region);
    } catch (IllegalArgumentException iae) {
        throw ApiException.newBuilder()
                .withApiErrors(DefaultApiError.AUTHENTICATION_ERROR_INVALID_REGION)
                .withExceptionCause(iae.getCause())
                .withExceptionMessage("Specified region is not valid.")
                .build();
    }
}

/**
 * Updates the KMS key policy in AWS for the given CMK
 */
protected void updateKmsKeyPolicy(String updatedPolicyJson, String awsKmsKeyArn, String kmsCMKRegion) {

    AWSKMSClient kmsClient = kmsClientFactory.getClient(kmsCMKRegion);

    kmsClient.putKeyPolicy(new PutKeyPolicyRequest()
            .withKeyId(awsKmsKeyArn)
            .withPolicyName("default")
            .withPolicy(updatedPolicyJson)
    );
}

/**
 * Get the state of the KMS key
 * @param kmsKeyId - The AWS KMS Key ID
 * @param region - The KMS key region
 * @return - KMS key state
 */
protected String getKmsKeyState(String kmsKeyId, String region) {

    AWSKMSClient kmsClient = kmsClientFactory.getClient(region);
    DescribeKeyRequest request = new DescribeKeyRequest().withKeyId(kmsKeyId);

    return kmsClient.describeKey(request)
            .getKeyMetadata()
            .getKeyState();
}

/**
 * Delete a CMK in AWS
 * @param kmsKeyId - The AWS KMS Key ID
 * @param region - The KMS key region
 */
protected void scheduleKmsKeyDeletion(String kmsKeyId, String region, Integer pendingWindowInDays) {

    final AWSKMSClient kmsClient = kmsClientFactory.getClient(region);
    final ScheduleKeyDeletionRequest scheduleKeyDeletionRequest = new ScheduleKeyDeletionRequest()
            .withKeyId(kmsKeyId)
            .withPendingWindowInDays(pendingWindowInDays);

    kmsClient.scheduleKeyDeletion(scheduleKeyDeletionRequest);
}

/**
 * Encrypts the data provided using KMS based on the provided region and key id.
 *
 * @param regionName Region where key is located
 * @param keyId Key id
 * @param data Data to be encrypted
 * @return encrypted data
 */
private byte[] encrypt(final String regionName, final String keyId, final byte[] data) {
    Region region;
    try {
        region = Region.getRegion(Regions.fromName(regionName));
    } catch (IllegalArgumentException iae) {
        throw ApiException.newBuilder()
                .withApiErrors(DefaultApiError.AUTH_IAM_ROLE_AWS_REGION_INVALID)
                .withExceptionCause(iae)
                .build();
    }

    final AWSKMSClient kmsClient = kmsClientFactory.getClient(region);

    try {
        final EncryptResult encryptResult =
                kmsClient.encrypt(new EncryptRequest().withKeyId(keyId).withPlaintext(ByteBuffer.wrap(data)));

        return encryptResult.getCiphertextBlob().array();
    } catch (AmazonClientException ace) {
        throw ApiException.newBuilder()
                .withApiErrors(DefaultApiError.INTERNAL_SERVER_ERROR)
                .withExceptionCause(ace)
                .withExceptionMessage(
                        String.format("Unexpected error communicating with AWS KMS for region %s.", regionName))
                .build();
    }
}

@Test
public void test_validatePolicy_validates_policy_when_validate_interval_has_passed() {
    String kmsKeyArn = "kms key arn";
    String awsIamRoleRecordId = "aws iam role record id";
    String kmsCMKRegion = "kmsCMKRegion";
    String policy = "policy";
    OffsetDateTime lastValidated = OffsetDateTime.of(2016, 1, 1, 1, 1,
            1, 1, ZoneOffset.UTC);
    OffsetDateTime now = OffsetDateTime.now();

    AWSKMSClient client = mock(AWSKMSClient.class);
    when(client.describeKey(anyObject())).thenReturn(
            new DescribeKeyResult()
                    .withKeyMetadata(
                            new KeyMetadata()
                                    .withKeyState(KeyState.Enabled)));

    when(kmsClientFactory.getClient(kmsCMKRegion)).thenReturn(client);

    GetKeyPolicyResult result = mock(GetKeyPolicyResult.class);
    when(result.getPolicy()).thenReturn(policy);
    when(client.getKeyPolicy(new GetKeyPolicyRequest().withKeyId(kmsKeyArn)
            .withPolicyName("default"))).thenReturn(result);
    when(kmsPolicyService.isPolicyValid(policy)).thenReturn(true);

    AwsIamRoleKmsKeyRecord kmsKey = mock(AwsIamRoleKmsKeyRecord.class);
    when(kmsKey.getAwsIamRoleId()).thenReturn(awsIamRoleRecordId);
    when(kmsKey.getAwsKmsKeyId()).thenReturn(kmsKeyArn);
    when(kmsKey.getAwsRegion()).thenReturn(kmsCMKRegion);
    when(kmsKey.getLastValidatedTs()).thenReturn(lastValidated);
    when(awsIamRoleDao.getKmsKey(awsIamRoleRecordId, kmsCMKRegion)).thenReturn(Optional.of(kmsKey));

    when(dateTimeSupplier.get()).thenReturn(now);
    kmsService.validateKeyAndPolicy(kmsKey, kmsKeyArn);

    verify(client, times(1)).getKeyPolicy(new GetKeyPolicyRequest().withKeyId(kmsKeyArn)
            .withPolicyName("default"));
    verify(kmsPolicyService, times(1)).isPolicyValid(policy);
}

@Test
public void test_validateKeyAndPolicy_does_not_throw_error_when_cannot_validate() {
    String keyId = "key-id";
    String iamPrincipalArn = "arn";
    String kmsCMKRegion = "kmsCMKRegion";
    String policy = "policy";
    OffsetDateTime lastValidated = OffsetDateTime.of(2016, 1, 1, 1, 1,
            1, 1, ZoneOffset.UTC);
    OffsetDateTime now = OffsetDateTime.now();
    when(dateTimeSupplier.get()).thenReturn(now);

    AwsIamRoleKmsKeyRecord kmsKey = mock(AwsIamRoleKmsKeyRecord.class);
    when(kmsKey.getAwsKmsKeyId()).thenReturn(keyId);
    when(kmsKey.getAwsIamRoleId()).thenReturn(iamPrincipalArn);
    when(kmsKey.getAwsRegion()).thenReturn(kmsCMKRegion);
    when(kmsKey.getLastValidatedTs()).thenReturn(lastValidated);

    AWSKMSClient client = mock(AWSKMSClient.class);
    when(kmsClientFactory.getClient(kmsCMKRegion)).thenReturn(client);

    GetKeyPolicyResult result = mock(GetKeyPolicyResult.class);
    when(result.getPolicy()).thenReturn(policy);
    when(client.getKeyPolicy(new GetKeyPolicyRequest().withKeyId(keyId).withPolicyName("default"))).thenThrow(AmazonServiceException.class);

    kmsService.validateKeyAndPolicy(kmsKey, iamPrincipalArn);

    verify(kmsPolicyService, never()).isPolicyValid(policy);
    verify(client, never()).putKeyPolicy(anyObject());
}

public Observable<Void> start(VertxContext<Server> vertxContext,
                              JsonObject config) {
    AwsKms _this = this;
    SfsVertx sfsVertx = vertxContext.vertx();
    Context context = sfsVertx.getOrCreateContext();
    return Defer.aVoid()
            .filter(aVoid -> started.compareAndSet(false, true))
            .flatMap(aVoid -> {
                String keyStoreAwsKmsEndpoint = ConfigHelper.getFieldOrEnv(config, "keystore.aws.kms.endpoint");
                Preconditions.checkArgument(keyStoreAwsKmsEndpoint != null, "keystore.aws.kms.endpoint is required");

                _this.keyId = ConfigHelper.getFieldOrEnv(config, "keystore.aws.kms.key_id");
                Preconditions.checkArgument(_this.keyId != null, "keystore.aws.kms.key_id is required");

                _this.accessKeyId = ConfigHelper.getFieldOrEnv(config, "keystore.aws.kms.access_key_id");
                Preconditions.checkArgument(_this.accessKeyId != null, "keystore.aws.kms.access_key_id is required");

                _this.secretKey = ConfigHelper.getFieldOrEnv(config, "keystore.aws.kms.secret_key");
                Preconditions.checkArgument(_this.secretKey != null, "keystore.aws.kms.secret_key is required");


                return RxHelper.executeBlocking(context, sfsVertx.getBackgroundPool(),
                        () -> {
                            kms = new AWSKMSClient(new AWSCredentials() {
                                @Override
                                public String getAWSAccessKeyId() {
                                    return _this.accessKeyId;
                                }

                                @Override
                                public String getAWSSecretKey() {
                                    return _this.secretKey;
                                }
                            });
                            kms.setEndpoint(keyStoreAwsKmsEndpoint);
                            return (Void) null;
                        });
            })
            .singleOrDefault(null);
}

/**
 * Authenticates with Cerberus and decrypts and sets the token and expiration details.
 *
 * @param iamPrincipalArn AWS IAM principal ARN used to auth with cerberus
 * @param region          AWS Region used in auth with cerberus
 */
protected void getAndSetToken(final String iamPrincipalArn, final Region region) {
    final AWSKMSClient kmsClient = new AWSKMSClient();
    kmsClient.setRegion(region);

    final String encryptedAuthData = getEncryptedAuthData(iamPrincipalArn, region);
    final VaultAuthResponse decryptedToken = decryptToken(kmsClient, encryptedAuthData);
    final DateTime expires = DateTime.now(DateTimeZone.UTC)
            .plusSeconds(decryptedToken.getLeaseDuration() - paddingTimeInSeconds);

    credentials = new TokenVaultCredentials(decryptedToken.getClientToken());
    expireDateTime = expires;
}

@Before
public void setup() throws Exception {
    kmsClient = mock(AWSKMSClient.class);
    urlResolver = mock(UrlResolver.class);
    provider = new InstanceRoleVaultCredentialsProvider(urlResolver);

    whenNew(AWSKMSClient.class).withAnyArguments().thenReturn(kmsClient);
    mockStatic(EC2MetadataUtils.class);
    mockGetCurrentRegion();
}

private String provisionKmsCmkForBackupRegion(String region) {
    Policy kmsPolicy = new Policy();
    final List<Statement> statements = new LinkedList<>();
    // allow the configured admin iam principals all permissions
    configStore.getBackupAdminIamPrincipals().forEach( principal -> {
        log.debug("Adding principal: {} to the CMK Policy for region {}", principal, region);
        statements.add(new Statement(Statement.Effect.Allow)
            .withId("Principal " + principal + " Has All Actions")
            .withPrincipals(new Principal(AWS_PROVIDER, principal, false))
            .withActions(KMSActions.AllKMSActions)
            .withResources(new Resource("*")));
    });

    kmsPolicy.setStatements(statements);

    String policyString = kmsPolicy.toJson();

    log.debug("Creating key for region {} with policy {}", region, policyString);

    AWSKMS kms = AWSKMSClient.builder().withCredentials(getAWSCredentialsProviderChain()).withRegion(region).build();
    CreateKeyResult createKeyResult = kms.createKey(
            new CreateKeyRequest()
                .withPolicy(policyString)
                .withBypassPolicyLockoutSafetyCheck(true)
                .withDescription(String.format("Cerberus Backup Encryption key for env: %S region: %s",
                        environmentMetadata.getName(), region))
                .withTags(
                        new Tag().withTagKey("env").withTagValue(environmentMetadata.getName()),
                        new Tag().withTagKey("region").withTagValue(region),
                        new Tag().withTagKey("cerberus-backup-key").withTagValue("true")

                )
    );

    String keyId = createKeyResult.getKeyMetadata().getKeyId();

    log.info("Created new backup KMS CMK with id: {} for region: {}", keyId, region);

    return keyId;
}

/**
 * Binds all the Amazon services used.
 */
@Override
protected void configure() {
    final Region region = Region.getRegion(Regions.fromName(regionName));
    bind(AmazonEC2.class).toInstance(createAmazonClientInstance(AmazonEC2Client.class, region));
    bind(AmazonCloudFormation.class).toInstance(createAmazonClientInstance(AmazonCloudFormationClient.class, region));
    bind(AmazonIdentityManagement.class).toInstance(createAmazonClientInstance(AmazonIdentityManagementClient.class, region));
    bind(AWSKMS.class).toInstance(createAmazonClientInstance(AWSKMSClient.class, region));
    bind(AmazonS3.class).toInstance(createAmazonClientInstance(AmazonS3Client.class, region));
    bind(AmazonAutoScaling.class).toInstance(createAmazonClientInstance(AmazonAutoScalingClient.class, region));
    bind(AWSSecurityTokenService.class).toInstance(createAmazonClientInstance(AWSSecurityTokenServiceClient.class, region));
    bind(AWSLambda.class).toInstance(createAmazonClientInstance(AWSLambdaClient.class, region));
    bind(AmazonSNS.class).toInstance(createAmazonClientInstance(AmazonSNSClient.class, region));
}

/**
 * Returns the KMS endpoint URL for the specified region.
 *
 * @param kmsClient  the KMS client
 * @param regionName the desired region
 * @return the endpoint URL for the specified region
 * @throws IllegalArgumentException if the endpoint cannot be determined
 */
private static String getKMSEndpointForRegion(AWSKMSClient kmsClient, String regionName) {
  requireNonNull(kmsClient, "kmsClient is null");
  requireNonNull(regionName, "regionName is null");

  com.amazonaws.regions.Region region = RegionUtils.getRegion(regionName);

  if (region == null) {
    throw new IllegalArgumentException(String.format("Unable to find the region %s", regionName));
  }

  String serviceName = kmsClient.getServiceName();
  String protocolPrefix = region.hasHttpsEndpoint(serviceName) ? "https://" : "http://";
  return protocolPrefix + region.getServiceEndpoint(serviceName);
}

/**
 * Using the given AWS Key, encrypt the given bytes
 *
 * @param awsKeyId   unique identifier for the customer master key
 * @param clearBytes the unencrypted bytes to encrypt
 * @return the encrypted bytes
 */
public static byte[] encrypt(String awsKeyId, Map<String, String> encryptionContext, byte[] clearBytes) {
    EncryptRequest encryptRequest = new EncryptRequest();
    encryptRequest.setKeyId(awsKeyId);
    encryptRequest.setPlaintext(ByteBuffer.wrap(clearBytes));
    encryptRequest.setEncryptionContext(encryptionContext);

    AWSKMSClient client = new AWSKMSClient();
    EncryptResult encryptResult = client.encrypt(encryptRequest);

    return encryptResult.getCiphertextBlob().array();
}