From d6720b5689b6bf15a6e131daddbd8d76be3ca092 Mon Sep 17 00:00:00 2001 From: zhouxiang Date: Tue, 10 May 2022 16:50:37 +0800 Subject: [PATCH] [ISSUE #3949] Interceptor * Add GlobalExceptionInterceptor * Catch AclException in AuthenticationInterceptor --- .../proxy/grpc/GrpcServerBuilder.java | 2 + .../AuthenticationInterceptor.java | 43 +++--- .../GlobalExceptionInterceptor.java | 128 ++++++++++++++++++ 3 files changed, 155 insertions(+), 18 deletions(-) create mode 100644 proxy/src/main/java/org/apache/rocketmq/proxy/grpc/interceptor/GlobalExceptionInterceptor.java 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 02e6d3a9df..bd72629814 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 @@ -45,6 +45,7 @@ import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.proxy.config.ProxyConfig; import org.apache.rocketmq.proxy.grpc.interceptor.AuthenticationInterceptor; import org.apache.rocketmq.proxy.grpc.interceptor.ContextInterceptor; +import org.apache.rocketmq.proxy.grpc.interceptor.GlobalExceptionInterceptor; import org.apache.rocketmq.proxy.grpc.interceptor.HeaderInterceptor; public class GrpcServerBuilder { @@ -150,6 +151,7 @@ public class GrpcServerBuilder { } this.serverBuilder + .intercept(new GlobalExceptionInterceptor()) .intercept(new ContextInterceptor()) .intercept(new HeaderInterceptor()); 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 a3b75034b1..d3c3063252 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 @@ -24,9 +24,12 @@ import io.grpc.Metadata; import io.grpc.ServerCall; import io.grpc.ServerCallHandler; import io.grpc.ServerInterceptor; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; import java.util.List; import org.apache.rocketmq.acl.AccessResource; 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.grpc.v2.adapter.RequestMapping; @@ -44,25 +47,29 @@ public class AuthenticationInterceptor implements ServerInterceptor { return new ForwardingServerCallListener.SimpleForwardingServerCallListener(next.startCall(call, headers)) { @Override public void onMessage(R message) { - 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); + 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); } - super.onMessage(message); } }; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/interceptor/GlobalExceptionInterceptor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/interceptor/GlobalExceptionInterceptor.java new file mode 100644 index 0000000000..0c34b15743 --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/interceptor/GlobalExceptionInterceptor.java @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.proxy.grpc.interceptor; + +import io.grpc.ForwardingServerCall; +import io.grpc.ForwardingServerCallListener; +import io.grpc.Metadata; +import io.grpc.ServerCall; +import io.grpc.ServerCallHandler; +import io.grpc.ServerInterceptor; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.logging.InternalLogger; +import org.apache.rocketmq.logging.InternalLoggerFactory; + +public class GlobalExceptionInterceptor implements ServerInterceptor { + private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.PROXY_LOGGER_NAME); + + @Override + public ServerCall.Listener interceptCall( + ServerCall call, + Metadata headers, + ServerCallHandler next + ) { + final ServerCall serverCall = new ClosableServerCall<>(call); + ServerCall.Listener delegate = next.startCall(serverCall, headers); + return new ForwardingServerCallListener.SimpleForwardingServerCallListener(delegate) { + @Override + public void onMessage(R message) { + try { + super.onMessage(message); + } catch (Throwable e) { + closeWithException(e); + } + } + + @Override + public void onHalfClose() { + try { + super.onHalfClose(); + } catch (Throwable e) { + closeWithException(e); + } + } + + @Override + public void onCancel() { + try { + super.onCancel(); + } catch (Throwable e) { + closeWithException(e); + } + } + + @Override + public void onComplete() { + try { + super.onComplete(); + } catch (Throwable e) { + closeWithException(e); + } + } + + @Override + public void onReady() { + try { + super.onReady(); + } catch (Throwable e) { + closeWithException(e); + } + } + + private void closeWithException(Throwable t) { + Metadata trailers = new Metadata(); + Status status = Status.INTERNAL.withDescription(t.getMessage()); + boolean printLog = true; + + if (t instanceof StatusRuntimeException) { + trailers = ((StatusRuntimeException) t).getTrailers(); + status = ((StatusRuntimeException) t).getStatus(); + // no error stack for permission denied. + if (status.getCode().value() == Status.PERMISSION_DENIED.getCode().value()) { + printLog = false; + } + } + + if (printLog) { + log.error("grpc server has exception. errorMsg:{}, e:", t.getMessage(), t); + } + + serverCall.close(status, trailers); + } + }; + } + + private static class ClosableServerCall extends + ForwardingServerCall.SimpleForwardingServerCall { + private boolean closeCalled = false; + + ClosableServerCall(ServerCall delegate) { + super(delegate); + } + + @Override + public synchronized void close(final Status status, final Metadata trailers) { + if (!closeCalled) { + closeCalled = true; + ClosableServerCall.super.close(status, trailers); + } + } + } +}