[FLINK-32375] Flink AWS Source AssumeRole in VPC

• Type: New Feature
• Status: Resolved
• Priority: Major
• Resolution: Duplicate
• Affects Version/s: 1.15.4
• Component/s: Connectors / AWS

Environment

Flink 1.15.4

running on Amazon KDA (managed flink)

runtime is running inside a VPC

input stream cross-account

• Assignee: Unassigned
• Reporter: Tomas Witzany
• Created: 19/Jun/23 07:52
• Updated: 21/Jun/23 11:18
• Resolved: 19/Jun/23 10:33

Description

Current way to configure auth against AWS supports assuming a role, but when you assume a role in a VPC without a NAT gateway, the global STS endpoint is not accessible. And there is no way to configure the provider to use a different endpoint.

This means that there currently is no supported way to configure AWS auth in such a situation. Note that you can add an sts endpoint to a VPC, but its always a regional endpoint, not the global endpoint.

Options on how you can configure this:

• configuring the aws DefaultsMode, by default legacy, to in-region:
  • environment variables - not possible in KDA
  • system variables - not possible in KDA
  • aws config file - not possible in KDA
• adding endpoint configuration options to the assume role provider

The piece of code that creates the provider and how it could be extended to support endpoint configuration (just an example)

private static AwsCredentialsProvider getAssumeRoleCredentialProvider(
        final Properties configProps, final String configPrefix) {
    return StsAssumeRoleCredentialsProvider.builder()
            .refreshRequest(
                    AssumeRoleRequest.builder()
                            .roleArn(
                                    configProps.getProperty(
                                            AWSConfigConstants.roleArn(configPrefix)))
                            .roleSessionName(
                                    configProps.getProperty(
                                            AWSConfigConstants.roleSessionName(configPrefix)))
                            .externalId(
                                    configProps.getProperty(
                                            AWSConfigConstants.externalId(configPrefix)))
                            .build())
            .stsClient(
                    StsClient.builder()
                            .credentialsProvider(
                                    getCredentialsProvider(
                                            configProps,
                                            AWSConfigConstants.roleCredentialsProvider(
                                                    configPrefix)))
                            .endpointOverride(new URI( // added code
                                    configProps.getProperty(AWSConfigConstants.endpointOverride(configPrefix)) // added code
                            )) // added code
                            .region(getRegion(configProps))
                            .build())
            .build();
} 

 

I am not entirely certain that there is no other way to configure this in my situation, my current plan is to build my own version of the connectors with this option supported. If a feature like this would be nice to have, I would be happy to share my results in a PR afterwards.

However,  if there is a better way to configure this, I would be happy to hear about it. If you know of some trick to do this in KDA, where you have limited options to configure things.

Some other options to attack this problem:

• trying to set system properties on the task manager before the kinesis source is initialized - this is hard as you dont have control over execution order, probably doable though with some hacks
• ask AWS support to set a system property with flink config file options - this is hard as it will involve aws support
• add a NAT gateway to the VPC - this will not be always an option because of security reasons

Issue Links

• duplicates: FLINK-29496 Add Configuration for STS endpoint when using ASSUME_ROLE credential provider