From c21203b7eabcec4a34d074ef92fab8314da7b5f6 Mon Sep 17 00:00:00 2001 From: jenkins Date: Sat, 15 Aug 2026 00:34:08 -0300 Subject: [PATCH] hermes: deliver generated media to Telegram --- knowledge/catalog/atlas.json | 8 +- knowledge/catalog/atlas.yaml | 6 +- services/comms/knowledge/catalog/atlas.json | 8 +- services/comms/knowledge/catalog/atlas.yaml | 6 +- services/hermes/chat-router.yaml | 2 +- services/hermes/chat-statefulset.yaml | 36 ++ services/hermes/kustomization.yaml | 1 + services/hermes/networkpolicy.yaml | 2 + services/hermes/router/main.go | 20 +- services/hermes/router/telegram.go | 4 +- services/hermes/router/telegram_media.go | 328 ++++++++++++++ services/hermes/router/telegram_media_test.go | 424 ++++++++++++++++++ .../hermes/scripts/telegram_media_server.py | 190 ++++++++ services/hermes/service.yaml | 4 + .../test_hermes_telegram_media_server.py | 130 ++++++ 15 files changed, 1155 insertions(+), 14 deletions(-) create mode 100644 services/hermes/router/telegram_media.go create mode 100644 services/hermes/router/telegram_media_test.go create mode 100644 services/hermes/scripts/telegram_media_server.py create mode 100644 testing/tests/test_hermes_telegram_media_server.py diff --git a/knowledge/catalog/atlas.json b/knowledge/catalog/atlas.json index d88e6d2b..6fcdd018 100644 --- a/knowledge/catalog/atlas.json +++ b/knowledge/catalog/atlas.json @@ -930,7 +930,7 @@ "serviceAccountName": "hermes-chat", "nodeSelector": {}, "images": [ - "registry.bstein.dev/bstein/hermes-chat-router@sha256:e3794fb8b9ee76d699893ba303d7080f54698c0a06064dc32293e99e988cabff" + "registry.bstein.dev/bstein/hermes-chat-router@sha256:a4010fdc5b6dce2696e6fec06eb38a3992817faa01ff3472122c870038b22bd2" ] }, { @@ -3141,6 +3141,12 @@ "port": 8642, "targetPort": "api", "protocol": "TCP" + }, + { + "name": "telegram-media", + "port": 8788, + "targetPort": "telegram-media", + "protocol": "TCP" } ] }, diff --git a/knowledge/catalog/atlas.yaml b/knowledge/catalog/atlas.yaml index 434335af..3ee78f4d 100644 --- a/knowledge/catalog/atlas.yaml +++ b/knowledge/catalog/atlas.yaml @@ -624,7 +624,7 @@ workloads: serviceAccountName: hermes-chat nodeSelector: {} images: - - registry.bstein.dev/bstein/hermes-chat-router@sha256:e3794fb8b9ee76d699893ba303d7080f54698c0a06064dc32293e99e988cabff + - registry.bstein.dev/bstein/hermes-chat-router@sha256:a4010fdc5b6dce2696e6fec06eb38a3992817faa01ff3472122c870038b22bd2 - kind: Deployment namespace: hermes name: hermes-chat-sandbox-0 @@ -2097,6 +2097,10 @@ services: port: 8642 targetPort: api protocol: TCP + - name: telegram-media + port: 8788 + targetPort: telegram-media + protocol: TCP - namespace: hermes name: hermes-claude-broker type: ClusterIP diff --git a/services/comms/knowledge/catalog/atlas.json b/services/comms/knowledge/catalog/atlas.json index d88e6d2b..6fcdd018 100644 --- a/services/comms/knowledge/catalog/atlas.json +++ b/services/comms/knowledge/catalog/atlas.json @@ -930,7 +930,7 @@ "serviceAccountName": "hermes-chat", "nodeSelector": {}, "images": [ - "registry.bstein.dev/bstein/hermes-chat-router@sha256:e3794fb8b9ee76d699893ba303d7080f54698c0a06064dc32293e99e988cabff" + "registry.bstein.dev/bstein/hermes-chat-router@sha256:a4010fdc5b6dce2696e6fec06eb38a3992817faa01ff3472122c870038b22bd2" ] }, { @@ -3141,6 +3141,12 @@ "port": 8642, "targetPort": "api", "protocol": "TCP" + }, + { + "name": "telegram-media", + "port": 8788, + "targetPort": "telegram-media", + "protocol": "TCP" } ] }, diff --git a/services/comms/knowledge/catalog/atlas.yaml b/services/comms/knowledge/catalog/atlas.yaml index 434335af..3ee78f4d 100644 --- a/services/comms/knowledge/catalog/atlas.yaml +++ b/services/comms/knowledge/catalog/atlas.yaml @@ -624,7 +624,7 @@ workloads: serviceAccountName: hermes-chat nodeSelector: {} images: - - registry.bstein.dev/bstein/hermes-chat-router@sha256:e3794fb8b9ee76d699893ba303d7080f54698c0a06064dc32293e99e988cabff + - registry.bstein.dev/bstein/hermes-chat-router@sha256:a4010fdc5b6dce2696e6fec06eb38a3992817faa01ff3472122c870038b22bd2 - kind: Deployment namespace: hermes name: hermes-chat-sandbox-0 @@ -2097,6 +2097,10 @@ services: port: 8642 targetPort: api protocol: TCP + - name: telegram-media + port: 8788 + targetPort: telegram-media + protocol: TCP - namespace: hermes name: hermes-claude-broker type: ClusterIP diff --git a/services/hermes/chat-router.yaml b/services/hermes/chat-router.yaml index 0538b73e..92b4f5fe 100644 --- a/services/hermes/chat-router.yaml +++ b/services/hermes/chat-router.yaml @@ -62,7 +62,7 @@ spec: values: [rpi5] containers: - name: router - image: registry.bstein.dev/bstein/hermes-chat-router@sha256:e3794fb8b9ee76d699893ba303d7080f54698c0a06064dc32293e99e988cabff + image: registry.bstein.dev/bstein/hermes-chat-router@sha256:a4010fdc5b6dce2696e6fec06eb38a3992817faa01ff3472122c870038b22bd2 imagePullPolicy: IfNotPresent ports: - {name: http, containerPort: 8080, protocol: TCP} diff --git a/services/hermes/chat-statefulset.yaml b/services/hermes/chat-statefulset.yaml index 55bf7fae..ffaa5d0e 100644 --- a/services/hermes/chat-statefulset.yaml +++ b/services/hermes/chat-statefulset.yaml @@ -327,6 +327,42 @@ spec: resources: requests: {cpu: 100m, memory: 256Mi} limits: {cpu: 750m, memory: 1Gi} + - name: telegram-media + image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107 + imagePullPolicy: IfNotPresent + command: [/opt/hermes/.venv/bin/python, /opt/coordinator/telegram_media_server.py] + ports: + - {name: telegram-media, containerPort: 8788, protocol: TCP} + env: + - name: POD_NAME + valueFrom: + fieldRef: + fieldPath: metadata.name + - {name: HERMES_MEDIA_RELAY_KEY_FILE, value: /opt/data/.env} + volumeMounts: + - {name: home, mountPath: /opt/data, readOnly: true} + - {name: workspace, mountPath: /opt/data/workspace, readOnly: true} + - {name: coordinator, mountPath: /opt/coordinator, readOnly: true} + readinessProbe: + httpGet: {path: /healthz, port: telegram-media} + initialDelaySeconds: 2 + periodSeconds: 10 + livenessProbe: + httpGet: {path: /healthz, port: telegram-media} + initialDelaySeconds: 10 + periodSeconds: 20 + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: [ALL] + readOnlyRootFilesystem: true + runAsUser: 10000 + runAsGroup: 10000 + seccompProfile: + type: RuntimeDefault + resources: + requests: {cpu: 10m, memory: 24Mi} + limits: {cpu: 100m, memory: 64Mi} volumes: - name: provider-auth persistentVolumeClaim: diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index b4cffbc5..094921b6 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -81,6 +81,7 @@ configMapGenerator: - patch_tui_gateway.py=scripts/patch_tui_gateway.py - patch_ttyd_index.py=scripts/patch_ttyd_index.py - routing_catalog.py=scripts/routing_catalog.py + - telegram_media_server.py=scripts/telegram_media_server.py options: disableNameSuffixHash: true - name: hermes-agent-kubeconfig diff --git a/services/hermes/networkpolicy.yaml b/services/hermes/networkpolicy.yaml index 005adb18..699c97b8 100644 --- a/services/hermes/networkpolicy.yaml +++ b/services/hermes/networkpolicy.yaml @@ -299,6 +299,7 @@ spec: ports: - {protocol: TCP, port: 8787} - {protocol: TCP, port: 8642} + - {protocol: TCP, port: 8788} egress: - to: - namespaceSelector: @@ -482,6 +483,7 @@ spec: ports: - {protocol: TCP, port: 8787} - {protocol: TCP, port: 8642} + - {protocol: TCP, port: 8788} - to: - namespaceSelector: matchLabels: diff --git a/services/hermes/router/main.go b/services/hermes/router/main.go index 2f59eb72..50201aa1 100644 --- a/services/hermes/router/main.go +++ b/services/hermes/router/main.go @@ -35,14 +35,15 @@ type tenantState struct { } type tenantRouter struct { - mu sync.Mutex - state tenantState - statePath string - slots int - backendURL func(int) string - backendAPIURL func(int) string - now func() time.Time - telegram *telegramBot + mu sync.Mutex + state tenantState + statePath string + slots int + backendURL func(int) string + backendAPIURL func(int) string + backendMediaURL func(int) string + now func() time.Time + telegram *telegramBot } var deniedPrefixes = []string{ @@ -396,6 +397,9 @@ func main() { router.backendAPIURL = func(slot int) string { return fmt.Sprintf("http://hermes-chat-tenant-%d.hermes-chat-tenant.hermes.svc.cluster.local:8642", slot) } + router.backendMediaURL = func(slot int) string { + return fmt.Sprintf("http://hermes-chat-tenant-%d.hermes-chat-tenant.hermes.svc.cluster.local:8788", slot) + } telegramConfig, err := readTelegramConfig(os.Getenv("TELEGRAM_CONFIG_PATH")) if err != nil { log.Printf("Telegram is disabled: configuration is unavailable") diff --git a/services/hermes/router/telegram.go b/services/hermes/router/telegram.go index 02a4e519..e84aa0d6 100644 --- a/services/hermes/router/telegram.go +++ b/services/hermes/router/telegram.go @@ -28,6 +28,7 @@ type telegramBot struct { router *tenantRouter apiBase string client *http.Client + mediaClient *http.Client agentClient *http.Client mu sync.RWMutex botUsername string @@ -89,6 +90,7 @@ func newTelegramBot(config telegramConfig, router *tenantRouter) *telegramBot { router: router, apiBase: "https://api.telegram.org/bot" + config.BotToken, client: &http.Client{Timeout: 70 * time.Second}, + mediaClient: newTenantMediaClient(), agentClient: &http.Client{Timeout: 15 * time.Minute}, workLimit: make(chan struct{}, 4), slotLocks: map[int]*sync.Mutex{}, @@ -289,7 +291,7 @@ func (bot *telegramBot) handleUpdate(update telegramUpdate) { _ = bot.sendText(message.Chat.ID, "Hermes could not answer right now. Please try again shortly.") return } - _ = bot.sendText(message.Chat.ID, reply) + _ = bot.sendReply(message.Chat.ID, slot, reply) } func (bot *telegramBot) askTenant(slot int, text string, updateID int64) (string, error) { diff --git a/services/hermes/router/telegram_media.go b/services/hermes/router/telegram_media.go new file mode 100644 index 00000000..90be976f --- /dev/null +++ b/services/hermes/router/telegram_media.go @@ -0,0 +1,328 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "mime" + "mime/multipart" + "net/http" + "net/url" + "os" + "path/filepath" + "regexp" + "strconv" + "strings" + "time" + "unicode/utf8" +) + +const ( + telegramCaptionLimit = 1000 + tenantMediaLimit = 50 << 20 +) + +var telegramMediaMarker = regexp.MustCompile(`MEDIA:([^\s\)\]]+)`) + +type telegramReply struct { + Text string + MediaPaths []string +} + +type tenantMedia struct { + Name string + MIME string + Path string +} + +func newTenantMediaClient() *http.Client { + return &http.Client{Timeout: 70 * time.Second} +} + +func parseTelegramReply(text string) telegramReply { + matches := telegramMediaMarker.FindAllStringSubmatch(text, -1) + paths := make([]string, 0, len(matches)) + for _, match := range matches { + if len(match) == 2 { + paths = append(paths, match[1]) + } + } + cleaned := telegramMediaMarker.ReplaceAllString(text, "") + lines := strings.Split(cleaned, "\n") + compact := make([]string, 0, len(lines)) + blank := false + for _, line := range lines { + line = strings.TrimRight(line, " \t") + if strings.TrimSpace(line) == "" { + if len(compact) > 0 && !blank { + compact = append(compact, "") + blank = true + } + continue + } + compact = append(compact, line) + blank = false + } + return telegramReply{ + Text: strings.TrimSpace(strings.Join(compact, "\n")), + MediaPaths: paths, + } +} + +func hasParentTraversal(path string) bool { + for _, component := range strings.Split(filepath.ToSlash(path), "/") { + if component == ".." { + return true + } + } + return false +} + +func pathWithin(path, root string) bool { + relative, err := filepath.Rel(root, path) + return err == nil && relative != ".." && !strings.HasPrefix(relative, ".."+string(filepath.Separator)) +} + +func normalizeTenantMediaPath(raw string) (string, error) { + path := strings.TrimSpace(raw) + if path == "" || strings.ContainsRune(path, '\x00') || strings.Contains(path, `\`) || hasParentTraversal(path) { + return "", errors.New("invalid media path") + } + if path == "/workspace" || strings.HasPrefix(path, "/workspace/") { + path = "/opt/data/workspace" + strings.TrimPrefix(path, "/workspace") + } + if !filepath.IsAbs(path) { + return "", errors.New("media path must be absolute") + } + path = filepath.Clean(path) + for _, root := range []string{"/opt/data/cache/images", "/opt/data/workspace"} { + if pathWithin(path, root) { + return path, nil + } + } + return "", errors.New("media path is outside the tenant media roots") +} + +func (bot *telegramBot) tenantMediaRequest(ctx context.Context, slot int, path string) (*http.Response, error) { + if bot.router == nil || bot.router.backendMediaURL == nil { + return nil, errors.New("tenant media API unavailable") + } + endpoint := bot.router.backendMediaURL(slot) + "/media?" + url.Values{ + "path": {path}, + }.Encode() + request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return nil, errors.New("create tenant media request") + } + request.Header.Set("Authorization", "Bearer "+bot.config.RelayKey) + request.Header.Set("X-Hermes-Tenant-Slot", strconv.Itoa(slot)) + return bot.mediaClient.Do(request) +} + +func (bot *telegramBot) fetchTenantMedia(ctx context.Context, slot int, path string) (tenantMedia, error) { + if bot.mediaClient == nil { + bot.mediaClient = newTenantMediaClient() + } + response, err := bot.tenantMediaRequest(ctx, slot, path) + if err != nil { + return tenantMedia{}, errors.New("tenant media unavailable") + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + return tenantMedia{}, errors.New("tenant media rejected") + } + temporary, err := os.CreateTemp("", "hermes-telegram-media-*") + if err != nil { + return tenantMedia{}, errors.New("stage tenant media") + } + temporaryPath := temporary.Name() + keep := false + defer func() { + _ = temporary.Close() + if !keep { + _ = os.Remove(temporaryPath) + } + }() + written, err := io.Copy(temporary, io.LimitReader(response.Body, tenantMediaLimit+1)) + if err != nil { + return tenantMedia{}, errors.New("read tenant media") + } + if written > tenantMediaLimit { + return tenantMedia{}, errors.New("tenant media exceeds upload limit") + } + declaredType, _, _ := mime.ParseMediaType(response.Header.Get("Content-Type")) + if _, err := temporary.Seek(0, io.SeekStart); err != nil { + return tenantMedia{}, errors.New("inspect tenant media") + } + header := make([]byte, 512) + read, readErr := temporary.Read(header) + if readErr != nil && readErr != io.EOF { + return tenantMedia{}, errors.New("inspect tenant media") + } + mediaType := http.DetectContentType(header[:read]) + if mediaType == "application/octet-stream" && declaredType != "" { + mediaType = declaredType + } + if err := temporary.Close(); err != nil { + return tenantMedia{}, errors.New("stage tenant media") + } + keep = true + return tenantMedia{Name: filepath.Base(path), MIME: mediaType, Path: temporaryPath}, nil +} + +func (bot *telegramBot) callMultipart( + ctx context.Context, + method string, + fields map[string]string, + fileField string, + media tenantMedia, +) error { + reader, writer := io.Pipe() + multipartWriter := multipart.NewWriter(writer) + request, err := http.NewRequestWithContext(ctx, http.MethodPost, bot.apiBase+"/"+method, reader) + if err != nil { + reader.Close() + writer.Close() + return errors.New("create Telegram request") + } + request.Header.Set("Content-Type", multipartWriter.FormDataContentType()) + writeDone := make(chan error, 1) + go func() { + var writeErr error + mediaFile, openErr := os.Open(media.Path) + if openErr != nil { + writeErr = openErr + } else { + defer mediaFile.Close() + } + for key, value := range fields { + if writeErr == nil { + writeErr = multipartWriter.WriteField(key, value) + } + } + if writeErr == nil { + var part io.Writer + part, writeErr = multipartWriter.CreateFormFile(fileField, media.Name) + if writeErr == nil { + _, writeErr = io.Copy(part, mediaFile) + } + } + if closeErr := multipartWriter.Close(); writeErr == nil { + writeErr = closeErr + } + if writeErr != nil { + _ = writer.CloseWithError(writeErr) + } else { + _ = writer.Close() + } + writeDone <- writeErr + }() + response, err := bot.client.Do(request) + if err != nil { + _ = reader.Close() + <-writeDone + return errors.New("Telegram API unavailable") + } + defer response.Body.Close() + if err := <-writeDone; err != nil { + return errors.New("encode Telegram upload") + } + body, err := io.ReadAll(io.LimitReader(response.Body, 2<<20)) + if err != nil { + return errors.New("read Telegram response") + } + var envelope struct { + OK bool `json:"ok"` + Description string `json:"description"` + ErrorCode int `json:"error_code"` + } + if err := json.Unmarshal(body, &envelope); err != nil { + return fmt.Errorf("Telegram API returned undecodable status %d", response.StatusCode) + } + if response.StatusCode != http.StatusOK || !envelope.OK { + description := strings.TrimSpace(envelope.Description) + if description == "" { + description = http.StatusText(response.StatusCode) + } + return fmt.Errorf("Telegram API error %d: %s", envelope.ErrorCode, description) + } + return nil +} + +func telegramPhotoMIME(mediaType string) bool { + return mediaType == "image/jpeg" || mediaType == "image/png" +} + +func (bot *telegramBot) sendMedia(chatID int64, media tenantMedia, caption string) error { + fields := map[string]string{"chat_id": strconv.FormatInt(chatID, 10)} + if caption != "" { + fields["caption"] = caption + } + if telegramPhotoMIME(media.MIME) { + ctx, cancel := context.WithTimeout(context.Background(), 70*time.Second) + err := bot.callMultipart(ctx, "sendPhoto", fields, "photo", media) + cancel() + if err == nil { + return nil + } + } + ctx, cancel := context.WithTimeout(context.Background(), 70*time.Second) + defer cancel() + return bot.callMultipart(ctx, "sendDocument", fields, "document", media) +} + +func (bot *telegramBot) sendReply(chatID int64, slot int, text string) error { + reply := parseTelegramReply(text) + if len(reply.MediaPaths) == 0 { + return bot.sendText(chatID, reply.Text) + } + + captionAvailable := reply.Text != "" && utf8.RuneCountInString(reply.Text) <= telegramCaptionLimit + captionSent := false + var firstError error + if reply.Text != "" && !captionAvailable { + if err := bot.sendText(chatID, reply.Text); err != nil { + firstError = err + } else { + captionSent = true + } + } + + for _, rawPath := range reply.MediaPaths { + path, err := normalizeTenantMediaPath(rawPath) + if err == nil { + ctx, cancel := context.WithTimeout(context.Background(), 70*time.Second) + media, fetchErr := bot.fetchTenantMedia(ctx, slot, path) + cancel() + err = fetchErr + if err == nil { + caption := "" + if captionAvailable && !captionSent { + caption = reply.Text + } + err = bot.sendMedia(chatID, media, caption) + _ = os.Remove(media.Path) + if err == nil && caption != "" { + captionSent = true + } + } + } + if err != nil { + if firstError == nil { + firstError = err + } + if fallbackErr := bot.sendText(chatID, "I created an attachment, but Telegram could not deliver it. You can still open it in Hermes WebUI."); fallbackErr != nil && firstError == nil { + firstError = fallbackErr + } + } + } + + if reply.Text != "" && !captionSent { + if err := bot.sendText(chatID, reply.Text); err != nil && firstError == nil { + firstError = err + } + } + return firstError +} diff --git a/services/hermes/router/telegram_media_test.go b/services/hermes/router/telegram_media_test.go new file mode 100644 index 00000000..ee36f3d0 --- /dev/null +++ b/services/hermes/router/telegram_media_test.go @@ -0,0 +1,424 @@ +package main + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "testing" +) + +type recordedTelegramRequest struct { + Method string + Fields map[string]string + FileField string + FileName string + FileData []byte +} + +type telegramAPIRecorder struct { + mu sync.Mutex + requests []recordedTelegramRequest + failures map[string]int +} + +func (recorder *telegramAPIRecorder) snapshot() []recordedTelegramRequest { + recorder.mu.Lock() + defer recorder.mu.Unlock() + return append([]recordedTelegramRequest(nil), recorder.requests...) +} + +func newFakeTelegramAPI(t *testing.T, failures map[string]int) (*httptest.Server, *telegramAPIRecorder) { + t.Helper() + recorder := &telegramAPIRecorder{failures: failures} + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + method := strings.TrimPrefix(request.URL.Path, "/") + recorded := recordedTelegramRequest{Method: method, Fields: map[string]string{}} + if strings.HasPrefix(request.Header.Get("Content-Type"), "multipart/form-data") { + if err := request.ParseMultipartForm(64 << 20); err != nil { + t.Fatalf("parse Telegram multipart request: %v", err) + } + for key, values := range request.MultipartForm.Value { + if len(values) > 0 { + recorded.Fields[key] = values[0] + } + } + for _, field := range []string{"photo", "document"} { + files := request.MultipartForm.File[field] + if len(files) == 0 { + continue + } + file, err := files[0].Open() + if err != nil { + t.Fatal(err) + } + recorded.FileData, err = io.ReadAll(file) + file.Close() + if err != nil { + t.Fatal(err) + } + recorded.FileField = field + recorded.FileName = files[0].Filename + } + } else { + if err := request.ParseForm(); err != nil { + t.Fatal(err) + } + for key, values := range request.Form { + if len(values) > 0 { + recorded.Fields[key] = values[0] + } + } + } + recorder.mu.Lock() + recorder.requests = append(recorder.requests, recorded) + fail := recorder.failures[method] > 0 + if fail { + recorder.failures[method]-- + } + recorder.mu.Unlock() + writer.Header().Set("Content-Type", "application/json") + if fail { + writer.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(writer).Encode(map[string]any{ + "ok": false, "error_code": 400, "description": "test upload rejection", + }) + return + } + _ = json.NewEncoder(writer).Encode(map[string]any{"ok": true, "result": map[string]any{}}) + })) + return server, recorder +} + +func newTelegramMediaTestBot(t *testing.T, telegramURL string, backendURL func(int) string) *telegramBot { + t.Helper() + router, err := newTenantRouter(filepath.Join(t.TempDir(), "state.json"), 2, backendURL) + if err != nil { + t.Fatal(err) + } + router.backendMediaURL = backendURL + bot := newTelegramBot(telegramConfig{RelayKey: "relay-secret"}, router) + bot.apiBase = telegramURL + return bot +} + +func pngFixture() []byte { + return append([]byte{0x89, 'P', 'N', 'G', '\r', '\n', 0x1a, '\n'}, []byte("telegram-test-image")...) +} + +func TestParseTelegramReplyRemovesOneOrMoreInternalMarkers(t *testing.T) { + reply := parseTelegramReply("Created these.\n\nMEDIA:/opt/data/cache/images/one.png\nMEDIA:/opt/data/workspace/two.png") + if reply.Text != "Created these." { + t.Fatalf("unexpected caption %q", reply.Text) + } + if len(reply.MediaPaths) != 2 || reply.MediaPaths[0] != "/opt/data/cache/images/one.png" || reply.MediaPaths[1] != "/opt/data/workspace/two.png" { + t.Fatalf("unexpected media paths: %#v", reply.MediaPaths) + } + if strings.Contains(reply.Text, "MEDIA:") || strings.Contains(reply.Text, "/opt/data") { + t.Fatalf("internal marker leaked into caption %q", reply.Text) + } +} + +func TestNormalizeTenantMediaPath(t *testing.T) { + allowed := map[string]string{ + "/workspace/./renders//portrait.png": "/opt/data/workspace/renders/portrait.png", + "/opt/data/workspace/./reports/result.pdf": "/opt/data/workspace/reports/result.pdf", + "/opt/data/cache/images//generated/image.png": "/opt/data/cache/images/generated/image.png", + } + for input, expected := range allowed { + actual, err := normalizeTenantMediaPath(input) + if err != nil || actual != expected { + t.Errorf("normalize %q: got %q, %v; want %q", input, actual, err, expected) + } + } + for _, input := range []string{ + "relative.png", + "/opt/data/cache/images/../secrets/token.png", + "/opt/data/workspace/../../other-tenant/image.png", + "/opt/data/cache/images-foreign/image.png", + "/opt/data/tenants/1/image.png", + `/opt/data/cache/images\..\secret.png`, + } { + if normalized, err := normalizeTenantMediaPath(input); err == nil { + t.Errorf("unsafe path %q normalized to %q", input, normalized) + } + } +} + +func TestSendReplyTextOnlyUsesSendMessage(t *testing.T) { + telegram, recorder := newFakeTelegramAPI(t, nil) + defer telegram.Close() + bot := newTelegramMediaTestBot(t, telegram.URL, func(int) string { + t.Fatal("text-only reply attempted a tenant media request") + return "" + }) + if err := bot.sendReply(42, 0, "Hello from Hermes"); err != nil { + t.Fatal(err) + } + requests := recorder.snapshot() + if len(requests) != 1 || requests[0].Method != "sendMessage" || requests[0].Fields["text"] != "Hello from Hermes" { + t.Fatalf("unexpected Telegram requests: %#v", requests) + } +} + +func TestSendReplyUploadsPhotoWithCaptionThroughTenantAuth(t *testing.T) { + var mediaRequests int + tenant := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + mediaRequests++ + if request.URL.Path != "/media" || request.URL.Query().Get("path") != "/opt/data/workspace/renders/portrait.png" { + t.Fatalf("unexpected tenant media URL %s", request.URL.String()) + } + if request.Header.Get("Authorization") != "Bearer relay-secret" { + t.Fatal("tenant media relay authentication was not set") + } + if request.Header.Get("X-Hermes-Tenant-Slot") != "1" { + t.Fatalf("unexpected tenant slot %q", request.Header.Get("X-Hermes-Tenant-Slot")) + } + writer.Header().Set("Content-Type", "image/png") + _, _ = writer.Write(pngFixture()) + })) + defer tenant.Close() + telegram, recorder := newFakeTelegramAPI(t, nil) + defer telegram.Close() + bot := newTelegramMediaTestBot(t, telegram.URL, func(slot int) string { + if slot != 1 { + t.Fatalf("requested media from slot %d", slot) + } + return tenant.URL + }) + if err := bot.sendReply(42, 1, "Created it.\nMEDIA:/workspace/./renders/portrait.png"); err != nil { + t.Fatal(err) + } + requests := recorder.snapshot() + if len(requests) != 1 || requests[0].Method != "sendPhoto" || requests[0].FileField != "photo" { + t.Fatalf("unexpected Telegram requests: %#v", requests) + } + if requests[0].Fields["caption"] != "Created it." || requests[0].Fields["chat_id"] != "42" { + t.Fatalf("unexpected photo fields: %#v", requests[0].Fields) + } + if requests[0].FileName != "portrait.png" || string(requests[0].FileData) != string(pngFixture()) { + t.Fatalf("unexpected uploaded photo: %#v", requests[0]) + } +} + +func TestSendReplyUploadsMultipleMarkersAndCaptionsOnlyFirst(t *testing.T) { + tenant := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + path := request.URL.Query().Get("path") + switch filepath.Ext(path) { + case ".png": + writer.Header().Set("Content-Type", "image/png") + _, _ = writer.Write(pngFixture()) + case ".pdf": + writer.Header().Set("Content-Type", "application/pdf") + _, _ = writer.Write([]byte("%PDF-1.4 test")) + default: + writer.WriteHeader(http.StatusNotFound) + } + })) + defer tenant.Close() + telegram, recorder := newFakeTelegramAPI(t, nil) + defer telegram.Close() + bot := newTelegramMediaTestBot(t, telegram.URL, func(int) string { return tenant.URL }) + err := bot.sendReply(7, 0, "Two files.\nMEDIA:/opt/data/cache/images/one.png\nMEDIA:/opt/data/workspace/two.pdf") + if err != nil { + t.Fatal(err) + } + requests := recorder.snapshot() + if len(requests) != 2 || requests[0].Method != "sendPhoto" || requests[1].Method != "sendDocument" { + t.Fatalf("unexpected Telegram methods: %#v", requests) + } + if requests[0].Fields["caption"] != "Two files." || requests[1].Fields["caption"] != "" { + t.Fatalf("caption was not scoped to first upload: %#v", requests) + } +} + +func TestSendReplySendsLongCaptionSeparately(t *testing.T) { + tenant := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + writer.Header().Set("Content-Type", "image/png") + _, _ = writer.Write(pngFixture()) + })) + defer tenant.Close() + telegram, recorder := newFakeTelegramAPI(t, nil) + defer telegram.Close() + bot := newTelegramMediaTestBot(t, telegram.URL, func(int) string { return tenant.URL }) + caption := strings.Repeat("detail ", 180) + if err := bot.sendReply(7, 0, caption+"\nMEDIA:/opt/data/cache/images/one.png"); err != nil { + t.Fatal(err) + } + requests := recorder.snapshot() + if len(requests) != 2 || requests[0].Method != "sendMessage" || requests[1].Method != "sendPhoto" { + t.Fatalf("long caption was not sent separately: %#v", requests) + } + if requests[0].Fields["text"] != strings.TrimSpace(caption) || requests[1].Fields["caption"] != "" { + t.Fatalf("long caption leaked into photo upload: %#v", requests) + } +} + +func TestSendReplyRejectsTraversalWithoutFetchingTenant(t *testing.T) { + telegram, recorder := newFakeTelegramAPI(t, nil) + defer telegram.Close() + bot := newTelegramMediaTestBot(t, telegram.URL, func(int) string { + t.Fatal("traversal path reached the tenant backend") + return "" + }) + if err := bot.sendReply(9, 0, "MEDIA:/opt/data/cache/images/../auth.json"); err == nil { + t.Fatal("expected traversal rejection") + } + assertOnlySafeFallback(t, recorder.snapshot()) +} + +func TestSendReplyRejectsTenantSymlinkEscape(t *testing.T) { + root := filepath.Join(t.TempDir(), "images") + outside := filepath.Join(t.TempDir(), "outside.png") + if err := os.MkdirAll(root, 0700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(outside, pngFixture(), 0600); err != nil { + t.Fatal(err) + } + if err := os.Symlink(outside, filepath.Join(root, "escape.png")); err != nil { + t.Fatal(err) + } + tenant := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + logical := request.URL.Query().Get("path") + relative := strings.TrimPrefix(logical, "/opt/data/cache/images/") + target, err := filepath.EvalSymlinks(filepath.Join(root, relative)) + if err != nil || !pathWithin(target, root) { + writer.WriteHeader(http.StatusForbidden) + return + } + _, _ = os.Open(target) + })) + defer tenant.Close() + telegram, recorder := newFakeTelegramAPI(t, nil) + defer telegram.Close() + bot := newTelegramMediaTestBot(t, telegram.URL, func(int) string { return tenant.URL }) + if err := bot.sendReply(9, 0, "MEDIA:/opt/data/cache/images/escape.png"); err == nil { + t.Fatal("expected tenant symlink rejection") + } + assertOnlySafeFallback(t, recorder.snapshot()) +} + +func TestSendReplyHandlesMissingTenantFile(t *testing.T) { + tenant := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + writer.WriteHeader(http.StatusNotFound) + })) + defer tenant.Close() + telegram, recorder := newFakeTelegramAPI(t, nil) + defer telegram.Close() + bot := newTelegramMediaTestBot(t, telegram.URL, func(int) string { return tenant.URL }) + if err := bot.sendReply(9, 0, "MEDIA:/opt/data/cache/images/missing.png"); err == nil { + t.Fatal("expected missing-file error") + } + assertOnlySafeFallback(t, recorder.snapshot()) +} + +func TestSendPhotoFailureFallsBackToDocument(t *testing.T) { + tenant := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + writer.Header().Set("Content-Type", "image/png") + _, _ = writer.Write(pngFixture()) + })) + defer tenant.Close() + telegram, recorder := newFakeTelegramAPI(t, map[string]int{"sendPhoto": 1}) + defer telegram.Close() + bot := newTelegramMediaTestBot(t, telegram.URL, func(int) string { return tenant.URL }) + if err := bot.sendReply(11, 0, "Portrait.\nMEDIA:/opt/data/cache/images/portrait.png"); err != nil { + t.Fatal(err) + } + requests := recorder.snapshot() + if len(requests) != 2 || requests[0].Method != "sendPhoto" || requests[1].Method != "sendDocument" { + t.Fatalf("photo did not fall back to document: %#v", requests) + } + if requests[1].Fields["caption"] != "Portrait." || requests[1].FileField != "document" { + t.Fatalf("document fallback lost content: %#v", requests[1]) + } +} + +func TestMislabeledImageUsesSniffedDocumentType(t *testing.T) { + tenant := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + writer.Header().Set("Content-Type", "image/png") + _, _ = writer.Write([]byte("%PDF-1.4 mislabeled")) + })) + defer tenant.Close() + telegram, recorder := newFakeTelegramAPI(t, nil) + defer telegram.Close() + bot := newTelegramMediaTestBot(t, telegram.URL, func(int) string { return tenant.URL }) + if err := bot.sendReply(11, 0, "MEDIA:/opt/data/cache/images/mislabeled.png"); err != nil { + t.Fatal(err) + } + requests := recorder.snapshot() + if len(requests) != 1 || requests[0].Method != "sendDocument" { + t.Fatalf("mislabeled media was not routed as a document: %#v", requests) + } +} + +func TestUploadFailureSendsSafeTextWithoutInternalPath(t *testing.T) { + tenant := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + writer.Header().Set("Content-Type", "image/png") + _, _ = writer.Write(pngFixture()) + })) + defer tenant.Close() + telegram, recorder := newFakeTelegramAPI(t, map[string]int{"sendPhoto": 1, "sendDocument": 1}) + defer telegram.Close() + bot := newTelegramMediaTestBot(t, telegram.URL, func(int) string { return tenant.URL }) + if err := bot.sendReply(11, 0, "Created it.\nMEDIA:/opt/data/cache/images/private.png"); err == nil { + t.Fatal("expected upload failure") + } + requests := recorder.snapshot() + if len(requests) != 4 || requests[2].Method != "sendMessage" || requests[3].Method != "sendMessage" { + t.Fatalf("unexpected failure requests: %#v", requests) + } + for _, request := range requests { + for _, value := range request.Fields { + if strings.Contains(value, "MEDIA:") || strings.Contains(value, "/opt/data/") { + t.Fatalf("internal path leaked to Telegram: %#v", requests) + } + } + } + if requests[2].Fields["text"] == "" || requests[3].Fields["text"] != "Created it." { + t.Fatalf("safe fallback or response text missing: %#v", requests) + } +} + +func TestSendReplyUsesOnlyLinkedTenantSlot(t *testing.T) { + tenantZero := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + t.Fatal("cross-tenant media request reached slot zero") + })) + defer tenantZero.Close() + tenantOne := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.Header.Get("X-Hermes-Tenant-Slot") != "1" { + t.Fatalf("unexpected tenant header %q", request.Header.Get("X-Hermes-Tenant-Slot")) + } + writer.Header().Set("Content-Type", "image/png") + _, _ = writer.Write(pngFixture()) + })) + defer tenantOne.Close() + telegram, recorder := newFakeTelegramAPI(t, nil) + defer telegram.Close() + bot := newTelegramMediaTestBot(t, telegram.URL, func(slot int) string { + return []string{tenantZero.URL, tenantOne.URL}[slot] + }) + if err := bot.sendReply(99, 1, "MEDIA:/opt/data/cache/images/tenant-one.png"); err != nil { + t.Fatal(err) + } + requests := recorder.snapshot() + if len(requests) != 1 || requests[0].Fields["chat_id"] != strconv.FormatInt(99, 10) { + t.Fatalf("unexpected Telegram delivery: %#v", requests) + } +} + +func assertOnlySafeFallback(t *testing.T, requests []recordedTelegramRequest) { + t.Helper() + if len(requests) != 1 || requests[0].Method != "sendMessage" { + t.Fatalf("expected only a safe fallback message, got %#v", requests) + } + text := requests[0].Fields["text"] + if text == "" || strings.Contains(text, "MEDIA:") || strings.Contains(text, "/opt/data/") { + t.Fatalf("unsafe fallback %q", text) + } +} diff --git a/services/hermes/scripts/telegram_media_server.py b/services/hermes/scripts/telegram_media_server.py new file mode 100644 index 00000000..2d905053 --- /dev/null +++ b/services/hermes/scripts/telegram_media_server.py @@ -0,0 +1,190 @@ +#!/usr/bin/env python3 +"""Serve one Hermes tenant's generated media to the Telegram router.""" + +from __future__ import annotations + +import hmac +import mimetypes +import os +from pathlib import Path, PurePosixPath +import posixpath +import shutil +import stat +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlsplit + + +ALLOWED_ROOTS = (Path("/opt/data/cache/images"), Path("/opt/data/workspace")) +MAX_MEDIA_BYTES = 50 << 20 + + +class MediaPathError(ValueError): + """Raised when a requested path violates the tenant media boundary.""" + + +def read_relay_key(path: Path) -> str: + """Read the shared relay key from the tenant's private environment file.""" + for line in path.read_text(encoding="utf-8").splitlines(): + key, separator, value = line.partition("=") + if separator and key.strip() == "API_SERVER_KEY" and value.strip(): + return value.strip() + raise RuntimeError("tenant relay key is unavailable") + + +def tenant_slot(pod_name: str) -> str: + """Extract the StatefulSet ordinal used to bind requests to this tenant.""" + prefix, separator, ordinal = pod_name.rpartition("-") + if not separator or not prefix or not ordinal.isdigit(): + raise RuntimeError("tenant pod name has no ordinal") + return ordinal + + +def normalize_media_path(raw_path: str, allowed_roots: tuple[Path, ...] = ALLOWED_ROOTS) -> tuple[Path, tuple[str, ...]]: + """Return an allowed root and clean relative components for an absolute path.""" + value = str(raw_path or "").strip() + if not value or "\x00" in value or "\\" in value or not value.startswith("/"): + raise MediaPathError("invalid media path") + original = PurePosixPath(value) + if ".." in original.parts: + raise MediaPathError("parent traversal is not allowed") + cleaned = PurePosixPath(posixpath.normpath(value)) + for root in allowed_roots: + logical_root = PurePosixPath(root.as_posix()) + try: + relative = cleaned.relative_to(logical_root) + except ValueError: + continue + components = tuple(part for part in relative.parts if part not in {"", "."}) + if not components: + raise MediaPathError("media path must name a file") + return root, components + raise MediaPathError("media path is outside allowed roots") + + +def open_directory_without_symlinks(path: Path) -> int: + """Open every component of an absolute directory path with O_NOFOLLOW.""" + if not path.is_absolute() or ".." in path.parts: + raise MediaPathError("invalid media root") + flags = os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC | os.O_NOFOLLOW + directory_fd = os.open("/", flags) + try: + for component in path.parts[1:]: + next_fd = os.open(component, flags, dir_fd=directory_fd) + os.close(directory_fd) + directory_fd = next_fd + return directory_fd + except Exception: + os.close(directory_fd) + raise + + +def open_media_file(raw_path: str, allowed_roots: tuple[Path, ...] = ALLOWED_ROOTS) -> tuple[int, os.stat_result, str]: + """Open a regular file beneath an allowed root without following symlinks.""" + root, components = normalize_media_path(raw_path, allowed_roots) + directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC | os.O_NOFOLLOW + file_flags = os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW + directory_fd = open_directory_without_symlinks(root) + try: + for component in components[:-1]: + next_fd = os.open(component, directory_flags, dir_fd=directory_fd) + os.close(directory_fd) + directory_fd = next_fd + file_fd = os.open(components[-1], file_flags, dir_fd=directory_fd) + finally: + os.close(directory_fd) + metadata = os.fstat(file_fd) + if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > MAX_MEDIA_BYTES: + os.close(file_fd) + raise MediaPathError("media is not an uploadable regular file") + return file_fd, metadata, components[-1] + + +class TenantMediaServer(ThreadingHTTPServer): + """HTTP server carrying immutable tenant authentication and root policy.""" + + daemon_threads = True + + def __init__( + self, + address: tuple[str, int], + relay_key: str, + slot: str, + allowed_roots: tuple[Path, ...] = ALLOWED_ROOTS, + ) -> None: + self.relay_key = relay_key + self.tenant_slot = slot + self.allowed_roots = allowed_roots + super().__init__(address, TenantMediaHandler) + + +class TenantMediaHandler(BaseHTTPRequestHandler): + """Authenticate the router and stream one tenant-private file descriptor.""" + + server: TenantMediaServer + + def log_message(self, _format: str, *_args: object) -> None: + """Keep request paths out of logs because they contain internal names.""" + + def _plain_status(self, status_code: int, message: bytes) -> None: + self.send_response(status_code) + self.send_header("Content-Type", "text/plain; charset=utf-8") + self.send_header("Content-Length", str(len(message))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(message) + + def _authorized(self) -> bool: + expected = "Bearer " + self.server.relay_key + supplied = self.headers.get("Authorization", "") + slot = self.headers.get("X-Hermes-Tenant-Slot", "") + return hmac.compare_digest(supplied.encode(), expected.encode()) and hmac.compare_digest( + slot.encode(), self.server.tenant_slot.encode() + ) + + def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + parsed = urlsplit(self.path) + if parsed.path == "/healthz": + self._plain_status(200, b"ok\n") + return + if parsed.path != "/media": + self._plain_status(404, b"not found\n") + return + if not self._authorized(): + self._plain_status(401, b"authentication required\n") + return + raw_path = parse_qs(parsed.query).get("path", [""])[0] + try: + file_fd, metadata, filename = open_media_file(raw_path, self.server.allowed_roots) + except MediaPathError: + self._plain_status(403, b"media path rejected\n") + return + except (FileNotFoundError, NotADirectoryError, PermissionError, OSError): + self._plain_status(404, b"media unavailable\n") + return + content_type = mimetypes.guess_type(filename)[0] or "application/octet-stream" + safe_filename = filename.replace('"', "_").replace("\r", "_").replace("\n", "_") + self.send_response(200) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(metadata.st_size)) + self.send_header("Content-Disposition", f'attachment; filename="{safe_filename}"') + self.send_header("Cache-Control", "no-store") + self.send_header("X-Content-Type-Options", "nosniff") + self.end_headers() + try: + with os.fdopen(file_fd, "rb") as media: + shutil.copyfileobj(media, self.wfile, length=64 << 10) + except (BrokenPipeError, ConnectionResetError): + return + + +def main() -> None: + """Start the tenant-local media handoff server.""" + relay_key_path = Path(os.getenv("HERMES_MEDIA_RELAY_KEY_FILE", "/opt/data/.env")) + relay_key = read_relay_key(relay_key_path) + slot = tenant_slot(os.getenv("POD_NAME", "")) + server = TenantMediaServer(("0.0.0.0", 8788), relay_key, slot) + server.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/services/hermes/service.yaml b/services/hermes/service.yaml index 00b85e6b..e8cddd11 100644 --- a/services/hermes/service.yaml +++ b/services/hermes/service.yaml @@ -58,6 +58,10 @@ spec: - name: api port: 8642 targetPort: api + - name: telegram-media + port: 8788 + targetPort: telegram-media + protocol: TCP --- apiVersion: v1 kind: Service diff --git a/testing/tests/test_hermes_telegram_media_server.py b/testing/tests/test_hermes_telegram_media_server.py new file mode 100644 index 00000000..61d66b25 --- /dev/null +++ b/testing/tests/test_hermes_telegram_media_server.py @@ -0,0 +1,130 @@ +"""Security and transport tests for the tenant-local Telegram media handoff.""" + +from __future__ import annotations + +import importlib.util +import os +from pathlib import Path +import threading +from urllib.error import HTTPError +from urllib.parse import urlencode +from urllib.request import Request, urlopen + +import pytest + + +SCRIPT = Path(__file__).parents[2] / "services/hermes/scripts/telegram_media_server.py" + + +@pytest.fixture(scope="module") +def media_server_module(): + """Load the deployed script without requiring it to be a Python package.""" + spec = importlib.util.spec_from_file_location("telegram_media_server", SCRIPT) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_normalize_media_path_allows_clean_paths_and_rejects_traversal(media_server_module, tmp_path): + root = tmp_path / "images" + root.mkdir() + selected, components = media_server_module.normalize_media_path( + f"{root}/./renders//portrait.png", (root,) + ) + assert selected == root + assert components == ("renders", "portrait.png") + + for unsafe in ( + f"{root}/../secret.png", + f"{root}-other/portrait.png", + "relative.png", + f"{root}\\..\\secret.png", + ): + with pytest.raises(media_server_module.MediaPathError): + media_server_module.normalize_media_path(unsafe, (root,)) + + +def test_open_media_file_rejects_symlink_escape_and_missing_file(media_server_module, tmp_path): + root = tmp_path / "images" + root.mkdir() + outside = tmp_path / "private.png" + outside.write_bytes(b"private") + (root / "escape.png").symlink_to(outside) + + with pytest.raises(OSError): + media_server_module.open_media_file(str(root / "escape.png"), (root,)) + with pytest.raises(FileNotFoundError): + media_server_module.open_media_file(str(root / "missing.png"), (root,)) + + +def test_open_media_file_rejects_symlink_in_allowed_root(media_server_module, tmp_path): + logical_parent = tmp_path / "tenant" + logical_parent.mkdir() + outside = tmp_path / "outside" + (outside / "images").mkdir(parents=True) + (outside / "images" / "private.png").write_bytes(b"private") + (logical_parent / "cache").symlink_to(outside) + logical_root = logical_parent / "cache" / "images" + + with pytest.raises(OSError): + media_server_module.open_media_file(str(logical_root / "private.png"), (logical_root,)) + + +def test_open_media_file_returns_only_regular_file_beneath_root(media_server_module, tmp_path): + root = tmp_path / "images" + nested = root / "generated" + nested.mkdir(parents=True) + expected = b"generated-image" + target = nested / "portrait.png" + target.write_bytes(expected) + + descriptor, metadata, filename = media_server_module.open_media_file(str(target), (root,)) + try: + assert os.read(descriptor, len(expected)) == expected + finally: + os.close(descriptor) + assert metadata.st_size == len(expected) + assert filename == "portrait.png" + + +def test_http_handoff_requires_relay_key_and_exact_tenant_slot(media_server_module, tmp_path): + root = tmp_path / "images" + root.mkdir() + target = root / "portrait.png" + target.write_bytes(b"generated-image") + server = media_server_module.TenantMediaServer( + ("127.0.0.1", 0), "relay-secret", "3", (root,) + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + url = f"http://127.0.0.1:{server.server_port}/media?{urlencode({'path': str(target)})}" + try: + for headers in ( + {"Authorization": "Bearer wrong", "X-Hermes-Tenant-Slot": "3"}, + {"Authorization": "Bearer relay-secret", "X-Hermes-Tenant-Slot": "2"}, + ): + with pytest.raises(HTTPError) as error: + urlopen(Request(url, headers=headers), timeout=2) + assert error.value.code == 401 + + request = Request( + url, + headers={ + "Authorization": "Bearer relay-secret", + "X-Hermes-Tenant-Slot": "3", + }, + ) + with urlopen(request, timeout=2) as response: + assert response.read() == b"generated-image" + assert response.headers.get_content_type() == "image/png" + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + +def test_tenant_slot_requires_statefulset_ordinal(media_server_module): + assert media_server_module.tenant_slot("hermes-chat-tenant-7") == "7" + with pytest.raises(RuntimeError): + media_server_module.tenant_slot("hermes-chat-tenant")