diff --git a/google-auth-library-java/oauth2_http/java/com/google/auth/mtls/MtlsUtils.java b/google-auth-library-java/oauth2_http/java/com/google/auth/mtls/MtlsUtils.java index a5c4c0f86e77..9cfa1f46d6eb 100644 --- a/google-auth-library-java/oauth2_http/java/com/google/auth/mtls/MtlsUtils.java +++ b/google-auth-library-java/oauth2_http/java/com/google/auth/mtls/MtlsUtils.java @@ -38,8 +38,10 @@ import java.io.FileInputStream; import java.io.IOException; import java.io.InputStream; +import java.security.MessageDigest; import java.util.Locale; import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; /** * Utility class for mTLS related operations. @@ -57,6 +59,129 @@ private MtlsUtils() { // Prevent instantiation for Utility class } + /** + * Returns if mutual TLS client certificate should be used. Delegates directly to + * getWorkloadCertPath to avoid duplicate logic. + */ + public static boolean useMtlsClientCertificate( + EnvironmentProvider envProvider, PropertyProvider propProvider) { + return getWorkloadCertPath(envProvider, propProvider) != null; + } + + /** + * Resolves and returns the path to the mutual TLS client certificate, or null if none should be + * used. + */ + public static @Nullable String getWorkloadCertPath( + EnvironmentProvider envProvider, PropertyProvider propProvider) { + String useClientCertificate = envProvider.getEnv("GOOGLE_API_USE_CLIENT_CERTIFICATE"); + if ("false".equalsIgnoreCase(useClientCertificate)) { + return null; + } + + String certConfigPath = envProvider.getEnv(CERTIFICATE_CONFIGURATION_ENV_VARIABLE); + if (!Strings.isNullOrEmpty(certConfigPath)) { + try { + WorkloadCertificateConfiguration config = + getWorkloadCertificateConfiguration(envProvider, propProvider, certConfigPath); + + File certFile = new File(config.getCertPath()); + File keyFile = new File(config.getPrivateKeyPath()); + if (!certFile.exists() || !keyFile.exists()) { + throw new IllegalStateException( + "Certificate config points to certificate/key files that do not exist on disk: " + + "cert_path=" + + config.getCertPath() + + ", key_path=" + + config.getPrivateKeyPath()); + } + return config.getCertPath(); + } catch (CertificateSourceUnavailableException e) { + // Certificate config file does not exist on disk -> safe fallback + } catch (IllegalStateException e) { + throw e; + } catch (Exception e) { + throw new IllegalStateException("Failed to parse certificate config: " + certConfigPath, e); + } + } else { + try { + WorkloadCertificateConfiguration config = + getWorkloadCertificateConfiguration(envProvider, propProvider, null); + File certFile = new File(config.getCertPath()); + File keyFile = new File(config.getPrivateKeyPath()); + if (certFile.exists() && keyFile.exists()) { + return config.getCertPath(); + } + } catch (CertificateSourceUnavailableException e) { + // Well-known gcloud certificate_config.json does not exist. Safe fallback to + // SPIFFE/well-known paths. + } catch (Exception e) { + // Ignore parsing errors for well-known config fallback + } + } + + String gkeCertPath = getGkeWorkloadCertPath(); + if (gkeCertPath != null) { + return gkeCertPath; + } + + String gceCertPath = getGceWorkloadCertPath(); + if (gceCertPath != null) { + return gceCertPath; + } + + return null; + } + + /** Dedicated GKE Fallback Resolution Path */ + public static @Nullable String getGkeWorkloadCertPath() { + String gkePath = "/var/run/secrets/workload-spiffe-credentials"; + File bundleFile = new File(gkePath, "credentialbundle.pem"); + if (bundleFile.exists()) { + return bundleFile.getAbsolutePath(); + } + return null; + } + + /** Dedicated GCE Fallback Resolution Path */ + public static @Nullable String getGceWorkloadCertPath() { + String gcePath = "/var/run/secrets/workload-spiffe-credentials"; + File certFile = new File(gcePath, "certificates.pem"); + File keyFile = new File(gcePath, "private_key.pem"); + if (certFile.exists() && keyFile.exists()) { + return certFile.getAbsolutePath(); + } + return null; + } + + /** Centralized SHA-256 Fingerprint Calculator */ + public static @Nullable String getCertificateFingerprint(@Nullable String certPath) { + if (certPath == null) { + return null; + } + File file = new File(certPath); + if (!file.exists()) { + return null; + } + try { + MessageDigest digest = MessageDigest.getInstance("SHA-256"); + try (FileInputStream fis = new FileInputStream(file)) { + byte[] byteArray = new byte[1024]; + int bytesCount; + while ((bytesCount = fis.read(byteArray)) != -1) { + digest.update(byteArray, 0, bytesCount); + } + } + StringBuilder sb = new StringBuilder(); + for (byte b : digest.digest()) { + sb.append(String.format("%02x", b)); + } + return sb.toString(); + } catch (Exception e) { + return null; + } + } + /** * Returns the path to the client certificate file specified by the loaded workload certificate * configuration. @@ -65,14 +190,17 @@ private MtlsUtils() { * @throws IOException if the certificate configuration cannot be found or loaded. */ public static String getCertificatePath( - EnvironmentProvider envProvider, PropertyProvider propProvider, String certConfigPathOverride) + EnvironmentProvider envProvider, + PropertyProvider propProvider, + @Nullable String certConfigPathOverride) throws IOException { String certPath = getWorkloadCertificateConfiguration(envProvider, propProvider, certConfigPathOverride) .getCertPath(); if (Strings.isNullOrEmpty(certPath)) { throw new CertificateSourceUnavailableException( - "Certificate configuration loaded successfully, but does not contain a 'certificate_file' path."); + "Certificate configuration loaded successfully, but does not contain a 'certificate_file'" + + " path."); } return certPath; } @@ -92,7 +220,9 @@ public static String getCertificatePath( * @throws IOException if the configuration file cannot be found, read, or parsed */ static WorkloadCertificateConfiguration getWorkloadCertificateConfiguration( - EnvironmentProvider envProvider, PropertyProvider propProvider, String certConfigPathOverride) + EnvironmentProvider envProvider, + PropertyProvider propProvider, + @Nullable String certConfigPathOverride) throws IOException { File certConfig; if (certConfigPathOverride != null) { diff --git a/google-auth-library-java/oauth2_http/javatests/com/google/auth/mtls/MtlsUtilsTest.java b/google-auth-library-java/oauth2_http/javatests/com/google/auth/mtls/MtlsUtilsTest.java index f3fdf05a4c32..2e4f4c94c15d 100644 --- a/google-auth-library-java/oauth2_http/javatests/com/google/auth/mtls/MtlsUtilsTest.java +++ b/google-auth-library-java/oauth2_http/javatests/com/google/auth/mtls/MtlsUtilsTest.java @@ -243,4 +243,95 @@ public String getProperty(String name, String def) { assertEquals("APPDATA environment variable is not set on Windows.", exception.getMessage()); } + + @Test + void useMtlsClientCertificate_trueWithNoCertsOnDisk_returnsFalseWithoutThrowing() { + EnvironmentProvider envProvider = + name -> "GOOGLE_API_USE_CLIENT_CERTIFICATE".equals(name) ? "true" : null; + PropertyProvider propProvider = (name, def) -> def; + + assertFalse(MtlsUtils.useMtlsClientCertificate(envProvider, propProvider)); + assertNull(MtlsUtils.getWorkloadCertPath(envProvider, propProvider)); + } + + @Test + void useMtlsClientCertificate_false_returnsFalse() { + EnvironmentProvider envProvider = + name -> "GOOGLE_API_USE_CLIENT_CERTIFICATE".equals(name) ? "false" : null; + PropertyProvider propProvider = (name, def) -> def; + + assertFalse(MtlsUtils.useMtlsClientCertificate(envProvider, propProvider)); + assertNull(MtlsUtils.getWorkloadCertPath(envProvider, propProvider)); + } + + @Test + void getWorkloadCertPath_missingConfigFile_returnsNullSafely() { + EnvironmentProvider envProvider = + name -> "GOOGLE_API_CERTIFICATE_CONFIG".equals(name) ? "/nonexistent/config.json" : null; + PropertyProvider propProvider = (name, def) -> def; + + assertNull(MtlsUtils.getWorkloadCertPath(envProvider, propProvider)); + } + + @Test + void getWorkloadCertPath_configPointsToMissingCertFiles_throwsIllegalStateException() + throws IOException { + Path configFile = tempDir.resolve("config.json"); + Files.write( + configFile, + "{\"cert_configs\":{\"workload\":{\"cert_path\":\"/nonexistent/cert.pem\",\"key_path\":\"/nonexistent/key.pem\"}}}" + .getBytes()); + + EnvironmentProvider envProvider = + name -> "GOOGLE_API_CERTIFICATE_CONFIG".equals(name) ? configFile.toString() : null; + PropertyProvider propProvider = (name, def) -> def; + + IllegalStateException exception = + assertThrows( + IllegalStateException.class, + () -> MtlsUtils.getWorkloadCertPath(envProvider, propProvider)); + assertTrue(exception.getMessage().contains("files that do not exist on disk")); + } + + @Test + void getWorkloadCertPath_validConfig_returnsCertPath() throws IOException { + Path certFile = tempDir.resolve("cert.pem"); + Path keyFile = tempDir.resolve("key.pem"); + Files.write(certFile, "dummy cert".getBytes()); + Files.write(keyFile, "dummy key".getBytes()); + + Path configFile = tempDir.resolve("config.json"); + String configJson = + String.format( + "{\"cert_configs\":{\"workload\":{\"cert_path\":\"%s\",\"key_path\":\"%s\"}}}", + certFile.toString().replace("\\", "\\\\"), keyFile.toString().replace("\\", "\\\\")); + Files.write(configFile, configJson.getBytes()); + + EnvironmentProvider envProvider = + name -> "GOOGLE_API_CERTIFICATE_CONFIG".equals(name) ? configFile.toString() : null; + PropertyProvider propProvider = (name, def) -> def; + + assertTrue(MtlsUtils.useMtlsClientCertificate(envProvider, propProvider)); + assertEquals(certFile.toString(), MtlsUtils.getWorkloadCertPath(envProvider, propProvider)); + } + + @Test + void getCertificateFingerprint_validFile_returnsSha256() throws IOException { + Path file = tempDir.resolve("test.crt"); + Files.write(file, "hello world".getBytes()); + + String fingerprint = MtlsUtils.getCertificateFingerprint(file.toString()); + assertNotNull(fingerprint); + assertEquals(64, fingerprint.length()); // SHA-256 hex string length + } + + @Test + void getGkeWorkloadCertPath_nonexistent_returnsNull() { + assertNull(MtlsUtils.getGkeWorkloadCertPath()); + } + + @Test + void getGceWorkloadCertPath_nonexistent_returnsNull() { + assertNull(MtlsUtils.getGceWorkloadCertPath()); + } } diff --git a/sdk-platform-java/gax-java/gax-grpc/pom.xml b/sdk-platform-java/gax-java/gax-grpc/pom.xml index d717ac0944b5..0abd59208e3c 100644 --- a/sdk-platform-java/gax-java/gax-grpc/pom.xml +++ b/sdk-platform-java/gax-java/gax-grpc/pom.xml @@ -162,8 +162,7 @@ maven-surefire-plugin - !InstantiatingGrpcChannelProviderTest#testLogDirectPathMisconfig_AttemptDirectPathNotSetAndAttemptDirectPathXdsSetViaEnv_warns,!InstantiatingGrpcChannelProviderTest#canUseDirectPath_directPathEnvVarNotSet_attemptDirectPathIsTrue,InstantiatingGrpcChannelProviderTest#testLogDirectPathMisconfigWrongCredential - + !InstantiatingGrpcChannelProviderTest#testLogDirectPathMisconfig_AttemptDirectPathNotSetAndAttemptDirectPathXdsSetViaEnv_warns diff --git a/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/ChannelPool.java b/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/ChannelPool.java index fab73a55dccf..f0e0db3d3017 100644 --- a/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/ChannelPool.java +++ b/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/ChannelPool.java @@ -31,6 +31,7 @@ import com.google.api.core.InternalApi; import com.google.api.gax.core.FixedExecutorProvider; +import com.google.api.gax.rpc.mtls.WorkloadCertificateUtils; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; @@ -72,18 +73,36 @@ @NullMarked class ChannelPool extends ManagedChannel { static final String CHANNEL_POOL_CONSECUTIVE_RESIZING_WARNING = - "The gRPC ChannelPool used in the client has been flagged to be repeatedly resizing (5+ times). See https://github.com/googleapis/google-cloud-java/blob/main/docs/grpc_channel_pool_guide.md for more information about this behavior."; + "The gRPC ChannelPool used in the client has been flagged to be repeatedly resizing (5+" + + " times). See" + + " https://github.com/googleapis/google-cloud-java/blob/main/docs/grpc_channel_pool_guide.md" + + " for more information about this behavior."; @VisibleForTesting static final Logger LOG = Logger.getLogger(ChannelPool.class.getName()); private static final java.time.Duration REFRESH_PERIOD = java.time.Duration.ofMinutes(50); private final ChannelPoolSettings settings; private final ChannelFactory channelFactory; private final FixedExecutorProvider backgroundExecutorProvider; + private final String workloadCertPath; private @Nullable ScheduledFuture refreshFuture = null; private @Nullable ScheduledFuture resizeFuture = null; + private static class DiskCheckResult { + final String fingerprint; + final long timestampNanos; + + DiskCheckResult(String fingerprint, long timestampNanos) { + this.fingerprint = fingerprint; + this.timestampNanos = timestampNanos; + } + } + + private volatile DiskCheckResult lastDiskCheck = null; + private final java.util.concurrent.locks.ReentrantLock diskCheckLock = + new java.util.concurrent.locks.ReentrantLock(); private final Object entryWriteLock = new Object(); + private volatile String activeCertFingerprint = ""; @VisibleForTesting final AtomicReference> entries = new AtomicReference<>(); private final AtomicInteger indexTicker = new AtomicInteger(); private final String authority; @@ -100,14 +119,15 @@ class ChannelPool extends ManagedChannel { static ChannelPool create( ChannelPoolSettings settings, ChannelFactory channelFactory, - @Nullable ScheduledExecutorService backgroundExecutor) + @Nullable ScheduledExecutorService backgroundExecutor, + @Nullable String workloadCertPath) throws IOException { FixedExecutorProvider executorProvider = backgroundExecutor == null ? FixedExecutorProvider.create(Executors.newSingleThreadScheduledExecutor(), true) : FixedExecutorProvider.create(backgroundExecutor, false); - return new ChannelPool(settings, channelFactory, executorProvider); + return new ChannelPool(settings, channelFactory, executorProvider, workloadCertPath); } /** @@ -121,11 +141,13 @@ static ChannelPool create( ChannelPool( ChannelPoolSettings settings, ChannelFactory channelFactory, - FixedExecutorProvider executorProvider) + FixedExecutorProvider executorProvider, + @Nullable String workloadCertPath) throws IOException { this.settings = settings; this.channelFactory = channelFactory; this.backgroundExecutorProvider = executorProvider; + this.workloadCertPath = workloadCertPath; ImmutableList.Builder initialListBuilder = ImmutableList.builder(); @@ -136,6 +158,11 @@ static ChannelPool create( entries.set(initialListBuilder.build()); authority = entries.get().get(0).channel.authority(); + if (workloadCertPath != null) { + this.activeCertFingerprint = + WorkloadCertificateUtils.getCertificateFingerprint(workloadCertPath); + } + if (!settings.isStaticSize()) { resizeFuture = backgroundExecutorProvider @@ -421,10 +448,57 @@ private void expand(int desiredSize) { private void refreshSafely() { try { - refresh(); + synchronized (entryWriteLock) { + if (workloadCertPath != null) { + String currentDiskFingerprint = getOrUpdateDiskFingerprint(workloadCertPath); + if (!currentDiskFingerprint.isEmpty()) { + this.activeCertFingerprint = currentDiskFingerprint; + } + } + refreshAll(); + } } catch (Exception e) { - LOG.log(Level.WARNING, "Failed to pre-emptively refresh channnels", e); + LOG.log(Level.WARNING, "Failed to pre-emptively refresh channels", e); + } + } + + private String getOrUpdateDiskFingerprint(String certPath) { + long now = System.nanoTime(); + DiskCheckResult cached = lastDiskCheck; + if (cached != null + && (now - cached.timestampNanos < java.util.concurrent.TimeUnit.SECONDS.toNanos(1))) { + return cached.fingerprint; + } + + diskCheckLock.lock(); + try { + cached = lastDiskCheck; + if (cached != null + && (now - cached.timestampNanos < java.util.concurrent.TimeUnit.SECONDS.toNanos(1))) { + return cached.fingerprint; + } + String fingerprint = WorkloadCertificateUtils.getCertificateFingerprint(certPath); + lastDiskCheck = new DiskCheckResult(fingerprint, System.nanoTime()); + return fingerprint; + } finally { + diskCheckLock.unlock(); + } + } + + @VisibleForTesting + void invalidateDiskFingerprintCache() { + this.lastDiskCheck = null; + } + + boolean shouldRefresh() { + if (workloadCertPath == null) { + return false; + } + String currentDiskFingerprint = getOrUpdateDiskFingerprint(workloadCertPath); + if (currentDiskFingerprint.isEmpty()) { + return false; } + return !currentDiskFingerprint.equalsIgnoreCase(activeCertFingerprint); } /** @@ -443,7 +517,36 @@ void refresh() { // - then thread2 will shut down channel that thread1 will put back into circulation (after it // replaces the list) synchronized (entryWriteLock) { - LOG.fine("Refreshing all channels"); + if (workloadCertPath == null) { + refreshAll(); + return; + } + String currentDiskFingerprint = getOrUpdateDiskFingerprint(workloadCertPath); + if (currentDiskFingerprint.isEmpty()) { + return; + } + + // Double-check fingerprint inside the lock + if (currentDiskFingerprint.equalsIgnoreCase(this.activeCertFingerprint)) { + LOG.fine( + "Channel pool was already refreshed by a concurrent thread, skipping duplicate" + + " refresh"); + return; + } + + this.activeCertFingerprint = currentDiskFingerprint; + refreshAll(); + } + } + + @InternalApi("Visible for testing") + void refreshAll() { + synchronized (entryWriteLock) { + LOG.fine( + "Refreshing all channels" + + (activeCertFingerprint == null + ? "" + : " with certificate fingerprint: " + activeCertFingerprint)); ArrayList newEntries = new ArrayList<>(entries.get()); for (int i = 0; i < newEntries.size(); i++) { @@ -621,12 +724,19 @@ public ClientCall newCall( } } - /** ClientCall wrapper that makes sure to decrement the outstanding RPC count on completion. */ + /** + * ClientCall wrapper that makes sure to decrement the outstanding RPC count on completion. + * + *

Contract: Exactly one call to {@link #start(Listener, Metadata)} is required to balance + * reference counts. Early cancellation before {@code start()} is recorded and safely decrements + * the reference count when {@code start()} is subsequently invoked. + */ static class ReleasingClientCall extends SimpleForwardingClientCall { private @Nullable CancellationException cancellationException; final Entry entry; private final AtomicBoolean wasClosed = new AtomicBoolean(); private final AtomicBoolean wasReleased = new AtomicBoolean(); + private final AtomicBoolean wasStarted = new AtomicBoolean(); public ReleasingClientCall(ClientCall delegate, Entry entry) { super(delegate); @@ -635,7 +745,11 @@ public ReleasingClientCall(ClientCall delegate, Entry entry) { @Override public void start(Listener responseListener, Metadata headers) { + wasStarted.set(true); if (cancellationException != null) { + if (wasReleased.compareAndSet(false, true)) { + entry.release(); + } throw new IllegalStateException("Call is already cancelled", cancellationException); } try { @@ -646,7 +760,8 @@ public void onClose(Status status, Metadata trailers) { if (!wasClosed.compareAndSet(false, true)) { LOG.log( Level.WARNING, - "Call is being closed more than once. Please make sure that onClose() is not being manually called."); + "Call is being closed more than once. Please make sure that onClose() is not" + + " being manually called."); return; } try { @@ -657,7 +772,8 @@ public void onClose(Status status, Metadata trailers) { } else { LOG.log( Level.WARNING, - "Entry was released before the call is closed. This may be due to an exception on start of the call."); + "Entry was released before the call is closed. This may be due to an" + + " exception on start of the call."); } } } @@ -670,7 +786,8 @@ public void onClose(Status status, Metadata trailers) { } else { LOG.log( Level.WARNING, - "The entry is already released. This indicates that onClose() has already been called previously"); + "The entry is already released. This indicates that onClose() has already been called" + + " previously"); } throw e; } @@ -679,7 +796,12 @@ public void onClose(Status status, Metadata trailers) { @Override public void cancel(@Nullable String message, @Nullable Throwable cause) { this.cancellationException = new CancellationException(message); - super.cancel(message, cause); + if (delegate() != null) { + super.cancel(message, cause); + } + if (!wasStarted.get() && wasReleased.compareAndSet(false, true)) { + entry.release(); + } } } } diff --git a/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/GrpcCallContext.java b/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/GrpcCallContext.java index 23f56c5f8951..428531848b12 100644 --- a/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/GrpcCallContext.java +++ b/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/GrpcCallContext.java @@ -99,6 +99,7 @@ public final class GrpcCallContext implements ApiCallContext { private final ApiCallContextOptions options; private final EndpointContext endpointContext; private final boolean isDirectPath; + @Nullable private final TransportChannel transportChannel; /** Returns an empty instance with a null channel and default {@link CallOptions}. */ public static GrpcCallContext createDefault() { @@ -115,7 +116,8 @@ public static GrpcCallContext createDefault() { null, null, null, - false); + false, + null); } /** Returns an instance with the given channel and {@link CallOptions}. */ @@ -133,7 +135,8 @@ public static GrpcCallContext of(Channel channel, CallOptions callOptions) { null, null, null, - false); + false, + null); } private GrpcCallContext( @@ -149,7 +152,8 @@ private GrpcCallContext( @Nullable RetrySettings retrySettings, @Nullable Set retryableCodes, @Nullable EndpointContext endpointContext, - boolean isDirectPath) { + boolean isDirectPath, + @Nullable TransportChannel transportChannel) { this.channel = channel; this.credentials = credentials; Preconditions.checkNotNull(callOptions); @@ -169,6 +173,7 @@ private GrpcCallContext( this.endpointContext = endpointContext == null ? EndpointContext.getDefaultInstance() : endpointContext; this.isDirectPath = isDirectPath; + this.transportChannel = transportChannel; } /** @@ -210,7 +215,13 @@ public GrpcCallContext withCredentials(Credentials newCredentials) { retrySettings, retryableCodes, endpointContext, - isDirectPath); + isDirectPath, + transportChannel); + } + + @Override + public TransportChannel getTransportChannel() { + return transportChannel; } @Override @@ -234,7 +245,8 @@ public GrpcCallContext withTransportChannel(TransportChannel inputChannel) { retrySettings, retryableCodes, endpointContext, - transportChannel.isDirectPath()); + transportChannel.isDirectPath(), + inputChannel); } @Override @@ -253,7 +265,8 @@ public GrpcCallContext withEndpointContext(EndpointContext endpointContext) { retrySettings, retryableCodes, endpointContext, - isDirectPath); + isDirectPath, + transportChannel); } /** This method is obsolete. Use {@link #withTimeoutDuration(java.time.Duration)} instead. */ @@ -271,7 +284,7 @@ public GrpcCallContext withTimeoutDuration(java.time.@Nullable Duration timeout) } // Prevent expanding timeouts - if (timeout != null && this.timeout != null && this.timeout.compareTo(timeout) <= 0) { + if (this.timeout != null && (timeout == null || this.timeout.compareTo(timeout) <= 0)) { return this; } @@ -288,7 +301,8 @@ public GrpcCallContext withTimeoutDuration(java.time.@Nullable Duration timeout) retrySettings, retryableCodes, endpointContext, - isDirectPath); + isDirectPath, + transportChannel); } @Override @@ -334,7 +348,8 @@ public GrpcCallContext withStreamWaitTimeoutDuration( retrySettings, retryableCodes, endpointContext, - isDirectPath); + isDirectPath, + transportChannel); } /** @@ -369,7 +384,8 @@ public GrpcCallContext withStreamIdleTimeoutDuration( retrySettings, retryableCodes, endpointContext, - isDirectPath); + isDirectPath, + transportChannel); } @BetaApi("The surface for channel affinity is not stable yet and may change in the future.") @@ -387,7 +403,8 @@ public GrpcCallContext withChannelAffinity(@Nullable Integer affinity) { retrySettings, retryableCodes, endpointContext, - isDirectPath); + isDirectPath, + transportChannel); } @BetaApi("The surface for extra headers is not stable yet and may change in the future.") @@ -409,7 +426,8 @@ public GrpcCallContext withExtraHeaders(Map> extraHeaders) retrySettings, retryableCodes, endpointContext, - isDirectPath); + isDirectPath, + transportChannel); } @Override @@ -432,7 +450,8 @@ public GrpcCallContext withRetrySettings(RetrySettings retrySettings) { retrySettings, retryableCodes, endpointContext, - isDirectPath); + isDirectPath, + transportChannel); } @Override @@ -455,7 +474,8 @@ public GrpcCallContext withRetryableCodes(Set retryableCodes) { retrySettings, retryableCodes, endpointContext, - isDirectPath); + isDirectPath, + transportChannel); } @Override @@ -542,6 +562,12 @@ public ApiCallContext merge(ApiCallContext inputCallContext) { newCallOptions = newCallOptions.withOption(TRACER_KEY, newTracer); } + TransportChannel newTransportChannel = grpcCallContext.transportChannel; + if (newTransportChannel == null + && (grpcCallContext.channel == null || grpcCallContext.channel.equals(channel))) { + newTransportChannel = transportChannel; + } + // The EndpointContext is not updated as there should be no reason for a user // to update this. return new GrpcCallContext( @@ -557,7 +583,8 @@ public ApiCallContext merge(ApiCallContext inputCallContext) { newRetrySettings, newRetryableCodes, endpointContext, - newIsDirectPath); + newIsDirectPath, + newTransportChannel); } /** The {@link Channel} set on this context. */ @@ -635,7 +662,8 @@ public GrpcCallContext withChannel(@Nullable Channel newChannel) { retrySettings, retryableCodes, endpointContext, - isDirectPath); + isDirectPath, + (newChannel == null || newChannel.equals(channel)) ? transportChannel : null); } /** Returns a new instance with the call options set to the given call options. */ @@ -653,7 +681,8 @@ public GrpcCallContext withCallOptions(CallOptions newCallOptions) { retrySettings, retryableCodes, endpointContext, - isDirectPath); + isDirectPath, + transportChannel); } public GrpcCallContext withRequestParamsDynamicHeaderOption(String requestParams) { @@ -698,7 +727,8 @@ public GrpcCallContext withOption(Key key, T value) { retrySettings, retryableCodes, endpointContext, - isDirectPath); + isDirectPath, + transportChannel); } /** {@inheritDoc} */ @@ -759,7 +789,8 @@ public int hashCode() { options, retrySettings, retryableCodes, - endpointContext); + endpointContext, + transportChannel); } @Override @@ -783,7 +814,8 @@ public boolean equals(@Nullable Object o) { && Objects.equals(options, that.options) && Objects.equals(retrySettings, that.retrySettings) && Objects.equals(retryableCodes, that.retryableCodes) - && Objects.equals(endpointContext, that.endpointContext); + && Objects.equals(endpointContext, that.endpointContext) + && Objects.equals(transportChannel, that.transportChannel); } Metadata getMetadata() { diff --git a/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/GrpcTransportChannel.java b/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/GrpcTransportChannel.java index e0a520facb17..31ede726f3f3 100644 --- a/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/GrpcTransportChannel.java +++ b/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/GrpcTransportChannel.java @@ -68,6 +68,23 @@ public Channel getChannel() { return getManagedChannel(); } + @Override + public void refresh() { + Channel channel = getChannel(); + if (channel instanceof ChannelPool) { + ((ChannelPool) channel).refresh(); + } + } + + @Override + public boolean shouldRefresh() { + Channel channel = getChannel(); + if (channel instanceof ChannelPool) { + return ((ChannelPool) channel).shouldRefresh(); + } + return false; + } + @Override public void shutdown() { getManagedChannel().shutdown(); diff --git a/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProvider.java b/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProvider.java index ac42396f006a..07ffa3b86b5e 100644 --- a/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProvider.java +++ b/sdk-platform-java/gax-java/gax-grpc/src/main/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProvider.java @@ -406,7 +406,8 @@ private TransportChannel createChannel() throws IOException { ChannelPool.create( channelPoolSettings, InstantiatingGrpcChannelProvider.this::createSingleChannel, - backgroundExecutor)) + backgroundExecutor, + certificateBasedAccess.getWorkloadCertPath())) .setDirectPath(this.canUseDirectPath()) .build(); } @@ -465,8 +466,9 @@ private void logDirectPathMisconfig() { level, "Env var " + DIRECT_PATH_ENV_ENABLE_XDS - + " was found and set to TRUE, but DirectPath was not enabled for this client. If this is intended for " - + "this client, please note that this is a misconfiguration and set the attemptDirectPath option as well."); + + " was found and set to TRUE, but DirectPath was not enabled for this client. If" + + " this is intended for this client, please note that this is a misconfiguration" + + " and set the attemptDirectPath option as well."); } // Case 2: Direct Path xDS was enabled via Builder. Direct Path Traffic Director must be set // (enabled with `setAttemptDirectPath(true)`) along with xDS. @@ -474,7 +476,9 @@ private void logDirectPathMisconfig() { else if (isDirectPathXdsEnabledViaBuilderOption()) { LOG.log( level, - "DirectPath is misconfigured. The DirectPath XDS option was set, but the attemptDirectPath option was not. Please set both the attemptDirectPath and attemptDirectPathXds options."); + "DirectPath is misconfigured. The DirectPath XDS option was set, but the" + + " attemptDirectPath option was not. Please set both the attemptDirectPath and" + + " attemptDirectPathXds options."); } } else { // Case 3: credential is not correctly set @@ -666,7 +670,8 @@ ChannelCredentials createS2ASecuredChannelCredentials() { // Fallback to plaintext connection to S2A. LOG.log( Level.INFO, - "Cannot establish an mTLS connection to S2A because autoconfig endpoint did not return a mtls address to reach S2A."); + "Cannot establish an mTLS connection to S2A because autoconfig endpoint did not" + + " return a mtls address to reach S2A."); s2aChannelCredentials = createPlaintextToS2AChannelCredentials(plaintextAddress); return s2aChannelCredentials; } @@ -685,7 +690,9 @@ ChannelCredentials createS2ASecuredChannelCredentials() { // Fallback to plaintext-to-S2A connection on error. LOG.log( Level.WARNING, - "Cannot establish an mTLS connection to S2A due to error creating MTLS to MDS TlsChannelCredentials credentials, falling back to plaintext connection to S2A: " + "Cannot establish an mTLS connection to S2A due to error creating MTLS to MDS" + + " TlsChannelCredentials credentials, falling back to plaintext connection to" + + " S2A: " + ignore.getMessage()); s2aChannelCredentials = createPlaintextToS2AChannelCredentials(plaintextAddress); return s2aChannelCredentials; @@ -1403,7 +1410,8 @@ public InstantiatingGrpcChannelProvider build() { "DefaultMtlsProviderFactory encountered unexpected IOException: " + e.getMessage()); LOG.log( Level.WARNING, - "mTLS configuration was detected on the device, but mTLS failed to initialize. Falling back to non-mTLS channel."); + "mTLS configuration was detected on the device, but mTLS failed to initialize." + + " Falling back to non-mTLS channel."); } } } diff --git a/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/ChannelPoolTest.java b/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/ChannelPoolTest.java index 5bfdc7754759..6cb7be99c9eb 100644 --- a/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/ChannelPoolTest.java +++ b/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/ChannelPoolTest.java @@ -81,13 +81,18 @@ class ChannelPoolTest { private static final int DEFAULT_AWAIT_TERMINATION_SEC = 10; private ChannelPool pool; + private java.nio.file.Path tempCert; @AfterEach - void cleanup() throws InterruptedException { + void cleanup() throws InterruptedException, IOException { if (pool != null) { pool.shutdown(); pool.awaitTermination(DEFAULT_AWAIT_TERMINATION_SEC, TimeUnit.SECONDS); } + if (tempCert != null) { + java.nio.file.Files.deleteIfExists(tempCert); + tempCert = null; + } } @Test @@ -101,6 +106,7 @@ void testAuthority() throws IOException { ChannelPool.create( ChannelPoolSettings.staticallySized(2), new FakeChannelFactory(Arrays.asList(sub1, sub2)), + null, null); assertThat(pool.authority()).isEqualTo("myAuth"); } @@ -117,6 +123,7 @@ void testRoundRobin() throws IOException { ChannelPool.create( ChannelPoolSettings.staticallySized(channels.size()), new FakeChannelFactory(channels), + null, null); verifyTargetChannel(pool, channels, sub1); @@ -195,6 +202,7 @@ void ensureEvenDistribution() throws InterruptedException, IOException { ChannelPool.create( ChannelPoolSettings.staticallySized(numChannels), new FakeChannelFactory(Arrays.asList(channels)), + null, null); int numThreads = 20; @@ -233,6 +241,7 @@ void channelPrimerShouldCallPoolConstruction() throws IOException { .setPreemptiveRefreshEnabled(true) .build(), new FakeChannelFactory(Arrays.asList(channel1, channel2), mockChannelPrimer), + null, null); Mockito.verify(mockChannelPrimer, Mockito.times(2)) .primeChannel(Mockito.any(ManagedChannel.class)); @@ -273,7 +282,8 @@ void channelPrimerIsCalledPeriodically() throws IOException { .setPreemptiveRefreshEnabled(true) .build(), channelFactory, - provider); + provider, + null); // 1 call during the creation Mockito.verify(mockChannelPrimer, Mockito.times(1)) .primeChannel(Mockito.any(ManagedChannel.class)); @@ -297,7 +307,7 @@ void callShouldCompleteAfterCreation() throws IOException { ManagedChannel replacementChannel = mock(ManagedChannel.class); FakeChannelFactory channelFactory = new FakeChannelFactory(ImmutableList.of(underlyingChannel, replacementChannel)); - pool = ChannelPool.create(ChannelPoolSettings.staticallySized(1), channelFactory, null); + pool = ChannelPool.create(ChannelPoolSettings.staticallySized(1), channelFactory, null, null); // create a mock call when new call comes to the underlying channel MockClientCall mockClientCall = new MockClientCall<>(1, Status.OK); @@ -322,7 +332,7 @@ void callShouldCompleteAfterCreation() throws IOException { ClientCall call = pool.newCall(FakeMethodDescriptor.create(), CallOptions.DEFAULT); - pool.refresh(); + pool.refreshAll(); // shutdown is not called because there is still an outstanding call, even if it hasn't started Mockito.verify(underlyingChannel, Mockito.after(200).never()).shutdown(); @@ -346,7 +356,7 @@ void callShouldCompleteAfterStarted() throws IOException { FakeChannelFactory channelFactory = new FakeChannelFactory(ImmutableList.of(underlyingChannel, replacementChannel)); - pool = ChannelPool.create(ChannelPoolSettings.staticallySized(1), channelFactory, null); + pool = ChannelPool.create(ChannelPoolSettings.staticallySized(1), channelFactory, null, null); // create a mock call when new call comes to the underlying channel MockClientCall mockClientCall = new MockClientCall<>(1, Status.OK); @@ -373,7 +383,7 @@ void callShouldCompleteAfterStarted() throws IOException { // start clientCall call.start(listener, new Metadata()); - pool.refresh(); + pool.refreshAll(); // shutdown is not called because there is still an outstanding call Mockito.verify(underlyingChannel, Mockito.after(200).never()).shutdown(); @@ -391,7 +401,7 @@ void channelShouldShutdown() throws IOException { FakeChannelFactory channelFactory = new FakeChannelFactory(ImmutableList.of(underlyingChannel, replacementChannel)); - pool = ChannelPool.create(ChannelPoolSettings.staticallySized(1), channelFactory, null); + pool = ChannelPool.create(ChannelPoolSettings.staticallySized(1), channelFactory, null, null); // create a mock call when new call comes to the underlying channel MockClientCall mockClientCall = new MockClientCall<>(1, Status.OK); @@ -422,11 +432,80 @@ void channelShouldShutdown() throws IOException { call.sendMessage("message"); // shutdown is not called because it has not been shutdown yet Mockito.verify(underlyingChannel, Mockito.after(200).never()).shutdown(); - pool.refresh(); + pool.refreshAll(); // shutdown is called because the outstanding call has completed Mockito.verify(underlyingChannel, Mockito.atLeastOnce()).shutdown(); } + @Test + void testCancelBeforeStartReleasesChannelEntry() throws IOException { + ManagedChannel underlyingChannel = mock(ManagedChannel.class); + ManagedChannel replacementChannel = mock(ManagedChannel.class); + FakeChannelFactory channelFactory = + new FakeChannelFactory(ImmutableList.of(underlyingChannel, replacementChannel)); + pool = ChannelPool.create(ChannelPoolSettings.staticallySized(1), channelFactory, null, null); + + ClientCall call = + pool.newCall(FakeMethodDescriptor.create(), CallOptions.DEFAULT); + + pool.refreshAll(); + Mockito.verify(underlyingChannel, Mockito.never()).shutdown(); + + call.cancel("Cancelled early", null); + Mockito.verify(underlyingChannel, Mockito.times(1)).shutdown(); + } + + @Test + void channelReactiveMTlsRefreshShouldConditionallySwapChannels() + throws IOException, InterruptedException { + ManagedChannel underlyingChannel1 = Mockito.mock(ManagedChannel.class); + ManagedChannel underlyingChannel2 = Mockito.mock(ManagedChannel.class); + + FakeChannelFactory channelFactory = + new FakeChannelFactory(ImmutableList.of(underlyingChannel1, underlyingChannel2)); + + // Create a temp file to act as the cert + tempCert = java.nio.file.Files.createTempFile("cert", ".pem"); + + java.nio.file.Path clientCert = + java.nio.file.Paths.get("src", "test", "resources", "client_cert.pem"); + java.nio.file.Files.copy( + clientCert, tempCert, java.nio.file.StandardCopyOption.REPLACE_EXISTING); + + ChannelPoolSettings channelPoolSettings = + ChannelPoolSettings.builder().setInitialChannelCount(1).build(); + + pool = ChannelPool.create(channelPoolSettings, channelFactory, null, tempCert.toString()); + + // Initially uses channel1 + pool.newCall(FakeMethodDescriptor.create(), CallOptions.DEFAULT); + Mockito.verify(underlyingChannel1, Mockito.times(1)) + .newCall(Mockito.>any(), Mockito.any(CallOptions.class)); + + // Try a reactive refresh *without* changing the cert content (should no-op) + pool.refresh(); + + // Verify it's STILL channel1 + pool.newCall(FakeMethodDescriptor.create(), CallOptions.DEFAULT); + Mockito.verify(underlyingChannel1, Mockito.times(2)) + .newCall(Mockito.>any(), Mockito.any(CallOptions.class)); + + // The ChannelPool caches fingerprints for 1000ms, wait for it to expire + pool.invalidateDiskFingerprintCache(); + + java.nio.file.Path rootCert = + java.nio.file.Paths.get("src", "test", "resources", "root_cert.pem"); + java.nio.file.Files.copy(rootCert, tempCert, java.nio.file.StandardCopyOption.REPLACE_EXISTING); + + // Try a reactive refresh *with* a changed cert content (should swap channels) + pool.refresh(); + + // Verify it is NOW channel2 + pool.newCall(FakeMethodDescriptor.create(), CallOptions.DEFAULT); + Mockito.verify(underlyingChannel2, Mockito.times(1)) + .newCall(Mockito.>any(), Mockito.any(CallOptions.class)); + } + @Test void channelRefreshShouldSwapChannels() throws IOException { ManagedChannel underlyingChannel1 = mock(ManagedChannel.class); @@ -450,7 +529,8 @@ void channelRefreshShouldSwapChannels() throws IOException { .setPreemptiveRefreshEnabled(true) .build(), channelFactory, - provider); + provider, + null); Mockito.reset(underlyingChannel1); pool.newCall(FakeMethodDescriptor.create(), CallOptions.DEFAULT); @@ -459,7 +539,7 @@ void channelRefreshShouldSwapChannels() throws IOException { .newCall(Mockito.>any(), Mockito.any(CallOptions.class)); // swap channel - pool.refresh(); + pool.refreshAll(); pool.newCall(FakeMethodDescriptor.create(), CallOptions.DEFAULT); @@ -467,6 +547,37 @@ void channelRefreshShouldSwapChannels() throws IOException { .newCall(Mockito.>any(), Mockito.any(CallOptions.class)); } + @Test + void testRefreshWithNullWorkloadCertPathSwapsChannel() throws IOException { + ScheduledExecutorService executor = + Mockito.mock(ScheduledExecutorService.class, Mockito.withSettings().withoutAnnotations()); + FixedExecutorProvider provider = FixedExecutorProvider.create(executor); + ManagedChannel underlyingChannel1 = Mockito.mock(ManagedChannel.class); + ManagedChannel underlyingChannel2 = Mockito.mock(ManagedChannel.class); + FakeChannelFactory channelFactory = + new FakeChannelFactory(ImmutableList.of(underlyingChannel1, underlyingChannel2)); + pool = + new ChannelPool( + ChannelPoolSettings.staticallySized(1).toBuilder() + .setPreemptiveRefreshEnabled(true) + .build(), + channelFactory, + provider, + null); + Mockito.reset(underlyingChannel1); + + pool.newCall(FakeMethodDescriptor.create(), CallOptions.DEFAULT); + Mockito.verify(underlyingChannel1, Mockito.only()) + .newCall(Mockito.>any(), Mockito.any(CallOptions.class)); + + // Calling refresh() when workloadCertPath is null should fall back to refreshAll() + pool.refresh(); + + pool.newCall(FakeMethodDescriptor.create(), CallOptions.DEFAULT); + Mockito.verify(underlyingChannel2, Mockito.only()) + .newCall(Mockito.>any(), Mockito.any(CallOptions.class)); + } + @Test void channelCountShouldNotChangeWhenOutstandingRpcsAreWithinLimits() throws Exception { ScheduledExecutorService executor = @@ -486,7 +597,8 @@ void channelCountShouldNotChangeWhenOutstandingRpcsAreWithinLimits() throws Exce .setMaxRpcsPerChannel(2) .build(), channelFactory, - provider); + provider, + null); assertThat(pool.entries.get()).hasSize(2); // Start the minimum number of @@ -553,7 +665,8 @@ void customResizeDeltaIsRespected() throws Exception { .setMaxResizeDelta(5) .build(), channelFactory, - provider); + provider, + null); assertThat(pool.entries.get()).hasSize(2); // Add 20 RPCs to push expansion @@ -586,7 +699,8 @@ void removedIdleChannelsAreShutdown() throws Exception { .setMaxRpcsPerChannel(2) .build(), channelFactory, - provider); + provider, + null); assertThat(pool.entries.get()).hasSize(2); // With no outstanding RPCs, the pool should shrink @@ -614,7 +728,8 @@ void removedActiveChannelsAreShutdown() throws Exception { .setMaxRpcsPerChannel(2) .build(), channelFactory, - provider); + provider, + null); assertThat(pool.entries.get()).hasSize(2); // Start 2 RPCs @@ -652,7 +767,7 @@ void testReleasingClientCallCancelEarly() throws IOException { Mockito.when(fakeChannel.newCall(Mockito.any(), Mockito.any())).thenReturn(mockClientCall); ChannelPoolSettings channelPoolSettings = ChannelPoolSettings.staticallySized(1); ChannelFactory factory = new FakeChannelFactory(ImmutableList.of(fakeChannel)); - pool = ChannelPool.create(channelPoolSettings, factory, null); + pool = ChannelPool.create(channelPoolSettings, factory, null, null); EndpointContext endpointContext = Mockito.mock(EndpointContext.class, Mockito.withSettings().withoutAnnotations()); @@ -717,7 +832,8 @@ void repeatedResizingLogsWarningOnExpand() throws Exception { .setMaxChannelCount(10) .build(), channelFactory, - provider); + provider, + null); assertThat(pool.entries.get()).hasSize(1); FakeLogHandler logHandler = new FakeLogHandler(); @@ -769,7 +885,8 @@ void repeatedResizingLogsWarningOnShrink() throws Exception { .setMaxChannelCount(10) .build(), channelFactory, - provider); + provider, + null); assertThat(pool.entries.get()).hasSize(10); FakeLogHandler logHandler = new FakeLogHandler(); @@ -805,7 +922,7 @@ void testDoubleRelease() throws Exception { ChannelPoolSettings channelPoolSettings = ChannelPoolSettings.staticallySized(1); ChannelFactory factory = new FakeChannelFactory(ImmutableList.of(fakeChannel)); - pool = ChannelPool.create(channelPoolSettings, factory, null); + pool = ChannelPool.create(channelPoolSettings, factory, null, null); EndpointContext endpointContext = Mockito.mock(EndpointContext.class, Mockito.withSettings().withoutAnnotations()); @@ -843,7 +960,8 @@ void testDoubleRelease() throws Exception { // Ensure that the channel pool properly logged the double call and kept the refCount correct assertThat(logHandler.getAllMessages()) .contains( - "Call is being closed more than once. Please make sure that onClose() is not being manually called."); + "Call is being closed more than once. Please make sure that onClose() is not being" + + " manually called."); assertThat(pool.entries.get()).hasSize(1); ChannelPool.Entry entry = pool.entries.get().get(0); assertThat(entry.outstandingRpcs.get()).isEqualTo(0); @@ -879,7 +997,8 @@ void minChannelsClampedToMaxChannelCountUnderHighLoad() throws Exception { .setMaxChannelCount(5) .build(), channelFactory, - provider); + provider, + null); assertThat(pool.entries.get()).hasSize(1); // Add 20 RPCs, which would require 10 channels (20/2) @@ -914,7 +1033,8 @@ void maxChannelsClampedToMinChannelCountUnderLowLoad() throws Exception { .setMaxChannelCount(10) .build(), channelFactory, - provider); + provider, + null); assertThat(pool.entries.get()).hasSize(5); // With no outstanding RPCs, the pool should want to shrink to 0 diff --git a/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcCallContextTest.java b/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcCallContextTest.java index e20767fdb8ed..59d5bbf568be 100644 --- a/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcCallContextTest.java +++ b/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcCallContextTest.java @@ -494,4 +494,54 @@ private static Map> createTestExtraHeaders(String... keyVal } return extraHeaders; } + + @Test + public void testEqualsAndHashCode() { + ManagedChannel managedChannel1 = org.mockito.Mockito.mock(ManagedChannel.class); + ManagedChannel managedChannel2 = org.mockito.Mockito.mock(ManagedChannel.class); + + GrpcTransportChannel transportChannel1 = GrpcTransportChannel.create(managedChannel1); + GrpcTransportChannel transportChannel2 = GrpcTransportChannel.create(managedChannel2); + + GrpcCallContext context1 = + GrpcCallContext.createDefault().withTransportChannel(transportChannel1); + GrpcCallContext context2 = + GrpcCallContext.createDefault().withTransportChannel(transportChannel1); + GrpcCallContext context3 = + GrpcCallContext.createDefault().withTransportChannel(transportChannel2); + + org.junit.jupiter.api.Assertions.assertEquals(context1, context2); + org.junit.jupiter.api.Assertions.assertEquals(context1.hashCode(), context2.hashCode()); + + org.junit.jupiter.api.Assertions.assertNotEquals(context1, context3); + } + + @Test + public void testMergeWithCustomChannelClearsTransportChannel() { + ManagedChannel defaultChannel = org.mockito.Mockito.mock(ManagedChannel.class); + ManagedChannel customChannel = org.mockito.Mockito.mock(ManagedChannel.class); + GrpcTransportChannel transportChannel = GrpcTransportChannel.create(defaultChannel); + + GrpcCallContext baseContext = + GrpcCallContext.createDefault().withTransportChannel(transportChannel); + GrpcCallContext overrideContext = GrpcCallContext.of(customChannel, CallOptions.DEFAULT); + + GrpcCallContext mergedContext = (GrpcCallContext) baseContext.merge(overrideContext); + assertEquals(customChannel, mergedContext.getChannel()); + assertNull(mergedContext.getTransportChannel()); + } + + @Test + public void testWithChannelWithCustomChannelClearsTransportChannel() { + ManagedChannel defaultChannel = org.mockito.Mockito.mock(ManagedChannel.class); + ManagedChannel customChannel = org.mockito.Mockito.mock(ManagedChannel.class); + GrpcTransportChannel transportChannel = GrpcTransportChannel.create(defaultChannel); + + GrpcCallContext baseContext = + GrpcCallContext.createDefault().withTransportChannel(transportChannel); + GrpcCallContext updatedContext = baseContext.withChannel(customChannel); + + assertEquals(customChannel, updatedContext.getChannel()); + assertNull(updatedContext.getTransportChannel()); + } } diff --git a/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcClientCallsTest.java b/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcClientCallsTest.java index 2aa9279e249f..6877eb1dbe1e 100644 --- a/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcClientCallsTest.java +++ b/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcClientCallsTest.java @@ -125,6 +125,7 @@ void testAffinity() throws IOException { ChannelPool.create( ChannelPoolSettings.staticallySized(2), new FakeChannelFactory(Arrays.asList(channel0, channel1)), + null, null); GrpcCallContext context = defaultCallContext.withChannel(pool); diff --git a/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcLoggingInterceptorTest.java b/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcLoggingInterceptorTest.java index fad4cd468b95..c93db599d575 100644 --- a/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcLoggingInterceptorTest.java +++ b/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/GrpcLoggingInterceptorTest.java @@ -32,7 +32,6 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -83,7 +82,7 @@ void testInterceptor_basic() { void testInterceptor_responseListener() { when(channel.newCall(Mockito.>any(), any(CallOptions.class))) .thenReturn(call); - GrpcLoggingInterceptor interceptor = spy(new GrpcLoggingInterceptor()); + GrpcLoggingInterceptor interceptor = new GrpcLoggingInterceptor(); Channel intercepted = ClientInterceptors.intercept(channel, interceptor); @SuppressWarnings("unchecked") ClientCall.Listener listener = mock(ClientCall.Listener.class); diff --git a/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProviderTest.java b/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProviderTest.java index be0365866615..c2127aede3b8 100644 --- a/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProviderTest.java +++ b/sdk-platform-java/gax-java/gax-grpc/src/test/java/com/google/api/gax/grpc/InstantiatingGrpcChannelProviderTest.java @@ -664,7 +664,9 @@ private void createAndCloseTransportChannel(InstantiatingGrpcChannelProvider pro createAndCloseTransportChannel(provider); assertThat(logHandler.getAllMessages()) .contains( - "DirectPath is misconfigured. The DirectPath XDS option was set, but the attemptDirectPath option was not. Please set both the attemptDirectPath and attemptDirectPathXds options."); + "DirectPath is misconfigured. The DirectPath XDS option was set, but the" + + " attemptDirectPath option was not. Please set both the attemptDirectPath and" + + " attemptDirectPathXds options."); InstantiatingGrpcChannelProvider.LOG.removeHandler(logHandler); } @@ -682,8 +684,10 @@ void testLogDirectPathMisconfig_AttemptDirectPathNotSetAndAttemptDirectPathXdsSe createAndCloseTransportChannel(provider); assertThat(logHandler.getAllMessages()) .contains( - "Env var GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS was found and set to TRUE, but DirectPath was not enabled for this client. If this is intended for " - + "this client, please note that this is a misconfiguration and set the attemptDirectPath option as well."); + "Env var GOOGLE_CLOUD_ENABLE_DIRECT_PATH_XDS was found and set to TRUE, but DirectPath" + + " was not enabled for this client. If this is intended for this client, please" + + " note that this is a misconfiguration and set the attemptDirectPath option as" + + " well."); InstantiatingGrpcChannelProvider.LOG.removeHandler(logHandler); } @@ -711,6 +715,7 @@ void testLogDirectPathMisconfigWrongCredential() throws Exception { InstantiatingGrpcChannelProvider.newBuilder() .setAttemptDirectPathXds() .setAttemptDirectPath(true) + .setEnvProvider(name -> null) .setHeaderProvider( mock(HeaderProvider.class, Mockito.withSettings().withoutAnnotations())) .setExecutor(mock(Executor.class)) @@ -877,12 +882,14 @@ public void canUseDirectPath_directPathEnvVarDisabled() throws IOException { @Test public void canUseDirectPath_directPathEnvVarNotSet_attemptDirectPathIsTrue() { System.setProperty("os.name", "Linux"); + EnvironmentProvider envProvider = name -> null; InstantiatingGrpcChannelProvider.Builder builder = InstantiatingGrpcChannelProvider.newBuilder() .setCertificateBasedAccess(certificateBasedAccess) .setAttemptDirectPath(true) .setCredentials(computeEngineCredentials) - .setEndpoint(DEFAULT_ENDPOINT); + .setEndpoint(DEFAULT_ENDPOINT) + .setEnvProvider(envProvider); InstantiatingGrpcChannelProvider provider = new InstantiatingGrpcChannelProvider(builder, GCE_PRODUCTION_NAME_AFTER_2016); Truth.assertThat(provider.canUseDirectPath()).isTrue(); @@ -891,12 +898,14 @@ public void canUseDirectPath_directPathEnvVarNotSet_attemptDirectPathIsTrue() { @Test public void canUseDirectPath_directPathEnvVarNotSet_attemptDirectPathIsFalse() { System.setProperty("os.name", "Linux"); + EnvironmentProvider envProvider = name -> null; InstantiatingGrpcChannelProvider.Builder builder = InstantiatingGrpcChannelProvider.newBuilder() .setCertificateBasedAccess(certificateBasedAccess) .setAttemptDirectPath(false) .setCredentials(computeEngineCredentials) - .setEndpoint(DEFAULT_ENDPOINT); + .setEndpoint(DEFAULT_ENDPOINT) + .setEnvProvider(envProvider); InstantiatingGrpcChannelProvider provider = new InstantiatingGrpcChannelProvider(builder, GCE_PRODUCTION_NAME_AFTER_2016); Truth.assertThat(provider.canUseDirectPath()).isFalse(); @@ -1201,7 +1210,8 @@ void createS2ASecuredChannelCredentials_bothS2AAddressesNull_returnsNull() { assertThat(provider.createS2ASecuredChannelCredentials()).isNotNull(); assertThat(logHandler.getAllMessages()) .contains( - "Cannot establish an mTLS connection to S2A because autoconfig endpoint did not return a mtls address to reach S2A."); + "Cannot establish an mTLS connection to S2A because autoconfig endpoint did not return" + + " a mtls address to reach S2A."); InstantiatingGrpcChannelProvider.LOG.removeHandler(logHandler); } @@ -1247,7 +1257,8 @@ void createS2ASecuredChannelCredentials_returnsPlaintextToS2AS2AChannelCredentia assertThat(provider.createS2ASecuredChannelCredentials()).isNotNull(); assertThat(logHandler.getAllMessages()) .contains( - "Cannot establish an mTLS connection to S2A because MTLS to MDS credentials do not exist on filesystem, falling back to plaintext connection to S2A"); + "Cannot establish an mTLS connection to S2A because MTLS to MDS credentials do not" + + " exist on filesystem, falling back to plaintext connection to S2A"); InstantiatingGrpcChannelProvider.LOG.removeHandler(logHandler); } diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallContext.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallContext.java index 2679b51860df..5a2c739ac345 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallContext.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonCallContext.java @@ -82,6 +82,7 @@ public final class HttpJsonCallContext implements ApiCallContext { private final @Nullable RetrySettings retrySettings; private final @Nullable ImmutableSet retryableCodes; private final EndpointContext endpointContext; + @Nullable private final TransportChannel transportChannel; /** Returns an empty instance. */ public static HttpJsonCallContext createDefault() { @@ -96,6 +97,7 @@ public static HttpJsonCallContext createDefault() { null, null, null, + null, null); } @@ -111,6 +113,7 @@ public static HttpJsonCallContext of(HttpJsonChannel channel, HttpJsonCallOption null, null, null, + null, null); } @@ -125,7 +128,8 @@ private HttpJsonCallContext( @Nullable ApiTracer tracer, @Nullable RetrySettings defaultRetrySettings, @Nullable Set defaultRetryableCodes, - @Nullable EndpointContext endpointContext) { + @Nullable EndpointContext endpointContext, + @Nullable TransportChannel transportChannel) { this.channel = channel; this.callOptions = callOptions; this.timeout = timeout; @@ -141,6 +145,7 @@ private HttpJsonCallContext( // a valid EndpointContext with user configurations after the client has been initialized. this.endpointContext = endpointContext == null ? EndpointContext.getDefaultInstance() : endpointContext; + this.transportChannel = transportChannel; } /** @@ -220,6 +225,11 @@ public HttpJsonCallContext merge(ApiCallContext inputCallContext) { newRetryableCodes = this.retryableCodes; } + TransportChannel newTransportChannel = httpJsonCallContext.transportChannel; + if (newTransportChannel == null) { + newTransportChannel = this.transportChannel; + } + // The EndpointContext is not updated as there should be no reason for a user // to update this. return new HttpJsonCallContext( @@ -233,7 +243,8 @@ public HttpJsonCallContext merge(ApiCallContext inputCallContext) { newTracer, newRetrySettings, newRetryableCodes, - endpointContext); + endpointContext, + newTransportChannel); } @Override @@ -251,7 +262,24 @@ public HttpJsonCallContext withTransportChannel(TransportChannel inputChannel) { "Expected HttpJsonTransportChannel, got " + inputChannel.getClass().getName()); } HttpJsonTransportChannel transportChannel = (HttpJsonTransportChannel) inputChannel; - return withChannel(transportChannel.getChannel()); + return new HttpJsonCallContext( + transportChannel.getChannel(), + this.callOptions, + this.timeout, + this.streamWaitTimeout, + this.streamIdleTimeout, + this.extraHeaders, + this.options, + this.tracer, + this.retrySettings, + this.retryableCodes, + this.endpointContext, + transportChannel); + } + + @Override + public TransportChannel getTransportChannel() { + return transportChannel; } /** This method is obsolete. Use {@link #withTimeoutDuration(java.time.Duration)} instead. */ @@ -275,7 +303,8 @@ public HttpJsonCallContext withEndpointContext(EndpointContext endpointContext) this.tracer, this.retrySettings, this.retryableCodes, - endpointContext); + endpointContext, + this.transportChannel); } @Override @@ -286,7 +315,7 @@ public HttpJsonCallContext withTimeoutDuration(java.time.Duration timeout) { } // Prevent expanding deadlines - if (timeout != null && this.timeout != null && this.timeout.compareTo(timeout) <= 0) { + if (this.timeout != null && (timeout == null || this.timeout.compareTo(timeout) <= 0)) { return this; } @@ -301,7 +330,8 @@ public HttpJsonCallContext withTimeoutDuration(java.time.Duration timeout) { this.tracer, this.retrySettings, this.retryableCodes, - this.endpointContext); + this.endpointContext, + this.transportChannel); } /** This method is obsolete. Use {@link #getTimeoutDuration()} instead. */ @@ -346,7 +376,8 @@ public HttpJsonCallContext withStreamWaitTimeoutDuration( this.tracer, this.retrySettings, this.retryableCodes, - this.endpointContext); + this.endpointContext, + this.transportChannel); } /** This method is obsolete. Use {@link #getStreamWaitTimeoutDuration()} instead. */ @@ -396,7 +427,8 @@ public HttpJsonCallContext withStreamIdleTimeoutDuration( this.tracer, this.retrySettings, this.retryableCodes, - this.endpointContext); + this.endpointContext, + this.transportChannel); } /** This method is obsolete. Use {@link #getStreamIdleTimeoutDuration()} instead. */ @@ -433,7 +465,8 @@ public ApiCallContext withExtraHeaders(Map> extraHeaders) { this.tracer, this.retrySettings, this.retryableCodes, - this.endpointContext); + this.endpointContext, + this.transportChannel); } @BetaApi("The surface for extra headers is not stable yet and may change in the future.") @@ -457,7 +490,8 @@ public ApiCallContext withOption(Key key, T value) { this.tracer, this.retrySettings, this.retryableCodes, - this.endpointContext); + this.endpointContext, + this.transportChannel); } /** {@inheritDoc} */ @@ -527,7 +561,8 @@ public HttpJsonCallContext withRetrySettings(RetrySettings retrySettings) { this.tracer, retrySettings, this.retryableCodes, - this.endpointContext); + this.endpointContext, + this.transportChannel); } @Override @@ -548,7 +583,8 @@ public HttpJsonCallContext withRetryableCodes(Set retryableCode this.tracer, this.retrySettings, retryableCodes, - this.endpointContext); + this.endpointContext, + this.transportChannel); } public HttpJsonCallContext withChannel(@Nullable HttpJsonChannel newChannel) { @@ -563,7 +599,8 @@ public HttpJsonCallContext withChannel(@Nullable HttpJsonChannel newChannel) { this.tracer, this.retrySettings, this.retryableCodes, - this.endpointContext); + this.endpointContext, + this.transportChannel); } public HttpJsonCallContext withCallOptions(HttpJsonCallOptions newCallOptions) { @@ -578,7 +615,8 @@ public HttpJsonCallContext withCallOptions(HttpJsonCallOptions newCallOptions) { this.tracer, this.retrySettings, this.retryableCodes, - this.endpointContext); + this.endpointContext, + this.transportChannel); } @Deprecated @@ -614,7 +652,8 @@ public HttpJsonCallContext withTracer(@Nonnull ApiTracer newTracer) { newTracer, this.retrySettings, this.retryableCodes, - this.endpointContext); + this.endpointContext, + this.transportChannel); } @Override @@ -634,7 +673,8 @@ public boolean equals(@Nullable Object o) { && Objects.equals(this.tracer, that.tracer) && Objects.equals(this.retrySettings, that.retrySettings) && Objects.equals(this.retryableCodes, that.retryableCodes) - && Objects.equals(this.endpointContext, that.endpointContext); + && Objects.equals(this.endpointContext, that.endpointContext) + && Objects.equals(this.transportChannel, that.transportChannel); } @Override @@ -648,6 +688,7 @@ public int hashCode() { tracer, retrySettings, retryableCodes, - endpointContext); + endpointContext, + transportChannel); } } diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonTransportChannel.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonTransportChannel.java index 813622b6a97e..a8333a589a4a 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonTransportChannel.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/HttpJsonTransportChannel.java @@ -64,6 +64,16 @@ public HttpJsonChannel getChannel() { return getManagedChannel(); } + @Override + public void refresh() { + getManagedChannel().refresh(); + } + + @Override + public boolean shouldRefresh() { + return getManagedChannel().shouldRefresh(); + } + @Override public void shutdown() { getManagedChannel().shutdown(); diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/InstantiatingHttpJsonChannelProvider.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/InstantiatingHttpJsonChannelProvider.java index 92ce4efe36aa..90ce27c2879d 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/InstantiatingHttpJsonChannelProvider.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/InstantiatingHttpJsonChannelProvider.java @@ -31,7 +31,6 @@ import com.google.api.client.http.HttpTransport; import com.google.api.client.http.javanet.NetHttpTransport; -import com.google.api.client.util.SslUtils; import com.google.api.core.InternalExtensionOnly; import com.google.api.gax.core.ExecutorProvider; import com.google.api.gax.rpc.FixedHeaderProvider; @@ -46,13 +45,11 @@ import java.io.IOException; import java.security.GeneralSecurityException; import java.security.KeyStore; -import java.security.Provider; import java.util.Map; import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; import java.util.logging.Level; import java.util.logging.Logger; -import javax.net.ssl.SSLContext; import org.jspecify.annotations.NullMarked; import org.jspecify.annotations.Nullable; @@ -195,58 +192,44 @@ public TransportChannelProvider withCredentials(Credentials credentials) { "InstantiatingHttpJsonChannelProvider doesn't need credentials"); } - HttpTransport createHttpTransport() throws IOException, GeneralSecurityException { - NetHttpTransport.Builder builder = new NetHttpTransport.Builder(); - configureMtls(builder); - HttpJsonConscryptUtils.configureConscryptSecurityProvider(builder); - return builder.build(); - } - - private NetHttpTransport.Builder configureMtls(NetHttpTransport.Builder builder) - throws IOException, GeneralSecurityException { - if (mtlsProvider == null || !certificateBasedAccess.useMtlsClientCertificate()) { - return builder; + @Nullable HttpTransport createHttpTransport() throws IOException, GeneralSecurityException { + if (mtlsProvider == null) { + return null; } - KeyStore mtlsKeyStore = mtlsProvider.getKeyStore(); - if (mtlsKeyStore == null) { - return builder; - } - builder.trustCertificates(null, mtlsKeyStore, ""); - Provider conscryptProvider = HttpJsonConscryptUtils.getConscryptProvider(); - if (conscryptProvider == null) { - // Fall back to standard JDK JSSE if Conscrypt provider is unavailable - return builder; + if (certificateBasedAccess.useMtlsClientCertificate()) { + KeyStore mtlsKeyStore = mtlsProvider.getKeyStore(); + if (mtlsKeyStore != null) { + NetHttpTransport.Builder builder = new NetHttpTransport.Builder(); + builder.trustCertificates(null, mtlsKeyStore, ""); + HttpJsonConscryptUtils.configureConscryptSecurityProvider(builder); + return builder.build(); + } } - // Explicitly initialize SSLContext with the Conscrypt provider so that the client certificate - // key managers - // and trust manager factory (TMF) are bound to Conscrypt's TLS implementation (supporting PQC - // key exchange). - SSLContext sslContext = SSLContext.getInstance("TLS", conscryptProvider); - SslUtils.initSslContext( - sslContext, - null, - SslUtils.getPkixTrustManagerFactory(), - mtlsKeyStore, - "", - SslUtils.getDefaultKeyManagerFactory()); - builder.setSslSocketFactory(sslContext.getSocketFactory()); - return builder; + return null; } private HttpJsonTransportChannel createChannel() throws IOException, GeneralSecurityException { - HttpTransport httpTransportToUse = httpTransport; - if (httpTransportToUse == null) { - httpTransportToUse = createHttpTransport(); - } + java.util.function.Supplier channelFactory = + () -> { + try { + HttpTransport httpTransportToUse = httpTransport; + if (httpTransportToUse == null) { + httpTransportToUse = createHttpTransport(); + } + return ManagedHttpJsonChannel.newBuilder() + .setEndpoint(endpoint) + .setExecutor(executor) + .setHttpTransport(httpTransportToUse) + .setManageHttpTransport(httpTransport == null) + .build(); + } catch (Exception e) { + throw new java.lang.RuntimeException( + "Failed to create fresh ManagedHttpJsonChannel", e); + } + }; - // Pass the executor to the ManagedChannel. If no executor was provided (or null), - // the channel will use a default executor for the calls. ManagedHttpJsonChannel channel = - ManagedHttpJsonChannel.newBuilder() - .setEndpoint(endpoint) - .setExecutor(executor) - .setHttpTransport(httpTransportToUse) - .build(); + new RefreshingHttpJsonChannel(channelFactory, certificateBasedAccess.getWorkloadCertPath()); HttpJsonClientInterceptor headerInterceptor = new HttpJsonHeaderInterceptor(headerProvider.getHeaders()); diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ManagedHttpJsonChannel.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ManagedHttpJsonChannel.java index 87767bee5c7f..f83f09bac486 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ManagedHttpJsonChannel.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ManagedHttpJsonChannel.java @@ -52,11 +52,12 @@ public class ManagedHttpJsonChannel implements HttpJsonChannel, BackgroundResour private final boolean usingDefaultExecutor; private final String endpoint; private final HttpTransport httpTransport; + private final boolean usingDefaultTransport; private final ScheduledExecutorService deadlineScheduledExecutorService; private boolean isTransportShutdown; protected ManagedHttpJsonChannel() { - this(null, true, null, null); + this(null, true, null, null, true); } String getEndpoint() { @@ -72,7 +73,8 @@ private ManagedHttpJsonChannel( @Nullable Executor executor, boolean usingDefaultExecutor, @Nullable String endpoint, - @Nullable HttpTransport httpTransport) { + @Nullable HttpTransport httpTransport, + boolean usingDefaultTransport) { this.executor = executor; this.usingDefaultExecutor = usingDefaultExecutor; this.endpoint = endpoint; @@ -82,6 +84,7 @@ private ManagedHttpJsonChannel( new NetHttpTransport.Builder()) .build() : httpTransport; + this.usingDefaultTransport = usingDefaultTransport || httpTransport == null; this.deadlineScheduledExecutorService = Executors.newSingleThreadScheduledExecutor(); } @@ -98,6 +101,12 @@ public HttpJsonClientCall newCall( deadlineScheduledExecutorService); } + public void refresh() {} + + public boolean shouldRefresh() { + return false; + } + @VisibleForTesting Executor getExecutor() { return executor; @@ -116,7 +125,9 @@ public synchronized void shutdown() { ((ExecutorService) executor).shutdown(); } deadlineScheduledExecutorService.shutdown(); - httpTransport.shutdown(); + if (usingDefaultTransport) { + httpTransport.shutdown(); + } isTransportShutdown = true; } catch (IOException e) { // TODO: Log this scenario once we implemented the Cloud SDK logging. @@ -158,7 +169,9 @@ public void shutdownNow() { ((ExecutorService) executor).shutdownNow(); } deadlineScheduledExecutorService.shutdownNow(); - httpTransport.shutdown(); + if (usingDefaultTransport) { + httpTransport.shutdown(); + } isTransportShutdown = true; } catch (IOException e) { // TODO: Log this scenario once we implemented the Cloud SDK logging. @@ -205,9 +218,11 @@ public static class Builder { private String endpoint; private HttpTransport httpTransport; private boolean usingDefaultExecutor; + private boolean usingDefaultTransport; private Builder() { this.usingDefaultExecutor = false; + this.usingDefaultTransport = false; } public Builder setExecutor(Executor executor) { @@ -225,6 +240,11 @@ public Builder setHttpTransport(HttpTransport httpTransport) { return this; } + Builder setManageHttpTransport(boolean manageHttpTransport) { + this.usingDefaultTransport = manageHttpTransport; + return this; + } + public ManagedHttpJsonChannel build() { Preconditions.checkNotNull(endpoint); @@ -237,14 +257,8 @@ public ManagedHttpJsonChannel build() { usingDefaultExecutor = true; } - if (httpTransport == null) { - httpTransport = - HttpJsonConscryptUtils.configureConscryptSecurityProvider( - new NetHttpTransport.Builder()) - .build(); - } - - return new ManagedHttpJsonChannel(executor, usingDefaultExecutor, endpoint, httpTransport); + return new ManagedHttpJsonChannel( + executor, usingDefaultExecutor, endpoint, httpTransport, usingDefaultTransport); } } } diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ManagedHttpJsonInterceptorChannel.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ManagedHttpJsonInterceptorChannel.java index eaaa8c3a7c56..e552608b6529 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ManagedHttpJsonInterceptorChannel.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ManagedHttpJsonInterceptorChannel.java @@ -29,7 +29,9 @@ */ package com.google.api.gax.httpjson; +import com.google.api.client.http.HttpTransport; import com.google.common.annotations.VisibleForTesting; +import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; import org.jspecify.annotations.NullMarked; @@ -51,12 +53,39 @@ ManagedHttpJsonChannel getChannel() { return channel; } + @Override + String getEndpoint() { + return channel.getEndpoint(); + } + + @Override + @VisibleForTesting + HttpTransport getHttpTransport() { + return channel.getHttpTransport(); + } + + @Override + @VisibleForTesting + Executor getExecutor() { + return channel.getExecutor(); + } + @Override public HttpJsonClientCall newCall( ApiMethodDescriptor methodDescriptor, HttpJsonCallOptions callOptions) { return interceptor.interceptCall(methodDescriptor, callOptions, channel); } + @Override + public void refresh() { + channel.refresh(); + } + + @Override + public boolean shouldRefresh() { + return channel.shouldRefresh(); + } + @Override public synchronized void shutdown() { channel.shutdown(); diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/RefreshingHttpJsonChannel.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/RefreshingHttpJsonChannel.java new file mode 100644 index 000000000000..c0855b16871b --- /dev/null +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/RefreshingHttpJsonChannel.java @@ -0,0 +1,395 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.httpjson; + +import com.google.api.client.http.HttpTransport; +import com.google.api.core.InternalApi; +import com.google.api.gax.httpjson.ForwardingHttpJsonClientCall.SimpleForwardingHttpJsonClientCall; +import com.google.api.gax.httpjson.ForwardingHttpJsonClientCallListener.SimpleForwardingHttpJsonClientCallListener; +import com.google.api.gax.rpc.mtls.WorkloadCertificateUtils; +import com.google.common.annotations.VisibleForTesting; +import java.util.concurrent.CancellationException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.jspecify.annotations.Nullable; + +/** + * An implementation of {@link ManagedHttpJsonChannel} that supports dynamic mTLS certificate + * rotation by thread-safely hot-swapping the underlying active HTTP/JSON channel while gracefully + * retiring older connections after all active in-flight requests complete. + */ +@InternalApi +public class RefreshingHttpJsonChannel extends ManagedHttpJsonChannel { + + private static final Logger LOG = Logger.getLogger(RefreshingHttpJsonChannel.class.getName()); + + private static class DiskCheckResult { + final String fingerprint; + final long timestampNanos; + + DiskCheckResult(String fingerprint, long timestampNanos) { + this.fingerprint = fingerprint; + this.timestampNanos = timestampNanos; + } + } + + private volatile DiskCheckResult lastDiskCheck = null; + private final java.util.concurrent.locks.ReentrantLock diskCheckLock = + new java.util.concurrent.locks.ReentrantLock(); + private final Supplier channelFactory; + private final String workloadCertPath; + private final AtomicReference activeEntry; + // Keep track of all entries to properly await their termination + private final java.util.concurrent.ConcurrentLinkedQueue allEntries = + new java.util.concurrent.ConcurrentLinkedQueue<>(); + private final Object refreshLock = new Object(); + private volatile String activeCertFingerprint = ""; + + public RefreshingHttpJsonChannel( + Supplier channelFactory, String workloadCertPath) { + this.channelFactory = channelFactory; + this.workloadCertPath = workloadCertPath; + ChannelEntry initial = new ChannelEntry(channelFactory.get()); + this.activeEntry = new AtomicReference<>(initial); + this.allEntries.add(initial); + if (workloadCertPath != null) { + this.activeCertFingerprint = getCertificateFingerprint(workloadCertPath); + } + } + + private String getOrUpdateDiskFingerprint(String certPath) { + long now = System.nanoTime(); + DiskCheckResult cached = lastDiskCheck; + if (cached != null + && (now - cached.timestampNanos < java.util.concurrent.TimeUnit.SECONDS.toNanos(1))) { + return cached.fingerprint; + } + + diskCheckLock.lock(); + try { + cached = lastDiskCheck; + if (cached != null + && (now - cached.timestampNanos < java.util.concurrent.TimeUnit.SECONDS.toNanos(1))) { + return cached.fingerprint; + } + String fingerprint = getCertificateFingerprint(certPath); + lastDiskCheck = new DiskCheckResult(fingerprint, System.nanoTime()); + return fingerprint; + } finally { + diskCheckLock.unlock(); + } + } + + // Visible for testing + protected String getWorkloadCertPath() { + return workloadCertPath; + } + + // Visible for testing + protected String getCertificateFingerprint(String certPath) { + return WorkloadCertificateUtils.getCertificateFingerprint(certPath); + } + + @Override + public boolean shouldRefresh() { + String certPath = getWorkloadCertPath(); + if (certPath == null) { + return false; + } + String currentDiskFingerprint = getOrUpdateDiskFingerprint(certPath); + if (currentDiskFingerprint.isEmpty()) { + return false; + } + return !currentDiskFingerprint.equalsIgnoreCase(activeCertFingerprint); + } + + @Override + public void refresh() { + synchronized (refreshLock) { + if (isShutdown()) { + return; + } + String certPath = getWorkloadCertPath(); + if (certPath == null) { + return; + } + String currentDiskFingerprint = getOrUpdateDiskFingerprint(certPath); + if (currentDiskFingerprint.isEmpty()) { + return; + } + + // Double-check inside refreshLock + if (currentDiskFingerprint.equalsIgnoreCase(this.activeCertFingerprint)) { + LOG.fine( + "HTTP/JSON channel was already refreshed by a concurrent thread, skipping duplicate" + + " refresh"); + return; + } + + LOG.info("mTLS certificate rotation detected. Triggering HTTP/JSON channel pool refresh."); + + // Prune terminated entries to prevent memory leak + allEntries.removeIf(entry -> entry.channel.isTerminated()); + + ChannelEntry newEntry = new ChannelEntry(channelFactory.get()); + allEntries.add(newEntry); + ChannelEntry oldEntry = activeEntry.getAndSet(newEntry); + this.activeCertFingerprint = currentDiskFingerprint; + + if (oldEntry != null) { + oldEntry.requestShutdown(); + } + } + } + + private ChannelEntry getRetainedEntry() { + while (true) { + ChannelEntry entry = activeEntry.get(); + if (entry.retain()) { + return entry; + } + if (entry == activeEntry.get()) { + throw new IllegalStateException("Channel has been shut down"); + } + } + } + + @Override + public HttpJsonClientCall newCall( + ApiMethodDescriptor methodDescriptor, HttpJsonCallOptions callOptions) { + ChannelEntry entry = getRetainedEntry(); + try { + HttpJsonClientCall delegateCall = + entry.channel.newCall(methodDescriptor, callOptions); + return new ReleasingHttpJsonClientCall<>(delegateCall, entry); + } catch (Exception e) { + entry.release(); + throw e; + } + } + + @Override + java.util.concurrent.Executor getExecutor() { + return activeEntry.get().channel.getExecutor(); + } + + @VisibleForTesting + ManagedHttpJsonChannel getActiveChannel() { + return activeEntry.get().channel; + } + + private volatile boolean isShuttingDown = false; + + @Override + public void shutdown() { + synchronized (refreshLock) { + isShuttingDown = true; + for (ChannelEntry entry : allEntries) { + entry.requestShutdown(); + } + } + } + + @Override + public boolean isShutdown() { + return isShuttingDown; + } + + @Override + public boolean isTerminated() { + for (ChannelEntry entry : allEntries) { + if (!entry.channel.isTerminated()) { + return false; + } + } + return true; + } + + @Override + public void shutdownNow() { + synchronized (refreshLock) { + isShuttingDown = true; + for (ChannelEntry entry : allEntries) { + entry.shutdownRequested.set(true); + entry.shutdownInitiated.set(true); + entry.channel.shutdownNow(); + } + } + } + + @VisibleForTesting + void invalidateDiskFingerprintCache() { + this.lastDiskCheck = null; + } + + @Override + public boolean awaitTermination(long duration, TimeUnit unit) throws InterruptedException { + long endNanos = System.nanoTime() + unit.toNanos(duration); + for (ChannelEntry entry : allEntries) { + if (entry.channel.isTerminated()) { + continue; + } + long remainingNanos = endNanos - System.nanoTime(); + if (remainingNanos <= 0) { + return false; + } + if (!entry.channel.awaitTermination(remainingNanos, TimeUnit.NANOSECONDS)) { + return false; + } + } + return true; + } + + @Override + public void close() { + shutdown(); + } + + @Override + String getEndpoint() { + return activeEntry.get().channel.getEndpoint(); + } + + @Override + @VisibleForTesting + HttpTransport getHttpTransport() { + return activeEntry.get().channel.getHttpTransport(); + } + + /** Internal container to manage request reference-counting and graceful shutdown. */ + private static class ChannelEntry { + private final ManagedHttpJsonChannel channel; + private final AtomicInteger outstandingCalls = new AtomicInteger(0); + private final AtomicBoolean shutdownRequested = new AtomicBoolean(false); + private final AtomicBoolean shutdownInitiated = new AtomicBoolean(false); + + ChannelEntry(ManagedHttpJsonChannel channel) { + this.channel = channel; + } + + boolean retain() { + outstandingCalls.incrementAndGet(); + if (shutdownRequested.get()) { + release(); + return false; + } + return true; + } + + void release() { + int count = outstandingCalls.decrementAndGet(); + if (shutdownRequested.get() && count == 0) { + shutdown(); + } + } + + void requestShutdown() { + shutdownRequested.set(true); + if (outstandingCalls.get() == 0) { + shutdown(); + } + } + + private void shutdown() { + if (shutdownInitiated.compareAndSet(false, true)) { + try { + channel.shutdown(); + } catch (Exception e) { + LOG.log(Level.WARNING, "Error shutting down retired HTTP/JSON channel", e); + } + } + } + } + + /** A client call decorator that decrements the entry counter upon call completion. */ + private static class ReleasingHttpJsonClientCall + extends SimpleForwardingHttpJsonClientCall { + + private @Nullable CancellationException cancellationException; + private final ChannelEntry entry; + private final AtomicBoolean wasClosed = new AtomicBoolean(false); + private final AtomicBoolean wasReleased = new AtomicBoolean(false); + private final AtomicBoolean wasStarted = new AtomicBoolean(false); + + ReleasingHttpJsonClientCall(HttpJsonClientCall delegate, ChannelEntry entry) { + super(delegate); + this.entry = entry; + } + + @Override + public void start(Listener responseListener, HttpJsonMetadata requestHeaders) { + wasStarted.set(true); + if (cancellationException != null) { + if (wasReleased.compareAndSet(false, true)) { + entry.release(); + } + throw new IllegalStateException("Call is already cancelled", cancellationException); + } + try { + super.start( + new SimpleForwardingHttpJsonClientCallListener(responseListener) { + @Override + public void onClose(int statusCode, HttpJsonMetadata trailers) { + if (!wasClosed.compareAndSet(false, true)) { + return; + } + try { + super.onClose(statusCode, trailers); + } finally { + if (wasReleased.compareAndSet(false, true)) { + entry.release(); + } + } + } + }, + requestHeaders); + } catch (Exception e) { + if (wasReleased.compareAndSet(false, true)) { + entry.release(); + } + throw e; + } + } + + @Override + public void cancel(@Nullable String message, @Nullable Throwable cause) { + this.cancellationException = new CancellationException(message); + super.cancel(message, cause); + if (!wasStarted.get() && wasReleased.compareAndSet(false, true)) { + entry.release(); + } + } + } +} diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/InstantiatingHttpJsonChannelProviderTest.java b/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/InstantiatingHttpJsonChannelProviderTest.java index 8c95c1d2e1c4..4482f2367a4a 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/InstantiatingHttpJsonChannelProviderTest.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/InstantiatingHttpJsonChannelProviderTest.java @@ -31,8 +31,9 @@ import static com.google.common.truth.Truth.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; -import com.google.api.client.http.javanet.NetHttpTransport; +import com.google.api.gax.rpc.HeaderProvider; import com.google.api.gax.rpc.TransportChannelProvider; import com.google.api.gax.rpc.mtls.AbstractMtlsTransportChannelTest; import com.google.api.gax.rpc.mtls.CertificateBasedAccess; @@ -46,6 +47,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.Mockito; class InstantiatingHttpJsonChannelProviderTest extends AbstractMtlsTransportChannelTest { @@ -55,9 +57,10 @@ class InstantiatingHttpJsonChannelProviderTest extends AbstractMtlsTransportChan @BeforeEach public void setup() throws IOException { - certificateBasedAccess = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "never" : "false"); + certificateBasedAccess = org.mockito.Mockito.mock(CertificateBasedAccess.class); + org.mockito.Mockito.when(certificateBasedAccess.getMtlsEndpointUsagePolicy()) + .thenReturn(CertificateBasedAccess.MtlsEndpointUsagePolicy.NEVER); + org.mockito.Mockito.when(certificateBasedAccess.useMtlsClientCertificate()).thenReturn(false); } @Test @@ -179,6 +182,39 @@ void managedChannelUsesCustomExecutor() throws IOException { instantiatingHttpJsonChannelProvider.getTransportChannel().shutdownNow(); } + @Test + void managedChannelDoesNotShutdownCustomHttpTransport() throws IOException { + com.google.api.client.http.HttpTransport mockHttpTransport = + org.mockito.Mockito.mock(com.google.api.client.http.HttpTransport.class); + + InstantiatingHttpJsonChannelProvider provider = + InstantiatingHttpJsonChannelProvider.newBuilder() + .setEndpoint(DEFAULT_ENDPOINT) + .setHttpTransport(mockHttpTransport) + .setCertificateBasedAccess(certificateBasedAccess) + .build(); + provider = (InstantiatingHttpJsonChannelProvider) provider.withHeaders(DEFAULT_HEADER_MAP); + + HttpJsonTransportChannel httpJsonTransportChannel = provider.getTransportChannel(); + + // Verify custom transport is injected + ManagedHttpJsonInterceptorChannel interceptorChannel = + (ManagedHttpJsonInterceptorChannel) httpJsonTransportChannel.getManagedChannel(); + ManagedHttpJsonInterceptorChannel managedHttpJsonChannel = + (ManagedHttpJsonInterceptorChannel) interceptorChannel.getChannel(); + RefreshingHttpJsonChannel refreshingHttpJsonChannel = + (RefreshingHttpJsonChannel) managedHttpJsonChannel.getChannel(); + ManagedHttpJsonChannel channel = refreshingHttpJsonChannel.getActiveChannel(); + + assertThat(channel.getHttpTransport()).isEqualTo(mockHttpTransport); + + // Perform a shutdown + provider.getTransportChannel().shutdownNow(); + + // Verify that shutdown() was NOT called on the custom HttpTransport + org.mockito.Mockito.verify(mockHttpTransport, org.mockito.Mockito.never()).shutdown(); + } + @Override protected Object getMtlsObjectFromTransportChannel( MtlsProvider provider, CertificateBasedAccess certificateBasedAccess) @@ -188,30 +224,10 @@ protected Object getMtlsObjectFromTransportChannel( .setEndpoint("localhost:8080") .setMtlsProvider(provider) .setCertificateBasedAccess(certificateBasedAccess) - .setHeaderProvider(Collections::emptyMap) - .setExecutor(Runnable::run) - .build(); - NetHttpTransport transport = (NetHttpTransport) channelProvider.createHttpTransport(); - return (transport != null && transport.isMtls()) ? transport : null; - } - - @Test - void testCreateHttpTransport_returnsValidTransport() throws Exception { - InstantiatingHttpJsonChannelProvider channelProvider = - InstantiatingHttpJsonChannelProvider.newBuilder() - .setEndpoint("localhost:8080") - .setHeaderProvider(Collections::emptyMap) - .setExecutor(Runnable::run) + .setHeaderProvider( + mock(HeaderProvider.class, Mockito.withSettings().withoutAnnotations())) + .setExecutor(mock(Executor.class)) .build(); - NetHttpTransport transport = (NetHttpTransport) channelProvider.createHttpTransport(); - assertThat(transport).isNotNull(); - } - - @Test - void testConfigureConscryptSecurityProvider_returnsConfiguredBuilder() { - NetHttpTransport.Builder builder = new NetHttpTransport.Builder(); - NetHttpTransport.Builder result = - HttpJsonConscryptUtils.configureConscryptSecurityProvider(builder); - assertThat(result).isSameInstanceAs(builder); + return channelProvider.createHttpTransport(); } } diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/RefreshingHttpJsonChannelTest.java b/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/RefreshingHttpJsonChannelTest.java new file mode 100644 index 000000000000..ea147deb74bc --- /dev/null +++ b/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/RefreshingHttpJsonChannelTest.java @@ -0,0 +1,392 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.httpjson; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; +import javax.annotation.Nullable; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class RefreshingHttpJsonChannelTest { + private static class FakeHttpJsonClientCall + extends HttpJsonClientCall { + private Listener listener; + + @Override + public void start(Listener responseListener, HttpJsonMetadata requestHeaders) { + this.listener = responseListener; + } + + @Override + public void request(int numMessages) {} + + @Override + public void cancel(@Nullable String message, @Nullable Throwable cause) {} + + @Override + public void sendMessage(RequestT message) {} + + @Override + public void halfClose() {} + } + + private static class FakeManagedHttpJsonChannel extends ManagedHttpJsonChannel { + private volatile boolean isShutdown = false; + private volatile boolean isTerminated = false; + private HttpJsonClientCall nextCall = null; + + @Override + String getEndpoint() { + return "https://fake.endpoint:443"; + } + + @Override + public void shutdown() { + isShutdown = true; + } + + @Override + public void shutdownNow() { + isShutdown = true; + isTerminated = true; + } + + @Override + public boolean isShutdown() { + return isShutdown; + } + + @Override + public boolean isTerminated() { + return isTerminated; + } + + @Override + public boolean awaitTermination(long duration, TimeUnit unit) { + return isTerminated; + } + + @Override + @SuppressWarnings("unchecked") + public HttpJsonClientCall newCall( + ApiMethodDescriptor methodDescriptor, + HttpJsonCallOptions callOptions) { + if (nextCall != null) { + return (HttpJsonClientCall) nextCall; + } + return new FakeHttpJsonClientCall<>(); + } + } + + private AtomicInteger channelFactoryCount; + private FakeManagedHttpJsonChannel lastCreatedChannel; + private String testCertPath = "/fake/path"; + private String testFingerprint = "fingerprint1"; + private boolean shouldThrowOnFactory = false; + private List createdChannels; + + private Supplier channelFactory = + () -> { + if (shouldThrowOnFactory) { + throw new RuntimeException("Simulated factory failure"); + } + channelFactoryCount.incrementAndGet(); + lastCreatedChannel = new FakeManagedHttpJsonChannel(); + return lastCreatedChannel; + }; + + @BeforeEach + void setUp() { + channelFactoryCount = new AtomicInteger(0); + testCertPath = "/fake/path"; + testFingerprint = "fingerprint1"; + shouldThrowOnFactory = false; + createdChannels = new ArrayList<>(); + } + + @AfterEach + void tearDown() { + for (RefreshingHttpJsonChannel channel : createdChannels) { + channel.shutdownNow(); + } + } + + private RefreshingHttpJsonChannel createTestChannel() { + RefreshingHttpJsonChannel ch = + new RefreshingHttpJsonChannel(channelFactory, "fake/cert/path.json") { + @Override + protected String getWorkloadCertPath() { + return testCertPath; + } + + @Override + protected String getCertificateFingerprint(String certPath) { + return testFingerprint; + } + }; + createdChannels.add(ch); + return ch; + } + + @Test + void testShouldRefreshNullCertPath() { + testCertPath = null; + RefreshingHttpJsonChannel channel = createTestChannel(); + assertFalse(channel.shouldRefresh()); + } + + @Test + void testShouldRefreshFalseWhenUnchanged() throws InterruptedException { + RefreshingHttpJsonChannel channel = createTestChannel(); + + channel.invalidateDiskFingerprintCache(); // Invalidate 1-second cache + assertFalse(channel.shouldRefresh()); + } + + @Test + void testShouldRefreshTrueWhenChanged() throws InterruptedException { + RefreshingHttpJsonChannel channel = createTestChannel(); + + channel.invalidateDiskFingerprintCache(); // Invalidate 1-second cache + + // Simulate disk fingerprint changing + testFingerprint = "fingerprint2"; + + assertTrue(channel.shouldRefresh()); + } + + @Test + void testRefreshSwapsChannel() throws InterruptedException { + RefreshingHttpJsonChannel channel = createTestChannel(); + FakeManagedHttpJsonChannel firstChannel = lastCreatedChannel; + assertEquals(1, channelFactoryCount.get()); + + channel.invalidateDiskFingerprintCache(); // Invalidate 1-second cache + + // Change fingerprint + testFingerprint = "fingerprint2"; + + // Act + channel.refresh(); + + // Verify a new channel was created and the old one retired + assertEquals(2, channelFactoryCount.get()); + FakeManagedHttpJsonChannel secondChannel = lastCreatedChannel; + + // The old channel should receive a shutdown request immediately since there are no active calls + assertTrue(firstChannel.isShutdown()); + assertFalse(secondChannel.isShutdown()); + } + + @Test + void testRefreshKeepsInFlightChannelsAlive() throws InterruptedException { + RefreshingHttpJsonChannel channel = createTestChannel(); + FakeManagedHttpJsonChannel firstChannel = lastCreatedChannel; + + // Simulate an in-flight API call + FakeHttpJsonClientCall fakeCall = new FakeHttpJsonClientCall<>(); + firstChannel.nextCall = fakeCall; + + HttpJsonClientCall activeCall = channel.newCall(null, null); + + channel.invalidateDiskFingerprintCache(); // Invalidate 1-second cache + + // Change fingerprint & refresh + testFingerprint = "fingerprint2"; + + channel.refresh(); + + // Verify a new channel was created + assertEquals(2, channelFactoryCount.get()); + + // IMPORTANT: The first channel should NOT be shut down yet because of the active call! + assertFalse(firstChannel.isShutdown()); + + // Now start the call + activeCall.start(new HttpJsonClientCall.Listener() {}, null); + + assertNotNull(fakeCall.listener); + + // Fire onClose + fakeCall.listener.onClose(0, null); + + // FIRST CHANNEL SHOULD BE SHUT DOWN NOW! + assertTrue(firstChannel.isShutdown()); + } + + @Test + void testCancelBeforeStartReleasesChannelEntry() { + RefreshingHttpJsonChannel channel = createTestChannel(); + FakeManagedHttpJsonChannel firstChannel = lastCreatedChannel; + + HttpJsonClientCall activeCall = channel.newCall(null, null); + + channel.invalidateDiskFingerprintCache(); + testFingerprint = "fingerprint2"; + channel.refresh(); + + // Because activeCall was created, the old channel should NOT be shut down yet + assertFalse(firstChannel.isShutdown()); + + // Cancel before start() is called + activeCall.cancel("Cancelled early", null); + + // Because cancel() safely released the entry, the old channel should now be shut down! + assertTrue(firstChannel.isShutdown()); + } + + @Test + void testRefreshDoesNotSpawnChannelWhenShutdown() throws InterruptedException { + RefreshingHttpJsonChannel channel = createTestChannel(); + FakeManagedHttpJsonChannel firstChannel = lastCreatedChannel; + assertEquals(1, channelFactoryCount.get()); + + // Simulate that the channel pool is shut down. + channel.shutdown(); + firstChannel.shutdown(); + + channel.invalidateDiskFingerprintCache(); // Invalidate 1-second cache + + // Change fingerprint + testFingerprint = "fingerprint2"; + + // Act + channel.refresh(); + + // Verify no new channel was spawned + assertEquals(1, channelFactoryCount.get()); + } + + @Test + void testRefreshFactoryExceptionDoesNotWedgeFingerprint() throws InterruptedException { + RefreshingHttpJsonChannel channel = createTestChannel(); + assertEquals(1, channelFactoryCount.get()); + + shouldThrowOnFactory = true; + channel.invalidateDiskFingerprintCache(); // Invalidate 1-second cache + testFingerprint = "fingerprint2"; + + assertThrows(RuntimeException.class, channel::refresh); + + // Because factory threw, activeCertFingerprint should NOT be updated to fingerprint2 + // Therefore shouldRefresh() should still return true + assertTrue(channel.shouldRefresh()); + + shouldThrowOnFactory = false; + channel.refresh(); + assertEquals(2, channelFactoryCount.get()); + assertFalse(channel.shouldRefresh()); + } + + @Test + void testShutdownNowSetsIsShutdown() { + RefreshingHttpJsonChannel channel = createTestChannel(); + assertFalse(channel.isShutdown()); + + channel.shutdownNow(); + + assertTrue(channel.isShutdown()); + } + + @Test + void testAwaitTerminationZeroTimeoutOnTerminatedChannelReturnsTrue() throws InterruptedException { + RefreshingHttpJsonChannel channel = createTestChannel(); + FakeManagedHttpJsonChannel firstChannel = lastCreatedChannel; + firstChannel.isTerminated = true; + + channel.shutdown(); + assertTrue(channel.awaitTermination(0, TimeUnit.MILLISECONDS)); + } + + @Test + void testChannelDelegationMethods() { + RefreshingHttpJsonChannel channel = createTestChannel(); + FakeManagedHttpJsonChannel firstChannel = lastCreatedChannel; + + assertEquals(firstChannel.getEndpoint(), channel.getEndpoint()); + assertEquals(firstChannel.getHttpTransport(), channel.getHttpTransport()); + assertEquals(firstChannel.getExecutor(), channel.getExecutor()); + } + + @Test + void testNewCallAfterShutdownNowThrowsIllegalStateException() { + RefreshingHttpJsonChannel channel = createTestChannel(); + channel.shutdownNow(); + + assertThrows( + IllegalStateException.class, + () -> channel.newCall(null, null), + "Channel has been shut down"); + } + + @Test + void testConcurrentNewCallDuringRefresh() throws InterruptedException { + RefreshingHttpJsonChannel channel = createTestChannel(); + int threadCount = 10; + java.util.concurrent.ExecutorService executorService = + java.util.concurrent.Executors.newFixedThreadPool(threadCount); + java.util.concurrent.CountDownLatch latch = + new java.util.concurrent.CountDownLatch(threadCount); + java.util.concurrent.atomic.AtomicInteger successCount = + new java.util.concurrent.atomic.AtomicInteger(0); + + for (int i = 0; i < threadCount; i++) { + executorService.submit( + () -> { + try { + channel.newCall(null, null); + successCount.incrementAndGet(); + } finally { + latch.countDown(); + } + }); + } + + channel.invalidateDiskFingerprintCache(); + testFingerprint = "fingerprint2"; + channel.refresh(); + + latch.await(5, TimeUnit.SECONDS); + executorService.shutdown(); + + assertEquals(threadCount, successCount.get()); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/retrying/RetrySettings.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/retrying/RetrySettings.java index d69fd310c2c0..e6729459b775 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/retrying/RetrySettings.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/retrying/RetrySettings.java @@ -189,7 +189,7 @@ public final org.threeten.bp.Duration getInitialRpcTimeout() { * connection has been terminated). * *

{@link #getTotalTimeout()} caps how long the logic should keep trying the RPC until it gives - * up completely. If {@link #getTotalTimeout()} is set, initialRpcTimeout should be <= + * up completely. If {@link #getTotalTimeout()} is set, initialRpcTimeout should be <= * totalTimeout. * *

If there are no configurations, Retries have the default initial RPC timeout value of {@code @@ -356,7 +356,7 @@ public final Builder setInitialRpcTimeout(org.threeten.bp.Duration initialTimeou * the connection has been terminated). * *

{@link #getTotalTimeout()} caps how long the logic should keep trying the RPC until it - * gives up completely. If {@link #getTotalTimeout()} is set, initialRpcTimeout should be <= + * gives up completely. If {@link #getTotalTimeout()} is set, initialRpcTimeout should be <= * totalTimeout. * *

If there are no configurations, Retries have the default initial RPC timeout value of @@ -491,7 +491,7 @@ public final org.threeten.bp.Duration getInitialRpcTimeout() { * the connection has been terminated). * *

{@link #getTotalTimeout()} caps how long the logic should keep trying the RPC until it - * gives up completely. If {@link #getTotalTimeout()} is set, initialRpcTimeout should be <= + * gives up completely. If {@link #getTotalTimeout()} is set, initialRpcTimeout should be <= * totalTimeout. * *

If there are no configurations, Retries have the default initial RPC timeout value of diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ApiCallContext.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ApiCallContext.java index 67b5e5b285d1..36d6fd966e55 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ApiCallContext.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ApiCallContext.java @@ -42,7 +42,6 @@ import java.util.Map; import java.util.Set; import javax.annotation.Nonnull; -import org.jspecify.annotations.NullMarked; import org.jspecify.annotations.Nullable; /** @@ -55,7 +54,6 @@ * *

This is transport specific and each transport has an implementation with its own options. */ -@NullMarked @InternalExtensionOnly public interface ApiCallContext extends RetryingContext { @@ -65,6 +63,18 @@ public interface ApiCallContext extends RetryingContext { /** Returns a new ApiCallContext with the given channel set. */ ApiCallContext withTransportChannel(TransportChannel channel); + /** + * Returns the {@link TransportChannel} associated with this call context, or {@code null} if none + * is set. + * + *

Note: By default, this method returns {@code null}. If an implementation does not override + * this method, automatic mTLS certificate rotation and channel refreshing in retrying callables + * will be disabled. + */ + default TransportChannel getTransportChannel() { + return null; + } + /** Returns a new ApiCallContext with the given Endpoint Context. */ ApiCallContext withEndpointContext(EndpointContext endpointContext); diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ApiResultRetryAlgorithm.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ApiResultRetryAlgorithm.java index 7d04d38d2605..97d24d441ad2 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ApiResultRetryAlgorithm.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ApiResultRetryAlgorithm.java @@ -37,7 +37,6 @@ @NullMarked class ApiResultRetryAlgorithm extends BasicResultRetryAlgorithm { - /** Returns true if previousThrowable is an {@link ApiException} that is retryable. */ @Override public boolean shouldRetry(Throwable previousThrowable, ResponseT previousResponse) { return (previousThrowable instanceof ApiException) @@ -53,6 +52,12 @@ public boolean shouldRetry(Throwable previousThrowable, ResponseT previousRespon @Override public boolean shouldRetry( RetryingContext context, Throwable previousThrowable, ResponseT previousResponse) { + // Check UnauthenticatedException retryability first to ensure mTLS certificate + // rotation retries take precedence over static method retry codes. + if (previousThrowable instanceof UnauthenticatedException + && ((UnauthenticatedException) previousThrowable).isRetryable()) { + return true; + } if (context.getRetryableCodes() != null) { // Ignore the isRetryable() value of the throwable if the RetryingContext has a specific list // of codes that should be retried. diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/AttemptCallable.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/AttemptCallable.java index 897e1f2dae8e..e9978959f85a 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/AttemptCallable.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/AttemptCallable.java @@ -86,7 +86,33 @@ public ResponseT call() { .attemptStarted(request, externalFuture.getAttemptSettings().getOverallAttemptCount()); ApiFuture internalFuture = callable.futureCall(request, callContext); - externalFuture.setAttemptFuture(internalFuture); + final ApiCallContext finalContext = callContext; + ApiFuture mappedFuture = + ApiFutures.catching( + internalFuture, + UnauthenticatedException.class, + unauthenticatedException -> { + TransportChannel transportChannel = finalContext.getTransportChannel(); + if (transportChannel != null && transportChannel.shouldRefresh()) { + transportChannel.refresh(); + UnauthenticatedException newEx = + new UnauthenticatedException( + unauthenticatedException.getMessage(), + unauthenticatedException.getCause(), + unauthenticatedException.getStatusCode(), + true, // isRetryable = true + unauthenticatedException.getErrorDetails()); + newEx.setStackTrace(unauthenticatedException.getStackTrace()); + for (Throwable suppressed : unauthenticatedException.getSuppressed()) { + newEx.addSuppressed(suppressed); + } + throw newEx; + } + throw unauthenticatedException; + }, + com.google.common.util.concurrent.MoreExecutors.directExecutor()); + + externalFuture.setAttemptFuture(mappedFuture); } catch (Throwable e) { externalFuture.setAttemptFuture(ApiFutures.immediateFailedFuture(e)); } diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/BidiStreamingCallable.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/BidiStreamingCallable.java index 97e22d6ee41f..980aa5b9ac26 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/BidiStreamingCallable.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/BidiStreamingCallable.java @@ -241,11 +241,53 @@ public BidiStreamingCallable withDefaultCallContext( return new BidiStreamingCallable() { @Override public ClientStream internalCall( - ResponseObserver responseObserver, + final ResponseObserver responseObserver, ClientStreamReadyObserver onReady, ApiCallContext thisCallContext) { - return BidiStreamingCallable.this.internalCall( - responseObserver, onReady, defaultCallContext.merge(thisCallContext)); + final ApiCallContext mergedContext = defaultCallContext.merge(thisCallContext); + ResponseObserver refreshingObserver = + new ResponseObserver() { + @Override + public void onStart(StreamController controller) { + responseObserver.onStart(controller); + } + + @Override + public void onResponse(ResponseT response) { + responseObserver.onResponse(response); + } + + @Override + public void onError(Throwable t) { + if (t instanceof UnauthenticatedException) { + TransportChannel transportChannel = mergedContext.getTransportChannel(); + if (transportChannel != null && transportChannel.shouldRefresh()) { + transportChannel.refresh(); + UnauthenticatedException causeEx = (UnauthenticatedException) t; + UnauthenticatedException newEx = + new UnauthenticatedException( + causeEx.getMessage(), + causeEx.getCause(), + causeEx.getStatusCode(), + true, // isRetryable = true + causeEx.getErrorDetails()); + newEx.setStackTrace(causeEx.getStackTrace()); + for (Throwable suppressed : causeEx.getSuppressed()) { + newEx.addSuppressed(suppressed); + } + t = newEx; + } + } + responseObserver.onError(t); + } + + @Override + public void onComplete() { + responseObserver.onComplete(); + } + }; + + return BidiStreamingCallable.this.internalCall(refreshingObserver, onReady, mergedContext); } }; } diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ClientStreamingCallable.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ClientStreamingCallable.java index 854ce3fd5de8..9bff9209fbe6 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ClientStreamingCallable.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ClientStreamingCallable.java @@ -77,9 +77,46 @@ public ClientStreamingCallable withDefaultCallContext( return new ClientStreamingCallable() { @Override public ApiStreamObserver clientStreamingCall( - ApiStreamObserver responseObserver, ApiCallContext thisCallContext) { - return ClientStreamingCallable.this.clientStreamingCall( - responseObserver, defaultCallContext.merge(thisCallContext)); + final ApiStreamObserver responseObserver, ApiCallContext thisCallContext) { + final ApiCallContext mergedContext = defaultCallContext.merge(thisCallContext); + ApiStreamObserver refreshingObserver = + new ApiStreamObserver() { + @Override + public void onNext(ResponseT response) { + responseObserver.onNext(response); + } + + @Override + public void onError(Throwable t) { + if (t instanceof UnauthenticatedException) { + TransportChannel transportChannel = mergedContext.getTransportChannel(); + if (transportChannel != null && transportChannel.shouldRefresh()) { + transportChannel.refresh(); + UnauthenticatedException causeEx = (UnauthenticatedException) t; + UnauthenticatedException newEx = + new UnauthenticatedException( + causeEx.getMessage(), + causeEx.getCause(), + causeEx.getStatusCode(), + true, // isRetryable = true + causeEx.getErrorDetails()); + newEx.setStackTrace(causeEx.getStackTrace()); + for (Throwable suppressed : causeEx.getSuppressed()) { + newEx.addSuppressed(suppressed); + } + t = newEx; + } + } + responseObserver.onError(t); + } + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + + return ClientStreamingCallable.this.clientStreamingCall(refreshingObserver, mergedContext); } }; } diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ServerStreamingAttemptCallable.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ServerStreamingAttemptCallable.java index af6e8014e38a..26951d355f79 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ServerStreamingAttemptCallable.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ServerStreamingAttemptCallable.java @@ -221,6 +221,7 @@ public Void call() { .getTracer() .attemptStarted(request, outerRetryingFuture.getAttemptSettings().getOverallAttemptCount()); + final ApiCallContext finalContext = attemptContext; innerCallable.call( request, new StateCheckingResponseObserver() { @@ -236,6 +237,31 @@ public void onResponseImpl(ResponseT response) { @Override public void onErrorImpl(Throwable t) { + Throwable cause = t; + if (cause instanceof com.google.api.gax.retrying.ServerStreamingAttemptException) { + cause = cause.getCause(); + } + if (cause instanceof UnauthenticatedException) { + TransportChannel transportChannel = finalContext.getTransportChannel(); + if (transportChannel != null && transportChannel.shouldRefresh()) { + transportChannel.refresh(); + UnauthenticatedException causeEx = (UnauthenticatedException) cause; + UnauthenticatedException newEx = + new UnauthenticatedException( + causeEx.getMessage(), + causeEx.getCause(), + causeEx.getStatusCode(), + true, // isRetryable = true + causeEx.getErrorDetails()); + newEx.setStackTrace(causeEx.getStackTrace()); + for (Throwable suppressed : causeEx.getSuppressed()) { + newEx.addSuppressed(suppressed); + } + cause = newEx; + + t = cause; + } + } onAttemptError(t); } diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/TransportChannel.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/TransportChannel.java index 1866092e28f9..de83dbc73861 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/TransportChannel.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/TransportChannel.java @@ -31,10 +31,8 @@ import com.google.api.core.InternalExtensionOnly; import com.google.api.gax.core.BackgroundResource; -import org.jspecify.annotations.NullMarked; /** Class whose instances can issue RPCs on a particular transport. */ -@NullMarked @InternalExtensionOnly public interface TransportChannel extends BackgroundResource { @@ -49,4 +47,20 @@ public interface TransportChannel extends BackgroundResource { * Returns an empty {@link ApiCallContext} that is compatible with this {@code TransportChannel}. */ ApiCallContext getEmptyCallContext(); + + /** + * Refreshes or recreates the underlying network connections of this transport channel. + * + *

By default, this is a no-op for transports that do not require stateful connection lifecycle + * management. + */ + default void refresh() {} + + /** + * Returns true if a certificate rotation has been detected on disk and the transport channel + * should be refreshed, or false otherwise. + */ + default boolean shouldRefresh() { + return false; + } } diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/mtls/CertificateBasedAccess.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/mtls/CertificateBasedAccess.java index 99f2cca9b53d..0dbdf60723d2 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/mtls/CertificateBasedAccess.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/mtls/CertificateBasedAccess.java @@ -32,22 +32,45 @@ import com.google.api.core.InternalApi; import com.google.api.gax.rpc.internal.EnvironmentProvider; -import org.jspecify.annotations.NullMarked; +import com.google.auth.mtls.MtlsUtils; +import com.google.auth.oauth2.PropertyProvider; +import java.io.IOException; /** * Utility class for handling certificate-based access configurations. * - *

This class handles the processing of GOOGLE_API_USE_CLIENT_CERTIFICATE and - * GOOGLE_API_USE_MTLS_ENDPOINT environment variables according to https://google.aip.dev/auth/4114 + *

This class handles the processing of GOOGLE_API_USE_CLIENT_CERTIFICATE, + * GOOGLE_API_CERTIFICATE_CONFIG, and GOOGLE_API_USE_MTLS_ENDPOINT configurations. */ -@NullMarked @InternalApi public class CertificateBasedAccess { private final EnvironmentProvider envProvider; - /** The EnvironmentProvider mechanism supports env var injection for unit tests. */ + @InternalApi + public interface FileExistenceProvider { + boolean exists(String path); + } + + @InternalApi + public interface FileContentReader { + String read(String path) throws IOException; + } + public CertificateBasedAccess(EnvironmentProvider envProvider) { + this( + envProvider, + path -> new java.io.File(path).isFile(), + path -> + new String( + java.nio.file.Files.readAllBytes(java.nio.file.Paths.get(path)), + java.nio.charset.StandardCharsets.UTF_8)); + } + + CertificateBasedAccess( + EnvironmentProvider envProvider, + FileExistenceProvider fileExistenceProvider, + FileContentReader fileContentReader) { this.envProvider = envProvider; } @@ -66,20 +89,35 @@ public enum MtlsEndpointUsagePolicy { ALWAYS; } + private com.google.auth.oauth2.EnvironmentProvider getAuthEnvProvider() { + return name -> envProvider.getenv(name); + } + + private PropertyProvider getAuthPropertyProvider() { + return System::getProperty; + } + /** Returns if mutual TLS client certificate should be used. */ public boolean useMtlsClientCertificate() { - String useClientCertificate = envProvider.getenv("GOOGLE_API_USE_CLIENT_CERTIFICATE"); - return "true".equals(useClientCertificate); + return MtlsUtils.useMtlsClientCertificate(getAuthEnvProvider(), getAuthPropertyProvider()); } /** Returns the current mutual TLS endpoint usage policy. */ public MtlsEndpointUsagePolicy getMtlsEndpointUsagePolicy() { String mtlsEndpointUsagePolicy = envProvider.getenv("GOOGLE_API_USE_MTLS_ENDPOINT"); - if ("never".equals(mtlsEndpointUsagePolicy)) { + if ("never".equalsIgnoreCase(mtlsEndpointUsagePolicy)) { return MtlsEndpointUsagePolicy.NEVER; - } else if ("always".equals(mtlsEndpointUsagePolicy)) { + } else if ("always".equalsIgnoreCase(mtlsEndpointUsagePolicy)) { return MtlsEndpointUsagePolicy.ALWAYS; } return MtlsEndpointUsagePolicy.AUTO; } + + /** + * Resolves and returns the path to the mutual TLS client certificate, or null if none should be + * used. + */ + public String getWorkloadCertPath() { + return MtlsUtils.getWorkloadCertPath(getAuthEnvProvider(), getAuthPropertyProvider()); + } } diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/mtls/WorkloadCertificateUtils.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/mtls/WorkloadCertificateUtils.java new file mode 100644 index 000000000000..b6dac90db34c --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/mtls/WorkloadCertificateUtils.java @@ -0,0 +1,45 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.rpc.mtls; + +import com.google.api.core.InternalApi; +import com.google.auth.mtls.MtlsUtils; + +/** Internal utility class for managing dynamic workload certificates. */ +@InternalApi +public class WorkloadCertificateUtils { + + private WorkloadCertificateUtils() {} + + public static String getCertificateFingerprint(String certPath) { + String fingerprint = MtlsUtils.getCertificateFingerprint(certPath); + return fingerprint != null ? fingerprint : ""; + } +} diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/AttemptCallableTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/AttemptCallableTest.java index 4b5da578862c..2916d8a1d019 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/AttemptCallableTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/AttemptCallableTest.java @@ -34,11 +34,15 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import com.google.api.core.ApiFuture; import com.google.api.core.SettableApiFuture; import com.google.api.gax.retrying.RetrySettings; import com.google.api.gax.retrying.RetryingFuture; import com.google.api.gax.retrying.TimedAttemptSettings; import com.google.api.gax.rpc.testing.FakeCallContext; +import com.google.api.gax.rpc.testing.FakeChannel; +import com.google.api.gax.rpc.testing.FakeStatusCode; +import com.google.api.gax.rpc.testing.FakeTransportChannel; import com.google.api.gax.tracing.ApiTracer; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -133,4 +137,50 @@ void testRpcTimeoutIsNotErased() { assertThat(capturedCallContext.getValue().getTimeoutDuration()).isEqualTo(callerTimeout); } + + @Test + void testUnauthenticatedExceptionReThrowPreservesContext() { + FakeTransportChannel transportChannel = + FakeTransportChannel.create(new FakeChannel()).setShouldRefresh(true); + ApiCallContext callContext = + FakeCallContext.createDefault().withTransportChannel(transportChannel); + + UnauthenticatedException originalEx = + new UnauthenticatedException( + "Expired cert", + new IllegalStateException("Root cause"), + FakeStatusCode.of(StatusCode.Code.UNAUTHENTICATED), + false); + originalEx.setStackTrace( + new StackTraceElement[] {new StackTraceElement("foo", "bar", "Baz.java", 123)}); + originalEx.addSuppressed(new RuntimeException("Suppressed cause")); + + SettableApiFuture failedFuture = SettableApiFuture.create(); + failedFuture.setException(originalEx); + when(mockInnerCallable.futureCall(Mockito.anyString(), Mockito.any())).thenReturn(failedFuture); + + AttemptCallable callable = + new AttemptCallable<>(mockInnerCallable, "fake-request", callContext); + callable.setExternalFuture(mockExternalFuture); + + callable.call(); + + ArgumentCaptor futureCaptor = ArgumentCaptor.forClass(ApiFuture.class); + Mockito.verify(mockExternalFuture, Mockito.times(2)).setAttemptFuture(futureCaptor.capture()); + + Throwable thrown = null; + try { + futureCaptor.getValue().get(); + } catch (Exception e) { + thrown = e.getCause(); + } + + assertThat(thrown).isInstanceOf(UnauthenticatedException.class); + UnauthenticatedException rethrown = (UnauthenticatedException) thrown; + assertThat(rethrown.isRetryable()).isTrue(); + assertThat(rethrown.getCause()).isEqualTo(originalEx.getCause()); + assertThat(rethrown.getStackTrace()).isEqualTo(originalEx.getStackTrace()); + assertThat(rethrown.getSuppressed().length).isEqualTo(1); + assertThat(rethrown.getSuppressed()[0]).isInstanceOf(RuntimeException.class); + } } diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/EndpointContextTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/EndpointContextTest.java index 4c1b5320de4c..c53ca1509430 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/EndpointContextTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/EndpointContextTest.java @@ -77,8 +77,10 @@ void mtlsEndpointResolver_switchToMtlsAllowedIsFalse() throws IOException { FakeMtlsProvider.createTestMtlsKeyStore(), "", throwExceptionForGetKeyStore); boolean switchToMtlsEndpointAllowed = false; CertificateBasedAccess certificateBasedAccess = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "auto" : "true"); + org.mockito.Mockito.mock(CertificateBasedAccess.class); + org.mockito.Mockito.when(certificateBasedAccess.getMtlsEndpointUsagePolicy()) + .thenReturn(CertificateBasedAccess.MtlsEndpointUsagePolicy.AUTO); + org.mockito.Mockito.when(certificateBasedAccess.useMtlsClientCertificate()).thenReturn(true); Truth.assertThat( defaultEndpointContextBuilder.mtlsEndpointResolver( DEFAULT_ENDPOINT, @@ -97,8 +99,10 @@ void mtlsEndpointResolver_switchToMtlsAllowedIsTrue_mtlsUsageAuto() throws IOExc FakeMtlsProvider.createTestMtlsKeyStore(), "", throwExceptionForGetKeyStore); boolean switchToMtlsEndpointAllowed = true; CertificateBasedAccess certificateBasedAccess = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "auto" : "true"); + org.mockito.Mockito.mock(CertificateBasedAccess.class); + org.mockito.Mockito.when(certificateBasedAccess.getMtlsEndpointUsagePolicy()) + .thenReturn(CertificateBasedAccess.MtlsEndpointUsagePolicy.AUTO); + org.mockito.Mockito.when(certificateBasedAccess.useMtlsClientCertificate()).thenReturn(true); Truth.assertThat( defaultEndpointContextBuilder.mtlsEndpointResolver( DEFAULT_ENDPOINT, @@ -117,8 +121,10 @@ void mtlsEndpointResolver_switchToMtlsAllowedIsTrue_mtlsUsageAlways() throws IOE FakeMtlsProvider.createTestMtlsKeyStore(), "", throwExceptionForGetKeyStore); boolean switchToMtlsEndpointAllowed = true; CertificateBasedAccess certificateBasedAccess = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "always" : "true"); + org.mockito.Mockito.mock(CertificateBasedAccess.class); + org.mockito.Mockito.when(certificateBasedAccess.getMtlsEndpointUsagePolicy()) + .thenReturn(CertificateBasedAccess.MtlsEndpointUsagePolicy.ALWAYS); + org.mockito.Mockito.when(certificateBasedAccess.useMtlsClientCertificate()).thenReturn(true); Truth.assertThat( defaultEndpointContextBuilder.mtlsEndpointResolver( DEFAULT_ENDPOINT, @@ -137,8 +143,10 @@ void mtlsEndpointResolver_switchToMtlsAllowedIsTrue_mtlsUsageNever() throws IOEx FakeMtlsProvider.createTestMtlsKeyStore(), "", throwExceptionForGetKeyStore); boolean switchToMtlsEndpointAllowed = true; CertificateBasedAccess certificateBasedAccess = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "never" : "true"); + org.mockito.Mockito.mock(CertificateBasedAccess.class); + org.mockito.Mockito.when(certificateBasedAccess.getMtlsEndpointUsagePolicy()) + .thenReturn(CertificateBasedAccess.MtlsEndpointUsagePolicy.NEVER); + org.mockito.Mockito.when(certificateBasedAccess.useMtlsClientCertificate()).thenReturn(true); Truth.assertThat( defaultEndpointContextBuilder.mtlsEndpointResolver( DEFAULT_ENDPOINT, @@ -156,8 +164,10 @@ void mtlsEndpointResolver_switchToMtlsAllowedIsTrue_useCertificateIsFalse_nullMt MtlsProvider mtlsProvider = new FakeMtlsProvider(null, "", throwExceptionForGetKeyStore); boolean switchToMtlsEndpointAllowed = true; CertificateBasedAccess certificateBasedAccess = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "auto" : "false"); + org.mockito.Mockito.mock(CertificateBasedAccess.class); + org.mockito.Mockito.when(certificateBasedAccess.getMtlsEndpointUsagePolicy()) + .thenReturn(CertificateBasedAccess.MtlsEndpointUsagePolicy.AUTO); + org.mockito.Mockito.when(certificateBasedAccess.useMtlsClientCertificate()).thenReturn(false); Truth.assertThat( defaultEndpointContextBuilder.mtlsEndpointResolver( DEFAULT_ENDPOINT, @@ -174,8 +184,10 @@ void mtlsEndpointResolver_getKeyStore_throwsIOException() throws IOException { MtlsProvider mtlsProvider = new FakeMtlsProvider(null, "", throwExceptionForGetKeyStore); boolean switchToMtlsEndpointAllowed = true; CertificateBasedAccess certificateBasedAccess = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "auto" : "true"); + org.mockito.Mockito.mock(CertificateBasedAccess.class); + org.mockito.Mockito.when(certificateBasedAccess.getMtlsEndpointUsagePolicy()) + .thenReturn(CertificateBasedAccess.MtlsEndpointUsagePolicy.AUTO); + org.mockito.Mockito.when(certificateBasedAccess.useMtlsClientCertificate()).thenReturn(true); assertThrows( IOException.class, () -> @@ -272,8 +284,10 @@ void endpointContextBuild_mtlsConfigured_GDU() throws IOException { MtlsProvider mtlsProvider = new FakeMtlsProvider(FakeMtlsProvider.createTestMtlsKeyStore(), "", false); CertificateBasedAccess certificateBasedAccess = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "always" : "true"); + org.mockito.Mockito.mock(CertificateBasedAccess.class); + org.mockito.Mockito.when(certificateBasedAccess.getMtlsEndpointUsagePolicy()) + .thenReturn(CertificateBasedAccess.MtlsEndpointUsagePolicy.ALWAYS); + org.mockito.Mockito.when(certificateBasedAccess.useMtlsClientCertificate()).thenReturn(true); EndpointContext endpointContext = defaultEndpointContextBuilder .setClientSettingsEndpoint(null) @@ -293,8 +307,10 @@ void endpointContextBuild_mtlsConfigured_nonGDU_throwsIllegalArgumentException() MtlsProvider mtlsProvider = new FakeMtlsProvider(FakeMtlsProvider.createTestMtlsKeyStore(), "", false); CertificateBasedAccess certificateBasedAccess = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "always" : "true"); + org.mockito.Mockito.mock(CertificateBasedAccess.class); + org.mockito.Mockito.when(certificateBasedAccess.getMtlsEndpointUsagePolicy()) + .thenReturn(CertificateBasedAccess.MtlsEndpointUsagePolicy.ALWAYS); + org.mockito.Mockito.when(certificateBasedAccess.useMtlsClientCertificate()).thenReturn(true); EndpointContext.Builder endpointContextBuilder = defaultEndpointContextBuilder .setUniverseDomain("random.com") diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ServerStreamingAttemptCallableTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ServerStreamingAttemptCallableTest.java index 4cdd3cc018cc..2a52cb2b2572 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ServerStreamingAttemptCallableTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ServerStreamingAttemptCallableTest.java @@ -29,6 +29,7 @@ */ package com.google.api.gax.rpc; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.Mockito.mock; import com.google.api.core.AbstractApiFuture; @@ -248,6 +249,46 @@ void testInitialRetry() { Truth.assertThat(call.getRequest()).isEqualTo("request > 0"); } + @Test + @SuppressWarnings("ConstantConditions") + void testUnauthenticatedRefresh() { + TransportChannel transportChannel = Mockito.mock(TransportChannel.class); + Mockito.when(transportChannel.shouldRefresh()).thenReturn(true); + + ApiCallContext context = Mockito.mock(ApiCallContext.class); + Mockito.when(context.getTransportChannel()).thenReturn(transportChannel); + Mockito.when(context.getTracer()).thenReturn(BaseApiTracer.getInstance()); + Mockito.when(context.getTimeoutDuration()).thenReturn(java.time.Duration.ofHours(5)); + + resumptionStrategy = new MyStreamResumptionStrategy(); + ServerStreamingAttemptCallable callable = createCallable(context); + callable.start(); + + MockServerStreamingCall call = innerCallable.popLastCall(); + + // Send initial error + UnauthenticatedException initialError = + new UnauthenticatedException( + "test", + null, + com.google.api.gax.rpc.testing.FakeStatusCode.of(Code.UNAUTHENTICATED), + false); + call.getController().getObserver().onError(initialError); + + // Should notify the outer future + ExecutionException ee = + assertThrows( + ExecutionException.class, + () -> fakeRetryingFuture.getAttemptResult().get(1, TimeUnit.SECONDS)); + Throwable outerError = ee.getCause(); + Mockito.verify(transportChannel).refresh(); + Truth.assertThat(outerError).isInstanceOf(ServerStreamingAttemptException.class); + Truth.assertThat(((ServerStreamingAttemptException) outerError).hasSeenResponses()).isFalse(); + Truth.assertThat(((ServerStreamingAttemptException) outerError).canResume()).isTrue(); + Truth.assertThat(outerError.getCause()).isInstanceOf(UnauthenticatedException.class); + Truth.assertThat(((UnauthenticatedException) outerError.getCause()).isRetryable()).isTrue(); + } + @Test @SuppressWarnings("ConstantConditions") void testMidRetry() { diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/StreamingCallableTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/StreamingCallableTest.java index 6f9584826893..c46522114318 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/StreamingCallableTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/StreamingCallableTest.java @@ -129,7 +129,7 @@ void testClientStreamingCall() { ClientStreamingCallable callable = stashCallable.withDefaultCallContext(defaultCallContext); callable.clientStreamingCall(observer); - assertSame(observer, stashCallable.getActualObserver()); + org.junit.jupiter.api.Assertions.assertNotNull(stashCallable.getActualObserver()); assertSame(defaultCallContext, stashCallable.getContext()); } @@ -158,7 +158,7 @@ void testClientStreamingCallWithContext() { ClientStreamingCallable callable = stashCallable.withDefaultCallContext(FakeCallContext.createDefault()); callable.clientStreamingCall(observer, context); - assertSame(observer, stashCallable.getActualObserver()); + org.junit.jupiter.api.Assertions.assertNotNull(stashCallable.getActualObserver()); FakeCallContext actualContext = (FakeCallContext) stashCallable.getContext(); assertSame(channel, actualContext.getChannel()); assertSame(credentials, actualContext.getCredentials()); diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/mtls/AbstractMtlsTransportChannelTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/mtls/AbstractMtlsTransportChannelTest.java index bea4674b765b..5ed563254965 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/mtls/AbstractMtlsTransportChannelTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/mtls/AbstractMtlsTransportChannelTest.java @@ -64,8 +64,10 @@ void testNotUseClientCertificate() throws IOException, GeneralSecurityException @Test void testUseClientCertificate() throws IOException, GeneralSecurityException { CertificateBasedAccess certificateBasedAccess = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "auto" : "true"); + org.mockito.Mockito.mock(CertificateBasedAccess.class); + org.mockito.Mockito.when(certificateBasedAccess.getMtlsEndpointUsagePolicy()) + .thenReturn(CertificateBasedAccess.MtlsEndpointUsagePolicy.AUTO); + org.mockito.Mockito.when(certificateBasedAccess.useMtlsClientCertificate()).thenReturn(true); MtlsProvider provider = new FakeMtlsProvider(FakeMtlsProvider.createTestMtlsKeyStore(), "", false); assertNotNull(getMtlsObjectFromTransportChannel(provider, certificateBasedAccess)); @@ -74,8 +76,10 @@ void testUseClientCertificate() throws IOException, GeneralSecurityException { @Test void testNoClientCertificate() throws IOException, GeneralSecurityException { CertificateBasedAccess certificateBasedAccess = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "auto" : "true"); + org.mockito.Mockito.mock(CertificateBasedAccess.class); + org.mockito.Mockito.when(certificateBasedAccess.getMtlsEndpointUsagePolicy()) + .thenReturn(CertificateBasedAccess.MtlsEndpointUsagePolicy.AUTO); + org.mockito.Mockito.when(certificateBasedAccess.useMtlsClientCertificate()).thenReturn(true); MtlsProvider provider = new FakeMtlsProvider(null, "", false); assertNull(getMtlsObjectFromTransportChannel(provider, certificateBasedAccess)); } @@ -84,8 +88,10 @@ void testNoClientCertificate() throws IOException, GeneralSecurityException { void testGetKeyStoreThrows() throws GeneralSecurityException { // Test the case where provider.getKeyStore() throws. CertificateBasedAccess certificateBasedAccess = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "auto" : "true"); + org.mockito.Mockito.mock(CertificateBasedAccess.class); + org.mockito.Mockito.when(certificateBasedAccess.getMtlsEndpointUsagePolicy()) + .thenReturn(CertificateBasedAccess.MtlsEndpointUsagePolicy.AUTO); + org.mockito.Mockito.when(certificateBasedAccess.useMtlsClientCertificate()).thenReturn(true); MtlsProvider provider = new FakeMtlsProvider(null, "", true); IOException actual = assertThrows( diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/mtls/CertificateBasedAccessTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/mtls/CertificateBasedAccessTest.java index e328e0af4799..a6ba55fba6a5 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/mtls/CertificateBasedAccessTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/mtls/CertificateBasedAccessTest.java @@ -32,52 +32,108 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.assertNull; +import java.util.HashMap; +import java.util.Map; import org.junit.jupiter.api.Test; class CertificateBasedAccessTest { + private static class TestEnv { + private final Map env = new HashMap<>(); + + void set(String key, String val) { + env.put(key, val); + } + + String get(String name) { + return env.get(name); + } + } + + private CertificateBasedAccess createCba(TestEnv env) { + return new CertificateBasedAccess(env::get); + } + @Test void testUseMtlsEndpointAlways() { - CertificateBasedAccess cba = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "always" : "false"); + TestEnv env = new TestEnv(); + env.set("GOOGLE_API_USE_MTLS_ENDPOINT", "always"); + CertificateBasedAccess cba = createCba(env); assertEquals( CertificateBasedAccess.MtlsEndpointUsagePolicy.ALWAYS, cba.getMtlsEndpointUsagePolicy()); } @Test void testUseMtlsEndpointAuto() { - CertificateBasedAccess cba = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "auto" : "false"); + TestEnv env = new TestEnv(); + env.set("GOOGLE_API_USE_MTLS_ENDPOINT", "auto"); + CertificateBasedAccess cba = createCba(env); assertEquals( CertificateBasedAccess.MtlsEndpointUsagePolicy.AUTO, cba.getMtlsEndpointUsagePolicy()); } @Test void testUseMtlsEndpointNever() { - CertificateBasedAccess cba = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_MTLS_ENDPOINT") ? "never" : "false"); + TestEnv env = new TestEnv(); + env.set("GOOGLE_API_USE_MTLS_ENDPOINT", "never"); + CertificateBasedAccess cba = createCba(env); + assertEquals( + CertificateBasedAccess.MtlsEndpointUsagePolicy.NEVER, cba.getMtlsEndpointUsagePolicy()); + } + + @Test + void testUseMtlsEndpointCaseInsensitive() { + TestEnv env = new TestEnv(); + env.set("GOOGLE_API_USE_MTLS_ENDPOINT", "ALWAYS"); + CertificateBasedAccess cba = createCba(env); + assertEquals( + CertificateBasedAccess.MtlsEndpointUsagePolicy.ALWAYS, cba.getMtlsEndpointUsagePolicy()); + + env.set("GOOGLE_API_USE_MTLS_ENDPOINT", "NEVER"); assertEquals( CertificateBasedAccess.MtlsEndpointUsagePolicy.NEVER, cba.getMtlsEndpointUsagePolicy()); } @Test - void testUseMtlsClientCertificateTrue() { - CertificateBasedAccess cba = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_CLIENT_CERTIFICATE") ? "true" : "auto"); - assertTrue(cba.useMtlsClientCertificate()); + void testUseMtlsClientCertificateExplicitTrueNoCredentials() { + TestEnv env = new TestEnv(); + env.set("GOOGLE_API_USE_CLIENT_CERTIFICATE", "true"); + CertificateBasedAccess cba = createCba(env); + // Explicit 'true' permits mTLS if certs exist, but if no certs are present, returns false/null + // cleanly (Row 3) + assertFalse(cba.useMtlsClientCertificate()); + assertNull(cba.getWorkloadCertPath()); + } + + @Test + void testUseMtlsClientCertificateExplicitFalse() { + TestEnv env = new TestEnv(); + env.set("GOOGLE_API_USE_CLIENT_CERTIFICATE", "false"); + + CertificateBasedAccess cba = createCba(env); + assertFalse(cba.useMtlsClientCertificate()); + assertNull(cba.getWorkloadCertPath()); } @Test - void testUseMtlsClientCertificateFalse() { - CertificateBasedAccess cba = - new CertificateBasedAccess( - name -> name.equals("GOOGLE_API_USE_CLIENT_CERTIFICATE") ? "false" : "auto"); + void testUseMtlsClientCertificateUnsetNoFiles() { + TestEnv env = new TestEnv(); + CertificateBasedAccess cba = createCba(env); + assertFalse(cba.useMtlsClientCertificate()); + assertNull(cba.getWorkloadCertPath()); + } + + @Test + void testUseMtlsClientCertificateConfigMissingConfigFile_returnsNullSafely() { + TestEnv env = new TestEnv(); + env.set("GOOGLE_API_CERTIFICATE_CONFIG", "/nonexistent/config.json"); + + CertificateBasedAccess cba = createCba(env); + + // Non-existent config file on disk returns false/null safely per Row 3 assertFalse(cba.useMtlsClientCertificate()); + assertNull(cba.getWorkloadCertPath()); } } diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/testing/FakeCallContext.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/testing/FakeCallContext.java index 1cdefe435d55..e41dc041220d 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/testing/FakeCallContext.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/testing/FakeCallContext.java @@ -241,6 +241,11 @@ public FakeChannel getChannel() { return channel; } + @Override + public TransportChannel getTransportChannel() { + return channel != null ? FakeTransportChannel.create(channel) : null; + } + @Override public java.time.Duration getTimeoutDuration() { return timeout; diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/testing/FakeChannel.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/testing/FakeChannel.java index b725da363b3b..6d676db40032 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/testing/FakeChannel.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/testing/FakeChannel.java @@ -32,4 +32,24 @@ import com.google.api.core.InternalApi; @InternalApi("for testing") -public class FakeChannel {} +public class FakeChannel { + private volatile boolean shouldRefresh = false; + private volatile int refreshCount = 0; + + public FakeChannel setShouldRefresh(boolean shouldRefresh) { + this.shouldRefresh = shouldRefresh; + return this; + } + + public boolean shouldRefresh() { + return shouldRefresh; + } + + public void refresh() { + refreshCount++; + } + + public int getRefreshCount() { + return refreshCount; + } +} diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/testing/FakeTransportChannel.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/testing/FakeTransportChannel.java index 0d4abac8f1c6..bbde9feb4807 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/testing/FakeTransportChannel.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/testing/FakeTransportChannel.java @@ -41,6 +41,33 @@ public class FakeTransportChannel implements TransportChannel { private volatile boolean isShutdown = false; private volatile Map headers; private volatile Executor executor; + private volatile boolean shouldRefresh = false; + private volatile int refreshCount = 0; + + public FakeTransportChannel setShouldRefresh(boolean shouldRefresh) { + if (channel != null) { + channel.setShouldRefresh(shouldRefresh); + } + this.shouldRefresh = shouldRefresh; + return this; + } + + @Override + public boolean shouldRefresh() { + return channel != null ? channel.shouldRefresh() : shouldRefresh; + } + + @Override + public void refresh() { + if (channel != null) { + channel.refresh(); + } + refreshCount++; + } + + public int getRefreshCount() { + return channel != null ? channel.getRefreshCount() : refreshCount; + } private FakeTransportChannel(FakeChannel channel) { this.channel = channel;