Add the converter process: take convert-asks, have the document server make the PDF with a token only it holds, and put it to the target link (issue 1596)

This commit is contained in:
2026-10-21 14:58:41 +02:00
parent caae838a2a
commit 278014fd79
9 changed files with 786 additions and 11 deletions
+13 -3
View File
@@ -7,6 +7,7 @@ import (
"encoding/json"
"fmt"
"net/url"
"regexp"
"sort"
"strings"
"time"
@@ -40,8 +41,12 @@ type Ask struct {
Source string `json:"source"`
Type string `json:"type"`
Target string `json:"target"`
// SHA256 is the source's SHA-256 when the asker knows it; the converter then names the conversion by it.
SHA256 string `json:"sha256,omitempty"`
}
var sha256Pattern = regexp.MustCompile(`^[0-9a-f]{64}$`)
// OpConvert is the one operation.
const OpConvert = "convert"
@@ -106,7 +111,7 @@ func checkLink(name, raw string) error {
// Convert queues one ask and waits for the converter's answer {size, sha256} (the PDF it put to target). Nothing is
// queued for a type the converter does not take or a link that is not one.
func (s *Service) Convert(ctx context.Context, source, typ, target string) (map[string]any, error) {
func (s *Service) Convert(ctx context.Context, source, typ, target, sha256 string) (map[string]any, error) {
t, err := cleanType(typ)
if err != nil {
return nil, err
@@ -117,8 +122,12 @@ func (s *Service) Convert(ctx context.Context, source, typ, target string) (map[
if err := checkLink("target", target); err != nil {
return nil, err
}
sha256 = strings.ToLower(strings.TrimSpace(sha256))
if sha256 != "" && !sha256Pattern.MatchString(sha256) {
return nil, fmt.Errorf("sha256 is the source's SHA-256, 64 hex digits, not %q", sha256)
}
at := s.Now().UTC()
ask := Ask{ID: newAskID(), Op: OpConvert, At: at, Deadline: at.Add(s.Wait), Source: strings.TrimSpace(source), Type: t, Target: strings.TrimSpace(target)}
ask := Ask{ID: newAskID(), Op: OpConvert, At: at, Deadline: at.Add(s.Wait), Source: strings.TrimSpace(source), Type: t, Target: strings.TrimSpace(target), SHA256: sha256}
if err := s.Asks.Publish(ask.ID, ask); err != nil {
return nil, fmt.Errorf("the ask for the converter could not be queued: %w", err)
}
@@ -158,10 +167,11 @@ func tools(s *Service) []stdio.Tool {
"source": map[string]any{"type": "string", "description": "a signed GET link of the document"},
"type": map[string]any{"type": "string", "description": "the document's type: " + strings.Join(Types, ", ")},
"target": map[string]any{"type": "string", "description": "a signed PUT link for the PDF"},
"sha256": map[string]any{"type": "string", "description": "the document's SHA-256, if known (optional)"},
},
Run: func(a map[string]any) (any, error) {
str := func(k string) string { v, _ := a[k].(string); return v }
return s.Convert(context.Background(), str("source"), str("type"), str("target"))
return s.Convert(context.Background(), str("source"), str("type"), str("target"), str("sha256"))
},
}}
}
@@ -52,7 +52,7 @@ const get, put = "http://novox.internal:20001/b/doc.docx?X-Amz-Signature=a", "ht
func TestConvertQueuesOneAskAndAnswersWhatTheConverterWrote(t *testing.T) {
s, b := rig(time.Second)
b.answerOn = func(a Ask) json.RawMessage { return json.RawMessage(`{"size": 4096, "sha256": "ab12"}`) }
got, err := s.Convert(context.Background(), get, "DOCX", put)
got, err := s.Convert(context.Background(), get, "DOCX", put, "")
if err != nil {
t.Fatal(err)
}
@@ -74,7 +74,7 @@ func TestConvertAnswersTheConvertersError(t *testing.T) {
b.answerOn = func(a Ask) json.RawMessage {
return json.RawMessage(`{"error": "-4: the document server could not download the file"}`)
}
if _, err := s.Convert(context.Background(), get, "doc", put); err == nil || !strings.Contains(err.Error(), "-4: the document server") {
if _, err := s.Convert(context.Background(), get, "doc", put, ""); err == nil || !strings.Contains(err.Error(), "-4: the document server") {
t.Fatalf("error %v", err)
}
}
@@ -83,12 +83,12 @@ func TestConvertAnswersTheConvertersError(t *testing.T) {
func TestConvertRefusesOtherTypes(t *testing.T) {
s, b := rig(time.Second)
for _, typ := range []string{"exe", "pdf", "", "docm", "jpg"} {
_, err := s.Convert(context.Background(), get, typ, put)
_, err := s.Convert(context.Background(), get, typ, put, "")
if err == nil || !strings.Contains(err.Error(), "docx") || !strings.Contains(err.Error(), "txt") {
t.Errorf("type %q: %v", typ, err)
}
}
if _, err := s.Convert(context.Background(), "file:///etc/passwd", "docx", put); err == nil {
if _, err := s.Convert(context.Background(), "file:///etc/passwd", "docx", put, ""); err == nil {
t.Error("a file link was taken")
}
if len(b.asks) != 0 {
@@ -102,7 +102,7 @@ func TestConvertRefusesOtherTypes(t *testing.T) {
// 1595/AC3: with no answer within the bound the verb says it timed out, with the ask id.
func TestConvertTimesOutNamingTheAsk(t *testing.T) {
s, b := rig(30 * time.Millisecond)
_, err := s.Convert(context.Background(), get, "docx", put)
_, err := s.Convert(context.Background(), get, "docx", put, "")
if err == nil || !strings.Contains(err.Error(), "timed out") || !strings.Contains(err.Error(), b.asks[0].ID) {
t.Fatalf("error %v", err)
}
@@ -0,0 +1,309 @@
package main
// The converter process (novox/hq ADR 0468 §7, ADR 0436 §1, §3, §6; issue 1596). It runs under an account of its own
// and a bus account of its own, is the one consumer of the work queue convert-asks, serves no verb, and alone reads
// the document server's token (the JWT). For each ask it has the document server turn the source into a PDF, fetches
// the PDF from the document server and puts it to the ask's target link, and writes {size, sha256} -- or the error --
// under the ask's id in the state convert-answers, where the tool side reads it.
import (
"bytes"
"context"
"crypto/hmac"
"crypto/rand"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"os"
"os/signal"
"strings"
"syscall"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/objectlink"
"git.novox.be/novox/mesh-sdk/go/process"
"github.com/nats-io/nats.go/jetstream"
)
const (
// AsksStream is the work-queue stream's bus name: the manifest's convert-asks, capitalised with its hyphen an underscore.
AsksStream = "CONVERT_ASKS"
// ConvertPath is the document server's conversion service, reached on the machine's own published port.
ConvertPath = "/ConvertService.ashx"
// StaleAfter is how old an ask may be when the converter takes it: older than twice the longest wait, nobody waits.
StaleAfter = 2 * ConvertWait
// ConvertRun is how long one conversion, its fetch and its put may take.
ConvertRun = 4 * time.Minute
)
// errorMeanings are the conversion service's error codes. Read from the conversion API's documentation, page
// https://api.onlyoffice.com/docs/docs-api/additional-api/conversion-api/error-codes/ on 2026-10-21.
var errorMeanings = map[int]string{
-1: "unknown error",
-2: "conversion timeout error",
-3: "conversion error",
-4: "error while downloading the document file to be converted",
-5: "incorrect password",
-6: "error while accessing the conversion result database",
-7: "input error",
-8: "invalid token",
-9: "the converter cannot determine the output file format",
-10: "size limit exceeded",
}
// meaningOf is an error code with its meaning, "-4: error while ...".
func meaningOf(code int) string {
if m, ok := errorMeanings[code]; ok {
return fmt.Sprintf("%d: %s", code, m)
}
return fmt.Sprintf("%d: a code the conversion API does not document", code)
}
// jwtHS256 signs a payload as a JSON Web Token with HMAC-SHA256.
func jwtHS256(secret string, payload []byte) string {
enc := base64.RawURLEncoding
signing := enc.EncodeToString([]byte(`{"alg":"HS256","typ":"JWT"}`)) + "." + enc.EncodeToString(payload)
mac := hmac.New(sha256.New, []byte(secret))
mac.Write([]byte(signing))
return signing + "." + enc.EncodeToString(mac.Sum(nil))
}
type answerStore interface {
Create(key string, value any) (uint64, error)
}
type action int
const (
actionAck action = iota
actionTerm
actionRetry
)
// converterWorker answers asks. Everything it touches is handed in, so a test drives it with fakes.
type converterWorker struct {
// JWT reads the document server's secret; only this process calls it.
JWT func() (string, error)
// Server is the document server's address, "http://127.0.0.1:9070".
Server string
// Docs talks to the document server, Store to the object store.
Docs, Store *http.Client
Answers answerStore
Now func() time.Time
Logf func(string, ...any)
NewKey func() string
}
func (w *converterWorker) logf(f string, a ...any) {
if w.Logf != nil {
w.Logf(f, a...)
}
}
// Handle takes one queued ask: it answers it under its id and says how the message is settled.
func (w *converterWorker) Handle(body []byte) (action, time.Duration) {
var ask Ask
if err := json.Unmarshal(body, &ask); err != nil || ask.ID == "" {
w.logf("[only-office] converter: a queued ask could not be read: %v", err)
return actionTerm, 0
}
if age := w.Now().Sub(ask.At); age > StaleAfter {
w.logf("[only-office] converter: ask %s is %s old; nobody waits for it", ask.ID, age.Round(time.Second))
return actionAck, 0
}
if !ask.Deadline.IsZero() && w.Now().After(ask.Deadline) {
w.answer(ask.ID, map[string]any{"error": "the ask was taken up after its deadline: nothing was converted for it; ask again"})
return actionTerm, 0
}
ctx, cancel := context.WithTimeout(context.Background(), ConvertRun)
defer cancel()
out, err := w.convert(ctx, ask)
if err != nil {
out = map[string]any{"error": err.Error()}
}
if err := w.answer(ask.ID, out); err != nil {
return actionRetry, 5 * time.Second
}
return actionAck, 0
}
func (w *converterWorker) answer(id string, v map[string]any) error {
if _, err := w.Answers.Create(id, v); err != nil && !errors.Is(err, stdio.ErrChanged) {
w.logf("[only-office] converter: the answer to %s could not be written: %v", id, err)
var api *jetstream.APIError
if errors.As(err, &api) {
return nil // a refused value would be refused again
}
return err
}
return nil
}
// convert does one ask: the document server converts, then the PDF is put to the target.
func (w *converterWorker) convert(ctx context.Context, ask Ask) (map[string]any, error) {
if ask.Op != OpConvert {
return nil, fmt.Errorf("the converter does not know the operation %q", ask.Op)
}
if _, err := cleanType(ask.Type); err != nil {
return nil, err
}
secret, err := w.JWT()
if err != nil {
return nil, fmt.Errorf("the document server's token cannot be read: %w", err)
}
key := ask.SHA256
if key == "" {
key = w.NewKey()
}
req := map[string]any{"async": false, "filetype": ask.Type, "key": key, "outputtype": "pdf", "url": ask.Source}
payload, _ := json.Marshal(req)
req["token"] = jwtHS256(secret, payload) // JWT_IN_BODY: the document server reads the token from the body
reqBody, _ := json.Marshal(req)
hreq, err := http.NewRequestWithContext(ctx, http.MethodPost, w.Server+ConvertPath, bytes.NewReader(reqBody))
if err != nil {
return nil, err
}
hreq.Header.Set("Content-Type", "application/json")
hreq.Header.Set("Accept", "application/json")
resp, err := w.Docs.Do(hreq)
if err != nil {
return nil, fmt.Errorf("the document server cannot be reached: %w", err)
}
defer resp.Body.Close()
raw, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
if err != nil {
return nil, fmt.Errorf("the document server's answer cannot be read: %w", err)
}
if resp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("the document server answered %d", resp.StatusCode)
}
var out struct {
EndConvert bool `json:"endConvert"`
FileURL string `json:"fileUrl"`
Error int `json:"error"`
}
if err := json.Unmarshal(raw, &out); err != nil {
return nil, fmt.Errorf("the document server's answer is not JSON: %w", err)
}
if out.Error != 0 {
return nil, fmt.Errorf("the document server refused the conversion, error %s", meaningOf(out.Error))
}
if !out.EndConvert || out.FileURL == "" {
return nil, errors.New("the document server did not finish the conversion: its answer has no file")
}
return w.deliver(ctx, out.FileURL, ask.Target)
}
// deliver fetches the PDF from the document server into a temporary file and puts it to the target.
func (w *converterWorker) deliver(ctx context.Context, fileURL, target string) (map[string]any, error) {
greq, err := http.NewRequestWithContext(ctx, http.MethodGet, fileURL, nil)
if err != nil {
return nil, err
}
gresp, err := w.Docs.Do(greq)
if err != nil {
return nil, fmt.Errorf("the converted file cannot be fetched from the document server: %w", err)
}
defer gresp.Body.Close()
if gresp.StatusCode != http.StatusOK {
return nil, fmt.Errorf("the document server answered %d for the converted file", gresp.StatusCode)
}
tmp, err := os.CreateTemp("", "converted-*.pdf")
if err != nil {
return nil, fmt.Errorf("a temporary file cannot be made: %w", err)
}
defer os.Remove(tmp.Name())
defer tmp.Close()
n, err := io.Copy(tmp, io.LimitReader(gresp.Body, objectlink.MaxDocument+1))
if err != nil {
return nil, fmt.Errorf("the converted file cannot be read: %w", err)
}
if n > objectlink.MaxDocument {
return nil, fmt.Errorf("the converted file is over %d bytes (novox/hq ADR 0468 §3)", objectlink.MaxDocument)
}
if _, err := tmp.Seek(0, io.SeekStart); err != nil {
return nil, err
}
res, err := objectlink.Put(ctx, w.Store, target, tmp, n, objectlink.MaxDocument)
if err != nil {
return nil, fmt.Errorf("the store did not take the PDF: %w", err)
}
return map[string]any{"size": res.Size, "sha256": res.SHA256}, nil
}
func newKey() string {
b := make([]byte, 16)
rand.Read(b)
return hex.EncodeToString(b)
}
func readSecret(path string) func() (string, error) {
return func() (string, error) {
raw, err := os.ReadFile(path)
if err != nil {
return "", err
}
s := strings.TrimSpace(string(raw))
if s == "" {
return "", fmt.Errorf("%s is empty", path)
}
return s, nil
}
}
// converter is `only-office converter`: it reads its own bus account, connects, and consumes convert-asks until stopped.
func converter() int {
brokerFile, jwtFile, server := os.Getenv("ONLY_OFFICE_BROKER"), os.Getenv("ONLY_OFFICE_JWT"), os.Getenv("ONLY_OFFICE_SERVER")
if brokerFile == "" || jwtFile == "" || server == "" {
fmt.Fprintln(os.Stderr, "[only-office] converter was started without ONLY_OFFICE_BROKER, ONLY_OFFICE_JWT or ONLY_OFFICE_SERVER")
return 2
}
account, err := process.ReadAccount(brokerFile)
if err != nil {
fmt.Fprintf(os.Stderr, "[only-office] converter: %v\n", err)
return 1
}
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
w := &converterWorker{JWT: readSecret(jwtFile), Server: strings.TrimRight(server, "/"),
Docs: &http.Client{Timeout: ConvertRun}, Store: &http.Client{Timeout: ConvertRun},
Now: time.Now, NewKey: newKey, Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, f+"\n", a...) }}
fmt.Fprintf(os.Stderr, "[only-office] converter is consuming %s as %s\n", AsksStream, account.User)
if err := runConverter(ctx, account, w); err != nil && ctx.Err() == nil {
fmt.Fprintf(os.Stderr, "[only-office] converter stopped: %v\n", err)
return 1
}
return 0
}
// runConverter connects as the converter's own account and consumes convert-asks until ctx ends, answering each ask in
// the state convert-answers. Apart from converter so a test can drive it over a bus of its own.
func runConverter(ctx context.Context, account process.Account, w *converterWorker) error {
conn, err := process.Connect(account)
if err != nil {
return fmt.Errorf("the bus cannot be reached: %w", err)
}
defer conn.Close()
w.Answers = conn.State(AnswersState)
return conn.Consume(ctx, AsksStream, func(msg *process.Message) {
act, delay := w.Handle(msg.Body)
var err error
switch act {
case actionAck:
err = msg.Ack()
case actionRetry:
err = msg.NakWithDelay(delay)
case actionTerm:
err = msg.Term()
}
if err != nil {
w.logf("[only-office] converter: settling %s: %v", msg.Subject, err)
}
})
}
@@ -0,0 +1,98 @@
//go:build integration
package main
// 1596/AC6: the tool side and the converter over a real bus (the shape of brother's device_integration_test.go,
// ADR 0436 §7): a JetStream server of the test's own, the work-queue stream convert-asks and its consumer, the state
// convert-answers, and a fake document server. The tool side's Convert queues an ask; the converter takes it and
// answers; Convert returns the answer. Compiled by the merge gate's `go vet -tags integration`; never run on this machine.
import (
"context"
"encoding/json"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/process"
"github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
)
// convertAskedSubject is the wire subject the controller composes for the stream's one event:
// "mesh.stream." + module + ".event." + event (mesh-controller internal/broker/modstreams.go).
const convertAskedSubject = "mesh.stream.only-office.event." + EventAsked
// natsBus is the tool side's bus: the ask published on the stream, the answer read from the state's bucket.
type natsBus struct {
js jetstream.JetStream
kv jetstream.KeyValue
}
func (b natsBus) Publish(id string, body any) error {
raw, _ := json.Marshal(body)
_, err := b.js.Publish(context.Background(), convertAskedSubject, raw, jetstream.WithMsgID(id))
return err
}
func (b natsBus) Answer(id string) (json.RawMessage, error) {
e, err := b.kv.Get(context.Background(), id)
if err != nil {
return nil, nil
}
return e.Value(), nil
}
func TestTheToolSideAndTheConverterConvertOverARealBus(t *testing.T) {
opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}
s, err := server.NewServer(opts)
if err != nil {
t.Fatal(err)
}
go s.Start()
if !s.ReadyForConnections(30 * time.Second) {
t.Fatal("a bus for this test did not come up within 30s")
}
t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
ctx := context.Background()
nc, err := nats.Connect(s.ClientURL())
if err != nil {
t.Fatal(err)
}
defer nc.Close()
js, err := jetstream.New(nc)
if err != nil {
t.Fatal(err)
}
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: AsksStream, Subjects: []string{convertAskedSubject},
Retention: jetstream.WorkQueuePolicy, Discard: jetstream.DiscardNew, MaxAge: 24 * time.Hour, MaxBytes: 8 << 20}); err != nil {
t.Fatal(err)
}
account := process.Account{URL: s.ClientURL(), Node: "novox", Module: "only-office", User: "novox.only-office.converter", Password: "pw"}
if _, err := js.CreateConsumer(ctx, AsksStream, jetstream.ConsumerConfig{Durable: account.QueueConsumerName(),
AckPolicy: jetstream.AckExplicitPolicy, AckWait: 30 * time.Second, MaxDeliver: 3}); err != nil {
t.Fatal(err)
}
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: account.Module + "_" + AnswersState, TTL: 10 * time.Minute})
if err != nil {
t.Fatal(err)
}
d, tgt := newFakeDocs(t), newFakeTarget(t)
w, _ := worker(d)
runCtx, cancel := context.WithCancel(ctx)
defer cancel()
go runConverter(runCtx, account, w)
svc := &Service{Asks: natsBus{js: js, kv: kv}, Now: time.Now, Wait: 20 * time.Second, Poll: 50 * time.Millisecond}
got, err := svc.Convert(ctx, "http://novox.internal:20001/b/in.docx?X-Amz-Signature=a", "docx", tgt.srv.URL+"/b/in.pdf?sig=1", "")
if err != nil {
t.Fatal(err)
}
if got["size"] != float64(len(d.pdf)) || string(tgt.got) != string(d.pdf) {
t.Fatalf("answer %v, target got %q", got, tgt.got)
}
if len(d.requests) != 1 {
t.Fatalf("%d conversion requests", len(d.requests))
}
}
@@ -0,0 +1,273 @@
package main
import (
"crypto/hmac"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"os"
"strings"
"sync"
"testing"
"time"
)
const jwtSecret = "module-jwt-secret"
// fakeDocs stands for the document server's conversion service (the request and answer shapes are the conversion API's:
// https://api.onlyoffice.com/docs/docs-api/additional-api/conversion-api/request/ and .../error-codes/, read
// 2026-10-21) and the file it serves back.
type fakeDocs struct {
mu sync.Mutex
requests []map[string]any
accept string
answer string // the JSON the service answers; {fileUrl} is replaced by the file's address
pdf []byte
srv *httptest.Server
}
func newFakeDocs(t *testing.T) *fakeDocs {
d := &fakeDocs{pdf: []byte("%PDF-1.7 converted"), answer: `{"endConvert": true, "fileUrl": "{fileUrl}", "percent": 100}`}
d.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
d.mu.Lock()
defer d.mu.Unlock()
switch r.URL.Path {
case ConvertPath:
var m map[string]any
json.NewDecoder(r.Body).Decode(&m)
d.requests = append(d.requests, m)
d.accept = r.Header.Get("Accept")
io.WriteString(w, strings.ReplaceAll(d.answer, "{fileUrl}", "http://"+r.Host+"/cache/out.pdf"))
case "/cache/out.pdf":
w.Write(d.pdf)
default:
w.WriteHeader(404)
}
}))
t.Cleanup(d.srv.Close)
return d
}
type fakeTarget struct {
mu sync.Mutex
got []byte
puts int
status int
srv *httptest.Server
}
func newFakeTarget(t *testing.T) *fakeTarget {
f := &fakeTarget{status: 200}
f.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
f.mu.Lock()
defer f.mu.Unlock()
if f.status != 200 {
w.WriteHeader(f.status)
io.WriteString(w, `<Error><Code>AccessDenied</Code><Message>Access Denied.</Message></Error>`)
return
}
f.got, _ = io.ReadAll(r.Body)
f.puts++
}))
t.Cleanup(f.srv.Close)
return f
}
type answers struct {
mu sync.Mutex
m map[string]map[string]any
}
func (a *answers) Create(key string, value any) (uint64, error) {
a.mu.Lock()
defer a.mu.Unlock()
raw, _ := json.Marshal(value)
var v map[string]any
json.Unmarshal(raw, &v)
a.m[key] = v
return 1, nil
}
func worker(d *fakeDocs) (*converterWorker, *answers) {
a := &answers{m: map[string]map[string]any{}}
return &converterWorker{JWT: func() (string, error) { return jwtSecret, nil }, Server: d.srv.URL,
Docs: http.DefaultClient, Store: http.DefaultClient, Answers: a, Now: time.Now,
NewKey: func() string { return "fresh-key" }}, a
}
func askBody(t *testing.T, target string, sha string) []byte {
raw, err := json.Marshal(Ask{ID: "ask1", Op: OpConvert, At: time.Now(), Deadline: time.Now().Add(time.Minute),
Source: "http://novox.internal:20001/b/in.docx?X-Amz-Signature=a", Type: "docx", Target: target, SHA256: sha})
if err != nil {
t.Fatal(err)
}
return raw
}
// verifies a JWT the way a document server does: three parts, HS256 over the first two with the secret.
func verifiesJWT(t *testing.T, token string) map[string]any {
t.Helper()
parts := strings.Split(token, ".")
if len(parts) != 3 {
t.Fatalf("token %q", token)
}
mac := hmac.New(sha256.New, []byte(jwtSecret))
mac.Write([]byte(parts[0] + "." + parts[1]))
if base64.RawURLEncoding.EncodeToString(mac.Sum(nil)) != parts[2] {
t.Fatal("the token does not verify with the module's secret")
}
payload, err := base64.RawURLEncoding.DecodeString(parts[1])
if err != nil {
t.Fatal(err)
}
var m map[string]any
json.Unmarshal(payload, &m)
return m
}
// 1596/AC1: the server receives one conversion request with the source link, the type, outputtype pdf and a JWT that
// verifies with the module's secret, and the PDF it returns is PUT to target.
func TestTheConverterPostsTheRequestAndPutsThePDF(t *testing.T) {
d, tgt := newFakeDocs(t), newFakeTarget(t)
w, _ := worker(d)
act, _ := w.Handle(askBody(t, tgt.srv.URL+"/b/in.pdf?sig=1", strings.Repeat("ab", 32)))
if act != actionAck {
t.Fatalf("action %v", act)
}
if len(d.requests) != 1 {
t.Fatalf("%d conversion requests", len(d.requests))
}
r := d.requests[0]
if r["url"] != "http://novox.internal:20001/b/in.docx?X-Amz-Signature=a" || r["filetype"] != "docx" || r["outputtype"] != "pdf" ||
r["async"] != false || r["key"] != strings.Repeat("ab", 32) || d.accept != "application/json" {
t.Fatalf("request %v accept %q", r, d.accept)
}
claims := verifiesJWT(t, r["token"].(string))
for _, k := range []string{"url", "filetype", "outputtype", "key"} {
if claims[k] != r[k] {
t.Errorf("the token's %s is %v, the body's %v", k, claims[k], r[k])
}
}
if string(tgt.got) != string(d.pdf) || tgt.puts != 1 {
t.Fatalf("the target got %q in %d puts", tgt.got, tgt.puts)
}
}
// 1596/AC1: without a known SHA-256 the key is a fresh one.
func TestTheKeyIsFreshWhenTheSourceHashIsNotKnown(t *testing.T) {
d, tgt := newFakeDocs(t), newFakeTarget(t)
w, _ := worker(d)
w.Handle(askBody(t, tgt.srv.URL+"/x", ""))
if d.requests[0]["key"] != "fresh-key" {
t.Fatalf("key %v", d.requests[0]["key"])
}
}
// 1596/AC2: the answer under the ask id carries the PUT body's size and SHA-256.
func TestTheAnswerCarriesTheSizeAndHashOfWhatWasPut(t *testing.T) {
d, tgt := newFakeDocs(t), newFakeTarget(t)
w, a := worker(d)
w.Handle(askBody(t, tgt.srv.URL+"/x", ""))
sum := sha256.Sum256(d.pdf)
got := a.m["ask1"]
if got["size"] != float64(len(d.pdf)) || got["sha256"] != hex.EncodeToString(sum[:]) || got["error"] != nil {
t.Fatalf("answer %v", got)
}
}
// 1596/AC3: a server answer with error -4 is answered with the code and its meaning, and nothing is PUT.
func TestAServerErrorIsAnsweredWithItsMeaningAndNothingIsPut(t *testing.T) {
d, tgt := newFakeDocs(t), newFakeTarget(t)
d.answer = `{"error": -4}`
w, a := worker(d)
w.Handle(askBody(t, tgt.srv.URL+"/x", ""))
msg, _ := a.m["ask1"]["error"].(string)
if !strings.Contains(msg, "-4: error while downloading the document file to be converted") {
t.Fatalf("answer %v", a.m["ask1"])
}
if tgt.puts != 0 {
t.Fatal("something was put")
}
if got := meaningOf(-8); got != "-8: invalid token" {
t.Fatalf("-8 reads %q", got)
}
}
// 1596/AC4: a failed PUT is answered as a failure naming the store's refusal; the ask is not answered as done.
func TestAFailedPutIsAFailureNamingTheStoresRefusal(t *testing.T) {
d, tgt := newFakeDocs(t), newFakeTarget(t)
tgt.status = 403
w, a := worker(d)
w.Handle(askBody(t, tgt.srv.URL+"/x", ""))
got := a.m["ask1"]
msg, _ := got["error"].(string)
if !strings.Contains(msg, "403") || !strings.Contains(msg, "AccessDenied") || got["size"] != nil || got["sha256"] != nil {
t.Fatalf("answer %v", got)
}
}
// An ask whose deadline passed is not converted, and an unreadable ask is ended.
func TestALateAskIsNotConverted(t *testing.T) {
d, tgt := newFakeDocs(t), newFakeTarget(t)
w, a := worker(d)
w.Now = func() time.Time { return time.Now().Add(2 * time.Minute) }
if act, _ := w.Handle(askBody(t, tgt.srv.URL+"/x", "")); act != actionTerm {
t.Fatalf("action %v", act)
}
if len(d.requests) != 0 || a.m["ask1"]["error"] == nil {
t.Fatalf("requests %d, answer %v", len(d.requests), a.m["ask1"])
}
if act, _ := w.Handle([]byte("nonsense")); act != actionTerm {
t.Fatal("an unreadable ask was kept")
}
}
// 1596/AC5: the process converter has no serves, is the one consumer of convert-asks and writes convert-answers, and
// its account owns the JWT file.
func TestManifestDeclaresTheConverterProcess(t *testing.T) {
m := manifest(t)
var proc, account map[string]any
for _, r := range m["resources"].([]any) {
res := r.(map[string]any)
if res["type"] == "process" && res["name"] == "only-office-converter" {
proc = res
}
if res["type"] == "user" && res["name"] == "only-office-converter" {
account = res
}
}
if proc == nil || account == nil {
t.Fatalf("process %v account %v", proc, account)
}
if _, serves := proc["serves"]; serves {
t.Error("the converter serves")
}
bus := proc["bus"].(map[string]any)
if q := bus["queues"].([]any); len(q) != 1 || q[0] != "convert-asks" {
t.Errorf("queues %v", q)
}
if s := bus["state"].([]any); len(s) != 1 || s[0] != AnswersState {
t.Errorf("state %v", s)
}
if proc["user"] != "only-office-converter" || account["shell"] != "/usr/bin/nologin" {
t.Errorf("user %v shell %v", proc["user"], account["shell"])
}
jwt := m["own-secrets"].(map[string]any)["jwt"].(map[string]any)
if jwt["owner"] != "only-office-converter" || jwt["path"] != "${dir:state}/jwt.secret" {
t.Errorf("jwt %v", jwt)
}
broker := m["own-secrets"].(map[string]any)["broker-converter"].(map[string]any)
env := proc["env"].(map[string]any)
if broker["owner"] != "only-office-converter" || env["ONLY_OFFICE_BROKER"] != broker["path"] || env["ONLY_OFFICE_JWT"] != jwt["path"] {
t.Errorf("broker %v env %v", broker, env)
}
// The container still gets the token: server.env names it.
raw, _ := os.ReadFile("../../files/server.env")
if !strings.Contains(string(raw), "JWT_SECRET=${secret:jwt}") {
t.Error("the document server is no longer given the token")
}
}
@@ -14,6 +14,9 @@ import (
)
func main() {
if len(os.Args) > 1 && os.Args[1] == "converter" {
os.Exit(converter())
}
if err := stdio.Serve("", tools(newService())); err != nil {
fmt.Fprintf(os.Stderr, "[only-office] %v\n", err)
os.Exit(1)
+18 -1
View File
@@ -2,4 +2,21 @@ module only-office
go 1.26.0
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021122348-6c857e1c025f
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021122348-6c857e1c025f
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
require (
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
github.com/google/go-tpm v0.9.8 // indirect
github.com/klauspost/compress v1.20.0 // indirect
github.com/minio/highwayhash v1.0.4 // indirect
github.com/nats-io/jwt/v2 v2.8.1 // indirect
github.com/nats-io/nkeys v0.4.16 // indirect
github.com/nats-io/nuid v1.0.1 // indirect
golang.org/x/crypto v0.57.0 // indirect
golang.org/x/sys v0.48.0 // indirect
golang.org/x/time v0.15.0 // indirect
)
+25
View File
@@ -1,2 +1,27 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021122348-6c857e1c025f h1:7J/9jOfSfjfYpf8MOj96HwQRViQUkBNMb71RHP2mjn8=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021122348-6c857e1c025f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
+42 -2
View File
@@ -46,7 +46,14 @@
"route": "${dir:state}/route.json"
},
"own-secrets": {
"jwt": "${dir:state}/jwt.secret"
"jwt": {
"path": "${dir:state}/jwt.secret",
"owner": "only-office-converter"
},
"broker-converter": {
"path": "${dir:state}/broker-converter.secret",
"owner": "only-office-converter"
}
},
"listens": [
{
@@ -107,10 +114,16 @@
{
"id": "state",
"type": "directory",
"mode": "0700",
"mode": "0711",
"place": ".",
"unknown": "program"
},
{
"id": "converter-account",
"type": "user",
"name": "only-office-converter",
"shell": "/usr/bin/nologin"
},
{
"id": "server-env",
"type": "file",
@@ -187,6 +200,33 @@
"${dir:fonts}:/usr/share/fonts/truetype/custom"
],
"secrets-in-environment": "run-document-server.sh regenerates JWT_SECRET from the environment on every start and overwrites local.json; not convertible"
},
{
"id": "converter",
"type": "process",
"name": "only-office-converter",
"artifact": "tools",
"run": [
"./only-office",
"converter"
],
"user": "only-office-converter",
"env": {
"ONLY_OFFICE_BROKER": "${dir:state}/broker-converter.secret",
"ONLY_OFFICE_JWT": "${dir:state}/jwt.secret",
"ONLY_OFFICE_SERVER": "http://127.0.0.1:9070"
},
"bus": {
"queues": [
"convert-asks"
],
"state": [
"convert-answers"
]
},
"health": {
"kind": "unit"
}
}
],
"build": {