Python 中进行 gRPC 认证(authentication)

在许多现代的微服务架构中，gRPC（Google Remote Procedure Call）因其高效的通讯性能和跨语言服务的接口定义能力而受到欢迎。安全的访问控制在 gRPC 中至关重要，它确保了只有合法认证的用户或服务能够调用 gRPC 服务。本文将一步步介绍如何在 Python 中实现 gRPC 认证，在开始之前，我们先准备一款 gRPC 调试工具。

gRPC 调试工具

为了优化 gRPC 的调试体验，选择一个可靠的调试工具至关重要。这里我比较推荐使用 Apifox —— 一款支持多种协议（如 http、https、WebSocket、Socket、gRPC、Dubbo 等）的全能接口测试工具。下面，我们将通过具体步骤演示如何在 Python 中进行 gRPC 认证，并借助 Apifox 进行高效调试。

立即体验 Apifox

Python 中进行 gRPC 认证

步骤 1: 环境搭建

首先，确保你安装了 Python 环境以及grpcio和grpcio-tools两个库。你可以使用 pip 安装它们：

pip install grpcio grpcio-tools

步骤 2: 定义你的 gRPC 服务

假设你已经有了一个.proto 文件定义了 gRPC 服务和消息类型。举个例子， example.proto：

syntax = "proto3";

package example;

// 定义了一个Greeter服务
service Greeter {
  // 发出问候的远程方法
  rpc SayHello (HelloRequest) returns (HelloReply) {}
}

// 问候的请求消息
message HelloRequest {
  string name = 1;
}

// 问候的回复消息
message HelloReply {
  string message = 1;
}

步骤 3: 生成服务代码

使用下面的命令从.proto文件生成 Python 代码：

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. example.proto

在当前目录中，你会看到生成的example_pb2.py和example_pb2_grpc.py文件。

步骤 4: 实现服务端

接下来在服务端实现认证机制。gRPC 支持多种认证方式，常见的有 SSL/TLS 认证和 Token 认证。

SSL/TLS 认证

import grpc
from example_pb2_grpc import GreeterServicer, add_GreeterServicer_to_server
from example_pb2 import HelloReply

# 创建gRPC服务器
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))

# 实现Greeter服务接口
class Greeter(GreeterServicer):
    def SayHello(self, request, context):
        return HelloReply(message='Hello, %s!' % request.name)

# 添加服务到服务器中
add_GreeterServicer_to_server(Greeter(), server)

# 生成服务器端证书和私钥
with open('server.crt', 'rb') as f:
    server_certificate = f.read()
with open('server.key', 'rb') as f:
    private_key = f.read()

# 创建SSL上下文
server_credentials = grpc.ssl_server_credentials(((private_key, server_certificate),))

# 为服务器添加安全策略
server.add_secure_port('[::]:50051', server_credentials)

# 启动gRPC服务器
server.start()
server.wait_for_termination()

Token 认证

import grpc
from example_pb2_grpc import GreeterServicer, add_GreeterServicer_to_server
from example_pb2 import HelloReply

# 创建Token认证的拦截器
class TokenAuthInterceptor(grpc.ServerInterceptor):
    def __init__(self, valid_token):
        self.valid_token = valid_token

    def intercept_service(self, continuation, handler_call_details):
        # 从context中获取metadata
        metadata = dict(handler_call_details.invocation_metadata)
        # 检查token是否合法
        if metadata.get('authorization') != self.valid_token:
            return grpc.unary_unary_rpc_terminator(
                grpc.StatusCode.UNAUTHENTICATED, 'Invalid token!')
        return continuation(handler_call_details)

# 创建gRPC服务器，并添加拦截器
interceptor = TokenAuthInterceptor(valid_token='your_secret_token')
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10), interceptors=(interceptor,))

# ...
# 后续代码与上面的SSL/TLS认证中一样

步骤 5: 实现客户端

客户端需要传递正确的认证信息以调用服务端的方法。

SSL/TLS 认证

import grpc
from example_pb2_grpc import GreeterStub
from example_pb2 import HelloRequest

# 从文件读取客户端证书
with open('client.crt', 'rb') as f:
    server_certificate = f.read()

# 创建SSL上下文
ssl_credentials = grpc.ssl_channel_credentials(root_certificates=server_certificate)

# 创建一个安全的gRPC通道
channel = grpc.secure_channel('localhost:50051', ssl_credentials)

# 创建服务存根
stub = GreeterStub(channel)

# 发起RPC调用
response = stub.SayHello(HelloRequest(name='World'))
print(response.message)

Token 认证

import grpc
from example_pb2_grpc import GreeterStub
from example_pb2 import HelloRequest

# 添加Token
def set_token(metadata):
    return (('authorization', 'your_secret_token'),)

# 创建gRPC通道，添加Token
channel = grpc.intercept_channel(
    grpc.insecure_channel('localhost:50051'),
    grpc.metadata_call_credentials(set_token)
)

# 创建服务存根
stub = GreeterStub(channel)

# 发起RPC调用
response = stub.SayHello(HelloRequest(name='World'))
print(response.message)