hermes: deliver generated media to Telegram

This commit is contained in:
jenkins 2026-08-15 00:34:08 -03:00
parent 9909f78386
commit c21203b7ea
15 changed files with 1155 additions and 14 deletions

View File

@ -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"
}
]
},

View File

@ -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

View File

@ -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"
}
]
},

View File

@ -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

View File

@ -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}

View File

@ -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:

View File

@ -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

View File

@ -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:

View File

@ -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")

View File

@ -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) {

View File

@ -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
}

View File

@ -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)
}
}

View File

@ -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()

View File

@ -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

View File

@ -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")