Python中为gRPC添加自定义加密的实现方法

Q1：拦截器的使用误区与正确姿势 #

Python的gRPC客户端拦截器确实可以实现类似C# CallInvoker的序列化前后处理，但你可能误解了intercept_unary_unary中request参数的含义：它是未序列化的原始请求对象，而非序列化后的字节数据。如果要在序列化后做加密处理，需要手动调用传入的序列化器处理原始请求，再对结果做加密操作；对于响应的反序列化前处理，则需捕获服务器返回的原始字节，解密后再传入反序列化器。

以下是完整的一元调用拦截器示例：

import grpc

class EncryptionInterceptor(grpc.UnaryUnaryClientInterceptor):
    def intercept_unary_unary(self, continuation, client_call_details, request):
        # 1. 调用原始序列化器生成字节数据
        serialized_request = client_call_details.request_serializer(request)
        # 2. 自定义加密处理
        encrypted_request = self._encrypt(serialized_request)
        # 3. 替换序列化器为直接返回加密后的字节
        modified_call_details = client_call_details._replace(
            request_serializer=lambda x: encrypted_request
        )
        # 4. 发起调用并处理响应
        response_future = continuation(modified_call_details, request)
        original_deserializer = client_call_details.response_deserializer
        
        def wrapped_deserializer(encrypted_response):
            decrypted_response = self._decrypt(encrypted_response)
            return original_deserializer(decrypted_response)
        
        # 替换响应的反序列化逻辑
        def handle_response(future):
            if future.exception() is None:
                result = future.result()
                return result._replace(response=wrapped_deserializer(result.response))
            return future.exception()
        
        response_future.add_done_callback(lambda f: handle_response(f))
        return response_future

    def _encrypt(self, data: bytes) -> bytes:
        # 你的自定义加密逻辑
        pass

    def _decrypt(self, data: bytes) -> bytes:
        # 你的自定义解密逻辑
        pass

不过这种方式需要针对一元、流式等不同调用类型分别实现拦截器，代码冗余度较高。

Q2：重写Channel类的方案可行性分析 #

你提出的重写grpc.Channel的思路是可行的，这是一种更简洁的全局处理方式，但你的代码存在几处错误：

• 类方法缺少self参数
• 响应处理函数错误使用了request_serializer，应调用原始的response_deserializer
• 未覆盖其他调用类型（如unary_stream、stream_unary、stream_stream）

修正后的完整代码示例如下：

import grpc

def encrypt(data: bytes) -> bytes:
    # 自定义加密逻辑
    pass

def decrypt(data: bytes) -> bytes:
    # 自定义解密逻辑
    pass

class EncryptedChannel(grpc.Channel):
    def unary_unary(self, method, request_serializer=None, response_deserializer=None, _registered_method=None):
        # 包装请求序列化器：先序列化再加密
        def wrapped_request_serializer(request):
            if request_serializer:
                raw_data = request_serializer(request)
            else:
                raw_data = request.SerializeToString()
            return encrypt(raw_data)
        
        # 包装响应反序列化器：先解密再反序列化
        def wrapped_response_deserializer(response_data):
            decrypted_data = decrypt(response_data)
            if response_deserializer:
                return response_deserializer(decrypted_data)
            else:
                # 针对protobuf消息的默认反序列化（需替换为你的实际消息类型）
                from your_proto_module import YourResponseMessage
                resp = YourResponseMessage()
                resp.ParseFromString(decrypted_data)
                return resp
        
        return super().unary_unary(
            method,
            wrapped_request_serializer,
            wrapped_response_deserializer,
            _registered_method
        )
    
    # 实现一元转流式调用的包装逻辑
    def unary_stream(self, method, request_serializer=None, response_deserializer=None, _registered_method=None):
        def wrapped_request_serializer(request):
            raw_data = request_serializer(request) if request_serializer else request.SerializeToString()
            return encrypt(raw_data)
        
        def wrapped_response_deserializer(response_data):
            decrypted_data = decrypt(response_data)
            if response_deserializer:
                return response_deserializer(decrypted_data)
            else:
                from your_proto_module import YourStreamResponseMessage
                resp = YourStreamResponseMessage()
                resp.ParseFromString(decrypted_data)
                return resp
        
        return super().unary_stream(
            method,
            wrapped_request_serializer,
            wrapped_response_deserializer,
            _registered_method
        )
    
    # 同理实现stream_unary和stream_stream方法

使用时直接用自定义Channel创建Stub即可：

from your_proto_module import your_service_pb2_grpc

channel = EncryptedChannel(target='your-server:port', options=[...])
stub = your_service_pb2_grpc.YourServiceStub(channel)

方案对比 #

• 拦截器方案：灵活性高，可针对特定方法做差异化处理，但需实现多类型拦截器，代码冗余。
• 重写Channel方案：全局生效，代码更简洁，适合所有请求/响应都需要加解密的场景，是更推荐的方案。

内容的提问来源于stack exchange，提问作者imakak