diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java index bd72629814..e380db3439 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/GrpcServerBuilder.java @@ -142,11 +142,8 @@ public class GrpcServerBuilder { public GrpcServerBuilder configInterceptor() { // grpc interceptors, including acl, logging etc. - if (ConfigurationManager.getProxyConfig().isEnableACL()) { - List accessValidators = ServiceProvider.load(ServiceProvider.ACL_VALIDATOR_ID, AccessValidator.class); - if (accessValidators.isEmpty()) { - throw new IllegalArgumentException("Load AccessValidator failed"); - } + List accessValidators = ServiceProvider.load(ServiceProvider.ACL_VALIDATOR_ID, AccessValidator.class); + if (!accessValidators.isEmpty()) { this.serverBuilder.intercept(new AuthenticationInterceptor(accessValidators)); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/interceptor/AuthenticationInterceptor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/interceptor/AuthenticationInterceptor.java index 409c2c8c16..5d3fc0f105 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/interceptor/AuthenticationInterceptor.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/interceptor/AuthenticationInterceptor.java @@ -32,6 +32,7 @@ import org.apache.rocketmq.acl.AccessValidator; import org.apache.rocketmq.acl.common.AclException; import org.apache.rocketmq.acl.common.MetadataHeader; import org.apache.rocketmq.acl.plain.PlainAccessResource; +import org.apache.rocketmq.proxy.config.ConfigurationManager; public class AuthenticationInterceptor implements ServerInterceptor { private final List accessValidatorList; @@ -46,28 +47,32 @@ public class AuthenticationInterceptor implements ServerInterceptor { return new ForwardingServerCallListener.SimpleForwardingServerCallListener(next.startCall(call, headers)) { @Override public void onMessage(R message) { - try { - GeneratedMessageV3 messageV3 = (GeneratedMessageV3) message; - MetadataHeader metadataHeader = MetadataHeader.builder() - .remoteAddress(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.REMOTE_ADDRESS)) - .namespace(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.NAMESPACE_ID)) - .authorization(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.AUTHORIZATION)) - .datetime(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.DATE_TIME)) - .sessionToken(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.SESSION_TOKEN)) - .requestId(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.REQUEST_ID)) - .language(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.LANGUAGE)) - .clientVersion(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.CLIENT_VERSION)) - .protocol(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.PROTOCOL_VERSION)) - .requestCode(RequestMapping.map(messageV3.getDescriptorForType().getFullName())) - .build(); - for (AccessValidator accessValidator : accessValidatorList) { - AccessResource accessResource = accessValidator.parse(messageV3, metadataHeader); - accessValidator.validate(accessResource); - addHeader(headers, messageV3, accessResource); + if (ConfigurationManager.getProxyConfig().isEnableACL()) { + try { + GeneratedMessageV3 messageV3 = (GeneratedMessageV3) message; + MetadataHeader metadataHeader = MetadataHeader.builder() + .remoteAddress(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.REMOTE_ADDRESS)) + .namespace(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.NAMESPACE_ID)) + .authorization(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.AUTHORIZATION)) + .datetime(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.DATE_TIME)) + .sessionToken(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.SESSION_TOKEN)) + .requestId(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.REQUEST_ID)) + .language(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.LANGUAGE)) + .clientVersion(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.CLIENT_VERSION)) + .protocol(InterceptorConstants.METADATA.get(Context.current()).get(InterceptorConstants.PROTOCOL_VERSION)) + .requestCode(RequestMapping.map(messageV3.getDescriptorForType().getFullName())) + .build(); + for (AccessValidator accessValidator : accessValidatorList) { + AccessResource accessResource = accessValidator.parse(messageV3, metadataHeader); + accessValidator.validate(accessResource); + addHeader(headers, messageV3, accessResource); + } + super.onMessage(message); + } catch (AclException aclException) { + throw new StatusRuntimeException(Status.PERMISSION_DENIED, headers); } + } else { super.onMessage(message); - } catch (AclException aclException) { - throw new StatusRuntimeException(Status.PERMISSION_DENIED, headers); } } };