From 278014fd7938cd269c942988ad9036f69dc49c18 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 14:58:41 +0200 Subject: [PATCH] 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) --- .../only-office/cmd/only-office/convert.go | 16 +- .../cmd/only-office/convert_test.go | 10 +- .../only-office/cmd/only-office/converter.go | 309 ++++++++++++++++++ .../only-office/converter_integration_test.go | 98 ++++++ .../cmd/only-office/converter_test.go | 273 ++++++++++++++++ modules/only-office/cmd/only-office/main.go | 3 + modules/only-office/go.mod | 19 +- modules/only-office/go.sum | 25 ++ modules/only-office/module.json | 44 ++- 9 files changed, 786 insertions(+), 11 deletions(-) create mode 100644 modules/only-office/cmd/only-office/converter.go create mode 100644 modules/only-office/cmd/only-office/converter_integration_test.go create mode 100644 modules/only-office/cmd/only-office/converter_test.go diff --git a/modules/only-office/cmd/only-office/convert.go b/modules/only-office/cmd/only-office/convert.go index f9d9b8d5..7576c958 100644 --- a/modules/only-office/cmd/only-office/convert.go +++ b/modules/only-office/cmd/only-office/convert.go @@ -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")) }, }} } diff --git a/modules/only-office/cmd/only-office/convert_test.go b/modules/only-office/cmd/only-office/convert_test.go index aa2b670b..4c72e180 100644 --- a/modules/only-office/cmd/only-office/convert_test.go +++ b/modules/only-office/cmd/only-office/convert_test.go @@ -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) } diff --git a/modules/only-office/cmd/only-office/converter.go b/modules/only-office/cmd/only-office/converter.go new file mode 100644 index 00000000..2cdc9fbb --- /dev/null +++ b/modules/only-office/cmd/only-office/converter.go @@ -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) + } + }) +} diff --git a/modules/only-office/cmd/only-office/converter_integration_test.go b/modules/only-office/cmd/only-office/converter_integration_test.go new file mode 100644 index 00000000..fd9378f1 --- /dev/null +++ b/modules/only-office/cmd/only-office/converter_integration_test.go @@ -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)) + } +} diff --git a/modules/only-office/cmd/only-office/converter_test.go b/modules/only-office/cmd/only-office/converter_test.go new file mode 100644 index 00000000..fe40c820 --- /dev/null +++ b/modules/only-office/cmd/only-office/converter_test.go @@ -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, `AccessDeniedAccess Denied.`) + 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") + } +} diff --git a/modules/only-office/cmd/only-office/main.go b/modules/only-office/cmd/only-office/main.go index c4a95df9..63f31170 100644 --- a/modules/only-office/cmd/only-office/main.go +++ b/modules/only-office/cmd/only-office/main.go @@ -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) diff --git a/modules/only-office/go.mod b/modules/only-office/go.mod index 2d1d9157..b06230a9 100644 --- a/modules/only-office/go.mod +++ b/modules/only-office/go.mod @@ -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 +) diff --git a/modules/only-office/go.sum b/modules/only-office/go.sum index 9728ce93..cd9cb802 100644 --- a/modules/only-office/go.sum +++ b/modules/only-office/go.sum @@ -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= diff --git a/modules/only-office/module.json b/modules/only-office/module.json index 717d5bbb..2c5e6ca6 100644 --- a/modules/only-office/module.json +++ b/modules/only-office/module.json @@ -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": {