brother: the module requires an s3-bucket, with mesh-mail's S3 client and store.json read by the device account alone (issue 1453, ADR 0468 clause 2)
This commit is contained in:
@@ -11,6 +11,9 @@ import (
|
||||
)
|
||||
|
||||
type manifest struct {
|
||||
Requires []string `json:"requires"`
|
||||
Binds map[string]string `json:"binds"`
|
||||
Secrets map[string]string `json:"secrets"`
|
||||
Emits []string `json:"emits"`
|
||||
OwnSecrets map[string]map[string]any `json:"own-secrets"`
|
||||
Seats []seat `json:"seats"`
|
||||
@@ -241,6 +244,7 @@ func TestTheAdminProcessIsDeclaredAsMeshMailsSenderIs(t *testing.T) {
|
||||
"BROTHER_SETTINGS": "${dir:state}/settings.json",
|
||||
"BROTHER_ADMIN_PASSWORD": m.OwnSecrets["admin-password"]["path"],
|
||||
"BROTHER_BROKER": m.OwnSecrets["broker-device"]["path"],
|
||||
"BROTHER_STORE": "${dir:state}/store.json",
|
||||
}
|
||||
if !reflect.DeepEqual(env, want) {
|
||||
t.Errorf("env = %v, want %v", env, want)
|
||||
@@ -298,3 +302,35 @@ func TestThePrinterAsksStreamIsAWorkQueue(t *testing.T) {
|
||||
t.Errorf("state = %v", m.State)
|
||||
}
|
||||
}
|
||||
|
||||
// 1453/AC8 (ADR 0468 clause 2): the module requires an s3-bucket as mesh-mail does, and its facts and key are
|
||||
// written into store.json, a file only the device account can read; the tools' bundle is not told where it is.
|
||||
func TestTheBucketIsRequiredAndItsKeyIsOnlyTheDeviceAccounts(t *testing.T) {
|
||||
m := readManifest(t)
|
||||
if !reflect.DeepEqual(m.Requires, []string{"s3-bucket"}) {
|
||||
t.Errorf("requires = %v", m.Requires)
|
||||
}
|
||||
if m.Binds["s3-bucket"] != "${dir:state}/store-binding.json" || m.Secrets["s3-bucket"] != "${dir:state}/store.secret" {
|
||||
t.Errorf("binds %v, secrets %v", m.Binds, m.Secrets)
|
||||
}
|
||||
r := resource(t, m, "store")
|
||||
if r["type"] != "file" || r["path"] != "${dir:state}/store.json" || r["mode"] != "0600" || r["owner"] != "brother-device" || r["template"] != "files/store.json" {
|
||||
t.Errorf("store = %v", r)
|
||||
}
|
||||
mine, _ := os.ReadFile(filepath.Join("..", "..", "files", "store.json"))
|
||||
theirs, err := os.ReadFile(filepath.Join("..", "..", "..", "mesh-mail", "files", "store.json"))
|
||||
if err != nil || string(mine) != string(theirs) {
|
||||
t.Errorf("files/store.json is not mesh-mail's (%v)", err)
|
||||
}
|
||||
p := resource(t, m, "device")
|
||||
if env, _ := p["env"].(map[string]any); env["BROTHER_STORE"] != "${dir:state}/store.json" {
|
||||
t.Errorf("the device process's BROTHER_STORE = %v", env["BROTHER_STORE"])
|
||||
}
|
||||
for _, a := range m.Build.Artifacts {
|
||||
if a["kind"] == "bundle" {
|
||||
if _, has := a["env"].(map[string]any)["BROTHER_STORE"]; has {
|
||||
t.Errorf("the tools' bundle is told where the bucket's key is")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,234 @@
|
||||
package main
|
||||
|
||||
// The module's bucket (novox/hq ADR 0468 clause 2): each scan's PDF kept once, by its SHA-256, and each upload
|
||||
// under uploads/, in the bucket the `minio` module provides this module through the `s3-bucket` provision. The
|
||||
// mesh writes the bucket's facts and its key into one file, store.json, which only the device process (account
|
||||
// brother-device) can read (BROTHER_STORE); this speaks S3 to it with the standard library alone, as mesh-mail's
|
||||
// store does (modules/mesh-mail/cmd/mesh-mail/objects.go): a HEAD, a PUT and a GET, each signed with AWS
|
||||
// Signature Version 4, path-style.
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Objects is the module's bucket.
|
||||
type Objects interface {
|
||||
// Bucket is the bucket's name, as an event names it.
|
||||
Bucket() string
|
||||
// Has says whether the key is kept.
|
||||
Has(ctx context.Context, key string) (bool, error)
|
||||
// Put keeps the bytes under the key.
|
||||
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)
|
||||
}
|
||||
|
||||
// Store keeps the bytes under the key once: a key already kept is left as it is, since a key that names the
|
||||
// content (the SHA-256 of a scan) names the same bytes.
|
||||
func Store(ctx context.Context, o Objects, key, mediaType string, data []byte) error {
|
||||
has, err := o.Has(ctx, key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if has {
|
||||
return nil
|
||||
}
|
||||
return o.Put(ctx, key, mediaType, data)
|
||||
}
|
||||
|
||||
// StoreFacts is the file the mesh writes from the s3-bucket binding and its secret.
|
||||
type StoreFacts struct {
|
||||
Scheme string `json:"scheme"`
|
||||
At string `json:"at"`
|
||||
Port string `json:"port"`
|
||||
Bucket string `json:"bucket"`
|
||||
Region string `json:"region"`
|
||||
Key string `json:"key"`
|
||||
Secret string `json:"secret"`
|
||||
}
|
||||
|
||||
// ReadStoreFacts reads the store's facts.
|
||||
func ReadStoreFacts(path string) (StoreFacts, error) {
|
||||
var f StoreFacts
|
||||
if path == "" {
|
||||
return f, errors.New("the module was started without its store's facts (BROTHER_STORE)")
|
||||
}
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return f, fmt.Errorf("the store's facts cannot be read: %w", err)
|
||||
}
|
||||
if err := json.Unmarshal(raw, &f); err != nil {
|
||||
return f, fmt.Errorf("the store's facts are not the module's: %v", err)
|
||||
}
|
||||
if f.At == "" || f.Bucket == "" || f.Key == "" || f.Secret == "" {
|
||||
return f, errors.New("the store's facts name no address, bucket or key")
|
||||
}
|
||||
if f.Scheme == "" {
|
||||
f.Scheme = "http"
|
||||
}
|
||||
if f.Region == "" {
|
||||
f.Region = "us-east-1"
|
||||
}
|
||||
return f, nil
|
||||
}
|
||||
|
||||
// S3 is the module's bucket over S3.
|
||||
type S3 struct {
|
||||
Endpoint string // scheme://host[:port]
|
||||
Name string
|
||||
Region string
|
||||
Access string
|
||||
Secret string
|
||||
Client *http.Client
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
// NewS3 is the store the facts name.
|
||||
func NewS3(f StoreFacts) *S3 {
|
||||
host := f.At
|
||||
if f.Port != "" {
|
||||
host += ":" + f.Port
|
||||
}
|
||||
return &S3{Endpoint: f.Scheme + "://" + host, Name: f.Bucket, Region: f.Region, Access: f.Key, Secret: f.Secret,
|
||||
Client: &http.Client{Timeout: 2 * time.Minute}, Now: time.Now}
|
||||
}
|
||||
|
||||
func (s *S3) Bucket() string { return s.Name }
|
||||
|
||||
func (s *S3) do(ctx context.Context, method, key, mediaType string, body []byte) (*http.Response, error) {
|
||||
u, err := url.Parse(s.Endpoint + "/" + s.Name + "/" + key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, method, u.String(), bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if mediaType != "" {
|
||||
req.Header.Set("Content-Type", mediaType)
|
||||
}
|
||||
req.ContentLength = int64(len(body))
|
||||
Sign(req, body, s.Access, s.Secret, s.Region, s.Now().UTC())
|
||||
return s.Client.Do(req)
|
||||
}
|
||||
|
||||
func (s *S3) Has(ctx context.Context, key string) (bool, error) {
|
||||
resp, err := s.do(ctx, http.MethodHead, key, "", nil)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("the module's bucket cannot be reached: %w", err)
|
||||
}
|
||||
resp.Body.Close()
|
||||
switch resp.StatusCode {
|
||||
case http.StatusOK:
|
||||
return true, nil
|
||||
case http.StatusNotFound:
|
||||
return false, nil
|
||||
}
|
||||
return false, fmt.Errorf("the module's bucket answered %s to asking for %s", resp.Status, key)
|
||||
}
|
||||
|
||||
func (s *S3) Put(ctx context.Context, key, mediaType string, data []byte) error {
|
||||
resp, err := s.do(ctx, http.MethodPut, key, mediaType, data)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the module's bucket cannot be reached: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode/100 != 2 {
|
||||
said, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
|
||||
return fmt.Errorf("the module's bucket refused %s: %s %s", key, resp.Status, strings.TrimSpace(string(said)))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *S3) Get(ctx context.Context, key string) ([]byte, error) {
|
||||
resp, err := s.do(ctx, http.MethodGet, key, "", nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the module's bucket cannot be reached: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("the module's bucket answered %s for %s", resp.Status, key)
|
||||
}
|
||||
return io.ReadAll(resp.Body)
|
||||
}
|
||||
|
||||
// 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) {
|
||||
payload := sha256.Sum256(body)
|
||||
payloadHex := hex.EncodeToString(payload[:])
|
||||
amzDate := now.Format("20060102T150405Z")
|
||||
day := now.Format("20060102")
|
||||
req.Header.Set("x-amz-date", amzDate)
|
||||
req.Header.Set("x-amz-content-sha256", payloadHex)
|
||||
|
||||
headers := map[string]string{"host": req.URL.Host}
|
||||
for k, v := range req.Header {
|
||||
lk := strings.ToLower(k)
|
||||
if lk == "user-agent" || lk == "content-length" || lk == "authorization" {
|
||||
continue
|
||||
}
|
||||
headers[lk] = strings.TrimSpace(strings.Join(v, ","))
|
||||
}
|
||||
names := make([]string, 0, len(headers))
|
||||
for k := range headers {
|
||||
names = append(names, k)
|
||||
}
|
||||
sort.Strings(names)
|
||||
var canonHeaders strings.Builder
|
||||
for _, k := range names {
|
||||
canonHeaders.WriteString(k + ":" + headers[k] + "\n")
|
||||
}
|
||||
signed := strings.Join(names, ";")
|
||||
path := req.URL.EscapedPath()
|
||||
if path == "" {
|
||||
path = "/"
|
||||
}
|
||||
canonical := strings.Join([]string{req.Method, path, canonicalQuery(req.URL.Query()), canonHeaders.String(), signed, payloadHex}, "\n")
|
||||
scope := day + "/" + region + "/s3/aws4_request"
|
||||
sum := sha256.Sum256([]byte(canonical))
|
||||
toSign := "AWS4-HMAC-SHA256\n" + amzDate + "\n" + scope + "\n" + hex.EncodeToString(sum[:])
|
||||
key := hmacOf([]byte("AWS4"+secret), day)
|
||||
key = hmacOf(key, region)
|
||||
key = hmacOf(key, "s3")
|
||||
key = hmacOf(key, "aws4_request")
|
||||
sig := hex.EncodeToString(hmacOf(key, toSign))
|
||||
req.Header.Set("Authorization", "AWS4-HMAC-SHA256 Credential="+access+"/"+scope+", SignedHeaders="+signed+", Signature="+sig)
|
||||
}
|
||||
|
||||
func canonicalQuery(q url.Values) string {
|
||||
keys := make([]string, 0, len(q))
|
||||
for k := range q {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
var parts []string
|
||||
for _, k := range keys {
|
||||
vs := append([]string(nil), q[k]...)
|
||||
sort.Strings(vs)
|
||||
for _, v := range vs {
|
||||
parts = append(parts, url.QueryEscape(k)+"="+strings.ReplaceAll(url.QueryEscape(v), "+", "%20"))
|
||||
}
|
||||
}
|
||||
return strings.Join(parts, "&")
|
||||
}
|
||||
|
||||
func hmacOf(key []byte, data string) []byte {
|
||||
h := hmac.New(sha256.New, key)
|
||||
h.Write([]byte(data))
|
||||
return h.Sum(nil)
|
||||
}
|
||||
@@ -0,0 +1,134 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// memObjects is the bucket in memory, for the tests of what the device process keeps.
|
||||
type memObjects struct {
|
||||
mu sync.Mutex
|
||||
kept map[string][]byte
|
||||
puts int
|
||||
}
|
||||
|
||||
func newMemObjects() *memObjects { return &memObjects{kept: map[string][]byte{}} }
|
||||
|
||||
func (m *memObjects) Bucket() string { return "brother-test" }
|
||||
|
||||
func (m *memObjects) Has(_ context.Context, key string) (bool, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
_, ok := m.kept[key]
|
||||
return ok, nil
|
||||
}
|
||||
|
||||
func (m *memObjects) Put(_ context.Context, key, _ string, data []byte) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.kept[key] = append([]byte(nil), data...)
|
||||
m.puts++
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *memObjects) Get(_ context.Context, key string) ([]byte, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
b, ok := m.kept[key]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("%s is not kept", key)
|
||||
}
|
||||
return b, nil
|
||||
}
|
||||
|
||||
func (m *memObjects) keys() []string {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
var k []string
|
||||
for key := range m.kept {
|
||||
k = append(k, key)
|
||||
}
|
||||
return k
|
||||
}
|
||||
|
||||
// 1453/AC8: the S3 client puts, heads and gets path-style under a Signature Version 4 header, and Store keeps
|
||||
// a key once.
|
||||
func TestTheBucketIsSpokenToPathStyleSignedAndKeptOnce(t *testing.T) {
|
||||
var mu sync.Mutex
|
||||
objects := map[string][]byte{}
|
||||
var puts int
|
||||
var auth string
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
auth = r.Header.Get("Authorization")
|
||||
if !strings.HasPrefix(r.URL.Path, "/scans/") {
|
||||
http.Error(w, "path-style: /<bucket>/<key>", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
key := strings.TrimPrefix(r.URL.Path, "/scans/")
|
||||
switch r.Method {
|
||||
case http.MethodPut:
|
||||
b, _ := io.ReadAll(r.Body)
|
||||
objects[key] = b
|
||||
puts++
|
||||
case http.MethodHead, http.MethodGet:
|
||||
b, ok := objects[key]
|
||||
if !ok {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
if r.Method == http.MethodGet {
|
||||
w.Write(b)
|
||||
}
|
||||
}
|
||||
}))
|
||||
defer srv.Close()
|
||||
host := strings.TrimPrefix(srv.URL, "http://")
|
||||
at, port, _ := strings.Cut(host, ":")
|
||||
s3 := NewS3(StoreFacts{Scheme: "http", At: at, Port: port, Bucket: "scans", Region: "us-east-1", Key: "AKID", Secret: "sek"})
|
||||
ctx := context.Background()
|
||||
for i := 0; i < 2; i++ {
|
||||
if err := Store(ctx, s3, "abc.pdf", "application/pdf", []byte("%PDF-x")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if puts != 1 {
|
||||
t.Errorf("%d puts, want the one: a key is kept once", puts)
|
||||
}
|
||||
got, err := s3.Get(ctx, "abc.pdf")
|
||||
if err != nil || string(got) != "%PDF-x" {
|
||||
t.Errorf("Get = %q, %v", got, err)
|
||||
}
|
||||
if !strings.HasPrefix(auth, "AWS4-HMAC-SHA256 Credential=AKID/") || !strings.Contains(auth, "SignedHeaders=") || strings.Contains(auth, "sek") {
|
||||
t.Errorf("Authorization = %q", auth)
|
||||
}
|
||||
if _, err := s3.Get(ctx, "none.pdf"); err == nil {
|
||||
t.Errorf("a key not kept was got")
|
||||
}
|
||||
}
|
||||
|
||||
// 1453: the store's facts are read from store.json, and one that names no bucket or key is refused by name.
|
||||
func TestTheStoresFactsAreRead(t *testing.T) {
|
||||
p := filepath.Join(t.TempDir(), "store.json")
|
||||
os.WriteFile(p, []byte(`{"at":"minio.example","port":"9000","bucket":"brother","key":"k","secret":"s"}`), 0o600)
|
||||
f, err := ReadStoreFacts(p)
|
||||
if err != nil || f.Scheme != "http" || f.Region != "us-east-1" || f.Bucket != "brother" {
|
||||
t.Errorf("facts = %+v, %v", f, err)
|
||||
}
|
||||
os.WriteFile(p, []byte(`{"at":"x"}`), 0o600)
|
||||
if _, err := ReadStoreFacts(p); err == nil || !strings.Contains(err.Error(), "bucket") {
|
||||
t.Errorf("error = %v", err)
|
||||
}
|
||||
if _, err := ReadStoreFacts(""); err == nil || !strings.Contains(err.Error(), "BROTHER_STORE") {
|
||||
t.Errorf("error = %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
{
|
||||
"scheme": "${bound:s3-bucket:scheme}",
|
||||
"at": "${bound:s3-bucket:at}",
|
||||
"port": "${bound:s3-bucket:port}",
|
||||
"bucket": "${bound:s3-bucket:bucket}",
|
||||
"region": "${bound:s3-bucket:region}",
|
||||
"key": "${bound:s3-bucket:as}",
|
||||
"secret": "${secret:s3-bucket}"
|
||||
}
|
||||
@@ -1,6 +1,15 @@
|
||||
{
|
||||
"module": "brother",
|
||||
"version": "1",
|
||||
"requires": [
|
||||
"s3-bucket"
|
||||
],
|
||||
"binds": {
|
||||
"s3-bucket": "${dir:state}/store-binding.json"
|
||||
},
|
||||
"secrets": {
|
||||
"s3-bucket": "${dir:state}/store.secret"
|
||||
},
|
||||
"emits": [
|
||||
"scan.added"
|
||||
],
|
||||
@@ -170,6 +179,14 @@
|
||||
"merge": "json",
|
||||
"template": "files/settings.json"
|
||||
},
|
||||
{
|
||||
"id": "store",
|
||||
"type": "file",
|
||||
"path": "${dir:state}/store.json",
|
||||
"owner": "brother-device",
|
||||
"mode": "0600",
|
||||
"template": "files/store.json"
|
||||
},
|
||||
{
|
||||
"id": "device",
|
||||
"type": "process",
|
||||
@@ -183,10 +200,12 @@
|
||||
"env": {
|
||||
"BROTHER_SETTINGS": "${dir:state}/settings.json",
|
||||
"BROTHER_ADMIN_PASSWORD": "${dir:state}/admin-password.secret",
|
||||
"BROTHER_BROKER": "${dir:state}/broker-device.secret"
|
||||
"BROTHER_BROKER": "${dir:state}/broker-device.secret",
|
||||
"BROTHER_STORE": "${dir:state}/store.json"
|
||||
},
|
||||
"restart-on": [
|
||||
"settings"
|
||||
"settings",
|
||||
"store"
|
||||
],
|
||||
"bus": {
|
||||
"queues": [
|
||||
|
||||
Reference in New Issue
Block a user