admin
db16371f47
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).
421 lines
14 KiB
Go
421 lines
14 KiB
Go
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)
|
|
}
|
|
}
|