基于Python的gRPC服务器方法权限控制方案咨询（支持离线）

Great choice going with mutual TLS for your offline gRPC service—this is exactly the right approach for trust and authentication when you can't rely on external token services like Google's. Let's break down how to implement method-level authorization step by step:

First, you'll need to establish mutual TLS (mTLS) so your server can verify client identities via their certificates. Here's the core setup for Python gRPC:

• Generate local CA and certificates: Since you're running offline, create a private CA, then issue server and client certificates signed by this CA. Each client type (e.g., "admin", "user", "device") should get a unique certificate with a distinct identifier (like a Common Name or custom extension).
• Configure the gRPC server: Load the CA certificate to validate client certificates, plus your server's own cert and key:

  import grpc
  from grpc import ssl_server_credentials
  
  def create_server_credentials():
      # Load CA cert (to verify clients)
      with open("ca.crt", "rb") as f:
          ca_cert = f.read()
      # Load server cert and key
      with open("server.crt", "rb") as f:
          server_cert = f.read()
      with open("server.key", "rb") as f:
          server_key = f.read()
      # Create credentials requiring client certs
      return ssl_server_credentials(
          [(server_key, server_cert)],
          root_certificates=ca_cert,
          require_client_auth=True
      )
  

• Configure clients: Each client type will use their own cert/key pair signed by your private CA to connect to the server.

Once mTLS is working, you need to pull the client's identity from their certificate. Use the gRPC context to access the peer certificate, then parse it with the cryptography library to extract identifiers:

from cryptography import x509
from cryptography.x509.oid import NameOID

def get_client_identity(context):
    # Get the peer certificate from the auth context
    peer_cert_pem = context.auth_context().get("peer_cert", None)
    if not peer_cert_pem:
        return None
    # Parse the PEM certificate
    cert = x509.load_pem_x509_certificate(peer_cert_pem)
    # Extract the Common Name (CN) as the client type
    client_cn = cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value
    return client_cn

You could also use custom certificate extensions if you need more granular identity data (e.g., a client_type field embedded in the cert).

You have two main options here—per-method checks or a centralized interceptor (preferred for clean code):

Option A: Centralized Interceptor (Recommended) #

Use a gRPC server interceptor to validate permissions for every method call, avoiding repetitive code in each service method. Define a permission map that links methods to allowed client types, then check against it:

from grpc import ServerInterceptor, StatusCode, intercept_server
from grpc._server import _Context

class AuthorizationInterceptor(ServerInterceptor):
    # Define your method-to-client-type permissions here
    PERMISSIONS = {
        "MyService.GetAdminData": ["admin_client"],
        "MyService.GetUserData": ["admin_client", "regular_user"],
        "MyService.DeviceSync": ["iot_device"]
    }

    def intercept_unary_unary(self, continuation, handler_call_details, request):
        # Get client identity from context
        client_identity = get_client_identity(_Context(handler_call_details.invocation_metadata))
        
        # Get the full method name (e.g., "MyService.GetAdminData")
        method_path = handler_call_details.method.split("/")
        full_method = f"{method_path[1]}.{method_path[-1]}"

        # Validate permissions
        allowed_clients = self.PERMISSIONS.get(full_method, [])
        if client_identity not in allowed_clients:
            return grpc.unary_unary_rpc_method_handler(
                lambda request, context: context.abort(
                    StatusCode.PERMISSION_DENIED,
                    f"Client {client_identity} is not allowed to call {full_method}"
                )
            )
        # Proceed to the actual method handler
        return continuation(handler_call_details, request)

# Attach the interceptor to your server
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
server = intercept_server(server, AuthorizationInterceptor())
server.add_secure_port("[::]:50051", create_server_credentials())

Option B: Per-Method Checks #

If you prefer explicit checks in each method (e.g., for highly custom logic), call the get_client_identity function directly in your service implementation:

class MyService(MyService_pb2_grpc.MyServiceServicer):
    def GetAdminData(self, request, context):
        client_identity = get_client_identity(context)
        if client_identity != "admin_client":
            context.abort(StatusCode.PERMISSION_DENIED, "Admin access required")
        # Proceed with method logic
        return MyService_pb2.AdminResponse(data="sensitive data")

• Certificate management: Store your private CA securely, and rotate client/server certificates before they expire. For offline use, generate all certs locally using tools like openssl or the cryptography library.
• Client type enforcement: Ensure each client type uses a unique certificate—never reuse certs across client categories, as this would break your permission model.
• Error clarity: Return specific PERMISSION_DENIED errors instead of generic ones, so clients understand why their request was rejected.

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