gRPC 除了提供 TLS 认证之外,还提供自定义认证方式。

在 gRPC 中定义了 PerRPCCredentials 接口,它的定义如下,它是 gRPC 提供的用于实现自定义认证的接口,**它的作用是将所需的安全认证信息添加到每一次 RPC 调用的 Context 中。

注意: PerRPCCredentials 要求 TLS 支持,否则会报 did not connect: grpc: the credentials require transport level security (use grpc.WithTransportCredentials() to set)

// PerRPCCredentials defines the common interface for the credentials which need to
// attach security information to every RPC (e.g., oauth2).
type PerRPCCredentials interface {
	// GetRequestMetadata gets the current request metadata, refreshing
	// tokens if required. This should be called by the transport layer on
	// each request, and the data should be populated in headers or other
	// context. If a status code is returned, it will be used as the status
	// for the RPC. uri is the URI of the entry point for the request.
	// When supported by the underlying implementation, ctx can be used for
	// timeout and cancellation. Additionally, RequestInfo data will be
	// available via ctx to this call.
	// TODO(zhaoq): Define the set of the qualified keys instead of leaving
	// it as an arbitrary string.
	GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error)
	// RequireTransportSecurity indicates whether the credentials requires
	// transport security.
	RequireTransportSecurity() bool


本节的代码基于 grpc-testing v0.0.1


// 定义 PerRPCCredentials
type Token struct {
	APPID  string
	Secret string
	Token  string

func (t Token) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
	return map[string]string{"app_id": t.APPID, "secret": t.Secret, "token": t.Token}, nil

func (t Token) RequireTransportSecurity() bool {
	return true

func main() {
	T := Token{
		APPID:  "app-id-testing",
		Secret: "secret-testing",
	// 假设 使用 OAuth2, 此处获取 Token 并启动协程定时刷新 Token
	T.Token = "token-testing"

	creds, err := credentials.NewClientTLSFromFile("./cert/ca.crt", "*")
	if err != nil {
		log.Fatalf("failed to load credentials: %v", err)
	// 添加 WithPerRPCCredentials Option
	conn, err := grpc.Dial("", grpc.WithTransportCredentials(creds), grpc.WithBlock(), grpc.WithPerRPCCredentials(T))
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	defer conn.Close()
	c := echo.NewEchoClient(conn)
	ctx, cancle := context.WithCancel(context.Background())
	defer cancle()
	r1, err := c.Echo(ctx, &echo.EchoRequest{Msg: "hello"})
	if err != nil {
		log.Fatalf("echo failed :%#v", err)
	log.Println("r1", r1.GetMsg())
	// 调用两次验证是否都传递了 metadata
	r2, err := c.Echo(ctx, &echo.EchoRequest{Msg: "hello"})
	if err != nil {
		log.Fatalf("echo failed :%#v", err)
	log.Println("r2", r2.GetMsg())


func (e *EchoServer) Echo(ctx context.Context, par *echo.EchoRequest) (*echo.EchoResponse, error) {
    // 获取 Metadata
	md, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return nil, errors.New("no Token")
	// TODO 验证 Token

	result := &echo.EchoResponse{
		Msg: par.GetMsg(),
	return result, nil

func main() {
	listen, err := net.Listen("tcp", ":8080")
	if err != nil {
		log.Fatalf("failed to listen: %v", err)

	// 启用 TLS
	creds, err := credentials.NewServerTLSFromFile("./cert/server.crt", "./cert/server.key")
	if err != nil {
		log.Fatal("create server TLS", err)
	s := grpc.NewServer(grpc.Creds(creds))
	echo.RegisterEchoServer(s, &EchoServer{})
	if err := s.Serve(listen); err != nil {
		log.Fatalf("failed to serve: %v", err)

以上就是 gRPC 自定义认证的实现过程,还是非常简单的