260 lines
8.8 KiB
Go
260 lines
8.8 KiB
Go
package main
|
|
|
|
// upload and download (novox/hq ADR 0468 §9, issue 1597): a file of this node goes into a bucket of the object store,
|
|
// and an object comes out of one into a file of this node, over a short signed link and never over the bus.
|
|
//
|
|
// upload sends a file's content off the node, so it is judged as a read (ADR 0394 §7): before anything else it
|
|
// refuses by name what a read refuses, a link's target included. download writes a file, so it is judged as a write
|
|
// (ADR 0394 §3 to §5): a path a module owns is refused without an override, the old file goes to the trash, and the
|
|
// act is one line in the act log.
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"regexp"
|
|
"strings"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"git.novox.be/novox/mesh-sdk/go/objectlink"
|
|
)
|
|
|
|
// LinkSeconds is how long the link asked for one transfer lives, and TransferTime how long a transfer may take: the
|
|
// tool runner cuts every tool call at 30 s, so a transfer that is not done in 25 s is refused rather than left to run
|
|
// on after the caller was told it failed. Nothing is kept of a transfer that did not finish.
|
|
const (
|
|
LinkSeconds = 300
|
|
TransferTime = 25 * time.Second
|
|
)
|
|
|
|
func (f *Files) bound() time.Duration {
|
|
if f.Bound > 0 {
|
|
return f.Bound
|
|
}
|
|
return TransferTime
|
|
}
|
|
|
|
// tooSlow says a transfer that ran out of time in words, naming the bound.
|
|
func (f *Files) tooSlow(err error, what string) error {
|
|
if errors.Is(err, context.DeadlineExceeded) || strings.Contains(err.Error(), "context deadline exceeded") ||
|
|
strings.Contains(err.Error(), "Client.Timeout") {
|
|
return fmt.Errorf("%s did not finish within %s (a tool call is cut at 30 s), so it is refused and nothing is kept", what, f.bound())
|
|
}
|
|
return err
|
|
}
|
|
|
|
var sha256Key = regexp.MustCompile(`^[0-9a-f]{64}$`)
|
|
|
|
// askLink asks the object store's seat for one link (ADR 0468 §8). The answer is {link, method, seconds, expires-at}.
|
|
func askLink(bucket, key, method string, seconds int) (string, error) {
|
|
raw, err := stdio.Ask("seat:object-store.link", map[string]any{"bucket": bucket, "key": key, "method": method, "seconds": seconds})
|
|
if err != nil {
|
|
return "", fmt.Errorf("the object store gave no %s link: %w", method, err)
|
|
}
|
|
var a struct {
|
|
Link string `json:"link"`
|
|
Answer json.RawMessage `json:"answer"`
|
|
}
|
|
if err := json.Unmarshal(raw, &a); err != nil {
|
|
return "", fmt.Errorf("the object store's answer is not JSON: %w", err)
|
|
}
|
|
if a.Link == "" && len(a.Answer) > 0 {
|
|
_ = json.Unmarshal(a.Answer, &a)
|
|
}
|
|
if a.Link == "" {
|
|
return "", errors.New("the object store's answer holds no link")
|
|
}
|
|
return a.Link, nil
|
|
}
|
|
|
|
func (f *Files) link(bucket, key, method string) (string, error) {
|
|
if f.Link == nil {
|
|
return "", errors.New("this holder has no way to ask the object store for a link")
|
|
}
|
|
return f.Link(bucket, key, method, LinkSeconds)
|
|
}
|
|
|
|
func (f *Files) client() *http.Client {
|
|
if f.HTTP != nil {
|
|
return f.HTTP
|
|
}
|
|
return &http.Client{Timeout: f.bound()}
|
|
}
|
|
|
|
// UploadAnswer is what upload answers.
|
|
type UploadAnswer struct {
|
|
Bucket string `json:"bucket"`
|
|
Key string `json:"key"`
|
|
Size int64 `json:"size"`
|
|
SHA256 string `json:"sha256"`
|
|
}
|
|
|
|
// Upload sends one file of this node into a bucket. The key is the file's SHA-256 when none is given; the hash is
|
|
// then made first (one read of the file), so the link is asked for the key it will be used for.
|
|
func (f *Files) Upload(raw, bucket, key string) (UploadAnswer, error) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), f.bound())
|
|
defer cancel()
|
|
if strings.TrimSpace(bucket) == "" {
|
|
return UploadAnswer{}, errors.New("bucket is needed: name the bucket the file goes into")
|
|
}
|
|
path, err := absolute(raw)
|
|
if err != nil {
|
|
return UploadAnswer{}, err
|
|
}
|
|
// Refused by name before anything else: no link is asked and nothing is read for a place a secret is kept.
|
|
if err := f.refusedRead(path); err != nil {
|
|
return UploadAnswer{}, err
|
|
}
|
|
in, err := os.Open(path)
|
|
if err != nil {
|
|
return UploadAnswer{}, fmt.Errorf("%s cannot be read by this account: %s", path, why(err))
|
|
}
|
|
defer in.Close()
|
|
info, err := in.Stat()
|
|
if err != nil {
|
|
return UploadAnswer{}, err
|
|
}
|
|
if !info.Mode().IsRegular() {
|
|
return UploadAnswer{}, fmt.Errorf("%s is a %s, not a file: upload sends one file", path, typeOf(info.Mode()))
|
|
}
|
|
if info.Size() > objectlink.MaxDocument {
|
|
return UploadAnswer{}, fmt.Errorf("%s is %d bytes, over the %d a document may be (novox/hq ADR 0468 §3)", path, info.Size(), objectlink.MaxDocument)
|
|
}
|
|
want := ""
|
|
if strings.TrimSpace(key) == "" {
|
|
sum, err := hashOf(in)
|
|
if err != nil {
|
|
return UploadAnswer{}, fmt.Errorf("%s could not be read: %v", path, err)
|
|
}
|
|
if _, err := in.Seek(0, io.SeekStart); err != nil {
|
|
return UploadAnswer{}, err
|
|
}
|
|
key, want = sum, sum
|
|
}
|
|
link, err := f.link(bucket, key, "put")
|
|
if err != nil {
|
|
return UploadAnswer{}, err
|
|
}
|
|
res, err := objectlink.Put(ctx, f.client(), link, in, info.Size(), objectlink.MaxDocument)
|
|
if err != nil {
|
|
return UploadAnswer{}, f.tooSlow(fmt.Errorf("%s could not be sent: %w", path, err), "the upload of "+path)
|
|
}
|
|
if want != "" && res.SHA256 != want {
|
|
return UploadAnswer{}, fmt.Errorf("%s changed while it was sent: it was stored under %s but hashes to %s", path, want, res.SHA256)
|
|
}
|
|
return UploadAnswer{Bucket: bucket, Key: key, Size: res.Size, SHA256: res.SHA256}, nil
|
|
}
|
|
|
|
func hashOf(r io.Reader) (string, error) {
|
|
h := sha256.New()
|
|
if _, err := io.Copy(h, r); err != nil {
|
|
return "", err
|
|
}
|
|
return hex.EncodeToString(h.Sum(nil)), nil
|
|
}
|
|
|
|
// DownloadAnswer is what download answers.
|
|
type DownloadAnswer struct {
|
|
ActAnswer
|
|
Size int64 `json:"size"`
|
|
SHA256 string `json:"sha256"`
|
|
}
|
|
|
|
// Download writes one object of a bucket to a file of this node. The bytes must hash to sha256 when it is given, or
|
|
// to the key when the key is itself a SHA-256 (the store keeps a document under its SHA-256, ADR 0468 §2); a
|
|
// mismatch leaves no file, and an old file is left as it was.
|
|
func (f *Files) Download(a Act, bucket, key, raw, wantSHA string) (DownloadAnswer, error) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if err := f.begin(a); err != nil {
|
|
return DownloadAnswer{}, err
|
|
}
|
|
if strings.TrimSpace(bucket) == "" || strings.TrimSpace(key) == "" {
|
|
return DownloadAnswer{}, errors.New("bucket and key are needed: name the object to fetch")
|
|
}
|
|
path, err := absolute(raw)
|
|
if err != nil {
|
|
return DownloadAnswer{}, err
|
|
}
|
|
hits, err := f.guard(a, false, path)
|
|
if err != nil {
|
|
return DownloadAnswer{}, err
|
|
}
|
|
dir := filepath.Dir(path)
|
|
if _, err := os.Stat(dir); err != nil {
|
|
return DownloadAnswer{}, fmt.Errorf("the folder of %s cannot be reached: %s", path, why(err))
|
|
}
|
|
expect := strings.ToLower(strings.TrimSpace(wantSHA))
|
|
if expect == "" && sha256Key.MatchString(key) {
|
|
expect = key
|
|
}
|
|
var perm uint32 = 0o644
|
|
want, exists := Ident{}, false
|
|
if old, err := os.Lstat(path); err == nil {
|
|
if !old.Mode().IsRegular() {
|
|
return DownloadAnswer{}, fmt.Errorf("%s is a %s, not a file: download replaces a file whole", path, typeOf(old.Mode()))
|
|
}
|
|
exists, perm = true, permOf(old.Mode())
|
|
if want, err = f.ownerOf(path); err != nil {
|
|
return DownloadAnswer{}, err
|
|
}
|
|
} else if want, err = f.ownerOf(dir); err != nil {
|
|
return DownloadAnswer{}, err
|
|
}
|
|
link, err := f.link(bucket, key, "get")
|
|
if err != nil {
|
|
return DownloadAnswer{}, err
|
|
}
|
|
tmp, err := os.CreateTemp(dir, ".download-*")
|
|
if err != nil {
|
|
return DownloadAnswer{}, fmt.Errorf("a file cannot be made in the folder of %s by this account: %s", path, why(err))
|
|
}
|
|
done := false
|
|
defer func() {
|
|
tmp.Close()
|
|
if !done {
|
|
os.Remove(tmp.Name())
|
|
}
|
|
}()
|
|
ctx, cancel := context.WithTimeout(context.Background(), f.bound())
|
|
defer cancel()
|
|
res, err := objectlink.Get(ctx, f.client(), link, tmp, objectlink.MaxDocument, expect)
|
|
if err != nil {
|
|
return DownloadAnswer{}, f.tooSlow(fmt.Errorf("%s/%s could not be fetched: %w", bucket, key, err), "the download of "+bucket+"/"+key)
|
|
}
|
|
if err := tmp.Close(); err != nil {
|
|
return DownloadAnswer{}, err
|
|
}
|
|
if err := f.setMode(tmp.Name(), perm); err != nil {
|
|
return DownloadAnswer{}, err
|
|
}
|
|
if err := f.chownTo(tmp.Name(), want); err != nil {
|
|
return DownloadAnswer{}, fmt.Errorf("the owner of %s could not be set: %v", path, err)
|
|
}
|
|
trashed := ""
|
|
if exists {
|
|
if trashed, err = f.toTrash(a, path); err != nil {
|
|
return DownloadAnswer{}, err
|
|
}
|
|
}
|
|
if err := os.Rename(tmp.Name(), path); err != nil {
|
|
if trashed != "" {
|
|
if _, rerr := f.restore(trashed); rerr != nil {
|
|
err = fmt.Errorf("%v (and the old file could not be put back from the trash, id %s: %v)", err, trashed, rerr)
|
|
}
|
|
}
|
|
return DownloadAnswer{}, err
|
|
}
|
|
done = true
|
|
ans, err := f.finish(a, "download", hits, ActAnswer{Done: "wrote " + path, Paths: []string{path}, Trashed: trashed})
|
|
return DownloadAnswer{ActAnswer: ans, Size: res.Size, SHA256: res.SHA256}, err
|
|
}
|