diff --git a/all/cmd/main/main.go b/all/cmd/main/main.go index 2d82897..0508873 100644 --- a/all/cmd/main/main.go +++ b/all/cmd/main/main.go @@ -45,7 +45,7 @@ func main() { _ = srv.Stop(ctx) }() - if err := srv.Start(config.Spec.Addr); err != nil && err != http.ErrServerClosed { + if err := srv.Start(config.Spec.Server.GRPC.Addr, config.Spec.Server.HTTP.Addr, config.Spec.DynamicRPC.Allow); err != nil && err != http.ErrServerClosed { panic(err) } } diff --git a/all/etc/default_dev.yaml b/all/etc/default_dev.yaml index 20c9c0c..757a768 100644 --- a/all/etc/default_dev.yaml +++ b/all/etc/default_dev.yaml @@ -1,6 +1,18 @@ Service: default -BindIP: 0.0.0.0 -Port: 12000 +Server: + GRPC: + BindIP: 0.0.0.0 + Port: 12000 + HTTP: + BindIP: 0.0.0.0 + Port: 12001 + +# Dynamic JSON-to-protobuf unary RPC endpoint: POST /rpc/{full.service}.{method} +# An empty allow list denies every dynamic call. Use "*" to expose all +# reflection-visible unary methods, or list full service/method names. +DynamicRPC: + Allow: + - "*" Databases: Driver: postgres @@ -38,7 +50,7 @@ Fts: AccessKeySecret: CHANGE_ME UseSSL: false Local: - Site: http://127.0.0.1:12000/files + Site: http://127.0.0.1:12001/files UploadDir: ./uploader/ FtsConfig: InputKey: file diff --git a/all/go.mod b/all/go.mod index 85a4f15..a1f54e3 100644 --- a/all/go.mod +++ b/all/go.mod @@ -44,11 +44,12 @@ require ( github.com/patrickmn/go-cache v2.1.0+incompatible go.etcd.io/etcd/client/v3 v3.7.1 go.yaml.in/yaml/v3 v3.0.5 - golang.org/x/net v0.57.0 google.golang.org/grpc v1.83.0 gorm.io/gorm v1.31.2 ) +require golang.org/x/net v0.57.0 // indirect + require ( filippo.io/edwards25519 v1.2.0 // indirect github.com/KyleBanks/depth v1.2.1 // indirect @@ -153,7 +154,7 @@ require ( google.golang.org/genproto v0.0.0-20260807164820-c8921c73eeea google.golang.org/genproto/googleapis/api v0.0.0-20260807164820-c8921c73eeea // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260807164820-c8921c73eeea // indirect - google.golang.org/protobuf v1.36.11 // indirect + google.golang.org/protobuf v1.36.11 gopkg.in/ini.v1 v1.67.3 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect gorm.io/driver/mysql v1.6.0 // indirect diff --git a/all/internal/config/config.go b/all/internal/config/config.go index d61cf8e..0a7c897 100644 --- a/all/internal/config/config.go +++ b/all/internal/config/config.go @@ -11,10 +11,13 @@ import ( senderService "bsm/full/module/base/sender/service" walletService "bsm/full/module/finance/wallet/service" "git.apinb.com/bsm-sdk/core/conf" + "git.apinb.com/bsm-sdk/core/printer" ) type SrvConfig struct { conf.Base `yaml:",inline"` + Server ServerConfig `yaml:"Server"` + DynamicRPC DynamicRPCConfig `yaml:"DynamicRPC"` Databases *conf.DBConf `yaml:"Databases"` MicroService *conf.MicroServiceConf `yaml:"MicroService"` Rpc map[string]conf.RpcConf `yaml:"Rpc"` @@ -28,16 +31,44 @@ type SrvConfig struct { Wallet *walletService.Config `yaml:"Wallet"` } +type ListenerConfig struct { + BindIP string `yaml:"BindIP"` + Port string `yaml:"Port"` + Addr string `yaml:"-"` +} + +type ServerConfig struct { + GRPC ListenerConfig `yaml:"GRPC"` + HTTP ListenerConfig `yaml:"HTTP"` +} + +type DynamicRPCConfig struct { + Allow []string `yaml:"Allow"` +} + var Spec SrvConfig func New(serviceKey string) { conf.New(serviceKey, &Spec) - Spec.Port = conf.CheckPort(Spec.Port) - Spec.BindIP = conf.CheckIP(Spec.BindIP) - Spec.Addr = net.JoinHostPort(Spec.BindIP, Spec.Port) + normalizeListener(&Spec.Server.GRPC) + normalizeListener(&Spec.Server.HTTP) + if Spec.Server.GRPC.Addr == Spec.Server.HTTP.Addr { + panic("gRPC and HTTP listeners must use different addresses") + } + // Keep the embedded base address meaningful for module configurations that + // still consume it, while all itself uses the two explicit listeners. + Spec.BindIP, Spec.Port, Spec.Addr = Spec.Server.HTTP.BindIP, Spec.Server.HTTP.Port, Spec.Server.HTTP.Addr conf.NotNil(Spec.Service, Spec.Cache) assignSharedConfig() - conf.PrintInfo(Spec.Addr) + conf.PrintInfo(Spec.Server.GRPC.Addr) + printer.Info("gRPC Address: %s", Spec.Server.GRPC.Addr) + printer.Info("HTTP Address: %s", Spec.Server.HTTP.Addr) +} + +func normalizeListener(listener *ListenerConfig) { + listener.Port = conf.CheckPort(listener.Port) + listener.BindIP = conf.CheckIP(listener.BindIP) + listener.Addr = net.JoinHostPort(listener.BindIP, listener.Port) } func assignSharedConfig() { diff --git a/all/internal/config/config_test.go b/all/internal/config/config_test.go index 3cbc7f9..349418b 100644 --- a/all/internal/config/config_test.go +++ b/all/internal/config/config_test.go @@ -20,6 +20,12 @@ func TestAllDevConfig(t *testing.T) { if len(cfg.Services) == 0 { t.Fatal("services must not be empty") } + if cfg.Server.GRPC.Port == "" || cfg.Server.HTTP.Port == "" || cfg.Server.GRPC.Port == cfg.Server.HTTP.Port { + t.Fatal("separate gRPC and HTTP ports are required") + } + if len(cfg.DynamicRPC.Allow) == 0 { + t.Fatal("dynamic RPC allow list must be explicit") + } if cfg.Fts == nil || cfg.Mgt == nil || cfg.Passport == nil || cfg.Sender == nil || cfg.Wallet == nil { t.Fatal("service-specific configuration is incomplete") } diff --git a/all/internal/server/dynamic.go b/all/internal/server/dynamic.go new file mode 100644 index 0000000..e2434b0 --- /dev/null +++ b/all/internal/server/dynamic.go @@ -0,0 +1,236 @@ +package server + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "strings" + "sync" + + "github.com/gin-gonic/gin" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" + reflectionv1 "google.golang.org/grpc/reflection/grpc_reflection_v1" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protodesc" + "google.golang.org/protobuf/reflect/protoreflect" + "google.golang.org/protobuf/reflect/protoregistry" + "google.golang.org/protobuf/types/descriptorpb" + "google.golang.org/protobuf/types/dynamicpb" +) + +const maxDynamicRPCBody = 4 << 20 + +type dynamicGateway struct { + conn *grpc.ClientConn + allow map[string]struct{} + mu sync.RWMutex + cache map[string]protoreflect.MethodDescriptor +} + +type dynamicRPCResponse struct { + Code int32 `json:"code"` + Message string `json:"message"` + Data json.RawMessage `json:"data,omitempty"` + Details []json.RawMessage `json:"details,omitempty"` +} + +func newDynamicGateway(grpcAddr string, allow []string) (*dynamicGateway, error) { + conn, err := grpc.NewClient(reflectionTarget(grpcAddr), grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return nil, fmt.Errorf("create dynamic gRPC client: %w", err) + } + allowed := make(map[string]struct{}, len(allow)) + for _, item := range allow { + if item = strings.TrimSpace(item); item != "" { + allowed[item] = struct{}{} + } + } + return &dynamicGateway{conn: conn, allow: allowed, cache: make(map[string]protoreflect.MethodDescriptor)}, nil +} + +func reflectionTarget(addr string) string { + host, port, err := net.SplitHostPort(addr) + if err != nil { + return addr + } + if host == "" || host == "0.0.0.0" || host == "::" { + host = "127.0.0.1" + } + return net.JoinHostPort(host, port) +} + +func (g *dynamicGateway) Close() error { return g.conn.Close() } + +func (g *dynamicGateway) handle(c *gin.Context) { + fullMethod := strings.TrimSpace(c.Param("method")) + serviceName, methodName, ok := splitFullMethod(fullMethod) + if !ok { + writeDynamicError(c, status.Error(codes.InvalidArgument, "method must be {full.service}.{method}")) + return + } + if !g.isAllowed(serviceName, fullMethod) { + writeDynamicError(c, status.Error(codes.PermissionDenied, "dynamic RPC method is not allowed")) + return + } + + descriptor, err := g.resolveMethod(c.Request.Context(), serviceName, methodName) + if err != nil { + writeDynamicError(c, err) + return + } + if descriptor.IsStreamingClient() || descriptor.IsStreamingServer() { + writeDynamicError(c, status.Error(codes.Unimplemented, "streaming RPC methods are not supported")) + return + } + + body, err := io.ReadAll(io.LimitReader(c.Request.Body, maxDynamicRPCBody+1)) + if err != nil { + writeDynamicError(c, status.Error(codes.InvalidArgument, "read request body: "+err.Error())) + return + } + if len(body) > maxDynamicRPCBody { + writeDynamicError(c, status.Error(codes.ResourceExhausted, "request body exceeds 4 MiB")) + return + } + + request := dynamicpb.NewMessage(descriptor.Input()) + if err := (protojson.UnmarshalOptions{DiscardUnknown: false}).Unmarshal(body, request); err != nil { + writeDynamicError(c, status.Error(codes.InvalidArgument, "invalid protobuf JSON: "+err.Error())) + return + } + response := dynamicpb.NewMessage(descriptor.Output()) + ctx := outgoingMetadata(c.Request) + grpcMethod := "/" + serviceName + "/" + methodName + if err := g.conn.Invoke(ctx, grpcMethod, request, response); err != nil { + writeDynamicError(c, err) + return + } + + data, err := (protojson.MarshalOptions{UseProtoNames: false}).Marshal(response) + if err != nil { + writeDynamicError(c, status.Error(codes.Internal, "marshal protobuf response: "+err.Error())) + return + } + c.JSON(http.StatusOK, dynamicRPCResponse{Code: int32(codes.OK), Message: codes.OK.String(), Data: data}) +} + +func splitFullMethod(value string) (string, string, bool) { + separator := strings.LastIndexByte(value, '.') + if separator <= 0 || separator == len(value)-1 { + return "", "", false + } + return value[:separator], value[separator+1:], true +} + +func (g *dynamicGateway) isAllowed(serviceName, fullMethod string) bool { + for _, key := range []string{"*", serviceName, fullMethod} { + if _, ok := g.allow[key]; ok { + return true + } + } + return false +} + +func (g *dynamicGateway) resolveMethod(ctx context.Context, serviceName, methodName string) (protoreflect.MethodDescriptor, error) { + cacheKey := serviceName + "." + methodName + g.mu.RLock() + method := g.cache[cacheKey] + g.mu.RUnlock() + if method != nil { + return method, nil + } + + stream, err := reflectionv1.NewServerReflectionClient(g.conn).ServerReflectionInfo(ctx) + if err != nil { + return nil, status.Error(codes.Unavailable, "open gRPC reflection stream: "+err.Error()) + } + if err := stream.Send(&reflectionv1.ServerReflectionRequest{ + MessageRequest: &reflectionv1.ServerReflectionRequest_FileContainingSymbol{FileContainingSymbol: serviceName}, + }); err != nil { + return nil, status.Error(codes.Unavailable, "query gRPC reflection: "+err.Error()) + } + reflectionResponse, err := stream.Recv() + if err != nil { + return nil, status.Error(codes.Unavailable, "read gRPC reflection response: "+err.Error()) + } + fileResponse := reflectionResponse.GetFileDescriptorResponse() + if fileResponse == nil { + if reflectionErr := reflectionResponse.GetErrorResponse(); reflectionErr != nil { + return nil, status.Error(codes.Code(reflectionErr.ErrorCode), reflectionErr.ErrorMessage) + } + return nil, status.Error(codes.NotFound, "service descriptor not found") + } + + set := &descriptorpb.FileDescriptorSet{} + for _, encoded := range fileResponse.FileDescriptorProto { + file := &descriptorpb.FileDescriptorProto{} + if err := proto.Unmarshal(encoded, file); err != nil { + return nil, status.Error(codes.Internal, "decode reflected descriptor: "+err.Error()) + } + set.File = append(set.File, file) + } + files, err := protodesc.NewFiles(set) + if err != nil { + return nil, status.Error(codes.Internal, "build reflected descriptors: "+err.Error()) + } + descriptor, err := files.FindDescriptorByName(protoreflect.FullName(serviceName)) + if err != nil { + if err == protoregistry.NotFound { + return nil, status.Error(codes.NotFound, "service not found") + } + return nil, status.Error(codes.Internal, "resolve service descriptor: "+err.Error()) + } + service, ok := descriptor.(protoreflect.ServiceDescriptor) + if !ok { + return nil, status.Error(codes.NotFound, "symbol is not a gRPC service") + } + method = service.Methods().ByName(protoreflect.Name(methodName)) + if method == nil { + return nil, status.Error(codes.NotFound, "method not found") + } + g.mu.Lock() + g.cache[cacheKey] = method + g.mu.Unlock() + return method, nil +} + +func outgoingMetadata(request *http.Request) context.Context { + pairs := make([]string, 0) + for name, values := range request.Header { + lower := strings.ToLower(name) + if lower != "authorization" && lower != "x-request-id" && !strings.HasPrefix(lower, "x-") { + continue + } + for _, value := range values { + pairs = append(pairs, lower, value) + } + } + return metadata.NewOutgoingContext(request.Context(), metadata.Pairs(pairs...)) +} + +func writeDynamicError(c *gin.Context, err error) { + grpcStatus := status.Convert(err) + details := make([]json.RawMessage, 0, len(grpcStatus.Details())) + for _, detail := range grpcStatus.Details() { + message, ok := detail.(proto.Message) + if !ok { + continue + } + if encoded, marshalErr := protojson.Marshal(message); marshalErr == nil { + details = append(details, encoded) + } + } + c.JSON(http.StatusOK, dynamicRPCResponse{ + Code: int32(grpcStatus.Code()), + Message: grpcStatus.Message(), + Details: details, + }) +} diff --git a/all/internal/server/dynamic_test.go b/all/internal/server/dynamic_test.go new file mode 100644 index 0000000..b32ffc8 --- /dev/null +++ b/all/internal/server/dynamic_test.go @@ -0,0 +1,97 @@ +package server + +import ( + "encoding/json" + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gin-gonic/gin" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/health" + healthpb "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/reflection" +) + +func TestDynamicGatewayInvokesUnaryRPC(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + grpcServer := grpc.NewServer() + healthServer := health.NewServer() + healthServer.SetServingStatus("", healthpb.HealthCheckResponse_SERVING) + healthpb.RegisterHealthServer(grpcServer, healthServer) + reflection.Register(grpcServer) + go func() { _ = grpcServer.Serve(listener) }() + t.Cleanup(func() { + grpcServer.Stop() + _ = listener.Close() + }) + + gateway, err := newDynamicGateway(listener.Addr().String(), []string{"grpc.health.v1.Health.Check"}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = gateway.Close() }) + + gin.SetMode(gin.TestMode) + engine := gin.New() + engine.POST("/rpc/:method", gateway.handle) + request := httptest.NewRequest(http.MethodPost, "/rpc/grpc.health.v1.Health.Check", strings.NewReader(`{"service":""}`)) + response := httptest.NewRecorder() + engine.ServeHTTP(response, request) + + if response.Code != http.StatusOK { + t.Fatalf("unexpected HTTP status: %d", response.Code) + } + var payload struct { + Code int32 `json:"code"` + Data json.RawMessage `json:"data"` + } + if err := json.Unmarshal(response.Body.Bytes(), &payload); err != nil { + t.Fatal(err) + } + if payload.Code != 0 || !strings.Contains(string(payload.Data), `"SERVING"`) { + t.Fatalf("unexpected response: %s", response.Body.String()) + } +} + +func TestDynamicGatewayDeniesMethodsByDefault(t *testing.T) { + gateway := &dynamicGateway{allow: map[string]struct{}{}} + gin.SetMode(gin.TestMode) + engine := gin.New() + engine.POST("/rpc/:method", gateway.handle) + request := httptest.NewRequest(http.MethodPost, "/rpc/grpc.health.v1.Health.Check", strings.NewReader(`{}`)) + response := httptest.NewRecorder() + engine.ServeHTTP(response, request) + if response.Code != http.StatusOK { + t.Fatalf("dynamic RPC errors must use HTTP 200, got %d", response.Code) + } + var payload dynamicRPCResponse + if err := json.Unmarshal(response.Body.Bytes(), &payload); err != nil { + t.Fatal(err) + } + if payload.Code != int32(codes.PermissionDenied) { + t.Fatalf("expected permission denied, got %s", response.Body.String()) + } +} + +func TestOutgoingMetadataFiltersHeaders(t *testing.T) { + request := httptest.NewRequest(http.MethodPost, "/", nil) + request.Header.Set("Authorization", "Bearer token") + request.Header.Set("X-Request-ID", "request-id") + request.Header.Set("Content-Type", "application/json") + ctx := outgoingMetadata(request) + forwarded, ok := metadata.FromOutgoingContext(ctx) + if !ok || len(forwarded.Get("authorization")) != 1 || len(forwarded.Get("x-request-id")) != 1 { + t.Fatalf("expected forwarded metadata: %v", forwarded) + } + if len(forwarded.Get("content-type")) != 0 { + t.Fatalf("content-type must not be forwarded: %v", forwarded) + } +} diff --git a/all/internal/server/server.go b/all/internal/server/server.go index 9de9432..cb1285d 100644 --- a/all/internal/server/server.go +++ b/all/internal/server/server.go @@ -3,16 +3,14 @@ package server import ( "bytes" "context" + "errors" "fmt" "net" "net/http" - "strings" "time" "github.com/gin-gonic/gin" gwRuntime "github.com/grpc-ecosystem/grpc-gateway/v2/runtime" - "golang.org/x/net/http2" - "golang.org/x/net/http2/h2c" "google.golang.org/grpc" "google.golang.org/grpc/reflection" ) @@ -21,7 +19,8 @@ type Server struct { GRPC *grpc.Server Gateway *gwRuntime.ServeMux HTTP *gin.Engine - server *http.Server + http *http.Server + dynamic *dynamicGateway } func New() *Server { @@ -36,12 +35,26 @@ func New() *Server { } } -func (s *Server) Start(addr string) error { - handler := h2c.NewHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.ProtoMajor == 2 && strings.HasPrefix(r.Header.Get("Content-Type"), "application/grpc") { - s.GRPC.ServeHTTP(w, r) - return - } +func (s *Server) Start(grpcAddr, httpAddr string, allow []string) error { + grpcListener, err := net.Listen("tcp", grpcAddr) + if err != nil { + return fmt.Errorf("listen gRPC on %s: %w", grpcAddr, err) + } + httpListener, err := net.Listen("tcp", httpAddr) + if err != nil { + _ = grpcListener.Close() + return fmt.Errorf("listen HTTP on %s: %w", httpAddr, err) + } + + s.dynamic, err = newDynamicGateway(grpcAddr, allow) + if err != nil { + _ = grpcListener.Close() + _ = httpListener.Close() + return err + } + s.HTTP.POST("/rpc/:method", s.dynamic.handle) + + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { recorder := newBufferedResponse() s.Gateway.ServeHTTP(recorder, r) if recorder.status != http.StatusNotFound { @@ -49,20 +62,25 @@ func (s *Server) Start(addr string) error { return } s.HTTP.ServeHTTP(w, r) - }), &http2.Server{}) - s.server = &http.Server{ - Addr: addr, + }) + s.http = &http.Server{ + Addr: httpAddr, Handler: handler, ReadHeaderTimeout: 10 * time.Second, IdleTimeout: 120 * time.Second, MaxHeaderBytes: 1 << 20, } - listener, err := net.Listen("tcp", addr) - if err != nil { - return err + fmt.Printf("all gRPC services listening on %s\n", grpcAddr) + fmt.Printf("all HTTP services listening on %s\n", httpAddr) + + errCh := make(chan error, 2) + go func() { errCh <- s.GRPC.Serve(grpcListener) }() + go func() { errCh <- s.http.Serve(httpListener) }() + serveErr := <-errCh + if errors.Is(serveErr, grpc.ErrServerStopped) || errors.Is(serveErr, http.ErrServerClosed) { + return http.ErrServerClosed } - fmt.Printf("all services listening on %s (gRPC + HTTP)\n", addr) - return s.server.Serve(listener) + return serveErr } func (s *Server) Stop(ctx context.Context) error { @@ -76,10 +94,13 @@ func (s *Server) Stop(ctx context.Context) error { case <-ctx.Done(): s.GRPC.Stop() } - if s.server == nil { - return ctx.Err() + if s.dynamic != nil { + _ = s.dynamic.Close() } - return s.server.Shutdown(ctx) + if s.http == nil { + return nil + } + return s.http.Shutdown(ctx) } type bufferedResponse struct {