diff --git a/cmd/wire.go b/cmd/wire.go index 74531f9..19169d4 100644 --- a/cmd/wire.go +++ b/cmd/wire.go @@ -17,6 +17,7 @@ import ( "kra/internal/integration/cache" "kra/internal/server" "kra/internal/server/handler" + "kra/internal/server/middleware" "kra/internal/server/router" "kra/internal/service" "kra/internal/worker" @@ -42,6 +43,7 @@ func wireApp(*conf.Server, *conf.Runtime, *slog.Logger, *logging.ReloadableLogge initialize.ProviderSet, wire.Bind(new(initialize.Backend), new(*data.Data)), wire.Bind(new(cache.RedisProvider), new(*data.Data)), + wire.Bind(new(middleware.TokenAuthenticator), new(*service.AuthService)), wire.Bind(new(biz.TaskMethodRegistry), new(*platformtask.Registry)), biz.ProviderSet, service.ProviderSet, diff --git a/cmd/wire_gen.go b/cmd/wire_gen.go index 41901ee..40b26f6 100644 --- a/cmd/wire_gen.go +++ b/cmd/wire_gen.go @@ -156,25 +156,20 @@ func wireApp(confServer *conf.Server, runtime *conf.Runtime, logger *slog.Logger v := handler.NewSet(authority, menu, api, permission, organization, announcement, handlerEmail, handlerPayment, task, media, audit, export, version, dictionary, parameter, apiToken, systemConfig, public, user, navigation, session, integrationConfig) routes := router.NewRoutes(v) taskMethods := worker.NewTaskMethods(taskUsecase, mediaUsecase, runtime) - mqReloadable, cleanup2, err := mq.New(store, logger) + moduleRuntime := app.Runtime(routes, taskMethods, registry) + websocketServer, cleanup2, err := websocket.New(store) if err != nil { cleanup() return nil, nil, err } - moduleRuntime, err := app.Runtime(routes, taskMethods, registry, catalog, mqReloadable) - if err != nil { - cleanup2() - cleanup() - return nil, nil, err - } - websocketServer, cleanup3, err := websocket.New(store) - if err != nil { - cleanup2() - cleanup() - return nil, nil, err - } engine := server.NewGinEngineWithRuntime(runtime, accessControlService, authService, securityService, auditRecorder, logger, string2, moduleRuntime, websocketServer) httpServer := server.NewGinServer(confServer, engine) + mqReloadable, cleanup3, err := mq.New(store, logger) + if err != nil { + cleanup2() + cleanup() + return nil, nil, err + } kratosApp := newApp(logger, httpServer, taskScheduler, auditRecorder, reloadableLogger, mqReloadable) return kratosApp, func() { cleanup3() diff --git a/internal/app/catalog.go b/internal/app/catalog.go index 634f1eb..594305e 100644 --- a/internal/app/catalog.go +++ b/internal/app/catalog.go @@ -7,7 +7,6 @@ import ( systemrouter "kra/internal/server/router" systemworker "kra/internal/worker" "kra/pkg/module" - "kra/pkg/mq" platformtask "kra/pkg/task" ) @@ -28,10 +27,7 @@ func TaskRegistry(catalog module.Catalog) *platformtask.Registry { } // Runtime composes HTTP route contributors from the enabled modules. -func Runtime(systemRoutes *systemrouter.Routes, systemTasks *systemworker.TaskMethods, registry *platformtask.Registry, catalog module.Catalog, subscriptions mq.SubscriptionRegistrar) (*module.Runtime, error) { +func Runtime(systemRoutes *systemrouter.Routes, systemTasks *systemworker.TaskMethods, registry *platformtask.Registry) *module.Runtime { platformtask.Apply(registry, systemTasks) - if err := mq.ApplySubscriptions(subscriptions, catalog.SubscriptionContributors()...); err != nil { - return nil, err - } - return module.NewRuntime(systemRoutes), nil + return module.NewRuntime(systemRoutes) } diff --git a/internal/integration/mq/emqx_test.go b/internal/integration/mq/emqx_test.go index 5d6bfc6..b493c93 100644 --- a/internal/integration/mq/emqx_test.go +++ b/internal/integration/mq/emqx_test.go @@ -12,10 +12,10 @@ import ( ) type fakeClient struct { - subscribed []string - unsubscribed []string - handlers map[string]platformmq.Handler - subscribeErrors map[string]error + subscribed []string + unsubscribed []string + handlers map[string]platformmq.Handler + subscribeErrors map[string]error unsubscribeErrors map[string]error } diff --git a/internal/server/middleware/access_log_test.go b/internal/server/middleware/access_log_test.go index f3fabf7..371ddcb 100644 --- a/internal/server/middleware/access_log_test.go +++ b/internal/server/middleware/access_log_test.go @@ -122,3 +122,25 @@ func TestAccessLogOmitsPaymentIntegrationConfigBody(t *testing.T) { t.Fatalf("payment configuration summary missing: %s", logText) } } + +func TestAccessLogRedactsWebSocketQueryToken(t *testing.T) { + gin.SetMode(gin.TestMode) + var logs bytes.Buffer + logger := slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelInfo})) + runtime := conf.NewRuntime(nil, &conf.AdminBackend{}) + engine := gin.New() + engine.Use(AccessLog(runtime, logger, "test")) + engine.GET("/ws", func(c *gin.Context) { c.Status(http.StatusSwitchingProtocols) }) + + request := httptest.NewRequest(http.MethodGet, "/ws?token=jwt-secret&room=orders", nil) + response := httptest.NewRecorder() + engine.ServeHTTP(response, request) + + logText := logs.String() + if strings.Contains(logText, "jwt-secret") { + t.Fatalf("websocket token leaked into access log: %s", logText) + } + if !strings.Contains(logText, "token=%2A%2A%2A") || !strings.Contains(logText, "room=orders") { + t.Fatalf("redacted query missing expected fields: %s", logText) + } +} diff --git a/internal/server/providers_system.go b/internal/server/providers_system.go index dd3c7d4..7ca4fb4 100644 --- a/internal/server/providers_system.go +++ b/internal/server/providers_system.go @@ -1,8 +1,6 @@ package server -import ( - "github.com/google/wire" -) +import "github.com/google/wire" // ProviderSet is server providers. var ProviderSet = wire.NewSet(NewGinEngineWithRuntime, NewGinServer) diff --git a/pkg/module/module.go b/pkg/module/module.go index ebf730a..432943e 100644 --- a/pkg/module/module.go +++ b/pkg/module/module.go @@ -6,7 +6,6 @@ package module import ( "github.com/gin-gonic/gin" "kra/pkg/database/migration" - "kra/pkg/mq" "kra/pkg/task" ) @@ -30,12 +29,11 @@ type TimedTask struct { } type Definition struct { - Name string - Migrations []migration.Step - Surface Surface - TimedTasks []TimedTask - Tasks []task.Method - SubscriptionContributors []mq.SubscriptionContributor + Name string + Migrations []migration.Step + Surface Surface + TimedTasks []TimedTask + Tasks []task.Method } type Catalog struct { @@ -75,16 +73,6 @@ func (c Catalog) TaskMethods() []task.Method { return methods } -// SubscriptionContributors returns the module declarations that are applied -// to the process-wide MQ runtime during application startup. -func (c Catalog) SubscriptionContributors() []mq.SubscriptionContributor { - var contributors []mq.SubscriptionContributor - for _, item := range c.Definitions { - contributors = append(contributors, item.SubscriptionContributors...) - } - return contributors -} - type RouteRegistrar interface { RegisterRoutes(public, private *gin.RouterGroup, engine *gin.Engine) } diff --git a/pkg/mq/rabbitmq.go b/pkg/mq/rabbitmq.go index 574a94c..d460ce9 100644 --- a/pkg/mq/rabbitmq.go +++ b/pkg/mq/rabbitmq.go @@ -191,6 +191,9 @@ func (c *RabbitMQ) Publish(ctx context.Context, topic string, payload []byte, qo if c == nil { return ErrUnavailable } + if ctx == nil { + ctx = context.Background() + } if c != nil && strings.TrimSpace(topic) == "" { topic = c.config.RoutingKey } @@ -231,6 +234,9 @@ func (c *RabbitMQ) Subscribe(ctx context.Context, topic string, qos byte, handle if c == nil { return ErrUnavailable } + if ctx == nil { + ctx = context.Background() + } if c != nil && strings.TrimSpace(topic) == "" { topic = c.config.RoutingKey } @@ -331,6 +337,9 @@ func (c *RabbitMQ) Unsubscribe(ctx context.Context, topics ...string) error { if c == nil { return ErrUnavailable } + if ctx == nil { + ctx = context.Background() + } if len(topics) == 0 { return errors.New("rabbitmq routing keys are empty") }