qianye1001 commented on code in PR #10826:
URL: https://github.com/apache/rocketmq/pull/10826#discussion_r3978481152
##########
proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java:
##########
@@ -81,17 +87,55 @@ public static void main(String[] args) {
TlsCertificateManager tlsCertificateManager = new
TlsCertificateManager();
PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(tlsCertificateManager);
- // create grpcServer
+ // create grpcServer (data plane). Capture the application
reference so the
+ // admin server can reuse the SAME GrpcChannelManager /
GrpcClientSettingsManager
+ // that the data plane uses to track online clients.
+ GrpcMessagingApplication dataPlaneApplication =
createServiceProcessor(messagingProcessor);
GrpcServer grpcServer = GrpcServerBuilder.newBuilder(executor,
ConfigurationManager.getProxyConfig().getGrpcServerPort(),
tlsCertificateManager)
- .addService(createServiceProcessor(messagingProcessor))
+ .addService(dataPlaneApplication)
.addService(ChannelzService.newInstance(100))
.addService(ProtoReflectionService.newInstance())
.configInterceptor()
.shutdownTime(ConfigurationManager.getProxyConfig().getGrpcShutdownTimeSeconds(),
TimeUnit.SECONDS)
.build();
PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(grpcServer);
+ // Dedicated admin gRPC server (control plane), gated by
grpcAdminServerEnable. Reuses
+ // the data plane's GrpcChannelManager so admin queries can see
the data-plane clients.
+ Integer adminPort =
ConfigurationManager.getProxyConfig().getGrpcAdminServerPort();
+ if (ConfigurationManager.getProxyConfig().isGrpcAdminServerEnable()
+ && adminPort != null && adminPort > 0) {
+ DefaultGrpcMessagingActivity dataPlaneActivity =
+ (DefaultGrpcMessagingActivity)
dataPlaneApplication.getGrpcMessagingActivity();
+ GrpcChannelManager sharedChannelManager =
dataPlaneActivity.getGrpcChannelManager();
+ GrpcClientSettingsManager sharedSettingsManager =
dataPlaneActivity.getGrpcClientSettingsManager();
+ DefaultMessagingProcessor defaultProcessor =
(DefaultMessagingProcessor) messagingProcessor;
+
+ // Forwards client-targeted admin RPCs to the proxy that owns
the client channel.
+ ProxyAdminForwarder adminForwarder =
+ new
ProxyAdminForwarder(defaultProcessor.getServiceManager());
+
PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(adminForwarder);
+
+ ProxyAdminGrpcService adminService = new ProxyAdminGrpcService(
+ defaultProcessor.getServiceManager(),
+ messagingProcessor,
+ sharedChannelManager,
+ sharedSettingsManager,
+ adminForwarder);
+ PROXY_START_AND_SHUTDOWN.appendStartAndShutdown(adminService);
+ GrpcServer adminGrpcServer =
GrpcServerBuilder.newBuilder(executor, adminPort, tlsCertificateManager)
+ .addService(adminService)
+ .configInterceptor()
+ // auth interceptor runs before the standard pipeline
+ .appendInterceptor(new ProxyAdminAuthInterceptor(
+ ConfigurationManager.getAuthConfig(),
messagingProcessor))
Review Comment:
**[P1] Normalize the transport channel ID before authentication**
Server interceptors run in reverse registration order, so appending
`ProxyAdminAuthInterceptor` here makes it execute before `HeaderInterceptor`.
At that point, `DefaultAuthenticationContextBuilder` reads `x-mq-channel-id`
directly from the inbound metadata, before it is replaced with the actual
transport channel ID.
With `StatefulAuthenticationStrategy` enabled, authentication results are
cached by channel ID + username. A request on another physical connection can
therefore supply the same metadata ID and username, reuse a successful cache
entry, and bypass signature verification.
I reproduced this locally against `0a2a49794d` using the real gRPC
interceptor chain and stateful authentication strategy with an in-memory
signature-checking provider: connection A populated the cache with a valid
signature; connection B used the same metadata channel ID and an invalid
signature, but still reached the service handler (expected 1 invocation, actual
2). The relevant ordering and authentication code are unchanged in `1e68ca8932`.
Please run `HeaderInterceptor` before authentication, or populate the
authentication context's channel ID directly from the server call's transport
attributes before evaluating it. This finding applies when stateful
authentication is enabled; the default `StatelessAuthenticationStrategy` is not
affected by this cache bypass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
