mesh-catalog/modules/mesh-mail/cmd/mesh-mail/objects.go

package main

// The attachment store (novox/hq ADR 0359 §3): each attachment kept once, by its SHA-256, 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 (MESH_MAIL_STORE); this speaks S3 to it with the standard library alone: a HEAD, a PUT and a GET, each
// signed with AWS Signature Version 4, path-style, as an S3-compatible object store answers them.

import (
	"bytes"
	"context"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"sort"
	"strings"
	"sync"
	"time"
)

// Objects is the attachment store.
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)
}

// Keep stores an attachment once by its SHA-256 and answers its reference: the same file mailed ten times is one
// object.
func Keep(ctx context.Context, o Objects, data []byte, mediaType, name string) (AttachmentRef, error) {
	sum := sha256.Sum256(data)
	key := hex.EncodeToString(sum[:])
	has, err := o.Has(ctx, key)
	if err != nil {
		return AttachmentRef{}, err
	}
	if !has {
		if err := o.Put(ctx, key, mediaType, data); err != nil {
			return AttachmentRef{}, err
		}
	}
	return AttachmentRef{Bucket: o.Bucket(), Key: key, Size: int64(len(data)), Type: mediaType, Name: name}, nil
}

// 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 (MESH_MAIL_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 attachment store 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 attachment store 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 attachment store 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 attachment store 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 attachment store 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 attachment store cannot be reached: %w", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("the attachment store 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)
}

// ---- in memory ----------------------------------------------------------------------------------------------------

type memObjects struct {
	mu   sync.Mutex
	name string
	kept map[string][]byte
	puts int
}

func newMemObjects() *memObjects { return &memObjects{name: "mail-test", kept: map[string][]byte{}} }

func (m *memObjects) Bucket() string { return m.name }

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
}