8. grpc 认证鉴权 —— oauth2 - 《gRPC 源码解读》

grpc 官方提供了对 oauth2 认证鉴权的实现 demo，放在 examples 目录的 features 目录的 authentication 目录下，我们来看一下源码实现

server

server 端源码实现如下：

server 端先调用了 tls 包下的 LoadX509KeyPair，通过 server 的公钥和私钥生成了一个 Certificate 结构体来保存证书信息。然后注册了一个校验 token 的方法到拦截器中，并将证书信息设置到 serverOption 中，构造 server 的时候层层透传进去，最终会被设置到 Server 里面 ServerOptions 结构中的 credentials.TransportCredentials 和 UnaryServerInterceptor 中。

我们来看看这两个结构什么时候会被调用，先梳理调用链路，在 s.Serve ——> s.handleRawConn ——> s.serveStreams ——> s.handleStream ——> s.processUnaryRPC 方法中有一行

可以看到调用了 md.Handler 方法，将 s.opts.unaryInt 这个结构传入了进去。s.opts.unaryInt 就是我们之前注册的 UnaryServerInterceptor 拦截器。md 是一个 MethodDesc 这个结构，包括了 MethodName 和 Handler

type MethodDesc struct {
    MethodName string
    Handler    methodHandler
}

这里会取出我们之前注册进去的结构，还记得我们介绍 helloworld 时 RegisterService 吗？至于如何取出 MethodName，源码中的设计非常复杂，经过了层层包装，这里不是本节重点就不赘述了。

func RegisterGreeterServer(s *grpc.Server, srv GreeterServer) {
    s.RegisterService(&_Greeter_serviceDesc, srv)
}
var _Greeter_serviceDesc = grpc.ServiceDesc{
    ServiceName: "helloworld.Greeter",
    HandlerType: (*GreeterServer)(nil),
    Methods: []grpc.MethodDesc{
        {
            MethodName: "SayHello",
            Handler:    _Greeter_SayHello_Handler,
        },
    },
    Streams:  []grpc.StreamDesc{},
    Metadata: "helloworld.proto",
}

我们看到 md.Handler 其实是 _Greeter_SayHello_Handler 这个结构，它也是在 pb 文件中生成的。

func _Greeter_SayHello_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
    in := new(HelloRequest)
    if err := dec(in); err != nil {
        return nil, err
    }
    if interceptor == nil {
        return srv.(GreeterServer).SayHello(ctx, in)
    }
    info := &grpc.UnaryServerInfo{
        Server:     srv,
        FullMethod: "/helloworld.Greeter/SayHello",
    }
    handler := func(ctx context.Context, req interface{}) (interface{}, error) {
        return srv.(GreeterServer).SayHello(ctx, req.(*HelloRequest))
    }
    return interceptor(ctx, in, info, handler)
}

这里调用了我们传入的 interceptor 方法。回到我们的调用：

reply, appErr := md.Handler(srv.server, ctx, df, s.opts.unaryInt)

    opts := []grpc.ServerOption{
        // The following grpc.ServerOption adds an interceptor for all unary
        // RPCs. To configure an interceptor for streaming RPCs, see:
        grpc.UnaryInterceptor(ensureValidToken),
        // Enable TLS for all incoming connections.
    }
    s := grpc.NewServer(opts...)

看 grpc.UnaryInterceptor 这个方法，其实是将 ensureValidToken 这个函数赋值给了 s.opts.unaryInt

所以之前我们执行的这一行

    return interceptor(ctx, in, info, handler)

其实是执行了 ensureValidToken 这个函数，这个函数就是我们在 server 端定义的 token 校验的函数。先取出我们传入的 metadata 数据，然后校验 token

    func ensureValidToken(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
        md, ok := metadata.FromIncomingContext(ctx)
        if !ok {
            return nil, errMissingMetadata
        }
        // The keys within metadata.MD are normalized to lowercase.
        // See: https://godoc.org/google.golang.org/grpc/metadata#New
        if !valid(md["authorization"]) {
            return nil, errInvalidToken
        }
        // Continue execution of handler after ensuring a valid token.
        return handler(ctx, req)
}

校验完 token 后，最终执行了 handler(ctx, req)

    handler := func(ctx context.Context, req interface{}) (interface{}, error) {
        return srv.(GreeterServer).SayHello(ctx, req.(*HelloRequest))
    }
    return interceptor(ctx, in, info, handler)

可以看到最终其实执行了 GreeterServer 的 SayHello 这个函数，也就是我们在 main 函数中定义的，这个函数就是我们在 server 端定义的提供 SayHello 给客户端回消息的函数。

// SayHello implements helloworld.GreeterServer
func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
    log.Printf("Received: %v", in.Name)
    return &pb.HelloReply{Message: "Hello " + in.Name}, nil
}

这里还可以额外说一下，md.Handler 执行完之后，其实 reply 就是 SayHello 的回包。

    reply, appErr := md.Handler(srv.server, ctx, df, s.opts.unaryInt)

获取到回包之后 server 执行了 sendResponse 方法，将回包发送给 client，这个方法我们之前已经剖析过了，最终会调用 http2Server 的 Write 方法。

if err := s.sendResponse(t, stream, reply, cp, opts, comp); err != nil {

看到这里，server 端对 token 的校验在哪里执行的我们已经清楚了。假如还没有被绕晕，那么恭喜你！可以继续完成 client 的挑战了。

client

可以看到 client 首先通过 NewOauthAccess 方法生成了包含 token 信息的 PerRPCCredentials 结构

func NewOauthAccess(token *oauth2.Token) credentials.PerRPCCredentials {
    return oauthAccess{token: *token}
}

然后再将 PerRPCCredentials 通过 grpc.WithPerRPCCredentials(perRPC) 添加到了到了 client 的 DialOptions 中的 transport.ConnectOptions 结构中的 [] credentials.PerRPCCredentials 结构中。

那么这个结构什么时候被使用呢，我们来看看。先梳理下调用链，在 client 调用的 Invoke ——> invoke ——> newClientStream ——> cs.newAttemptLocked ——> cs.cc.getTransport ——> pick ——> acw.getAddrConn().getReadyTransport() ——> ac.connect() ——> ac.resetTransport() ——> ac.tryAllAddrs ——> ac.createTransport ——> transport.NewClientTransport ——> newHTTP2Client 这个方法里面，有这么一段代码，先取出 []credentials.PerRPCCredentials 中的所有 PerRPCCredentials 添加到 perRPCCreds 中。

    transportCreds := opts.TransportCredentials
    perRPCCreds := opts.PerRPCCredentials
    if b := opts.CredsBundle; b != nil {
        if t := b.TransportCredentials(); t != nil {
            transportCreds = t
        }
        if t := b.PerRPCCredentials(); t != nil {
            perRPCCreds = append(perRPCCreds, t)
        }
    }

然后再将 perRPCCreds 赋值给 http2Client 的 perRPCCreds 属性

t := &http2Client{
    ...
    perRPCCreds:           perRPCCreds,
    ...
}

那么 perRPCCreds 属性什么时候被用呢？来继续跟踪，newClientStream 方法中有一段代码

    op := func(a *csAttempt) error { return a.newStream() }
    if err := cs.withRetry(op, func() { cs.bufferForRetryLocked(0, op) }); err != nil {
        cs.finish(err)
        return nil, err
    }

这里调用了 csAttempt 的 newStream ——> a.t.NewStream (http2Client 的 NewStream) ——> createHeaderFields ——> getTrAuthData 方法

func (t *http2Client) getTrAuthData(ctx context.Context, audience string) (map[string]string, error) {
    if len(t.perRPCCreds) == 0 {
        return nil, nil
    }
    authData := map[string]string{}
    for _, c := range t.perRPCCreds {
        data, err := c.GetRequestMetadata(ctx, audience)
        if err != nil {
            if _, ok := status.FromError(err); ok {
                return nil, err
            }
            return nil, status.Errorf(codes.Unauthenticated, "transport: %v", err)
        }
        for k, v := range data {
            // Capital header names are illegal in HTTP/2.
            k = strings.ToLower(k)
            authData[k] = v
        }
    }
    return authData, nil
}

这个方法，通过调用 GetRequestMetadata 取出 token 信息，这里会调用 oauth 的 GetRequestMetadata 方法，按照指定格式拼装成一个 map[string]string{} 的形式

func (s *serviceAccount) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
    s.mu.Lock()
    defer s.mu.Unlock()
    if !s.t.Valid() {
        var err error
        s.t, err = s.config.TokenSource(ctx).Token()
        if err != nil {
            return nil, err
        }
    }
    return map[string]string{
        "authorization": s.t.Type() + " " + s.t.AccessToken,

然后将以 map[string]string{} 的形式组装成一个 string map 返回，如下：

至此，client 和 server 的数据流转过程被打通