diff --git a/modules/brother/README.md b/modules/brother/README.md index ff4449cc..5d924552 100644 --- a/modules/brother/README.md +++ b/modules/brother/README.md @@ -25,6 +25,14 @@ the module runs no process of its own, and it runs as the node's operator accoun 3. Crops the glass page (a feeder page is a whole sheet and is never cropped): the lid's level is measured in the corner farthest from where documents lie, a pixel is document when it is darker than the lid, and the rows and columns holding enough of them give the box. No document, or one filling nearly the whole glass, keeps the whole image (`cropped` false), and so does `crop` false. + With `apart` true (opt-in; without it the glass gives one document, as it always did), several documents lying apart on the glass are each cut out and filed as a one-page PDF of its own (own + object, own `scan.added`, own file); the answer then lists them as `documents` (each with its `path`) and + `count`, in reading order, and one that could not be kept or filed is named in `failed` without holding back + the others. The rule (`Boxes` in `crop.go`): the picture is cut into blocks of 1 mm; a block is document when a + quarter of its pixels are darker than the lid; blocks not at least three across (dust, JPEG ringing) are dropped; + blocks within 10 mm of each other are one region, so a white band narrower than that does not split a document; + a region under 20 mm on a side is a stain, not a document; regions whose boxes overlap are one document (a frame + and the text inside it). One document, or none, is cropped exactly as before. 4. Files a PDF holding the JPEGs, one page for the glass and one per image for the feeder (so six for three sheets both sides), each page as large as its image really is: written whole beside its place as `.scan-*.tmp` and renamed to `scan-YYYY-MM-DD-HHMMSS.pdf` (local time); a taken name gets `-2`, `-3`, and no file is ever overwritten. @@ -55,9 +63,10 @@ makes it (mode 0755) when it files the first scan. | verb (tool) | does | |---|---| -| `scanner.scan` | `source` (`glass`, default, or `feeder`) and `duplex` (feeder only): asks the device process to scan, crop (glass only), keep the PDF in the bucket under its sha256 and say `scan.added`; then fetches it in parts, checks the sha256 and files it in `scan-folder`. Waits up to 180 s. Answers the event's body and `path` | +| `scanner.scan` | `source` (`glass`, default, or `feeder`), `duplex` (feeder only) and `apart` (glass only): asks the device process to scan, crop (glass only), keep the PDF in the bucket under its sha256 and say `scan.added`; then fetches it in parts, checks the sha256 and files it in `scan-folder`, in the background, so a scan is filed whether or not the caller still waits. A glass scan is waited for up to 180 s. Several glass pages in one PDF (a page turned over, a stapled document): `document` `start` scans page 1 into a new document and answers its id (nothing is filed yet), `document` with that id scans the next page, and `document` with the id and `finish` true closes it into one multi-page PDF, filed once with one `scan.added` (nothing is scanned). The pages are kept in the bucket under `pending//` (not in memory, so a restart of the device process loses none); a document that gets no page for 30 minutes is closed and filed so by the tool process (a restart of the tool process forgets that timer; `finish` still works). The pages in the bucket are not deleted, the bucket having no delete. A feeder scan is waited for 25 s (the mesh console gives up after 30 s); if it is not done, the answer is `started: true` with its `id`, and `scanner.scan {id}` reads the outcome (`state` running, done with the event's body and `path`, or failed with `why`). The device process ends a feeder job at 20 s a page (at least 3 min, at most 30 min). A scan whose PDF the tool could not fetch (over 100 MiB) is refused before it is kept or announced. Answers the event's body and `path` | | `scanner.document` | one part of a kept scan (`object`, `part` from 1): `part`, `parts`, `size`, `sha256` (of the whole), `data` (base64). Join the parts and check the sha256 | -| `scanner.status` | whether the scanner answers, its state (Idle or Processing) and its document feeder's state; a scanner that does not answer is `answers: false` with why. This is how its health is read: the module has no long-running resource to declare a health check on | +| `scanner.status` | whether the scanner answers, its state (Idle or Processing) and its document feeder's state; a scanner that does not answer is `answers: false` with why. | +| `brother_health` | the module's health, asked by the node-engine every 60 s (ADR 0240) as the tool check of the `watch` process, beside the device process's unit check: the certificates the printer serves, one for each key type (ECDSA and RSA, told apart by the cipher suites offered under TLS 1.2), read with no login and compared with the pinned `printer-certificate` list. Unhealthy when one is not pinned, `why` naming the new fingerprint, so the pin can be changed before the admin pages are needed. With `printer` or `printer-certificate` not given it answers `state: waits` with `waits` naming each setting (the SDK's health.Waiting, ADR 0283), not an error. Each handshake is bounded by 3 s and the whole check by 7 s. A printer that does not answer is healthy, with `why` saying nothing was compared (it may be switched off). The process's liveness is judged by the node-engine without a declaration | | `scanner.list` | the newest filed scans, newest first (`limit`, default 20). Read-only | ## The printer seat @@ -66,7 +75,7 @@ makes it (mode 0755) when it files the first scan. |---|---| | `printer.status` | without a login: over IPP (`Get-Printer-Attributes`, encoded by hand, on port 631) the state (idle, processing, stopped), the state reasons (such as `toner-low-warning`) and the markers' names, levels (percent; below 0 is unknown) and colours; over eSCL the scanner's state and its feeder's. A part that does not answer is `answers: false` with why. Needs `printer` and `scanner` | | `printer.upload` | one part of a document to print later (`upload` id, `part`, `parts`, `sha256` of the whole, `data` base64): stored as `uploads/.pdf` or `.jpg` once all parts came and the sha256 matches; a wrong sha256, a part out of range or a document that is neither PDF nor JPEG stores nothing | -| `printer.print` | prints a kept scan (`object`) or a completed upload (`upload`) as an IPP Print-Job (operation 0x0002, port 631) in its own format, `copies` 1 to 99 (default 1), `sides` one-sided, two-sided-long-edge or two-sided-short-edge (default one-sided), `media` (default `iso_a4_210x297mm`); answers the job's id and state | +| `printer.print` | prints a file on the machine the tool runs on (`path`, absolute, a regular file under the tool account's home folder after symbolic links are resolved; any other path is refused with the one message "path refused: only a file under the home folder", which says nothing of whether it exists: the tool reads it, uploads it in parts and prints it in one call; PDF or JPEG by its bytes, never its name), a kept scan (`object`) or a completed upload (`upload`, for modules that send bytes over the bus) as an IPP Print-Job (operation 0x0002, port 631) in its own format, `copies` 1 to 99 (default 1), `sides` one-sided, two-sided-long-edge or two-sided-short-edge (default one-sided), `media` (default `iso_a4_210x297mm`); answers the job's id and state (with `path`, the upload id and the size when `path` was given) | | `printer.settings` | asks the device process (below) to log in, read one admin page and log out, and waits up to 60 s for its answer. Argument `page`: a path on the admin site, starting with `/` and ending in `.html` (default `/general/status.html`). Answers the page's title and forms with their fields, each with its current value (a radio group as one field with the checked button's value, a checkbox with `checked`, a select with the selected option) and `options`; the page's links only with `links: true`, hidden fields only with `hidden: true`; a password field's value is never given. Read-only: it sends the login and the logout and nothing else | ### Trust and the password @@ -104,7 +113,7 @@ writes each answer, once, into the state `device-answers` under the ask's id (te title, links and form fields, or `error`. It alone reads the password, pins the certificate, logs in, reads the page and logs out. An ask older than two minutes is dropped without contacting the printer; at most three deliveries. It reads `settings.json` (mode 0644, no secret in it) and the state directory is 0711 so that it can -reach its own files; both are the choices that let its account read what it needs. Its health is its unit. +reach its own files; both are the choices that let its account read what it needs. The device process is judged by its unit; `brother_health` (the printer's certificates against the pins) is the tool check of the `watch` process beside it. The login reads the form each time: the device generates the password field's name, so the code takes the name from the form, posts it with the form's hidden fields, and keeps the session cookie. A login is failed when the answer is diff --git a/modules/brother/cmd/brother/admin.go b/modules/brother/cmd/brother/admin.go index 79114abe..f6524530 100644 --- a/modules/brother/cmd/brother/admin.go +++ b/modules/brother/cmd/brother/admin.go @@ -2,6 +2,7 @@ package main import ( "context" + "errors" "fmt" "io" "net/http" @@ -40,6 +41,9 @@ func NewAdmin(host, fingerprint, password string) (*Admin, error) { HTTP: &http.Client{ Timeout: 30 * time.Second, Jar: jar, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + return checkRedirect(host, req, via) + }, Transport: &http.Transport{ TLSClientConfig: pinnedTLS(fingerprint), DisableKeepAlives: true, @@ -48,6 +52,21 @@ func NewAdmin(host, fingerprint, password string) (*Admin, error) { }, nil } +// checkRedirect follows a redirect only on https on the printer's own host, and never with a body: a 307 or 308 +// answer to the login post would make the client send the post again, the password with it (issue 1515). +func checkRedirect(host string, req *http.Request, via []*http.Request) error { + if len(via) >= 10 { + return errors.New("the printer redirected more than ten times") + } + if req.URL.Scheme != "https" || req.URL.Host != host { + return fmt.Errorf("the printer redirected to %s, not to https on the printer %s: not followed", req.URL.Scheme+"://"+req.URL.Host, host) + } + if req.Method != http.MethodGet && req.Method != http.MethodHead { + return fmt.Errorf("the printer redirected a %s with its body (a 307 or 308): not followed, so the password is not sent again", via[len(via)-1].Method) + } + return nil +} + func (a *Admin) url(path string) string { return "https://" + a.Host + path } // fetch is one request; the body is read whole, up to maxPage. The final URL is the one after redirects. diff --git a/modules/brother/cmd/brother/admin_test.go b/modules/brother/cmd/brother/admin_test.go index bd786582..5f033d96 100644 --- a/modules/brother/cmd/brother/admin_test.go +++ b/modules/brother/cmd/brother/admin_test.go @@ -300,3 +300,52 @@ func TestThePasswordIsNotPostedOverPlainHTTP(t *testing.T) { t.Errorf("%d posts reached the server", posted) } } + +// 1515: a 307 or 308 after the login post would make the HTTP client send the post again, the password with +// it, to the place the answer names; a redirect off https or off the printer's host is refused as well. +func TestNoRedirectResendsThePasswordOrLeavesHTTPSOnThePrinter(t *testing.T) { + const loginPage = `
` + cases := []struct { + name string + status int + location string // "SAME" is a path on the printer itself + }{ + {"307 on the printer", 307, "/again"}, + {"308 on the printer", 308, "/again"}, + {"307 to plain http", 307, "http://192.0.2.1/steal"}, + {"302 to another host", 302, "https://192.0.2.1/steal"}, + {"302 to plain http", 302, "http://192.0.2.1/steal"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + var mu sync.Mutex + var posts []string + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + defer mu.Unlock() + switch { + case r.Method == http.MethodGet && r.URL.Path == "/": + w.Write([]byte(loginPage)) + case r.Method == http.MethodPost: + r.ParseForm() + posts = append(posts, r.URL.Path+" "+r.PostForm.Get("pw")) + http.Redirect(w, r, c.location, c.status) + default: + w.Write([]byte("ok")) + } + })) + srv.StartTLS() + defer srv.Close() + a, _ := NewAdmin(strings.TrimPrefix(srv.URL, "https://"), Fingerprint(srv.TLS.Certificates[0].Certificate[0]), "hunter2") + err := a.Login(context.Background()) + if err == nil { + t.Fatalf("the login went on after a %d to %s", c.status, c.location) + } + mu.Lock() + defer mu.Unlock() + if len(posts) != 1 { + t.Errorf("the password was posted %d times: %v", len(posts), posts) + } + }) + } +} diff --git a/modules/brother/cmd/brother/capture.go b/modules/brother/cmd/brother/capture.go index c917874c..92fa4048 100644 --- a/modules/brother/cmd/brother/capture.go +++ b/modules/brother/cmd/brother/capture.go @@ -34,11 +34,29 @@ type Capture struct { type ScanRequest struct { Feeder bool Duplex bool + // Apart cuts each of several documents lying apart on the glass out as its own PDF (issue 1528); without it + // the glass gives one document, as it always did. + Apart bool } // ParseSource reads the scanner.scan arguments source (glass, the default, or feeder) and duplex. Duplex is for the // feeder: the glass holds one sheet. func ParseSource(source string, duplex bool) (ScanRequest, error) { + return ParseScan(source, duplex, false) +} + +// ParseScan is ParseSource with apart: several documents on the glass, each its own file; a feeder scan is of +// sheets and has none lying apart. +func ParseScan(source string, duplex, apart bool) (ScanRequest, error) { + req, err := parseSource(source, duplex) + if err == nil && apart && req.Feeder { + return ScanRequest{}, fmt.Errorf("apart is for the glass: the feeder takes each sheet as its own page already") + } + req.Apart = apart + return req, err +} + +func parseSource(source string, duplex bool) (ScanRequest, error) { switch source { case "", "glass": if duplex { @@ -51,15 +69,38 @@ func ParseSource(source string, duplex bool) (ScanRequest, error) { return ScanRequest{}, fmt.Errorf("source %q is not glass or feeder", source) } -// Capture scans the glass or the feeder and builds the PDF. The glass page is cropped when the settings say so; a -// feeder page is a whole sheet and never cropped. Nothing is kept and nothing is said here; an error says why -// the scanner gave no page. The resolution is the image's own, never the one asked (1440/AC2). -func (s *Service) Capture(ctx context.Context, req ScanRequest) (Capture, error) { +// Capture scans the glass or the feeder and builds the PDFs: one for a feeder job (every page of it), and for the +// glass one for each separate document lying on the platen when crop is set (issue 1528), in reading order. A glass +// page is cropped when the settings say so; a feeder page is a whole sheet and never cropped. Nothing is kept and +// nothing is said here; an error says why the scanner gave no page. The resolution is the image's own, never the +// one asked (1440/AC2). +func (s *Service) Capture(ctx context.Context, req ScanRequest) ([]Capture, error) { at := s.Now() scans, err := NewClient(s.Settings.Scanner).ScanPages(ctx, ScanOptions{Resolution: s.Settings.Resolution, ColorMode: s.Settings.ColorMode, Feeder: req.Feeder, Duplex: req.Duplex}) if err != nil { - return Capture{}, err + return nil, err + } + if !req.Feeder { + dpi, err := Density(scans[0]) + if err != nil { + return nil, fmt.Errorf("page 1: %w", err) + } + cuts, err := Crop(scans[0], s.Settings.Crop) + one := []Cropped{cuts} + if req.Apart { + one, err = CropAll(scans[0], true, dpi) + } + if err != nil { + return nil, err + } + out := make([]Capture, len(one)) + for i, cut := range one { + if out[i], err = captureOf([]PDFPage{{JPEG: cut.JPEG, DPI: dpi}}, cut, dpi, at); err != nil { + return nil, err + } + } + return out, nil } pages := make([]PDFPage, len(scans)) var first Cropped @@ -67,25 +108,26 @@ func (s *Service) Capture(ctx context.Context, req ScanRequest) (Capture, error) for i, scan := range scans { dpi, err := Density(scan) if err != nil { - return Capture{}, fmt.Errorf("page %d: %w", i+1, err) + return nil, fmt.Errorf("page %d: %w", i+1, err) } - cut := Cropped{JPEG: scan} - if !req.Feeder { - if cut, err = Crop(scan, s.Settings.Crop); err != nil { - return Capture{}, err - } - } else { - cfg, err := jpeg.DecodeConfig(bytes.NewReader(scan)) - if err != nil { - return Capture{}, fmt.Errorf("page %d: the scanned image cannot be read: %w", i+1, err) - } - cut.Width, cut.Height = cfg.Width, cfg.Height + cfg, err := jpeg.DecodeConfig(bytes.NewReader(scan)) + if err != nil { + return nil, fmt.Errorf("page %d: the scanned image cannot be read: %w", i+1, err) } - pages[i] = PDFPage{JPEG: cut.JPEG, DPI: dpi} + pages[i] = PDFPage{JPEG: scan, DPI: dpi} if i == 0 { - first, firstDPI = cut, dpi + first, firstDPI = Cropped{JPEG: scan, Width: cfg.Width, Height: cfg.Height}, dpi } } + c, err := captureOf(pages, first, firstDPI, at) + if err != nil { + return nil, err + } + return []Capture{c}, nil +} + +// captureOf is the PDF of the pages, with what the event says of it from the first page. +func captureOf(pages []PDFPage, first Cropped, firstDPI int, at time.Time) (Capture, error) { pdf, err := PDFPages(pages) if err != nil { return Capture{}, err @@ -112,3 +154,29 @@ func (c Capture) Body(node, colorMode string) map[string]any { "scanned-at": c.At.UTC().Format(time.RFC3339), } } + +// GlassImage is a page scanned from the glass. +type GlassImage struct { + JPEG []byte + DPI, Width, Height int + Cropped bool +} + +// GlassPage scans one page from the glass for a document of several pages: cropped to its document when the settings +// say so (the first document, never split in several), with the resolution the scanner gave. +func (s *Service) GlassPage(ctx context.Context) (page GlassImage, err error) { + scans, err := NewClient(s.Settings.Scanner).ScanPages(ctx, ScanOptions{Resolution: s.Settings.Resolution, ColorMode: s.Settings.ColorMode}) + if err != nil { + return page, err + } + dpi, err := Density(scans[0]) + if err != nil { + return page, err + } + cut, err := Crop(scans[0], s.Settings.Crop) + if err != nil { + return page, err + } + page.JPEG, page.DPI, page.Width, page.Height, page.Cropped = cut.JPEG, dpi, cut.Width, cut.Height, cut.Cropped + return page, nil +} diff --git a/modules/brother/cmd/brother/certcheck.go b/modules/brother/cmd/brother/certcheck.go new file mode 100644 index 00000000..f3e56453 --- /dev/null +++ b/modules/brother/cmd/brother/certcheck.go @@ -0,0 +1,143 @@ +package main + +import ( + "context" + "crypto/tls" + "fmt" + "net" + "slices" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/health" +) + +// The module's health (issue 1510, novox/hq ADR 0240): the node-engine asks the tool brother_health, and the tool +// asks the printer for the certificates it serves, one for each key type, with no login, and compares them with +// the pinned list (the config value printer-certificate). A certificate that is not pinned (the printer was reset, +// new certificates were made, a firmware update changed them) is unhealthy and names the new fingerprint, so the +// operator can approve the new pin before the admin pages are needed. A printer that does not answer is not +// unhealthy (it is switched off, or asleep): nothing can be compared, and the answer says so. + +// keyKinds are the two certificates the printer serves for one name. Which one it serves depends on the cipher +// suites the client offers, so each handshake offers only one kind's, under TLS 1.2 (where the suite decides). +var keyKinds = []struct { + Name string + Suites []uint16 +}{ + {"ecdsa", []uint16{ + tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256, tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384, + tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256, tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA, tls.TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA, + }}, + {"rsa", []uint16{ + tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256, tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384, + tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256, tls.TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA, tls.TLS_ECDHE_RSA_WITH_AES_256_CBC_SHA, + tls.TLS_RSA_WITH_AES_128_GCM_SHA256, tls.TLS_RSA_WITH_AES_256_GCM_SHA384, tls.TLS_RSA_WITH_AES_128_CBC_SHA, tls.TLS_RSA_WITH_AES_256_CBC_SHA, + }}, +} + +// certDial bounds one handshake and certTotal the whole check (two handshakes), so the node-engine's look, which has +// ten seconds, is always answered (issue 1510). +var ( + certDial = 3 * time.Second + certTotal = 7 * time.Second +) + +// Served is a certificate the printer served to a handshake of one key type. +type Served struct { + Key string `json:"key"` + Fingerprint string `json:"fingerprint,omitempty"` + Pinned bool `json:"pinned"` + Note string `json:"note,omitempty"` +} + +// CertHealth is brother_health's answer: healthy, why when it is not (or what could not be compared), and what the +// printer served. +type CertHealth struct { + Healthy bool `json:"healthy"` + State string `json:"state"` // healthy, unhealthy or waits (ADR 0283) + Why string `json:"why"` + Served []Served `json:"served"` +} + +// CertWaits is the answer while printer or printer-certificate is not given: the module waits for the operator, +// which is not a fault (ADR 0283), and names each setting as the SDK's health.Wait asks. +func CertWaits(s Settings) (health.Answer, bool) { + var waits []health.Wait + var names []string + for _, c := range []struct{ name, value, what string }{ + {"printer", s.Printer, "the printer's address"}, + {"printer-certificate", s.PrinterCertificate, "the SHA-256 fingerprints of the certificates the printer serves"}, + } { + if c.value == "" { + waits = append(waits, health.Wait{Part: "the printer's certificate check", Setting: c.name, What: c.what}) + names = append(names, c.name) + } + } + if len(waits) == 0 { + return health.Answer{}, false + } + return health.Waiting("waits for the operator to give the config value "+strings.Join(names, " and "), waits...), true +} + +// serveAddr is the host with the HTTPS port when it names none. +func serveAddr(host string) string { + if _, _, err := net.SplitHostPort(host); err == nil { + return host + } + return net.JoinHostPort(strings.Trim(host, "[]"), "443") +} + +// served is the leaf's fingerprint the printer gives a handshake that offers only these suites, or why it gave none. +func served(ctx context.Context, addr string, suites []uint16) (string, error) { + ctx, cancel := context.WithTimeout(ctx, certDial) + defer cancel() + d := tls.Dialer{NetDialer: &net.Dialer{Timeout: certDial}, Config: &tls.Config{ + MinVersion: tls.VersionTLS12, MaxVersion: tls.VersionTLS12, CipherSuites: suites, + // Nothing is sent over this connection, and the certificate is only read, to be compared with the pins. + InsecureSkipVerify: true, + }} + conn, err := d.DialContext(ctx, "tcp", addr) + if err != nil { + return "", err + } + defer conn.Close() + state := conn.(*tls.Conn).ConnectionState() + if len(state.PeerCertificates) == 0 { + return "", fmt.Errorf("the printer served no certificate") + } + return Fingerprint(state.PeerCertificates[0].Raw), nil +} + +// CheckCertificates compares what the printer serves with the pins. The login is never used. +func CheckCertificates(ctx context.Context, host, pins string) CertHealth { + ctx, cancel := context.WithTimeout(ctx, certTotal) + defer cancel() + want := pinList(pins) + addr := serveAddr(host) + h := CertHealth{Healthy: true, State: "healthy"} + var changed, unreachable []string + for _, k := range keyKinds { + s := Served{Key: k.Name} + fp, err := served(ctx, addr, k.Suites) + if err != nil { + s.Note = err.Error() + unreachable = append(unreachable, k.Name+": "+err.Error()) + } else { + s.Fingerprint, s.Pinned = fp, slices.Contains(want, fp) + if !s.Pinned { + changed = append(changed, fmt.Sprintf("the %s certificate is now %s", k.Name, fp)) + } + } + h.Served = append(h.Served, s) + } + switch { + case len(changed) > 0: + h.Healthy, h.State = false, "unhealthy" + h.Why = fmt.Sprintf("the printer serves a certificate that is not pinned: %s; the pinned printer-certificate is %s. If the printer was reset or given new certificates, set printer-certificate to the new fingerprint(s) (with the old ones kept if they still serve)", + strings.Join(changed, "; "), strings.Join(want, ", ")) + case len(unreachable) == len(keyKinds): + h.Why = "the printer gave no certificate, so none was compared (it may be switched off or asleep): " + strings.Join(unreachable, "; ") + } + return h +} diff --git a/modules/brother/cmd/brother/certcheck_test.go b/modules/brother/cmd/brother/certcheck_test.go new file mode 100644 index 00000000..3f0cf2f9 --- /dev/null +++ b/modules/brother/cmd/brother/certcheck_test.go @@ -0,0 +1,160 @@ +package main + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/rsa" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/json" + "math/big" + "net" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/health" +) + +// selfSigned is a self-signed certificate over the key. +func selfSigned(t *testing.T, key any, pub any) tls.Certificate { + t.Helper() + tpl := &x509.Certificate{SerialNumber: big.NewInt(time.Now().UnixNano()), Subject: pkix.Name{CommonName: "printer.test"}, + NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(24 * time.Hour), DNSNames: []string{"printer.test"}} + der, err := x509.CreateCertificate(rand.Reader, tpl, tpl, pub, key) + if err != nil { + t.Fatal(err) + } + return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key} +} + +// twoCertPrinter is a TLS server with an ECDSA and an RSA certificate, as the printer has; it answers its address +// and the two fingerprints. +func twoCertPrinter(t *testing.T) (addr, ecdsaFP, rsaFP string) { + t.Helper() + ek, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + rk, _ := rsa.GenerateKey(rand.Reader, 2048) + ec, rc := selfSigned(t, ek, &ek.PublicKey), selfSigned(t, rk, &rk.PublicKey) + ln, err := tls.Listen("tcp", "127.0.0.1:0", &tls.Config{Certificates: []tls.Certificate{ec, rc}}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { ln.Close() }) + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + go func() { c.(*tls.Conn).Handshake(); c.Close() }() + } + }() + return ln.Addr().String(), Fingerprint(ec.Certificate[0]), Fingerprint(rc.Certificate[0]) +} + +// 1510: both certificates pinned is healthy; a changed one is unhealthy and names the new fingerprint; none +// pinned names both. +func TestTheHealthComparesBothServedCertificatesWithThePins(t *testing.T) { + addr, e, r := twoCertPrinter(t) + ctx := context.Background() + if h := CheckCertificates(ctx, addr, e+", "+r); !h.Healthy || len(h.Served) != 2 || !h.Served[0].Pinned || !h.Served[1].Pinned { + t.Errorf("both pinned: %+v", h) + } + old := strings.Repeat("AB:", 31) + "AB" + h := CheckCertificates(ctx, addr, e+", "+old) // the RSA certificate changed + if h.Healthy || !strings.Contains(h.Why, r) || strings.Contains(h.Why, "ecdsa certificate is now") { + t.Errorf("rsa changed: %+v", h) + } + h = CheckCertificates(ctx, addr, old+", "+r) // the ECDSA certificate changed + if h.Healthy || !strings.Contains(h.Why, e) || strings.Contains(h.Why, "rsa certificate is now") { + t.Errorf("ecdsa changed: %+v", h) + } + h = CheckCertificates(ctx, addr, old) + if h.Healthy || !strings.Contains(h.Why, e) || !strings.Contains(h.Why, r) || !strings.Contains(h.Why, old) { + t.Errorf("none pinned: %+v", h) + } +} + +// 1510: a printer that does not answer is not an unhealthy module: nothing can be compared, and the answer says so. +func TestAPrinterThatDoesNotAnswerIsNotUnhealthy(t *testing.T) { + ln, _ := net.Listen("tcp", "127.0.0.1:0") + addr := ln.Addr().String() + ln.Close() + h := CheckCertificates(context.Background(), addr, strings.Repeat("AB:", 31)+"AB") + if !h.Healthy || !strings.Contains(h.Why, "none was compared") { + t.Errorf("%+v", h) + } +} + +// 1510: brother_health is a tool of the module, read from the settings at each call. +func TestBrotherHealthIsAToolThatReadsTheSettings(t *testing.T) { + addr, e, r := twoCertPrinter(t) + js := `{"printer":"` + addr + `","printer-certificate":"` + e + `, ` + r + `"}` + run := tool(t, newRig(t, js, "pw").tl, "brother_health").Run + got, err := run(map[string]any{}) + if err != nil { + t.Fatal(err) + } + if h, ok := got.(CertHealth); !ok || !h.Healthy { + t.Errorf("answer = %+v", got) + } + bad := `{"printer":"` + addr + `","printer-certificate":"` + strings.Repeat("AB:", 31) + `AB"}` + got, _ = tool(t, newRig(t, bad, "pw").tl, "brother_health").Run(map[string]any{}) + if h := got.(CertHealth); h.Healthy { + t.Errorf("a wrong pin is healthy: %+v", h) + } +} + +// 1510: with the printer not given the answer is waits, read by the SDK as the engine reads it: state waits, each +// wait naming exactly one setting, a part and what is wanted; not an error and not unhealthy. +func TestHealthWaitsForTheSettingsTheOperatorHasYetToGive(t *testing.T) { + for js, want := range map[string][]string{`{}`: {"printer", "printer-certificate"}, `{"printer":"192.0.2.10"}`: {"printer-certificate"}} { + got, err := tool(t, newRig(t, js, "pw").tl, "brother_health").Run(map[string]any{}) + if err != nil { + t.Fatal(err) + } + raw, _ := json.Marshal(got) + a, err := health.Read(raw) + if err != nil || a.State != health.StateWaits || len(a.Waits) != len(want) { + t.Fatalf("%s: read as %+v, %v", js, a, err) + } + for i, w := range a.Waits { + if w.Setting != want[i] || w.Check() != nil { + t.Errorf("%s: wait %d = %+v (%v)", js, i, w, w.Check()) + } + } + } +} + +// 1510: each handshake is bounded by 3 s and the whole check by 7 s, and a printer that accepts and never answers +// ends the check at the bound with the module healthy (nothing compared). +func TestTheCheckIsBoundedAndAHungPrinterEndsItAtTheBound(t *testing.T) { + if certDial != 3*time.Second || certTotal != 7*time.Second { + t.Fatalf("bounds = %s and %s, want 3 s and 7 s", certDial, certTotal) + } + ln, _ := net.Listen("tcp", "127.0.0.1:0") + defer ln.Close() + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + defer c.Close() // held open, never answered + } + }() + oldD, oldT := certDial, certTotal + certDial, certTotal = 200*time.Millisecond, 500*time.Millisecond + t.Cleanup(func() { certDial, certTotal = oldD, oldT }) + start := time.Now() + h := CheckCertificates(context.Background(), ln.Addr().String(), strings.Repeat("AB:", 31)+"AB") + if took := time.Since(start); took > 700*time.Millisecond { + t.Errorf("the check took %s", took) + } + if !h.Healthy || !strings.Contains(h.Why, "none was compared") { + t.Errorf("%+v", h) + } +} diff --git a/modules/brother/cmd/brother/crop.go b/modules/brother/cmd/brother/crop.go index 62554ecb..47033c32 100644 --- a/modules/brother/cmd/brother/crop.go +++ b/modules/brother/cmd/brother/crop.go @@ -6,6 +6,7 @@ import ( "image" "image/color" "image/jpeg" + "sort" ) // Cropping cuts the document out of the platen. The scanner sends the whole glass, a document in one corner @@ -48,21 +49,7 @@ func Box(img image.Image) (image.Rectangle, bool) { if w < 2*margin || h < 2*margin { return b, false } - side := lidBlock - if side > w/2 { - side = w / 2 - } - if side > h/2 { - side = h / 2 - } - sum := 0 - for y := b.Max.Y - side; y < b.Max.Y; y++ { - for x := b.Max.X - side; x < b.Max.X; x++ { - sum += grayAt(img, x, y) - } - } - lid := int(float64(sum)/float64(side*side) + 0.5) - limit := lid - darker + limit := lidLimit(img) cols := make([]int, w) rows := make([]int, h) @@ -86,6 +73,28 @@ func Box(img image.Image) (image.Rectangle, bool) { return box, true } +// lidLimit is the level under which a pixel is document: the lid's level, measured in the corner block farthest +// from the origin, less darker. +func lidLimit(img image.Image) int { + b := img.Bounds() + w, h := b.Dx(), b.Dy() + side := lidBlock + if side > w/2 { + side = w / 2 + } + if side > h/2 { + side = h / 2 + } + sum := 0 + for y := b.Max.Y - side; y < b.Max.Y; y++ { + for x := b.Max.X - side; x < b.Max.X; x++ { + sum += grayAt(img, x, y) + } + } + lid := int(float64(sum)/float64(side*side) + 0.5) + return lid - darker +} + // span is the first and last index whose count is over the floor, or -1 when none is. func span(counts []int, floor int) (int, int) { first, last := -1, -1 @@ -136,3 +145,199 @@ func Crop(scan []byte, crop bool) (Cropped, error) { } return Cropped{JPEG: out.Bytes(), Width: box.Dx(), Height: box.Dy(), Cropped: true}, nil } + +// Several documents on the glass (issue 1528). Where the rows-and-columns box of Box would be one image around +// all of them, Boxes finds each document apart from the others: the picture is cut into blocks of about a +// millimetre, a block is document when a quarter of its pixels are darker than the lid (dust is not), blocks +// within gapMM of each other are one region (so a white area inside a document, or a gap in it, does not split +// it), and a region of at least minMM on both sides is a document. Regions whose boxes touch or overlap +// (a text block inside a document's frame) are one document. +const ( + // blockMM is the side of a block, in millimetres. + blockMM = 1.0 + // blockShare is the fraction of a block's pixels that must be document for the block to count. + blockShare = 0.25 + // gapMM is how near two regions must be to be one document, in millimetres: documents lying farther apart than + // this are separate ones. + gapMM = 10.0 + // minMM is the least a region may measure, on both sides, to be a document and not a speck or a stain. + minMM = 20.0 +) + +// Boxes is the box of each separate document on the platen, in reading order (top to bottom by the box's top, +// then left to right), at the image's resolution dpi. It is Box's white-detection (the lid's level and darker) on +// a grid of blocks. A box that fills (almost) the platen is no crop: then the whole image is the answer, as one +// box with found false. +func Boxes(img image.Image, dpi int) ([]image.Rectangle, bool) { + b := img.Bounds() + w, h := b.Dx(), b.Dy() + if dpi <= 0 || w < 2*margin || h < 2*margin { + return []image.Rectangle{b}, false + } + bs := max(2, int(blockMM*float64(dpi)/25.4+0.5)) + gw, gh := (w+bs-1)/bs, (h+bs-1)/bs + limit := lidLimit(img) + doc := make([]bool, gw*gh) + for gy := 0; gy < gh; gy++ { + for gx := 0; gx < gw; gx++ { + dark, total := 0, 0 + for y := b.Min.Y + gy*bs; y < min(b.Min.Y+(gy+1)*bs, b.Max.Y); y++ { + for x := b.Min.X + gx*bs; x < min(b.Min.X+(gx+1)*bs, b.Max.X); x++ { + total++ + if grayAt(img, x, y) < limit { + dark++ + } + } + } + doc[gy*gw+gx] = float64(dark) >= blockShare*float64(total) + } + } + // Dust and the ringing of the JPEG around it make single blocks, or a pair, of "document": an opening (every + // block must have all eight neighbours document to survive, and survivors give back their neighbours that are + // document) removes what is not at least three blocks across. + eroded := make([]bool, gw*gh) + for gy := 1; gy < gh-1; gy++ { + for gx := 1; gx < gw-1; gx++ { + all := true + for y := gy - 1; y <= gy+1 && all; y++ { + for x := gx - 1; x <= gx+1; x++ { + if !doc[y*gw+x] { + all = false + break + } + } + } + eroded[gy*gw+gx] = all + } + } + opened := make([]bool, gw*gh) + for gy := 0; gy < gh; gy++ { + for gx := 0; gx < gw; gx++ { + if !eroded[gy*gw+gx] { + continue + } + for y := gy - 1; y <= gy+1; y++ { + for x := gx - 1; x <= gx+1; x++ { + opened[y*gw+x] = true + } + } + } + } + doc = opened + // Join the blocks within the gap: a block is near when it is within gapBlocks/2 of a document block in each + // direction (so two blocks up to gapBlocks apart meet), then the nearby blocks are labelled in 4-connected + // regions; each region's box is that of the document blocks in it. + radius := int(gapMM * float64(dpi) / 25.4 / float64(bs) / 2) + near := make([]bool, gw*gh) + for gy := 0; gy < gh; gy++ { + for gx := 0; gx < gw; gx++ { + if !doc[gy*gw+gx] { + continue + } + for y := max(0, gy-radius); y <= min(gh-1, gy+radius); y++ { + for x := max(0, gx-radius); x <= min(gw-1, gx+radius); x++ { + near[y*gw+x] = true + } + } + } + } + seen := make([]bool, gw*gh) + var boxes []image.Rectangle + for start := range near { + if !near[start] || seen[start] { + continue + } + minX, minY, maxX, maxY := gw, gh, -1, -1 + stack := []int{start} + seen[start] = true + for len(stack) > 0 { + c := stack[len(stack)-1] + stack = stack[:len(stack)-1] + cx, cy := c%gw, c/gw + if doc[c] { + minX, minY, maxX, maxY = min(minX, cx), min(minY, cy), max(maxX, cx), max(maxY, cy) + } + for _, d := range [4][2]int{{1, 0}, {-1, 0}, {0, 1}, {0, -1}} { + x, y := cx+d[0], cy+d[1] + if x < 0 || y < 0 || x >= gw || y >= gh || !near[y*gw+x] || seen[y*gw+x] { + continue + } + seen[y*gw+x] = true + stack = append(stack, y*gw+x) + } + } + if maxX < 0 { + continue + } + r := image.Rect(minX*bs, minY*bs, (maxX+1)*bs, (maxY+1)*bs) + if float64(r.Dx())/float64(dpi)*25.4 < minMM || float64(r.Dy())/float64(dpi)*25.4 < minMM { + continue + } + boxes = append(boxes, r) + } + // Boxes that touch or overlap are one document: a frame and the text inside it, or a faint paper whose + // parts the lid's white hides, are one however far apart their blocks lie. + kept := boxes + for merged := true; merged; { + merged = false + for i := 0; i < len(kept) && !merged; i++ { + for j := i + 1; j < len(kept); j++ { + if kept[i].Overlaps(kept[j]) { + kept[i] = kept[i].Union(kept[j]) + kept = append(kept[:j], kept[j+1:]...) + merged = true + break + } + } + } + } + if len(kept) == 0 { + return []image.Rectangle{b}, false + } + out := make([]image.Rectangle, len(kept)) + for i, r := range kept { + out[i] = image.Rect(max(0, r.Min.X-margin), max(0, r.Min.Y-margin), min(w, r.Max.X+margin), min(h, r.Max.Y+margin)).Add(b.Min) + } + sort.SliceStable(out, func(i, j int) bool { + if out[i].Min.Y != out[j].Min.Y { + return out[i].Min.Y < out[j].Min.Y + } + return out[i].Min.X < out[j].Min.X + }) + if len(out) == 1 && float64(out[0].Dx()*out[0].Dy()) >= wholeShare*float64(w*h) { + return []image.Rectangle{b}, false + } + return out, true +} + +// CropAll cuts each separate document out of a scanned glass JPEG when crop is set and there are two or more; one +// document, none or crop off is Crop's answer, exactly as before. dpi is the scan's resolution. +func CropAll(scan []byte, crop bool, dpi int) ([]Cropped, error) { + one, err := Crop(scan, crop) + if err != nil || !crop { + return []Cropped{one}, err + } + img, err := jpeg.Decode(bytes.NewReader(scan)) + if err != nil { + return nil, fmt.Errorf("the scanned image cannot be read: %w", err) + } + boxes, found := Boxes(img, dpi) + if !found || len(boxes) < 2 { + return []Cropped{one}, nil + } + sub, ok := img.(interface { + SubImage(image.Rectangle) image.Image + }) + if !ok { + return []Cropped{one}, nil + } + out := make([]Cropped, 0, len(boxes)) + for _, box := range boxes { + var buf bytes.Buffer + if err := jpeg.Encode(&buf, sub.SubImage(box), &jpeg.Options{Quality: jpegQuality}); err != nil { + return nil, fmt.Errorf("the cropped image cannot be encoded: %w", err) + } + out = append(out, Cropped{JPEG: buf.Bytes(), Width: box.Dx(), Height: box.Dy(), Cropped: true}) + } + return out, nil +} diff --git a/modules/brother/cmd/brother/crop_test.go b/modules/brother/cmd/brother/crop_test.go index b61a263c..84b766a4 100644 --- a/modules/brother/cmd/brother/crop_test.go +++ b/modules/brother/cmd/brother/crop_test.go @@ -123,3 +123,107 @@ func TestCropFindsADocumentInTheMiddleOfTheGlass(t *testing.T) { t.Fatalf("box = %v, want about (200,300)-(500,700)", box) } } + +// onWhite is a lid-white image of w x h with the rectangles drawn on it in gray (generated, not captured: the +// thresholds of Boxes are in millimetres, and this is 100 dpi, 1 mm about 4 px). +func onWhite(t *testing.T, seed int64, rects ...image.Rectangle) []byte { + t.Helper() + g := lidWhite(800, 1100, seed) + for _, r := range rects { + draw.Draw(g, r, &image.Uniform{color.Gray{200}}, image.Point{}, draw.Src) + } + return encode(t, g) +} + +// 1528: documents lying apart on the white are each cut out, in reading order; one document, as before, is one. +func TestCropAllCutsEachDocumentApartFromTheOthers(t *testing.T) { + a := image.Rect(100, 100, 300, 300) // 51 mm square + b := image.Rect(450, 100, 650, 400) // 38 mm right of a + c := image.Rect(100, 600, 400, 900) + for name, tc := range map[string]struct { + rects []image.Rectangle + want []image.Rectangle + }{ + "one": {[]image.Rectangle{a}, nil}, + "two": {[]image.Rectangle{b, a}, []image.Rectangle{a, b}}, + "three": {[]image.Rectangle{c, b, a}, []image.Rectangle{a, b, c}}, + } { + got, err := CropAll(onWhite(t, 5, tc.rects...), true, 100) + if err != nil { + t.Fatal(err) + } + if tc.want == nil { + one, _ := Crop(onWhite(t, 5, a), true) + if len(got) != 1 || !bytes.Equal(got[0].JPEG, one.JPEG) { + t.Errorf("%s: one document is not Crop's answer", name) + } + continue + } + if len(got) != len(tc.want) { + t.Fatalf("%s: %d documents, want %d", name, len(got), len(tc.want)) + } + for i, w := range tc.want { + // about the rectangle, within the margin and a block of rounding + if !got[i].Cropped || got[i].Width < w.Dx() || got[i].Width > w.Dx()+2*margin+2*4 || got[i].Height < w.Dy() || got[i].Height > w.Dy()+2*margin+2*4 { + t.Errorf("%s: document %d is %dx%d, want about %dx%d", name, i, got[i].Width, got[i].Height, w.Dx(), w.Dy()) + } + cfg, err := jpeg.DecodeConfig(bytes.NewReader(got[i].JPEG)) + if err != nil || cfg.Width != got[i].Width { + t.Errorf("%s: document %d JPEG %+v %v", name, i, cfg, err) + } + } + } +} + +// 1528: one document with a white area inside it is not split: a frame with an island inside it, a sheet with a +// white band across it narrower than the gap, and specks of dust beside it. +func TestCropAllDoesNotSplitOneDocumentWithWhiteInsideIt(t *testing.T) { + g := lidWhite(800, 1100, 6) + frame := image.Rect(100, 100, 600, 800) + draw.Draw(g, frame, &image.Uniform{color.Gray{200}}, image.Point{}, draw.Src) + draw.Draw(g, frame.Inset(50), &image.Uniform{color.Gray{255}}, image.Point{}, draw.Src) // white inside the frame + draw.Draw(g, image.Rect(300, 400, 400, 500), &image.Uniform{color.Gray{200}}, image.Point{}, draw.Src) // an island in it + got, err := CropAll(encode(t, g), true, 100) + if err != nil { + t.Fatal(err) + } + if len(got) != 1 { + t.Fatalf("a frame with an island inside was cut into %d documents", len(got)) + } + + g = lidWhite(800, 1100, 7) + draw.Draw(g, image.Rect(100, 100, 500, 500), &image.Uniform{color.Gray{200}}, image.Point{}, draw.Src) + draw.Draw(g, image.Rect(100, 530, 500, 900), &image.Uniform{color.Gray{200}}, image.Point{}, draw.Src) // 30 px = 7.6 mm of white between + got, err = CropAll(encode(t, g), true, 100) + if err != nil || len(got) != 1 { + t.Fatalf("a sheet with a narrow white band across it: %d documents, %v", len(got), err) + } +} + +// 1528: a speck smaller than the minimum is no document, and crop off files the whole platen as one. +func TestCropAllIgnoresSpecksAndCropOffIsOneWholeImage(t *testing.T) { + scan := onWhite(t, 8, image.Rect(100, 100, 300, 300), image.Rect(600, 800, 620, 815)) // 5 x 4 mm stain + got, err := CropAll(scan, true, 100) + if err != nil || len(got) != 1 { + t.Fatalf("a stain made %d documents (%v)", len(got), err) + } + two := onWhite(t, 9, image.Rect(100, 100, 300, 300), image.Rect(450, 100, 650, 400)) + got, err = CropAll(two, false, 100) + if err != nil || len(got) != 1 || got[0].Cropped { + t.Fatalf("crop off: %d documents, cropped %v", len(got), got[0].Cropped) + } +} + +// 1528: the captured scan (one receipt) is one document, the same bytes Crop gives. +func TestCropAllOfTheCapturedReceiptIsCropsAnswer(t *testing.T) { + scan := fixture(t, "scan.jpg") + got, err := CropAll(scan, true, 200) + one, _ := Crop(scan, true) + if err != nil || len(got) != 1 || !bytes.Equal(got[0].JPEG, one.JPEG) { + t.Fatalf("%d documents (%v), want Crop's one", len(got), err) + } + img, _ := jpeg.Decode(bytes.NewReader(scan)) + if boxes, _ := Boxes(img, 200); len(boxes) != 1 { + t.Errorf("the receipt is %d regions: %v", len(boxes), boxes) + } +} diff --git a/modules/brother/cmd/brother/device_boundary_test.go b/modules/brother/cmd/brother/device_boundary_test.go index 5c355612..57964f42 100644 --- a/modules/brother/cmd/brother/device_boundary_test.go +++ b/modules/brother/cmd/brother/device_boundary_test.go @@ -17,7 +17,7 @@ const secretPackage = "brother/cmd/brother/internal/secret" // toolFiles are the files the tools' bundle runs (the scanner's, the printer.status's and the printer.settings' // ask), as opposed to the device process's: deviceproc.go, admin.go, pin.go and html.go. -var toolFiles = []string{"main.go", "printer.go", "scan.go", "settings.go", "ipp.go", "escl.go", "file.go", "crop.go", "pdf.go", "parts.go"} +var toolFiles = []string{"main.go", "printer.go", "scan.go", "settings.go", "ipp.go", "escl.go", "file.go", "crop.go", "pdf.go", "parts.go", "certcheck.go"} // 1453: none of the tools' files imports the password-reading package or names the password's file. func TestTheToolsNeverOpenTheAdminPassword(t *testing.T) { diff --git a/modules/brother/cmd/brother/deviceproc.go b/modules/brother/cmd/brother/deviceproc.go index 2910c1fc..134846af 100644 --- a/modules/brother/cmd/brother/deviceproc.go +++ b/modules/brother/cmd/brother/deviceproc.go @@ -13,6 +13,7 @@ import ( "fmt" "os" "os/signal" + "strings" "sync" "syscall" "time" @@ -64,7 +65,7 @@ type deviceWorker struct { Answers Answers Node string // Capture scans; nil is the device's own (a test puts a stand-in). - Capture func(ctx context.Context, req ScanRequest) (Capture, error) + Capture func(ctx context.Context, req ScanRequest) ([]Capture, error) // answerPart is the raw size of a document part in an answer; zero is AnswerPartSize (a test makes it larger). answerPart int Now func() time.Time @@ -72,6 +73,8 @@ type deviceWorker struct { mu sync.Mutex pending map[string]*pendingUpload + // docs serialises the work on open documents. + docs sync.Mutex } func (w *deviceWorker) logf(format string, a ...any) { @@ -99,12 +102,27 @@ func (w *deviceWorker) Handle(body []byte, early func()) (action, time.Duration) w.logf("[brother] device: ask %s is %s old; nobody waits for it, the device is not touched", ask.ID, age.Round(time.Second)) return actionAck, 0 } + if !ask.Deadline.IsZero() && w.Now().After(ask.Deadline) { + // The asker has stopped waiting for this ask to be taken up: nothing is scanned or printed for it (issue + // 1539). The reason is its answer, for a late reader, and the message ends here. + w.logf("[brother] device: ask %s was taken up after its deadline %s; the device is not touched", ask.ID, ask.Deadline.Format(time.RFC3339)) + if _, err := w.Answers.Create(ask.ID, map[string]any{"error": "the ask was taken up after its deadline (the device process was busy): nothing was done for it; ask again"}); err != nil && !errors.Is(err, stdio.ErrChanged) { + w.logf("[brother] device: the answer to %s could not be written: %v", ask.ID, err) + } + return actionTerm, 0 + } settled := false if (ask.Op == OpScan || ask.Op == OpPrint) && early != nil { early() settled = true } - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + // A scan of the feeder may run as long as FeederMaxBound (the scan itself ends sooner by FeederBound, which + // scales with its pages) and the PDF is then stored; every other ask is done in 5 minutes. + bound := 5 * time.Minute + if ask.Op == OpScan && ask.Source == "feeder" { + bound = FeederMaxBound + 5*time.Minute + } + ctx, cancel := context.WithTimeout(context.Background(), bound) defer cancel() var answer map[string]any if got, err := w.run(ctx, ask); err != nil { @@ -152,6 +170,8 @@ func (w *deviceWorker) run(ctx context.Context, ask Ask) (map[string]any, error) return w.upload(ctx, ask) case OpPrint: return w.print(ctx, ask) + case OpOpen: + return w.openDocuments(ctx) } return nil, fmt.Errorf("the ask %s names an operation, %q, the device process does not do", ask.ID, ask.Op) } @@ -161,7 +181,7 @@ func (w *deviceWorker) run(ctx context.Context, ask Ask) (map[string]any, error) // kept nor said when the scanner gives no page (1440/AC6). The answer is the event's body with key, size and the // number of parts. func (w *deviceWorker) scan(ctx context.Context, ask Ask) (map[string]any, error) { - req, err := ParseSource(ask.Source, ask.Duplex) + req, err := ParseScan(ask.Source, ask.Duplex, ask.Apart) if err != nil { return nil, err } @@ -177,14 +197,61 @@ func (w *deviceWorker) scan(ctx context.Context, ask Ask) (map[string]any, error return nil, err } svc := &Service{Settings: set, Node: w.Node, Now: w.Now} + if ask.Document != "" { + return w.scanDocument(ctx, ask, svc, objects) + } capture := svc.Capture if w.Capture != nil { capture = w.Capture } - c, err := capture(ctx, req) + caps, err := capture(ctx, req) if err != nil { return nil, err } + return w.keep(ctx, objects, set, caps) +} + +// keep stores each captured PDF in the bucket and says scan.added for it, in that order. One document is answered +// as today: the event's body with key, size and the number of parts. Several (documents lying apart on the glass) +// are answered as documents, a list of those, each filed on its own by the tool. A document that could not be kept +// or announced is named in failed and does not hold back the others: what is announced is always filed. +func (w *deviceWorker) keep(ctx context.Context, objects Objects, set Settings, caps []Capture) (map[string]any, error) { + if len(caps) == 0 { + return nil, fmt.Errorf("the scanner gave no document") + } + var docs []map[string]any + var failed []string + for i, c := range caps { + one, err := w.keepOne(ctx, objects, set, c) + if err != nil { + if len(caps) == 1 { + return nil, err + } + failed = append(failed, fmt.Sprintf("document %d: %v", i+1, err)) + continue + } + docs = append(docs, one) + } + if len(docs) == 0 { + return nil, fmt.Errorf("no document could be kept: %s", strings.Join(failed, "; ")) + } + if len(caps) == 1 { + return docs[0], nil + } + out := map[string]any{"documents": docs} + if len(failed) > 0 { + out["failed"] = failed + } + return out, nil +} + +// keepOne keeps one captured PDF and announces it. +func (w *deviceWorker) keepOne(ctx context.Context, objects Objects, set Settings, c Capture) (map[string]any, error) { + // A scan the tool could not take is not kept nor announced: an announced scan is always filed (issue 1539). + if most := MostAnswerParts * AnswerPartSize; len(c.PDF) > most { + return nil, fmt.Errorf("the scan's PDF is %d bytes, over the %d the tool can fetch (%d parts of %d): it is not kept; scan with a lower resolution or fewer sheets", + len(c.PDF), most, MostAnswerParts, AnswerPartSize) + } body := c.Body(w.Node, set.ColorMode) key := body["object"].(string) if err := Store(ctx, objects, key, "application/pdf", c.PDF); err != nil { @@ -500,3 +567,174 @@ func runDevice(ctx context.Context, account process.Account, w *deviceWorker) er } }) } + +// An open document (issue 1538) is kept in the bucket, not in memory, so that a restart of the device process +// loses none of its pages: each page's JPEG under pending//.jpg and a manifest, pending//manifest.json, +// that says how many pages there are and, once it is closed, the key of its PDF. The pages are not deleted when it +// is closed (the bucket has no delete); they are the bucket's to age out. +const mostDocumentPages = 100 + +type docPage struct { + Key string `json:"key"` + DPI int `json:"dpi"` + Width int `json:"width"` + Height int `json:"height"` + Cropped bool `json:"cropped"` +} + +type docManifest struct { + Started time.Time `json:"started"` + Updated time.Time `json:"updated"` // the last page's time + Pages []docPage `json:"pages"` + Finished string `json:"finished,omitempty"` // the closed document's object key +} + +func manifestKey(id string) string { return "pending/" + id + "/manifest.json" } + +func loadManifest(ctx context.Context, o Objects, id string) (*docManifest, error) { + has, err := o.Has(ctx, manifestKey(id)) + if err != nil { + return nil, err + } + if !has { + return nil, fmt.Errorf("no document %s is open: its id is wrong, or it was closed", id) + } + raw, err := o.Get(ctx, manifestKey(id)) + if err != nil { + return nil, err + } + var m docManifest + if err := json.Unmarshal(raw, &m); err != nil { + return nil, fmt.Errorf("document %s: its manifest cannot be read: %w", id, err) + } + if m.Finished != "" { + return nil, fmt.Errorf("document %s is closed already, as %s", id, m.Finished) + } + return &m, nil +} + +func saveManifest(ctx context.Context, o Objects, id string, m *docManifest) error { + raw, _ := json.Marshal(m) + return o.Put(ctx, manifestKey(id), "application/json", raw) +} + +// scanDocument adds a glass page to an open document (document start opens one, with the ask's id as its id), or +// with finish closes the document into one multi-page PDF, kept and announced once like any scan. +func (w *deviceWorker) scanDocument(ctx context.Context, ask Ask, svc *Service, objects Objects) (map[string]any, error) { + if ask.Source == "feeder" || ask.Duplex { + return nil, fmt.Errorf("a document of several pages is scanned from the glass, page by page; the feeder takes the whole stack in one scan") + } + id := ask.Document + if id == "start" { + if ask.Finish { + return nil, fmt.Errorf("finish closes an open document: name its id, not start") + } + id = ask.ID + } + if !uploadPattern.MatchString(id) { + return nil, fmt.Errorf("document %q is not an id: letters, digits, - and _, at most 64", id) + } + w.docs.Lock() + defer w.docs.Unlock() + var m *docManifest + if ask.Document == "start" { + m = &docManifest{Started: w.Now().UTC(), Updated: w.Now().UTC()} + } else { + var err error + if m, err = loadManifest(ctx, objects, id); err != nil { + return nil, err + } + } + if ask.Finish { + return w.finishDocument(ctx, id, m, svc, objects) + } + if len(m.Pages) >= mostDocumentPages { + return nil, fmt.Errorf("document %s has %d pages, the most one takes: close it", id, len(m.Pages)) + } + page, err := svc.GlassPage(ctx) + if err != nil { + return nil, err + } + key := fmt.Sprintf("pending/%s/%03d.jpg", id, len(m.Pages)+1) + if err := objects.Put(ctx, key, "image/jpeg", page.JPEG); err != nil { + return nil, err + } + m.Updated = w.Now().UTC() + m.Pages = append(m.Pages, docPage{Key: key, DPI: page.DPI, Width: page.Width, Height: page.Height, Cropped: page.Cropped}) + if err := saveManifest(ctx, objects, id, m); err != nil { + return nil, err + } + return map[string]any{"document": id, "pages": len(m.Pages), "state": "open", + "next": "scanner.scan {document: \"" + id + "\"} adds a page; scanner.scan {document: \"" + id + "\", finish: true} closes it into one PDF"}, nil +} + +// finishDocument builds the open document's pages into one PDF and keeps and announces it. +func (w *deviceWorker) finishDocument(ctx context.Context, id string, m *docManifest, svc *Service, objects Objects) (map[string]any, error) { + if len(m.Pages) == 0 { + return nil, fmt.Errorf("document %s has no page", id) + } + pages := make([]PDFPage, len(m.Pages)) + for i, p := range m.Pages { + jpg, err := objects.Get(ctx, p.Key) + if err != nil { + return nil, fmt.Errorf("document %s: page %d is not kept: %w", id, i+1, err) + } + pages[i] = PDFPage{JPEG: jpg, DPI: p.DPI} + } + first := m.Pages[0] + c, err := captureOf(pages, Cropped{Width: first.Width, Height: first.Height, Cropped: first.Cropped}, first.DPI, w.Now()) + if err != nil { + return nil, err + } + set, err := w.Settings() + if err != nil { + return nil, err + } + answer, err := w.keepOne(ctx, objects, set, c) + if err != nil { + return nil, err + } + m.Finished = answer["key"].(string) + if err := saveManifest(ctx, objects, id, m); err != nil { + return nil, fmt.Errorf("document %s is kept as %s, but could not be marked closed: %w", id, m.Finished, err) + } + // The pages are not needed any more: delete them. A bucket that refuses leaves them, and the answer says so. + for _, p := range m.Pages { + if err := objects.Delete(ctx, p.Key); err != nil { + answer["pending-kept"] = fmt.Sprintf("the pages of document %s could not be deleted from the bucket and stay under pending/%s/: %v", id, id, err) + break + } + } + return answer, nil +} + +// openDocuments lists the documents of several pages that are open in the bucket, with the time of each one's last +// page, so that the tool can arm their expiry again after it was restarted. +func (w *deviceWorker) openDocuments(ctx context.Context) (map[string]any, error) { + objects, err := w.Objects() + if err != nil { + return nil, err + } + keys, err := objects.List(ctx, "pending/") + if err != nil { + return nil, err + } + open := []map[string]any{} + for _, k := range keys { + rest, ok := strings.CutPrefix(k, "pending/") + id, name, _ := strings.Cut(rest, "/") + if !ok || name != "manifest.json" || !uploadPattern.MatchString(id) { + continue + } + m, err := loadManifest(ctx, objects, id) + if err != nil { + continue // closed, or unreadable: not open + } + at := m.Updated + if at.IsZero() { + at = m.Started + } + open = append(open, map[string]any{"id": id, "pages": len(m.Pages), "updated": at.UTC().Format(time.RFC3339)}) + } + return map[string]any{"open": open}, nil +} diff --git a/modules/brother/cmd/brother/documents_test.go b/modules/brother/cmd/brother/documents_test.go index ceaff0ba..a876c540 100644 --- a/modules/brother/cmd/brother/documents_test.go +++ b/modules/brother/cmd/brother/documents_test.go @@ -474,8 +474,8 @@ func TestAScanLargerThanTheStatesValueLimitIsFiledWholeAndItsAnswerIsSmall(t *te if len(big) <= StateMaxValue { t.Fatal("the stand-in is not larger than the state's value limit") } - r.worker.Capture = func(ctx context.Context, req ScanRequest) (Capture, error) { - return Capture{PDF: big, SHA256: sha256Hex(big), Resolution: 300, At: fixedNow}, nil + r.worker.Capture = func(ctx context.Context, req ScanRequest) ([]Capture, error) { + return []Capture{{PDF: big, SHA256: sha256Hex(big), Resolution: 300, At: fixedNow}}, nil } body, err := r.scan(t) if err != nil { @@ -539,3 +539,91 @@ func TestOnlyATooLargeAnswerEndsTheAskATransientFailureIsRetried(t *testing.T) { t.Errorf("a too large answer: action %v, want term", act) } } + +// home makes a home folder for the tool account and returns it. +func home(t *testing.T) string { + t.Helper() + h := t.TempDir() + t.Setenv("HOME", h) + return h +} + +// 1551: printer.print with a path reads the file, uploads it in parts and prints it in one call; a document of +// several parts arrives whole, and the answer has the job's id and state. +func TestPrinterPrintTakesAPathAndUploadsAndPrintsInOneCall(t *testing.T) { + fake := newIPPFake(t, 31, 5) + r := newRig(t, `{"printer":"192.0.2.10"}`, "pw") + doc := pdfOf(2*PartSize + 1234) + file := filepath.Join(home(t), "letter.bin") // the name says nothing: the bytes are read + if err := os.WriteFile(file, doc, 0o600); err != nil { + t.Fatal(err) + } + got, err := r.call(t, "printer.print", map[string]any{"path": file, "copies": float64(2)}) + if err != nil { + t.Fatal(err) + } + m := got.(map[string]any) + if m["job-id"] != float64(31) || m["job-state"] != "processing" || m["format"] != "application/pdf" || m["bytes"] != len(doc) { + t.Errorf("answer = %v", m) + } + attrs, sent := fake.request(t) + if !bytes.Equal(sent, doc) || attrs.Attributes["copies"][0].Integer != 2 { + t.Errorf("sent %d bytes (want %d), attributes %v", len(sent), len(doc), attrs.Attributes) + } + ops := r.bus.ops() + if len(ops) != 4 || ops[0] != OpUpload || ops[2] != OpUpload || ops[3] != OpPrint { + t.Errorf("asks = %v, want three upload parts and a print", ops) + } +} + +// 1551: only a regular file under the tool account's home folder is read, after symbolic links are resolved; every +// other path is refused in one fixed way that says nothing of whether it exists or how large it is; a file in the +// home folder that is no PDF or JPEG by its bytes is refused by name of the type. +func TestPrinterPrintReadsOnlyAFileUnderTheHomeFolder(t *testing.T) { + fake := newIPPFake(t, 1, 5) + r := newRig(t, `{"printer":"192.0.2.10"}`, "pw") + h := home(t) + outside := t.TempDir() + secret := filepath.Join(outside, "secret.pdf") + os.WriteFile(secret, pdfOf(3000), 0o600) + link := filepath.Join(h, "link.pdf") + os.Symlink(secret, link) + dirLink := filepath.Join(h, "dirlink") + os.Symlink(outside, dirLink) + os.Mkdir(filepath.Join(h, "folder"), 0o700) + for name, path := range map[string]string{ + "outside home": secret, + "link out of home": link, + "folder link out": filepath.Join(dirLink, "secret.pdf"), + "dot-dot out of home": filepath.Join(h, "..", filepath.Base(outside), "secret.pdf"), + "missing in home": filepath.Join(h, "none.pdf"), + "missing outside home": filepath.Join(outside, "none.pdf"), + "folder": filepath.Join(h, "folder"), + "the home folder itself": h, + "relative": "secret.pdf", + "proc": "/proc/self/environ", + } { + _, err := r.call(t, "printer.print", map[string]any{"path": path}) + if err == nil || err.Error() != PathRefused { + t.Errorf("%s: error = %v, want exactly %q", name, err, PathRefused) + } + } + text := filepath.Join(h, "fake.pdf") + os.WriteFile(text, []byte("just text, named .pdf"), 0o600) + if _, err := r.call(t, "printer.print", map[string]any{"path": text}); err == nil || !strings.Contains(err.Error(), "neither a PDF nor a JPEG") { + t.Errorf("text in home: error = %v", err) + } + if _, err := r.call(t, "printer.print", map[string]any{"path": text, "object": sha256Hex([]byte("x")) + ".pdf"}); err == nil { + t.Error("path with object accepted") + } + if r.bus.asks() != 0 || fake.body != nil { + t.Errorf("%d asks queued, printer sent %d bytes", r.bus.asks(), len(fake.body)) + } + // A symbolic link inside the home folder to a file inside it is fine. + inside := filepath.Join(h, "real.pdf") + os.WriteFile(inside, pdfOf(3000), 0o600) + os.Symlink(inside, filepath.Join(h, "ok.pdf")) + if _, err := r.call(t, "printer.print", map[string]any{"path": filepath.Join(h, "ok.pdf")}); err != nil { + t.Errorf("a link within home: %v", err) + } +} diff --git a/modules/brother/cmd/brother/escl.go b/modules/brother/cmd/brother/escl.go index a16f1e31..4ae2f7a1 100644 --- a/modules/brother/cmd/brother/escl.go +++ b/modules/brother/cmd/brother/escl.go @@ -128,8 +128,21 @@ type ScanOptions struct { ColorMode string Feeder bool Duplex bool + // Bound is how long a feeder job may have run when it has this many pages and is asked for the next; nil is + // FeederBound. The job is deleted and an error given when it has run longer. + Bound func(pages int) time.Duration } +// FeederBound is how long a feeder job may run when it has this many pages (issue 1539): 20 seconds for each, +// at least 3 minutes and at most 30. The scanner gives a duplex sheet about every 3.4 seconds, so this is generous +// and only ends a job that is stuck. +func FeederBound(pages int) time.Duration { + return min(max(time.Duration(pages)*20*time.Second, 3*time.Minute), FeederMaxBound) +} + +// FeederMaxBound is the longest a feeder job may run. +const FeederMaxBound = 30 * time.Minute + // A4 in the 300ths of an inch the region is asked in; a region is cut to what the input can scan. const a4Width, a4Height = 2480, 3508 @@ -202,7 +215,7 @@ func (c *Client) ScanPages(ctx context.Context, o ScanOptions) ([][]byte, error) } doc := job + "/NextDocument" if o.Feeder { - pages, err := c.feederPages(ctx, doc) + pages, err := c.feederPages(ctx, doc, o.Bound) if err != nil { c.deleteJob(job) return nil, err @@ -226,9 +239,18 @@ func (c *Client) ScanPages(ctx context.Context, o ScanOptions) ([][]byte, error) // feederPages asks NextDocument until the scanner answers 404, which ends the job. The first page missing means no // paper was in the feeder. -func (c *Client) feederPages(ctx context.Context, doc string) ([][]byte, error) { +func (c *Client) feederPages(ctx context.Context, doc string, bound func(int) time.Duration) ([][]byte, error) { + if bound == nil { + bound = FeederBound + } + start := time.Now() var pages [][]byte for len(pages) <= maxFeederPages { + if len(pages) > 0 { + if run, limit := time.Since(start), bound(len(pages)); run > limit { + return nil, fmt.Errorf("the feeder scan has run %s and has %d pages, past its bound of %s: it is stopped", run.Round(time.Second), len(pages), limit) + } + } got, err := c.get(ctx, doc) if err != nil { return nil, fmt.Errorf("the scanned page %d cannot be fetched: %w", len(pages)+1, err) diff --git a/modules/brother/cmd/brother/feeder_test.go b/modules/brother/cmd/brother/feeder_test.go index 66dd63f8..8d4d43ec 100644 --- a/modules/brother/cmd/brother/feeder_test.go +++ b/modules/brother/cmd/brother/feeder_test.go @@ -3,9 +3,11 @@ package main import ( "bytes" "context" + "errors" "os" "strings" "testing" + "time" ) // 1452: the Brother ignores the whole body when it holds a line break before the closing root tag, and runs its @@ -163,3 +165,143 @@ func TestDuplexOnTheGlassIsRefusedAndNothingIsAsked(t *testing.T) { t.Errorf("%d asks queued, %d jobs posted", r.bus.asks(), f.posts) } } + +// pdfCapture is a stand-in capture of a small PDF, ready to be released. +func standInCapture(release <-chan struct{}, fail error) func(ctx context.Context, req ScanRequest) ([]Capture, error) { + return func(ctx context.Context, req ScanRequest) ([]Capture, error) { + <-release + if fail != nil { + return nil, fail + } + pdf := pdfOf(5000) + return []Capture{{PDF: pdf, SHA256: sha256Hex(pdf), Resolution: 200, Pages: 3, At: fixedNow}}, nil + } +} + +// waitFor polls until the condition holds. +func waitFor(t *testing.T, what string, cond func() bool) { + t.Helper() + for i := 0; i < 400; i++ { + if cond() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", what) +} + +// 1552 and 1539: a feeder scan not done within the quick wait is answered as started with its id; the scan goes on +// and is filed although the caller has gone; the id reads the outcome, running and then done with the path. +func TestALongFeederScanAnswersStartedAndIsFiledWithoutTheCaller(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, folder := scanRig(t, f, Settings{}) + old := quickWait + quickWait = 50 * time.Millisecond + t.Cleanup(func() { quickWait = old }) + release := make(chan struct{}) + r.worker.Capture = standInCapture(release, nil) + + got, err := r.call(t, "scanner.scan", scanArgs("feeder", false)) + if err != nil { + t.Fatal(err) + } + start := got.(map[string]any) + id, _ := start["id"].(string) + if start["started"] != true || id == "" || start["path"] != nil { + t.Fatalf("answer = %v, want started with an id", start) + } + running, err := r.call(t, "scanner.scan", map[string]any{"id": id}) + if err != nil || running.(map[string]any)["state"] != "running" { + t.Fatalf("while it runs: %v, %v", running, err) + } + close(release) + // Nobody asks again: the scan is filed all the same. + waitFor(t, "the scan to be filed", func() bool { e, _ := os.ReadDir(folder); return len(e) == 1 }) + // The file is renamed into place before the job is marked done: read the outcome until it is. + var d map[string]any + waitFor(t, "the outcome to be done", func() bool { + got, err := r.call(t, "scanner.scan", map[string]any{"id": id}) + if err != nil { + t.Fatal(err) + } + d = got.(map[string]any) + return d["state"] != "running" + }) + if d["state"] != "done" || d["pages"] != float64(3) || d["path"] == nil { + t.Fatalf("outcome = %v, want done with pages and the path", d) + } + if _, err := os.Stat(d["path"].(string)); err != nil { + t.Error(err) + } +} + +// 1552: a scan that fails after the caller was told it started is read as failed, with why; an unknown id is an +// error that says so. +func TestAFailedLongScanIsReadAsFailedWithWhy(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, _ := scanRig(t, f, Settings{}) + old := quickWait + quickWait = 50 * time.Millisecond + t.Cleanup(func() { quickWait = old }) + release := make(chan struct{}) + r.worker.Capture = standInCapture(release, errors.New("the scanner dropped the job: 410 Gone")) + start, err := r.call(t, "scanner.scan", scanArgs("feeder", true)) + if err != nil { + t.Fatal(err) + } + id := start.(map[string]any)["id"].(string) + close(release) + var out map[string]any + waitFor(t, "the failure", func() bool { + got, _ := r.call(t, "scanner.scan", map[string]any{"id": id}) + out = got.(map[string]any) + return out["state"] == "failed" + }) + if !strings.Contains(out["why"].(string), "410 Gone") { + t.Errorf("why = %v", out["why"]) + } + if _, err := r.call(t, "scanner.scan", map[string]any{"id": "nosuchid"}); err == nil || !strings.Contains(err.Error(), "nosuchid") { + t.Errorf("an unknown id: %v", err) + } +} + +// 1539: the feeder job's bound scales with its pages, from 3 to 30 minutes. +func TestTheFeederBoundScalesWithThePages(t *testing.T) { + for pages, want := range map[int]time.Duration{1: 3 * time.Minute, 9: 3 * time.Minute, 30: 10 * time.Minute, 100: 30 * time.Minute, 500: 30 * time.Minute} { + if got := FeederBound(pages); got != want { + t.Errorf("FeederBound(%d) = %s, want %s", pages, got, want) + } + } + f := newFakeScanner(t, fixture(t, "scan.jpg")) + f.sheets = 3 + _, err := NewClient(f.base()).ScanPages(context.Background(), ScanOptions{Resolution: 200, ColorMode: "color", Feeder: true, + Bound: func(int) time.Duration { return 0 }}) + if err == nil || !strings.Contains(err.Error(), "bound") { + t.Fatalf("error = %v, want the bound", err) + } + if f.open || f.deletes != 1 { + t.Errorf("job open %v, %d DELETEs, want the job deleted", f.open, f.deletes) + } +} + +// 1539: a scan too large for the tool to fetch is not kept nor announced: it is refused with its size. +func TestAScanTooLargeToFetchIsNeitherKeptNorAnnounced(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, folder := scanRig(t, f, Settings{}) + release := make(chan struct{}) + close(release) + huge := make([]byte, MostAnswerParts*AnswerPartSize+1) + r.worker.Capture = func(ctx context.Context, req ScanRequest) ([]Capture, error) { + return []Capture{{PDF: huge, SHA256: "x", Resolution: 200, Pages: 99, At: fixedNow}}, nil + } + _, err := r.call(t, "scanner.scan", scanArgs("feeder", false)) + if err == nil || !strings.Contains(err.Error(), "not kept") { + t.Fatalf("error = %v", err) + } + if len(*r.said) != 0 { + t.Errorf("%d events said for a scan not kept", len(*r.said)) + } + if e, _ := os.ReadDir(folder); len(e) != 0 { + t.Errorf("filed: %v", e) + } +} diff --git a/modules/brother/cmd/brother/file.go b/modules/brother/cmd/brother/file.go index 44de023a..3e68324d 100644 --- a/modules/brother/cmd/brother/file.go +++ b/modules/brother/cmd/brother/file.go @@ -1,6 +1,7 @@ package main import ( + "errors" "fmt" "os" "path/filepath" @@ -122,3 +123,41 @@ func List(folder string, limit int) ([]Entry, error) { } return out, nil } + +// PathRefused is the one answer to a path that may not be printed from: it says nothing of whether the path exists, +// what it is or how large (issue 1551). +const PathRefused = "path refused: only a file under the home folder" + +// ReadDocument is the file at path, read whole: a regular file under the tool account's home folder, of at most +// most bytes. The path is resolved through its symbolic links first and the real path must still lie under the +// home folder (which is itself resolved); anything else, a missing path included, is PathRefused and no more. +func ReadDocument(path string, most int64) ([]byte, error) { + home, err := os.UserHomeDir() + if err != nil || home == "" || !filepath.IsAbs(path) { + return nil, errors.New(PathRefused) + } + realHome, err := filepath.EvalSymlinks(home) + if err != nil { + return nil, errors.New(PathRefused) + } + real, err := filepath.EvalSymlinks(path) + if err != nil { + return nil, errors.New(PathRefused) + } + rel, err := filepath.Rel(realHome, real) + if err != nil || rel == "." || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { + return nil, errors.New(PathRefused) + } + info, err := os.Stat(real) + if err != nil || !info.Mode().IsRegular() { + return nil, errors.New(PathRefused) + } + if info.Size() > most { + return nil, fmt.Errorf("file %s is %d bytes, over the %d the module takes", path, info.Size(), most) + } + doc, err := os.ReadFile(real) + if err != nil { + return nil, fmt.Errorf("file %s cannot be read: %w", path, err) + } + return doc, nil +} diff --git a/modules/brother/cmd/brother/glass_test.go b/modules/brother/cmd/brother/glass_test.go new file mode 100644 index 00000000..ef89f713 --- /dev/null +++ b/modules/brother/cmd/brother/glass_test.go @@ -0,0 +1,322 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "image" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// withDensity puts a JFIF header of dpi into a JPEG that has none (Go's encoder writes none). +func withDensity(jpg []byte, dpi int) []byte { + app0 := []byte{0xFF, 0xE0, 0, 16, 'J', 'F', 'I', 'F', 0, 1, 1, 1, byte(dpi >> 8), byte(dpi), byte(dpi >> 8), byte(dpi), 0, 0} + return append(append(append([]byte{}, jpg[:2]...), app0...), jpg[2:]...) +} + +// 1528: three receipts lying apart on the glass are three PDFs, each its own object, own scan.added and own file +// in the scan folder, one page each; the answer lists every document. +func TestSeveralDocumentsOnTheGlassAreFiledEachOnItsOwn(t *testing.T) { + scan := withDensity(onWhite(t, 11, image.Rect(100, 100, 300, 300), image.Rect(450, 100, 650, 400), image.Rect(100, 600, 400, 900)), 100) + f := newFakeScanner(t, scan) + r, folder := scanRig(t, f, Settings{Crop: true, Resolution: 100}) + got, err := r.call(t, "scanner.scan", map[string]any{"apart": true}) + if err != nil { + t.Fatal(err) + } + body := got.(map[string]any) + docs, _ := body["documents"].([]any) + if len(docs) != 3 || body["count"] != 3 { + t.Fatalf("answer = %v, want three documents", body) + } + seen := map[string]bool{} + for _, d := range docs { + m := d.(map[string]any) + path, _ := m["path"].(string) + if _, err := os.Stat(path); err != nil || m["pages"] != float64(1) || m["cropped"] != true || seen[path] { + t.Errorf("document %v: %v", m, err) + } + seen[path] = true + } + if e, _ := os.ReadDir(folder); len(e) != 3 { + t.Errorf("%d files in the scan folder, want 3", len(e)) + } + out := *r.said + if len(out) != 3 { + t.Fatalf("%d events, want one each", len(out)) + } + keys := map[any]bool{} + for _, ev := range out { + keys[ev.body["object"]] = true + if !ev.seen { + t.Error("an event was said before its PDF was kept") + } + } + if len(keys) != 3 { + t.Errorf("the events name %d different objects, want 3", len(keys)) + } +} + +// 1528: one document on the glass is answered as today: a flat body with its path. +func TestOneDocumentOnTheGlassIsAnsweredAsBefore(t *testing.T) { + scan := withDensity(onWhite(t, 12, image.Rect(100, 100, 300, 300)), 100) + f := newFakeScanner(t, scan) + r, _ := scanRig(t, f, Settings{Crop: true, Resolution: 100}) + body, err := r.scan(t) + if err != nil { + t.Fatal(err) + } + if body["documents"] != nil || body["path"] == nil || len(*r.said) != 1 { + t.Errorf("answer = %v, %d events", body, len(*r.said)) + } +} + +func docArgs(document string, finish bool) map[string]any { + a := map[string]any{"document": document} + if finish { + a["finish"] = true + } + return a +} + +// 1538: start scans page 1 into a new document and files nothing; the id adds the next page; finish closes the +// pages into one PDF, one object, one scan.added, one file, and scans nothing. +func TestAGlassDocumentOfTwoPagesIsOnePDFFiledOnce(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, folder := scanRig(t, f, Settings{Crop: false, Resolution: 200}) + got, err := r.call(t, "scanner.scan", docArgs("start", false)) + if err != nil { + t.Fatal(err) + } + id, _ := got.(map[string]any)["document"].(string) + if id == "" || got.(map[string]any)["pages"] != float64(1) { + t.Fatalf("start answered %v", got) + } + if got, err = r.call(t, "scanner.scan", docArgs(id, false)); err != nil || got.(map[string]any)["pages"] != float64(2) { + t.Fatalf("second page: %v, %v", got, err) + } + if e, _ := os.ReadDir(folder); len(e) != 0 || len(*r.said) != 0 { + t.Fatalf("an open document filed %d files and said %d events", len(e), len(*r.said)) + } + posts := f.posts + got, err = r.call(t, "scanner.scan", docArgs(id, true)) + if err != nil { + t.Fatal(err) + } + body := got.(map[string]any) + if body["pages"] != float64(2) || body["path"] == nil || f.posts != posts { + t.Fatalf("finish answered %v (%d scans after, %d before)", body, f.posts, posts) + } + pdf, _ := os.ReadFile(body["path"].(string)) + if !bytes.Contains(pdf, []byte("/Count 2")) || bytes.Count(pdf, []byte("/Type /Page /Parent")) != 2 { + t.Error("the filed PDF does not have two pages") + } + if e, _ := os.ReadDir(folder); len(e) != 1 || len(*r.said) != 1 || (*r.said)[0].body["pages"] != 2 { + t.Errorf("%d files, %d events", len(e), len(*r.said)) + } + // A closed document takes no more. + if _, err := r.call(t, "scanner.scan", docArgs(id, false)); err == nil || !strings.Contains(err.Error(), "closed already") { + t.Errorf("a page into a closed document: %v", err) + } +} + +// 1538: what is refused: finish with start, an unknown id, a feeder document, an id that is no id. +func TestGlassDocumentsRefuseWhatIsNotWellFormed(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, _ := scanRig(t, f, Settings{}) + for name, args := range map[string]map[string]any{ + "finish with start": docArgs("start", true), + "finish alone": {"finish": true}, + "unknown id": docArgs("nosuchdocument", false), + "unknown finish": docArgs("nosuchdocument", true), + "bad id": docArgs("../../etc", false), + "on the feeder": {"document": "start", "source": "feeder"}, + } { + if _, err := r.call(t, "scanner.scan", args); err == nil { + t.Errorf("%s: accepted", name) + } + } + if f.posts != 0 { + t.Errorf("%d scans were put for refused asks", f.posts) + } +} + +// 1538: an open document that gets no page is closed and filed as one PDF by the tool: its pages are not lost. +func TestAnUnfinishedGlassDocumentIsFiledAfterItsExpiry(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, folder := scanRig(t, f, Settings{Crop: false, Resolution: 200}) + old := docExpiry + docExpiry = 150 * time.Millisecond + t.Cleanup(func() { docExpiry = old }) + got, err := r.call(t, "scanner.scan", docArgs("start", false)) + if err != nil { + t.Fatal(err) + } + id := got.(map[string]any)["document"].(string) + if _, err := r.call(t, "scanner.scan", docArgs(id, false)); err != nil { + t.Fatal(err) + } + waitFor(t, "the expired document to be filed", func() bool { e, _ := os.ReadDir(folder); return len(e) == 1 }) + e, _ := os.ReadDir(folder) + pdf, _ := os.ReadFile(filepath.Join(folder, e[0].Name())) + if !bytes.Contains(pdf, []byte("/Count 2")) || len(*r.said) != 1 { + t.Errorf("the expired document: %d events, two pages %v", len(*r.said), bytes.Contains(pdf, []byte("/Count 2"))) + } +} + +// 1538: finishing a document does not wait for the expiry: its timer is gone, so nothing is filed twice. +func TestAFinishedGlassDocumentIsNotFiledAgainByItsExpiry(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, folder := scanRig(t, f, Settings{Crop: false, Resolution: 200}) + old := docExpiry + docExpiry = 150 * time.Millisecond + t.Cleanup(func() { docExpiry = old }) + got, _ := r.call(t, "scanner.scan", docArgs("start", false)) + id := got.(map[string]any)["document"].(string) + if _, err := r.call(t, "scanner.scan", docArgs(id, true)); err != nil { + t.Fatal(err) + } + time.Sleep(400 * time.Millisecond) + if e, _ := os.ReadDir(folder); len(e) != 1 { + t.Errorf("%d files, want 1", len(e)) + } +} + +// 1539: an ask carries its deadline, and a device process that takes it up later does nothing for it: nothing is +// scanned, the answer says why, and the message ends. +func TestAnAskTakenUpAfterItsDeadlineIsDroppedBeforeAnyWork(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, folder := scanRig(t, f, Settings{}) + scans := 0 + r.worker.Capture = func(ctx context.Context, req ScanRequest) ([]Capture, error) { + scans++ + return nil, errors.New("scanned") + } + late, _ := json.Marshal(Ask{ID: "late", Op: OpScan, At: fixedNow.Add(-4 * time.Minute), Deadline: fixedNow.Add(-time.Second)}) + if act, _ := r.worker.Handle(late, func() { t.Error("a late ask was acknowledged early as work to be done") }); act != actionTerm { + t.Errorf("action = %v, want the message ended", act) + } + if scans != 0 || len(*r.said) != 0 { + t.Errorf("%d scans, %d events for a late ask", scans, len(*r.said)) + } + if raw := r.answers.get("late"); !strings.Contains(string(raw), "deadline") { + t.Errorf("answer = %s", raw) + } + onTime, _ := json.Marshal(Ask{ID: "ontime", Op: OpScan, At: fixedNow, Deadline: fixedNow.Add(time.Minute)}) + r.worker.Handle(onTime, nil) + if scans != 1 { + t.Errorf("an ask in time was not scanned") + } + _ = folder +} + +// 1539: a scan the device process took up in time but ran past the tool's start-by wait is still filed: the tool +// follows it for the run's own bound. +func TestAScanThatRunsPastTheStartByWaitIsStillFiled(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, folder := scanRig(t, f, Settings{}) + old := scanWait + scanWait = 100 * time.Millisecond + t.Cleanup(func() { scanWait = old }) + release := make(chan struct{}) + r.worker.Capture = standInCapture(release, nil) + got, err := r.call(t, "scanner.scan", map[string]any{}) + if err != nil || got.(map[string]any)["started"] != true { + t.Fatalf("answer = %v, %v, want started", got, err) + } + time.Sleep(300 * time.Millisecond) // past the start-by wait; the answer comes only now + close(release) + waitFor(t, "the late scan to be filed", func() bool { e, _ := os.ReadDir(folder); return len(e) == 1 }) +} + +// 1538: an expired document whose closing fails is not forgotten: it is tried again. +func TestAnExpiredDocumentWhoseClosingFailsIsClosedOnTheNextTry(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, folder := scanRig(t, f, Settings{Crop: false, Resolution: 200}) + oldE, oldR := docExpiry, docRetry + docExpiry, docRetry = 100*time.Millisecond, 100*time.Millisecond + t.Cleanup(func() { docExpiry, docRetry = oldE, oldR }) + askBus = &flakyBus{fakeAskBus: &fakeAskBus{worker: r.worker, answers: r.answers}, fails: 2, + match: func(a Ask) bool { return a.Op == OpScan && a.Finish }} + if _, err := r.call(t, "scanner.scan", docArgs("start", false)); err != nil { + t.Fatal(err) + } + waitFor(t, "the document to be filed on the third try", func() bool { e, _ := os.ReadDir(folder); return len(e) == 1 }) +} + +// 1538: after a restart of the tool process the open documents are found again in the bucket and their expiry +// armed from the time of their last page; closing a document deletes its pages. +func TestAnOpenDocumentIsFoundAgainAfterARestartAndItsPagesAreDeletedWhenClosed(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, folder := scanRig(t, f, Settings{Crop: false, Resolution: 200}) + got, err := r.call(t, "scanner.scan", docArgs("start", false)) + if err != nil { + t.Fatal(err) + } + id := got.(map[string]any)["document"].(string) + // A new tool process: no timers. Its Recover finds the document and arms it. + old := docExpiry + docExpiry = 200 * time.Millisecond + t.Cleanup(func() { docExpiry = old }) + s := &Service{Settings: Settings{Scanner: f.base(), Folder: folder, Resolution: 200, ColorMode: "color"}, Now: func() time.Time { return fixedNow }, + Asks: askBus, AnswerWait: 5 * time.Second, ScanWait: 5 * time.Second, AnswerPoll: 5 * time.Millisecond, Jobs: NewJobs()} + if err := s.Recover(context.Background()); err != nil { + t.Fatal(err) + } + waitFor(t, "the recovered document to be closed", func() bool { e, _ := os.ReadDir(folder); return len(e) == 1 }) + waitFor(t, "its pages to be deleted", func() bool { + for _, k := range r.objects.keys() { + if strings.HasPrefix(k, "pending/"+id+"/") && strings.HasSuffix(k, ".jpg") { + return false + } + } + return true + }) + var hasManifest bool + for _, k := range r.objects.keys() { + hasManifest = hasManifest || k == "pending/"+id+"/manifest.json" + } + if !hasManifest { + t.Error("the manifest that says the document is closed was deleted too") + } +} + +// 1552: a finished job's outcome is forgotten after jobKeep. +func TestAFinishedJobIsForgottenAfterAnHour(t *testing.T) { + old := jobKeep + jobKeep = 30 * time.Millisecond + t.Cleanup(func() { jobKeep = old }) + j := NewJobs() + first := j.put("a") + close(first.done) + time.Sleep(60 * time.Millisecond) + j.put("b") + if j.get("a") != nil || j.get("b") == nil { + t.Errorf("a = %v, b = %v", j.get("a"), j.get("b")) + } +} + +// 1528: without apart three documents on the glass are one, cut around all of them, as before. +func TestWithoutApartSeveralDocumentsOnTheGlassAreOneScan(t *testing.T) { + scan := withDensity(onWhite(t, 11, image.Rect(100, 100, 300, 300), image.Rect(450, 100, 650, 400), image.Rect(100, 600, 400, 900)), 100) + f := newFakeScanner(t, scan) + r, folder := scanRig(t, f, Settings{Crop: true, Resolution: 100}) + body, err := r.scan(t) + if err != nil { + t.Fatal(err) + } + if body["documents"] != nil || body["path"] == nil || len(*r.said) != 1 { + t.Errorf("answer = %v, %d events", body, len(*r.said)) + } + if e, _ := os.ReadDir(folder); len(e) != 1 { + t.Errorf("%d files, want 1", len(e)) + } + if _, err := r.call(t, "scanner.scan", map[string]any{"apart": true, "source": "feeder"}); err == nil || !strings.Contains(err.Error(), "apart is for the glass") { + t.Errorf("apart on the feeder: %v", err) + } +} diff --git a/modules/brother/cmd/brother/main.go b/modules/brother/cmd/brother/main.go index 99ec4b1f..d33a82ad 100644 --- a/modules/brother/cmd/brother/main.go +++ b/modules/brother/cmd/brother/main.go @@ -4,6 +4,8 @@ import ( "context" "fmt" "os" + "os/signal" + "syscall" "time" stdio "git.novox.be/novox/mesh-sdk/go" @@ -13,12 +15,26 @@ func main() { if len(os.Args) > 1 && os.Args[1] == "device" { os.Exit(device()) } + if len(os.Args) > 1 && os.Args[1] == "watch" { + os.Exit(watch()) + } if err := stdio.Serve("", tools(os.Getenv("BROTHER_SETTINGS"), time.Now)); err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } } +// watch is `brother watch`: the process that carries the tool check brother_health beside the device process's unit +// check, as fail2ban's watch does (ADR 0240: a tool check is only beside a check the module does not run itself). +// It does nothing itself; the node-engine asks the tool, and this stays up until it is stopped. +func watch() int { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + fmt.Fprintln(os.Stderr, "[brother] watch is up; the node-engine asks brother_health") + <-ctx.Done() + return 0 +} + // intArg reads a whole number out of a tool's arguments: a JSON number, which arrives as a float64. func intArg(args map[string]any, name string) (int, error) { v, ok := args[name] @@ -85,13 +101,15 @@ func askFrom(args map[string]any) (Ask, error) { // The settings are read at each call, so a changed config value is the next call's. None opens a credential: what // needs one is asked of the device process. func tools(settingsFile string, now func() time.Time) []stdio.Tool { + jobs := NewJobs() service := func() (*Service, error) { s, err := LoadSettings(settingsFile) if err != nil { return nil, err } return &Service{Settings: s, Node: os.Getenv("MESH_NODE"), Now: now, Asks: askBus, - AnswerWait: answerWait, ScanWait: scanWait, AnswerPoll: answerPoll}, nil + AnswerWait: answerWait, ScanWait: scanWait, AnswerPoll: answerPoll, + QuickWait: quickWait, FeederWait: feederWait, Jobs: jobs}, nil } // ask is a tool that reads its arguments, and answers what the service gives. ask := func(run func(s *Service, ctx context.Context, a Ask) (map[string]any, error), bound time.Duration) func(map[string]any) (any, error) { @@ -129,16 +147,40 @@ func tools(settingsFile string, now func() time.Time) []stdio.Tool { }, { Name: "printer.print", - Description: "Print a document on the printer: a scan kept in the bucket (object) or a completed upload (upload). The device " + - "process sends it as an IPP Print-Job in its own format (PDF or JPEG) and answers the job's id and state.", + Description: "Print a document on the printer: a file under the home folder of the account the tool runs as, on the machine it runs on (path: reads it, uploads it in parts " + + "and prints it, in one call), a scan kept in the bucket (object), or a completed upload (upload; for a module that sends " + + "bytes over the bus). Only PDF and JPEG, told by the file's bytes and not its name. The device " + + "process sends it as an IPP Print-Job in its own format and answers the job's id and state (with path, the upload id and the size).", Input: map[string]any{"type": "object", "properties": map[string]any{ + "path": str("an absolute path of a PDF or JPEG file under the tool account's home folder (symbolic links are resolved and the real path must lie there too; any other path is refused in one fixed way)"), "object": str("a scan's key, as scan.added names it"), "upload": str("a completed upload's id"), "copies": num("1 to 99 (default 1)"), "sides": str("one-sided (default), two-sided-long-edge or two-sided-short-edge"), "media": str("the paper's size name (default " + DefaultMedia + ")"), }}, - Run: ask(func(s *Service, ctx context.Context, a Ask) (map[string]any, error) { return s.PrinterPrint(ctx, a) }, 2*time.Minute), + Run: func(args map[string]any) (any, error) { + path, err := strArg(args, "path") + if err != nil { + return nil, err + } + a, err := askFrom(args) + if err != nil { + return nil, err + } + s, err := service() + if err != nil { + return nil, err + } + if path == "" { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + return s.PrinterPrint(ctx, a) + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute) + defer cancel() + return s.PrintFile(ctx, path, a) + }, }, { Name: "scanner.document", @@ -216,16 +258,65 @@ func tools(settingsFile string, now func() time.Time) []stdio.Tool { Description: "Scan with the configured network scanner, from the glass (source glass, the default: one page, cut out of " + "the platen's white when crop is set) or from the document feeder (source feeder: every sheet in it, whole, up to 50; " + "duplex true scans both sides of each sheet). The device process keeps the result as one PDF in the module's bucket " + - "under its sha256 (one page for the glass, one page per side for the feeder) and says scan.added on the bus; this " + - "tool then fetches it and files it in the scan folder. Waits up to three minutes. Answers the event's body: " + + "under its sha256 (one page for the glass, one page per side for the feeder; with apart true and several documents lying apart on the glass, each is its own one-page PDF with its own object and scan.added) and says scan.added on the bus; this " + + "tool then fetches it and files it in the scan folder, in the background, so a scan is filed even when you stop waiting. " + + "A glass scan is waited for (up to three minutes). A feeder scan is waited for 25 seconds (the mesh console gives up on a " + + "call after 30): when it is not done by then the answer is {started: true, id, state: running}; the scan goes on (the " + + "device process lets a feeder run last 20 seconds a page, 3 to 30 minutes) and is filed when done, and " + + "scanner.scan {id} reads the outcome: state running, state done with the answer below, or state failed with why. " + + "The id is known to the tool process that started the scan; scan.added also announces it. " + + "Several pages of the glass in one PDF (a page turned over, a stapled document): document start scans page 1 and answers {document: , pages: 1}; " + + "document scans the next page; document with finish true closes it into one PDF, filed once, answered like a scan. " + + "An open document that gets no page for 30 minutes is closed and filed so by the tool, and its pages are kept meanwhile in the bucket. Answers the event's body: " + "the object's key, its size and sha256, the first page's real size in millimetres, the resolution the scanner really gave, " + - "whether it was cropped, the number of pages and the file's path. The image itself is never sent. " + + "whether it was cropped, the number of pages and the file's path; for apart true and several documents on the glass {documents: [that, for each], count} (a failure of one is in failed). The image itself is never sent. " + "Duplex on the glass is refused.", Input: map[string]any{"type": "object", "properties": map[string]any{ - "source": map[string]any{"type": "string", "enum": []string{"glass", "feeder"}, "description": "glass (default) or feeder"}, - "duplex": map[string]any{"type": "boolean", "description": "feeder only: scan both sides of each sheet (default false)"}, + "source": map[string]any{"type": "string", "enum": []string{"glass", "feeder"}, "description": "glass (default) or feeder"}, + "id": map[string]any{"type": "string", "description": "read the outcome of the scan with this id (from an answer that said started) instead of scanning"}, + "document": map[string]any{"type": "string", "description": "glass only, a document of several pages in one PDF: start opens one and scans its first page; the id of an open document scans the next page into it"}, + "finish": map[string]any{"type": "boolean", "description": "with document: the id of an open document: close it into one PDF, filed once (nothing is scanned)"}, + "duplex": map[string]any{"type": "boolean", "description": "feeder only: scan both sides of each sheet (default false)"}, }}, Run: func(args map[string]any) (any, error) { + s, err := service() + if err != nil { + return nil, err + } + apart := false + if v, ok := args["apart"]; ok { + if apart, ok = v.(bool); !ok { + return nil, fmt.Errorf("apart is true or false") + } + } + id, err := strArg(args, "id") + if err != nil { + return nil, err + } + if id != "" { + return s.ScanResult(id) + } + document, err := strArg(args, "document") + if err != nil { + return nil, err + } + finish := false + if v, ok := args["finish"]; ok { + if finish, ok = v.(bool); !ok { + return nil, fmt.Errorf("finish is true or false") + } + } + if document != "" || finish { + if document == "" { + return nil, fmt.Errorf("finish closes a document: name its id in document") + } + if v, _ := args["source"].(string); v == "feeder" { + return nil, fmt.Errorf("a document of several pages is scanned from the glass; the feeder takes the whole stack in one scan") + } + ctx, cancel := context.WithTimeout(context.Background(), scanWait+10*time.Minute) + defer cancel() + return s.ScanPage(ctx, document, finish) + } source, err := strArg(args, "source") if err != nil { return nil, err @@ -236,13 +327,40 @@ func tools(settingsFile string, now func() time.Time) []stdio.Tool { return nil, fmt.Errorf("duplex is true or false") } } + ctx, cancel := context.WithTimeout(context.Background(), feederWait+10*time.Minute) + defer cancel() + return s.ScanNow(ctx, source, duplex, apart) + }, + }, + { + Name: "brother_health", + Description: "The module's health, asked by the node-engine (novox/hq ADR 0240): the certificates the printer serves, one for each " + + "key type (ECDSA and RSA), read with no login and compared with the pinned printer-certificate list. Answers healthy, why, " + + "and what was served with each fingerprint and whether it is pinned. Not healthy when a certificate served is not pinned (the " + + "printer was reset or given new certificates): why names the new fingerprint, so the pin can be changed before the admin " + + "pages are needed. A printer that does not answer is healthy, with why saying nothing was compared. Read-only.", + Run: func(map[string]any) (any, error) { s, err := service() if err != nil { return nil, err } - ctx, cancel := context.WithTimeout(context.Background(), scanWait+10*time.Minute) + // The node-engine asks this every minute, so it is also what brings back, after a restart of this + // process, the expiry of the documents left open in the bucket (issue 1538). + if s.Settings.Missing() == nil && jobs.tryRecover() { + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + if err := s.Recover(ctx); err != nil { + jobs.recoverFailed() + } + }() + } + if w, waits := CertWaits(s.Settings); waits { + return w, nil + } + ctx, cancel := context.WithTimeout(context.Background(), certTotal+time.Second) defer cancel() - return s.ScanNow(ctx, source, duplex) + return CheckCertificates(ctx, s.Settings.Printer, s.Settings.PrinterCertificate), nil }, }, { diff --git a/modules/brother/cmd/brother/manifest_test.go b/modules/brother/cmd/brother/manifest_test.go index 2291186f..6782b773 100644 --- a/modules/brother/cmd/brother/manifest_test.go +++ b/modules/brother/cmd/brother/manifest_test.go @@ -395,3 +395,17 @@ func TestEachConfigPatternAdmitsARealValueWhole(t *testing.T) { } } } + +// 1510: the certificate check is a tool check on a process of its own beside the device process's unit check, as +// fail2ban's watch is (ADR 0227 rule 8, ADR 0240: a tool check only beside a check of another kind). +func TestTheCertificateCheckIsAToolCheckBesideTheUnitCheck(t *testing.T) { + m := readManifest(t) + w := resource(t, m, "watch") + h, _ := w["health"].(map[string]any) + if w["type"] != "process" || w["artifact"] != "tools" || h["kind"] != "tool" || h["tool"] != "brother_health" || !reflect.DeepEqual(w["run"], []any{"./brother", "watch"}) { + t.Errorf("watch = %v", w) + } + if !slices.Contains(m.Tools, "brother_health") { + t.Errorf("tools = %v", m.Tools) + } +} diff --git a/modules/brother/cmd/brother/objects.go b/modules/brother/cmd/brother/objects.go index 6370f055..ffad97dc 100644 --- a/modules/brother/cmd/brother/objects.go +++ b/modules/brother/cmd/brother/objects.go @@ -14,6 +14,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "encoding/xml" "errors" "fmt" "io" @@ -35,6 +36,10 @@ type Objects interface { Put(ctx context.Context, key, mediaType string, data []byte) error // Get is the bytes kept under the key. Get(ctx context.Context, key string) ([]byte, error) + // List is the keys kept that start with the prefix, in order (the first thousand). + List(ctx context.Context, prefix string) ([]string, error) + // Delete removes the key; a key not kept is not an error. + Delete(ctx context.Context, key string) error } // Store keeps the bytes under the key once: a key already kept is left as it is, since a key that names the @@ -166,6 +171,37 @@ func (s *S3) Get(ctx context.Context, key string) ([]byte, error) { return io.ReadAll(resp.Body) } +func (s *S3) List(ctx context.Context, prefix string) ([]string, error) { + resp, err := s.do(ctx, http.MethodGet, "?list-type=2&prefix="+url.QueryEscape(prefix), "", nil) + if err != nil { + return nil, fmt.Errorf("the module's bucket cannot be reached: %w", err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("the module's bucket answered %s to listing %s", resp.Status, prefix) + } + var out struct { + Keys []string `xml:"Contents>Key"` + } + if err := xml.Unmarshal(body, &out); err != nil { + return nil, fmt.Errorf("the module's bucket listed %s in a form that cannot be read: %w", prefix, err) + } + return out.Keys, nil +} + +func (s *S3) Delete(ctx context.Context, key string) error { + resp, err := s.do(ctx, http.MethodDelete, key, "", nil) + if err != nil { + return fmt.Errorf("the module's bucket cannot be reached: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 && resp.StatusCode != http.StatusNotFound { + return fmt.Errorf("the module's bucket refused to delete %s: %s", key, resp.Status) + } + return nil +} + // Sign signs a request with AWS Signature Version 4 for S3: every header it carries but the agent and the length is // signed, with the payload's hash. func Sign(req *http.Request, body []byte, access, secret, region string, now time.Time) { diff --git a/modules/brother/cmd/brother/objects_test.go b/modules/brother/cmd/brother/objects_test.go index 2ba39fbc..1bcfbdc1 100644 --- a/modules/brother/cmd/brother/objects_test.go +++ b/modules/brother/cmd/brother/objects_test.go @@ -8,6 +8,7 @@ import ( "net/http/httptest" "os" "path/filepath" + "sort" "strings" "sync" "testing" @@ -49,6 +50,24 @@ func (m *memObjects) Get(_ context.Context, key string) ([]byte, error) { return b, nil } +func (m *memObjects) List(_ context.Context, prefix string) ([]string, error) { + var out []string + for _, k := range m.keys() { + if strings.HasPrefix(k, prefix) { + out = append(out, k) + } + } + sort.Strings(out) + return out, nil +} + +func (m *memObjects) Delete(_ context.Context, key string) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.kept, key) + return nil +} + func (m *memObjects) keys() []string { m.mu.Lock() defer m.mu.Unlock() @@ -132,3 +151,28 @@ func TestTheStoresFactsAreRead(t *testing.T) { t.Errorf("error = %v", err) } } + +// 1538: the bucket lists keys by prefix (ListObjectsV2, signed, path-style) and deletes one. +func TestTheBucketListsByPrefixAndDeletes(t *testing.T) { + var gotQuery, gotDelete string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + gotQuery = r.URL.RawQuery + w.Write([]byte(`pending/a/manifest.jsonpending/b/manifest.json`)) + case http.MethodDelete: + gotDelete = r.URL.Path + w.WriteHeader(http.StatusNoContent) + } + })) + defer srv.Close() + at, port, _ := strings.Cut(strings.TrimPrefix(srv.URL, "http://"), ":") + s3 := NewS3(StoreFacts{Scheme: "http", At: at, Port: port, Bucket: "scans", Region: "us-east-1", Key: "AKID", Secret: "sek"}) + keys, err := s3.List(context.Background(), "pending/") + if err != nil || len(keys) != 2 || keys[1] != "pending/b/manifest.json" || !strings.Contains(gotQuery, "prefix=pending%2F") || !strings.Contains(gotQuery, "list-type=2") { + t.Errorf("List = %v, %v (query %q)", keys, err, gotQuery) + } + if err := s3.Delete(context.Background(), "pending/a/001.jpg"); err != nil || gotDelete != "/scans/pending/a/001.jpg" { + t.Errorf("Delete: %v, %q", err, gotDelete) + } +} diff --git a/modules/brother/cmd/brother/printer.go b/modules/brother/cmd/brother/printer.go index d149dcaf..a7acc06e 100644 --- a/modules/brother/cmd/brother/printer.go +++ b/modules/brother/cmd/brother/printer.go @@ -39,6 +39,12 @@ const ( // AnswerWait is how long a verb waits for the device process; ScanWait how long a scan does. AnswerWait = 60 * time.Second ScanWait = 180 * time.Second + // QuickWait is how long a feeder scan is waited for in one call: the mesh console gives up on a call after + // 30 seconds, so a longer scan is answered as started, with its id, and read later (issue 1552). + QuickWait = 25 * time.Second + // FeederWait is how long the tool waits in the background for a feeder scan: the longest the device process + // lets one run (FeederMaxBound) and two minutes to store it. + FeederWait = FeederMaxBound + 2*time.Minute // StaleAfter is how old an ask may be when the device process takes it: older than twice the longest wait, // nobody waits for it any more, so the device is not touched for it. StaleAfter = 2 * ScanWait @@ -51,6 +57,8 @@ const ( OpDocument = "document" OpUpload = "upload" OpPrint = "print" + // OpOpen asks which documents of several pages are open in the bucket (issue 1538). + OpOpen = "open-documents" ) // Ask is what a verb queues: the operation, when it was asked, and the operation's own arguments. @@ -58,6 +66,9 @@ type Ask struct { ID string `json:"id"` Op string `json:"op"` At time.Time `json:"asked-at"` + // Deadline is when the asker stops waiting for the device process to take the ask up: one taken after it is + // dropped before any work, so nothing is scanned or printed for an asker that has gone (issue 1539). + Deadline time.Time `json:"deadline,omitempty"` Page string `json:"page,omitempty"` // settings Links bool `json:"links,omitempty"` // settings: the page's links are answered too @@ -73,6 +84,10 @@ type Ask struct { Media string `json:"media,omitempty"` // print Source string `json:"source,omitempty"` // scan: glass (default) or feeder Duplex bool `json:"duplex,omitempty"` // scan: both sides of each sheet, feeder only + Apart bool `json:"apart,omitempty"` // scan: glass only, each of several documents lying apart is its own file + + Document string `json:"document,omitempty"` // scan: start, or an open document's id: a glass page into it + Finish bool `json:"finish,omitempty"` // scan: close the document named in Document into one PDF } // AskBus is how the tools reach the device process: they queue an ask and read an answer back. @@ -99,6 +114,12 @@ func (sdkAskBus) Answer(id string) (json.RawMessage, error) { var ( answerWait = AnswerWait scanWait = ScanWait + quickWait = QuickWait + feederWait = FeederWait + // scanRun and feederRun are how long the device process may take over a scan once it has taken it up (its own + // bounds, deviceproc.go); the tool follows a scan for its start-by wait and this. + scanRun = 5 * time.Minute + feederRun = FeederMaxBound + 5*time.Minute answerPoll = 500 * time.Millisecond ) @@ -114,6 +135,28 @@ func newAskID() string { // call queues the ask and waits for the device process's answer, at most wait. An answer that is an error is // the error. Nothing here reads, opens or sends a credential (ADR 0436 ยง12). func (s *Service) call(ctx context.Context, ask Ask, wait time.Duration) (map[string]any, error) { + id, err := s.queue(ask, wait) + if err != nil { + return nil, err + } + return s.await(ctx, id, wait) +} + +// queue puts the ask on the work queue and answers its id, under which the answer is written. The asker waits +// startBy for the device process to take the ask up (0 is for as long as it is waited for). +func (s *Service) queue(ask Ask, startBy time.Duration) (string, error) { + ask.ID, ask.At = newAskID(), s.Now().UTC() + if startBy > 0 { + ask.Deadline = ask.At.Add(startBy) + } + if err := s.Asks.Publish(ask.ID, ask); err != nil { + return "", fmt.Errorf("the ask for the device process could not be queued: %w", err) + } + return ask.ID, nil +} + +// await waits at most wait for the answer to the ask with this id. An answer that is an error is the error. +func (s *Service) await(ctx context.Context, id string, wait time.Duration) (map[string]any, error) { poll := s.AnswerPoll if wait <= 0 { wait = AnswerWait @@ -121,14 +164,10 @@ func (s *Service) call(ctx context.Context, ask Ask, wait time.Duration) (map[st if poll <= 0 { poll = 500 * time.Millisecond } - ask.ID, ask.At = newAskID(), s.Now().UTC() - if err := s.Asks.Publish(ask.ID, ask); err != nil { - return nil, fmt.Errorf("the ask for the device process could not be queued: %w", err) - } ctx, cancel := context.WithTimeout(ctx, wait) defer cancel() for { - raw, err := s.Asks.Answer(ask.ID) + raw, err := s.Asks.Answer(id) if err != nil { return nil, fmt.Errorf("the device process's answers cannot be read: %w", err) } @@ -145,7 +184,7 @@ func (s *Service) call(ctx context.Context, ask Ask, wait time.Duration) (map[st select { case <-ctx.Done(): return nil, fmt.Errorf("the device process did not answer within %s: it may not be running, or the device is slow (ask %s)", - wait, ask.ID) + wait, id) case <-time.After(poll): } } @@ -268,6 +307,48 @@ func (s *Service) PrinterUpload(ctx context.Context, a Ask) (map[string]any, err return s.call(ctx, a, s.AnswerWait) } +// PrintFile prints a PDF or a JPEG that lies on the machine the tool runs on: it reads the file, uploads it in +// parts under an upload id of its own and prints that, in one call (issue 1551). The type is read from the +// file's first bytes, never its name; any other file is refused before anything is queued. The answer is the +// print's, with the upload id and the file's size. +func (s *Service) PrintFile(ctx context.Context, path string, a Ask) (map[string]any, error) { + if a.Object != "" || a.Upload != "" { + return nil, fmt.Errorf("name path, or object, or upload: one of the three") + } + // The path is checked first, and refused in one fixed way that tells nothing about the file. + doc, err := ReadDocument(path, MostParts*PartSize) + if err != nil { + return nil, err + } + a.Upload = "p-" + newAskID() + if err := checkPrint(&a); err != nil { + return nil, err + } + if s.Settings.Printer == "" { + return nil, unsetError("printer") + } + if _, _, err := SniffFormat(doc); err != nil { + return nil, fmt.Errorf("file %s: %w", path, err) + } + sum, n := sha256Hex(doc), partCount(len(doc)) + for i := 1; i <= n; i++ { + part, _, _ := cutPartOf(doc, i, PartSize) + up, err := s.PrinterUpload(ctx, Ask{Upload: a.Upload, Part: i, Parts: n, SHA256: sum, Data: base64.StdEncoding.EncodeToString(part)}) + if err != nil { + return nil, fmt.Errorf("part %d of %d of %s could not be uploaded: %w", i, n, path, err) + } + if i == n && up["complete"] != true { + return nil, fmt.Errorf("the upload of %s is not complete after its last part: %v", path, up) + } + } + out, err := s.PrinterPrint(ctx, a) + if err != nil { + return nil, fmt.Errorf("%s is uploaded as %s but could not be printed: %w", path, a.Upload, err) + } + out["path"], out["upload"], out["bytes"] = path, a.Upload, len(doc) + return out, nil +} + // PrinterPrint asks the device process to print a kept scan or a completed upload. func (s *Service) PrinterPrint(ctx context.Context, a Ask) (map[string]any, error) { a.Op = OpPrint diff --git a/modules/brother/cmd/brother/scan.go b/modules/brother/cmd/brother/scan.go index c97c3471..69cd3b36 100644 --- a/modules/brother/cmd/brother/scan.go +++ b/modules/brother/cmd/brother/scan.go @@ -4,6 +4,8 @@ import ( "context" "encoding/base64" "fmt" + "strings" + "sync" "time" ) @@ -23,23 +25,259 @@ type Service struct { // AnswerWait is how long a verb waits for the device process's answer (ScanWait for a scan); AnswerPoll how // often it looks. AnswerWait, ScanWait, AnswerPoll time.Duration + // QuickWait is how long a feeder scan is waited for in one call, FeederWait how long it is followed. + QuickWait, FeederWait time.Duration + // Jobs are the scans this process follows. + Jobs *Jobs +} + +// Jobs are the scans this tool process has started and not yet been asked the outcome of. A scan runs in the +// background from the moment it is asked for, so that it is filed whether or not the caller still waits (issue +// 1539); the caller waits for it, and reads its outcome later with its id when it did not finish in time. +type Jobs struct { + mu sync.Mutex + m map[string]*job + open map[string]*time.Timer // open documents: the timer that closes each + recovered bool // the open documents of the bucket were looked for +} + +// arm (re)starts the timer that closes the open document after d. +func (j *Jobs) arm(doc string, d time.Duration, closeIt func()) { + j.mu.Lock() + defer j.mu.Unlock() + if j.open == nil { + j.open = map[string]*time.Timer{} + } + if t := j.open[doc]; t != nil { + t.Stop() + } + j.open[doc] = time.AfterFunc(d, func() { + j.disarm(doc) + closeIt() + }) +} + +// disarm forgets the open document's timer. +func (j *Jobs) disarm(doc string) { + j.mu.Lock() + defer j.mu.Unlock() + if t := j.open[doc]; t != nil { + t.Stop() + delete(j.open, doc) + } +} + +type job struct { + at time.Time + done chan struct{} + body map[string]any // set when done + err error // set when failed +} + +// jobKeep is how long a finished job's outcome can be read with its id. +var jobKeep = time.Hour + +// docRetry is how long after a failed closing of an expired document it is tried again. +var docRetry = 5 * time.Minute + +// NewJobs is an empty set of jobs. +func NewJobs() *Jobs { return &Jobs{m: map[string]*job{}} } + +func (j *Jobs) get(id string) *job { + if j == nil { + return nil + } + j.mu.Lock() + defer j.mu.Unlock() + return j.m[id] +} + +func (j *Jobs) put(id string) *job { + j.mu.Lock() + defer j.mu.Unlock() + if j.m == nil { + j.m = map[string]*job{} + } + // A job nobody has asked after for jobKeep is forgotten (a finished one only). + for k, o := range j.m { + select { + case <-o.done: + if time.Since(o.at) > jobKeep { + delete(j.m, k) + } + default: + } + } + x := &job{at: time.Now(), done: make(chan struct{})} + j.m[id] = x + return x } // ScanNow asks the device process to scan, crop, build the PDF, keep it in the bucket and say scan.added, waits for // the stored scan's key, fetches its parts through the device process, joins them, checks their SHA-256 and files // the PDF in the scan folder (temp and rename, no overwrite). It answers the event's body and the path. Nothing is // asked before scanner and scan-folder are known to be set (1440/AC7); the tool never reads a credential. -func (s *Service) ScanNow(ctx context.Context, source string, duplex bool) (map[string]any, error) { +// +// The scan is followed by a goroutine of its own to the end, so it is filed even when the caller has gone. A glass +// scan is waited for (ScanWait); a feeder scan only for QuickWait, and when it is not finished by then the answer is +// started: true with the scan's id, and ScanResult reads the outcome later. +func (s *Service) ScanNow(ctx context.Context, source string, duplex, apart bool) (map[string]any, error) { if err := s.Settings.Missing(); err != nil { return nil, err } - if _, err := ParseSource(source, duplex); err != nil { - return nil, err - } - ans, err := s.call(ctx, Ask{Op: OpScan, Source: source, Duplex: duplex}, s.ScanWait) + req, err := ParseScan(source, duplex, apart) if err != nil { return nil, err } + run, quick := scanRun, s.ScanWait + if req.Feeder { + run, quick = feederRun, s.QuickWait + } + return s.runScan(ctx, Ask{Op: OpScan, Source: source, Duplex: duplex, Apart: apart}, run, quick) +} + +// runScan queues the ask and follows it in the background to its end (filing included); the caller is given the +// outcome if it comes within quick, and started: true with the id otherwise. +func (s *Service) runScan(ctx context.Context, ask Ask, run, quick time.Duration) (map[string]any, error) { + id, j, err := s.startScan(ask, run) + if err != nil { + return nil, err + } + select { + case <-j.done: + return j.body, j.err + case <-time.After(quick): + return map[string]any{"started": true, "id": id, "state": "running", + "note": "the scan goes on and is filed when it is done; read its outcome with scanner.scan {id}, or wait for scan.added"}, nil + case <-ctx.Done(): + return nil, fmt.Errorf("the caller went away; the scan %s goes on and is filed: %w", id, ctx.Err()) + } +} + +// startScan queues the ask and starts the goroutine that follows it, however long the caller waits. +// +// The ask carries a deadline, ScanWait from now: the device process drops it unscanned if it is taken up later (it +// may be busy with a long feeder scan). One taken up in time may run for run, and the goroutine waits for that +// too, so a scan that is announced is always filed, however long the caller has waited. +func (s *Service) startScan(ask Ask, run time.Duration) (string, *job, error) { + startBy := s.ScanWait + bound := startBy + run + id, err := s.queue(ask, startBy) + if err != nil { + return "", nil, err + } + j := s.Jobs.put(id) + go func() { + // The goroutine outlives the caller: a context of its own, bounded by the scan's own bound and the time + // to fetch the parts. + ctx, cancel := context.WithTimeout(context.Background(), bound+10*time.Minute) + defer cancel() + j.body, j.err = s.finishScan(ctx, id, bound, ask) + close(j.done) + }() + return id, j, nil +} + +// docExpiry is how long an open document waits for its next page before its pages are filed as one PDF anyway. +var docExpiry = 30 * time.Minute + +// ScanPage scans the glass into a document of several pages (issue 1538). document is start, which opens a +// document and scans its first page into it, or the id of an open one, which scans the next page into it; with +// finish the document is closed into one PDF, filed once (nothing is scanned). An open document that gets no page +// for docExpiry is closed and filed so, by this process: its pages are never lost. +func (s *Service) ScanPage(ctx context.Context, document string, finish bool) (map[string]any, error) { + if err := s.Settings.Missing(); err != nil { + return nil, err + } + if document != "start" && !uploadPattern.MatchString(document) { + return nil, fmt.Errorf("document is start, or the id of an open document: letters, digits, - and _, at most 64") + } + if finish && document == "start" { + return nil, fmt.Errorf("finish closes an open document: name its id in document, not start") + } + return s.runScan(ctx, Ask{Op: OpScan, Document: document, Finish: finish}, scanRun, s.ScanWait) +} + +// ScanResult is the outcome of the scan with this id: its body and path when it is done, state running when it is +// not, and state failed with why when it failed. A scan this process does not know (it was restarted, or the id is +// wrong) is an error. +func (s *Service) ScanResult(id string) (map[string]any, error) { + j := s.Jobs.get(id) + if j == nil { + return nil, fmt.Errorf("no scan %s is known to this tool: its id is wrong, or the tool was restarted since it was asked for (a scan the device process kept is announced by scan.added)", id) + } + select { + case <-j.done: + if j.err != nil { + return map[string]any{"id": id, "state": "failed", "why": j.err.Error()}, nil + } + out := map[string]any{"id": id, "state": "done"} + for k, v := range j.body { + out[k] = v + } + return out, nil + default: + return map[string]any{"started": true, "id": id, "state": "running"}, nil + } +} + +// finishScan waits for the device process's answer to the scan ask, then fetches the stored scan and files it. +func (s *Service) finishScan(ctx context.Context, id string, wait time.Duration, ask Ask) (map[string]any, error) { + ans, err := s.await(ctx, id, wait) + if err != nil { + return nil, err + } + if doc, _ := ans["document"].(string); doc != "" && ans["key"] == nil { + // A page was added to an open document: nothing to file yet, and the document is closed if it is left open. + s.armDocument(doc, docExpiry, 0) + return ans, nil + } + if ask.Document != "" && ask.Finish { + out, err := s.fileAnswer(ctx, ans) + if err == nil { + s.Jobs.disarm(ask.Document) + } + return out, err + } + return s.fileAnswer(ctx, ans) +} + +// fileAnswer fetches the scan the device process's answer names and files it; an answer that lists documents (several +// lying apart on the glass, issue 1528) is each fetched and filed as a PDF of its own, and answered as documents, +// in the order the scanner found them. One that cannot be filed does not hold back the others: it is named in failed. +func (s *Service) fileAnswer(ctx context.Context, ans map[string]any) (map[string]any, error) { + list, many := ans["documents"].([]any) + if !many { + return s.fileOne(ctx, ans) + } + var filed []any + var failed []string + for i, d := range list { + one, _ := d.(map[string]any) + body, err := s.fileOne(ctx, one) + if err != nil { + failed = append(failed, fmt.Sprintf("document %d: %v", i+1, err)) + continue + } + filed = append(filed, body) + } + if bad, ok := ans["failed"].([]any); ok { + for _, b := range bad { + failed = append(failed, fmt.Sprint(b)) + } + } + if len(filed) == 0 { + return nil, fmt.Errorf("no document could be filed: %s", strings.Join(failed, "; ")) + } + out := map[string]any{"documents": filed, "count": len(filed)} + if len(failed) > 0 { + out["failed"] = failed + } + return out, nil +} + +// fileOne fetches the one scan the answer names and files it. +func (s *Service) fileOne(ctx context.Context, ans map[string]any) (map[string]any, error) { key, _ := ans["key"].(string) sum, _ := ans["sha256"].(string) parts, _ := ans["parts"].(float64) @@ -49,6 +287,10 @@ func (s *Service) ScanNow(ctx context.Context, source string, duplex bool) (map[ got := make([][]byte, 0, int(parts)) for i := 1; i <= int(parts); i++ { p, err := s.fetchPart(ctx, key, i) + if err != nil && ctx.Err() == nil { + // A part fetch is a read, safe to repeat: one retry rides out a short bus outage (issue 1536). + p, err = s.fetchPart(ctx, key, i) + } if err != nil { return nil, fmt.Errorf("the scan is kept as %s, but its part %d could not be fetched: %w", key, i, err) } @@ -114,3 +356,61 @@ func (s *Service) ScanList(limit int) (map[string]any, error) { } return map[string]any{"folder": s.Settings.Folder, "scans": scans}, nil } + +// armDocument closes the open document in d, from the expiry on: if that fails (the device process dropped the ask, +// the bucket was away) it is tried again after docRetry, up to ten times, unless the document is gone or closed +// already. A document is never forgotten while it can still be filed. +func (s *Service) armDocument(doc string, d time.Duration, attempts int) { + s.Jobs.arm(doc, d, func() { + _, j, err := s.startScan(Ask{Op: OpScan, Document: doc, Finish: true}, scanRun) + if err == nil { + <-j.done + err = j.err + } + if err == nil || attempts >= 10 || strings.Contains(err.Error(), "is closed already") || strings.Contains(err.Error(), "its id is wrong") { + return + } + s.armDocument(doc, docRetry, attempts+1) + }) +} + +// Recover arms the expiry of every document the device process finds open in the bucket, so that a restart of +// this process forgets none (issue 1538). A document gets what is left of its time since its last page. +func (s *Service) Recover(ctx context.Context) error { + ans, err := s.call(ctx, Ask{Op: OpOpen}, s.AnswerWait) + if err != nil { + return err + } + open, _ := ans["open"].([]any) + for _, o := range open { + m, _ := o.(map[string]any) + id, _ := m["id"].(string) + if id == "" { + continue + } + left := docExpiry + if at, err := time.Parse(time.RFC3339, fmt.Sprint(m["updated"])); err == nil { + left = max(docExpiry-s.Now().Sub(at), time.Second) + } + s.armDocument(id, left, 0) + } + return nil +} + +// tryRecover says whether this process has still to recover its open documents, and takes the turn; failed gives it +// back. +func (j *Jobs) tryRecover() bool { + j.mu.Lock() + defer j.mu.Unlock() + if j.recovered { + return false + } + j.recovered = true + return true +} + +func (j *Jobs) recoverFailed() { + j.mu.Lock() + j.recovered = false + j.mu.Unlock() +} diff --git a/modules/brother/cmd/brother/service_test.go b/modules/brother/cmd/brother/service_test.go index 01f6fc8b..11eea84c 100644 --- a/modules/brother/cmd/brother/service_test.go +++ b/modules/brother/cmd/brother/service_test.go @@ -9,6 +9,7 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "time" @@ -334,3 +335,65 @@ func TestLoadSettingsLaysTheDefaultsUnder(t *testing.T) { t.Fatalf("settings = %+v", s) } } + +// flakyBus fails the first n document asks with an error answer, as a bus outage would end one wait, and passes +// everything else on. +type flakyBus struct { + *fakeAskBus + mu sync.Mutex + fails int + // match says which asks fail; nil is the document fetches. + match func(Ask) bool +} + +func (b *flakyBus) Publish(id string, body any) error { + raw, _ := json.Marshal(body) + var ask Ask + json.Unmarshal(raw, &ask) + b.mu.Lock() + fail := (b.match == nil && ask.Op == OpDocument || b.match != nil && b.match(ask)) && b.fails > 0 + if fail { + b.fails-- + } + b.mu.Unlock() + if fail { + b.answers.Create(id, map[string]any{"error": "nats: timeout"}) + return nil + } + return b.fakeAskBus.Publish(id, body) +} + +// 1536: a part fetch that fails once is asked again, and the scan is filed; one that fails twice is a failure +// naming the stored object, so it can be filed later. +func TestAFailedPartFetchIsAskedOnceMoreBeforeTheScanFails(t *testing.T) { + f := newFakeScanner(t, fixture(t, "scan.jpg")) + r, _ := scanRig(t, f, Settings{Crop: false}) + r.bus = nil + flaky := &flakyBus{fakeAskBus: &fakeAskBus{worker: r.worker, answers: r.answers}, fails: 1} + askBus = flaky + body, err := r.scan(t) + if err != nil { + t.Fatalf("one failed part fetch ended the scan: %v", err) + } + if _, err := os.Stat(body["path"].(string)); err != nil { + t.Errorf("the scan is not filed: %v", err) + } + + f2 := newFakeScanner(t, fixture(t, "scan.jpg")) + r2, folder := scanRig(t, f2, Settings{Crop: false}) + askBus = &flakyBus{fakeAskBus: &fakeAskBus{worker: r2.worker, answers: r2.answers}, fails: 2} + _, err = r2.scan(t) + if err == nil { + t.Fatal("two failed part fetches did not fail the scan") + } + key := "" + if got := *r2.said; len(got) == 1 { + key, _ = got[0].body["object"].(string) + } + if key == "" || !strings.Contains(err.Error(), key) { + t.Errorf("the failure %q does not name the stored object %q", err, key) + } + if entries, _ := os.ReadDir(folder); len(entries) != 0 { + t.Errorf("a scan whose parts failed was filed: %v", entries) + } +} diff --git a/modules/brother/module.json b/modules/brother/module.json index e30094fb..57559d56 100644 --- a/modules/brother/module.json +++ b/modules/brother/module.json @@ -154,7 +154,8 @@ "printer.status", "printer.settings", "printer.print", - "printer.upload" + "printer.upload", + "brother_health" ], "resources": [ { @@ -232,6 +233,25 @@ "health": { "kind": "unit" } + }, + { + "id": "watch", + "type": "process", + "name": "brother-watch", + "artifact": "tools", + "run": [ + "./brother", + "watch" + ], + "user": "brother-device", + "health": { + "kind": "tool", + "tool": "brother_health", + "interval": "60s", + "timeout": "10s", + "looks": 2, + "grace": "60s" + } } ], "build": {