From 7c13a446ac905d56383c3a992af07f89604b9384 Mon Sep 17 00:00:00 2001 From: nfebe Date: Fri, 17 Jul 2026 13:07:05 +0100 Subject: [PATCH] fix(observ): Address the observability review follow-ups Notification targets no longer expose their credentials. A target's URL carries its secret inline (an SMTP password, a webhook token), and it was returned in full by the settings API and written to a world-readable file. The URL is now masked in API responses and the file is owner-only, and saving a target back with a masked URL keeps the stored secret rather than overwriting it. The health watcher no longer blocks health reads while it restarts a container: the restart, a blocking operation, now runs without the lock that those reads need. Turning auto-restart on or off now works from an assistant session that is not scoped to a single deployment. It is a host-wide setting, so it is gated on permission to change settings instead of demanding a deployment it cannot name. --- internal/api/ai_tools.go | 50 +++++++---- internal/api/notifications_test.go | 137 +++++++++++++++++++++++++++++ internal/api/server.go | 2 +- internal/notify/notify.go | 38 +++++++- internal/notify/notify_test.go | 37 ++++++++ internal/observ/health.go | 45 +++++++--- internal/observ/health_test.go | 26 ++++++ internal/observ/tools.go | 1 + pkg/pluginapi/pluginapi.go | 6 +- 9 files changed, 310 insertions(+), 32 deletions(-) create mode 100644 internal/api/notifications_test.go diff --git a/internal/api/ai_tools.go b/internal/api/ai_tools.go index 6d51e45..ed63240 100644 --- a/internal/api/ai_tools.go +++ b/internal/api/ai_tools.go @@ -400,20 +400,19 @@ func (s *Server) pluginInfos() []pluginapi.Info { return s.pluginHost.Infos() } -// pluginToolMutates reports whether a namespaced plugin tool changes state, so the caller -// applies the write-access gate. -func (s *Server) pluginToolMutates(plugin, tool string) bool { +// pluginToolSpec returns the advertised spec for a namespaced plugin tool. +func (s *Server) pluginToolSpec(plugin, tool string) (pluginapi.ToolSpec, bool) { for _, info := range s.pluginInfos() { if info.Name != plugin { continue } for _, t := range info.Tools { if t.Name == tool { - return t.Mutates + return t, true } } } - return false + return pluginapi.ToolSpec{}, false } // runAITool executes one tool call, returning the textual result the @@ -429,18 +428,26 @@ func (s *Server) runAITool(c *gin.Context, boundDeployment string, call ai.ToolC // Plugin-provided tools are dispatched to the owning plugin over its socket. if plugin, tool, ok := parsePluginToolName(call.Name); ok { - if s.pluginToolMutates(plugin, tool) { - // A state-changing tool requires write access to a deployment, and the plugin is - // told which one so it can scope the mutation to that deployment rather than act - // on an arbitrary resource the caller named. - dep, err := s.toolAllowedDeploymentWrite(c, boundDeployment, args) - if err != nil { - return "Error: " + err.Error() - } - if args == nil { - args = map[string]interface{}{} + if spec, found := s.pluginToolSpec(plugin, tool); found && spec.Mutates { + if spec.Global { + // A host-wide setting has no deployment scope; gate it on settings-write so + // it works in an unscoped assistant session instead of demanding a deployment. + if err := s.toolAllowedGlobalWrite(c); err != nil { + return "Error: " + err.Error() + } + } else { + // A state-changing tool requires write access to a deployment, and the plugin is + // told which one so it can scope the mutation to that deployment rather than act + // on an arbitrary resource the caller named. + dep, err := s.toolAllowedDeploymentWrite(c, boundDeployment, args) + if err != nil { + return "Error: " + err.Error() + } + if args == nil { + args = map[string]interface{}{} + } + args["_deployment"] = dep } - args["_deployment"] = dep } result, err := s.pluginHost.ExecTool(plugin, tool, args) if err != nil { @@ -473,3 +480,14 @@ func (s *Server) toolAllowedDeploymentWrite(c *gin.Context, boundDeployment stri } return name, nil } + +// toolAllowedGlobalWrite gates a mutating tool that changes a host-wide setting rather than a +// deployment. It needs settings-write; a nil actor means auth is disabled, which is allowed as +// elsewhere. +func (s *Server) toolAllowedGlobalWrite(c *gin.Context) error { + actor := auth.GetActorFromContext(c) + if actor != nil && !actor.HasPermission(auth.PermSettingsWrite) { + return fmt.Errorf("you do not have permission to change this setting") + } + return nil +} diff --git a/internal/api/notifications_test.go b/internal/api/notifications_test.go new file mode 100644 index 0000000..60b927c --- /dev/null +++ b/internal/api/notifications_test.go @@ -0,0 +1,137 @@ +package api + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/flatrun/agent/internal/auth" + "github.com/flatrun/agent/internal/contextkeys" + "github.com/flatrun/agent/internal/notify" + "github.com/gin-gonic/gin" +) + +func setupNotifyTest(t *testing.T) (*Server, *gin.Engine) { + t.Helper() + gin.SetMode(gin.TestMode) + s := &Server{ + notify: notify.NewService(t.TempDir()), + pluginToken: "plugin-token", + } + r := gin.New() + r.GET("/notifications/targets", s.getNotificationTargets) + r.PUT("/notifications/targets", s.updateNotificationTargets) + r.POST("/internal/notify/emit", s.emitNotification) + return s, r +} + +func TestGetNotificationTargetsMasksSecret(t *testing.T) { + s, r := setupNotifyTest(t) + if err := s.notify.Save(notify.Config{Targets: []notify.Target{ + {ID: "1", Name: "email", URL: "smtp://user:secret@host:587", Enabled: true}, + }}); err != nil { + t.Fatal(err) + } + + w := httptest.NewRecorder() + r.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/notifications/targets", nil)) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d", w.Code) + } + body := w.Body.String() + if strings.Contains(body, "secret") { + t.Errorf("response leaked the target credential: %s", body) + } + if !strings.Contains(body, notify.MaskedURL) { + t.Errorf("response should mask the target URL: %s", body) + } +} + +func TestUpdateNotificationTargetsPreservesMaskedSecret(t *testing.T) { + s, r := setupNotifyTest(t) + const real = "smtp://user:secret@host:587" + if err := s.notify.Save(notify.Config{Targets: []notify.Target{ + {ID: "1", Name: "email", URL: real, Enabled: true}, + }}); err != nil { + t.Fatal(err) + } + + payload, _ := json.Marshal(notify.Config{Targets: []notify.Target{ + {ID: "1", Name: "renamed", URL: notify.MaskedURL, Enabled: true}, + }}) + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPut, "/notifications/targets", bytes.NewReader(payload)) + req.Header.Set("Content-Type", "application/json") + r.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status = %d, body = %s", w.Code, w.Body.String()) + } + stored := s.notify.Load() + if len(stored.Targets) != 1 || stored.Targets[0].URL != real { + t.Errorf("saving a masked URL must keep the stored secret, got %+v", stored.Targets) + } + if stored.Targets[0].Name != "renamed" { + t.Errorf("name change should apply, got %q", stored.Targets[0].Name) + } +} + +func TestEmitNotificationRequiresPluginToken(t *testing.T) { + _, r := setupNotifyTest(t) + body := func() *bytes.Reader { return bytes.NewReader([]byte(`{"title":"t","message":"m"}`)) } + + cases := []struct { + name string + token string + status int + }{ + {"no token", "", http.StatusUnauthorized}, + {"wrong token", "nope", http.StatusUnauthorized}, + {"correct token", "plugin-token", http.StatusAccepted}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + w := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/internal/notify/emit", body()) + req.Header.Set("Content-Type", "application/json") + if tc.token != "" { + req.Header.Set("X-Plugin-Token", tc.token) + } + r.ServeHTTP(w, req) + if w.Code != tc.status { + t.Errorf("status = %d, want %d", w.Code, tc.status) + } + }) + } +} + +func TestToolAllowedGlobalWrite(t *testing.T) { + s := &Server{} + ctx := func(actor *auth.ActorContext) *gin.Context { + c, _ := gin.CreateTestContext(httptest.NewRecorder()) + if actor != nil { + c.Set(contextkeys.Actor, actor) + } + return c + } + + if err := s.toolAllowedGlobalWrite(ctx(&auth.ActorContext{Role: auth.RoleAdmin})); err != nil { + t.Errorf("admin should be allowed: %v", err) + } + if err := s.toolAllowedGlobalWrite(ctx(&auth.ActorContext{ + Role: auth.RoleViewer, + Permissions: []string{string(auth.PermSettingsWrite)}, + })); err != nil { + t.Errorf("actor with settings-write should be allowed: %v", err) + } + if err := s.toolAllowedGlobalWrite(ctx(&auth.ActorContext{Role: auth.RoleViewer})); err == nil { + t.Error("actor without settings-write should be denied") + } + if err := s.toolAllowedGlobalWrite(ctx(nil)); err != nil { + t.Errorf("nil actor (auth disabled) should be allowed: %v", err) + } +} diff --git a/internal/api/server.go b/internal/api/server.go index c3b227b..39213f0 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -3106,7 +3106,7 @@ func (s *Server) updateNotificationTargets(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) return } - if err := s.notify.Save(cfg); err != nil { + if err := s.notify.Update(cfg); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } diff --git a/internal/notify/notify.go b/internal/notify/notify.go index 50d0cd7..fd6e6fc 100644 --- a/internal/notify/notify.go +++ b/internal/notify/notify.go @@ -5,6 +5,7 @@ package notify import ( + "encoding/json" "fmt" "os" "path/filepath" @@ -14,6 +15,11 @@ import ( "gopkg.in/yaml.v3" ) +// MaskedURL stands in for a target URL in API responses. A shoutrrr URL carries +// its credentials inline (an SMTP password, a webhook token), so the real URL is +// never returned to a client; it is written verbatim only to the on-disk store. +const MaskedURL = "********" + // Target is one delivery destination. URL is a shoutrrr service URL, e.g. // "smtp://user:pass@host:587/?from=x&to=y" or "generic+https://example.com/hook". type Target struct { @@ -23,6 +29,17 @@ type Target struct { Enabled bool `yaml:"enabled" json:"enabled"` } +// MarshalJSON masks the credential-bearing URL. YAML persistence does not use +// this path, so the stored file keeps the real URL. +func (t Target) MarshalJSON() ([]byte, error) { + type alias Target + masked := alias(t) + if masked.URL != "" { + masked.URL = MaskedURL + } + return json.Marshal(masked) +} + // Config is the persisted notification settings. type Config struct { Targets []Target `yaml:"targets" json:"targets"` @@ -57,14 +74,31 @@ func (s *Service) Load() Config { func (s *Service) Save(cfg Config) error { s.mu.Lock() defer s.mu.Unlock() - if err := os.MkdirAll(filepath.Dir(s.path), 0755); err != nil { + if err := os.MkdirAll(filepath.Dir(s.path), 0700); err != nil { return err } data, err := yaml.Marshal(cfg) if err != nil { return err } - return os.WriteFile(s.path, data, 0644) + return os.WriteFile(s.path, data, 0600) +} + +// Update saves targets, restoring the stored URL for any target whose incoming +// URL is the mask: a client that received a masked target and saved it back +// unchanged must not overwrite the real URL with the mask. +func (s *Service) Update(cfg Config) error { + stored := s.Load() + byID := make(map[string]string, len(stored.Targets)) + for _, t := range stored.Targets { + byID[t.ID] = t.URL + } + for i := range cfg.Targets { + if cfg.Targets[i].URL == MaskedURL { + cfg.Targets[i].URL = byID[cfg.Targets[i].ID] + } + } + return s.Save(cfg) } // Test sends a message to a single URL, so an admin can verify a target before saving it. diff --git a/internal/notify/notify_test.go b/internal/notify/notify_test.go index 38f0374..aa8ec2e 100644 --- a/internal/notify/notify_test.go +++ b/internal/notify/notify_test.go @@ -1,10 +1,47 @@ package notify import ( + "encoding/json" "errors" + "strings" "testing" ) +func TestTargetJSONMasksURL(t *testing.T) { + b, err := json.Marshal(Target{ID: "1", Name: "email", URL: "smtp://user:secret@host:587", Enabled: true}) + if err != nil { + t.Fatal(err) + } + out := string(b) + if strings.Contains(out, "secret") { + t.Errorf("credential leaked in JSON: %s", out) + } + if !strings.Contains(out, MaskedURL) { + t.Errorf("URL should be masked: %s", out) + } +} + +func TestUpdatePreservesMaskedURL(t *testing.T) { + s := NewService(t.TempDir()) + const real = "smtp://user:secret@host:587" + if err := s.Save(Config{Targets: []Target{{ID: "1", Name: "email", URL: real, Enabled: true}}}); err != nil { + t.Fatal(err) + } + + // A client that received a masked URL saves it back with only a name change. + if err := s.Update(Config{Targets: []Target{{ID: "1", Name: "renamed", URL: MaskedURL, Enabled: true}}}); err != nil { + t.Fatal(err) + } + + got := s.Load() + if len(got.Targets) != 1 || got.Targets[0].URL != real { + t.Errorf("masked update must keep the stored URL, got %+v", got.Targets) + } + if got.Targets[0].Name != "renamed" { + t.Errorf("non-secret changes should still apply, got name %q", got.Targets[0].Name) + } +} + func TestServiceRoundTripAndNotify(t *testing.T) { s := NewService(t.TempDir()) diff --git a/internal/observ/health.go b/internal/observ/health.go index 3aa7441..773adb5 100644 --- a/internal/observ/health.go +++ b/internal/observ/health.go @@ -132,14 +132,21 @@ func NewHealthWatcher(source HealthSource, restart RestartFunc, interval, cooldo } // checkOnce reads health once and restarts any unhealthy container past its cooldown. +// The restart call is a blocking docker operation, so it runs without the lock held: +// otherwise a health read (Snapshot/Events) would stall for the whole restart. The +// decision is taken under the lock, the restarts run unlocked, and the results are +// recorded under the lock again. checkOnce is only ever called from the single Run +// goroutine, so there is no concurrent writer to race with between those phases. func (w *HealthWatcher) checkOnce() { states, err := w.source() if err != nil { return } - w.mu.Lock() - defer w.mu.Unlock() + var toRestart []ContainerHealth + var exhaustedEvents []ExhaustedEvent + + w.mu.Lock() w.health = make(map[string]ContainerHealth, len(states)) for _, s := range states { w.health[s.Container] = s @@ -161,14 +168,12 @@ func (w *HealthWatcher) checkOnce() { if w.attempts[s.Container] >= maxRestartAttempts { if !w.exhausted[s.Container] { w.exhausted[s.Container] = true - if w.onExhausted != nil { - go w.onExhausted(ExhaustedEvent{ - Container: s.Container, - Deployment: s.Deployment, - Attempts: w.attempts[s.Container], - At: w.now(), - }) - } + exhaustedEvents = append(exhaustedEvents, ExhaustedEvent{ + Container: s.Container, + Deployment: s.Deployment, + Attempts: w.attempts[s.Container], + At: w.now(), + }) } continue } @@ -176,16 +181,32 @@ func (w *HealthWatcher) checkOnce() { if seen && w.now().Sub(last) < w.cooldown { continue } + toRestart = append(toRestart, s) + } + w.mu.Unlock() + + if w.onExhausted != nil { + for _, ev := range exhaustedEvents { + go w.onExhausted(ev) + } + } + + for _, s := range toRestart { if err := w.restart(s.Container); err != nil { continue } - w.lastHeal[s.Container] = w.now() + now := w.now() + ev := RecoveryEvent{Container: s.Container, Deployment: s.Deployment, At: now} + + w.mu.Lock() + w.lastHeal[s.Container] = now w.attempts[s.Container]++ - ev := RecoveryEvent{Container: s.Container, Deployment: s.Deployment, At: w.now()} w.events = append(w.events, ev) if len(w.events) > maxRecoveryEvents { w.events = w.events[len(w.events)-maxRecoveryEvents:] } + w.mu.Unlock() + if w.onRecover != nil { go w.onRecover(ev) } diff --git a/internal/observ/health_test.go b/internal/observ/health_test.go index 9d8309d..6fc0ab1 100644 --- a/internal/observ/health_test.go +++ b/internal/observ/health_test.go @@ -5,6 +5,32 @@ import ( "time" ) +// A restart is a blocking docker call; health reads must not stall for its +// duration, so checkOnce must not hold the lock across it. +func TestCheckOnceDoesNotBlockReadsDuringRestart(t *testing.T) { + states := []ContainerHealth{{Container: "web", Deployment: "d", Status: HealthUnhealthy}} + started := make(chan struct{}) + release := make(chan struct{}) + w := NewHealthWatcher( + func() ([]ContainerHealth, error) { return states, nil }, + func(string) error { close(started); <-release; return nil }, + time.Second, time.Minute, + ) + + go w.checkOnce() + <-started // a restart is now in progress + + done := make(chan struct{}) + go func() { w.Snapshot(); close(done) }() + select { + case <-done: + case <-time.After(2 * time.Second): + close(release) + t.Fatal("Snapshot blocked while a restart was in progress") + } + close(release) +} + func TestHealthWatcherRestartsUnhealthy(t *testing.T) { states := []ContainerHealth{ {Container: "shop-web", Deployment: "shop", Status: HealthUnhealthy}, diff --git a/internal/observ/tools.go b/internal/observ/tools.go index aad94cf..cf0c47c 100644 --- a/internal/observ/tools.go +++ b/internal/observ/tools.go @@ -98,6 +98,7 @@ func buildTools(store *Store, watcher *HealthWatcher, cfgStore *ConfigStore) []p "required": []string{"enabled"}, }, Mutates: true, + Global: true, }, Run: func(args map[string]any) (string, error) { enabled, _ := args["enabled"].(bool) diff --git a/pkg/pluginapi/pluginapi.go b/pkg/pluginapi/pluginapi.go index da192ac..25d0591 100644 --- a/pkg/pluginapi/pluginapi.go +++ b/pkg/pluginapi/pluginapi.go @@ -37,12 +37,16 @@ type Info struct { // ToolSpec declares a tool the plugin exposes to the AI assistant. The host advertises it to // the model and dispatches calls back to the plugin's /_plugin/tools/exec endpoint. Mutates -// marks a tool that changes state, so the host can gate it on write access. +// marks a tool that changes state, so the host can gate it on write access. Global marks a +// mutating tool that acts on a host-wide setting rather than a single deployment, so the host +// gates it on a settings-write permission instead of requiring a deployment scope (which such +// a tool has no way to provide, and which would make it fail in an unscoped assistant session). type ToolSpec struct { Name string `json:"name"` Description string `json:"description"` Parameters map[string]any `json:"parameters,omitempty"` Mutates bool `json:"mutates,omitempty"` + Global bool `json:"global,omitempty"` } // ToolExecPath is where the host POSTs {name, args} to invoke a plugin tool.