294 lines
10 KiB
Go
294 lines
10 KiB
Go
package storageagent
|
||
|
||
import (
|
||
"bytes"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"os"
|
||
"path"
|
||
"path/filepath"
|
||
"strings"
|
||
)
|
||
|
||
// Запись в хранилище (docs/library-and-storage.md#запись-в-хранилище): только импорт одобренных загрузок,
|
||
// только создание новых файлов. Перезаписи и удаления файлов библиотеки в коде агента нет.
|
||
|
||
// Paths — каталоги агента.
|
||
type Paths struct {
|
||
Library string // ro-точки NFS: <Library>/<source_id> (видны api, nginx, analyzer)
|
||
LibraryRW string // rw-точки NFS-источников с accepts_uploads; видны только агенту. Пусто — запись в NFS выключена
|
||
HostRW string // LIBRARY_HOST на запись (local-источники); пусто — запись в local выключена
|
||
Uploads string // staging очереди на добавление (только чтение)
|
||
FileUID int // владелец новых файлов и каталогов; < 0 — не менять
|
||
FileGID int
|
||
}
|
||
|
||
// WriteRequest — тело POST /internal/files.
|
||
type WriteRequest struct {
|
||
SourceID string `json:"source_id"`
|
||
LocalPath *string `json:"local_path,omitempty"` // для local-источника: папка внутри LIBRARY_HOST
|
||
RelPath string `json:"rel_path"` // желаемый путь в хранилище
|
||
StagingPath string `json:"staging_path"` // путь файла относительно каталога staging
|
||
SHA256 string `json:"sha256"` // ожидаемый хэш (hex), копия сверяется с ним
|
||
}
|
||
|
||
// WriteResult — где файл оказался (при конфликте имени — с суффиксом « (2)»).
|
||
type WriteResult struct {
|
||
RelPath string `json:"rel_path"`
|
||
Size int64 `json:"size"`
|
||
Renamed bool `json:"renamed"`
|
||
}
|
||
|
||
// WriteTestRequest — тело POST /internal/files/test.
|
||
type WriteTestRequest struct {
|
||
SourceID string `json:"source_id"`
|
||
LocalPath *string `json:"local_path,omitempty"`
|
||
}
|
||
|
||
// WriteTestResult — можно ли писать в хранилище.
|
||
type WriteTestResult struct {
|
||
OK bool `json:"ok"`
|
||
Message string `json:"message,omitempty"`
|
||
}
|
||
|
||
const maxConflicts = 100
|
||
|
||
// errBadPath — путь вне корня, абсолютный или с «..».
|
||
var errBadPath = errors.New("path must be relative, without .. and hidden segments")
|
||
|
||
// cleanRel проверяет относительный путь с `/` и возвращает его в каноническом виде.
|
||
func cleanRel(rel string) (string, error) {
|
||
if rel == "" || strings.HasPrefix(rel, "/") || strings.ContainsAny(rel, "\\\x00") {
|
||
return "", errBadPath
|
||
}
|
||
for _, seg := range strings.Split(rel, "/") {
|
||
if seg == "" || seg == "." || seg == ".." || strings.HasPrefix(seg, ".") {
|
||
return "", errBadPath
|
||
}
|
||
}
|
||
return path.Clean(rel), nil
|
||
}
|
||
|
||
// writeRoot — каталог источника на запись.
|
||
func (a *agent) writeRoot(sourceID string, localPath *string) (string, error) {
|
||
if localPath != nil {
|
||
if a.paths.HostRW == "" {
|
||
return "", errors.New("writing to local sources is disabled")
|
||
}
|
||
root := a.paths.HostRW
|
||
if *localPath != "" {
|
||
rel, err := cleanRel(strings.Trim(*localPath, "/"))
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
root = filepath.Join(root, filepath.FromSlash(rel))
|
||
}
|
||
if info, err := os.Stat(root); err != nil || !info.IsDir() {
|
||
return "", errors.New("source folder not found")
|
||
}
|
||
return root, nil
|
||
}
|
||
if a.paths.LibraryRW == "" {
|
||
return "", errors.New("writing to NFS sources is disabled")
|
||
}
|
||
root, err := mountPoint(a.paths.LibraryRW, sourceID)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
if _, err := os.Stat(root); err != nil {
|
||
return "", errors.New("source is not mounted for writing (accepts_uploads is off?)")
|
||
}
|
||
return root, nil
|
||
}
|
||
|
||
func (a *agent) writeFile(w http.ResponseWriter, r *http.Request) {
|
||
var req WriteRequest
|
||
if !a.decode(w, r, &req) {
|
||
return
|
||
}
|
||
root, err := a.writeRoot(req.SourceID, req.LocalPath)
|
||
if err != nil {
|
||
writeValidation(w, &ValidationError{Field: "source_id", Reason: err.Error()})
|
||
return
|
||
}
|
||
rel, err := cleanRel(req.RelPath)
|
||
if err != nil {
|
||
writeValidation(w, &ValidationError{Field: "rel_path", Reason: err.Error()})
|
||
return
|
||
}
|
||
staging, err := cleanRel(req.StagingPath)
|
||
if err != nil || a.paths.Uploads == "" {
|
||
writeValidation(w, &ValidationError{Field: "staging_path", Reason: "invalid staging path"})
|
||
return
|
||
}
|
||
want, err := hex.DecodeString(req.SHA256)
|
||
if err != nil || len(want) != sha256.Size {
|
||
writeValidation(w, &ValidationError{Field: "sha256", Reason: "must be 64 hex characters"})
|
||
return
|
||
}
|
||
res, err := a.write(root, rel, filepath.Join(a.paths.Uploads, filepath.FromSlash(staging)), want)
|
||
if err != nil {
|
||
a.log.ErrorContext(r.Context(), "write file", "source_id", req.SourceID, "rel_path", rel, "err", err)
|
||
writeJSON(w, http.StatusInternalServerError, map[string]any{
|
||
"status": http.StatusInternalServerError, "code": "storage.write_failed", "detail": err.Error(),
|
||
})
|
||
return
|
||
}
|
||
a.log.InfoContext(r.Context(), "file written", "source_id", req.SourceID, "rel_path", res.RelPath, "size", res.Size)
|
||
writeJSON(w, http.StatusOK, res)
|
||
}
|
||
|
||
// write копирует staging сразу под итоговым именем (O_EXCL — существующий файл не перезаписывается, при
|
||
// конфликте — « (2)»), fsync, сверяет SHA-256 копии. На музыкальный диск не попадает ничего, кроме трека и его
|
||
// картинок cover/artist: ни временных файлов, ни служебных пометок. При любой ошибке недописанный файл и созданные для него пустые
|
||
// каталоги удаляются — агент убирает только то, что создал сам.
|
||
func (a *agent) write(root, rel, staging string, want []byte) (WriteResult, error) {
|
||
dir := filepath.Dir(filepath.Join(root, filepath.FromSlash(rel)))
|
||
created, err := a.mkdirs(root, dir)
|
||
if err != nil {
|
||
return WriteResult{}, err
|
||
}
|
||
fail := func(err error) (WriteResult, error) {
|
||
removeEmpty(created)
|
||
return WriteResult{}, err
|
||
}
|
||
|
||
ext := path.Ext(rel)
|
||
base := strings.TrimSuffix(rel, ext)
|
||
for n := 1; n <= maxConflicts; n++ {
|
||
name := rel
|
||
if n > 1 {
|
||
name = fmt.Sprintf("%s (%d)%s", base, n, ext)
|
||
}
|
||
target := filepath.Join(root, filepath.FromSlash(name))
|
||
dst, err := os.OpenFile(target, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o644)
|
||
if os.IsExist(err) {
|
||
continue
|
||
}
|
||
if err != nil {
|
||
return fail(fmt.Errorf("create: %w", err))
|
||
}
|
||
size, err := copyInto(dst, staging)
|
||
if err == nil {
|
||
var got []byte
|
||
if got, err = fileSHA256(target); err == nil && !bytes.Equal(got, want) {
|
||
err = fmt.Errorf("checksum mismatch: copy %x, expected %x", got, want)
|
||
}
|
||
}
|
||
if err != nil {
|
||
_ = os.Remove(target) // наш собственный недописанный / испорченный файл
|
||
return fail(err)
|
||
}
|
||
a.chown(target)
|
||
syncDir(dir)
|
||
return WriteResult{RelPath: name, Size: size, Renamed: n > 1}, nil
|
||
}
|
||
return fail(errors.New("too many files with the same name"))
|
||
}
|
||
|
||
// mkdirs создаёт недостающие каталоги от root до dir (0755, владелец — FileUID/FileGID) и возвращает созданные.
|
||
func (a *agent) mkdirs(root, dir string) ([]string, error) {
|
||
rel, err := filepath.Rel(root, dir)
|
||
if err != nil || strings.HasPrefix(rel, "..") {
|
||
return nil, errBadPath
|
||
}
|
||
var created []string
|
||
cur := root
|
||
for _, seg := range strings.Split(rel, string(filepath.Separator)) {
|
||
if seg == "." || seg == "" {
|
||
continue
|
||
}
|
||
cur = filepath.Join(cur, seg)
|
||
if err := os.Mkdir(cur, 0o755); err != nil {
|
||
if os.IsExist(err) {
|
||
continue
|
||
}
|
||
removeEmpty(created)
|
||
return nil, fmt.Errorf("mkdir: %w", err)
|
||
}
|
||
created = append(created, cur)
|
||
a.chown(cur)
|
||
}
|
||
return created, nil
|
||
}
|
||
|
||
// removeEmpty удаляет созданные каталоги от глубоких к верхним; os.Remove не трогает непустые.
|
||
func removeEmpty(dirs []string) {
|
||
for i := len(dirs) - 1; i >= 0; i-- {
|
||
_ = os.Remove(dirs[i])
|
||
}
|
||
}
|
||
|
||
// copyInto копирует staging в открытый файл и делает fsync; файл закрывается.
|
||
func copyInto(dst *os.File, staging string) (int64, error) {
|
||
src, err := os.Open(staging)
|
||
if err != nil {
|
||
_ = dst.Close()
|
||
return 0, fmt.Errorf("staging: %w", err)
|
||
}
|
||
defer func() { _ = src.Close() }()
|
||
n, err := io.Copy(dst, src)
|
||
if err == nil {
|
||
err = dst.Sync()
|
||
}
|
||
if cerr := dst.Close(); err == nil {
|
||
err = cerr
|
||
}
|
||
if err != nil {
|
||
return 0, fmt.Errorf("copy: %w", err)
|
||
}
|
||
return n, nil
|
||
}
|
||
|
||
func (a *agent) chown(p string) {
|
||
if a.paths.FileUID >= 0 {
|
||
if err := os.Lchown(p, a.paths.FileUID, a.paths.FileGID); err != nil {
|
||
a.log.Warn("chown failed", "path", p, "err", err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// testWrite проверяет право записи в корень источника, ничего не создавая на диске (access(W_OK);
|
||
// на NFS ответ даёт сам сервер с учётом root_squash и прав экспорта).
|
||
func (a *agent) testWrite(w http.ResponseWriter, r *http.Request) {
|
||
var req WriteTestRequest
|
||
if !a.decode(w, r, &req) {
|
||
return
|
||
}
|
||
root, err := a.writeRoot(req.SourceID, req.LocalPath)
|
||
if err != nil {
|
||
writeJSON(w, http.StatusOK, WriteTestResult{Message: err.Error()})
|
||
return
|
||
}
|
||
if err := canWrite(root); err != nil {
|
||
writeJSON(w, http.StatusOK, WriteTestResult{Message: "cannot write: " + err.Error()})
|
||
return
|
||
}
|
||
writeJSON(w, http.StatusOK, WriteTestResult{OK: true, Message: "write access ok"})
|
||
}
|
||
|
||
func fileSHA256(p string) ([]byte, error) {
|
||
f, err := os.Open(p)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer func() { _ = f.Close() }()
|
||
h := sha256.New()
|
||
if _, err := io.Copy(h, f); err != nil {
|
||
return nil, err
|
||
}
|
||
return h.Sum(nil), nil
|
||
}
|
||
|
||
// syncDir — fsync каталога, чтобы новое имя пережило сбой питания (на Windows не поддерживается — молча).
|
||
func syncDir(dir string) {
|
||
if d, err := os.Open(dir); err == nil {
|
||
_ = d.Sync()
|
||
_ = d.Close()
|
||
}
|
||
}
|