golang工程——grpc-gateway 转发http header中自定义字段到grpc上下文元数据

http header 转发到 grpc上下文

grpc网关可以将请求体内容转发到grpc对应消息中。那如何获取http header头中的信息，本文将介绍如何将http header转发到grpc上下文并采用拦截器，获取http header中的内容。 有些http header中的内置字段是会转发的比如Authorization，但是狠多自定义字段是转发不了的。

本文实现http header中自定义字段转发到grpc上下文并采用拦截器做个简单鉴权

代码可以参考前面几篇grpc-gateway博客

grpc-gateway入门,环境+简单案例

grpc-gateway proto定义http路由

grpc-gateway定义http路由

网关代码修改

如果要转发http header中的自定义内容，生成的网关代码需要进行修改，增加一些网关服务器选项

• runtime.WithIncomingHeaderMatcher： 请求http header 设置转发哪些到grpc上下文
• runtime.WithOutgoingHeaderMatcher: 响应后,grpc上下文转发到http头部

gateway.go

package gateway

import (
    "context"
    "flag"
    "fmt"
    "net/http"

    "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    _ "google.golang.org/grpc/grpclog"

    gw "user/proto"  // Update
)

var (
    // command-line options:
    // gRPC server endpoint
    grpcServerEndpoint = flag.String("grpc-server-endpoint",  "localhost:50051", "gRPC server endpoint")
)

func Run() error {
    ctx := context.Background()
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()
    // 请求时,将http header中某些字段转发到grpc上下文
    inComingOpt :=  runtime.WithIncomingHeaderMatcher(func(s string) (string, bool) {
        fmt.Println("header:" + s)
        switch s {
        case "Service-Authorization":
            fmt.Println("Service-Authorization hit")
            return "Service-Authorization", true
        default:
            return "", false
        }
    })
    // 响应后,grpc上下文转发到http头部
    outGoingOpt := runtime.WithOutgoingHeaderMatcher(func(s string) (string, bool) {
       return "", false
    })
    // Register gRPC server endpoint
    // Note: Make sure the gRPC server is running properly and accessible
    mux := runtime.NewServeMux(inComingOpt, outGoingOpt)
    
    //添加文件上传处理函数
    mux.HandlePath("POST", "/upload", uploadHandler)
    
    opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
    err := gw.RegisterUserHandlerFromEndpoint(ctx, mux,  *grpcServerEndpoint, opts)
    if err != nil {
        return err
    }

    // Start HTTP server (and proxy calls to gRPC server endpoint)
    return http.ListenAndServe(":8081", mux)
}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

文件上传接口修改，因为这是自定义的网关路由接口，需要自己将Header中的字段转发到grpc中

upload.go

package gateway

import (
    "context"
    "fmt"
    "github.com/golang/protobuf/jsonpb"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    "google.golang.org/grpc/metadata"
    "io"
    "net/http"
    "user/proto"
)

func uploadHandler(w http.ResponseWriter, r *http.Request, pathParams map[string]string) {
    // 先从request解析文件
    err := r.ParseForm()
    if err != nil {
        http.Error(w, fmt.Sprintf("上传失败:%s", err.Error()), http.StatusInternalServerError)
    }

    f, header, err :=r.FormFile("attachment")

    if err != nil {
        http.Error(w, fmt.Sprintf("上传失败:%s", err.Error()), http.StatusInternalServerError)
    }

    defer f.Close()

    // 访问grpc server端, 实际生产用连接池

    conn, err := grpc.Dial(*grpcServerEndpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))

    if err != nil {
        http.Error(w, fmt.Sprintf("上传失败:%s", err.Error()), http.StatusInternalServerError)
    }

    defer conn.Close()

    c := proto.NewUserClient(conn)

    ctx := context.Background()
    ctx = metadata.NewOutgoingContext(ctx, metadata.New(map[string]string{
        "file_name":header.Filename,
        "service-authorization":r.Header.Get("Service-Authorization"),
    }))
    stream, err := c.Upload(ctx)

    if err != nil {
        http.Error(w, fmt.Sprintf("上传失败:%s", err.Error()), http.StatusInternalServerError)
    }

    // 读文件流 转发给grpc
    buf := make([]byte, 512)
    for {
        n, err := f.Read(buf)
        if err != nil && err != io.EOF{
            http.Error(w, fmt.Sprintf("上传失败:%s", err.Error()), http.StatusInternalServerError)
        }
        if n == 0 {
            break
        }

        stream.Send(&proto.UploadRequest{
            Content: buf[:n],
            Size: int64(n),
        })
    }

    res, err := stream.CloseAndRecv()
    if err != nil {
        http.Error(w, fmt.Sprintf("上传失败:%s", err.Error()), http.StatusInternalServerError)
    }

    m := jsonpb.Marshaler{}
    str, _ := m.MarshalToString(res)
    if err != nil {
        http.Error(w, fmt.Sprintf("上传失败:%s", err.Error()), http.StatusInternalServerError)
    }

    w.Header().Add("Content-Type", "application/json")
    fmt.Fprintf(w, str)

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

grpc服务代码修改

拦截器，从上下文中获取元数据进行业务操作即可

interceptor.go

package server

import (
    "context"
    "errors"
    "fmt"
    "google.golang.org/grpc"
    "google.golang.org/grpc/metadata"
    "strings"
)

func UnaryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
    err = auth(ctx)
    if err != nil {
        return nil, err
    }

    return handler(ctx, req)
}

func StreamInterceptor(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
    err := auth(ss.Context())
    if err != nil {
        return err
    }
    return handler(srv, ss)
}


func auth(ctx context.Context) error {
    md, ok := metadata.FromIncomingContext(ctx)
    fmt.Println("meta:", md)
    // 实际应用中，返回前端提示需模糊化,详细错误可以打印日志
    if !ok {
        return errors.New("获取元数据失败，身份校验失败")
    }
    // 转发过来都是小写
    authorization := md["service-authorization"]
    if len(authorization) < 1 {
        return errors.New("获取身份令牌失败，身份校验失败")
    }
    token := strings.TrimPrefix(authorization[0], "Bearer ")
    if token != bearerToken {
        return errors.New("身份令牌对比失败，身份校验失败")
    }
    return nil
}

// 测试用
var bearerToken = "sdfdlsdhgeiasdxzasqqqy2ybfhhu2gyvb"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

并将拦截器注册到grpc服务中

s := grpc.NewServer(grpc.UnaryInterceptor(server.UnaryInterceptor), grpc.StreamInterceptor(server.StreamInterceptor))
1

重点还是网关代码修改，增加转发header的逻辑。