From 171347ff18e97f959e00313e14a5a4e272883d10 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Mon, 7 Sep 2026 22:32:52 +0000 Subject: [PATCH] feat(distributed): release exact staged keys Request inputs need transport-neutral cleanup after backend calls. Add authenticated exact-key deletion for HTTP and coordinated cache eviction before shared-object deletion for S3/NATS. Preserve URL metacharacters as filename data, reject unsafe keys, and remove upload sidecars while pruning empty request directories. Assisted-by: Codex:gpt-6 --- core/services/messaging/subjects.go | 6 + core/services/nodes/file_stager.go | 26 ++- core/services/nodes/file_stager_http.go | 30 +++ .../nodes/file_stager_release_test.go | 174 ++++++++++++++++++ core/services/nodes/file_stager_s3.go | 32 ++++ core/services/nodes/file_transfer_server.go | 53 ++++++ core/services/nodes/router_test.go | 2 + core/services/worker/file_staging.go | 120 ++++++++++++ .../worker/file_staging_release_test.go | 135 ++++++++++++++ 9 files changed, 577 insertions(+), 1 deletion(-) create mode 100644 core/services/nodes/file_stager_release_test.go create mode 100644 core/services/worker/file_staging_release_test.go diff --git a/core/services/messaging/subjects.go b/core/services/messaging/subjects.go index c1f4cf8bf..c53a5d187 100644 --- a/core/services/messaging/subjects.go +++ b/core/services/messaging/subjects.go @@ -439,6 +439,12 @@ func SubjectNodeFilesStage(nodeID string) string { return subjectNodePrefix + sanitizeSubjectToken(nodeID) + ".files.stage" } +// SubjectNodeFilesRelease tells a serve-backend node to evict one ephemeral cache key. +// Reply: {error} +func SubjectNodeFilesRelease(nodeID string) string { + return subjectNodePrefix + sanitizeSubjectToken(nodeID) + ".files.release" +} + // SubjectNodeFilesTemp tells a serve-backend node to allocate a temp file. // Reply: {local_path, error} func SubjectNodeFilesTemp(nodeID string) string { diff --git a/core/services/nodes/file_stager.go b/core/services/nodes/file_stager.go index c5ee38556..6b18cd0ff 100644 --- a/core/services/nodes/file_stager.go +++ b/core/services/nodes/file_stager.go @@ -1,6 +1,11 @@ package nodes -import "context" +import ( + "context" + "fmt" + "path" + "strings" +) // FileStager abstracts file transfer between frontend and backend nodes // in distributed mode. Two implementations exist: @@ -29,7 +34,26 @@ type FileStager interface { // StageRemoteToStore uploads a remote file to shared storage. StageRemoteToStore(ctx context.Context, nodeID, remotePath, key string) error + // ReleaseRemote removes one ephemeral key from the remote node. + ReleaseRemote(ctx context.Context, nodeID, key string) error + // ListRemoteDir returns relative file paths within a directory on the remote node. // keyPrefix is a storage-style key prefix (e.g. "models/mymodel"). ListRemoteDir(ctx context.Context, nodeID, keyPrefix string) ([]string, error) } + +func validateEphemeralReleaseKey(key string) error { + if strings.Contains(key, "\\") || path.Clean(key) != key { + return fmt.Errorf("invalid ephemeral key %q", key) + } + parts := strings.Split(key, "/") + if len(parts) != 4 || parts[0] != "ephemeral" { + return fmt.Errorf("release key %q must identify one file below ephemeral/", key) + } + for _, part := range parts[1:] { + if part == "" || part == "." || part == ".." { + return fmt.Errorf("invalid ephemeral key %q", key) + } + } + return nil +} diff --git a/core/services/nodes/file_stager_http.go b/core/services/nodes/file_stager_http.go index 79047aad6..681eaa3d4 100644 --- a/core/services/nodes/file_stager_http.go +++ b/core/services/nodes/file_stager_http.go @@ -10,6 +10,7 @@ import ( "io" "net" "net/http" + "net/url" "os" "path/filepath" "strconv" @@ -81,6 +82,35 @@ func NewHTTPFileStager(httpAddrFor func(nodeID string) (string, error), token st } } +// ReleaseRemote removes one exact ephemeral key from a backend node. +func (h *HTTPFileStager) ReleaseRemote(ctx context.Context, nodeID, key string) error { + if err := validateEphemeralReleaseKey(key); err != nil { + return err + } + addr, err := h.httpAddrFor(nodeID) + if err != nil { + return fmt.Errorf("resolving HTTP address for node %s: %w", nodeID, err) + } + releaseURL := (&url.URL{Scheme: "http", Host: addr, Path: "/v1/files/" + key}).String() + req, err := http.NewRequestWithContext(ctx, http.MethodDelete, releaseURL, nil) + if err != nil { + return fmt.Errorf("creating release request for %q: %w", key, err) + } + if h.token != "" { + req.Header.Set("Authorization", "Bearer "+h.token) + } + resp, err := h.client.Do(req) + if err != nil { + return fmt.Errorf("releasing %q from node %s: %w", key, nodeID, err) + } + defer resp.Body.Close() + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return fmt.Errorf("releasing %q from node %s: status %d: %s", key, nodeID, resp.StatusCode, strings.TrimSpace(string(body))) + } + return nil +} + func (h *HTTPFileStager) EnsureRemote(ctx context.Context, nodeID, localPath, key string) (string, error) { xlog.Debug("Staging file to remote node via HTTP", "node", nodeID, "localPath", localPath, "key", key) diff --git a/core/services/nodes/file_stager_release_test.go b/core/services/nodes/file_stager_release_test.go new file mode 100644 index 000000000..9592fe5af --- /dev/null +++ b/core/services/nodes/file_stager_release_test.go @@ -0,0 +1,174 @@ +package nodes + +import ( + "context" + "encoding/json" + "net" + "os" + "path/filepath" + "strings" + "time" + + "github.com/mudler/LocalAI/core/services/messaging" + "github.com/mudler/LocalAI/core/services/storage" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +type releaseTestSubscription struct{} + +func (releaseTestSubscription) Unsubscribe() error { return nil } + +type releaseTestMessaging struct { + subject string + payload []byte + onRequest func() + requestCalled bool +} + +func (m *releaseTestMessaging) Publish(string, any) error { return nil } +func (m *releaseTestMessaging) Subscribe(string, func([]byte)) (messaging.Subscription, error) { + return releaseTestSubscription{}, nil +} +func (m *releaseTestMessaging) QueueSubscribe(string, string, func([]byte)) (messaging.Subscription, error) { + return releaseTestSubscription{}, nil +} +func (m *releaseTestMessaging) QueueSubscribeReply(string, string, func([]byte, func([]byte))) (messaging.Subscription, error) { + return releaseTestSubscription{}, nil +} +func (m *releaseTestMessaging) SubscribeReply(string, func([]byte, func([]byte))) (messaging.Subscription, error) { + return releaseTestSubscription{}, nil +} +func (m *releaseTestMessaging) Request(subject string, data []byte, _ time.Duration) ([]byte, error) { + m.subject = subject + m.payload = append([]byte(nil), data...) + m.requestCalled = true + if m.onRequest != nil { + m.onRequest() + } + return []byte(`{}`), nil +} +func (m *releaseTestMessaging) IsConnected() bool { return true } +func (m *releaseTestMessaging) Close() {} + +var _ = Describe("File stager exact-key release", func() { + startReleaseServer := func(stagingDir, token string) (*HTTPFileStager, func()) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + Expect(err).NotTo(HaveOccurred()) + server, err := StartFileTransferServerWithListener( + listener, + stagingDir, + GinkgoT().TempDir(), + GinkgoT().TempDir(), + token, + 0, + ) + Expect(err).NotTo(HaveOccurred()) + return NewHTTPFileStager(func(string) (string, error) { + return listener.Addr().String(), nil + }, token), func() { + Expect(server.Shutdown(context.Background())).To(Succeed()) + } + } + + It("transmits URL metacharacters as the exact key", func() { + stagingDir := GinkgoT().TempDir() + categoryDir := filepath.Join(stagingDir, "ephemeral", "request-id", "audio") + Expect(os.MkdirAll(categoryDir, 0750)).To(Succeed()) + + key := "ephemeral/request-id/audio/name ?#%2F.wav" + exactPath := filepath.Join(categoryDir, "name ?#%2F.wav") + wrongPath := filepath.Join(categoryDir, "name ") + for _, path := range []string{exactPath, exactPath + hashSidecarSuffix, exactPath + targetSidecarSuffix, wrongPath} { + Expect(os.WriteFile(path, []byte("data"), 0640)).To(Succeed()) + } + + stager, stop := startReleaseServer(stagingDir, "release-token") + DeferCleanup(stop) + Expect(stager.ReleaseRemote(context.Background(), "node-1", key)).To(Succeed()) + + Expect(exactPath).NotTo(BeAnExistingFile()) + Expect(exactPath + hashSidecarSuffix).NotTo(BeAnExistingFile()) + Expect(exactPath + targetSidecarSuffix).NotTo(BeAnExistingFile()) + Expect(wrongPath).To(BeAnExistingFile()) + }) + + It("is idempotent and prunes empty category and request directories", func() { + stagingDir := GinkgoT().TempDir() + path := filepath.Join(stagingDir, "ephemeral", "request-id", "audio", "input.wav") + Expect(os.MkdirAll(filepath.Dir(path), 0750)).To(Succeed()) + Expect(os.WriteFile(path, []byte("data"), 0640)).To(Succeed()) + + stager, stop := startReleaseServer(stagingDir, "release-token") + DeferCleanup(stop) + for range 2 { + Expect(stager.ReleaseRemote(context.Background(), "node-1", "ephemeral/request-id/audio/input.wav")).To(Succeed()) + } + + Expect(filepath.Join(stagingDir, "ephemeral", "request-id", "audio")).NotTo(BeADirectory()) + Expect(filepath.Join(stagingDir, "ephemeral", "request-id")).NotTo(BeADirectory()) + Expect(filepath.Join(stagingDir, "ephemeral")).To(BeADirectory()) + }) + + It("rejects non-ephemeral and traversing keys before making a request", func() { + resolved := false + stager := NewHTTPFileStager(func(string) (string, error) { + resolved = true + return "127.0.0.1:1", nil + }, "token") + + for _, key := range []string{ + "models/model.gguf", + "ephemeral/../models/model.gguf", + "ephemeral/request-id/../../model.gguf", + "/ephemeral/request-id/audio/input.wav", + } { + Expect(stager.ReleaseRemote(context.Background(), "node-1", key)).NotTo(Succeed(), key) + } + Expect(resolved).To(BeFalse()) + }) + + It("rejects symlink escapes", func() { + stagingDir := GinkgoT().TempDir() + outsideDir := GinkgoT().TempDir() + outsidePath := filepath.Join(outsideDir, "input.wav") + Expect(os.WriteFile(outsidePath, []byte("keep"), 0640)).To(Succeed()) + requestDir := filepath.Join(stagingDir, "ephemeral", "request-id") + Expect(os.MkdirAll(requestDir, 0750)).To(Succeed()) + Expect(os.Symlink(outsideDir, filepath.Join(requestDir, "audio"))).To(Succeed()) + + stager, stop := startReleaseServer(stagingDir, "release-token") + DeferCleanup(stop) + Expect(stager.ReleaseRemote(context.Background(), "node-1", "ephemeral/request-id/audio/input.wav")).NotTo(Succeed()) + Expect(outsidePath).To(BeAnExistingFile()) + }) + + It("evicts the worker cache before deleting the shared object", func() { + storeRoot := GinkgoT().TempDir() + cacheRoot := GinkgoT().TempDir() + store, err := storage.NewFilesystemStore(storeRoot) + Expect(err).NotTo(HaveOccurred()) + fm, err := storage.NewFileManager(store, cacheRoot) + Expect(err).NotTo(HaveOccurred()) + key := "ephemeral/request-id/audio/input.wav" + Expect(store.Put(context.Background(), key, strings.NewReader("shared"))).To(Succeed()) + + client := &releaseTestMessaging{} + client.onRequest = func() { + exists, existsErr := store.Exists(context.Background(), key) + Expect(existsErr).NotTo(HaveOccurred()) + Expect(exists).To(BeTrue()) + } + stager := NewS3NATSFileStager(fm, client) + Expect(stager.ReleaseRemote(context.Background(), "node.one", key)).To(Succeed()) + + Expect(client.requestCalled).To(BeTrue()) + Expect(client.subject).To(Equal(messaging.SubjectNodeFilesRelease("node.one"))) + var payload fileReleaseRequest + Expect(json.Unmarshal(client.payload, &payload)).To(Succeed()) + Expect(payload.Key).To(Equal(key)) + exists, err := store.Exists(context.Background(), key) + Expect(err).NotTo(HaveOccurred()) + Expect(exists).To(BeFalse()) + }) +}) diff --git a/core/services/nodes/file_stager_s3.go b/core/services/nodes/file_stager_s3.go index 0d3847b7c..e63e0a629 100644 --- a/core/services/nodes/file_stager_s3.go +++ b/core/services/nodes/file_stager_s3.go @@ -48,6 +48,14 @@ type fileStageReply struct { Error string `json:"error,omitempty"` } +type fileReleaseRequest struct { + Key string `json:"key"` +} + +type fileReleaseReply struct { + Error string `json:"error,omitempty"` +} + type fileTempRequest struct{} type fileTempReply struct { @@ -181,3 +189,27 @@ func (s *S3NATSFileStager) StageRemoteToStore(ctx context.Context, nodeID, remot return nil } + +// ReleaseRemote evicts one exact ephemeral key from the worker before deleting +// the shared object. +func (s *S3NATSFileStager) ReleaseRemote(ctx context.Context, nodeID, key string) error { + if err := validateEphemeralReleaseKey(key); err != nil { + return err + } + reply, err := messaging.RequestJSON[fileReleaseRequest, fileReleaseReply]( + s.nats, + messaging.SubjectNodeFilesRelease(nodeID), + fileReleaseRequest{Key: key}, + 30*time.Second, + ) + if err != nil { + return err + } + if reply.Error != "" { + return fmt.Errorf("backend release failed: %s", reply.Error) + } + if err := s.fm.Delete(ctx, key); err != nil { + return fmt.Errorf("deleting shared object %q: %w", key, err) + } + return nil +} diff --git a/core/services/nodes/file_transfer_server.go b/core/services/nodes/file_transfer_server.go index 49cf04549..2d8feff76 100644 --- a/core/services/nodes/file_transfer_server.go +++ b/core/services/nodes/file_transfer_server.go @@ -119,6 +119,8 @@ func StartFileTransferServerWithReadiness(lis net.Listener, stagingDir, modelsDi handleUpload(w, r, stagingDir, modelsDir, dataDir, key, maxUploadSize) case http.MethodGet: handleDownload(w, r, stagingDir, modelsDir, dataDir, key) + case http.MethodDelete: + handleRelease(w, r, stagingDir, key) case http.MethodPost: if key == "temp" { handleAllocTemp(w, r, stagingDir) @@ -182,6 +184,57 @@ func StartFileTransferServerWithReadiness(lis net.Listener, stagingDir, modelsDi return server, nil } +func handleRelease(w http.ResponseWriter, _ *http.Request, stagingDir, key string) { + if err := validateEphemeralReleaseKey(key); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + filePath := filepath.Join(stagingDir, filepath.FromSlash(key)) + if err := validatePathInDir(filePath, stagingDir); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if info, err := os.Lstat(filePath); err == nil && info.IsDir() { + http.Error(w, "release key identifies a directory", http.StatusBadRequest) + return + } else if err != nil && !os.IsNotExist(err) { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + for _, path := range []string{filePath, filePath + hashSidecarSuffix, filePath + targetSidecarSuffix} { + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + } + categoryDir := filepath.Dir(filePath) + requestDir := filepath.Dir(categoryDir) + for _, dir := range []string{categoryDir, requestDir} { + if err := pruneEmptyDir(dir); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + } + w.WriteHeader(http.StatusNoContent) +} + +func pruneEmptyDir(dir string) error { + entries, err := os.ReadDir(dir) + if os.IsNotExist(err) { + return nil + } + if err != nil { + return err + } + if len(entries) != 0 { + return nil + } + if err := os.Remove(dir); err != nil && !os.IsNotExist(err) { + return err + } + return nil +} + func handleHead(w http.ResponseWriter, r *http.Request, stagingDir, modelsDir, dataDir, key string) { if key == "" { http.Error(w, "key is required", http.StatusBadRequest) diff --git a/core/services/nodes/router_test.go b/core/services/nodes/router_test.go index 10c646329..c6140bc1b 100644 --- a/core/services/nodes/router_test.go +++ b/core/services/nodes/router_test.go @@ -52,6 +52,8 @@ func (f *fakeFileStager) AllocRemoteTemp(_ context.Context, _ string) (string, e func (f *fakeFileStager) StageRemoteToStore(_ context.Context, _, _, _ string) error { return nil } +func (f *fakeFileStager) ReleaseRemote(_ context.Context, _, _ string) error { return nil } + func (f *fakeFileStager) ListRemoteDir(_ context.Context, _, _ string) ([]string, error) { return nil, nil } diff --git a/core/services/worker/file_staging.go b/core/services/worker/file_staging.go index 019afcba9..6293551fc 100644 --- a/core/services/worker/file_staging.go +++ b/core/services/worker/file_staging.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "os" + "path" "path/filepath" "strings" @@ -57,6 +58,9 @@ func (cfg *Config) subscribeFileStaging(natsClient messaging.MessagingClient, no if err != nil { return fmt.Errorf("initializing file manager: %w", err) } + if err := subscribeFileRelease(natsClient, nodeID, fm, cacheDir); err != nil { + return err + } // Subscribe: files.ensure — download S3 key to local, reply with local path if _, err := natsClient.SubscribeReply(messaging.SubjectNodeFilesEnsure(nodeID), func(data []byte, reply func([]byte)) { @@ -199,3 +203,119 @@ func (cfg *Config) subscribeFileStaging(natsClient messaging.MessagingClient, no xlog.Info("Subscribed to file staging NATS subjects", "nodeID", nodeID) return nil } + +func subscribeFileRelease(natsClient messaging.MessagingClient, nodeID string, fm *storage.FileManager, cacheDir string) error { + if _, err := natsClient.SubscribeReply(messaging.SubjectNodeFilesRelease(nodeID), func(data []byte, reply func([]byte)) { + var req struct { + Key string `json:"key"` + } + if err := json.Unmarshal(data, &req); err != nil { + replyJSON(reply, map[string]string{"error": "invalid request"}) + return + } + cachePath, err := fm.CachePath(req.Key) + if err == nil { + err = releaseEphemeralCachePath(cacheDir, req.Key, cachePath) + } + if err != nil { + replyJSON(reply, map[string]string{"error": err.Error()}) + return + } + replyJSON(reply, map[string]string{}) + }); err != nil { + return fmt.Errorf("subscribing to files.release events: %w", err) + } + return nil +} + +func releaseEphemeralCacheKey(cacheDir, key string) error { + return releaseEphemeralCachePath(cacheDir, key, filepath.Join(cacheDir, filepath.FromSlash(key))) +} + +func releaseEphemeralCachePath(cacheDir, key, filePath string) error { + if err := validateEphemeralCacheKey(key); err != nil { + return err + } + if err := validateReleasePath(filePath, cacheDir); err != nil { + return err + } + if info, err := os.Lstat(filePath); err == nil && info.IsDir() { + return fmt.Errorf("release key identifies a directory") + } else if err != nil && !os.IsNotExist(err) { + return err + } + for _, path := range []string{filePath, filePath + ".sha256", filePath + ".sha256.target"} { + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + return err + } + } + for _, dir := range []string{filepath.Dir(filePath), filepath.Dir(filepath.Dir(filePath))} { + entries, err := os.ReadDir(dir) + if os.IsNotExist(err) { + continue + } + if err != nil { + return err + } + if len(entries) == 0 { + if err := os.Remove(dir); err != nil && !os.IsNotExist(err) { + return err + } + } + } + return nil +} + +func validateEphemeralCacheKey(key string) error { + if strings.Contains(key, "\\") || path.Clean(key) != key { + return fmt.Errorf("invalid ephemeral key %q", key) + } + parts := strings.Split(key, "/") + if len(parts) != 4 || parts[0] != "ephemeral" { + return fmt.Errorf("release key %q must identify one file below ephemeral/", key) + } + for _, part := range parts[1:] { + if part == "" || part == "." || part == ".." { + return fmt.Errorf("invalid ephemeral key %q", key) + } + } + return nil +} + +func validateReleasePath(targetPath, baseDir string) error { + absBase, err := filepath.Abs(baseDir) + if err != nil { + return err + } + realBase, err := filepath.EvalSymlinks(absBase) + if err != nil { + return err + } + absTarget, err := filepath.Abs(targetPath) + if err != nil { + return err + } + realTarget, err := filepath.EvalSymlinks(absTarget) + if err != nil { + remaining := filepath.Base(absTarget) + dir := filepath.Dir(absTarget) + for { + resolved, resolveErr := filepath.EvalSymlinks(dir) + if resolveErr == nil { + realTarget = filepath.Join(resolved, remaining) + break + } + remaining = filepath.Join(filepath.Base(dir), remaining) + parent := filepath.Dir(dir) + if parent == dir { + realTarget = filepath.Clean(absTarget) + break + } + dir = parent + } + } + if realTarget != realBase && !strings.HasPrefix(realTarget, realBase+string(filepath.Separator)) { + return fmt.Errorf("path %q resolves outside ephemeral cache", targetPath) + } + return nil +} diff --git a/core/services/worker/file_staging_release_test.go b/core/services/worker/file_staging_release_test.go new file mode 100644 index 000000000..969b738bf --- /dev/null +++ b/core/services/worker/file_staging_release_test.go @@ -0,0 +1,135 @@ +package worker + +import ( + "encoding/json" + "os" + "path/filepath" + "time" + + "github.com/mudler/LocalAI/core/services/messaging" + "github.com/mudler/LocalAI/core/services/storage" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +type releaseSubscription struct{} + +func (releaseSubscription) Unsubscribe() error { return nil } + +type releaseMessagingClient struct { + subject string + handler func([]byte, func([]byte)) +} + +func (m *releaseMessagingClient) Publish(string, any) error { return nil } +func (m *releaseMessagingClient) Subscribe(string, func([]byte)) (messaging.Subscription, error) { + return releaseSubscription{}, nil +} +func (m *releaseMessagingClient) QueueSubscribe(string, string, func([]byte)) (messaging.Subscription, error) { + return releaseSubscription{}, nil +} +func (m *releaseMessagingClient) QueueSubscribeReply(string, string, func([]byte, func([]byte))) (messaging.Subscription, error) { + return releaseSubscription{}, nil +} +func (m *releaseMessagingClient) SubscribeReply(subject string, handler func([]byte, func([]byte))) (messaging.Subscription, error) { + m.subject = subject + m.handler = handler + return releaseSubscription{}, nil +} +func (m *releaseMessagingClient) Request(string, []byte, time.Duration) ([]byte, error) { + return nil, nil +} +func (m *releaseMessagingClient) IsConnected() bool { return true } +func (m *releaseMessagingClient) Close() {} + +var _ = Describe("Worker exact-key staging release", func() { + It("removes only the exact cache file and upload sidecars", func() { + cacheDir := GinkgoT().TempDir() + categoryDir := filepath.Join(cacheDir, "ephemeral", "request-id", "audio") + Expect(os.MkdirAll(categoryDir, 0750)).To(Succeed()) + target := filepath.Join(categoryDir, "input.wav") + sibling := filepath.Join(categoryDir, "keep.wav") + for _, path := range []string{target, target + ".sha256", target + ".sha256.target", sibling} { + Expect(os.WriteFile(path, []byte("data"), 0640)).To(Succeed()) + } + + Expect(releaseEphemeralCacheKey(cacheDir, "ephemeral/request-id/audio/input.wav")).To(Succeed()) + Expect(target).NotTo(BeAnExistingFile()) + Expect(target + ".sha256").NotTo(BeAnExistingFile()) + Expect(target + ".sha256.target").NotTo(BeAnExistingFile()) + Expect(sibling).To(BeAnExistingFile()) + Expect(categoryDir).To(BeADirectory()) + }) + + It("succeeds for a missing file and prunes empty category and request directories", func() { + cacheDir := GinkgoT().TempDir() + categoryDir := filepath.Join(cacheDir, "ephemeral", "request-id", "audio") + Expect(os.MkdirAll(categoryDir, 0750)).To(Succeed()) + + for range 2 { + Expect(releaseEphemeralCacheKey(cacheDir, "ephemeral/request-id/audio/missing.wav")).To(Succeed()) + } + Expect(categoryDir).NotTo(BeADirectory()) + Expect(filepath.Dir(categoryDir)).NotTo(BeADirectory()) + Expect(filepath.Join(cacheDir, "ephemeral")).To(BeADirectory()) + }) + + It("rejects traversal and symlink escapes", func() { + cacheDir := GinkgoT().TempDir() + outsideDir := GinkgoT().TempDir() + outsidePath := filepath.Join(outsideDir, "input.wav") + Expect(os.WriteFile(outsidePath, []byte("keep"), 0640)).To(Succeed()) + requestDir := filepath.Join(cacheDir, "ephemeral", "request-id") + Expect(os.MkdirAll(requestDir, 0750)).To(Succeed()) + Expect(os.Symlink(outsideDir, filepath.Join(requestDir, "audio"))).To(Succeed()) + + for _, key := range []string{ + "models/model.gguf", + "ephemeral/../models/model.gguf", + "ephemeral/request-id/audio/../../model.gguf", + "ephemeral/request-id/audio/input.wav", + } { + Expect(releaseEphemeralCacheKey(cacheDir, key)).NotTo(Succeed(), key) + } + Expect(outsidePath).To(BeAnExistingFile()) + }) + + It("registers an exact release handler", func() { + cacheDir := GinkgoT().TempDir() + path := filepath.Join(cacheDir, "ephemeral", "request-id", "audio", "input.wav") + Expect(os.MkdirAll(filepath.Dir(path), 0750)).To(Succeed()) + Expect(os.WriteFile(path, []byte("data"), 0640)).To(Succeed()) + fm, err := storage.NewFileManager(nil, cacheDir) + Expect(err).NotTo(HaveOccurred()) + client := &releaseMessagingClient{} + + Expect(subscribeFileRelease(client, "node.one", fm, cacheDir)).To(Succeed()) + Expect(client.subject).To(Equal(messaging.SubjectNodeFilesRelease("node.one"))) + request, err := json.Marshal(map[string]string{"key": "ephemeral/request-id/audio/input.wav"}) + Expect(err).NotTo(HaveOccurred()) + var response []byte + client.handler(request, func(data []byte) { response = append([]byte(nil), data...) }) + + var reply map[string]string + Expect(json.Unmarshal(response, &reply)).To(Succeed()) + Expect(reply["error"]).To(BeEmpty()) + Expect(path).NotTo(BeAnExistingFile()) + }) + + It("returns validation errors through the release handler", func() { + cacheDir := GinkgoT().TempDir() + fm, err := storage.NewFileManager(nil, cacheDir) + Expect(err).NotTo(HaveOccurred()) + client := &releaseMessagingClient{} + Expect(subscribeFileRelease(client, "node-1", fm, cacheDir)).To(Succeed()) + + request, err := json.Marshal(map[string]string{"key": "models/model.gguf"}) + Expect(err).NotTo(HaveOccurred()) + var response []byte + client.handler(request, func(data []byte) { response = append([]byte(nil), data...) }) + + var reply map[string]string + Expect(json.Unmarshal(response, &reply)).To(Succeed()) + Expect(reply["error"]).NotTo(BeEmpty()) + }) +})