From 70b1ef906f3f18581bd7085243d8b7b08bc0ddf0 Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Mon, 17 Jan 2022 20:41:57 -0600 Subject: [PATCH 1/4] switch on req type --- server/grpc.go | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/server/grpc.go b/server/grpc.go index f24823915..5468ab84a 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -1655,7 +1655,16 @@ func Valid(ctx context.Context, method string, auth *authn.Auth, req interface{} if !ok { ua = []string{""} } - logger.Infof("GRPC: %v, %v, %v, %v, %v, %v", ip, ua, method, uinfo.UserID, uinfo.UserName, req) + + switch r := req.(type) { + case *pb.QueryPQLRequest: + logger.Infof("GRPC: %v, %v, %v, %v, %v, %+v", ip, ua, method, uinfo.UserID, uinfo.UserName, r) + case *pb.QuerySQLRequest: + logger.Infof("GRPC: %v, %v, %v, %v, %v, %+v", ip, ua, method, uinfo.UserID, uinfo.UserName, r) + default: + logger.Infof("GRPC: %v, %v, %v, %v, %v, %v", ip, ua, method, uinfo.UserID, uinfo.UserName) + + } return context.WithValue(ctx, "userinfo", uinfo), nil } From 695321e6c098125db8a1280bf1249e948f66a181 Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Mon, 17 Jan 2022 20:47:27 -0600 Subject: [PATCH 2/4] print attr --- server/grpc.go | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/server/grpc.go b/server/grpc.go index 5468ab84a..ee6875a42 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -1658,12 +1658,11 @@ func Valid(ctx context.Context, method string, auth *authn.Auth, req interface{} switch r := req.(type) { case *pb.QueryPQLRequest: - logger.Infof("GRPC: %v, %v, %v, %v, %v, %+v", ip, ua, method, uinfo.UserID, uinfo.UserName, r) + logger.Infof("GRPC: %v, %v, %v, %v, %v, %s", ip, ua, method, uinfo.UserID, uinfo.UserName, r.Pql) case *pb.QuerySQLRequest: - logger.Infof("GRPC: %v, %v, %v, %v, %v, %+v", ip, ua, method, uinfo.UserID, uinfo.UserName, r) + logger.Infof("GRPC: %v, %v, %v, %v, %v, %s", ip, ua, method, uinfo.UserID, uinfo.UserName, r.Sql) default: - logger.Infof("GRPC: %v, %v, %v, %v, %v, %v", ip, ua, method, uinfo.UserID, uinfo.UserName) - + logger.Infof("GRPC: %v, %v, %v, %v, %v", ip, ua, method, uinfo.UserID, uinfo.UserName) } return context.WithValue(ctx, "userinfo", uinfo), nil From 592fcbb05b72cb8fb2b3ab7eb1ad90be6d96f583 Mon Sep 17 00:00:00 2001 From: reesporte Date: Wed, 19 Jan 2022 21:20:08 -0600 Subject: [PATCH 3/4] one logger to rule them all unify logging method, actually log query for streaming and unary requests --- server/grpc.go | 61 +++++++++++++++++++++++++++++--------------------- 1 file changed, 35 insertions(+), 26 deletions(-) diff --git a/server/grpc.go b/server/grpc.go index ee6875a42..3a0e67fe2 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -165,8 +165,8 @@ func isAllowed(requested []string, allowed []string) bool { // QuerySQL handles the SQL request and sends RowResponses to the stream. func (h *GRPCHandler) QuerySQL(req *pb.QuerySQLRequest, stream pb.Pilosa_QuerySQLServer) error { ctx := stream.Context() - uinfo := ctx.Value("userinfo") - if uinfo != nil { + uinfo, ok := ctx.Value("userinfo").(*authn.UserInfo) + if ok && uinfo != nil { // authz m := sql.NewMapper() parsed, err := m.MapSQL(req.Sql) @@ -174,13 +174,14 @@ func (h *GRPCHandler) QuerySQL(req *pb.QuerySQLRequest, stream pb.Pilosa_QuerySQ return errors.Wrap(err, "parsing SQL") } - allowed := h.perms.GetAuthorizedIndexList(uinfo.(*authn.UserInfo).Groups, authz.Read) - if !h.perms.IsAdmin(uinfo.(*authn.UserInfo).Groups) { + allowed := h.perms.GetAuthorizedIndexList(uinfo.Groups, authz.Read) + if !h.perms.IsAdmin(uinfo.Groups) { if !isAllowed(parsed.Tables, allowed) { return status.Error(codes.PermissionDenied, "insufficient permissions to access requested tables") } ctx = context.WithValue(ctx, "indices", allowed) } + LogQuery(ctx, "QuerySQL", req, h.queryLogger) } start := time.Now() @@ -272,6 +273,7 @@ func (h *GRPCHandler) QueryPQL(req *pb.QueryPQLRequest, stream pb.Pilosa_QueryPQ return status.Error(codes.PermissionDenied, "insufficient permissions to access requested indexes") } } + LogQuery(ctx, "QueryPQL", req, h.queryLogger) } t := time.Now() resp, err := h.api.Query(stream.Context(), &query) @@ -693,6 +695,8 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe h.logger.Infof("DEPRECATED: Inspect is deprecated, please use Extract() instead.") }) + LogQuery(stream.Context(), "Inspect", req, h.queryLogger) + index, err := h.api.Index(stream.Context(), req.Index) if err != nil { return errToStatusError(err) @@ -1561,16 +1565,17 @@ func NewGRPCServer(opts ...grpcServerOption) (*grpcServer, error) { if server.auth != nil { gopts = append(gopts, grpc.UnaryInterceptor( func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { - ctx, err := Valid(ctx, info.FullMethod, server.auth, req, server.queryLogger) + ctx, err := Valid(ctx, server.auth) if err != nil { return nil, err } + LogQuery(ctx, info.FullMethod, req, server.logger) return handler(ctx, req) }, )) gopts = append(gopts, grpc.StreamInterceptor( func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { - ctx, err := Valid(ss.Context(), info.FullMethod, server.auth, srv, server.queryLogger) + ctx, err := Valid(ss.Context(), server.auth) if err != nil { return err } @@ -1597,6 +1602,29 @@ func NewGRPCServer(opts ...grpcServerOption) (*grpcServer, error) { return server, nil } +// LogQuery logs requests +func LogQuery(ctx context.Context, method string, req interface{}, logger logger.Logger) { + uinfo, ok := ctx.Value("userinfo").(*authn.UserInfo) + md, _ := metadata.FromIncomingContext(ctx) + p, ok := peer.FromContext(ctx) + ip := "" + if ok { + ip = p.Addr.String() + } + ua, ok := md["user-agent"] + if !ok { + ua = []string{""} + } + switch r := req.(type) { + case *pb.QueryPQLRequest: + logger.Infof("GRPC: %v, %v, %v, %v, %v, %s", ip, ua, method, uinfo.UserID, uinfo.UserName, r.Pql) + case *pb.QuerySQLRequest: + logger.Infof("GRPC: %v, %v, %v, %v, %v, %s", ip, ua, method, uinfo.UserID, uinfo.UserName, r.Sql) + default: + logger.Infof("GRPC: %v, %v, %v, %v, %v", ip, ua, method, uinfo.UserID, uinfo.UserName) + } +} + // wrappedStream wraps around the embedded grpc.ServerStream, and intercepts the RecvMsg and // SendMsg method call. type wrappedStream struct { @@ -1616,7 +1644,7 @@ func (w *wrappedStream) SendMsg(m interface{}) error { return w.ServerStream.SendMsg(m) } -func Valid(ctx context.Context, method string, auth *authn.Auth, req interface{}, logger logger.Logger) (context.Context, error) { +func Valid(ctx context.Context, auth *authn.Auth) (context.Context, error) { md, ok := metadata.FromIncomingContext(ctx) if !ok { return ctx, status.Errorf(codes.InvalidArgument, "missing metadata") @@ -1646,24 +1674,5 @@ func Valid(ctx context.Context, method string, auth *authn.Auth, req interface{} return ctx, status.Errorf(codes.Unauthenticated, err.Error()) } - p, ok := peer.FromContext(ctx) - ip := "" - if ok { - ip = p.Addr.String() - } - ua, ok := md["user-agent"] - if !ok { - ua = []string{""} - } - - switch r := req.(type) { - case *pb.QueryPQLRequest: - logger.Infof("GRPC: %v, %v, %v, %v, %v, %s", ip, ua, method, uinfo.UserID, uinfo.UserName, r.Pql) - case *pb.QuerySQLRequest: - logger.Infof("GRPC: %v, %v, %v, %v, %v, %s", ip, ua, method, uinfo.UserID, uinfo.UserName, r.Sql) - default: - logger.Infof("GRPC: %v, %v, %v, %v, %v", ip, ua, method, uinfo.UserID, uinfo.UserName) - } - return context.WithValue(ctx, "userinfo", uinfo), nil } From 81fcd9c22860f0458e3c8cf7df2954f34a2469bd Mon Sep 17 00:00:00 2001 From: reesporte Date: Thu, 20 Jan 2022 14:14:25 -0600 Subject: [PATCH 4/4] fix merge conflicts --- server/grpc.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/grpc.go b/server/grpc.go index fa0623ec3..4fc8fd3bd 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -182,7 +182,7 @@ func (h *GRPCHandler) QuerySQL(req *pb.QuerySQLRequest, stream pb.Pilosa_QuerySQ } allowed := h.perms.GetAuthorizedIndexList(uinfo.Groups, perm) - if !h.perms.IsAdmin(uinfo.(*authn.UserInfo).Groups) { + if !h.perms.IsAdmin(uinfo.Groups) { if !isAllowed(parsed.Tables, allowed) { return status.Error(codes.PermissionDenied, "insufficient permissions to access requested tables") }