diff --git a/.gitignore b/.gitignore index 3d4ec33d549e..66e2d3a4dc15 100644 --- a/.gitignore +++ b/.gitignore @@ -55,3 +55,4 @@ derby.log # generated by docs build *.pyc +/.vscode/settings.json diff --git a/cdap-api/src/main/java/io/cdap/cdap/api/security/store/SecureStoreManager.java b/cdap-api/src/main/java/io/cdap/cdap/api/security/store/SecureStoreManager.java index 30372c80f2a8..258b6d068188 100644 --- a/cdap-api/src/main/java/io/cdap/cdap/api/security/store/SecureStoreManager.java +++ b/cdap-api/src/main/java/io/cdap/cdap/api/security/store/SecureStoreManager.java @@ -17,6 +17,7 @@ package io.cdap.cdap.api.security.store; import io.cdap.cdap.api.annotation.Beta; +import io.cdap.cdap.api.security.store.lease.SecureStoreLease; import java.io.IOException; import java.util.Map; import javax.annotation.Nullable; @@ -50,4 +51,29 @@ void put(String namespace, String name, String data, @Nullable String descriptio * @throws Exception If the specified namespace or name does not exist */ void delete(String namespace, String name) throws Exception; + + /** + * Attempts to acquire a lease lock on a secret resource. + * + * @param namespace The namespace that this key belongs to + * @param name Name of the element + * @param timeoutMs Lock timeout in milliseconds before lease is considered expired + * @return {@link SecureStoreLease} indicating acquisition success and lock details + * @throws IOException If lock acquisition fails due to underlying storage errors + */ + default SecureStoreLease acquireLease(String namespace, String name, long timeoutMs) throws IOException { + return SecureStoreLease.failed(); + } + + /** + * Releases an acquired lease lock on a secret resource. + * + * @param namespace The namespace that this key belongs to + * @param name Name of the element + * @param lease {@link SecureStoreLease} to release + * @throws IOException If lock release fails due to underlying storage errors + */ + default void releaseLease(String namespace, String name, SecureStoreLease lease) throws IOException { + // default no-op + } } diff --git a/cdap-api/src/main/java/io/cdap/cdap/api/security/store/lease/SecureStoreLease.java b/cdap-api/src/main/java/io/cdap/cdap/api/security/store/lease/SecureStoreLease.java new file mode 100644 index 000000000000..b3748187dede --- /dev/null +++ b/cdap-api/src/main/java/io/cdap/cdap/api/security/store/lease/SecureStoreLease.java @@ -0,0 +1,86 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * 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.cdap.cdap.api.security.store.lease; + +import java.util.Objects; + +/** + * API model representing a distributed lease lock on a secret or credential update in SecureStore. + *

+ * Note: This is the caller-facing API counterpart to {@code io.cdap.cdap.securestore.spi.lease.SecretLease} + * in the SPI layer, mirroring the {@link io.cdap.cdap.api.security.store.SecureStoreMetadata} vs {@code SecretMetadata} + * architectural pattern in CDAP. + */ +public class SecureStoreLease { + private final boolean acquired; + private final String lockTimestamp; + private final String lockHolder; + + private SecureStoreLease(boolean acquired, String lockTimestamp, String lockHolder) { + this.acquired = acquired; + this.lockTimestamp = lockTimestamp; + this.lockHolder = lockHolder; + } + + public static SecureStoreLease acquired(String lockTimestamp, String lockHolder) { + return new SecureStoreLease(true, lockTimestamp, lockHolder); + } + + public static SecureStoreLease failed() { + return new SecureStoreLease(false, null, null); + } + + public boolean isAcquired() { + return acquired; + } + + public String getLockTimestamp() { + return lockTimestamp; + } + + public String getLockHolder() { + return lockHolder; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + SecureStoreLease that = (SecureStoreLease) o; + return acquired == that.acquired + && Objects.equals(lockTimestamp, that.lockTimestamp) + && Objects.equals(lockHolder, that.lockHolder); + } + + @Override + public int hashCode() { + return Objects.hash(acquired, lockTimestamp, lockHolder); + } + + @Override + public String toString() { + return "SecureStoreLease{" + + "acquired=" + acquired + + ", lockTimestamp='" + lockTimestamp + '\'' + + ", lockHolder='" + lockHolder + '\'' + + '}'; + } +} diff --git a/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/DefaultAdmin.java b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/DefaultAdmin.java index c84e59e3074a..a3c6723fd6f9 100644 --- a/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/DefaultAdmin.java +++ b/cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/DefaultAdmin.java @@ -22,6 +22,7 @@ import io.cdap.cdap.api.messaging.TopicAlreadyExistsException; import io.cdap.cdap.api.messaging.TopicNotFoundException; import io.cdap.cdap.api.security.store.SecureStoreManager; +import io.cdap.cdap.api.security.store.lease.SecureStoreLease; import io.cdap.cdap.common.NamespaceNotFoundException; import io.cdap.cdap.common.namespace.NamespaceQueryAdmin; import io.cdap.cdap.common.service.Retries; @@ -83,6 +84,28 @@ public void delete(String namespace, String name) throws Exception { Retries.runWithRetries(() -> secureStoreManager.delete(namespace, name), retryStrategy); } + @Override + public SecureStoreLease acquireLease(String namespace, String name, long timeoutMs) throws IOException { + try { + return Retries.callWithRetries(() -> secureStoreManager.acquireLease(namespace, name, timeoutMs), retryStrategy); + } catch (IOException | RuntimeException e) { + throw e; + } catch (Exception e) { + throw new IOException(e); + } + } + + @Override + public void releaseLease(String namespace, String name, SecureStoreLease lease) throws IOException { + try { + Retries.runWithRetries(() -> secureStoreManager.releaseLease(namespace, name, lease), retryStrategy); + } catch (IOException | RuntimeException e) { + throw e; + } catch (Exception e) { + throw new IOException(e); + } + } + @Override public void createTopic(final String topic) throws TopicAlreadyExistsException, IOException { if (messagingAdmin == null) { diff --git a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/OAuthAccessToken.java b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/OAuthAccessToken.java index 7afff882947b..f2dc0b5ade96 100644 --- a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/OAuthAccessToken.java +++ b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/OAuthAccessToken.java @@ -21,15 +21,42 @@ */ public class OAuthAccessToken { private final String accessToken; + private final long expiresAt; + private final String identityUrl; public OAuthAccessToken(String accessToken) { + this(accessToken, 0L, null); + } + + public OAuthAccessToken(String accessToken, long expiresAt) { + this(accessToken, expiresAt, null); + } + + public OAuthAccessToken(String accessToken, long expiresAt, String identityUrl) { this.accessToken = accessToken; + this.expiresAt = expiresAt; + this.identityUrl = identityUrl; } public String getAccessToken() { return accessToken; } + public long getExpiresAt() { + return expiresAt; + } + + public String getIdentityUrl() { + return identityUrl; + } + + public boolean isExpired(long safetyBufferMs) { + if (expiresAt <= 0) { + return false; + } + return (expiresAt - System.currentTimeMillis()) <= safetyBufferMs; + } + public static Builder newBuilder() { return new Builder(); } @@ -39,6 +66,8 @@ public static Builder newBuilder() { */ public static class Builder { private String accessToken; + private long expiresAt; + private String identityUrl; public Builder() {} @@ -47,9 +76,19 @@ public Builder withAccessToken(String accessToken) { return this; } + public Builder withExpiresAt(long expiresAt) { + this.expiresAt = expiresAt; + return this; + } + + public Builder withIdentityUrl(String identityUrl) { + this.identityUrl = identityUrl; + return this; + } + public OAuthAccessToken build() { Preconditions.checkNotNull(accessToken, "OAuth access token missing"); - return new OAuthAccessToken(accessToken); + return new OAuthAccessToken(accessToken, expiresAt, identityUrl); } } } diff --git a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/OAuthProvider.java b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/OAuthProvider.java index 0aeea2777325..1bd074e33961 100644 --- a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/OAuthProvider.java +++ b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/OAuthProvider.java @@ -32,6 +32,8 @@ public class OAuthProvider { // Optional string to send as a USER_AGENT header @Nullable private final String userAgent; + private final AuthType authType; + private final RefreshType refreshType; public OAuthProvider(String name, String loginURL, @@ -39,12 +41,25 @@ public OAuthProvider(String name, @Nullable OAuthClientCredentials clientCreds, @Nullable CredentialEncodingStrategy strategy, @Nullable String userAgent) { + this(name, loginURL, tokenRefreshURL, clientCreds, strategy, userAgent, AuthType.STANDARD, RefreshType.STANDARD); + } + + public OAuthProvider(String name, + String loginURL, + String tokenRefreshURL, + @Nullable OAuthClientCredentials clientCreds, + @Nullable CredentialEncodingStrategy strategy, + @Nullable String userAgent, + @Nullable AuthType authType, + @Nullable RefreshType refreshType) { this.name = name; this.loginURL = loginURL; this.tokenRefreshURL = tokenRefreshURL; this.clientCreds = clientCreds; this.strategy = strategy; this.userAgent = userAgent; + this.authType = authType == null ? AuthType.STANDARD : authType; + this.refreshType = refreshType == null ? RefreshType.STANDARD : refreshType; } public String getName() { @@ -74,6 +89,14 @@ public String getUserAgent() { return userAgent; } + public AuthType getAuthType() { + return authType; + } + + public RefreshType getRefreshType() { + return refreshType; + } + public enum CredentialEncodingStrategy { // (default) Sends client ID & secret as part of the POST request body FORM_BODY, @@ -81,6 +104,16 @@ public enum CredentialEncodingStrategy { BASIC_AUTH, } + public enum AuthType { + STANDARD, + PKCE + } + + public enum RefreshType { + STANDARD, + RTR + } + public static Builder newBuilder() { return new Builder(); } @@ -95,6 +128,8 @@ public static class Builder { private OAuthClientCredentials clientCreds; private CredentialEncodingStrategy strategy; private String userAgent; + private AuthType authType; + private RefreshType refreshType; public Builder() {} @@ -128,6 +163,16 @@ public Builder withUserAgent(@Nullable String userAgent) { return this; } + public Builder withAuthType(@Nullable AuthType authType) { + this.authType = authType; + return this; + } + + public Builder withRefreshType(@Nullable RefreshType refreshType) { + this.refreshType = refreshType; + return this; + } + public OAuthProvider build() { Preconditions.checkNotNull(name, "OAuth provider name missing"); Preconditions.checkNotNull(loginURL, "Login URL missing"); @@ -136,7 +181,13 @@ public OAuthProvider build() { if (strategy == null) { this.strategy = CredentialEncodingStrategy.FORM_BODY; } - return new OAuthProvider(name, loginURL, tokenRefreshURL, clientCreds, strategy, userAgent); + if (authType == null) { + this.authType = AuthType.STANDARD; + } + if (refreshType == null) { + this.refreshType = RefreshType.STANDARD; + } + return new OAuthProvider(name, loginURL, tokenRefreshURL, clientCreds, strategy, userAgent, authType, refreshType); } } } diff --git a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/OAuthStore.java b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/OAuthStore.java index 6f3a2ea67de3..baeb910d72ca 100644 --- a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/OAuthStore.java +++ b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/OAuthStore.java @@ -18,6 +18,7 @@ import com.google.gson.JsonSyntaxException; import io.cdap.cdap.api.security.store.SecureStore; import io.cdap.cdap.api.security.store.SecureStoreManager; +import io.cdap.cdap.api.security.store.lease.SecureStoreLease; import io.cdap.cdap.proto.id.NamespaceId; import io.cdap.cdap.spi.data.InvalidFieldException; import io.cdap.cdap.spi.data.StructuredRow; @@ -29,27 +30,38 @@ import io.cdap.cdap.spi.data.table.field.Fields; import io.cdap.cdap.spi.data.transaction.TransactionRunner; import io.cdap.cdap.spi.data.transaction.TransactionRunners; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Map; +import java.util.Objects; import java.util.Optional; +import java.util.ServiceLoader; /** * Schema for OAuth store. */ public class OAuthStore { + private static final Logger LOG = LoggerFactory.getLogger(OAuthStore.class); private static final String OAUTH_PROVIDER_COL = "oauthprovider"; private static final String LOGIN_URL_COL = "loginurl"; private static final String TOKEN_REFRESH_URL_COL = "tokenrefreshurl"; private static final String CREDENTIAL_ENCODING_STRATEGY_COL = "credentialencodingstrategy"; private static final String USER_AGENT_COL = "useragent"; + private static final String OAUTH_TYPE_COL = "oauthtype"; + private static final String AUTH_TYPE_COL = "authtype"; + private static final String REFRESH_TYPE_COL = "refreshtype"; private static final String CLIENT_CREDS_KEY_PREFIX = "oauthclientcreds"; private static final String ACCESS_TOKEN_KEY_PREFIX = "oauthaccesstoken"; private static final String REFRESH_TOKEN_KEY_PREFIX = "oauthrefreshtoken"; private static final Gson GSON = new Gson(); + private final TransactionRunner transactionRunner; private final SecureStore secureStore; private final SecureStoreManager secureStoreManager; @@ -61,10 +73,23 @@ public class OAuthStore { Fields.stringType(LOGIN_URL_COL), Fields.stringType(TOKEN_REFRESH_URL_COL), Fields.stringType(CREDENTIAL_ENCODING_STRATEGY_COL), - Fields.stringType(USER_AGENT_COL)) + Fields.stringType(USER_AGENT_COL), + Fields.stringType(OAUTH_TYPE_COL), + Fields.stringType(AUTH_TYPE_COL), + Fields.stringType(REFRESH_TYPE_COL)) .withPrimaryKeys(OAUTH_PROVIDER_COL) .build(); + public static final StructuredTableSpecification LEASE_TABLE_SPEC = new StructuredTableSpecification.Builder() + .withId(new StructuredTableId("app_oauth_leases")) + .withFields(Fields.stringType("provider"), + Fields.stringType("credential_id"), + Fields.stringType("state"), + Fields.longType("lock_timestamp"), + Fields.stringType("lock_holder")) + .withPrimaryKeys("provider", "credential_id") + .build(); + public OAuthStore( TransactionRunner transactionRunner, SecureStore secureStore, @@ -74,6 +99,18 @@ public OAuthStore( this.secureStoreManager = secureStoreManager; } + public SecureStoreLease acquireLease(String provider, String credentialId, long timeoutMs) throws IOException { + String namespace = NamespaceId.SYSTEM.getNamespace(); + String key = getRefreshTokenKey(provider, credentialId); + return secureStoreManager.acquireLease(namespace, key, timeoutMs); + } + + public void releaseLease(String provider, String credentialId, SecureStoreLease lease) throws IOException { + String namespace = NamespaceId.SYSTEM.getNamespace(); + String key = getRefreshTokenKey(provider, credentialId); + secureStoreManager.releaseLease(namespace, key, lease); + } + /** * Create/update an OAuth provider. * @@ -311,6 +348,17 @@ private static OAuthProvider fromRow(StructuredRow row, OAuthClientCredentials c String tokenRefreshURL = row.getString(TOKEN_REFRESH_URL_COL); String credentialEncodingStrategy = row.getString(CREDENTIAL_ENCODING_STRATEGY_COL); String userAgent = row.getString(USER_AGENT_COL); + String authTypeStr = row.getString(AUTH_TYPE_COL); + String refreshTypeStr = row.getString(REFRESH_TYPE_COL); + + if (authTypeStr == null && refreshTypeStr == null) { + String oauthTypeStr = row.getString(OAUTH_TYPE_COL); + if ("REFRESH_TOKEN_ROTATION".equalsIgnoreCase(oauthTypeStr)) { + refreshTypeStr = "RTR"; + } else if ("PKCE".equalsIgnoreCase(oauthTypeStr)) { + authTypeStr = "PKCE"; + } + } return OAuthProvider.newBuilder() .withName(name) @@ -321,11 +369,17 @@ private static OAuthProvider fromRow(StructuredRow row, OAuthClientCredentials c Optional.ofNullable(credentialEncodingStrategy) .map(OAuthProvider.CredentialEncodingStrategy::valueOf).orElse(null)) .withUserAgent(userAgent) + .withAuthType( + Optional.ofNullable(authTypeStr) + .map(OAuthProvider.AuthType::valueOf).orElse(OAuthProvider.AuthType.STANDARD)) + .withRefreshType( + Optional.ofNullable(refreshTypeStr) + .map(OAuthProvider.RefreshType::valueOf).orElse(OAuthProvider.RefreshType.STANDARD)) .build(); } private static List> getRow(OAuthProvider oauthProvider) { - List> fields = new ArrayList<>(3); + List> fields = new ArrayList<>(7); fields.add(Fields.stringField(OAUTH_PROVIDER_COL, oauthProvider.getName())); fields.add(Fields.stringField(LOGIN_URL_COL, oauthProvider.getLoginURL())); fields.add(Fields.stringField(TOKEN_REFRESH_URL_COL, oauthProvider.getTokenRefreshURL())); @@ -333,6 +387,8 @@ private static List> getRow(OAuthProvider oauthProvider) { CREDENTIAL_ENCODING_STRATEGY_COL, oauthProvider.getCredentialEncodingStrategy().toString())); fields.add(Fields.stringField(USER_AGENT_COL, oauthProvider.getUserAgent())); + fields.add(Fields.stringField(AUTH_TYPE_COL, oauthProvider.getAuthType().toString())); + fields.add(Fields.stringField(REFRESH_TYPE_COL, oauthProvider.getRefreshType().toString())); return fields; } } diff --git a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/PutOAuthProviderRequest.java b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/PutOAuthProviderRequest.java index b872f881ae56..a008ed26c73c 100644 --- a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/PutOAuthProviderRequest.java +++ b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/PutOAuthProviderRequest.java @@ -27,6 +27,8 @@ public class PutOAuthProviderRequest { private final String clientSecret; private final OAuthProvider.CredentialEncodingStrategy strategy; private final String userAgent; + private final OAuthProvider.AuthType authType; + private final OAuthProvider.RefreshType refreshType; public PutOAuthProviderRequest( String loginURL, @@ -35,12 +37,26 @@ public PutOAuthProviderRequest( String clientSecret, OAuthProvider.CredentialEncodingStrategy strategy, String userAgent) { + this(loginURL, tokenRefreshURL, clientId, clientSecret, strategy, userAgent, OAuthProvider.AuthType.STANDARD, OAuthProvider.RefreshType.STANDARD); + } + + public PutOAuthProviderRequest( + String loginURL, + String tokenRefreshURL, + String clientId, + String clientSecret, + OAuthProvider.CredentialEncodingStrategy strategy, + String userAgent, + OAuthProvider.AuthType authType, + OAuthProvider.RefreshType refreshType) { this.loginURL = loginURL; this.tokenRefreshURL = tokenRefreshURL; this.clientId = clientId; this.clientSecret = clientSecret; this.strategy = strategy; this.userAgent = userAgent; + this.authType = authType == null ? OAuthProvider.AuthType.STANDARD : authType; + this.refreshType = refreshType == null ? OAuthProvider.RefreshType.STANDARD : refreshType; } public String getLoginURL() { @@ -66,4 +82,12 @@ public OAuthProvider.CredentialEncodingStrategy getCredentialEncodingStrategy() public String getUserAgent() { return userAgent; } + + public OAuthProvider.AuthType getAuthType() { + return authType == null ? OAuthProvider.AuthType.STANDARD : authType; + } + + public OAuthProvider.RefreshType getRefreshType() { + return refreshType == null ? OAuthProvider.RefreshType.STANDARD : refreshType; + } } diff --git a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/RefreshTokenResponse.java b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/RefreshTokenResponse.java index b61a1a5a870f..3bcf9945c857 100644 --- a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/RefreshTokenResponse.java +++ b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/oauth/RefreshTokenResponse.java @@ -36,6 +36,8 @@ public class RefreshTokenResponse { private final String tokenType; @SerializedName("issued_at") private final String issuedAt; + @SerializedName("expires_in") + private final long expiresIn; public RefreshTokenResponse( String accessToken, @@ -45,7 +47,8 @@ public RefreshTokenResponse( String instanceURL, String id, String tokenType, - String issuedAt) { + String issuedAt, + long expiresIn) { this.accessToken = accessToken; this.refreshToken = refreshToken; this.signature = signature; @@ -54,6 +57,7 @@ public RefreshTokenResponse( this.id = id; this.tokenType = tokenType; this.issuedAt = issuedAt; + this.expiresIn = expiresIn; } public String getAccessToken() { @@ -88,4 +92,7 @@ public String getIssuedAt() { return issuedAt; } + public long getExpiresIn() { + return expiresIn; + } } diff --git a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/service/OAuthHandler.java b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/service/OAuthHandler.java index 673d19c38c8e..2b44891b9725 100644 --- a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/service/OAuthHandler.java +++ b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/service/OAuthHandler.java @@ -20,6 +20,7 @@ import com.google.gson.Gson; import com.google.gson.GsonBuilder; import com.google.gson.JsonSyntaxException; +import io.cdap.cdap.api.security.store.lease.SecureStoreLease; import io.cdap.cdap.api.service.http.AbstractSystemHttpServiceHandler; import io.cdap.cdap.api.service.http.HttpServiceRequest; import io.cdap.cdap.api.service.http.HttpServiceResponder; @@ -141,6 +142,8 @@ public void putOAuthProvider(HttpServiceRequest request, HttpServiceResponder re .withClientCredentials(clientCredentials) .withCredentialEncodingStrategy(strategy) .withUserAgent(userAgent) + .withAuthType(putOAuthProviderRequest.getAuthType()) + .withRefreshType(putOAuthProviderRequest.getRefreshType()) .build(); oauthStore.writeProvider(provider, reuseClientCredentials); responder.sendStatus(HttpURLConnection.HTTP_OK); @@ -252,10 +255,15 @@ public void putOAuthCredential(HttpServiceRequest request, HttpServiceResponder .withRedirectURI(putOAuthCredentialRequest.getRedirectURI()) .build(); oauthStore.writeRefreshToken(provider, credentialId, refreshToken); + + // For RTR, also store the initial Access Token in OAuthStore + if (OAuthProvider.RefreshType.RTR.equals(oauthProvider.getRefreshType())) { + writeAccessTokenFromResponse(provider, credentialId, refreshTokenResponse); + } } catch (NullPointerException e) { throw new OAuthServiceException(HttpURLConnection.HTTP_BAD_REQUEST, e.getMessage(), e); } catch (OAuthStoreException e) { - throw new OAuthServiceException(HttpURLConnection.HTTP_BAD_REQUEST, "Failed to write refresh token", e); + throw new OAuthServiceException(HttpURLConnection.HTTP_BAD_REQUEST, "Failed to write credentials", e); } } else { // Refresh token call gave us an access token without a refresh token. @@ -294,19 +302,172 @@ public void getOAuthCredential(HttpServiceRequest request, HttpServiceResponder @PathParam("credential") String credentialId) { try { OAuthProvider oauthProvider = getProvider(provider); - Optional oAuthAccessToken = getAccessToken(provider, credentialId); - // If found, send the long-lived access token - if (oAuthAccessToken.isPresent()) { - responder.sendString(GSON.toJson( - new GetAccessTokenResponse(oAuthAccessToken.get().getAccessToken(), ""))); + if (OAuthProvider.RefreshType.RTR.equals(oauthProvider.getRefreshType())) { + getOAuthCredentialWithRefreshTokenRotation(responder, oauthProvider, provider, credentialId); + } else { + getOAuthCredentialStandard(responder, oauthProvider, provider, credentialId); + } + } catch (OAuthServiceException e) { + e.respond(responder); + } + } + + private void getOAuthCredentialStandard(HttpServiceResponder responder, + OAuthProvider oauthProvider, + String provider, + String credentialId) throws OAuthServiceException { + // 1. Check if long-lived access token is stored (for permanent token providers) + Optional oAuthAccessToken = getAccessToken(provider, credentialId); + if (oAuthAccessToken.isPresent()) { + responder.sendString(GSON.toJson( + new GetAccessTokenResponse(oAuthAccessToken.get().getAccessToken(), ""))); + return; + } + + // 2. Fetch refresh token from store + OAuthRefreshToken refreshToken = getRefreshToken(provider, credentialId); + + // 3. Request short-lived access token from 3rd-party API + HttpResponse response; + try { + response = HttpRequests.execute(createGetAccessTokenRequest(oauthProvider, refreshToken.getRefreshToken())); + } catch (IOException e) { + throw new OAuthServiceException(HttpURLConnection.HTTP_INTERNAL_ERROR, "Failed to fetch refresh token", e); + } + + if (response.getResponseCode() != 200) { + throw new OAuthServiceException( + response.getResponseCode(), + "Request for refresh token did not return 200. Response code: " + + response.getResponseCode() + + " , response message: " + + response.getResponseMessage() + + " , response body: " + + response.getResponseBodyAsString()); + } + + RefreshTokenResponse refreshTokenResponse; + try { + refreshTokenResponse = GSON.fromJson(response.getResponseBodyAsString(), RefreshTokenResponse.class); + } catch (JsonSyntaxException e) { + throw new OAuthServiceException(HttpURLConnection.HTTP_INTERNAL_ERROR, "Error parsing JSON response", e); + } + + if (refreshTokenResponse.getAccessToken() == null || refreshTokenResponse.getAccessToken().isEmpty()) { + throw new OAuthServiceException( + HttpURLConnection.HTTP_BAD_REQUEST, + "Access token response body does not have access token: " + response.getResponseBodyAsString()); + } + + // Standard flow: Return access token to caller without writing back to store + responder.sendString(GSON.toJson( + new GetAccessTokenResponse(refreshTokenResponse.getAccessToken(), refreshTokenResponse.getInstanceURL()))); + } + + private boolean isAccessTokenValid(OAuthAccessToken token) { + // Rule 1: If expiresAt is present (> 0), use it with 10-minute safety buffer + if (token.getExpiresAt() > 0) { + return !token.isExpired(600_000L); + } + + // Rule 2: Else if id / identityUrl is present, validate against identity URL + if (token.getIdentityUrl() != null && !token.getIdentityUrl().isEmpty()) { + return isAccessTokenValidViaIdentityUrl(token.getAccessToken(), token.getIdentityUrl()); + } + + // Rule 3: Otherwise -> refresh + return false; + } + + private boolean isAccessTokenValidViaIdentityUrl(String accessToken, String identityUrl) { + try { + HttpRequest request = HttpRequest.get(new URL(identityUrl)) + .addHeader("Authorization", "Bearer " + accessToken) + .build(); + HttpResponse response = HttpRequests.execute(request); + return response.getResponseCode() == 200; + } catch (Exception e) { + LOG.warn("Failed to validate access token via identity URL {}: {}", identityUrl, e.getMessage()); + return false; + } + } + + private void getOAuthCredentialWithRefreshTokenRotation(HttpServiceResponder responder, + OAuthProvider oauthProvider, + String provider, + String credentialId) throws OAuthServiceException { + // 1. Check if a valid cached access token is already available + Optional oAuthAccessToken = getAccessToken(provider, credentialId); + if (oAuthAccessToken.isPresent() && isAccessTokenValid(oAuthAccessToken.get())) { + LOG.debug("Returning valid cached access token for provider {} credential {}", provider, credentialId); + responder.sendString(GSON.toJson( + new GetAccessTokenResponse(oAuthAccessToken.get().getAccessToken(), ""))); + return; + } + + // 2. Try to acquire lease lock + SecureStoreLease lease = acquireLeaseLock(provider, credentialId, 100_000L); + + // 3. If lock is held by another process -> wait for published token + if (lease == null || !lease.isAcquired()) { + LOG.info("Lease lock held by another process for provider {} credential {}. Waiting for new access token...", + provider, credentialId); + Optional waitedResponse = waitForNewAccessToken(provider, credentialId, 100_000L, 500L); + if (waitedResponse.isPresent()) { + responder.sendString(GSON.toJson(waitedResponse.get())); return; } - // If no long-lived access token was found, request a short-lived access token from the 3rd-party API using the - // stored refresh token - OAuthRefreshToken refreshToken = getRefreshToken(provider, credentialId); + // Timeout occurred while waiting for winner -> Attempt to acquire lease lock again! + lease = acquireLeaseLock(provider, credentialId, 100_000L); + if (lease == null || !lease.isAcquired()) { + throw new OAuthServiceException(HttpURLConnection.HTTP_CLIENT_TIMEOUT, + "Timed out waiting for OAuth access token refresh for " + credentialId); + } + } + + // 4. Winner (either initial or fallback after timeout) executes token refresh and persistence + GetAccessTokenResponse response = refreshAndStoreTokens(oauthProvider, lease, provider, credentialId); + responder.sendString(GSON.toJson(response)); + } + private SecureStoreLease acquireLeaseLock(String provider, String credentialId, long timeoutMs) throws OAuthServiceException { + try { + return oauthStore.acquireLease(provider, credentialId, timeoutMs); + } catch (IOException e) { + throw new OAuthServiceException(HttpURLConnection.HTTP_NOT_FOUND, + "Refresh token credential not found for " + credentialId, e); + } + } + + private void writeAccessTokenFromResponse(String provider, String credentialId, RefreshTokenResponse refreshTokenResponse) + throws OAuthStoreException { + if (refreshTokenResponse.getAccessToken() == null || refreshTokenResponse.getAccessToken().isEmpty()) { + return; + } + + long expiresInSeconds = refreshTokenResponse.getExpiresIn(); + long expiresAt = expiresInSeconds > 0 + ? System.currentTimeMillis() + (expiresInSeconds * 1000L) + : 0L; + String identityUrl = refreshTokenResponse.getId(); + + OAuthAccessToken accessToken = OAuthAccessToken.newBuilder() + .withAccessToken(refreshTokenResponse.getAccessToken()) + .withExpiresAt(expiresAt) + .withIdentityUrl(identityUrl) + .build(); + + oauthStore.writeAccessToken(provider, credentialId, accessToken); + } + + private GetAccessTokenResponse refreshAndStoreTokens(OAuthProvider oauthProvider, + SecureStoreLease lease, + String provider, + String credentialId) throws OAuthServiceException { + try { + OAuthRefreshToken refreshToken = getRefreshToken(provider, credentialId); HttpResponse response; try { response = HttpRequests.execute(createGetAccessTokenRequest(oauthProvider, refreshToken.getRefreshToken())); @@ -317,12 +478,7 @@ public void getOAuthCredential(HttpServiceRequest request, HttpServiceResponder if (response.getResponseCode() != 200) { throw new OAuthServiceException( response.getResponseCode(), - "Request for refresh token did not return 200. Response code: " - + response.getResponseCode() - + " , response message: " - + response.getResponseMessage() - + " , response body: " - + response.getResponseBodyAsString()); + "Request for refresh token did not return 200: " + response.getResponseBodyAsString()); } RefreshTokenResponse refreshTokenResponse; @@ -332,39 +488,67 @@ public void getOAuthCredential(HttpServiceRequest request, HttpServiceResponder throw new OAuthServiceException(HttpURLConnection.HTTP_INTERNAL_ERROR, "Error parsing JSON response", e); } - boolean hasRefreshToken = refreshTokenResponse.getRefreshToken() != null - && !refreshTokenResponse.getRefreshToken().isEmpty(); - boolean hasAccessToken = refreshTokenResponse.getAccessToken() != null - && !refreshTokenResponse.getAccessToken().isEmpty(); + if (refreshTokenResponse.getAccessToken() == null || refreshTokenResponse.getAccessToken().isEmpty()) { + throw new OAuthServiceException(HttpURLConnection.HTTP_BAD_REQUEST, "Access token missing in response"); + } - if (!hasAccessToken) { - throw new OAuthServiceException( - HttpURLConnection.HTTP_BAD_REQUEST, - String.format( - "Access token response body does not have access token. The actual response received : %s", - response.getResponseBodyAsString())); + // Mandatory Writes for RTR: + // Write Refresh Token v2 FIRST + if (refreshTokenResponse.getRefreshToken() != null && !refreshTokenResponse.getRefreshToken().isEmpty()) { + OAuthRefreshToken rotatedRefreshToken = OAuthRefreshToken.newBuilder() + .withRefreshToken(refreshTokenResponse.getRefreshToken()) + .withRedirectURI(refreshToken.getRedirectURI()) + .build(); + try { + oauthStore.writeRefreshToken(provider, credentialId, rotatedRefreshToken); + } catch (OAuthStoreException e) { + throw new OAuthServiceException(HttpURLConnection.HTTP_INTERNAL_ERROR, "Failed to write rotated refresh token", e); + } } - // API has given us a new refresh token - if (hasRefreshToken && !refreshToken.getRefreshToken().equals(refreshTokenResponse.getRefreshToken())) { - OAuthRefreshToken newRefreshToken = OAuthRefreshToken.newBuilder() - .withRefreshToken(refreshTokenResponse.getRefreshToken()) - .withRedirectURI(refreshToken.getRedirectURI()) - .build(); + // Write Access Token v2 SECOND + try { + writeAccessTokenFromResponse(provider, credentialId, refreshTokenResponse); + } catch (OAuthStoreException e) { + throw new OAuthServiceException(HttpURLConnection.HTTP_INTERNAL_ERROR, "Failed to write access token", e); + } + return new GetAccessTokenResponse(refreshTokenResponse.getAccessToken(), refreshTokenResponse.getInstanceURL()); + } finally { + if (lease != null && lease.isAcquired()) { try { - oauthStore.writeRefreshToken(provider, credentialId, newRefreshToken); - } catch (OAuthStoreException e) { - throw new OAuthServiceException( - HttpURLConnection.HTTP_INTERNAL_ERROR, "An error occurred while writing the new refresh token"); + oauthStore.releaseLease(provider, credentialId, lease); + } catch (Exception e) { + LOG.warn("Failed to release lease lock for provider {} credential {}: {}", + provider, credentialId, e.getMessage()); } } + } + } - responder.sendString(GSON.toJson( - new GetAccessTokenResponse(refreshTokenResponse.getAccessToken(), refreshTokenResponse.getInstanceURL()))); - } catch (OAuthServiceException e) { - e.respond(responder); + private Optional waitForNewAccessToken( + String provider, String credentialId, long maxWaitMs, long pollIntervalMs) { + long startTime = System.currentTimeMillis(); + + while (System.currentTimeMillis() - startTime < maxWaitMs) { + try { + Thread.sleep(pollIntervalMs); + + // Poll OAuthStore to check if lock winner published a valid access token + Optional accessToken = getAccessToken(provider, credentialId); + if (accessToken.isPresent() && isAccessTokenValid(accessToken.get())) { + return Optional.of(new GetAccessTokenResponse(accessToken.get().getAccessToken(), "")); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return Optional.empty(); + } catch (Exception e) { + LOG.debug("Waiting for new access token for provider {} credential {}: {}", + provider, credentialId, e.getMessage()); + } } + + return Optional.empty(); } @GET diff --git a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/service/StudioService.java b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/service/StudioService.java index 9bab52706283..3162bcfa1fac 100644 --- a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/service/StudioService.java +++ b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/main/java/io/cdap/cdap/datapipeline/service/StudioService.java @@ -61,6 +61,7 @@ protected void configure() { addHandler(new OAuthHandler()); createTable(DraftStore.TABLE_SPEC); createTable(OAuthStore.TABLE_SPEC); + createTable(OAuthStore.LEASE_TABLE_SPEC); createTable(ConnectionStore.CONNECTION_TABLE_SPEC); setProperties(Collections.singletonMap(CONNECTION_TYPE_CONFIG, GSON.toJson(connectionConfig))); } diff --git a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/test/java/io/cdap/cdap/datapipeline/OAuthStoreTest.java b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/test/java/io/cdap/cdap/datapipeline/OAuthStoreTest.java index b560e18478a2..ba8bc348129b 100644 --- a/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/test/java/io/cdap/cdap/datapipeline/OAuthStoreTest.java +++ b/cdap-app-templates/cdap-etl/cdap-data-pipeline-base/src/test/java/io/cdap/cdap/datapipeline/OAuthStoreTest.java @@ -12,6 +12,8 @@ * */ +package io.cdap.cdap.datapipeline; + import java.nio.charset.StandardCharsets; import java.util.Optional; @@ -20,6 +22,7 @@ import io.cdap.cdap.api.security.store.SecureStore; import io.cdap.cdap.api.security.store.SecureStoreManager; +import io.cdap.cdap.api.security.store.lease.SecureStoreLease; import io.cdap.cdap.datapipeline.oauth.OAuthAccessToken; import io.cdap.cdap.datapipeline.oauth.OAuthProvider; import io.cdap.cdap.datapipeline.oauth.OAuthRefreshToken; @@ -34,6 +37,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.any; +import static org.mockito.Mockito.anyLong; import static org.mockito.Mockito.anyString; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doNothing; @@ -95,6 +99,39 @@ public void testGetProviderWithNullCredentialStrategy() throws Exception { assertTrue(provider.isPresent()); assertEquals(provider.get().getCredentialEncodingStrategy(), OAuthProvider.CredentialEncodingStrategy.FORM_BODY); + assertEquals(provider.get().getAuthType(), OAuthProvider.AuthType.STANDARD); + assertEquals(provider.get().getRefreshType(), OAuthProvider.RefreshType.STANDARD); + } + + @Test + public void testGetProviderWithRefreshTokenRotationOAuthType() throws Exception { + String clientCredsJson = "{\"clientId\":\"test-client\",\"clientSecret\":\"test-secret\"}"; + when(mockSecureStore.getData(any(), any())).thenReturn( + clientCredsJson.getBytes(StandardCharsets.UTF_8)); + + doAnswer(invocation -> { + TxRunnable runnable = invocation.getArgument(0); + StructuredTableContext mockContext = mock(StructuredTableContext.class); + when(mockContext.getTable(any())).thenReturn(mockTable); + runnable.run(mockContext); + return null; + }).when(mockTransactionRunner).run(any(TxRunnable.class)); + + when(mockRow.getString("oauthprovider")).thenReturn(PROVIDER_NAME); + when(mockRow.getString("loginurl")).thenReturn(LOGIN_URL); + when(mockRow.getString("tokenrefreshurl")).thenReturn(TOKEN_REFRESH_URL); + when(mockRow.getString("credentialencodingstrategy")).thenReturn("FORM_BODY"); + when(mockRow.getString("useragent")).thenReturn(USER_AGENT); + when(mockRow.getString("refreshtype")).thenReturn("RTR"); + when(mockRow.getString("authtype")).thenReturn("PKCE"); + + when(mockTable.read(any())).thenReturn(Optional.of(mockRow)); + + Optional provider = oauthStore.getProvider(PROVIDER_NAME); + + assertTrue(provider.isPresent()); + assertEquals(provider.get().getAuthType(), OAuthProvider.AuthType.PKCE); + assertEquals(provider.get().getRefreshType(), OAuthProvider.RefreshType.RTR); } @Test @@ -194,4 +231,27 @@ public NotFoundException(String message) { verify(mockSecureStoreManager, times(1)).delete(any(), any()); verify(mockTable, times(0)).delete(any()); } + + @Test + public void testAcquireAndReleaseDatabaseLease() throws Exception { + doAnswer(invocation -> { + TxRunnable runnable = invocation.getArgument(0); + StructuredTableContext mockContext = mock(StructuredTableContext.class); + when(mockContext.getTable(any())).thenReturn(mockTable); + runnable.run(mockContext); + return null; + }).when(mockTransactionRunner).run(any(TxRunnable.class)); + + when(mockTable.read(any())).thenReturn(Optional.empty()); + doNothing().when(mockTable).upsert(any()); + + when(mockSecureStoreManager.acquireLease(anyString(), anyString(), anyLong())) + .thenReturn(io.cdap.cdap.api.security.store.lease.SecureStoreLease.acquired("12345", "worker-1")); + + io.cdap.cdap.api.security.store.lease.SecureStoreLease lease = oauthStore.acquireLease(PROVIDER_NAME, "cred-1", 30000L); + assertTrue(lease.isAcquired()); + + oauthStore.releaseLease(PROVIDER_NAME, "cred-1", lease); + verify(mockSecureStoreManager, times(1)).releaseLease(anyString(), anyString(), eq(lease)); + } } diff --git a/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/CloudSecretManagerClient.java b/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/CloudSecretManagerClient.java index 88d91345e3aa..e832c40460cf 100644 --- a/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/CloudSecretManagerClient.java +++ b/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/CloudSecretManagerClient.java @@ -19,6 +19,7 @@ import com.google.api.gax.core.CredentialsProvider; import com.google.api.gax.core.FixedCredentialsProvider; import com.google.api.gax.rpc.ApiException; +import com.google.api.gax.rpc.StatusCode; import com.google.auth.oauth2.GoogleCredentials; import com.google.cloud.ServiceOptions; import com.google.cloud.secretmanager.v1.AddSecretVersionRequest; @@ -171,6 +172,43 @@ public void updateSecret(WrappedSecret wrappedSecret) { FieldMask.newBuilder().addPaths("annotations").build()); } + /** + * Conditionally updates annotations on the specified secret using an ETag for optimistic concurrency control. + * + * @param namespace CDAP secret namespace + * @param name CDAP secret name + * @param annotationsToUpdate Map of annotations to update + * @param etag The expected ETag of the secret + * @return {@code true} if update succeeded, {@code false} if ETag mismatch occurred (FAILED_PRECONDITION) + * @throws ApiException if another Google API failure occurs. + */ + public boolean updateSecretWithEtag(String namespace, + String name, + Map annotationsToUpdate, + String etag) { + String resourceName = getSecretResourceName(namespace, name); + Secret.Builder secretBuilder = Secret.newBuilder() + .setName(resourceName) + .setEtag(etag); + + for (Map.Entry entry : annotationsToUpdate.entrySet()) { + secretBuilder.putAnnotations(entry.getKey(), entry.getValue()); + } + + try { + secretManager.updateSecret( + secretBuilder.build(), + FieldMask.newBuilder().addPaths("annotations").build()); + return true; + } catch (ApiException e) { + if (e.getStatusCode().getCode() == StatusCode.Code.FAILED_PRECONDITION) { + LOG.debug("Optimistic lock failure (ETag mismatch) for secret {} in namespace {}", name, namespace); + return false; + } + throw e; + } + } + /** * Deletes the specified secret. * diff --git a/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/GcpSecretManager.java b/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/GcpSecretManager.java index ccb7ff2487ee..b639e541f41e 100644 --- a/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/GcpSecretManager.java +++ b/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/GcpSecretManager.java @@ -27,6 +27,7 @@ import java.io.IOException; import java.util.Arrays; import java.util.Collection; +import java.util.Map; import java.util.stream.Collectors; /** @@ -143,4 +144,18 @@ public void delete(String namespace, String name) throws SecretNotFoundException public void destroy(SecretManagerContext context) { client.destroy(); } + + /** + * Performs an optimistic atomic update on secret annotations using an ETag. + * + * @return {@code true} if lock/update succeeded, {@code false} if ETag mismatch occurred. + */ + public boolean updateSecretWithEtag(String namespace, String name, Map annotations, String etag) + throws IOException { + try { + return client.updateSecretWithEtag(namespace, name, annotations, etag); + } catch (ApiException e) { + throw new IOException("Secret Manager updateSecretWithEtag API call failed", e); + } + } } diff --git a/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/GcpSecretManagerLeaseStrategy.java b/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/GcpSecretManagerLeaseStrategy.java new file mode 100644 index 000000000000..83aa69c83ac6 --- /dev/null +++ b/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/GcpSecretManagerLeaseStrategy.java @@ -0,0 +1,192 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * 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.cdap.cdap.securestore.gcp.cloudsecretmanager; + +import com.google.api.gax.rpc.ApiException; +import com.google.api.gax.rpc.StatusCode; +import com.google.common.annotations.VisibleForTesting; +import io.cdap.cdap.securestore.spi.lease.SecretLease; +import io.cdap.cdap.securestore.spi.lease.SecretLeaseStrategy; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; + +/** + * GCP Secret Manager implementation of {@link SecretLeaseStrategy}. + * Performs Lease-Before-Refresh directly on the Refresh Token Secret annotations using GCP ETag + * optimistic concurrency control across multiple CDF instances. + */ +public class GcpSecretManagerLeaseStrategy implements SecretLeaseStrategy { + private static final Logger LOG = LoggerFactory.getLogger(GcpSecretManagerLeaseStrategy.class); + public static final String PROVIDER_NAME = "gcp-secretmanager"; + + private static final String ANNOTATION_STATE = "state"; + private static final String ANNOTATION_LOCK_TIMESTAMP = "lock_timestamp"; + private static final String ANNOTATION_LOCK_HOLDER = "lock_holder"; + + private static final String STATE_IDLE = "idle"; + private static final String STATE_REFRESHING = "refreshing"; + + private CloudSecretManagerClient client; + private String workerId; + + public GcpSecretManagerLeaseStrategy() { + this(Collections.emptyMap()); + } + + public GcpSecretManagerLeaseStrategy(Map properties) { + this.workerId = generateWorkerId(properties); + } + + private static String generateWorkerId(Map properties) { + String instanceName = properties != null ? properties.get("instance.name") : null; + if (instanceName == null || instanceName.isEmpty()) { + instanceName = System.getenv("HOSTNAME"); + } + if (instanceName == null || instanceName.isEmpty()) { + try { + instanceName = java.net.InetAddress.getLocalHost().getHostName(); + } catch (Exception e) { + instanceName = "cdf-instance"; + } + } + return String.format("%s:%s", instanceName, UUID.randomUUID()); + } + + @VisibleForTesting + public GcpSecretManagerLeaseStrategy(CloudSecretManagerClient client) { + this.client = client; + this.workerId = generateWorkerId(Collections.emptyMap()); + } + + @Override + public synchronized void initialize(Map properties) throws IOException { + if (properties != null && !properties.isEmpty()) { + this.workerId = generateWorkerId(properties); + } + if (this.client == null) { + this.client = new CloudSecretManagerClient(properties); + } + LOG.info("Initialized GcpSecretManagerLeaseStrategy for workerId '{}' with {} properties", + workerId, properties != null ? properties.size() : 0); + } + + @Override + public String getName() { + return PROVIDER_NAME; + } + + @Override + public SecretLease acquireLease(String namespace, String key, long timeoutMs) throws IOException { + ensureClientInitialized(); + long now = System.currentTimeMillis(); + + try { + // Read current Refresh Token Secret and its metadata (ETag & annotations) + WrappedSecret refreshSecret = client.getSecret(namespace, key); + String currentEtag = refreshSecret.getEtag() == null ? "" : refreshSecret.getEtag(); + + String state = refreshSecret.getAnnotation(ANNOTATION_STATE, STATE_IDLE); + String lockTimestampStr = refreshSecret.getAnnotation(ANNOTATION_LOCK_TIMESTAMP, "0"); + long lockTimestamp = 0L; + try { + lockTimestamp = Long.parseLong(lockTimestampStr); + } catch (NumberFormatException e) { + lockTimestamp = 0L; + } + + boolean isExpired = (now - lockTimestamp) > timeoutMs; + boolean isLockedByAnother = STATE_REFRESHING.equalsIgnoreCase(state) && !isExpired; + + if (isLockedByAnother) { + LOG.info("Lease lock currently held for Refresh Token Secret '{}' in namespace '{}' (holder: {}, timestamp: {})", + key, namespace, refreshSecret.getAnnotation(ANNOTATION_LOCK_HOLDER, "unknown"), lockTimestampStr); + return SecretLease.failed(); + } + + // Perform conditional update on Refresh Token Secret annotations using currentEtag + Map annotationsToUpdate = new HashMap<>(refreshSecret.getAnnotations()); + annotationsToUpdate.put(ANNOTATION_STATE, STATE_REFRESHING); + annotationsToUpdate.put(ANNOTATION_LOCK_TIMESTAMP, String.valueOf(now)); + annotationsToUpdate.put(ANNOTATION_LOCK_HOLDER, workerId); + + boolean acquired = client.updateSecretWithEtag(namespace, key, annotationsToUpdate, currentEtag); + + if (acquired) { + LOG.info("Acquired lease lock on Refresh Token Secret '{}' in namespace '{}' for worker '{}' at timestamp {}", + key, namespace, workerId, now); + return SecretLease.acquired(String.valueOf(now), workerId); + } else { + LOG.info("ETag mismatch or lock contention while acquiring lease on Refresh Token Secret '{}' in namespace '{}' (state: {})", + key, namespace, state); + return SecretLease.failed(); + } + + } catch (ApiException e) { + LOG.error("GCP ApiException while acquiring lease on Refresh Token Secret '{}' in namespace '{}': {} (StatusCode: {})", + key, namespace, e.getMessage(), e.getStatusCode().getCode(), e); + if (e.getStatusCode().getCode() == StatusCode.Code.NOT_FOUND) { + throw new IOException("Refresh Token Secret '" + key + "' not found in namespace '" + namespace + "'", e); + } + throw new IOException("Failed to acquire lease lock on Refresh Token Secret " + key, e); + } catch (InvalidSecretException e) { + LOG.error("InvalidSecretException while parsing Refresh Token Secret '{}' in namespace '{}': {}", + key, namespace, e.getMessage(), e); + throw new IOException("Failed to parse Refresh Token Secret metadata for " + key, e); + } catch (Exception e) { + LOG.error("Unexpected exception while acquiring lease on Refresh Token Secret '{}' in namespace '{}': {}", + key, namespace, e.getMessage(), e); + throw new IOException("Unexpected error acquiring lease on Refresh Token Secret " + key, e); + } + } + + @Override + public void releaseLease(String namespace, String key, SecretLease lease) throws IOException { + if (lease == null || !lease.isAcquired()) { + return; + } + + ensureClientInitialized(); + + try { + WrappedSecret refreshSecret = client.getSecret(namespace, key); + String currentEtag = refreshSecret.getEtag() == null ? "" : refreshSecret.getEtag(); + + Map annotationsToUpdate = new HashMap<>(refreshSecret.getAnnotations()); + annotationsToUpdate.put(ANNOTATION_STATE, STATE_IDLE); + annotationsToUpdate.put(ANNOTATION_LOCK_TIMESTAMP, "0"); + annotationsToUpdate.put(ANNOTATION_LOCK_HOLDER, ""); + + client.updateSecretWithEtag(namespace, key, annotationsToUpdate, currentEtag); + LOG.info("Released lease lock on Refresh Token Secret {} in namespace {}", key, namespace); + } catch (Exception e) { + LOG.warn("Failed to release lease lock on Refresh Token Secret {} in namespace {}: {}", + key, namespace, e.getMessage()); + } + } + + private synchronized void ensureClientInitialized() throws IOException { + if (this.client == null) { + this.client = new CloudSecretManagerClient(Collections.emptyMap()); + } + } +} diff --git a/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/WrappedSecret.java b/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/WrappedSecret.java index d4cee5040065..e29a4e641b33 100644 --- a/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/WrappedSecret.java +++ b/cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/WrappedSecret.java @@ -24,8 +24,11 @@ import com.google.gson.JsonSyntaxException; import com.google.protobuf.util.Timestamps; import io.cdap.cdap.securestore.spi.secret.SecretMetadata; +import javax.annotation.Nullable; import java.lang.reflect.Type; +import java.util.Collections; +import java.util.HashMap; import java.util.Map; import java.util.Optional; @@ -42,10 +45,22 @@ public final class WrappedSecret { private final String namespace; private final SecretMetadata secretMetadata; + @Nullable + private final String etag; + private final Map annotations; private WrappedSecret(String namespace, SecretMetadata secretMetadata) { + this(namespace, secretMetadata, null, Collections.emptyMap()); + } + + private WrappedSecret(String namespace, + SecretMetadata secretMetadata, + @Nullable String etag, + Map annotations) { this.namespace = namespace; this.secretMetadata = secretMetadata; + this.etag = etag; + this.annotations = annotations == null ? Collections.emptyMap() : annotations; } /** Constructs a new WrappedSecret from a CDAP Secret. */ @@ -60,7 +75,7 @@ public static WrappedSecret fromMetadata(String namespace, SecretMetadata metada public static WrappedSecret fromGcpSecret(Secret secret) throws InvalidSecretException { String namespace = getNamespace(secret); SecretMetadata metadata = toSecretMetadata(secret); - return new WrappedSecret(namespace, metadata); + return new WrappedSecret(namespace, metadata, secret.getEtag(), secret.getAnnotationsMap()); } public String getNamespace() { @@ -71,13 +86,34 @@ public SecretMetadata getCdapSecretMetadata() { return secretMetadata; } + @Nullable + public String getEtag() { + return etag; + } + + public Map getAnnotations() { + return annotations; + } + + public String getAnnotation(String key, String defaultValue) { + return annotations.getOrDefault(key, defaultValue); + } + /** * Returns a new GCP {@link Secret} representing the underlying SecretMetadata. * * @param resourceName Value to set for the "name" field needed for update operations. */ public Secret getGcpSecret(String resourceName) { - return Secret.newBuilder() + return getGcpSecret(resourceName, null); + } + + /** + * Returns a new GCP {@link Secret} representing the underlying SecretMetadata including etag + * and additional annotations. + */ + public Secret getGcpSecret(String resourceName, @Nullable Map additionalAnnotations) { + Secret.Builder builder = Secret.newBuilder() // Set replication policy to automatic (as opposed to user-managed) and do not specify a // CMEK (use google-managed key). .setReplication(Replication.newBuilder().setAutomatic(Automatic.getDefaultInstance())) @@ -85,16 +121,32 @@ public Secret getGcpSecret(String resourceName) { .putAnnotations("cdap_namespace", namespace) .putAnnotations("cdap_secret_name", secretMetadata.getName()) .putAnnotations("cdap_description", Optional.ofNullable(secretMetadata.getDescription()).orElse("")) - .putAnnotations("cdap_props", serializeProps(secretMetadata.getProperties())) - .build(); + .putAnnotations("cdap_props", serializeProps(secretMetadata.getProperties())); + + if (etag != null && !etag.isEmpty()) { + builder.setEtag(etag); + } + if (additionalAnnotations != null) { + for (Map.Entry entry : additionalAnnotations.entrySet()) { + builder.putAnnotations(entry.getKey(), entry.getValue()); + } + } + return builder.build(); } private static SecretMetadata toSecretMetadata(Secret secret) throws InvalidSecretException { + Map props = new HashMap<>(deserializeProps(secret.getAnnotationsOrDefault("cdap_props", "{}"))); + if (!secret.getEtag().isEmpty()) { + props.put("etag", secret.getEtag()); + } + for (Map.Entry entry : secret.getAnnotationsMap().entrySet()) { + props.putIfAbsent(entry.getKey(), entry.getValue()); + } return new SecretMetadata( secret.getAnnotationsOrDefault("cdap_secret_name", ""), secret.getAnnotationsOrDefault("cdap_description", ""), Timestamps.toMillis(secret.getCreateTime()), - deserializeProps(secret.getAnnotationsOrDefault("cdap_props", "{}"))); + props); } private static String getNamespace(Secret secret) { diff --git a/cdap-securestore-ext-gcp-secretstore/src/main/resources/META-INF/services/io.cdap.cdap.securestore.spi.lease.SecretLeaseStrategy b/cdap-securestore-ext-gcp-secretstore/src/main/resources/META-INF/services/io.cdap.cdap.securestore.spi.lease.SecretLeaseStrategy new file mode 100644 index 000000000000..81bc4875b90c --- /dev/null +++ b/cdap-securestore-ext-gcp-secretstore/src/main/resources/META-INF/services/io.cdap.cdap.securestore.spi.lease.SecretLeaseStrategy @@ -0,0 +1,17 @@ +# +# Copyright © 2026 Cask Data, Inc. +# +# 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. +# + +io.cdap.cdap.securestore.gcp.cloudsecretmanager.GcpSecretManagerLeaseStrategy diff --git a/cdap-securestore-ext-gcp-secretstore/src/test/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/GcpSecretManagerTest.java b/cdap-securestore-ext-gcp-secretstore/src/test/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/GcpSecretManagerTest.java index 4e5445dfa0bc..c4e189387c5b 100644 --- a/cdap-securestore-ext-gcp-secretstore/src/test/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/GcpSecretManagerTest.java +++ b/cdap-securestore-ext-gcp-secretstore/src/test/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/GcpSecretManagerTest.java @@ -36,7 +36,9 @@ import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -174,6 +176,58 @@ public void delete_wrapsApiExceptions() throws Exception { assertThrows(IOException.class, () -> secretManager.delete(NAMESPACE, "example")); } + @Test + public void updateSecretWithEtag_success() throws Exception { + when(client.updateSecretWithEtag(eq(NAMESPACE), eq("example"), ArgumentMatchers.any(), eq("etag-123"))) + .thenReturn(true); + + boolean success = secretManager.updateSecretWithEtag( + NAMESPACE, "example", ImmutableMap.of("state", "refreshing"), "etag-123"); + assertTrue(success); + } + + @Test + public void updateSecretWithEtag_etagMismatchFailure() throws Exception { + when(client.updateSecretWithEtag(eq(NAMESPACE), eq("example"), ArgumentMatchers.any(), eq("etag-123"))) + .thenReturn(false); + + boolean success = secretManager.updateSecretWithEtag( + NAMESPACE, "example", ImmutableMap.of("state", "refreshing"), "etag-123"); + assertFalse(success); + } + + @Test + public void testAcquireGcpLeaseSuccess() throws Exception { + GcpSecretManagerLeaseStrategy leaseStrategy = new GcpSecretManagerLeaseStrategy(client); + SecretMetadata metadata = createMetadata("salesforce"); + WrappedSecret wrappedSecret = WrappedSecret.fromMetadata(NAMESPACE, metadata); + when(client.getSecret(eq(NAMESPACE), eq("salesforce"))).thenReturn(wrappedSecret); + when(client.updateSecretWithEtag( + eq(NAMESPACE), eq("salesforce"), ArgumentMatchers.any(), ArgumentMatchers.any())) + .thenReturn(true); + + io.cdap.cdap.securestore.spi.lease.SecretLease lease = + leaseStrategy.acquireLease(NAMESPACE, "salesforce", 30000L); + assertTrue(lease.isAcquired()); + + leaseStrategy.releaseLease(NAMESPACE, "salesforce", lease); + } + + @Test + public void testAcquireGcpLeaseEtagMismatchFailure() throws Exception { + GcpSecretManagerLeaseStrategy leaseStrategy = new GcpSecretManagerLeaseStrategy(client); + SecretMetadata metadata = createMetadata("salesforce"); + WrappedSecret wrappedSecret = WrappedSecret.fromMetadata(NAMESPACE, metadata); + when(client.getSecret(eq(NAMESPACE), eq("salesforce"))).thenReturn(wrappedSecret); + when(client.updateSecretWithEtag( + eq(NAMESPACE), eq("salesforce"), ArgumentMatchers.any(), ArgumentMatchers.any())) + .thenReturn(false); + + io.cdap.cdap.securestore.spi.lease.SecretLease lease = + leaseStrategy.acquireLease(NAMESPACE, "salesforce", 30000L); + assertFalse(lease.isAcquired()); + } + private static Secret createSecret(String name) { return new Secret(name.getBytes(), createMetadata(name)); } diff --git a/cdap-securestore-spi/src/main/java/io/cdap/cdap/securestore/spi/lease/SecretLease.java b/cdap-securestore-spi/src/main/java/io/cdap/cdap/securestore/spi/lease/SecretLease.java new file mode 100644 index 000000000000..68038e60c196 --- /dev/null +++ b/cdap-securestore-spi/src/main/java/io/cdap/cdap/securestore/spi/lease/SecretLease.java @@ -0,0 +1,86 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * 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.cdap.cdap.securestore.spi.lease; + +import java.util.Objects; + +/** + * SPI model representing a distributed lease lock on a secret in the underlying storage plugin. + *

+ * Note: This is the plugin-facing SPI counterpart to {@code io.cdap.cdap.api.security.store.lease.SecureStoreLease} + * in the API layer, mirroring the {@link io.cdap.cdap.securestore.spi.secret.SecretMetadata} vs + * {@code SecureStoreMetadata} architectural pattern in CDAP. + */ +public class SecretLease { + private final boolean acquired; + private final String lockTimestamp; + private final String lockHolder; + + private SecretLease(boolean acquired, String lockTimestamp, String lockHolder) { + this.acquired = acquired; + this.lockTimestamp = lockTimestamp; + this.lockHolder = lockHolder; + } + + public static SecretLease acquired(String lockTimestamp, String lockHolder) { + return new SecretLease(true, lockTimestamp, lockHolder); + } + + public static SecretLease failed() { + return new SecretLease(false, null, null); + } + + public boolean isAcquired() { + return acquired; + } + + public String getLockTimestamp() { + return lockTimestamp; + } + + public String getLockHolder() { + return lockHolder; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + SecretLease that = (SecretLease) o; + return acquired == that.acquired + && Objects.equals(lockTimestamp, that.lockTimestamp) + && Objects.equals(lockHolder, that.lockHolder); + } + + @Override + public int hashCode() { + return Objects.hash(acquired, lockTimestamp, lockHolder); + } + + @Override + public String toString() { + return "SecretLease{" + + "acquired=" + acquired + + ", lockTimestamp='" + lockTimestamp + '\'' + + ", lockHolder='" + lockHolder + '\'' + + '}'; + } +} diff --git a/cdap-securestore-spi/src/main/java/io/cdap/cdap/securestore/spi/lease/SecretLeaseStrategy.java b/cdap-securestore-spi/src/main/java/io/cdap/cdap/securestore/spi/lease/SecretLeaseStrategy.java new file mode 100644 index 000000000000..81fbdbd217de --- /dev/null +++ b/cdap-securestore-spi/src/main/java/io/cdap/cdap/securestore/spi/lease/SecretLeaseStrategy.java @@ -0,0 +1,62 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * 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.cdap.cdap.securestore.spi.lease; + +import java.io.IOException; +import java.util.Map; + +/** + * SPI for acquiring and releasing distributed lease locks during secret updates or token rotations. + */ +public interface SecretLeaseStrategy { + + /** + * Initializes the strategy with configuration properties. + * + * @param properties configuration properties + * @throws IOException if initialization fails + */ + default void initialize(Map properties) throws IOException { + // default no-op + } + + /** + * Returns the unique name of this lease strategy implementation. + */ + String getName(); + + /** + * Attempts to acquire a lease lock on a secret resource. + * + * @param namespace namespace or provider domain of the secret + * @param key unique name or credential ID of the secret + * @param timeoutMs lock timeout in milliseconds before lease is considered expired + * @return {@link SecretLease} indicating acquisition success and lock details + * @throws IOException if lock acquisition fails due to underlying storage errors + */ + SecretLease acquireLease(String namespace, String key, long timeoutMs) throws IOException; + + /** + * Releases an acquired lease lock on a secret resource. + * + * @param namespace namespace or provider domain of the secret + * @param key unique name or credential ID of the secret + * @param lease {@link SecretLease} to release + * @throws IOException if lock release fails due to underlying storage errors + */ + void releaseLease(String namespace, String key, SecretLease lease) throws IOException; +} diff --git a/cdap-security/src/main/java/io/cdap/cdap/security/guice/SecureStoreServerModule.java b/cdap-security/src/main/java/io/cdap/cdap/security/guice/SecureStoreServerModule.java index 46e989627722..bcfcc1cab87f 100644 --- a/cdap-security/src/main/java/io/cdap/cdap/security/guice/SecureStoreServerModule.java +++ b/cdap-security/src/main/java/io/cdap/cdap/security/guice/SecureStoreServerModule.java @@ -29,6 +29,7 @@ import io.cdap.cdap.common.conf.CConfiguration; import io.cdap.cdap.common.conf.Constants; import io.cdap.cdap.common.conf.SConfiguration; +import io.cdap.cdap.securestore.spi.lease.SecretLeaseStrategy; import io.cdap.cdap.security.store.DefaultSecureStoreService; import io.cdap.cdap.security.store.DummySecureStoreService; import io.cdap.cdap.security.store.FileSecureStoreService; @@ -36,7 +37,15 @@ import io.cdap.cdap.security.store.SecureStoreUtils; import io.cdap.cdap.security.store.file.FileSecureStoreCodec; import io.cdap.cdap.security.store.file.SecureStoreDataCodecV2; +import io.cdap.cdap.security.store.lease.StructuredTableLeaseStrategy; +import io.cdap.cdap.security.store.secretmanager.SecretLeaseStrategyExtensionLoader; import io.cdap.cdap.security.store.secretmanager.SecretManagerSecureStoreService; +import io.cdap.cdap.spi.data.transaction.TransactionRunner; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.util.Collections; +import java.util.Map; +import javax.annotation.Nullable; /** * Server side guice bindings for secure store service related classes. @@ -57,9 +66,11 @@ protected void configure() { bind(SecureStore.class).to(DefaultSecureStoreService.class); bind(SecureStoreManager.class).to(DefaultSecureStoreService.class); bind(SecureStoreService.class).to(DefaultSecureStoreService.class); + bind(SecretLeaseStrategy.class).toProvider(SecretLeaseStrategyProvider.class).in(Scopes.SINGLETON); expose(SecureStore.class); expose(SecureStoreManager.class); expose(SecureStoreService.class); + expose(SecretLeaseStrategy.class); } /** @@ -113,4 +124,44 @@ public SecureStoreService get() { return injector.getInstance(SecretManagerSecureStoreService.class); } } + + @Singleton + private static final class SecretLeaseStrategyProvider implements Provider { + private static final Logger LOG = LoggerFactory.getLogger(SecretLeaseStrategyProvider.class); + private final CConfiguration cConf; + private final Injector injector; + + @Inject + private SecretLeaseStrategyProvider(CConfiguration cConf, Injector injector) { + this.cConf = cConf; + this.injector = injector; + } + + @Override + public SecretLeaseStrategy get() { + String provider = cConf.get(Constants.Security.Store.PROVIDER); + String extDir = cConf.get(Constants.Security.Store.EXTENSIONS_DIR); + if (extDir != null && provider != null) { + try { + SecretLeaseStrategy strategy = new SecretLeaseStrategyExtensionLoader(extDir).get(provider); + if (strategy != null) { + String prefix = String.format("%s%s.", Constants.Security.Store.PROPERTY_PREFIX, provider); + Map properties = Collections.unmodifiableMap(cConf.getPropsWithPrefix(prefix)); + strategy.initialize(properties); + LOG.info("Discovered and initialized extension SecretLeaseStrategy '{}' for provider '{}'", strategy.getName(), provider); + return strategy; + } + } catch (Exception e) { + LOG.warn("Failed to load extension SecretLeaseStrategy for provider '{}'", provider, e); + } + } + try { + TransactionRunner transactionRunner = injector.getInstance(TransactionRunner.class); + return new StructuredTableLeaseStrategy(transactionRunner); + } catch (Exception e) { + LOG.warn("TransactionRunner not bound in Guice, returning null SecretLeaseStrategy", e); + return null; + } + } + } } diff --git a/cdap-security/src/main/java/io/cdap/cdap/security/store/DefaultSecureStoreService.java b/cdap-security/src/main/java/io/cdap/cdap/security/store/DefaultSecureStoreService.java index 268b3cb93b59..1b34ed391c12 100644 --- a/cdap-security/src/main/java/io/cdap/cdap/security/store/DefaultSecureStoreService.java +++ b/cdap-security/src/main/java/io/cdap/cdap/security/store/DefaultSecureStoreService.java @@ -32,6 +32,7 @@ import io.cdap.cdap.security.spi.authentication.AuthenticationContext; import io.cdap.cdap.security.spi.authorization.AccessEnforcer; import io.cdap.cdap.security.spi.authorization.UnauthorizedException; +import io.cdap.cdap.api.security.store.lease.SecureStoreLease; import java.io.IOException; import java.util.List; import java.util.Map; @@ -152,4 +153,14 @@ protected void startUp() throws Exception { protected void shutDown() throws Exception { secureStoreService.stopAndWait(); } + + @Override + public SecureStoreLease acquireLease(String namespace, String name, long timeoutMs) throws IOException { + return secureStoreService.acquireLease(namespace, name, timeoutMs); + } + + @Override + public void releaseLease(String namespace, String name, SecureStoreLease lease) throws IOException { + secureStoreService.releaseLease(namespace, name, lease); + } } diff --git a/cdap-security/src/main/java/io/cdap/cdap/security/store/SecureStoreHandler.java b/cdap-security/src/main/java/io/cdap/cdap/security/store/SecureStoreHandler.java index 474fc4bdf84d..83ace9308389 100644 --- a/cdap-security/src/main/java/io/cdap/cdap/security/store/SecureStoreHandler.java +++ b/cdap-security/src/main/java/io/cdap/cdap/security/store/SecureStoreHandler.java @@ -46,9 +46,12 @@ import java.nio.charset.StandardCharsets; import javax.ws.rs.DELETE; import javax.ws.rs.GET; +import javax.ws.rs.POST; import javax.ws.rs.PUT; import javax.ws.rs.Path; import javax.ws.rs.PathParam; +import javax.ws.rs.QueryParam; +import io.cdap.cdap.api.security.store.lease.SecureStoreLease; /** * Exposes REST APIs for {@link SecureStore} and {@link SecureStoreManager}. @@ -127,6 +130,37 @@ public void getMetadata(HttpRequest httpRequest, HttpResponder httpResponder, httpResponder.sendJson(HttpResponseStatus.OK, GSON.toJson(metadata)); } + @Path("/{key-name}/lease") + @POST + public void acquireLease(HttpRequest httpRequest, HttpResponder httpResponder, + @PathParam("namespace-id") String namespace, + @PathParam("key-name") String name, + @QueryParam("timeoutMs") long timeoutMs) throws Exception { + SecureStoreLease lease = secureStoreManager.acquireLease(namespace, name, timeoutMs); + httpResponder.sendJson(HttpResponseStatus.OK, GSON.toJson(lease)); + } + + @Path("/{key-name}/lease") + @DELETE + public void releaseLease(FullHttpRequest httpRequest, HttpResponder httpResponder, + @PathParam("namespace-id") String namespace, + @PathParam("key-name") String name) throws Exception { + SecureStoreLease lease = parseLeaseBody(httpRequest); + secureStoreManager.releaseLease(namespace, name, lease); + httpResponder.sendStatus(HttpResponseStatus.OK); + } + + private SecureStoreLease parseLeaseBody(FullHttpRequest request) throws IOException { + ByteBuf content = request.content(); + if (!content.isReadable()) { + return null; + } + try (Reader reader = new InputStreamReader(new ByteBufInputStream(content), + StandardCharsets.UTF_8)) { + return GSON.fromJson(reader, SecureStoreLease.class); + } + } + @Path("/") @GET public void list(HttpRequest httpRequest, HttpResponder httpResponder, diff --git a/cdap-security/src/main/java/io/cdap/cdap/security/store/client/RemoteSecureStore.java b/cdap-security/src/main/java/io/cdap/cdap/security/store/client/RemoteSecureStore.java index 90124369fe75..ae72528c8c79 100644 --- a/cdap-security/src/main/java/io/cdap/cdap/security/store/client/RemoteSecureStore.java +++ b/cdap-security/src/main/java/io/cdap/cdap/security/store/client/RemoteSecureStore.java @@ -25,6 +25,7 @@ import io.cdap.cdap.api.security.store.SecureStoreData; import io.cdap.cdap.api.security.store.SecureStoreManager; import io.cdap.cdap.api.security.store.SecureStoreMetadata; +import io.cdap.cdap.api.security.store.lease.SecureStoreLease; import io.cdap.cdap.common.SecureKeyAlreadyExistsException; import io.cdap.cdap.common.SecureKeyNotFoundException; import io.cdap.cdap.common.conf.Constants; @@ -121,6 +122,38 @@ public void delete(String namespace, String name) throws Exception { namespace, name)); } + @Override + public SecureStoreLease acquireLease(String namespace, String name, long timeoutMs) throws IOException { + try { + HttpRequest request = remoteClient.requestBuilder(HttpMethod.POST, + createPath(namespace, name) + "/lease?timeoutMs=" + timeoutMs).build(); + HttpResponse response = remoteClient.execute(request, Idempotency.NONE); + handleResponse(response, namespace, name, + String.format("Error occurred while acquiring lease for key %s:%s", namespace, name)); + return GSON.fromJson(response.getResponseBodyAsString(), SecureStoreLease.class); + } catch (IOException e) { + throw e; + } catch (Exception e) { + throw new IOException(e); + } + } + + @Override + public void releaseLease(String namespace, String name, SecureStoreLease lease) throws IOException { + try { + HttpRequest request = remoteClient.requestBuilder(HttpMethod.DELETE, + createPath(namespace, name) + "/lease") + .withBody(GSON.toJson(lease)).build(); + HttpResponse response = remoteClient.execute(request, Idempotency.NONE); + handleResponse(response, namespace, name, + String.format("Error occurred while releasing lease for key %s:%s", namespace, name)); + } catch (IOException e) { + throw e; + } catch (Exception e) { + throw new IOException(e); + } + } + /** * Handles error based on response code. */ diff --git a/cdap-security/src/main/java/io/cdap/cdap/security/store/lease/StructuredTableLeaseStrategy.java b/cdap-security/src/main/java/io/cdap/cdap/security/store/lease/StructuredTableLeaseStrategy.java new file mode 100644 index 000000000000..a90c0a3db8a5 --- /dev/null +++ b/cdap-security/src/main/java/io/cdap/cdap/security/store/lease/StructuredTableLeaseStrategy.java @@ -0,0 +1,166 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * 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.cdap.cdap.security.store.lease; + +import io.cdap.cdap.securestore.spi.lease.SecretLease; +import io.cdap.cdap.securestore.spi.lease.SecretLeaseStrategy; +import io.cdap.cdap.spi.data.InvalidFieldException; +import io.cdap.cdap.spi.data.StructuredRow; +import io.cdap.cdap.spi.data.StructuredTable; +import io.cdap.cdap.spi.data.TableNotFoundException; +import io.cdap.cdap.spi.data.table.StructuredTableId; +import io.cdap.cdap.spi.data.table.field.Field; +import io.cdap.cdap.spi.data.table.field.Fields; +import io.cdap.cdap.spi.data.transaction.TransactionRunner; +import io.cdap.cdap.spi.data.transaction.TransactionRunners; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Default database StructuredTable implementation of {@link SecretLeaseStrategy}. + */ +public class StructuredTableLeaseStrategy implements SecretLeaseStrategy { + private static final Logger LOG = LoggerFactory.getLogger(StructuredTableLeaseStrategy.class); + public static final String PROVIDER_NAME = "structured-table"; + + public static final StructuredTableId LEASE_TABLE_ID = new StructuredTableId("app_oauth_leases"); + public static final String LEASE_PROVIDER_COL = "provider"; + public static final String LEASE_CREDENTIAL_COL = "credential_id"; + public static final String LEASE_STATE_COL = "state"; + public static final String LEASE_LOCK_TIMESTAMP_COL = "lock_timestamp"; + public static final String LEASE_LOCK_HOLDER_COL = "lock_holder"; + + private static final String STATE_IDLE = "idle"; + private static final String STATE_REFRESHING = "refreshing"; + + private final TransactionRunner transactionRunner; + private final String workerId; + + public StructuredTableLeaseStrategy(TransactionRunner transactionRunner) { + this(transactionRunner, Collections.emptyMap()); + } + + public StructuredTableLeaseStrategy(TransactionRunner transactionRunner, Map properties) { + this.transactionRunner = transactionRunner; + this.workerId = generateWorkerId(properties); + } + + private static String generateWorkerId(Map properties) { + String instanceName = properties != null ? properties.get("instance.name") : null; + if (instanceName == null || instanceName.isEmpty()) { + instanceName = System.getenv("HOSTNAME"); + } + if (instanceName == null || instanceName.isEmpty()) { + try { + instanceName = java.net.InetAddress.getLocalHost().getHostName(); + } catch (Exception e) { + instanceName = "cdf-instance"; + } + } + return String.format("%s:%s", instanceName, UUID.randomUUID()); + } + + @Override + public String getName() { + return PROVIDER_NAME; + } + + @Override + public SecretLease acquireLease(String namespace, String key, long timeoutMs) throws IOException { + AtomicReference leaseRef = new AtomicReference<>(SecretLease.failed()); + long now = System.currentTimeMillis(); + + try { + TransactionRunners.run(transactionRunner, context -> { + StructuredTable table = context.getTable(LEASE_TABLE_ID); + List> primaryKey = getPrimaryKey(namespace, key); + Optional rowOpt = table.read(primaryKey); + + String state = STATE_IDLE; + long lockTimestamp = 0L; + + if (rowOpt.isPresent()) { + StructuredRow row = rowOpt.get(); + state = Optional.ofNullable(row.getString(LEASE_STATE_COL)).orElse(STATE_IDLE); + Long timestampObj = row.getLong(LEASE_LOCK_TIMESTAMP_COL); + lockTimestamp = timestampObj == null ? 0L : timestampObj; + } + + boolean isExpired = (now - lockTimestamp) > timeoutMs; + boolean isLockedByAnother = STATE_REFRESHING.equalsIgnoreCase(state) && !isExpired; + + if (!isLockedByAnother) { + List> fields = new ArrayList<>(5); + fields.add(Fields.stringField(LEASE_PROVIDER_COL, namespace)); + fields.add(Fields.stringField(LEASE_CREDENTIAL_COL, key)); + fields.add(Fields.stringField(LEASE_STATE_COL, STATE_REFRESHING)); + fields.add(Fields.longField(LEASE_LOCK_TIMESTAMP_COL, now)); + fields.add(Fields.stringField(LEASE_LOCK_HOLDER_COL, workerId)); + + table.upsert(fields); + leaseRef.set(SecretLease.acquired(String.valueOf(now), workerId)); + } + }, TableNotFoundException.class, InvalidFieldException.class); + + return leaseRef.get(); + } catch (TableNotFoundException e) { + throw new IOException("OAuth leases table not found", e); + } catch (Exception e) { + throw new IOException("Failed to acquire lease lock from StructuredTable", e); + } + } + + @Override + public void releaseLease(String namespace, String key, SecretLease lease) throws IOException { + if (lease == null || !lease.isAcquired()) { + return; + } + + try { + TransactionRunners.run(transactionRunner, context -> { + StructuredTable table = context.getTable(LEASE_TABLE_ID); + List> fields = new ArrayList<>(5); + fields.add(Fields.stringField(LEASE_PROVIDER_COL, namespace)); + fields.add(Fields.stringField(LEASE_CREDENTIAL_COL, key)); + fields.add(Fields.stringField(LEASE_STATE_COL, STATE_IDLE)); + fields.add(Fields.longField(LEASE_LOCK_TIMESTAMP_COL, 0L)); + fields.add(Fields.stringField(LEASE_LOCK_HOLDER_COL, "")); + + table.upsert(fields); + }, TableNotFoundException.class, InvalidFieldException.class); + } catch (Exception e) { + LOG.warn("Failed to release lease lock in StructuredTable for namespace {} key {}: {}", + namespace, key, e.getMessage()); + } + } + + private static List> getPrimaryKey(String namespace, String key) { + List> primaryKey = new ArrayList<>(2); + primaryKey.add(Fields.stringField(LEASE_PROVIDER_COL, namespace)); + primaryKey.add(Fields.stringField(LEASE_CREDENTIAL_COL, key)); + return primaryKey; + } +} diff --git a/cdap-security/src/main/java/io/cdap/cdap/security/store/secretmanager/SecretLeaseStrategyExtensionLoader.java b/cdap-security/src/main/java/io/cdap/cdap/security/store/secretmanager/SecretLeaseStrategyExtensionLoader.java new file mode 100644 index 000000000000..42ba2b996ea3 --- /dev/null +++ b/cdap-security/src/main/java/io/cdap/cdap/security/store/secretmanager/SecretLeaseStrategyExtensionLoader.java @@ -0,0 +1,54 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * 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.cdap.cdap.security.store.secretmanager; + +import io.cdap.cdap.securestore.spi.lease.SecretLeaseStrategy; +import io.cdap.cdap.common.lang.FilterClassLoader; +import io.cdap.cdap.extension.AbstractExtensionLoader; + +import java.util.Collections; +import java.util.Set; + +/** + * Extension loader for SecretLeaseStrategy implementations in cdap-security. + */ +public class SecretLeaseStrategyExtensionLoader extends AbstractExtensionLoader { + + public SecretLeaseStrategyExtensionLoader(String extensionDir) { + super(extensionDir); + } + + @Override + protected Set getSupportedTypesForProvider(SecretLeaseStrategy strategy) { + return Collections.singleton(strategy.getName()); + } + + @Override + protected FilterClassLoader.Filter getExtensionParentClassLoaderFilter() { + return new FilterClassLoader.Filter() { + @Override + public boolean acceptResource(String resource) { + return resource.startsWith("io/cdap/cdap/securestore/spi"); + } + + @Override + public boolean acceptPackage(String packageName) { + return packageName.startsWith("io.cdap.cdap.securestore.spi"); + } + }; + } +} diff --git a/cdap-security/src/main/java/io/cdap/cdap/security/store/secretmanager/SecretManagerSecureStoreService.java b/cdap-security/src/main/java/io/cdap/cdap/security/store/secretmanager/SecretManagerSecureStoreService.java index ebdf19ba1d40..69aff36732b2 100644 --- a/cdap-security/src/main/java/io/cdap/cdap/security/store/secretmanager/SecretManagerSecureStoreService.java +++ b/cdap-security/src/main/java/io/cdap/cdap/security/store/secretmanager/SecretManagerSecureStoreService.java @@ -36,9 +36,14 @@ import io.cdap.cdap.securestore.spi.secret.Secret; import io.cdap.cdap.securestore.spi.secret.SecretMetadata; import io.cdap.cdap.security.store.SecureStoreService; +import io.cdap.cdap.api.security.store.lease.SecureStoreLease; +import io.cdap.cdap.securestore.spi.lease.SecretLeaseStrategy; +import io.cdap.cdap.security.store.lease.StructuredTableLeaseStrategy; +import io.cdap.cdap.spi.data.transaction.TransactionRunner; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Map; import javax.annotation.Nullable; @@ -53,29 +58,53 @@ public class SecretManagerSecureStoreService extends AbstractIdleService impleme private static final Logger LOG = LoggerFactory.getLogger(SecretManagerSecureStoreService.class); + private final CConfiguration cConf; + private final TransactionRunner transactionRunner; private final NamespaceQueryAdmin namespaceQueryAdmin; private final SecretManagerContext context; private final String type; private SecretManager secretManager; + private volatile SecretLeaseStrategy leaseStrategy; @Inject SecretManagerSecureStoreService(CConfiguration cConf, NamespaceQueryAdmin namespaceQueryAdmin, - SecretStore store) { - this(namespaceQueryAdmin, + SecretStore store, @Nullable TransactionRunner transactionRunner) { + this(cConf, namespaceQueryAdmin, new DefaultSecretManagerContext(cConf, store), cConf.get(Constants.Security.Store.PROVIDER), new SecretManagerExtensionLoader(cConf.get(Constants.Security.Store.EXTENSIONS_DIR)) - .get(cConf.get(Constants.Security.Store.PROVIDER))); + .get(cConf.get(Constants.Security.Store.PROVIDER)), + null, + transactionRunner); } @VisibleForTesting SecretManagerSecureStoreService(NamespaceQueryAdmin namespaceQueryAdmin, SecretManagerContext context, String type, SecretManager secretManager) { + this(namespaceQueryAdmin, context, type, secretManager, null); + } + + @VisibleForTesting + SecretManagerSecureStoreService(NamespaceQueryAdmin namespaceQueryAdmin, + SecretManagerContext context, + String type, SecretManager secretManager, SecretLeaseStrategy leaseStrategy) { + this(CConfiguration.create(), namespaceQueryAdmin, context, type, secretManager, leaseStrategy, null); + } + + SecretManagerSecureStoreService(CConfiguration cConf, + NamespaceQueryAdmin namespaceQueryAdmin, + SecretManagerContext context, + String type, SecretManager secretManager, + @Nullable SecretLeaseStrategy leaseStrategy, + @Nullable TransactionRunner transactionRunner) { + this.cConf = cConf; this.namespaceQueryAdmin = namespaceQueryAdmin; this.context = context; this.type = type; this.secretManager = secretManager; + this.leaseStrategy = leaseStrategy; + this.transactionRunner = transactionRunner; } @Override @@ -197,4 +226,95 @@ private void destroySecretManager() { LOG.warn("Error occurred while stopping {}.", getClass().getSimpleName(), e); } } + + private SecretLeaseStrategy discoverLeaseStrategy() { + String provider = cConf.get(Constants.Security.Store.PROVIDER); + String extDir = cConf.get(Constants.Security.Store.EXTENSIONS_DIR); + LOG.info("Discovering SecretLeaseStrategy for provider '{}' in extension dir '{}'", provider, extDir); + SecretLeaseStrategy strategy = new SecretLeaseStrategyExtensionLoader(extDir).get(provider); + if (strategy != null) { + try { + Map props = context != null ? context.getProperties() : Collections.emptyMap(); + strategy.initialize(props); + LOG.info("Loaded and initialized SecretLeaseStrategy extension '{}' for provider '{}' with {} properties", + strategy.getName(), provider, props.size()); + return strategy; + } catch (IOException e) { + LOG.error("Failed to initialize SecretLeaseStrategy extension '{}' for provider '{}'", strategy.getName(), provider, e); + } + } + if (transactionRunner != null) { + LOG.info("Using fallback StructuredTableLeaseStrategy for provider '{}'", provider); + return new StructuredTableLeaseStrategy(transactionRunner, context != null ? context.getProperties() : Collections.emptyMap()); + } + LOG.warn("No SecretLeaseStrategy or TransactionRunner found for provider '{}'. Using no-op SecretLeaseStrategy.", provider); + return new SecretLeaseStrategy() { + @Override + public String getName() { + return "noop"; + } + + @Override + public io.cdap.cdap.securestore.spi.lease.SecretLease acquireLease(String namespace, String key, long timeoutMs) { + return io.cdap.cdap.securestore.spi.lease.SecretLease.failed(); + } + + @Override + public void releaseLease(String namespace, String key, io.cdap.cdap.securestore.spi.lease.SecretLease lease) { + } + }; + } + + private SecretLeaseStrategy getOrLoadLeaseStrategy() { + SecretLeaseStrategy strategy = this.leaseStrategy; + if (strategy != null) { + return strategy; + } + synchronized (this) { + strategy = this.leaseStrategy; + if (strategy == null) { + strategy = discoverLeaseStrategy(); + this.leaseStrategy = strategy; + } + return strategy; + } + } + + @Override + public SecureStoreLease acquireLease(String namespace, String name, long timeoutMs) throws IOException { + SecretLeaseStrategy strategy = getOrLoadLeaseStrategy(); + if (strategy == null) { + LOG.warn("Cannot acquire lease for namespace '{}' name '{}': lease strategy is null", namespace, name); + return SecureStoreLease.failed(); + } + try { + io.cdap.cdap.securestore.spi.lease.SecretLease spiLease = strategy.acquireLease(namespace, name, timeoutMs); + if (spiLease != null && spiLease.isAcquired()) { + LOG.info("Successfully acquired lease via '{}' for namespace '{}' name '{}' (holder: {}, timestamp: {})", + strategy.getName(), namespace, name, spiLease.getLockHolder(), spiLease.getLockTimestamp()); + return SecureStoreLease.acquired(spiLease.getLockTimestamp(), spiLease.getLockHolder()); + } + LOG.info("Lease acquisition via '{}' returned unacquired/held lease for namespace '{}' name '{}'", + strategy.getName(), namespace, name); + } catch (IOException | RuntimeException e) { + LOG.error("Exception occurred while acquiring lease via '{}' for namespace '{}' name '{}': {}", + strategy.getName(), namespace, name, e.getMessage(), e); + throw e; + } catch (Exception e) { + LOG.error("Unexpected exception occurred while acquiring lease via '{}' for namespace '{}' name '{}': {}", + strategy.getName(), namespace, name, e.getMessage(), e); + throw new IOException(e); + } + return SecureStoreLease.failed(); + } + + @Override + public void releaseLease(String namespace, String name, SecureStoreLease lease) throws IOException { + SecretLeaseStrategy strategy = getOrLoadLeaseStrategy(); + if (strategy != null && lease != null && lease.isAcquired()) { + io.cdap.cdap.securestore.spi.lease.SecretLease spiLease = + io.cdap.cdap.securestore.spi.lease.SecretLease.acquired(lease.getLockTimestamp(), lease.getLockHolder()); + strategy.releaseLease(namespace, name, spiLease); + } + } } diff --git a/cdap-system-app-api/src/main/java/io/cdap/cdap/api/service/http/SystemHttpServiceContext.java b/cdap-system-app-api/src/main/java/io/cdap/cdap/api/service/http/SystemHttpServiceContext.java index 51f3eef489be..83277a3cc5a6 100644 --- a/cdap-system-app-api/src/main/java/io/cdap/cdap/api/service/http/SystemHttpServiceContext.java +++ b/cdap-system-app-api/src/main/java/io/cdap/cdap/api/service/http/SystemHttpServiceContext.java @@ -27,6 +27,7 @@ import io.cdap.cdap.spi.data.transaction.TransactionRunner; import java.io.IOException; import java.util.Map; +import javax.annotation.Nullable; /** * A System HttpServiceContext that exposes capabilities beyond those available to service contexts