diff --git a/docs/src/main/sphinx/object-storage/metastores.md b/docs/src/main/sphinx/object-storage/metastores.md index 5fec3303d28e..641d31483c0c 100644 --- a/docs/src/main/sphinx/object-storage/metastores.md +++ b/docs/src/main/sphinx/object-storage/metastores.md @@ -432,6 +432,130 @@ properties: - `false` ::: + +(hive-glue-metastore-security-mapping)= +### Security mapping + +Trino supports flexible security mapping for Glue, allowing for separate +credentials or IAM roles for specific users. The IAM role +for a specific query can be selected from a list of allowed roles by providing +it as an *extra credential*. + +Each security mapping entry may specify one or more match criteria. +If multiple criteria are specified, all criteria must match. +The following match criteria are available: + +- `user`: Regular expression to match against username. Example: `alice|bob` +- `group`: Regular expression to match against any of the groups that the user + belongs to. Example: `finance|sales` + +The security mapping must provide one or more configuration settings: + +- `accessKey` and `secretKey`: AWS access key and secret key. This overrides + any globally configured credentials, such as access key or instance credentials. +- `iamRole`: IAM role to use if no user provided role is specified as an + extra credential. This overrides any globally configured IAM role. This role + is allowed to be specified as an extra credential, although specifying it + explicitly has no effect. +- `roleSessionName`: Optional role session name to use with `iamRole`. This can only + be used when `iamRole` is specified. If `roleSessionName` includes the string + `${USER}`, then the `${USER}` portion of the string is replaced with the + current session's username. If `roleSessionName` is not specified, it defaults + to `trino-session`. +- `allowedIamRoles`: IAM roles that are allowed to be specified as an extra + credential. This is useful because a particular AWS account may have permissions + to use many roles, but a specific user should only be allowed to use a subset + of those roles. +- `endpoint`: The Glue catalog endpoint server. This optional property can be used + to override Glue endpoints on a per-user basis. +- `region`: The Glue region to connect to. This optional property can be used + to override Glue regions on a per-user basis. + +The security mapping entries are processed in the order listed in the JSON configuration. +You can specify the default configuration by not including any match criteria for the last entry in the list. + +In addition to the preceding rules, the default mapping can contain the optional +`useClusterDefault` boolean property set to `true` to use the default IAM configuration. +It cannot be used with any other configuration settings. + +If no mapping entry matches and no default is configured, access is denied. + +The configuration JSON is read from a file via `hive.metastore.glue.security-mapping.config-file` +or from an HTTP endpoint via `hive.metastore.glue.security-mapping.config-uri`. + +Example JSON configuration: + +```json +{ + "mappings": [ + { + "user": "bob|charlie", + "iamRole": "arn:aws:iam::123456789101:role/test_default", + "allowedIamRoles": [ + "arn:aws:iam::123456789101:role/test1", + "arn:aws:iam::123456789101:role/test2", + "arn:aws:iam::123456789101:role/test3" + ] + }, + { + "user": "alice", + "accessKey": "AKIAxxxaccess", + "secretKey": "iXbXxxxsecret" + }, + { + "user": "danny", + "iamRole": "arn:aws:iam::123456789101:role/regional-user", + "endpoint": "https://bucket.vpce-1a2b3c4d-5e6f.s3.us-east-1.vpce.amazonaws.com", + "region": "us-east-1" + }, + { + "user": "test.*", + "iamRole": "arn:aws:iam::123456789101:role/test_users" + }, + { + "group": "finance", + "iamRole": "arn:aws:iam::123456789101:role/finance_users" + }, + { + "iamRole": "arn:aws:iam::123456789101:role/default" + } + ] +} +``` + +:::{list-table} Security mapping properties +:header-rows: 1 + +* - Property name + - Description +* - `s3.security-mapping.enabled` + - Activate the security mapping feature. Defaults to `false`. + Must be set to `true` for all other properties be used. +* - `s3.security-mapping.config-file` + - Path to the JSON configuration file containing security mappings. +* - `s3.security-mapping.config-uri` + - HTTP endpoint URI containing security mappings. +* - `s3.security-mapping.json-pointer` + - A JSON pointer (RFC 6901) to mappings inside the JSON retrieved from the + configuration file or HTTP endpoint. The default is the root of the document. +* - `s3.security-mapping.iam-role-credential-name` + - The name of the *extra credential* used to provide the IAM role. +* - `s3.security-mapping.kms-key-id-credential-name` + - The name of the *extra credential* used to provide the KMS-managed key ID. +* - `s3.security-mapping.sse-customer-key-credential-name` + - The name of the *extra credential* used to provide the server-side encryption with customer-provided keys (SSE-C). +* - `s3.security-mapping.refresh-period` + - How often to refresh the security mapping configuration, specified as a + {ref}`prop-type-duration`. By default, the configuration is not refreshed. +* - `s3.security-mapping.colon-replacement` + - The character or characters to be used instead of a colon character + when specifying an IAM role name as an extra credential. + Any instances of this replacement value in the extra credential value + are converted to a colon. + Choose a value not used in any of your IAM ARNs. +::: + + (iceberg-glue-catalog)= ### Iceberg-specific Glue catalog configuration properties diff --git a/lib/trino-filesystem-s3/pom.xml b/lib/trino-filesystem-s3/pom.xml index 77fee4f9864a..235f2afe4193 100644 --- a/lib/trino-filesystem-s3/pom.xml +++ b/lib/trino-filesystem-s3/pom.xml @@ -23,6 +23,11 @@ jackson-annotations + + com.fasterxml.jackson.core + jackson-core + + com.google.errorprone error_prone_annotations @@ -83,21 +88,14 @@ io.trino trino-filesystem - io.trino - trino-memory-context + trino-iam-aws io.trino - trino-plugin-toolkit - - - io.airlift - bootstrap - - + trino-memory-context diff --git a/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMapping.java b/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMapping.java index 064040d06295..b755452b6f04 100644 --- a/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMapping.java +++ b/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMapping.java @@ -18,11 +18,9 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; import io.trino.filesystem.Location; +import io.trino.iam.aws.IAMSecurityMapping; import io.trino.spi.security.ConnectorIdentity; -import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; -import software.amazon.awssdk.auth.credentials.AwsCredentials; -import java.util.Collection; import java.util.List; import java.util.Optional; import java.util.Set; @@ -33,21 +31,13 @@ import static java.util.Objects.requireNonNull; public final class S3SecurityMapping + extends IAMSecurityMapping { - private final Predicate user; - private final Predicate> group; private final Predicate prefix; - private final Optional iamRole; - private final Optional roleSessionName; - private final Set allowedIamRoles; private final Optional kmsKeyId; private final Set allowedKmsKeyIds; private final Optional sseCustomerKey; private final Set allowedSseCustomerKeys; - private final Optional credentials; - private final boolean useClusterDefault; - private final Optional endpoint; - private final Optional region; @JsonCreator public S3SecurityMapping( @@ -67,25 +57,14 @@ public S3SecurityMapping( @JsonProperty("endpoint") Optional endpoint, @JsonProperty("region") Optional region) { - this.user = user - .map(S3SecurityMapping::toPredicate) - .orElse(_ -> true); - this.group = group - .map(S3SecurityMapping::toPredicate) - .map(S3SecurityMapping::anyMatch) - .orElse(_ -> true); + super(user, group, iamRole, roleSessionName, allowedIamRoles, accessKey, secretKey, useClusterDefault, endpoint, region); + this.prefix = prefix .map(Location::of) .map(S3Location::new) .map(S3SecurityMapping::prefixPredicate) .orElse(_ -> true); - this.iamRole = requireNonNull(iamRole, "iamRole is null"); - this.roleSessionName = requireNonNull(roleSessionName, "roleSessionName is null"); - checkArgument(roleSessionName.isEmpty() || iamRole.isPresent(), "iamRole must be provided when roleSessionName is provided"); - - this.allowedIamRoles = ImmutableSet.copyOf(allowedIamRoles.orElse(ImmutableList.of())); - this.kmsKeyId = requireNonNull(kmsKeyId, "kmsKeyId is null"); this.allowedKmsKeyIds = ImmutableSet.copyOf(allowedKmsKeyIds.orElse(ImmutableList.of())); @@ -94,43 +73,14 @@ public S3SecurityMapping( this.allowedSseCustomerKeys = allowedSseCustomerKeys.map(ImmutableSet::copyOf).orElse(ImmutableSet.of()); - requireNonNull(accessKey, "accessKey is null"); - requireNonNull(secretKey, "secretKey is null"); - checkArgument(accessKey.isPresent() == secretKey.isPresent(), "accessKey and secretKey must be provided together"); - this.credentials = accessKey.map(access -> AwsBasicCredentials.create(access, secretKey.get())); - - this.useClusterDefault = useClusterDefault.orElse(false); - boolean roleOrCredentialsArePresent = !this.allowedIamRoles.isEmpty() || iamRole.isPresent() || credentials.isPresent(); - checkArgument(this.useClusterDefault != roleOrCredentialsArePresent, "must either allow useClusterDefault role or provide role and/or credentials"); - checkArgument(!this.useClusterDefault || this.kmsKeyId.isEmpty(), "KMS key ID cannot be provided together with useClusterDefault"); checkArgument(!this.useClusterDefault || this.sseCustomerKey.isEmpty(), "SSE Customer key cannot be provided together with useClusterDefault"); checkArgument(this.kmsKeyId.isEmpty() || this.sseCustomerKey.isEmpty(), "SSE Customer key cannot be provided together with KMS key ID"); - - this.endpoint = requireNonNull(endpoint, "endpoint is null"); - this.region = requireNonNull(region, "region is null"); } boolean matches(ConnectorIdentity identity, S3Location location) { - return user.test(identity.getUser()) && - group.test(identity.getGroups()) && - prefix.test(location); - } - - public Optional iamRole() - { - return iamRole; - } - - public Optional roleSessionName() - { - return roleSessionName; - } - - public Set allowedIamRoles() - { - return allowedIamRoles; + return super.matches(identity) && prefix.test(location); } public Optional kmsKeyId() @@ -153,39 +103,9 @@ public Set allowedSseCustomerKeys() return allowedSseCustomerKeys; } - public Optional credentials() - { - return credentials; - } - - public boolean useClusterDefault() - { - return useClusterDefault; - } - - public Optional endpoint() - { - return endpoint; - } - - public Optional region() - { - return region; - } - private static Predicate prefixPredicate(S3Location prefix) { return value -> prefix.bucket().equals(value.bucket()) && value.key().startsWith(prefix.key()); } - - private static Predicate toPredicate(Pattern pattern) - { - return value -> pattern.matcher(value).matches(); - } - - private static Predicate> anyMatch(Predicate predicate) - { - return values -> values.stream().anyMatch(predicate); - } } diff --git a/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingConfig.java b/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingConfig.java index 8bf2b8cbe499..9774d9f2fc9f 100644 --- a/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingConfig.java +++ b/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingConfig.java @@ -15,30 +15,19 @@ import io.airlift.configuration.Config; import io.airlift.configuration.ConfigDescription; -import io.airlift.configuration.validation.FileExists; import io.airlift.units.Duration; +import io.trino.iam.aws.IAMSecurityMappingConfig; import jakarta.validation.constraints.AssertTrue; -import jakarta.validation.constraints.NotNull; import java.io.File; import java.net.URI; import java.util.Optional; public class S3SecurityMappingConfig + extends IAMSecurityMappingConfig { - private File configFile; - private URI configUri; - private String jsonPointer = ""; - private String roleCredentialName; private String kmsKeyIdCredentialName; private String sseCustomerKeyCredentialName; - private Duration refreshPeriod; - private String colonReplacement; - - public Optional<@FileExists File> getConfigFile() - { - return Optional.ofNullable(configFile); - } @Config("s3.security-mapping.config-file") @ConfigDescription("Path to the JSON security mappings file") @@ -48,11 +37,6 @@ public S3SecurityMappingConfig setConfigFile(File configFile) return this; } - public Optional getConfigUri() - { - return Optional.ofNullable(configUri); - } - @Config("s3.security-mapping.config-uri") @ConfigDescription("HTTP URI of the JSON security mappings") public S3SecurityMappingConfig setConfigUri(URI configUri) @@ -61,12 +45,6 @@ public S3SecurityMappingConfig setConfigUri(URI configUri) return this; } - @NotNull - public String getJsonPointer() - { - return jsonPointer; - } - @Config("s3.security-mapping.json-pointer") @ConfigDescription("JSON pointer (RFC 6901) to mappings inside JSON config") public S3SecurityMappingConfig setJsonPointer(String jsonPointer) @@ -75,11 +53,6 @@ public S3SecurityMappingConfig setJsonPointer(String jsonPointer) return this; } - public Optional getRoleCredentialName() - { - return Optional.ofNullable(roleCredentialName); - } - @Config("s3.security-mapping.iam-role-credential-name") @ConfigDescription("Name of the extra credential used to provide IAM role") public S3SecurityMappingConfig setRoleCredentialName(String roleCredentialName) @@ -114,11 +87,6 @@ public S3SecurityMappingConfig setSseCustomerKeyCredentialName(String sseCustome return this; } - public Optional getRefreshPeriod() - { - return Optional.ofNullable(refreshPeriod); - } - @Config("s3.security-mapping.refresh-period") @ConfigDescription("How often to refresh the security mapping configuration") public S3SecurityMappingConfig setRefreshPeriod(Duration refreshPeriod) @@ -127,11 +95,6 @@ public S3SecurityMappingConfig setRefreshPeriod(Duration refreshPeriod) return this; } - public Optional getColonReplacement() - { - return Optional.ofNullable(colonReplacement); - } - @Config("s3.security-mapping.colon-replacement") @ConfigDescription("Value used in place of colon for IAM role name in extra credentials") public S3SecurityMappingConfig setColonReplacement(String colonReplacement) @@ -140,9 +103,10 @@ public S3SecurityMappingConfig setColonReplacement(String colonReplacement) return this; } + @Override @AssertTrue(message = "Exactly one of s3.security-mapping.config-file or s3.security-mapping.config-uri must be set") public boolean validateMappingsConfig() { - return (configFile == null) != (configUri == null); + return super.validateMappingsConfig(); } } diff --git a/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingProvider.java b/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingProvider.java index 008fb835f6bf..bf7764ea756f 100644 --- a/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingProvider.java +++ b/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingProvider.java @@ -14,50 +14,29 @@ package io.trino.filesystem.s3; import com.google.inject.Inject; -import io.airlift.units.Duration; import io.trino.filesystem.Location; +import io.trino.iam.aws.IAMSecurityMappingProvider; import io.trino.spi.security.AccessDeniedException; import io.trino.spi.security.ConnectorIdentity; import java.util.Optional; import java.util.function.Supplier; -import static com.google.common.base.Suppliers.memoize; -import static com.google.common.base.Suppliers.memoizeWithExpiration; -import static com.google.common.base.Verify.verify; import static java.util.Objects.requireNonNull; -import static java.util.concurrent.TimeUnit.MILLISECONDS; final class S3SecurityMappingProvider + extends IAMSecurityMappingProvider { - private final Supplier mappingsProvider; - private final Optional roleCredentialName; private final Optional kmsKeyIdCredentialName; private final Optional sseCustomerKeyCredentialName; - private final Optional colonReplacement; @Inject - public S3SecurityMappingProvider(S3SecurityMappingConfig config, Supplier mappingsProvider) + public S3SecurityMappingProvider(S3SecurityMappingConfig config, Supplier mappings) { - this(mappingsProvider(mappingsProvider, config.getRefreshPeriod()), - config.getRoleCredentialName(), - config.getKmsKeyIdCredentialName(), - config.getSseCustomerKeyCredentialName(), - config.getColonReplacement()); - } + super(config, mappings); - public S3SecurityMappingProvider( - Supplier mappingsProvider, - Optional roleCredentialName, - Optional kmsKeyIdCredentialName, - Optional sseCustomerKeyCredentialName, - Optional colonReplacement) - { - this.mappingsProvider = requireNonNull(mappingsProvider, "mappingsProvider is null"); - this.roleCredentialName = requireNonNull(roleCredentialName, "roleCredentialName is null"); - this.kmsKeyIdCredentialName = requireNonNull(kmsKeyIdCredentialName, "kmsKeyIdCredentialName is null"); - this.sseCustomerKeyCredentialName = requireNonNull(sseCustomerKeyCredentialName, "customerKeyCredentialName is null"); - this.colonReplacement = requireNonNull(colonReplacement, "colonReplacement is null"); + this.kmsKeyIdCredentialName = requireNonNull(config.getKmsKeyIdCredentialName(), "kmsKeyIdCredentialName is null"); + this.sseCustomerKeyCredentialName = requireNonNull(config.getSseCustomerKeyCredentialName(), "sseCustomerKeyCredentialName is null"); } public Optional getMapping(ConnectorIdentity identity, Location location) @@ -79,38 +58,6 @@ public Optional getMapping(ConnectorIdentity identity, mapping.region())); } - private Optional selectRole(S3SecurityMapping mapping, ConnectorIdentity identity) - { - Optional optionalSelected = getRoleFromExtraCredential(identity); - - if (optionalSelected.isEmpty()) { - if (!mapping.allowedIamRoles().isEmpty() && mapping.iamRole().isEmpty()) { - throw new AccessDeniedException("No S3 role selected and mapping has no default role"); - } - verify(mapping.iamRole().isPresent() || mapping.credentials().isPresent(), "mapping must have role or credential"); - return mapping.iamRole(); - } - - String selected = optionalSelected.get(); - - // selected role must match default or be allowed - if (!selected.equals(mapping.iamRole().orElse(null)) && - !mapping.allowedIamRoles().contains(selected)) { - throw new AccessDeniedException("Selected S3 role is not allowed: " + selected); - } - - return optionalSelected; - } - - private Optional getRoleFromExtraCredential(ConnectorIdentity identity) - { - return roleCredentialName - .map(name -> identity.getExtraCredentials().get(name)) - .map(role -> colonReplacement - .map(replacement -> role.replace(replacement, ":")) - .orElse(role)); - } - private Optional selectKmsKeyId(S3SecurityMapping mapping, ConnectorIdentity identity) { Optional userSelected = getKmsKeyIdFromExtraCredential(identity); @@ -161,11 +108,4 @@ private Optional getSseCustomerKeyFromExtraCredential(ConnectorIdentity { return sseCustomerKeyCredentialName.map(name -> identity.getExtraCredentials().get(name)); } - - private static Supplier mappingsProvider(Supplier supplier, Optional refreshPeriod) - { - return refreshPeriod - .map(refresh -> memoizeWithExpiration(supplier::get, refresh.toMillis(), MILLISECONDS)) - .orElseGet(() -> memoize(supplier::get)); - } } diff --git a/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappings.java b/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappings.java index 22ecb7945e03..e9b63b2d8724 100644 --- a/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappings.java +++ b/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappings.java @@ -15,27 +15,24 @@ import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; -import com.google.common.collect.ImmutableList; +import io.trino.iam.aws.IAMSecurityMappings; import io.trino.spi.security.ConnectorIdentity; import java.util.List; import java.util.Optional; -import static java.util.Objects.requireNonNull; - public final class S3SecurityMappings + extends IAMSecurityMappings { - private final List mappings; - @JsonCreator public S3SecurityMappings(@JsonProperty("mappings") List mappings) { - this.mappings = ImmutableList.copyOf(requireNonNull(mappings, "mappings is null")); + super(mappings); } Optional getMapping(ConnectorIdentity identity, S3Location location) { - return mappings.stream() + return getMappings().stream() .filter(mapping -> mapping.matches(identity, location)) .findFirst(); } diff --git a/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingsFileSource.java b/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingsFileSource.java index aa96fdc26912..580d310119b0 100644 --- a/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingsFileSource.java +++ b/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingsFileSource.java @@ -13,29 +13,19 @@ */ package io.trino.filesystem.s3; +import com.fasterxml.jackson.core.type.TypeReference; import com.google.inject.Inject; +import io.trino.iam.aws.IAMSecurityMappingsFileSource; -import java.io.File; import java.util.function.Supplier; -import static io.trino.plugin.base.util.JsonUtils.parseJson; - class S3SecurityMappingsFileSource + extends IAMSecurityMappingsFileSource implements Supplier { - private final File configFile; - private final String jsonPointer; - @Inject public S3SecurityMappingsFileSource(S3SecurityMappingConfig config) { - this.configFile = config.getConfigFile().orElseThrow(); - this.jsonPointer = config.getJsonPointer(); - } - - @Override - public S3SecurityMappings get() - { - return parseJson(configFile.toPath(), jsonPointer, S3SecurityMappings.class); + super(config, new TypeReference() {}); } } diff --git a/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingsUriSource.java b/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingsUriSource.java index e77a17348b77..4063227c8fb2 100644 --- a/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingsUriSource.java +++ b/lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3SecurityMappingsUriSource.java @@ -13,52 +13,21 @@ */ package io.trino.filesystem.s3; -import com.google.common.annotations.VisibleForTesting; +import com.fasterxml.jackson.core.type.TypeReference; import com.google.inject.Inject; import io.airlift.http.client.HttpClient; -import io.airlift.http.client.HttpStatus; -import io.airlift.http.client.Request; -import io.airlift.http.client.StringResponseHandler.StringResponse; import io.trino.filesystem.s3.S3FileSystemModule.ForS3SecurityMapping; +import io.trino.iam.aws.IAMSecurityMappingsUriSource; -import java.net.URI; import java.util.function.Supplier; -import static io.airlift.http.client.Request.Builder.prepareGet; -import static io.airlift.http.client.StringResponseHandler.createStringResponseHandler; -import static io.trino.plugin.base.util.JsonUtils.parseJson; -import static java.util.Objects.requireNonNull; - class S3SecurityMappingsUriSource + extends IAMSecurityMappingsUriSource implements Supplier { - private final URI configUri; - private final HttpClient httpClient; - private final String jsonPointer; - @Inject public S3SecurityMappingsUriSource(S3SecurityMappingConfig config, @ForS3SecurityMapping HttpClient httpClient) { - this.configUri = config.getConfigUri().orElseThrow(); - this.httpClient = requireNonNull(httpClient, "httpClient is null"); - this.jsonPointer = config.getJsonPointer(); - } - - @Override - public S3SecurityMappings get() - { - return parseJson(getRawJsonString(), jsonPointer, S3SecurityMappings.class); - } - - @VisibleForTesting - String getRawJsonString() - { - Request request = prepareGet().setUri(configUri).build(); - StringResponse response = httpClient.execute(request, createStringResponseHandler()); - int status = response.getStatusCode(); - if (status != HttpStatus.OK.code()) { - throw new RuntimeException("Request to '%s' returned unexpected status code: %s".formatted(configUri, status)); - } - return response.getBody(); + super(config, httpClient, new TypeReference() {}); } } diff --git a/lib/trino-filesystem-s3/src/test/java/io/trino/filesystem/s3/TestS3SecurityMapping.java b/lib/trino-filesystem-s3/src/test/java/io/trino/filesystem/s3/TestS3SecurityMapping.java index b3b0809ec581..80722ced1e8f 100644 --- a/lib/trino-filesystem-s3/src/test/java/io/trino/filesystem/s3/TestS3SecurityMapping.java +++ b/lib/trino-filesystem-s3/src/test/java/io/trino/filesystem/s3/TestS3SecurityMapping.java @@ -149,7 +149,7 @@ public void testMapping() assertMappingFails( provider, path("s3://bar/test"), - "No S3 role selected and mapping has no default role"); + "No IAM role selected and mapping has no default role"); // matches prefix and user selected one of the allowed roles assertMapping( @@ -164,7 +164,7 @@ public void testMapping() path("s3://bar/test") .withUser("bob") .withExtraCredentialIamRole("bogus"), - "Selected S3 role is not allowed: bogus"); + "Selected IAM role is not allowed: bogus"); // verify that colon replacement works String roleWithoutColon = "arn#aws#iam##123456789101#role/allow_bucket_2"; @@ -181,12 +181,6 @@ public void testMapping() path("s3://bar/abc/data/test.csv"), MappingResult.iamRole("arn:aws:iam::123456789101:role/allow_path")); - // matches empty rule at the end -- default role used - assertMapping( - provider, - MappingSelector.empty(), - MappingResult.iamRole("arn:aws:iam::123456789101:role/default")); - // matches prefix -- default role used assertMapping( provider, @@ -206,193 +200,6 @@ public void testMapping() path("s3://xyz/bar") .withExtraCredentialIamRole("arn:aws:iam::123456789101:role/allow_bar"), MappingResult.iamRole("arn:aws:iam::123456789101:role/allow_bar")); - - // matches user -- default role used - assertMapping( - provider, - MappingSelector.empty() - .withUser("alice"), - MappingResult.iamRole("alice_role")); - - // matches user and user selected default role - assertMapping( - provider, - MappingSelector.empty() - .withUser("alice") - .withExtraCredentialIamRole("alice_role"), - MappingResult.iamRole("alice_role")); - - // matches user and selected role not allowed - assertMappingFails( - provider, - MappingSelector.empty() - .withUser("alice") - .withExtraCredentialIamRole("bogus"), - "Selected S3 role is not allowed: bogus"); - - // verify that the first matching rule is used - // matches prefix earlier in the file and selected role not allowed - assertMappingFails( - provider, - path("s3://bar/test") - .withUser("alice") - .withExtraCredentialIamRole("alice_role"), - "Selected S3 role is not allowed: alice_role"); - - // matches user regex -- default role used - assertMapping( - provider, - MappingSelector.empty() - .withUser("bob"), - MappingResult.iamRole("bob_and_charlie_role")); - - // matches group -- default role used - assertMapping( - provider, - MappingSelector.empty() - .withGroups("finance"), - MappingResult.iamRole("finance_role")); - - // matches group regex -- default role used - assertMapping( - provider, - MappingSelector.empty() - .withGroups("eng"), - MappingResult.iamRole("hr_and_eng_group")); - - // verify that all constraints must match - // matches user but not group -- uses empty mapping at the end - assertMapping( - provider, - MappingSelector.empty() - .withUser("danny"), - MappingResult.iamRole("arn:aws:iam::123456789101:role/default")); - - // matches group but not user -- uses empty mapping at the end - assertMapping( - provider, - MappingSelector.empty() - .withGroups("hq"), - MappingResult.iamRole("arn:aws:iam::123456789101:role/default")); - - // matches user and group - assertMapping( - provider, - MappingSelector.empty() - .withUser("danny") - .withGroups("hq"), - MappingResult.iamRole("danny_hq_role")); - - // matches prefix -- mapping provides credentials and endpoint - assertMapping( - provider, - path("s3://endpointbucket/bar"), - credentials("AKIAxxxaccess", "iXbXxxxsecret") - .withEndpoint("http://localhost:7753")); - - // matches prefix -- mapping provides credentials and region - assertMapping( - provider, - path("s3://regionalbucket/bar"), - credentials("AKIAxxxaccess", "iXbXxxxsecret") - .withRegion("us-west-2")); - - // matches role session name - assertMapping( - provider, - path("s3://somebucket/"), - MappingResult.iamRole("arn:aws:iam::1234567891012:role/default") - .withRoleSessionName("iam-trino-session")); - } - - @Test - public void testMappingWithFallbackToClusterDefault() - { - S3SecurityMappingConfig mappingConfig = new S3SecurityMappingConfig() - .setConfigFile(getResourceFile("security-mapping-with-fallback-to-cluster-default.json")); - - var provider = new S3SecurityMappingProvider(mappingConfig, new S3SecurityMappingsFileSource(mappingConfig)); - - // matches prefix -- uses the role from the mapping - assertMapping( - provider, - path("s3://bar/abc/data/test.csv"), - MappingResult.iamRole("arn:aws:iam::123456789101:role/allow_path")); - - // doesn't match any rule except default rule at the end - assertThat(getMapping(provider, MappingSelector.empty())).isEmpty(); - } - - @Test - public void testMappingWithoutFallback() - { - S3SecurityMappingConfig mappingConfig = new S3SecurityMappingConfig() - .setConfigFile(getResourceFile("security-mapping-without-fallback.json")); - - var provider = new S3SecurityMappingProvider(mappingConfig, new S3SecurityMappingsFileSource(mappingConfig)); - - // matches prefix - return role from the mapping - assertMapping( - provider, - path("s3://bar/abc/data/test.csv"), - MappingResult.iamRole("arn:aws:iam::123456789101:role/allow_path")); - - // doesn't match any rule - assertMappingFails( - provider, - MappingSelector.empty(), - "No matching S3 security mapping"); - } - - @Test - public void testMappingWithoutRoleCredentialsFallbackShouldFail() - { - assertThatThrownBy(() -> - new S3SecurityMapping( - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty())) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("must either allow useClusterDefault role or provide role and/or credentials"); - } - - @Test - public void testMappingWithRoleAndFallbackShouldFail() - { - Optional iamRole = Optional.of("arn:aws:iam::123456789101:role/allow_path"); - Optional useClusterDefault = Optional.of(true); - - assertThatThrownBy(() -> - new S3SecurityMapping( - Optional.empty(), - Optional.empty(), - Optional.empty(), - iamRole, - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - useClusterDefault, - Optional.empty(), - Optional.empty())) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("must either allow useClusterDefault role or provide role and/or credentials"); } @Test @@ -476,32 +283,6 @@ public void testMappingWithSseCustomerAndKMSKeysShouldFail() .hasMessage("SSE Customer key cannot be provided together with KMS key ID"); } - @Test - public void testMappingWithRoleSessionNameWithoutIamRoleShouldFail() - { - Optional roleSessionName = Optional.of("iam-trino-session"); - - assertThatThrownBy(() -> - new S3SecurityMapping( - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - roleSessionName, - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty())) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("iamRole must be provided when roleSessionName is provided"); - } - private File getResourceFile(String name) { return new File(getResource(getClass(), name).getFile()); @@ -589,11 +370,6 @@ public MappingSelector withUser(String user) return new MappingSelector(user, groups, location, extraCredentialIamRole, extraCredentialKmsKeyId, extraCredentialCustomerKey); } - public MappingSelector withGroups(String... groups) - { - return new MappingSelector(user, ImmutableSet.copyOf(groups), location, extraCredentialIamRole, extraCredentialKmsKeyId, extraCredentialCustomerKey); - } - public ConnectorIdentity identity() { Map extraCredentials = new HashMap<>(); @@ -665,11 +441,6 @@ private MappingResult( this.region = requireNonNull(region, "region is null"); } - public MappingResult withEndpoint(String endpoint) - { - return new MappingResult(accessKey, secretKey, iamRole, Optional.empty(), kmsKeyId, sseCustomerKey, Optional.of(endpoint), region); - } - public MappingResult withKmsKeyId(String kmsKeyId) { return new MappingResult(accessKey, secretKey, iamRole, Optional.empty(), Optional.of(kmsKeyId), Optional.empty(), endpoint, region); @@ -680,16 +451,6 @@ public MappingResult withSseCustomerKey(String customerKey) return new MappingResult(accessKey, secretKey, iamRole, Optional.empty(), Optional.empty(), Optional.of(customerKey), endpoint, region); } - public MappingResult withRegion(String region) - { - return new MappingResult(accessKey, secretKey, iamRole, Optional.empty(), kmsKeyId, sseCustomerKey, endpoint, Optional.of(region)); - } - - public MappingResult withRoleSessionName(String roleSessionName) - { - return new MappingResult(accessKey, secretKey, iamRole, Optional.of(roleSessionName), kmsKeyId, sseCustomerKey, Optional.empty(), region); - } - public Optional accessKey() { return accessKey; diff --git a/lib/trino-filesystem-s3/src/test/resources/io/trino/filesystem/s3/security-mapping-with-fallback-to-cluster-default.json b/lib/trino-filesystem-s3/src/test/resources/io/trino/filesystem/s3/security-mapping-with-fallback-to-cluster-default.json deleted file mode 100644 index 45b98ed39849..000000000000 --- a/lib/trino-filesystem-s3/src/test/resources/io/trino/filesystem/s3/security-mapping-with-fallback-to-cluster-default.json +++ /dev/null @@ -1,11 +0,0 @@ -{ - "mappings": [ - { - "prefix": "s3://bar/abc", - "iamRole": "arn:aws:iam::123456789101:role/allow_path" - }, - { - "useClusterDefault": "true" - } - ] -} diff --git a/lib/trino-filesystem-s3/src/test/resources/io/trino/filesystem/s3/security-mapping-without-fallback.json b/lib/trino-filesystem-s3/src/test/resources/io/trino/filesystem/s3/security-mapping-without-fallback.json deleted file mode 100644 index 6700f8b1dec0..000000000000 --- a/lib/trino-filesystem-s3/src/test/resources/io/trino/filesystem/s3/security-mapping-without-fallback.json +++ /dev/null @@ -1,8 +0,0 @@ -{ - "mappings": [ - { - "prefix": "s3://bar/abc", - "iamRole": "arn:aws:iam::123456789101:role/allow_path" - } - ] -} diff --git a/lib/trino-filesystem-s3/src/test/resources/io/trino/filesystem/s3/security-mapping.json b/lib/trino-filesystem-s3/src/test/resources/io/trino/filesystem/s3/security-mapping.json index 44e0d58d4a59..d727ba5f3d02 100644 --- a/lib/trino-filesystem-s3/src/test/resources/io/trino/filesystem/s3/security-mapping.json +++ b/lib/trino-filesystem-s3/src/test/resources/io/trino/filesystem/s3/security-mapping.json @@ -1,106 +1,99 @@ { - "mappings": [ - { - "prefix": "s3://bar/abc", - "iamRole": "arn:aws:iam::123456789101:role/allow_path" - }, - { - "prefix": "s3://bar/", - "allowedIamRoles": [ - "arn:aws:iam::123456789101:role/allow_bucket_1", - "arn:aws:iam::123456789101:role/allow_bucket_2", - "arn:aws:iam::123456789101:role/allow_bucket_3" - ] - }, - { - "prefix": "s3://xyz/", - "iamRole": "arn:aws:iam::123456789101:role/allow_default", - "allowedIamRoles": [ - "arn:aws:iam::123456789101:role/allow_foo", - "arn:aws:iam::123456789101:role/allow_bar" - ] - }, - { - "prefix": "s3://foo/", - "accessKey": "AKIAxxxaccess", - "secretKey": "iXbXxxxsecret", - "kmsKeyId": "kmsKey_10", - "allowedKmsKeyIds": ["kmsKey_11"] - }, - { - "prefix": "s3://foo_all_keys_allowed/", - "accessKey": "AKIAxxxaccess", - "secretKey": "iXbXxxxsecret", - "kmsKeyId": "kmsKey_10", - "allowedKmsKeyIds": ["*"] - }, - { - "prefix": "s3://foo_no_default_key/", - "accessKey": "AKIAxxxaccess", - "secretKey": "iXbXxxxsecret", - "allowedKmsKeyIds": ["kmsKey_11", "kmsKey_12"] - }, - { - "prefix": "s3://baz/", - "accessKey": "AKIAxxxaccess", - "secretKey": "iXbXxxxsecret", - "sseCustomerKey": "customerKey_10", - "allowedSseCustomerKeys": ["customerKey_11"] - }, - { - "prefix": "s3://baz_all_customer_keys_allowed/", - "accessKey": "AKIAxxxaccess", - "secretKey": "iXbXxxxsecret", - "sseCustomerKey": "customerKey_10", - "allowedSseCustomerKeys": ["*"] - }, - { - "prefix": "s3://baz_no_customer_default_key/", - "accessKey": "AKIAxxxaccess", - "secretKey": "iXbXxxxsecret", - "allowedSseCustomerKeys": ["customerKey_11", "customerKey_12"] - }, - { - "user": "alice", - "iamRole": "alice_role" - }, - { - "user": "bob|charlie", - "iamRole": "bob_and_charlie_role" - }, - { - "group": "finance", - "iamRole": "finance_role" - }, - { - "group": "hr|eng", - "iamRole": "hr_and_eng_group" - }, - { - "user": "danny", - "group": "hq", - "iamRole": "danny_hq_role" - }, - { - "prefix": "s3://endpointbucket/", - "accessKey": "AKIAxxxaccess", - "secretKey": "iXbXxxxsecret", - "endpoint": "http://localhost:7753" - }, - { - "prefix": "s3://regionalbucket/", - "accessKey": "AKIAxxxaccess", - "secretKey": "iXbXxxxsecret", - "region": "us-west-2" - }, - { - "prefix": "s3://somebucket/", - "iamRole": "arn:aws:iam::1234567891012:role/default", - "roleSessionName": "iam-trino-session" - }, - { - "useClusterDefault": "false", - "iamRole": "arn:aws:iam::123456789101:role/default" - } - ] + "mappings": [ + { + "prefix": "s3://bar/abc", + "iamRole": "arn:aws:iam::123456789101:role/allow_path" + }, + { + "prefix": "s3://bar/", + "allowedIamRoles": [ + "arn:aws:iam::123456789101:role/allow_bucket_1", + "arn:aws:iam::123456789101:role/allow_bucket_2", + "arn:aws:iam::123456789101:role/allow_bucket_3" + ] + }, + { + "prefix": "s3://xyz/", + "iamRole": "arn:aws:iam::123456789101:role/allow_default", + "allowedIamRoles": [ + "arn:aws:iam::123456789101:role/allow_foo", + "arn:aws:iam::123456789101:role/allow_bar" + ] + }, + { + "prefix": "s3://foo/", + "accessKey": "AKIAxxxaccess", + "secretKey": "iXbXxxxsecret", + "kmsKeyId": "kmsKey_10", + "allowedKmsKeyIds": [ + "kmsKey_11" + ] + }, + { + "prefix": "s3://foo_all_keys_allowed/", + "accessKey": "AKIAxxxaccess", + "secretKey": "iXbXxxxsecret", + "kmsKeyId": "kmsKey_10", + "allowedKmsKeyIds": [ + "*" + ] + }, + { + "prefix": "s3://foo_no_default_key/", + "accessKey": "AKIAxxxaccess", + "secretKey": "iXbXxxxsecret", + "allowedKmsKeyIds": [ + "kmsKey_11", + "kmsKey_12" + ] + }, + { + "prefix": "s3://baz/", + "accessKey": "AKIAxxxaccess", + "secretKey": "iXbXxxxsecret", + "sseCustomerKey": "customerKey_10", + "allowedSseCustomerKeys": [ + "customerKey_11" + ] + }, + { + "prefix": "s3://baz_all_customer_keys_allowed/", + "accessKey": "AKIAxxxaccess", + "secretKey": "iXbXxxxsecret", + "sseCustomerKey": "customerKey_10", + "allowedSseCustomerKeys": [ + "*" + ] + }, + { + "prefix": "s3://baz_no_customer_default_key/", + "accessKey": "AKIAxxxaccess", + "secretKey": "iXbXxxxsecret", + "allowedSseCustomerKeys": [ + "customerKey_11", + "customerKey_12" + ] + }, + { + "prefix": "s3://endpointbucket/", + "accessKey": "AKIAxxxaccess", + "secretKey": "iXbXxxxsecret", + "endpoint": "http://localhost:7753" + }, + { + "prefix": "s3://regionalbucket/", + "accessKey": "AKIAxxxaccess", + "secretKey": "iXbXxxxsecret", + "region": "us-west-2" + }, + { + "prefix": "s3://somebucket/", + "iamRole": "arn:aws:iam::1234567891012:role/default", + "roleSessionName": "iam-trino-session" + }, + { + "useClusterDefault": "false", + "iamRole": "arn:aws:iam::123456789101:role/default" + } + ] } diff --git a/lib/trino-iam-aws/pom.xml b/lib/trino-iam-aws/pom.xml new file mode 100644 index 000000000000..11b54ff66e8e --- /dev/null +++ b/lib/trino-iam-aws/pom.xml @@ -0,0 +1,150 @@ + + + 4.0.0 + + + io.trino + trino-root + 478-SNAPSHOT + ../../pom.xml + + + trino-iam-aws + ${project.artifactId} + Trino AWS IAM Security Mapping Library + + + true + + + + + + com.fasterxml.jackson.core + jackson-annotations + + + + com.fasterxml.jackson.core + jackson-core + + + + com.google.guava + guava + + + + com.google.inject + guice + classes + + + + io.airlift + configuration + + + + io.airlift + http-client + + + + io.airlift + units + + + + io.trino + trino-plugin-toolkit + + + io.airlift + log-manager + + + + + + io.trino + trino-spi + + + + jakarta.validation + jakarta.validation-api + + + + software.amazon.awssdk + auth + + + + software.amazon.awssdk + identity-spi + + + + io.airlift + junit-extensions + test + + + + org.assertj + assertj-core + test + + + org.junit.jupiter + junit-jupiter-api + test + + + + org.junit.jupiter + junit-jupiter-engine + test + + + + + + + + org.apache.maven.plugins + maven-dependency-plugin + + + software.amazon.awssdk:identity-spi + + + + + + + + + default + + true + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + **/Test*.java + + + + + + + + + diff --git a/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMapping.java b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMapping.java new file mode 100644 index 000000000000..ae7c8f36cccf --- /dev/null +++ b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMapping.java @@ -0,0 +1,148 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.iam.aws; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import com.google.inject.BindingAnnotation; +import io.trino.spi.security.ConnectorIdentity; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentials; + +import java.lang.annotation.Retention; +import java.lang.annotation.Target; +import java.util.Collection; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.function.Predicate; +import java.util.regex.Pattern; + +import static com.google.common.base.Preconditions.checkArgument; +import static java.lang.annotation.ElementType.FIELD; +import static java.lang.annotation.ElementType.METHOD; +import static java.lang.annotation.ElementType.PARAMETER; +import static java.lang.annotation.RetentionPolicy.RUNTIME; +import static java.util.Objects.requireNonNull; + +@JsonIgnoreProperties(ignoreUnknown = true) +public class IAMSecurityMapping +{ + private final Predicate user; + private final Predicate> group; + private final Optional iamRole; + private final Optional roleSessionName; + private final Set allowedIamRoles; + private final Optional credentials; + protected final boolean useClusterDefault; + private final Optional endpoint; + private final Optional region; + + public IAMSecurityMapping( + @JsonProperty("user") Optional user, + @JsonProperty("group") Optional group, + @JsonProperty("iamRole") Optional iamRole, + @JsonProperty("roleSessionName") Optional roleSessionName, + @JsonProperty("allowedIamRoles") Optional> allowedIamRoles, + @JsonProperty("accessKey") Optional accessKey, + @JsonProperty("secretKey") Optional secretKey, + @JsonProperty("useClusterDefault") Optional useClusterDefault, + @JsonProperty("endpoint") Optional endpoint, + @JsonProperty("region") Optional region) + { + this.user = user + .map(IAMSecurityMapping::toPredicate) + .orElse(_ -> true); + this.group = group + .map(IAMSecurityMapping::toPredicate) + .map(IAMSecurityMapping::anyMatch) + .orElse(_ -> true); + + this.iamRole = requireNonNull(iamRole, "iamRole is null"); + this.roleSessionName = requireNonNull(roleSessionName, "roleSessionName is null"); + checkArgument(roleSessionName.isEmpty() || iamRole.isPresent(), "iamRole must be provided when roleSessionName is provided"); + + this.allowedIamRoles = ImmutableSet.copyOf(allowedIamRoles.orElse(ImmutableList.of())); + + requireNonNull(accessKey, "accessKey is null"); + requireNonNull(secretKey, "secretKey is null"); + checkArgument(accessKey.isPresent() == secretKey.isPresent(), "accessKey and secretKey must be provided together"); + this.credentials = accessKey.map(access -> AwsBasicCredentials.create(access, secretKey.get())); + + this.useClusterDefault = useClusterDefault.orElse(false); + boolean roleOrCredentialsArePresent = !this.allowedIamRoles.isEmpty() || iamRole.isPresent() || credentials.isPresent(); + checkArgument(this.useClusterDefault != roleOrCredentialsArePresent, "must either allow useClusterDefault role or provide role and/or credentials"); + + this.endpoint = requireNonNull(endpoint, "endpoint is null"); + this.region = requireNonNull(region, "region is null"); + } + + protected boolean matches(ConnectorIdentity identity) + { + return user.test(identity.getUser()) && + group.test(identity.getGroups()); + } + + public Optional iamRole() + { + return iamRole; + } + + public Optional roleSessionName() + { + return roleSessionName; + } + + public Set allowedIamRoles() + { + return allowedIamRoles; + } + + public Optional credentials() + { + return credentials; + } + + public boolean useClusterDefault() + { + return useClusterDefault; + } + + public Optional endpoint() + { + return endpoint; + } + + public Optional region() + { + return region; + } + + private static Predicate toPredicate(Pattern pattern) + { + return value -> pattern.matcher(value).matches(); + } + + private static Predicate> anyMatch(Predicate predicate) + { + return values -> values.stream().anyMatch(predicate); + } + + @Retention(RUNTIME) + @Target({FIELD, PARAMETER, METHOD}) + @BindingAnnotation + public @interface ForIAMSecurityMapping {} +} diff --git a/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingConfig.java b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingConfig.java new file mode 100644 index 000000000000..984a169d04c4 --- /dev/null +++ b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingConfig.java @@ -0,0 +1,91 @@ +package io.trino.iam.aws; +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +import io.airlift.configuration.validation.FileExists; +import io.airlift.units.Duration; +import jakarta.validation.constraints.NotNull; + +import java.io.File; +import java.net.URI; +import java.util.Optional; + +public class IAMSecurityMappingConfig +{ + protected File configFile; + protected URI configUri; + protected String jsonPointer = ""; + protected String roleCredentialName; + protected Duration refreshPeriod; + protected String colonReplacement; + + public Optional<@FileExists File> getConfigFile() + { + return Optional.ofNullable(configFile); + } + + public IAMSecurityMappingConfig setConfigFileInternal(File configFile) + { + this.configFile = configFile; + return this; + } + + public Optional getConfigUri() + { + return Optional.ofNullable(configUri); + } + + public IAMSecurityMappingConfig setConfigUriInternal(URI configUri) + { + this.configUri = configUri; + return this; + } + + @NotNull + public String getJsonPointer() + { + return jsonPointer; + } + + public Optional getRoleCredentialName() + { + return Optional.ofNullable(roleCredentialName); + } + + public IAMSecurityMappingConfig setRoleCredentialNameInternal(String roleCredentialName) + { + this.roleCredentialName = roleCredentialName; + return this; + } + + public Optional getRefreshPeriod() + { + return Optional.ofNullable(refreshPeriod); + } + + public Optional getColonReplacement() + { + return Optional.ofNullable(colonReplacement); + } + + public IAMSecurityMappingConfig setColonReplacementInternal(String colonReplacement) + { + this.colonReplacement = colonReplacement; + return this; + } + + public boolean validateMappingsConfig() + { + return (configFile == null) != (configUri == null); + } +} diff --git a/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingProvider.java b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingProvider.java new file mode 100644 index 000000000000..2044950f52b4 --- /dev/null +++ b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingProvider.java @@ -0,0 +1,109 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.iam.aws; + +import com.google.inject.Inject; +import io.airlift.units.Duration; +import io.trino.spi.security.AccessDeniedException; +import io.trino.spi.security.ConnectorIdentity; + +import java.util.Optional; +import java.util.function.Supplier; + +import static com.google.common.base.Suppliers.memoize; +import static com.google.common.base.Suppliers.memoizeWithExpiration; +import static com.google.common.base.Verify.verify; +import static java.util.Objects.requireNonNull; +import static java.util.concurrent.TimeUnit.MILLISECONDS; + +public class IAMSecurityMappingProvider, E extends IAMSecurityMapping, C extends IAMSecurityMappingConfig> +{ + protected final Supplier mappingsProvider; + private final Optional roleCredentialName; + private final Optional colonReplacement; + + @Inject + public IAMSecurityMappingProvider(C config, Supplier mappingsProvider) + { + this(mappingsProvider(mappingsProvider, config.getRefreshPeriod()), + config.getRoleCredentialName(), + config.getColonReplacement()); + } + + public IAMSecurityMappingProvider( + Supplier mappingsProvider, + Optional roleCredentialName, + Optional colonReplacement) + { + this.mappingsProvider = requireNonNull(mappingsProvider, "mappingsProvider is null"); + this.roleCredentialName = requireNonNull(roleCredentialName, "roleCredentialName is null"); + this.colonReplacement = requireNonNull(colonReplacement, "colonReplacement is null"); + } + + public Optional getMapping(ConnectorIdentity identity) + { + IAMSecurityMapping mapping = mappingsProvider.get().getMapping(identity) + .orElseThrow(() -> new AccessDeniedException("No matching IAM security mapping")); + + if (mapping.useClusterDefault()) { + return Optional.empty(); + } + + return Optional.of(new IAMSecurityMappingResult( + mapping.credentials(), + selectRole(mapping, identity), + mapping.roleSessionName().map(name -> name.replace("${USER}", identity.getUser())), + mapping.endpoint(), + mapping.region())); + } + + protected Optional selectRole(IAMSecurityMapping mapping, ConnectorIdentity identity) + { + Optional optionalSelected = getRoleFromExtraCredential(identity); + + if (optionalSelected.isEmpty()) { + if (!mapping.allowedIamRoles().isEmpty() && mapping.iamRole().isEmpty()) { + throw new AccessDeniedException("No IAM role selected and mapping has no default role"); + } + verify(mapping.iamRole().isPresent() || mapping.credentials().isPresent(), "mapping must have role or credential"); + return mapping.iamRole(); + } + + String selected = optionalSelected.get(); + + // selected role must match default or be allowed + if (!selected.equals(mapping.iamRole().orElse(null)) && + !mapping.allowedIamRoles().contains(selected)) { + throw new AccessDeniedException("Selected IAM role is not allowed: " + selected); + } + + return optionalSelected; + } + + private Optional getRoleFromExtraCredential(ConnectorIdentity identity) + { + return roleCredentialName + .map(name -> identity.getExtraCredentials().get(name)) + .map(role -> colonReplacement + .map(replacement -> role.replace(replacement, ":")) + .orElse(role)); + } + + private static Supplier mappingsProvider(Supplier supplier, Optional refreshPeriod) + { + return refreshPeriod + .map(refresh -> memoizeWithExpiration(supplier::get, refresh.toMillis(), MILLISECONDS)) + .orElseGet(() -> memoize(supplier::get)); + } +} diff --git a/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingResult.java b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingResult.java new file mode 100644 index 000000000000..dbe99ea4086f --- /dev/null +++ b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingResult.java @@ -0,0 +1,44 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.iam.aws; + +import software.amazon.awssdk.auth.credentials.AwsCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; + +import java.util.Optional; + +import static java.util.Objects.requireNonNull; + +public record IAMSecurityMappingResult( + Optional credentials, + Optional iamRole, + Optional roleSessionName, + Optional endpoint, + Optional region) +{ + public IAMSecurityMappingResult + { + requireNonNull(credentials, "credentials is null"); + requireNonNull(iamRole, "iamRole is null"); + requireNonNull(roleSessionName, "roleSessionName is null"); + requireNonNull(endpoint, "endpoint is null"); + requireNonNull(region, "region is null"); + } + + public Optional credentialsProvider() + { + return credentials.map(StaticCredentialsProvider::create); + } +} diff --git a/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappings.java b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappings.java new file mode 100644 index 000000000000..8733e826ccae --- /dev/null +++ b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappings.java @@ -0,0 +1,47 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.iam.aws; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.google.common.collect.ImmutableList; +import io.trino.spi.security.ConnectorIdentity; + +import java.util.List; +import java.util.Optional; + +import static java.util.Objects.requireNonNull; + +public class IAMSecurityMappings +{ + protected final List mappings; + + @JsonCreator + public IAMSecurityMappings(@JsonProperty("mappings") List mappings) + { + this.mappings = ImmutableList.copyOf(requireNonNull(mappings, "mappings is null")); + } + + protected List getMappings() + { + return mappings; + } + + Optional getMapping(ConnectorIdentity identity) + { + return mappings.stream() + .filter(mapping -> mapping.matches(identity)) + .findFirst(); + } +} diff --git a/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingsFileSource.java b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingsFileSource.java new file mode 100644 index 000000000000..ddb71f53edc6 --- /dev/null +++ b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingsFileSource.java @@ -0,0 +1,44 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.iam.aws; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.google.inject.Inject; + +import java.io.File; +import java.util.function.Supplier; + +import static io.trino.plugin.base.util.JsonUtils.parseJson; + +public class IAMSecurityMappingsFileSource, E extends IAMSecurityMapping, C extends IAMSecurityMappingConfig> + implements Supplier +{ + private final File configFile; + private final String jsonPointer; + private final TypeReference mappingsTypeReference; + + @Inject + public IAMSecurityMappingsFileSource(C config, TypeReference mappingsTypeReference) + { + this.configFile = config.getConfigFile().orElseThrow(); + this.jsonPointer = config.getJsonPointer(); + this.mappingsTypeReference = mappingsTypeReference; + } + + @Override + public T get() + { + return parseJson(configFile.toPath(), jsonPointer, mappingsTypeReference); + } +} diff --git a/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingsUriSource.java b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingsUriSource.java new file mode 100644 index 000000000000..8db95d7357f8 --- /dev/null +++ b/lib/trino-iam-aws/src/main/java/io/trino/iam/aws/IAMSecurityMappingsUriSource.java @@ -0,0 +1,66 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.iam.aws; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.google.common.annotations.VisibleForTesting; +import com.google.inject.Inject; +import io.airlift.http.client.HttpClient; +import io.airlift.http.client.HttpStatus; +import io.airlift.http.client.Request; +import io.airlift.http.client.StringResponseHandler.StringResponse; + +import java.net.URI; +import java.util.function.Supplier; + +import static io.airlift.http.client.Request.Builder.prepareGet; +import static io.airlift.http.client.StringResponseHandler.createStringResponseHandler; +import static io.trino.plugin.base.util.JsonUtils.parseJson; +import static java.util.Objects.requireNonNull; + +public class IAMSecurityMappingsUriSource, E extends IAMSecurityMapping, C extends IAMSecurityMappingConfig> + implements Supplier +{ + private final URI configUri; + private final HttpClient httpClient; + private final String jsonPointer; + private final TypeReference mappingsTypeReference; + + @Inject + public IAMSecurityMappingsUriSource(C config, HttpClient httpClient, TypeReference mappingsTypeReference) + { + this.configUri = config.getConfigUri().orElseThrow(); + this.httpClient = requireNonNull(httpClient, "httpClient is null"); + this.jsonPointer = config.getJsonPointer(); + this.mappingsTypeReference = mappingsTypeReference; + } + + @Override + public T get() + { + return parseJson(getRawJsonString(), jsonPointer, mappingsTypeReference); + } + + @VisibleForTesting + public String getRawJsonString() + { + Request request = prepareGet().setUri(configUri).build(); + StringResponse response = httpClient.execute(request, createStringResponseHandler()); + int status = response.getStatusCode(); + if (status != HttpStatus.OK.code()) { + throw new RuntimeException("Request to '%s' returned unexpected status code: %s".formatted(configUri, status)); + } + return response.getBody(); + } +} diff --git a/lib/trino-iam-aws/src/test/java/io/trino/iam/aws/TestIAMSecurityMapping.java b/lib/trino-iam-aws/src/test/java/io/trino/iam/aws/TestIAMSecurityMapping.java new file mode 100644 index 000000000000..bfb8f75dad95 --- /dev/null +++ b/lib/trino-iam-aws/src/test/java/io/trino/iam/aws/TestIAMSecurityMapping.java @@ -0,0 +1,479 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.iam.aws; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.google.common.collect.ImmutableSet; +import io.trino.spi.security.AccessDeniedException; +import io.trino.spi.security.ConnectorIdentity; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity; + +import java.io.File; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +import static com.google.common.io.Resources.getResource; +import static io.trino.iam.aws.TestIAMSecurityMapping.MappingResult.credentials; +import static java.util.Objects.requireNonNull; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +public class TestIAMSecurityMapping +{ + private static final String IAM_ROLE_CREDENTIAL_NAME = "IAM_ROLE_CREDENTIAL_NAME"; + private static final String DEFAULT_USER = "testuser"; + + @Test + public void testMapping() + { + IAMSecurityMappingConfig mappingConfig = new IAMSecurityMappingConfig() + .setConfigFileInternal(getResourceFile("security-mapping.json")) + .setRoleCredentialNameInternal(IAM_ROLE_CREDENTIAL_NAME) + .setColonReplacementInternal("#"); + TypeReference> typeRef = new TypeReference<>() {}; + var provider = new IAMSecurityMappingProvider<>(mappingConfig, new IAMSecurityMappingsFileSource<>(mappingConfig, typeRef)); + + // matches user - mapping provides credentials + assertMapping( + provider, + MappingSelector.empty() + .withUser("jane"), + credentials("AKIAxxxaccess", "iXbXxxxsecret")); + + // matches user, no role selected and mapping has no default role + assertMappingFails( + provider, + MappingSelector.empty() + .withUser("john"), + "No IAM role selected and mapping has no default role"); + + // matches user and selected one of the allowed roles + assertMapping( + provider, + MappingSelector.empty() + .withUser("john") + .withExtraCredentialIamRole("arn:aws:iam::123456789101:role/role_a"), + MappingResult.iamRole("arn:aws:iam::123456789101:role/role_a")); + + // matches user and selected one of the allowed roles + assertMapping( + provider, + MappingSelector.empty() + .withUser("john") + .withExtraCredentialIamRole("arn:aws:iam::123456789101:role/role_b"), + MappingResult.iamRole("arn:aws:iam::123456789101:role/role_b")); + + // user selected a role not in allowed list + assertMappingFails( + provider, + MappingSelector.empty() + .withUser("john") + .withExtraCredentialIamRole("bogus"), + "Selected IAM role is not allowed: bogus"); + + // verify that colon replacement works + String roleWithoutColon = "arn#aws#iam##123456789101#role/role_b"; + assertThat(roleWithoutColon).doesNotContain(":"); + assertMapping( + provider, + MappingSelector.empty() + .withUser("john") + .withExtraCredentialIamRole(roleWithoutColon), + MappingResult.iamRole("arn:aws:iam::123456789101:role/role_b")); + + // matches user -- default role used + assertMapping( + provider, + MappingSelector.empty() + .withUser("alice"), + MappingResult.iamRole("alice_role")); + + // matches user and user selected default role + assertMapping( + provider, + MappingSelector.empty() + .withUser("alice") + .withExtraCredentialIamRole("alice_role"), + MappingResult.iamRole("alice_role")); + + // verify that the first matching rule is used + // matches prefix earlier in the file and selected role not allowed + assertMappingFails( + provider, + MappingSelector.empty() + .withUser("alice") + .withExtraCredentialIamRole("alice_other_role"), + "Selected IAM role is not allowed: alice_other_role"); + + // matches empty rule at the end -- default role used + assertMapping( + provider, + MappingSelector.empty(), + MappingResult.iamRole("arn:aws:iam::123456789101:role/default")); + + // matches user regex -- default role used + assertMapping( + provider, + MappingSelector.empty() + .withUser("bob"), + MappingResult.iamRole("bob_and_charlie_role")); + + // matches group -- default role used + assertMapping( + provider, + MappingSelector.empty() + .withGroups("finance"), + MappingResult.iamRole("finance_role")); + + // matches group regex -- default role used + assertMapping( + provider, + MappingSelector.empty() + .withGroups("eng"), + MappingResult.iamRole("hr_and_eng_group")); + + // verify that all constraints must match + // matches user but not group -- uses empty mapping at the end + assertMapping( + provider, + MappingSelector.empty() + .withUser("danny"), + MappingResult.iamRole("arn:aws:iam::123456789101:role/default")); + + // matches group but not user -- uses empty mapping at the end + assertMapping( + provider, + MappingSelector.empty() + .withGroups("hq"), + MappingResult.iamRole("arn:aws:iam::123456789101:role/default")); + + // matches user and group + assertMapping( + provider, + MappingSelector.empty() + .withUser("danny") + .withGroups("hq"), + MappingResult.iamRole("danny_hq_role")); + + // matches user -- mapping provides credentials and endpoint + assertMapping( + provider, + MappingSelector.empty() + .withUser("ryan"), + credentials("AKIAxxxaccess", "iXbXxxxsecret") + .withEndpoint("http://localhost:7753")); + + // matches user -- mapping provides credentials and region + assertMapping( + provider, + MappingSelector.empty() + .withUser("emily"), + credentials("AKIAxxxaccess", "iXbXxxxsecret") + .withRegion("us-west-2")); + + // matches role session name + assertMapping( + provider, + MappingSelector.empty() + .withUser("jack"), + MappingResult.iamRole("arn:aws:iam::1234567891012:role/default") + .withRoleSessionName("iam-trino-session")); + } + + @Test + public void testMappingWithFallbackToClusterDefault() + { + IAMSecurityMappingConfig mappingConfig = new IAMSecurityMappingConfig() + .setConfigFileInternal(getResourceFile("security-mapping-with-fallback-to-cluster-default.json")); + + TypeReference> typeRef = new TypeReference<>() {}; + var provider = new IAMSecurityMappingProvider<>(mappingConfig, new IAMSecurityMappingsFileSource<>(mappingConfig, typeRef)); + + // matches user -- uses the role from the mapping + assertMapping( + provider, + MappingSelector.empty(), + MappingResult.iamRole("arn:aws:iam::123456789101:role/example")); + + // doesn't match any rule except default rule at the end + assertThat(getMapping(provider, MappingSelector.empty().withUser("fake"))).isEmpty(); + } + + @Test + public void testMappingWithoutFallback() + { + IAMSecurityMappingConfig mappingConfig = new IAMSecurityMappingConfig() + .setConfigFileInternal(getResourceFile("security-mapping-without-fallback.json")); + + TypeReference> typeRef = new TypeReference<>() {}; + var provider = new IAMSecurityMappingProvider<>(mappingConfig, new IAMSecurityMappingsFileSource<>(mappingConfig, typeRef)); + + // matches user - return role from the mapping + assertMapping( + provider, + MappingSelector.empty() + .withUser("non-default"), + MappingResult.iamRole("arn:aws:iam::123456789101:role/example")); + + // doesn't match any rule + assertMappingFails( + provider, + MappingSelector.empty(), + "No matching IAM security mapping"); + } + + @Test + public void testMappingWithoutRoleCredentialsFallbackShouldFail() + { + assertThatThrownBy(() -> + new IAMSecurityMapping( + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("must either allow useClusterDefault role or provide role and/or credentials"); + } + + @Test + public void testMappingWithRoleAndFallbackShouldFail() + { + Optional iamRole = Optional.of("arn:aws:iam::123456789101:role/allow_path"); + Optional useClusterDefault = Optional.of(true); + + assertThatThrownBy(() -> + new IAMSecurityMapping( + Optional.empty(), + Optional.empty(), + iamRole, + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + useClusterDefault, + Optional.empty(), + Optional.empty())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("must either allow useClusterDefault role or provide role and/or credentials"); + } + + @Test + public void testMappingWithRoleSessionNameWithoutIamRoleShouldFail() + { + Optional roleSessionName = Optional.of("iam-trino-session"); + + assertThatThrownBy(() -> + new IAMSecurityMapping( + Optional.empty(), + Optional.empty(), + Optional.empty(), + roleSessionName, + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("iamRole must be provided when roleSessionName is provided"); + } + + private File getResourceFile(String name) + { + return new File(getResource(getClass(), name).getFile()); + } + + private static void assertMapping(IAMSecurityMappingProvider, IAMSecurityMapping, IAMSecurityMappingConfig> provider, MappingSelector selector, MappingResult expected) + { + Optional mapping = getMapping(provider, selector); + + assertThat(mapping).isPresent().get().satisfies(actual -> { + assertThat(actual.credentials().map(AwsCredentialsIdentity::accessKeyId)).isEqualTo(expected.accessKey()); + assertThat(actual.credentials().map(AwsCredentialsIdentity::secretAccessKey)).isEqualTo(expected.secretKey()); + assertThat(actual.iamRole()).isEqualTo(expected.iamRole()); + assertThat(actual.roleSessionName()).isEqualTo(expected.roleSessionName()); + assertThat(actual.endpoint()).isEqualTo(expected.endpoint()); + assertThat(actual.region()).isEqualTo(expected.region()); + }); + } + + private static void assertMappingFails(IAMSecurityMappingProvider, IAMSecurityMapping, IAMSecurityMappingConfig> provider, MappingSelector selector, String message) + { + assertThatThrownBy(() -> getMapping(provider, selector)) + .isInstanceOf(AccessDeniedException.class) + .hasMessage("Access Denied: " + message); + } + + private static Optional getMapping(IAMSecurityMappingProvider, IAMSecurityMapping, IAMSecurityMappingConfig> provider, MappingSelector selector) + { + return provider.getMapping(selector.identity()); + } + + public static class MappingSelector + { + public static MappingSelector empty() + { + return new MappingSelector(DEFAULT_USER, ImmutableSet.of(), Optional.empty()); + } + + private final String user; + private final Set groups; + private final Optional extraCredentialIamRole; + + private MappingSelector(String user, Set groups, Optional extraCredentialIamRole) + { + this.user = requireNonNull(user, "user is null"); + this.groups = ImmutableSet.copyOf(requireNonNull(groups, "groups is null")); + this.extraCredentialIamRole = requireNonNull(extraCredentialIamRole, "extraCredentialIamRole is null"); + } + + public MappingSelector withExtraCredentialIamRole(String role) + { + return new MappingSelector(user, groups, Optional.of(role)); + } + + public MappingSelector withUser(String user) + { + return new MappingSelector(user, groups, extraCredentialIamRole); + } + + public MappingSelector withGroups(String... groups) + { + return new MappingSelector(user, ImmutableSet.copyOf(groups), extraCredentialIamRole); + } + + public ConnectorIdentity identity() + { + Map extraCredentials = new HashMap<>(); + extraCredentialIamRole.ifPresent(role -> extraCredentials.put(IAM_ROLE_CREDENTIAL_NAME, role)); + + return ConnectorIdentity.forUser(user) + .withGroups(groups) + .withExtraCredentials(extraCredentials) + .build(); + } + } + + public static class MappingResult + { + public static MappingResult credentials(String accessKey, String secretKey) + { + return new MappingResult( + Optional.of(accessKey), + Optional.of(secretKey), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty()); + } + + public static MappingResult iamRole(String role) + { + return new MappingResult( + Optional.empty(), + Optional.empty(), + Optional.of(role), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty()); + } + + private final Optional accessKey; + private final Optional secretKey; + private final Optional iamRole; + private final Optional roleSessionName; + private final Optional kmsKeyId; + private final Optional sseCustomerKey; + private final Optional endpoint; + private final Optional region; + + private MappingResult( + Optional accessKey, + Optional secretKey, + Optional iamRole, + Optional roleSessionName, + Optional kmsKeyId, + Optional sseCustomerKey, + Optional endpoint, + Optional region) + { + this.accessKey = requireNonNull(accessKey, "accessKey is null"); + this.secretKey = requireNonNull(secretKey, "secretKey is null"); + this.iamRole = requireNonNull(iamRole, "role is null"); + this.kmsKeyId = requireNonNull(kmsKeyId, "kmsKeyId is null"); + this.sseCustomerKey = requireNonNull(sseCustomerKey, "sseCustomerKey is null"); + this.endpoint = requireNonNull(endpoint, "endpoint is null"); + this.roleSessionName = requireNonNull(roleSessionName, "roleSessionName is null"); + this.region = requireNonNull(region, "region is null"); + } + + public MappingResult withEndpoint(String endpoint) + { + return new MappingResult(accessKey, secretKey, iamRole, Optional.empty(), kmsKeyId, sseCustomerKey, Optional.of(endpoint), region); + } + + public MappingResult withRegion(String region) + { + return new MappingResult(accessKey, secretKey, iamRole, Optional.empty(), kmsKeyId, sseCustomerKey, endpoint, Optional.of(region)); + } + + public MappingResult withRoleSessionName(String roleSessionName) + { + return new MappingResult(accessKey, secretKey, iamRole, Optional.of(roleSessionName), kmsKeyId, sseCustomerKey, Optional.empty(), region); + } + + public Optional accessKey() + { + return accessKey; + } + + public Optional secretKey() + { + return secretKey; + } + + public Optional iamRole() + { + return iamRole; + } + + public Optional roleSessionName() + { + return roleSessionName; + } + + public Optional endpoint() + { + return endpoint; + } + + public Optional region() + { + return region; + } + } +} diff --git a/lib/trino-filesystem-s3/src/test/java/io/trino/filesystem/s3/TestS3SecurityMappingsUriSource.java b/lib/trino-iam-aws/src/test/java/io/trino/iam/aws/TestIAMSecurityMappingsUriSource.java similarity index 74% rename from lib/trino-filesystem-s3/src/test/java/io/trino/filesystem/s3/TestS3SecurityMappingsUriSource.java rename to lib/trino-iam-aws/src/test/java/io/trino/iam/aws/TestIAMSecurityMappingsUriSource.java index b6ea26467a00..7dafbf40b2df 100644 --- a/lib/trino-filesystem-s3/src/test/java/io/trino/filesystem/s3/TestS3SecurityMappingsUriSource.java +++ b/lib/trino-iam-aws/src/test/java/io/trino/iam/aws/TestIAMSecurityMappingsUriSource.java @@ -11,8 +11,9 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package io.trino.filesystem.s3; +package io.trino.iam.aws; +import com.fasterxml.jackson.core.type.TypeReference; import io.airlift.http.client.HttpStatus; import io.airlift.http.client.Response; import io.airlift.http.client.testing.TestingHttpClient; @@ -24,7 +25,7 @@ import static io.airlift.http.client.testing.TestingResponse.mockResponse; import static org.assertj.core.api.Assertions.assertThat; -public class TestS3SecurityMappingsUriSource +public class TestIAMSecurityMappingsUriSource { private static final String MOCK_MAPPINGS_RESPONSE = "{\"mappings\": [{\"iamRole\":\"arn:aws:iam::test\",\"user\":\"test\"}]}"; @@ -33,8 +34,9 @@ public class TestS3SecurityMappingsUriSource public void testGetRawJson() { Response response = mockResponse(HttpStatus.OK, JSON_UTF_8, MOCK_MAPPINGS_RESPONSE); - S3SecurityMappingConfig config = new S3SecurityMappingConfig().setConfigUri(URI.create("http://test:1234/api/endpoint")); - var provider = new S3SecurityMappingsUriSource(config, new TestingHttpClient(_ -> response)); + IAMSecurityMappingConfig config = new IAMSecurityMappingConfig().setConfigUriInternal(URI.create("http://test:1234/api/endpoint")); + var typeRef = new TypeReference>() {}; + var provider = new IAMSecurityMappingsUriSource<>(config, new TestingHttpClient(_ -> response), typeRef); String result = provider.getRawJsonString(); assertThat(result).isEqualTo(MOCK_MAPPINGS_RESPONSE); } diff --git a/lib/trino-iam-aws/src/test/resources/io/trino/iam/aws/security-mapping-with-fallback-to-cluster-default.json b/lib/trino-iam-aws/src/test/resources/io/trino/iam/aws/security-mapping-with-fallback-to-cluster-default.json new file mode 100644 index 000000000000..0a7e278dc695 --- /dev/null +++ b/lib/trino-iam-aws/src/test/resources/io/trino/iam/aws/security-mapping-with-fallback-to-cluster-default.json @@ -0,0 +1,11 @@ +{ + "mappings": [ + { + "user": "testuser", + "iamRole": "arn:aws:iam::123456789101:role/example" + }, + { + "useClusterDefault": "true" + } + ] +} diff --git a/lib/trino-iam-aws/src/test/resources/io/trino/iam/aws/security-mapping-without-fallback.json b/lib/trino-iam-aws/src/test/resources/io/trino/iam/aws/security-mapping-without-fallback.json new file mode 100644 index 000000000000..cbacc4dce48e --- /dev/null +++ b/lib/trino-iam-aws/src/test/resources/io/trino/iam/aws/security-mapping-without-fallback.json @@ -0,0 +1,8 @@ +{ + "mappings": [ + { + "user": "non-default", + "iamRole": "arn:aws:iam::123456789101:role/example" + } + ] +} diff --git a/lib/trino-iam-aws/src/test/resources/io/trino/iam/aws/security-mapping.json b/lib/trino-iam-aws/src/test/resources/io/trino/iam/aws/security-mapping.json new file mode 100644 index 000000000000..002495b08957 --- /dev/null +++ b/lib/trino-iam-aws/src/test/resources/io/trino/iam/aws/security-mapping.json @@ -0,0 +1,58 @@ +{ + "mappings": [ + { + "user": "jane", + "accessKey": "AKIAxxxaccess", + "secretKey": "iXbXxxxsecret" + }, + { + "user": "john", + "allowedIamRoles": [ + "arn:aws:iam::123456789101:role/role_a", + "arn:aws:iam::123456789101:role/role_b" + ] + }, + { + "user": "alice", + "iamRole": "alice_role" + }, + { + "user": "bob|charlie", + "iamRole": "bob_and_charlie_role" + }, + { + "group": "finance", + "iamRole": "finance_role" + }, + { + "group": "hr|eng", + "iamRole": "hr_and_eng_group" + }, + { + "user": "danny", + "group": "hq", + "iamRole": "danny_hq_role" + }, + { + "user": "ryan", + "accessKey": "AKIAxxxaccess", + "secretKey": "iXbXxxxsecret", + "endpoint": "http://localhost:7753" + }, + { + "user": "emily", + "accessKey": "AKIAxxxaccess", + "secretKey": "iXbXxxxsecret", + "region": "us-west-2" + }, + { + "user": "jack", + "iamRole": "arn:aws:iam::1234567891012:role/default", + "roleSessionName": "iam-trino-session" + }, + { + "useClusterDefault": "false", + "iamRole": "arn:aws:iam::123456789101:role/default" + } + ] +} diff --git a/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/util/JsonUtils.java b/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/util/JsonUtils.java index d2abef713c81..6839e388b53a 100644 --- a/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/util/JsonUtils.java +++ b/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/util/JsonUtils.java @@ -20,6 +20,7 @@ import com.fasterxml.jackson.core.StreamReadConstraints; import com.fasterxml.jackson.core.StreamReadFeature; import com.fasterxml.jackson.core.StreamWriteFeature; +import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.core.util.JsonRecyclerPools; import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.JsonNode; @@ -43,11 +44,11 @@ public final class JsonUtils { - private static final ObjectMapper OBJECT_MAPPER = new ObjectMapperProvider().get() - .enable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES) - .enable(MapperFeature.ACCEPT_CASE_INSENSITIVE_ENUMS); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapperProvider().get().enable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES).enable(MapperFeature.ACCEPT_CASE_INSENSITIVE_ENUMS); - private JsonUtils() {} + private JsonUtils() + { + } public static T parseJson(Path path, Class javaType) { @@ -151,6 +152,71 @@ private static T parseJson(JsonNode node, String jsonPointer, Class javaT return jsonTreeToValue(mappingsNode, javaType); } + public static T parseJson(Path path, TypeReference typeReference) + { + if (!path.isAbsolute()) { + path = path.toAbsolutePath(); + } + + checkArgument(exists(path), "File does not exist: %s", path); + checkArgument(isReadable(path), "File is not readable: %s", path); + + try { + byte[] json = Files.readAllBytes(path); + return parseJson(json, typeReference); + } + catch (IOException | RuntimeException e) { + throw new IllegalArgumentException(format("Invalid JSON file '%s' for '%s'", path, typeReference.getType()), e); + } + } + + public static T parseJson(byte[] jsonBytes, TypeReference typeReference) + { + requireNonNull(jsonBytes, "jsonBytes is null"); + requireNonNull(typeReference, "typeReference is null"); + + try (JsonParser parser = OBJECT_MAPPER.createParser(jsonBytes)) { + T value = OBJECT_MAPPER.readValue(parser, typeReference); + checkArgument(parser.nextToken() == null, "Found characters after the expected end of input"); + return value; + } + catch (IOException e) { + throw new UncheckedIOException("Could not parse JSON", e); + } + } + + public static T parseJson(String json, String jsonPointer, TypeReference typeReference) + { + JsonNode node = parseJson(json, JsonNode.class); + return parseJson(node, jsonPointer, typeReference); + } + + private static T parseJson(JsonNode node, String jsonPointer, TypeReference typeReference) + { + JsonNode mappingsNode = node.at(jsonPointer); + try { + return OBJECT_MAPPER.treeToValue(mappingsNode, OBJECT_MAPPER.getTypeFactory().constructType(typeReference.getType())); + } + catch (JsonProcessingException e) { + throw new UncheckedIOException("Failed to convert JSON tree node using TypeReference", e); + } + } + + public static T parseJson(Path path, String jsonPointer, TypeReference typeReference) + { + JsonNode node = parseJson(path, JsonNode.class); + JsonNode mappingsNode = node.at(jsonPointer); + if (mappingsNode.isMissingNode()) { + throw new IllegalArgumentException("JSON pointer not found: " + jsonPointer); + } + try { + return OBJECT_MAPPER.readValue(mappingsNode.traverse(), typeReference); + } + catch (IOException e) { + throw new UncheckedIOException("Failed to parse JSON at pointer", e); + } + } + public static JsonFactory jsonFactory() { return jsonFactoryBuilder().build(); diff --git a/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/metastore/glue/DeltaLakeGlueMetastoreTableOperationsProvider.java b/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/metastore/glue/DeltaLakeGlueMetastoreTableOperationsProvider.java index bf8ba3aea2f4..01846edd45d9 100644 --- a/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/metastore/glue/DeltaLakeGlueMetastoreTableOperationsProvider.java +++ b/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/metastore/glue/DeltaLakeGlueMetastoreTableOperationsProvider.java @@ -17,6 +17,7 @@ import io.trino.plugin.deltalake.metastore.DeltaLakeTableOperations; import io.trino.plugin.deltalake.metastore.DeltaLakeTableOperationsProvider; import io.trino.plugin.hive.metastore.glue.GlueCache; +import io.trino.plugin.hive.metastore.glue.GlueClientFactory; import io.trino.plugin.hive.metastore.glue.GlueMetastoreStats; import io.trino.spi.connector.ConnectorSession; import software.amazon.awssdk.services.glue.GlueClient; @@ -26,17 +27,17 @@ public class DeltaLakeGlueMetastoreTableOperationsProvider implements DeltaLakeTableOperationsProvider { - private final GlueClient glueClient; + private final GlueClientFactory glueClientFactory; private final GlueCache glueCache; private final GlueMetastoreStats stats; @Inject public DeltaLakeGlueMetastoreTableOperationsProvider( - GlueClient glueClient, + GlueClientFactory glueClientFactory, GlueCache glueCache, GlueMetastoreStats stats) { - this.glueClient = requireNonNull(glueClient, "glueClient is null"); + this.glueClientFactory = requireNonNull(glueClientFactory, "glueClientFactory is null"); this.glueCache = requireNonNull(glueCache, "glueCache is null"); this.stats = requireNonNull(stats, "stats is null"); } @@ -44,6 +45,7 @@ public DeltaLakeGlueMetastoreTableOperationsProvider( @Override public DeltaLakeTableOperations createTableOperations(ConnectorSession session) { + GlueClient glueClient = glueClientFactory.create(session.getIdentity()); return new DeltaLakeGlueMetastoreTableOperations(glueClient, glueCache, stats); } } diff --git a/plugin/trino-hive/pom.xml b/plugin/trino-hive/pom.xml index a8de44dc2689..5f804234e63d 100644 --- a/plugin/trino-hive/pom.xml +++ b/plugin/trino-hive/pom.xml @@ -66,6 +66,11 @@ configuration + + io.airlift + http-client + + io.airlift json @@ -111,6 +116,11 @@ trino-hive-formats + + io.trino + trino-iam-aws + + io.trino trino-memory-context diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/ForGlueHttpClient.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/ForGlueHttpClient.java new file mode 100644 index 000000000000..561a5da14dfc --- /dev/null +++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/ForGlueHttpClient.java @@ -0,0 +1,29 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hive.metastore.glue; + +import com.google.inject.BindingAnnotation; + +import java.lang.annotation.Retention; +import java.lang.annotation.Target; + +import static java.lang.annotation.ElementType.FIELD; +import static java.lang.annotation.ElementType.METHOD; +import static java.lang.annotation.ElementType.PARAMETER; +import static java.lang.annotation.RetentionPolicy.RUNTIME; + +@Retention(RUNTIME) +@Target({FIELD, PARAMETER, METHOD}) +@BindingAnnotation +public @interface ForGlueHttpClient {} diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/ForStsHttpClient.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/ForStsHttpClient.java new file mode 100644 index 000000000000..a04ac509cbcc --- /dev/null +++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/ForStsHttpClient.java @@ -0,0 +1,29 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hive.metastore.glue; + +import com.google.inject.BindingAnnotation; + +import java.lang.annotation.Retention; +import java.lang.annotation.Target; + +import static java.lang.annotation.ElementType.FIELD; +import static java.lang.annotation.ElementType.METHOD; +import static java.lang.annotation.ElementType.PARAMETER; +import static java.lang.annotation.RetentionPolicy.RUNTIME; + +@Retention(RUNTIME) +@Target({FIELD, PARAMETER, METHOD}) +@BindingAnnotation +public @interface ForStsHttpClient {} diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueClientFactory.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueClientFactory.java new file mode 100644 index 000000000000..ae6f32f833cb --- /dev/null +++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueClientFactory.java @@ -0,0 +1,170 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hive.metastore.glue; + +import com.google.common.collect.ImmutableList; +import com.google.inject.Inject; +import io.trino.iam.aws.IAMSecurityMapping; +import io.trino.iam.aws.IAMSecurityMappingProvider; +import io.trino.iam.aws.IAMSecurityMappingResult; +import io.trino.iam.aws.IAMSecurityMappings; +import io.trino.spi.security.ConnectorIdentity; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.core.interceptor.ExecutionInterceptor; +import software.amazon.awssdk.http.SdkHttpClient; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.regions.providers.DefaultAwsRegionProviderChain; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.glue.GlueClientBuilder; +import software.amazon.awssdk.services.glue.model.ConcurrentModificationException; +import software.amazon.awssdk.services.sts.StsClient; +import software.amazon.awssdk.services.sts.StsClientBuilder; +import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider; +import software.amazon.awssdk.services.sts.auth.StsWebIdentityTokenFileCredentialsProvider; + +import java.util.Optional; +import java.util.Set; + +import static java.util.Objects.requireNonNull; + +public class GlueClientFactory +{ + private final GlueHiveMetastoreConfig config; + private final Optional, IAMSecurityMapping, GlueSecurityMappingConfig>> mappingProvider; + private final Optional staticCredentialsProvider; + private final @ForGlueHiveMetastore Set executionInterceptors; + private final @ForGlueHttpClient SdkHttpClient glueHttpClient; + private final @ForStsHttpClient SdkHttpClient stsHttpClient; + + @Inject + public GlueClientFactory( + GlueHiveMetastoreConfig config, + Optional, IAMSecurityMapping, GlueSecurityMappingConfig>> mappingProvider, + @ForGlueHiveMetastore Set executionInterceptors, + @ForGlueHttpClient SdkHttpClient glueHttpClient, + @ForStsHttpClient SdkHttpClient stsHttpClient) + { + this.config = requireNonNull(config, "config is null"); + this.mappingProvider = mappingProvider; + this.staticCredentialsProvider = getStaticCredentialsProvider(config); + this.executionInterceptors = Set.copyOf(executionInterceptors); + this.glueHttpClient = requireNonNull(glueHttpClient, "glueHttpClient is null"); + this.stsHttpClient = requireNonNull(stsHttpClient, "stsHttpClient is null"); + } + + public GlueClient create(ConnectorIdentity identity) + { + GlueClientBuilder builder = GlueClient.builder(); + builder.overrideConfiguration(override -> override + .executionInterceptors(ImmutableList.copyOf(executionInterceptors)) + .retryStrategy(retryBuilder -> retryBuilder + .retryOnException(t -> t instanceof ConcurrentModificationException) + .maxAttempts(config.getMaxGlueErrorRetries()))); + + Optional credentials = resolveCredentials(identity); + if (credentials.isPresent()) { + builder.credentialsProvider(credentials.get()); + } + else { + staticCredentialsProvider.ifPresent(builder::credentialsProvider); + } + + if (config.getGlueEndpointUrl().isPresent()) { + builder.endpointOverride(config.getGlueEndpointUrl().get()); + builder.region(Region.of(config.getGlueRegion().orElseThrow())); + } + else if (config.getGlueRegion().isPresent()) { + builder.region(Region.of(config.getGlueRegion().get())); + } + else if (config.getPinGlueClientToCurrentRegion()) { + builder.region(DefaultAwsRegionProviderChain.builder().build().getRegion()); + } + + builder.httpClient(glueHttpClient); + + return builder.build(); + } + + private Optional resolveCredentials(ConnectorIdentity identity) + { + if (mappingProvider.isPresent()) { + Optional mapping = mappingProvider.flatMap(provider -> provider.getMapping(identity)); + if (mapping.isPresent()) { + IAMSecurityMappingResult iamMapping = mapping.get(); + return Optional.of(StsAssumeRoleCredentialsProvider.builder() + .refreshRequest(req -> req + .roleArn(iamMapping.iamRole().orElseThrow()) + .roleSessionName(iamMapping.roleSessionName().orElse("trino-session"))) + .stsClient(getStsClient()) + .asyncCredentialUpdateEnabled(true) + .build()); + } + } + + if (config.isUseWebIdentityTokenCredentialsProvider()) { + return Optional.of(StsWebIdentityTokenFileCredentialsProvider.builder() + .stsClient(getStsClient()) + .asyncCredentialUpdateEnabled(true) + .build()); + } + + if (config.getIamRole().isPresent()) { + return Optional.of(StsAssumeRoleCredentialsProvider.builder() + .refreshRequest(req -> req + .roleArn(config.getIamRole().get()) + .roleSessionName("trino-session") + .externalId(config.getExternalId().orElse(null))) + .stsClient(getStsClient()) + .asyncCredentialUpdateEnabled(true) + .build()); + } + + return Optional.empty(); + } + + private StsClient getStsClient() + { + StsClientBuilder sts = StsClient.builder(); + staticCredentialsProvider.ifPresent(sts::credentialsProvider); + + if (config.getGlueStsEndpointUrl().isPresent() && config.getGlueStsRegion().isPresent()) { + return sts.endpointOverride(config.getGlueStsEndpointUrl().get()) + .region(Region.of(config.getGlueStsRegion().get())) + .build(); + } + + if (config.getGlueStsRegion().isPresent()) { + return sts.region(Region.of(config.getGlueStsRegion().get())).build(); + } + + if (config.getPinGlueClientToCurrentRegion()) { + return sts.region(DefaultAwsRegionProviderChain.builder().build().getRegion()).build(); + } + + sts.httpClient(stsHttpClient); + + return sts.build(); + } + + private static Optional getStaticCredentialsProvider(GlueHiveMetastoreConfig config) + { + if (config.getAwsAccessKey().isPresent() && config.getAwsSecretKey().isPresent()) { + return Optional.of(StaticCredentialsProvider.create( + AwsBasicCredentials.create(config.getAwsAccessKey().get(), config.getAwsSecretKey().get()))); + } + return Optional.empty(); + } +} diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastore.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastore.java index 40a19fb41e87..394ed3c82bae 100644 --- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastore.java +++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastore.java @@ -18,7 +18,6 @@ import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Lists; -import com.google.inject.Inject; import io.airlift.log.Logger; import io.trino.filesystem.Location; import io.trino.filesystem.TrinoFileSystem; @@ -180,7 +179,6 @@ public class GlueHiveMetastore private final Predicate tableVisibilityFilter; private final ExecutorService executor; - @Inject public GlueHiveMetastore( GlueClient glueClient, GlueCache glueCache, diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastoreConfig.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastoreConfig.java index 09bb6bd8809b..17ab436f9fd2 100644 --- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastoreConfig.java +++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastoreConfig.java @@ -17,6 +17,7 @@ import io.airlift.configuration.ConfigDescription; import io.airlift.configuration.ConfigSecuritySensitive; import io.airlift.configuration.DefunctConfig; +import io.trino.iam.aws.IAMSecurityMappingConfig; import jakarta.validation.constraints.Max; import jakarta.validation.constraints.Min; @@ -32,6 +33,7 @@ "hive.metastore.glue.aws-credentials-provider", }) public class GlueHiveMetastoreConfig + extends IAMSecurityMappingConfig { private Optional glueRegion = Optional.empty(); private Optional glueEndpointUrl = Optional.empty(); diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastoreFactory.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastoreFactory.java index 8b5075e606e6..b7d39c6e228c 100644 --- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastoreFactory.java +++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastoreFactory.java @@ -15,40 +15,80 @@ import com.google.inject.Inject; import io.opentelemetry.api.trace.Tracer; +import io.trino.filesystem.TrinoFileSystemFactory; import io.trino.metastore.HiveMetastore; import io.trino.metastore.HiveMetastoreFactory; import io.trino.metastore.tracing.TracingHiveMetastore; +import io.trino.spi.catalog.CatalogName; import io.trino.spi.security.ConnectorIdentity; +import software.amazon.awssdk.services.glue.GlueClient; import java.util.Optional; +import java.util.Set; + +import static java.util.Objects.requireNonNull; public class GlueHiveMetastoreFactory implements HiveMetastoreFactory { - private final HiveMetastore metastore; + private final Tracer tracer; + private final GlueClientFactory glueClientFactory; + private final GlueCache glueCache; + private final GlueMetastoreStats glueStats; + private final TrinoFileSystemFactory fileSystemFactory; + private final GlueHiveMetastoreConfig config; + private final CatalogName catalogName; + private final Set visibleTableKinds; - // Glue metastore does not support impersonation, so just use single shared instance @Inject - public GlueHiveMetastoreFactory(GlueHiveMetastore metastore, Tracer tracer) + public GlueHiveMetastoreFactory( + Tracer tracer, + GlueClientFactory glueClientFactory, + GlueCache glueCache, + GlueMetastoreStats glueStats, + TrinoFileSystemFactory fileSystemFactory, + GlueHiveMetastoreConfig config, + CatalogName catalogName, + Set visibleTableKinds) { - this.metastore = new TracingHiveMetastore(tracer, metastore); + this.tracer = requireNonNull(tracer, "tracer is null"); + this.glueClientFactory = requireNonNull(glueClientFactory, "glueClientFactory is null"); + this.glueCache = requireNonNull(glueCache, "glueCache is null"); + this.glueStats = requireNonNull(glueStats, "glueStats is null"); + this.fileSystemFactory = requireNonNull(fileSystemFactory, "fileSystemFactory is null"); + this.config = requireNonNull(config, "config is null"); + this.catalogName = requireNonNull(catalogName, "catalogName is null"); + this.visibleTableKinds = requireNonNull(visibleTableKinds, "visibleTableKinds is null"); } @Override - public boolean hasBuiltInCaching() + public HiveMetastore createMetastore(Optional identity) { - return true; + if (identity.isEmpty()) { + throw new IllegalStateException("ConnectorIdentity must be provided"); + } + + GlueClient glueClient = glueClientFactory.create(identity.get()); + GlueHiveMetastore metastore = new GlueHiveMetastore( + glueClient, + glueCache, + glueStats, + fileSystemFactory, + config, + catalogName, + visibleTableKinds); + return new TracingHiveMetastore(tracer, metastore); } @Override - public boolean isImpersonationEnabled() + public boolean hasBuiltInCaching() { - return false; + return true; } @Override - public HiveMetastore createMetastore(Optional identity) + public boolean isImpersonationEnabled() { - return metastore; + return false; } } diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueMetastoreModule.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueMetastoreModule.java index 262dae64e633..62dc4c900874 100644 --- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueMetastoreModule.java +++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueMetastoreModule.java @@ -13,20 +13,29 @@ */ package io.trino.plugin.hive.metastore.glue; -import com.google.common.collect.ImmutableList; +import com.fasterxml.jackson.core.type.TypeReference; +import com.google.common.base.VerifyException; import com.google.inject.Binder; +import com.google.inject.BindingAnnotation; import com.google.inject.Inject; import com.google.inject.Key; import com.google.inject.Provider; import com.google.inject.Provides; import com.google.inject.Scopes; import com.google.inject.Singleton; +import com.google.inject.TypeLiteral; import com.google.inject.multibindings.Multibinder; import com.google.inject.multibindings.ProvidesIntoOptional; +import com.google.inject.util.Providers; import io.airlift.configuration.AbstractConfigurationAwareModule; import io.airlift.units.Duration; import io.opentelemetry.api.OpenTelemetry; import io.opentelemetry.instrumentation.awssdk.v2_2.AwsSdkTelemetry; +import io.trino.iam.aws.IAMSecurityMapping; +import io.trino.iam.aws.IAMSecurityMappingProvider; +import io.trino.iam.aws.IAMSecurityMappings; +import io.trino.iam.aws.IAMSecurityMappingsFileSource; +import io.trino.iam.aws.IAMSecurityMappingsUriSource; import io.trino.metastore.HiveMetastoreFactory; import io.trino.metastore.RawHiveMetastoreFactory; import io.trino.metastore.cache.CachingHiveMetastoreConfig; @@ -34,33 +43,29 @@ import io.trino.plugin.hive.HideDeltaLakeTables; import io.trino.spi.Node; import io.trino.spi.catalog.CatalogName; -import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; -import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; import software.amazon.awssdk.core.interceptor.ExecutionInterceptor; +import software.amazon.awssdk.http.SdkHttpClient; import software.amazon.awssdk.http.apache.ApacheHttpClient; -import software.amazon.awssdk.regions.Region; -import software.amazon.awssdk.regions.providers.DefaultAwsRegionProviderChain; -import software.amazon.awssdk.retries.api.BackoffStrategy; -import software.amazon.awssdk.services.glue.GlueClient; -import software.amazon.awssdk.services.glue.GlueClientBuilder; -import software.amazon.awssdk.services.glue.model.ConcurrentModificationException; -import software.amazon.awssdk.services.sts.StsClient; -import software.amazon.awssdk.services.sts.StsClientBuilder; -import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider; -import software.amazon.awssdk.services.sts.auth.StsWebIdentityTokenFileCredentialsProvider; +import java.lang.annotation.Retention; +import java.lang.annotation.Target; import java.util.EnumSet; -import java.util.Optional; import java.util.Set; +import java.util.function.Supplier; -import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkState; +import static com.google.inject.Scopes.SINGLETON; import static com.google.inject.multibindings.Multibinder.newSetBinder; import static com.google.inject.multibindings.OptionalBinder.newOptionalBinder; import static com.google.inject.multibindings.ProvidesIntoOptional.Type.DEFAULT; import static io.airlift.configuration.ConfigBinder.configBinder; -import static io.trino.plugin.base.ClosingBinder.closingBinder; +import static io.airlift.http.client.HttpClientBinder.httpClientBinder; +import static java.lang.annotation.ElementType.FIELD; +import static java.lang.annotation.ElementType.METHOD; +import static java.lang.annotation.ElementType.PARAMETER; +import static java.lang.annotation.RetentionPolicy.RUNTIME; import static java.util.Objects.requireNonNull; +import static java.util.concurrent.TimeUnit.SECONDS; import static org.weakref.jmx.guice.ExportBinder.newExporter; public final class GlueMetastoreModule @@ -71,10 +76,18 @@ protected void setup(Binder binder) { configBinder(binder).bindConfig(GlueHiveMetastoreConfig.class); + if (buildConfigObject(GlueSecurityMappingEnabledConfig.class).isEnabled()) { + install(new GlueMetastoreModule.GlueSecurityMappingModule()); + } + else { + newOptionalBinder(binder, new TypeLiteral, IAMSecurityMapping, GlueSecurityMappingConfig>>() {}) + .setDefault() + .toProvider(Providers.of(null)); + } + binder.bind(GlueHiveMetastoreFactory.class).in(Scopes.SINGLETON); - binder.bind(GlueHiveMetastore.class).in(Scopes.SINGLETON); binder.bind(GlueMetastoreStats.class).in(Scopes.SINGLETON); - newExporter(binder).export(GlueHiveMetastore.class).withGeneratedName(); + binder.bind(GlueClientFactory.class).in(Scopes.SINGLETON); newExporter(binder).export(GlueMetastoreStats.class).withGeneratedName(); newOptionalBinder(binder, Key.get(HiveMetastoreFactory.class, RawHiveMetastoreFactory.class)) .setDefault() @@ -86,8 +99,6 @@ protected void setup(Binder binder) executionInterceptorMultibinder.addBinding().toProvider(TelemetryExecutionInterceptorProvider.class).in(Scopes.SINGLETON); executionInterceptorMultibinder.addBinding().to(GlueHiveExecutionInterceptor.class).in(Scopes.SINGLETON); executionInterceptorMultibinder.addBinding().to(GlueCatalogIdInterceptor.class).in(Scopes.SINGLETON); - - closingBinder(binder).registerCloseable(GlueClient.class); } @Override @@ -123,7 +134,7 @@ public static GlueCache createGlueCache(CachingHiveMetastoreConfig config, Catal // Note: while we could skip CachingHiveMetastoreModule altogether on workers, we retain it so that catalog // configuration can remain identical for all nodes, making cluster configuration easier. boolean enabled = currentNode.isCoordinator() && - (metadataCacheTtl.toMillis() > 0 || statsCacheTtl.toMillis() > 0); + (metadataCacheTtl.toMillis() > 0 || statsCacheTtl.toMillis() > 0); checkState(config.isPartitionCacheEnabled(), "Disabling partitions cache is not supported with Glue v2"); checkState(config.isCacheMissing(), "Disabling cache missing is not supported with Glue v2"); @@ -144,87 +155,20 @@ public static GlueCache createGlueCache(CachingHiveMetastoreConfig config, Catal @Provides @Singleton - public static GlueClient createGlueClient(GlueHiveMetastoreConfig config, @ForGlueHiveMetastore Set executionInterceptors) - { - GlueClientBuilder glue = GlueClient.builder(); - - glue.overrideConfiguration(builder -> builder - .executionInterceptors(ImmutableList.copyOf(executionInterceptors)) - .retryStrategy(retryBuilder -> retryBuilder - .retryOnException(throwable -> throwable instanceof ConcurrentModificationException) - .backoffStrategy(BackoffStrategy.exponentialDelay( - java.time.Duration.ofMillis(20), - java.time.Duration.ofMillis(1500))) - .maxAttempts(config.getMaxGlueErrorRetries()))); - - Optional staticCredentialsProvider = getStaticCredentialsProvider(config); - - if (config.isUseWebIdentityTokenCredentialsProvider()) { - glue.credentialsProvider(StsWebIdentityTokenFileCredentialsProvider.builder() - .stsClient(getStsClient(config, staticCredentialsProvider)) - .asyncCredentialUpdateEnabled(true) - .build()); - } - else if (config.getIamRole().isPresent()) { - glue.credentialsProvider(StsAssumeRoleCredentialsProvider.builder() - .refreshRequest(request -> request - .roleArn(config.getIamRole().get()) - .roleSessionName("trino-session") - .externalId(config.getExternalId().orElse(null))) - .stsClient(getStsClient(config, staticCredentialsProvider)) - .asyncCredentialUpdateEnabled(true) - .build()); - } - else { - staticCredentialsProvider.ifPresent(glue::credentialsProvider); - } - - ApacheHttpClient.Builder httpClient = ApacheHttpClient.builder() - .maxConnections(config.getMaxGlueConnections()); - - if (config.getGlueEndpointUrl().isPresent()) { - checkArgument(config.getGlueRegion().isPresent(), "Glue region must be set when Glue endpoint URL is set"); - glue.region(Region.of(config.getGlueRegion().get())); - glue.endpointOverride(config.getGlueEndpointUrl().get()); - } - else if (config.getGlueRegion().isPresent()) { - glue.region(Region.of(config.getGlueRegion().get())); - } - else if (config.getPinGlueClientToCurrentRegion()) { - glue.region(DefaultAwsRegionProviderChain.builder().build().getRegion()); - } - - glue.httpClientBuilder(httpClient); - - return glue.build(); - } - - private static Optional getStaticCredentialsProvider(GlueHiveMetastoreConfig config) + @ForGlueHttpClient + public static SdkHttpClient createGlueSdkHttpClient(GlueHiveMetastoreConfig config) { - if (config.getAwsAccessKey().isPresent() && config.getAwsSecretKey().isPresent()) { - return Optional.of(StaticCredentialsProvider.create( - AwsBasicCredentials.create(config.getAwsAccessKey().get(), config.getAwsSecretKey().get()))); - } - return Optional.empty(); + return ApacheHttpClient.builder() + .maxConnections(config.getMaxGlueConnections()) + .build(); } - private static StsClient getStsClient(GlueHiveMetastoreConfig config, Optional staticCredentialsProvider) + @Provides + @Singleton + @ForStsHttpClient + public static SdkHttpClient createStsSdkHttpClient() { - StsClientBuilder sts = StsClient.builder(); - staticCredentialsProvider.ifPresent(sts::credentialsProvider); - - if (config.getGlueStsEndpointUrl().isPresent() && config.getGlueStsRegion().isPresent()) { - sts.endpointOverride(config.getGlueStsEndpointUrl().get()) - .region(Region.of(config.getGlueStsRegion().get())); - } - else if (config.getGlueStsRegion().isPresent()) { - sts.region(Region.of(config.getGlueStsRegion().get())); - } - else if (config.getPinGlueClientToCurrentRegion()) { - sts.region(DefaultAwsRegionProviderChain.builder().build().getRegion()); - } - - return sts.build(); + return ApacheHttpClient.builder().build(); } private static class TelemetryExecutionInterceptorProvider @@ -248,4 +192,45 @@ public ExecutionInterceptor get() .newExecutionInterceptor(); } } + + public static class GlueSecurityMappingModule + extends AbstractConfigurationAwareModule + { + @Override + protected void setup(Binder binder) + { + GlueSecurityMappingConfig config = buildConfigObject(GlueSecurityMappingConfig.class); + newOptionalBinder( + binder, + new TypeLiteral, IAMSecurityMapping, GlueSecurityMappingConfig>>() {}) + .setBinding() + .to(GlueSecurityMappingProvider.class) + .in(Scopes.SINGLETON); + + var mappingsBinder = binder.bind(new Key>>() + { + }); + if (config.getConfigFile().isPresent()) { + mappingsBinder.toInstance(new IAMSecurityMappingsFileSource<>( + config, new TypeReference<>() {})); + } + else if (config.getConfigUri().isPresent()) { + mappingsBinder.to(new TypeLiteral, IAMSecurityMapping, GlueSecurityMappingConfig>>() {}).in(SINGLETON); + httpClientBinder(binder).bindHttpClient("glue-security-mapping", GlueMetastoreModule.ForGlueSecurityMapping.class) + .withConfigDefaults(httpConfig -> httpConfig + .setRequestTimeout(new Duration(10, SECONDS)) + .setSelectorCount(1) + .setMinThreads(1)); + } + else { + throw new VerifyException("No security mapping source configured"); + } + } + } + + @Retention(RUNTIME) + @Target({FIELD, PARAMETER, METHOD}) + @BindingAnnotation + public @interface ForGlueSecurityMapping + { } } diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueSecurityMappingConfig.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueSecurityMappingConfig.java new file mode 100644 index 000000000000..4685cf3d60dc --- /dev/null +++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueSecurityMappingConfig.java @@ -0,0 +1,82 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hive.metastore.glue; + +import io.airlift.configuration.Config; +import io.airlift.configuration.ConfigDescription; +import io.airlift.units.Duration; +import io.trino.iam.aws.IAMSecurityMappingConfig; +import jakarta.validation.constraints.AssertTrue; + +import java.io.File; +import java.net.URI; + +public class GlueSecurityMappingConfig + extends IAMSecurityMappingConfig +{ + @Config("hive.metastore.glue.security-mapping.config-file") + @ConfigDescription("Path to the JSON security mappings file") + public GlueSecurityMappingConfig setConfigFile(File configFile) + { + this.configFile = configFile; + return this; + } + + @Config("hive.metastore.glue.security-mapping.config-uri") + @ConfigDescription("HTTP URI of the JSON security mappings") + public GlueSecurityMappingConfig setConfigUri(URI configUri) + { + this.configUri = configUri; + return this; + } + + @Config("hive.metastore.glue.security-mapping.json-pointer") + @ConfigDescription("JSON pointer (RFC 6901) to mappings inside JSON config") + public GlueSecurityMappingConfig setJsonPointer(String jsonPointer) + { + this.jsonPointer = jsonPointer; + return this; + } + + @Config("hive.metastore.glue.security-mapping.iam-role-credential-name") + @ConfigDescription("Name of the extra credential used to provide IAM role") + public GlueSecurityMappingConfig setRoleCredentialName(String roleCredentialName) + { + this.roleCredentialName = roleCredentialName; + return this; + } + + @Config("hive.metastore.glue.security-mapping.refresh-period") + @ConfigDescription("How often to refresh the security mapping configuration") + public GlueSecurityMappingConfig setRefreshPeriod(Duration refreshPeriod) + { + this.refreshPeriod = refreshPeriod; + return this; + } + + @Config("hive.metastore.glue.security-mapping.colon-replacement") + @ConfigDescription("Value used in place of colon for IAM role name in extra credentials") + public GlueSecurityMappingConfig setColonReplacement(String colonReplacement) + { + this.colonReplacement = colonReplacement; + return this; + } + + @Override + @AssertTrue(message = "Exactly one of hive.metastore.glue.security-mapping.config-file or hive.metastore.glue.security-mapping.config-uri must be set") + public boolean validateMappingsConfig() + { + return super.validateMappingsConfig(); + } +} diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueSecurityMappingEnabledConfig.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueSecurityMappingEnabledConfig.java new file mode 100644 index 000000000000..0a63145c3817 --- /dev/null +++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueSecurityMappingEnabledConfig.java @@ -0,0 +1,33 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hive.metastore.glue; + +import io.airlift.configuration.Config; + +public class GlueSecurityMappingEnabledConfig +{ + private boolean enabled; + + public boolean isEnabled() + { + return enabled; + } + + @Config("hive.metastore.glue.security-mapping.enabled") + public GlueSecurityMappingEnabledConfig setEnabled(boolean enabled) + { + this.enabled = enabled; + return this; + } +} diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueSecurityMappingProvider.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueSecurityMappingProvider.java new file mode 100644 index 000000000000..09a911629508 --- /dev/null +++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueSecurityMappingProvider.java @@ -0,0 +1,33 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hive.metastore.glue; + +import com.google.inject.Inject; +import io.trino.iam.aws.IAMSecurityMapping; +import io.trino.iam.aws.IAMSecurityMappingProvider; +import io.trino.iam.aws.IAMSecurityMappings; + +import java.util.function.Supplier; + +public class GlueSecurityMappingProvider + extends IAMSecurityMappingProvider, IAMSecurityMapping, GlueSecurityMappingConfig> +{ + @Inject + public GlueSecurityMappingProvider( + GlueSecurityMappingConfig config, + Supplier> mappings) + { + super(config, mappings); + } +} diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveGlueMetadataListing.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveGlueMetadataListing.java index 56a498e61a7d..82103e1dd412 100644 --- a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveGlueMetadataListing.java +++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveGlueMetadataListing.java @@ -18,24 +18,30 @@ import com.google.common.collect.ImmutableSet; import io.airlift.log.Logger; import io.trino.Session; +import io.trino.plugin.hive.metastore.glue.GlueClientFactory; import io.trino.plugin.hive.metastore.glue.GlueHiveMetastore; import io.trino.plugin.hive.metastore.glue.GlueHiveMetastoreConfig; import io.trino.plugin.tpch.TpchPlugin; +import io.trino.spi.block.TestingSession; +import io.trino.spi.connector.ConnectorSession; +import io.trino.spi.security.ConnectorIdentity; import io.trino.testing.AbstractTestQueryFramework; import io.trino.testing.DistributedQueryRunner; import io.trino.testing.QueryRunner; import io.trino.tpch.TpchTable; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Test; +import software.amazon.awssdk.http.SdkHttpClient; +import software.amazon.awssdk.http.apache.ApacheHttpClient; import software.amazon.awssdk.services.glue.GlueClient; import software.amazon.awssdk.services.glue.model.CreateTableRequest; import software.amazon.awssdk.services.glue.model.TableInput; import java.nio.file.Path; import java.util.List; +import java.util.Optional; import java.util.Set; -import static io.trino.plugin.hive.metastore.glue.GlueMetastoreModule.createGlueClient; import static io.trino.plugin.hive.metastore.glue.TestingGlueHiveMetastore.createTestingGlueHiveMetastore; import static io.trino.plugin.tpch.TpchMetadata.TINY_SCHEMA_NAME; import static io.trino.testing.QueryAssertions.copyTpchTables; @@ -141,7 +147,16 @@ private void createBrokenTable(List tablesInput, Path dataDirectory) { GlueHiveMetastoreConfig glueConfig = new GlueHiveMetastoreConfig() .setDefaultWarehouseDir(dataDirectory.toString()); - try (GlueClient glueClient = createGlueClient(glueConfig, ImmutableSet.of())) { + SdkHttpClient sdkHttpClient = ApacheHttpClient.builder().build(); + GlueClientFactory glueClientFactory = new GlueClientFactory( + glueConfig, + Optional.empty(), + ImmutableSet.of(), + sdkHttpClient, + sdkHttpClient); + ConnectorSession session = TestingSession.SESSION; + ConnectorIdentity identity = session.getIdentity(); + try (GlueClient glueClient = glueClientFactory.create(identity)) { for (TableInput tableInput : tablesInput) { CreateTableRequest createTableRequest = CreateTableRequest.builder() .databaseName(tpchSchema) diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestGlueSecurityMappingConfig.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestGlueSecurityMappingConfig.java new file mode 100644 index 000000000000..dcc9621cdebf --- /dev/null +++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestGlueSecurityMappingConfig.java @@ -0,0 +1,109 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hive.metastore.glue; + +import com.google.common.collect.ImmutableMap; +import io.airlift.configuration.ConfigurationFactory; +import io.airlift.configuration.validation.FileExists; +import io.airlift.units.Duration; +import org.junit.jupiter.api.Test; + +import java.io.File; +import java.io.IOException; +import java.net.URI; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Map; +import java.util.UUID; + +import static io.airlift.configuration.testing.ConfigAssertions.assertRecordedDefaults; +import static io.airlift.configuration.testing.ConfigAssertions.recordDefaults; +import static io.airlift.testing.ValidationAssertions.assertFailsValidation; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.assertj.core.api.Assertions.assertThat; + +public class TestGlueSecurityMappingConfig +{ + @Test + public void testDefaults() + { + assertRecordedDefaults(recordDefaults(GlueSecurityMappingConfig.class) + .setJsonPointer("") + .setConfigFile(null) + .setConfigUri(null) + .setRoleCredentialName(null) + .setRefreshPeriod(null) + .setColonReplacement(null)); + } + + @Test + public void testExplicitPropertyMappingsWithFile() + throws IOException + { + Path securityMappingConfigFile = Files.createTempFile(null, null); + + Map properties = ImmutableMap.builder() + .put("hive.metastore.glue.security-mapping.config-file", securityMappingConfigFile.toString()) + .put("hive.metastore.glue.security-mapping.json-pointer", "/data") + .put("hive.metastore.glue.security-mapping.iam-role-credential-name", "iam-role-credential-name") + .put("hive.metastore.glue.security-mapping.refresh-period", "13s") + .put("hive.metastore.glue.security-mapping.colon-replacement", "#") + .buildOrThrow(); + + ConfigurationFactory configurationFactory = new ConfigurationFactory(properties); + GlueSecurityMappingConfig config = configurationFactory.build(GlueSecurityMappingConfig.class); + + assertThat(config.getConfigFile()).contains(securityMappingConfigFile.toFile()); + assertThat(config.getConfigUri()).isEmpty(); + assertThat(config.getJsonPointer()).isEqualTo("/data"); + assertThat(config.getRoleCredentialName()).contains("iam-role-credential-name"); + assertThat(config.getRefreshPeriod()).contains(new Duration(13, SECONDS)); + assertThat(config.getColonReplacement()).contains("#"); + } + + @Test + public void testExplicitPropertyMappingsWithUrl() + { + Map properties = ImmutableMap.builder() + .put("hive.metastore.glue.security-mapping.config-uri", "http://test:1234/example") + .put("hive.metastore.glue.security-mapping.json-pointer", "/data") + .put("hive.metastore.glue.security-mapping.iam-role-credential-name", "iam-role-credential-name") + .put("hive.metastore.glue.security-mapping.sse-customer-key-credential-name", "sse-customer-key-credential-name") + .put("hive.metastore.glue.security-mapping.refresh-period", "13s") + .put("hive.metastore.glue.security-mapping.colon-replacement", "#") + .buildOrThrow(); + + ConfigurationFactory configurationFactory = new ConfigurationFactory(properties); + GlueSecurityMappingConfig config = configurationFactory.build(GlueSecurityMappingConfig.class); + + assertThat(config.getConfigFile()).isEmpty(); + assertThat(config.getConfigUri()).contains(URI.create("http://test:1234/example")); + assertThat(config.getJsonPointer()).isEqualTo("/data"); + assertThat(config.getRoleCredentialName()).contains("iam-role-credential-name"); + assertThat(config.getRefreshPeriod()).contains(new Duration(13, SECONDS)); + assertThat(config.getColonReplacement()).contains("#"); + } + + @Test + public void testConfigFileDoesNotExist() + { + File file = new File("/doesNotExist-" + UUID.randomUUID()); + assertFailsValidation( + new GlueSecurityMappingConfig() + .setConfigFile(file), + "configFile", + "file does not exist: " + file, + FileExists.class); + } +} diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestHiveConcurrentModificationGlueMetastore.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestHiveConcurrentModificationGlueMetastore.java index b9048e9bc326..2c227f615fad 100644 --- a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestHiveConcurrentModificationGlueMetastore.java +++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestHiveConcurrentModificationGlueMetastore.java @@ -14,10 +14,15 @@ package io.trino.plugin.hive.metastore.glue; import com.google.common.collect.ImmutableSet; +import io.trino.spi.block.TestingSession; +import io.trino.spi.connector.ConnectorSession; +import io.trino.spi.security.ConnectorIdentity; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInstance; import software.amazon.awssdk.awscore.exception.AwsErrorDetails; import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration; +import software.amazon.awssdk.http.SdkHttpClient; +import software.amazon.awssdk.http.apache.ApacheHttpClient; import software.amazon.awssdk.retries.api.RefreshRetryTokenRequest; import software.amazon.awssdk.retries.api.RefreshRetryTokenResponse; import software.amazon.awssdk.retries.api.RetryStrategy; @@ -25,7 +30,8 @@ import software.amazon.awssdk.services.glue.GlueClient; import software.amazon.awssdk.services.glue.model.ConcurrentModificationException; -import static io.trino.plugin.hive.metastore.glue.GlueMetastoreModule.createGlueClient; +import java.util.Optional; + import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.junit.jupiter.api.TestInstance.Lifecycle.PER_CLASS; @@ -36,7 +42,18 @@ public class TestHiveConcurrentModificationGlueMetastore @Test public void testGlueClientShouldRetryConcurrentModificationException() { - try (GlueClient glueClient = createGlueClient(new GlueHiveMetastoreConfig(), ImmutableSet.of())) { + GlueHiveMetastoreConfig config = new GlueHiveMetastoreConfig(); + SdkHttpClient sdkHttpClient = ApacheHttpClient.builder().build(); + GlueClientFactory glueClientFactory = new GlueClientFactory( + config, + Optional.empty(), + ImmutableSet.of(), + sdkHttpClient, + sdkHttpClient); + ConnectorSession session = TestingSession.SESSION; + ConnectorIdentity identity = session.getIdentity(); + + try (GlueClient glueClient = glueClientFactory.create(identity)) { ClientOverrideConfiguration clientOverrideConfiguration = glueClient.serviceClientConfiguration().overrideConfiguration(); RetryStrategy retryStrategy = clientOverrideConfiguration.retryStrategy().orElseThrow(); diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestingGlueHiveMetastore.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestingGlueHiveMetastore.java index 77f08c12049b..ebac145073de 100644 --- a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestingGlueHiveMetastore.java +++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestingGlueHiveMetastore.java @@ -15,18 +15,23 @@ import com.google.common.collect.ImmutableSet; import io.trino.plugin.hive.metastore.glue.GlueHiveMetastore.TableKind; +import io.trino.spi.block.TestingSession; import io.trino.spi.catalog.CatalogName; +import io.trino.spi.connector.ConnectorSession; +import io.trino.spi.security.ConnectorIdentity; +import software.amazon.awssdk.http.SdkHttpClient; +import software.amazon.awssdk.http.apache.ApacheHttpClient; import software.amazon.awssdk.services.glue.GlueClient; import java.io.IOException; import java.net.URI; import java.nio.file.Path; import java.util.EnumSet; +import java.util.Optional; import java.util.function.Consumer; import static com.google.common.base.Verify.verify; import static io.trino.plugin.hive.HiveTestUtils.HDFS_FILE_SYSTEM_FACTORY; -import static io.trino.plugin.hive.metastore.glue.GlueMetastoreModule.createGlueClient; import static java.nio.file.Files.createDirectories; import static java.nio.file.Files.exists; import static java.nio.file.Files.isDirectory; @@ -53,7 +58,16 @@ public static GlueHiveMetastore createTestingGlueHiveMetastore(URI warehouseUri, { GlueHiveMetastoreConfig glueConfig = new GlueHiveMetastoreConfig() .setDefaultWarehouseDir(warehouseUri.toString()); - GlueClient glueClient = createGlueClient(glueConfig, ImmutableSet.of()); + SdkHttpClient sdkHttpClient = ApacheHttpClient.builder().build(); + GlueClientFactory glueClientFactory = new GlueClientFactory( + glueConfig, + Optional.empty(), + ImmutableSet.of(), + sdkHttpClient, + sdkHttpClient); + ConnectorSession session = TestingSession.SESSION; + ConnectorIdentity identity = session.getIdentity(); + GlueClient glueClient = glueClientFactory.create(identity); registerResource.accept(glueClient); return new GlueHiveMetastore( glueClient, diff --git a/plugin/trino-iceberg/pom.xml b/plugin/trino-iceberg/pom.xml index 38f5db7e3616..0f9794e0f129 100644 --- a/plugin/trino-iceberg/pom.xml +++ b/plugin/trino-iceberg/pom.xml @@ -289,6 +289,17 @@ jmxutils + + software.amazon.awssdk + apache-client + + + commons-logging + commons-logging + + + + software.amazon.awssdk auth @@ -304,6 +315,11 @@ glue + + software.amazon.awssdk + http-client-spi + + software.amazon.awssdk identity-spi @@ -704,6 +720,8 @@ org.apache.parquet:parquet-common + software.amazon.awssdk:apache-client + software.amazon.awssdk:http-client-spi software.amazon.awssdk:aws-core software.amazon.awssdk:utils diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/GlueIcebergTableOperationsProvider.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/GlueIcebergTableOperationsProvider.java index 095597884d7b..8e5cae578d8a 100644 --- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/GlueIcebergTableOperationsProvider.java +++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/GlueIcebergTableOperationsProvider.java @@ -15,6 +15,7 @@ import com.google.inject.Inject; import io.trino.filesystem.TrinoFileSystemFactory; +import io.trino.plugin.hive.metastore.glue.GlueClientFactory; import io.trino.plugin.hive.metastore.glue.GlueMetastoreStats; import io.trino.plugin.iceberg.catalog.IcebergTableOperations; import io.trino.plugin.iceberg.catalog.IcebergTableOperationsProvider; @@ -36,7 +37,7 @@ public class GlueIcebergTableOperationsProvider private final ForwardingFileIoFactory fileIoFactory; private final TypeManager typeManager; private final boolean cacheTableMetadata; - private final GlueClient glueClient; + private final GlueClientFactory glueClientFactory; private final GlueMetastoreStats stats; @Inject @@ -46,14 +47,14 @@ public GlueIcebergTableOperationsProvider( TypeManager typeManager, IcebergGlueCatalogConfig catalogConfig, GlueMetastoreStats stats, - GlueClient glueClient) + GlueClientFactory glueClientFactory) { this.fileSystemFactory = requireNonNull(fileSystemFactory, "fileSystemFactory is null"); this.fileIoFactory = requireNonNull(fileIoFactory, "fileIoFactory is null"); this.typeManager = requireNonNull(typeManager, "typeManager is null"); this.cacheTableMetadata = catalogConfig.isCacheTableMetadata(); this.stats = requireNonNull(stats, "stats is null"); - this.glueClient = requireNonNull(glueClient, "glueClient is null"); + this.glueClientFactory = requireNonNull(glueClientFactory, "glueClientFactory is null"); } @Override @@ -65,6 +66,7 @@ public IcebergTableOperations createTableOperations( Optional owner, Optional location) { + GlueClient glueClient = glueClientFactory.create(session.getIdentity()); return new GlueIcebergTableOperations( typeManager, cacheTableMetadata, diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/TrinoGlueCatalogFactory.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/TrinoGlueCatalogFactory.java index 1aae3ab1a709..45a671b85dc0 100644 --- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/TrinoGlueCatalogFactory.java +++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/TrinoGlueCatalogFactory.java @@ -17,6 +17,7 @@ import io.airlift.concurrent.BoundedExecutor; import io.trino.filesystem.TrinoFileSystemFactory; import io.trino.plugin.hive.NodeVersion; +import io.trino.plugin.hive.metastore.glue.GlueClientFactory; import io.trino.plugin.hive.metastore.glue.GlueHiveMetastoreConfig; import io.trino.plugin.hive.metastore.glue.GlueMetastoreStats; import io.trino.plugin.hive.security.UsingSystemSecurity; @@ -31,7 +32,6 @@ import io.trino.spi.type.TypeManager; import org.weakref.jmx.Flatten; import org.weakref.jmx.Managed; -import software.amazon.awssdk.services.glue.GlueClient; import java.util.Optional; import java.util.concurrent.Executor; @@ -51,7 +51,7 @@ public class TrinoGlueCatalogFactory private final IcebergTableOperationsProvider tableOperationsProvider; private final String trinoVersion; private final Optional defaultSchemaLocation; - private final GlueClient glueClient; + private final GlueClientFactory glueClientFactory; private final boolean isUniqueTableLocation; private final boolean hideMaterializedViewStorageTable; private final GlueMetastoreStats stats; @@ -71,7 +71,7 @@ public TrinoGlueCatalogFactory( IcebergGlueCatalogConfig catalogConfig, @UsingSystemSecurity boolean usingSystemSecurity, GlueMetastoreStats stats, - GlueClient glueClient, + GlueClientFactory glueClientFactory, @ForIcebergMetadata ExecutorService metadataExecutorService) { this.catalogName = requireNonNull(catalogName, "catalogName is null"); @@ -82,7 +82,7 @@ public TrinoGlueCatalogFactory( this.tableOperationsProvider = requireNonNull(tableOperationsProvider, "tableOperationsProvider is null"); this.trinoVersion = nodeVersion.toString(); this.defaultSchemaLocation = glueConfig.getDefaultWarehouseDir(); - this.glueClient = requireNonNull(glueClient, "glueClient is null"); + this.glueClientFactory = requireNonNull(glueClientFactory, "glueClientFactory is null"); this.isUniqueTableLocation = icebergConfig.isUniqueTableLocation(); this.hideMaterializedViewStorageTable = icebergConfig.isHideMaterializedViewStorageTable(); this.stats = requireNonNull(stats, "stats is null"); @@ -113,7 +113,7 @@ public TrinoCatalog create(ConnectorIdentity identity) cacheTableMetadata, tableOperationsProvider, trinoVersion, - glueClient, + glueClientFactory.create(identity), stats, isUsingSystemSecurity, defaultSchemaLocation, diff --git a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/glue/TestTrinoGlueCatalog.java b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/glue/TestTrinoGlueCatalog.java index 79400318e0be..8231fffb4726 100644 --- a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/glue/TestTrinoGlueCatalog.java +++ b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/glue/TestTrinoGlueCatalog.java @@ -15,10 +15,13 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; import io.airlift.log.Logger; import io.trino.filesystem.TrinoFileSystemFactory; import io.trino.metastore.TableInfo; import io.trino.plugin.hive.NodeVersion; +import io.trino.plugin.hive.metastore.glue.GlueClientFactory; +import io.trino.plugin.hive.metastore.glue.GlueHiveMetastoreConfig; import io.trino.plugin.hive.metastore.glue.GlueMetastoreStats; import io.trino.plugin.iceberg.CommitTaskData; import io.trino.plugin.iceberg.IcebergConfig; @@ -26,16 +29,20 @@ import io.trino.plugin.iceberg.TableStatisticsWriter; import io.trino.plugin.iceberg.catalog.BaseTrinoCatalogTest; import io.trino.plugin.iceberg.catalog.TrinoCatalog; +import io.trino.spi.block.TestingSession; import io.trino.spi.catalog.CatalogName; import io.trino.spi.connector.ConnectorMaterializedViewDefinition; import io.trino.spi.connector.ConnectorMetadata; import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.MaterializedViewNotFoundException; import io.trino.spi.connector.SchemaTableName; +import io.trino.spi.security.ConnectorIdentity; import io.trino.spi.security.PrincipalType; import io.trino.spi.security.TrinoPrincipal; import io.trino.spi.type.TestingTypeManager; import org.junit.jupiter.api.Test; +import software.amazon.awssdk.http.SdkHttpClient; +import software.amazon.awssdk.http.apache.ApacheHttpClient; import software.amazon.awssdk.services.glue.GlueClient; import java.io.File; @@ -76,7 +83,16 @@ protected TrinoCatalog createTrinoCatalog(boolean useUniqueTableLocations) private TrinoCatalog createGlueTrinoCatalog(boolean useUniqueTableLocations, boolean useSystemSecurity) { - GlueClient glueClient = GlueClient.create(); + GlueHiveMetastoreConfig glueConfig = new GlueHiveMetastoreConfig(); + SdkHttpClient sdkHttpClient = ApacheHttpClient.builder().build(); + GlueClientFactory glueClientFactory = new GlueClientFactory( + glueConfig, + Optional.empty(), + ImmutableSet.of(), + sdkHttpClient, + sdkHttpClient); + ConnectorSession session = TestingSession.SESSION; + ConnectorIdentity identity = session.getIdentity(); IcebergGlueCatalogConfig catalogConfig = new IcebergGlueCatalogConfig(); return new TrinoGlueCatalog( new CatalogName("catalog_name"), @@ -90,9 +106,9 @@ private TrinoCatalog createGlueTrinoCatalog(boolean useUniqueTableLocations, boo TESTING_TYPE_MANAGER, catalogConfig, new GlueMetastoreStats(), - glueClient), + glueClientFactory), "test", - glueClient, + glueClientFactory.create(identity), new GlueMetastoreStats(), useSystemSecurity, Optional.empty(), @@ -111,7 +127,17 @@ public void testNonLowercaseGlueDatabase() // Trino schema names are always lowercase (until https://github.com/trinodb/trino/issues/17) String trinoSchemaName = databaseName.toLowerCase(ENGLISH); - GlueClient glueClient = GlueClient.create(); + GlueHiveMetastoreConfig glueConfig = new GlueHiveMetastoreConfig(); + SdkHttpClient sdkHttpClient = ApacheHttpClient.builder().build(); + GlueClientFactory glueClientFactory = new GlueClientFactory( + glueConfig, + Optional.empty(), + ImmutableSet.of(), + sdkHttpClient, + sdkHttpClient); + ConnectorSession session = TestingSession.SESSION; + ConnectorIdentity identity = session.getIdentity(); + GlueClient glueClient = glueClientFactory.create(identity); glueClient.createDatabase(database -> database .databaseInput(input -> input // Currently this is actually stored in lowercase @@ -217,7 +243,17 @@ public void testDefaultLocation() tmpDirectory.toFile().deleteOnExit(); TrinoFileSystemFactory fileSystemFactory = HDFS_FILE_SYSTEM_FACTORY; - GlueClient glueClient = GlueClient.create(); + GlueHiveMetastoreConfig glueConfig = new GlueHiveMetastoreConfig(); + SdkHttpClient sdkHttpClient = ApacheHttpClient.builder().build(); + GlueClientFactory glueClientFactory = new GlueClientFactory( + glueConfig, + Optional.empty(), + ImmutableSet.of(), + sdkHttpClient, + sdkHttpClient); + ConnectorSession session = TestingSession.SESSION; + ConnectorIdentity identity = session.getIdentity(); + GlueClient glueClient = glueClientFactory.create(identity); IcebergGlueCatalogConfig catalogConfig = new IcebergGlueCatalogConfig(); TrinoCatalog catalogWithDefaultLocation = new TrinoGlueCatalog( new CatalogName("catalog_name"), @@ -231,7 +267,7 @@ public void testDefaultLocation() TESTING_TYPE_MANAGER, catalogConfig, new GlueMetastoreStats(), - glueClient), + glueClientFactory), "test", glueClient, new GlueMetastoreStats(), diff --git a/pom.xml b/pom.xml index d855c46ac8f1..36c19908efba 100644 --- a/pom.xml +++ b/pom.xml @@ -52,6 +52,7 @@ lib/trino-geospatial-toolkit lib/trino-hdfs lib/trino-hive-formats + lib/trino-iam-aws lib/trino-matching lib/trino-memory-context lib/trino-metastore @@ -1156,6 +1157,12 @@ ${project.version} + + io.trino + trino-iam-aws + ${project.version} + + io.trino trino-iceberg @@ -2597,7 +2604,8 @@ Project name must be set in the pom.xml - Trino requires Temurin or Oracle JDK for development. Other vendors are not recommended due to lack of testing coverage. + Trino requires Temurin or Oracle JDK for development. Other vendors are not + recommended due to lack of testing coverage. Eclipse Adoptium