v0.128.0: chunked browser .fab upload on /import (tunnel-proof)

Cloudflare edge caps request bodies (~100 MB, probed live: 120 MiB -> edge 413, 80 MiB -> origin), so the client slices the file into 64 MiB strictly-sequential chunks; the server streams each to a .part file in the default drive's exports dir and finalize renames atomically. Scan/validate/import pipeline untouched. upload/{init,chunk,finalize,abort} inside ServeExportAPI (inherits auth+CSRF); single-flight; offset==received or 409+echo; free-space gate; collision -> lowest-free "name (N).fab"; startup GC of *.part-*; 15-min idle abort. appexport.DiskFree exported (seam web.uploadDiskFree). Scenarios A-F tested, red-proofs run (traversal / out-of-order / overwrite).
2026-07-13 21:09:20 +02:00
parent 59e4ca70de
commit db16371f47
11 changed files with 1055 additions and 8 deletions
@@ -0,0 +1,420 @@
+package web
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-controller/internal/appexport"
+	"gitea.dooplex.hu/admin/felhom-controller/internal/logx"
+)
+
+// Browser .fab upload (v0.128.0) — chunked, tunnel-proof. Cloudflare caps request bodies at
+// ~100 MB on the free plan (step-0 probe 2026-07-13: a 120 MiB POST to the real tunnel got an
+// edge HTTP 413 before the origin saw it; 80 MiB passed through), so the client slices the file
+// and the server appends strictly sequential chunks to a .part file in the DEFAULT drive's
+// exports dir; finalize renames atomically. The existing bundle scan + import pipeline take over
+// untouched — the landing dir is exactly what isValidExportPath and ScanForBundles already cover.
+//
+// No client-side sha256 — deliberate: WebCrypto cannot stream-hash multi-GB files, and the .fab
+// format self-validates at import (manifest + crypto checks in the import pipeline). Transport
+// integrity = strict sequential offsets + exact declared final size + the format's own validation.
+//
+// Upload state is in-memory only: a controller restart loses the .part (the browser re-uploads —
+// honest and simple; no resume machinery). Startup GC removes stray *.part-* files; an upload
+// idle for uploadIdleTimeout is aborted server-side.
+
+const (
+	// uploadChunkBytes is the chunk size handed to the client. 64 MiB clears the ~100 MB
+	// Cloudflare edge cap with headroom (step-0 probe above).
+	uploadChunkBytes = 64 << 20
+	// uploadMaxChunkBody is the server-side per-request body cap — above the advertised chunk
+	// size, below the edge cap.
+	uploadMaxChunkBody = 96 << 20
+	// uploadFreeSpaceMargin is the extra free space required beyond the declared file size.
+	uploadFreeSpaceMargin = 1 << 30
+)
+
+// uploadIdleTimeout aborts an upload with no chunk activity (package var: test seam).
+var uploadIdleTimeout = 15 * time.Minute
+
+// uploadDiskFree is the free-space probe (package var: test seam — scenario C fakes statfs).
+var uploadDiskFree = appexport.DiskFree
+
+// uploadJob is one in-flight browser upload.
+type uploadJob struct {
+	id        string
+	finalName string // sanitized target name ("app.fab")
+	partPath  string // the .part-<random> temp file in the exports dir
+	declared  int64  // size_bytes from init
+	received  int64  // bytes appended so far
+	file      *os.File
+	writing   bool // a chunk body is streaming to the file right now
+	timer     *time.Timer
+}
+
+// uploadState is the single-flight slot: ONE upload at a time (second init → 409).
+type uploadState struct {
+	mu  sync.Mutex
+	cur *uploadJob
+}
+
+// uploadJSON writes the upload API envelope with an explicit status code.
+func uploadJSON(w http.ResponseWriter, code int, v map[string]interface{}) {
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	w.WriteHeader(code)
+	json.NewEncoder(w).Encode(v)
+}
+
+// sanitizeUploadFilename reduces a client filename to a safe base name: charset
+// [A-Za-z0-9._ -], mandatory .fab suffix, non-empty stem, no leading dot, no traversal.
+func sanitizeUploadFilename(name string) (string, string) {
+	base := filepath.Base(strings.TrimSpace(name))
+	if !strings.HasSuffix(base, ".fab") {
+		return "", "Csak .fab fájl tölthető fel."
+	}
+	stem := strings.TrimSuffix(base, ".fab")
+	if stem == "" || strings.HasPrefix(stem, ".") {
+		return "", "Érvénytelen fájlnév."
+	}
+	for _, r := range base {
+		switch {
+		case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9':
+		case r == '.' || r == '_' || r == ' ' || r == '-':
+		default:
+			return "", "Érvénytelen fájlnév."
+		}
+	}
+	return base, ""
+}
+
+// uploadGB formats bytes as a one-decimal GB string for the Hungarian space error.
+func uploadGB(b int64) string {
+	return fmt.Sprintf("%.1f", float64(b)/float64(1<<30))
+}
+
+// lowestFreeUploadName returns name if unused, else the first "<stem> (N).fab" (N≥1) that does
+// not exist — computed from the ORIGINAL requested name every time, so a re-run finds its own
+// prior "(N)" and never produces "(1)(1)".
+func lowestFreeUploadName(dir, name string) string {
+	if _, err := os.Stat(filepath.Join(dir, name)); os.IsNotExist(err) {
+		return name
+	}
+	ext := filepath.Ext(name)
+	stem := strings.TrimSuffix(name, ext)
+	for n := 1; ; n++ {
+		cand := fmt.Sprintf("%s (%d)%s", stem, n, ext)
+		if _, err := os.Stat(filepath.Join(dir, cand)); os.IsNotExist(err) {
+			return cand
+		}
+	}
+}
+
+// randomUploadToken returns a hex token from crypto/rand.
+func randomUploadToken(nBytes int) (string, error) {
+	b := make([]byte, nBytes)
+	if _, err := rand.Read(b); err != nil {
+		return "", err
+	}
+	return hex.EncodeToString(b), nil
+}
+
+// apiUploadInit — POST /api/export/upload/init {filename, size_bytes}. Claims the single-flight
+// slot, gates on free space, creates the .part file in the DEFAULT drive's exports dir.
+func (s *Server) apiUploadInit(w http.ResponseWriter, r *http.Request) {
+	r.Body = http.MaxBytesReader(w, r.Body, 1<<20)
+	var req struct {
+		Filename  string `json:"filename"`
+		SizeBytes int64  `json:"size_bytes"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		jsonError(w, "Invalid request body", http.StatusBadRequest)
+		return
+	}
+	name, humanErr := sanitizeUploadFilename(req.Filename)
+	if humanErr != "" {
+		jsonError(w, humanErr, http.StatusBadRequest)
+		return
+	}
+	if req.SizeBytes <= 0 {
+		jsonError(w, "Érvénytelen fájlméret.", http.StatusBadRequest)
+		return
+	}
+
+	drive := s.settings.GetDefaultStoragePath()
+	if drive == "" {
+		jsonError(w, "Nincs alapértelmezett tároló beállítva.", http.StatusConflict)
+		return
+	}
+	exportDir := appexport.ExportDir(drive)
+	if err := os.MkdirAll(exportDir, 0755); err != nil {
+		logx.Warnf(s.logger, "[web] fab upload init: mkdir %s: %v", exportDir, err)
+		jsonError(w, "A tároló nem írható.", http.StatusInternalServerError)
+		return
+	}
+	free := uploadDiskFree(exportDir)
+	if free < req.SizeBytes+uploadFreeSpaceMargin {
+		jsonError(w, fmt.Sprintf(
+			"Nincs elég szabad hely a tárolón (szükséges: %s GB, elérhető: %s GB).",
+			uploadGB(req.SizeBytes), uploadGB(free)), http.StatusConflict)
+		return
+	}
+
+	id, err := randomUploadToken(16)
+	if err != nil {
+		jsonError(w, "Belső hiba.", http.StatusInternalServerError)
+		return
+	}
+	partRand, err := randomUploadToken(8)
+	if err != nil {
+		jsonError(w, "Belső hiba.", http.StatusInternalServerError)
+		return
+	}
+	job := &uploadJob{
+		id:        id,
+		finalName: name,
+		partPath:  filepath.Join(exportDir, name+".part-"+partRand),
+		declared:  req.SizeBytes,
+	}
+
+	s.fabUpload.mu.Lock()
+	if s.fabUpload.cur != nil {
+		s.fabUpload.mu.Unlock()
+		jsonError(w, "Már folyamatban van egy feltöltés.", http.StatusConflict)
+		return
+	}
+	f, err := os.OpenFile(job.partPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0644)
+	if err != nil {
+		s.fabUpload.mu.Unlock()
+		logx.Warnf(s.logger, "[web] fab upload init: create part: %v", err)
+		jsonError(w, "A tároló nem írható.", http.StatusInternalServerError)
+		return
+	}
+	job.file = f
+	job.timer = time.AfterFunc(uploadIdleTimeout, func() { s.expireIdleUpload(job) })
+	s.fabUpload.cur = job
+	s.fabUpload.mu.Unlock()
+
+	logx.Infof(s.logger, "[web] fab upload started: %s (%d bytes declared) → %s",
+		job.finalName, job.declared, exportDir)
+	uploadJSON(w, http.StatusOK, map[string]interface{}{
+		"ok":             true,
+		"upload_id":      job.id,
+		"chunk_bytes":    uploadChunkBytes,
+		"received_bytes": 0,
+	})
+}
+
+// apiUploadChunk — POST /api/export/upload/chunk?id=…&offset=N with a raw octet-stream body.
+// offset MUST equal bytes received so far (strictly sequential); a mismatch answers 409 with the
+// current received_bytes so the client re-syncs one step. Streams via io.Copy — no RAM
+// proportional to the chunk.
+func (s *Server) apiUploadChunk(w http.ResponseWriter, r *http.Request) {
+	id := r.URL.Query().Get("id")
+	offset, perr := strconv.ParseInt(r.URL.Query().Get("offset"), 10, 64)
+
+	s.fabUpload.mu.Lock()
+	job := s.fabUpload.cur
+	if job == nil || job.id != id || id == "" {
+		s.fabUpload.mu.Unlock()
+		jsonError(w, "Ismeretlen feltöltés.", http.StatusNotFound)
+		return
+	}
+	if job.writing {
+		s.fabUpload.mu.Unlock()
+		uploadJSON(w, http.StatusConflict, map[string]interface{}{
+			"ok": false, "error": "chunk already in flight", "received_bytes": job.received,
+		})
+		return
+	}
+	if perr != nil || offset != job.received {
+		received := job.received
+		s.fabUpload.mu.Unlock()
+		uploadJSON(w, http.StatusConflict, map[string]interface{}{
+			"ok": false, "error": "offset mismatch", "received_bytes": received,
+		})
+		return
+	}
+	job.writing = true
+	job.timer.Stop() // no idle-expiry underneath an in-flight body
+	s.fabUpload.mu.Unlock()
+
+	n, err := io.Copy(job.file, http.MaxBytesReader(w, r.Body, uploadMaxChunkBody))
+
+	s.fabUpload.mu.Lock()
+	job.writing = false
+	current := s.fabUpload.cur == job
+	if current {
+		job.received += n
+		job.timer.Reset(uploadIdleTimeout)
+	}
+	received := job.received
+	s.fabUpload.mu.Unlock()
+
+	if !current {
+		// Aborted/expired while the body streamed — the .part is already gone.
+		jsonError(w, "A feltöltés megszakadt.", http.StatusGone)
+		return
+	}
+	if err != nil {
+		var mbe *http.MaxBytesError
+		code := http.StatusInternalServerError
+		if errors.As(err, &mbe) {
+			code = http.StatusRequestEntityTooLarge
+		}
+		logx.Debugf(s.logger, "[web] fab upload chunk failed at %d bytes: %v", received, err)
+		uploadJSON(w, code, map[string]interface{}{
+			"ok": false, "error": "chunk write failed", "received_bytes": received,
+		})
+		return
+	}
+	logx.Debugf(s.logger, "[web] fab upload chunk ok: +%d → %d/%d bytes", n, received, job.declared)
+	uploadJSON(w, http.StatusOK, map[string]interface{}{
+		"ok": true, "received_bytes": received,
+	})
+}
+
+// apiUploadFinalize — POST /api/export/upload/finalize {upload_id}. Exact-size check, fsync,
+// atomic rename to the final name (lowest-free " (N)" on collision). The page then re-runs the
+// existing bundle scan — no second lister.
+func (s *Server) apiUploadFinalize(w http.ResponseWriter, r *http.Request) {
+	var req struct {
+		UploadID string `json:"upload_id"`
+	}
+	r.Body = http.MaxBytesReader(w, r.Body, 1<<20)
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		jsonError(w, "Invalid request body", http.StatusBadRequest)
+		return
+	}
+
+	s.fabUpload.mu.Lock()
+	job := s.fabUpload.cur
+	if job == nil || job.id != req.UploadID || req.UploadID == "" {
+		s.fabUpload.mu.Unlock()
+		jsonError(w, "Ismeretlen feltöltés.", http.StatusNotFound)
+		return
+	}
+	if job.writing {
+		s.fabUpload.mu.Unlock()
+		jsonError(w, "Egy adatcsomag még feltöltés alatt van.", http.StatusConflict)
+		return
+	}
+	job.timer.Stop()
+	s.fabUpload.cur = nil // the slot frees on every finalize outcome
+	s.fabUpload.mu.Unlock()
+
+	if job.received != job.declared {
+		job.file.Close()
+		os.Remove(job.partPath)
+		logx.Warnf(s.logger, "[web] fab upload finalize size mismatch: got %d, declared %d — part deleted",
+			job.received, job.declared)
+		uploadJSON(w, http.StatusUnprocessableEntity, map[string]interface{}{
+			"ok": false, "error": "A feltöltött méret nem egyezik — próbálja újra.",
+		})
+		return
+	}
+	if err := job.file.Sync(); err != nil {
+		job.file.Close()
+		os.Remove(job.partPath)
+		logx.Warnf(s.logger, "[web] fab upload finalize fsync: %v", err)
+		jsonError(w, "A fájl mentése sikertelen.", http.StatusInternalServerError)
+		return
+	}
+	if err := job.file.Close(); err != nil {
+		os.Remove(job.partPath)
+		jsonError(w, "A fájl mentése sikertelen.", http.StatusInternalServerError)
+		return
+	}
+	dir := filepath.Dir(job.partPath)
+	finalName := lowestFreeUploadName(dir, job.finalName)
+	if err := os.Rename(job.partPath, filepath.Join(dir, finalName)); err != nil {
+		os.Remove(job.partPath)
+		logx.Warnf(s.logger, "[web] fab upload finalize rename: %v", err)
+		jsonError(w, "A fájl mentése sikertelen.", http.StatusInternalServerError)
+		return
+	}
+	logx.Infof(s.logger, "[web] fab upload finished: %s (%d bytes)", finalName, job.declared)
+	uploadJSON(w, http.StatusOK, map[string]interface{}{
+		"ok": true, "filename": finalName,
+	})
+}
+
+// apiUploadAbort — POST /api/export/upload/abort {upload_id} → delete the .part, free the slot.
+func (s *Server) apiUploadAbort(w http.ResponseWriter, r *http.Request) {
+	var req struct {
+		UploadID string `json:"upload_id"`
+	}
+	r.Body = http.MaxBytesReader(w, r.Body, 1<<20)
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		jsonError(w, "Invalid request body", http.StatusBadRequest)
+		return
+	}
+
+	s.fabUpload.mu.Lock()
+	job := s.fabUpload.cur
+	if job == nil || job.id != req.UploadID || req.UploadID == "" {
+		s.fabUpload.mu.Unlock()
+		jsonError(w, "Ismeretlen feltöltés.", http.StatusNotFound)
+		return
+	}
+	job.timer.Stop()
+	s.fabUpload.cur = nil
+	s.fabUpload.mu.Unlock()
+
+	job.file.Close()
+	os.Remove(job.partPath)
+	logx.Infof(s.logger, "[web] fab upload aborted by client at %d/%d bytes", job.received, job.declared)
+	uploadJSON(w, http.StatusOK, map[string]interface{}{"ok": true})
+}
+
+// expireIdleUpload fires from the idle timer: an upload with no chunk for uploadIdleTimeout is
+// aborted server-side and its .part deleted. A body streaming right now is never expired (the
+// timer is stopped for the write's duration and re-armed after).
+func (s *Server) expireIdleUpload(job *uploadJob) {
+	s.fabUpload.mu.Lock()
+	if s.fabUpload.cur != job || job.writing {
+		s.fabUpload.mu.Unlock()
+		return
+	}
+	s.fabUpload.cur = nil
+	s.fabUpload.mu.Unlock()
+
+	job.file.Close()
+	os.Remove(job.partPath)
+	logx.Warnf(s.logger, "[web] fab upload idle-expired at %d/%d bytes — part deleted",
+		job.received, job.declared)
+}
+
+// CleanupStaleUploadParts removes stray *.part-* files from every registered drive's exports dir
+// (startup GC — upload state is in-memory, so a restart strands at most one .part per crash).
+func (s *Server) CleanupStaleUploadParts() {
+	removed := 0
+	for _, sp := range s.settings.GetStoragePaths() {
+		dir := appexport.ExportDir(sp.Path)
+		entries, err := os.ReadDir(dir)
+		if err != nil {
+			continue
+		}
+		for _, e := range entries {
+			if e.IsDir() || !strings.Contains(e.Name(), ".part-") {
+				continue
+			}
+			if err := os.Remove(filepath.Join(dir, e.Name())); err == nil {
+				removed++
+			}
+		}
+	}
+	if removed > 0 {
+		logx.Infof(s.logger, "[web] startup GC: removed %d stale upload part file(s)", removed)
+	}
+}