bluebird/internal/app/upload.go

package app

import (
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"io"
	"net"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"syscall"
	"time"
	"unicode"

	"filippo.io/age"
	"golang.org/x/sys/unix"
)

type ManifestFile struct {
	Name string `json:"name"`
	Size int64  `json:"size"`
}
type Manifest struct {
	Files     []ManifestFile `json:"files"`
	Days      int            `json:"days"`
	Downloads int            `json:"downloads"`
	Password  string         `json:"password"`
	Renew     bool           `json:"renew"`
}

func reserveSize(n int64, encrypted bool) int64 {
	if encrypted {
		n += 1024 + ((n/65536)+1)*16
	}
	return ((n+4095)/4096)*4096 + 4096
}
func cleanName(s string) string {
	s = strings.ReplaceAll(s, "\\", "/")
	s = filepath.Base(s)
	s = strings.Map(func(r rune) rune {
		if unicode.IsControl(r) {
			return -1
		}
		return r
	}, s)
	return strings.TrimSpace(s)
}
func objectPath(id string) string { return "objects/" + id[:2] + "/" + id }
func (a *App) capacity() (used, reserved, metadata int64, err error) {
	err = a.db.QueryRow("SELECT COALESCE(sum(stored_bytes),0) FROM objects").Scan(&used)
	if err != nil {
		return
	}
	err = a.db.QueryRow("SELECT COALESCE(sum(reserved),0) FROM files WHERE object_id IS NULL AND reserved>0 AND status!='deleted'").Scan(&reserved)
	if err != nil {
		return
	}
	for _, p := range []string{"db/bluebird.db", "db/bluebird.db-wal", "db/bluebird.db-shm"} {
		s, e := a.root.Stat(p)
		if e == nil {
			metadata += allocated(s)
		} else if !errors.Is(e, os.ErrNotExist) {
			err = e
			return
		}
	}
	metadata += a.extraDisk
	return
}
func allocated(s os.FileInfo) int64 {
	if st, ok := s.Sys().(*syscall.Stat_t); ok {
		return max(st.Blocks*512, s.Size())
	}
	if st, ok := s.Sys().(*unix.Stat_t); ok {
		return max(st.Blocks*512, s.Size())
	}
	return ((s.Size() + 4095) / 4096) * 4096
}
func (a *App) createUpload(w http.ResponseWriter, r *http.Request) {
	if !a.allow(r, "upload", 20) {
		fail(w, 429, "上传请求过于频繁")
		return
	}
	var m Manifest
	if !decode(w, r, &m) {
		return
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	p, e := a.policy()
	if e != nil {
		a.serverError(w, e)
		return
	}
	if !p.AnonymousUpload {
		fail(w, 403, "管理员已关闭上传")
		return
	}
	if m.Days == 0 {
		m.Days = p.DefaultDays
	}
	if m.Downloads == 0 {
		m.Downloads = p.DefaultDownloads
	}
	if m.Days < 1 || m.Days > p.MaxInitialDays || m.Downloads < 1 || m.Downloads > p.MaxDownloads || len(m.Files) < 1 || len(m.Files) > p.MaxFiles {
		fail(w, 400, "文件数量、期限或下载次数超出策略")
		return
	}
	if p.RequirePassword && m.Password == "" {
		fail(w, 400, "当前策略要求取件密码")
		return
	}
	hash := ""
	if m.Password != "" {
		if len(m.Password) < 8 || len(m.Password) > 128 {
			fail(w, 400, "取件密码需为 8–128 字节")
			return
		}
		select {
		case a.passwordSlots <- struct{}{}:
		default:
			fail(w, 429, "密码验证繁忙")
			return
		}
		hash, e = hashPassword(m.Password)
		<-a.passwordSlots
		if e != nil {
			fail(w, 400, e.Error())
			return
		}
	}
	var total, reservation int64
	for i, f := range m.Files {
		m.Files[i].Name = cleanName(f.Name)
		if m.Files[i].Name == "" || m.Files[i].Name == "." || m.Files[i].Name == ".." || len(m.Files[i].Name) > 240 || f.Size < 0 || f.Size > p.MaxFileBytes || total > p.MaxBatchBytes-f.Size {
			fail(w, 400, "文件名、文件大小或批次总大小无效")
			return
		}
		total += f.Size
		reservation += reserveSize(f.Size, a.cfg.Encryption)
	}
	if total > p.MaxBatchBytes {
		fail(w, 400, "批次总大小超出限制")
		return
	}
	var count int
	if e = a.db.QueryRow("SELECT count(*) FROM files").Scan(&count); e != nil {
		a.serverError(w, e)
		return
	}
	if count+len(m.Files) > a.cfg.MaxRecords {
		fail(w, 507, "文件记录已满")
		return
	}
	used, reserved, meta, e := a.capacity()
	if e != nil {
		a.serverError(w, e)
		return
	}
	var disk unix.Statfs_t
	if e = unix.Statfs(a.cfg.DataDir, &disk); e != nil {
		a.serverError(w, e)
		return
	}
	free := int64(disk.Bavail) * int64(disk.Bsize)
	limits, e := a.runtimeSettings()
	if e != nil {
		a.serverError(w, e)
		return
	}
	if used+reserved+reservation > limits.FileBudget || used+reserved+reservation+meta > limits.DataBudget || free-reserved-reservation < limits.MinFreeBytes || (disk.Files > 0 && disk.Ffree < uint64(len(m.Files)+32)) {
		fail(w, 507, "可用存储不足，暂时无法接收文件")
		return
	}
	visitor, e := a.visitor(w, r)
	if e != nil {
		fail(w, 429, "匿名会话暂不可用")
		return
	}
	ip := a.remoteIP(r)
	day := a.now().UTC().Format("2006-01-02")
	subjects := []string{"ip:" + a.keyed(ip), "visitor:" + visitor}
	tx, e := a.db.Begin()
	if e != nil {
		a.serverError(w, e)
		return
	}
	defer tx.Rollback()
	for i, subject := range subjects {
		if _, e = tx.Exec("INSERT OR IGNORE INTO usage VALUES(?,?,0,0)", subject, day); e != nil {
			a.serverError(w, e)
			return
		}
		var b int64
		var n int
		if e = tx.QueryRow("SELECT bytes,uploads FROM usage WHERE subject=? AND day=?", subject, day).Scan(&b, &n); e != nil {
			a.serverError(w, e)
			return
		}
		limit := p.IPDailyUploads
		if i == 1 {
			limit = p.SessionDailyUploads
		}
		if n >= limit || (i == 0 && b > p.IPDailyBytes-total) {
			fail(w, 429, "今日上传次数或流量额度已用完")
			return
		}
		if _, e = tx.Exec("UPDATE usage SET bytes=bytes+?,uploads=uploads+1 WHERE subject=? AND day=?", total, subject, day); e != nil {
			a.serverError(w, e)
			return
		}
	}
	id, manage := randomID(), randomToken()
	now := a.now().Unix()
	renewal := 0
	if m.Renew {
		renewal = p.RenewalDays
	}
	_, e = tx.Exec("INSERT INTO parcels(id,visitor,ip,state,created,password_hash,manage_hash,days,downloads,renewal_days,renewal_mode,lifetime_days,upload_expires,usage_day) VALUES(?,?,?,'uploading',?,?,?,?,?,?,?,?,?,?)", id, visitor, ip, now, hash, tokenHash(manage), m.Days, m.Downloads, renewal, p.RenewalMode, p.MaxLifetimeDays, now+3600, day)
	if e != nil {
		a.serverError(w, e)
		return
	}
	out := []map[string]any{}
	for _, f := range m.Files {
		fid := randomID()
		if _, e = tx.Exec("INSERT INTO files(id,parcel,name,size,status,reserved,remaining,charged) VALUES(?,?,?,?,'pending',?,?,?)", fid, id, f.Name, f.Size, reserveSize(f.Size, a.cfg.Encryption), m.Downloads, f.Size); e != nil {
			a.serverError(w, e)
			return
		}
		out = append(out, map[string]any{"id": fid, "name": f.Name, "size": f.Size})
	}
	if e = tx.Commit(); e != nil {
		a.serverError(w, e)
		return
	}
	a.audit("upload.create", id, ip)
	respond(w, 201, map[string]any{"id": id, "manage_token": manage, "files": out})
}
func (a *App) owner(r *http.Request, id string) bool {
	var owner, manage string
	if e := a.db.QueryRow("SELECT visitor,manage_hash FROM parcels WHERE id=?", id).Scan(&owner, &manage); e != nil {
		return false
	}
	token := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
	if token != "" && tokenHash(token) == manage {
		return true
	}
	v := a.currentVisitor(r)
	return v != "" && v == owner
}
func (a *App) uploadFile(w http.ResponseWriter, r *http.Request) {
	id, fid := r.PathValue("parcel"), r.PathValue("file")
	ip := a.remoteIP(r)
	if !a.allow(r, "file-transfer", 120) {
		fail(w, 429, "文件传输请求过于频繁")
		return
	}
	if !a.acquire("upload", ip) {
		fail(w, 429, "上传并发已满，请稍后重试")
		return
	}
	defer a.release("upload", ip)
	a.deadlines(w, r)
	a.mu.Lock()
	if !a.owner(r, id) {
		a.mu.Unlock()
		fail(w, 404, "上传会话不存在")
		return
	}
	var state, status string
	var expected, reserved, expiry int64
	e := a.db.QueryRow("SELECT p.state,f.status,f.size,f.reserved,p.upload_expires FROM files f JOIN parcels p ON p.id=f.parcel WHERE f.id=? AND p.id=?", fid, id).Scan(&state, &status, &expected, &reserved, &expiry)
	if e != nil || state != "uploading" || expiry <= a.now().Unix() {
		a.mu.Unlock()
		fail(w, 409, "上传会话已失效")
		return
	}
	if status == "ready" {
		a.mu.Unlock()
		respond(w, 200, map[string]bool{"ready": true})
		return
	}
	if a.activeFiles[fid] {
		a.mu.Unlock()
		fail(w, 409, "该文件正在上传")
		return
	}
	var attempts int
	var sessionIP, visitor, day string
	if e = a.db.QueryRow("SELECT f.attempts,p.ip,p.visitor,p.usage_day FROM files f JOIN parcels p ON p.id=f.parcel WHERE f.id=?", fid).Scan(&attempts, &sessionIP, &visitor, &day); e != nil {
		a.mu.Unlock()
		a.serverError(w, e)
		return
	}
	policy, pe := a.policy()
	if pe != nil {
		a.mu.Unlock()
		a.serverError(w, pe)
		return
	}
	if !policy.AnonymousUpload {
		a.mu.Unlock()
		fail(w, 403, "上传已关闭")
		return
	}
	tx, e := a.db.Begin()
	if e != nil {
		a.mu.Unlock()
		a.serverError(w, e)
		return
	}
	if attempts > 0 {
		var charged int64
		e = tx.QueryRow("SELECT bytes FROM usage WHERE subject=? AND day=?", "ip:"+a.keyed(sessionIP), day).Scan(&charged)
		if e == nil && charged > policy.IPDailyBytes-expected {
			tx.Rollback()
			a.mu.Unlock()
			fail(w, 429, "重试上传超过日流量限制")
			return
		}
		if e == nil {
			for _, subject := range []string{"ip:" + a.keyed(sessionIP), "visitor:" + visitor} {
				_, e = tx.Exec("UPDATE usage SET bytes=bytes+? WHERE subject=? AND day=?", expected, subject, day)
				if e != nil {
					break
				}
			}
		}
		if e == nil {
			_, e = tx.Exec("UPDATE files SET charged=charged+? WHERE id=?", expected, fid)
		}
	}
	if e == nil {
		_, e = tx.Exec("UPDATE files SET attempts=attempts+1 WHERE id=?", fid)
	}
	if e == nil {
		e = tx.Commit()
	} else {
		tx.Rollback()
	}
	if e != nil {
		a.mu.Unlock()
		a.serverError(w, e)
		return
	}
	a.activeFiles[fid] = true
	a.mu.Unlock()
	defer func() { a.mu.Lock(); delete(a.activeFiles, fid); a.mu.Unlock() }()
	tmp := "tmp/" + fid
	_ = a.root.Remove(tmp)
	f, e := a.root.OpenFile(tmp, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0600)
	if e != nil {
		a.serverError(w, e)
		return
	}
	defer a.root.Remove(tmp)
	var dst io.Writer = f
	var enc io.WriteCloser
	if a.cfg.Encryption {
		enc, e = age.Encrypt(f, a.identity.Recipient())
		if e != nil {
			f.Close()
			a.serverError(w, e)
			return
		}
		dst = enc
	}
	h := sha256.New()
	r.Body = http.MaxBytesReader(w, r.Body, expected)
	totalDeadline := time.Now().Add(time.Duration(a.cfg.MaxTransferSeconds) * time.Second)
	if sessionEnd := time.Unix(expiry, 0); sessionEnd.Before(totalDeadline) {
		totalDeadline = sessionEnd
	}
	src := &idleReader{reader: r.Body, controller: http.NewResponseController(w), total: totalDeadline, idle: time.Duration(a.cfg.TransferIdleSeconds) * time.Second}
	received, e := io.CopyBuffer(io.MultiWriter(dst, h), src, make([]byte, 128<<10))
	if e == nil && received != expected {
		e = errors.New("实际大小与声明不一致")
	}
	if e == nil && enc != nil {
		e = enc.Close()
	}
	if e == nil {
		e = f.Sync()
	}
	st, se := f.Stat()
	ce := f.Close()
	if e == nil {
		e = ce
	}
	if e == nil {
		e = se
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	if _, ue := a.db.Exec("UPDATE files SET received=received+? WHERE id=?", received, fid); ue != nil {
		a.serverError(w, ue)
		return
	}
	if e != nil {
		fail(w, 400, "文件接收失败或大小与声明不一致，请重试")
		return
	}
	stored := allocated(st)
	if stored > reserved {
		fail(w, 507, "文件实际落盘开销超出预留")
		return
	}
	if e = a.db.QueryRow("SELECT state FROM parcels WHERE id=?", id).Scan(&state); e != nil || state != "uploading" || expiry <= a.now().Unix() {
		fail(w, 409, "上传已取消")
		return
	}
	keyID := "plain"
	if a.cfg.Encryption {
		keyID = tokenHash(a.identity.Recipient().String())
	}
	digest := a.keyed(hex.EncodeToString(h.Sum(nil)) + "/" + keyID)
	if !a.cfg.Deduplication {
		digest += "/" + randomID()
	}
	var oid string
	e = a.db.QueryRow("SELECT id FROM objects WHERE digest=? AND size=?", digest, expected).Scan(&oid)
	if e != nil && !errors.Is(e, sql.ErrNoRows) {
		a.serverError(w, e)
		return
	}
	newObject := oid == ""
	committed := false
	defer func() {
		if newObject && oid != "" && !committed {
			if de := a.root.Remove(objectPath(oid)); de != nil && !errors.Is(de, os.ErrNotExist) {
				a.extraDisk += stored
				a.log.Error("orphan removal failed", "error", de)
			}
		}
	}()
	if newObject {
		oid = randomID()
		path := objectPath(oid)
		if e = a.root.MkdirAll(filepath.Dir(path), 0700); e != nil {
			a.serverError(w, e)
			return
		}
		if e = a.root.Rename(tmp, path); e != nil {
			a.serverError(w, e)
			return
		}
		if e = syncDir(a.root, filepath.Dir(path)); e != nil {
			a.serverError(w, e)
			return
		}
		if e = syncDir(a.root, "tmp"); e != nil {
			a.serverError(w, e)
			return
		}
	} else {
		if _, e = a.root.Stat(objectPath(oid)); e != nil {
			a.serverError(w, e)
			return
		}
		if e = a.root.Remove(tmp); e != nil {
			a.serverError(w, e)
			return
		}
	}
	tx, e = a.db.Begin()
	if e != nil {
		a.serverError(w, e)
		return
	}
	defer tx.Rollback()
	if newObject {
		_, e = tx.Exec("INSERT INTO objects VALUES(?,?,?,?,?,?,?)", oid, digest, expected, stored, a.cfg.Encryption, keyID, a.now().Unix())
		if e != nil {
			_ = a.root.Remove(objectPath(oid))
			a.serverError(w, e)
			return
		}
	}
	var attached sql.Result
	attached, e = tx.Exec("UPDATE files SET object_id=?,status='ready',reserved=0 WHERE id=? AND status='pending'", oid, fid)
	if e == nil {
		n, _ := attached.RowsAffected()
		if n != 1 {
			e = errors.New("文件已被管理员删除")
		}
	}
	if e == nil {
		e = tx.Commit()
	}
	if e != nil {
		if newObject {
			_ = a.root.Remove(objectPath(oid))
		}
		a.serverError(w, e)
		return
	}
	committed = true
	respond(w, 200, map[string]bool{"ready": true})
}
func (a *App) publish(w http.ResponseWriter, r *http.Request) {
	a.mu.Lock()
	defer a.mu.Unlock()
	id := r.PathValue("parcel")
	if !a.owner(r, id) {
		fail(w, 404, "包裹不存在")
		return
	}
	var state string
	var days, lifetime int
	var uploadExpiry int64
	e := a.db.QueryRow("SELECT state,days,lifetime_days,upload_expires FROM parcels WHERE id=?", id).Scan(&state, &days, &lifetime, &uploadExpiry)
	if e != nil {
		a.serverError(w, e)
		return
	}
	if state == "ready" {
		respond(w, 200, map[string]any{"published": true, "message": "已发布；如丢失取件码，请重新生成"})
		return
	}
	policy, pe := a.policy()
	if pe != nil {
		a.serverError(w, pe)
		return
	}
	if !policy.AnonymousUpload {
		fail(w, 403, "上传已关闭")
		return
	}
	if state != "uploading" || uploadExpiry <= a.now().Unix() {
		fail(w, 409, "上传会话已失效")
		return
	}
	var pending int
	e = a.db.QueryRow("SELECT count(*) FROM files WHERE parcel=? AND status!='ready'", id).Scan(&pending)
	if e != nil {
		a.serverError(w, e)
		return
	}
	if pending > 0 {
		fail(w, 409, "尚有文件未上传完成")
		return
	}
	now := a.now().Unix()
	expires := now + int64(days)*86400
	hard := now + int64(lifetime)*86400
	code := pickupCode()
	tx, e := a.db.Begin()
	if e != nil {
		a.serverError(w, e)
		return
	}
	defer tx.Rollback()
	_, e = tx.Exec("UPDATE parcels SET state='ready',created=?,expires=?,hard_expires=?,code_hash=? WHERE id=?", now, expires, hard, a.keyed(code), id)
	if e == nil {
		_, e = tx.Exec("UPDATE files SET expires=?,hard_expires=? WHERE parcel=?", expires, hard, id)
	}
	if e == nil {
		e = tx.Commit()
	}
	if e != nil {
		a.serverError(w, e)
		return
	}
	if e = a.settleUsage(id); e != nil {
		a.serverError(w, e)
		return
	}
	a.audit("parcel.publish", id, a.remoteIP(r))
	respond(w, 200, a.shareResult(r, code, expires))
}
func (a *App) shareResult(r *http.Request, code string, expires int64) map[string]any {
	links := []map[string]string{}
	for _, base := range a.cfg.accessOrigins() {
		label := "公网HTTP"
		if strings.HasPrefix(base, "https://") {
			label = "公网HTTPS"
		}
		u, _ := url.Parse(base)
		ip := net.ParseIP(u.Hostname())
		if ip != nil && (ip.IsPrivate() || ip.IsLoopback()) {
			label = "内网入口"
		}
		links = append(links, map[string]string{"label": label + " · " + base, "url": base + "/#/pickup/" + code})
	}
	return map[string]any{"code": code, "url": a.accessFor(r).base + "/#/pickup/" + code, "urls": links, "expires": expires}
}
func (a *App) regenerate(w http.ResponseWriter, r *http.Request) {
	a.mu.Lock()
	defer a.mu.Unlock()
	id := r.PathValue("parcel")
	if !a.owner(r, id) {
		fail(w, 404, "包裹不存在")
		return
	}
	var expiry int64
	if e := a.db.QueryRow("SELECT expires FROM parcels WHERE id=? AND state='ready' AND EXISTS(SELECT 1 FROM files WHERE parcel=parcels.id AND status='ready' AND remaining>0 AND expires>?)", id, a.now().Unix()).Scan(&expiry); e != nil || expiry <= a.now().Unix() {
		fail(w, 409, "包裹不可取件")
		return
	}
	code := pickupCode()
	if _, e := a.db.Exec("UPDATE parcels SET code_hash=?,version=version+1 WHERE id=?", a.keyed(code), id); e != nil {
		a.serverError(w, e)
		return
	}
	respond(w, 200, a.shareResult(r, code, expiry))
}
func (a *App) revoke(w http.ResponseWriter, r *http.Request) {
	a.mu.Lock()
	defer a.mu.Unlock()
	id := r.PathValue("parcel")
	if !a.owner(r, id) {
		fail(w, 404, "包裹不存在")
		return
	}
	if _, e := a.db.Exec("UPDATE parcels SET state='revoked',version=version+1 WHERE id=?", id); e != nil {
		a.serverError(w, e)
		return
	}
	a.audit("parcel.revoke", id, a.remoteIP(r))
	respond(w, 200, map[string]bool{"revoked": true})
}
func (a *App) uploadStatus(w http.ResponseWriter, r *http.Request) {
	a.mu.Lock()
	defer a.mu.Unlock()
	id := r.PathValue("parcel")
	if !a.owner(r, id) {
		fail(w, 404, "包裹不存在")
		return
	}
	var state string
	var created, expires int64
	if e := a.db.QueryRow("SELECT state,created,expires FROM parcels WHERE id=?", id).Scan(&state, &created, &expires); e != nil {
		a.serverError(w, e)
		return
	}
	files, e := a.fileList(id)
	if e != nil {
		a.serverError(w, e)
		return
	}
	respond(w, 200, map[string]any{"id": id, "state": state, "created": created, "expires": expires, "files": files})
}
func (a *App) mine(w http.ResponseWriter, r *http.Request) {
	a.mu.Lock()
	defer a.mu.Unlock()
	v := a.currentVisitor(r)
	if v == "" {
		respond(w, 200, []any{})
		return
	}
	rows, e := a.db.Query("SELECT p.id,p.state,p.created,p.expires,(SELECT count(*) FROM files WHERE parcel=p.id),(SELECT COALESCE(sum(size),0) FROM files WHERE parcel=p.id) FROM parcels p WHERE visitor=? ORDER BY created DESC LIMIT 100", v)
	if e != nil {
		a.serverError(w, e)
		return
	}
	defer rows.Close()
	out := []map[string]any{}
	for rows.Next() {
		var id, state string
		var created, expires, size int64
		var n int
		if e = rows.Scan(&id, &state, &created, &expires, &n, &size); e != nil {
			a.serverError(w, e)
			return
		}
		out = append(out, map[string]any{"id": id, "state": state, "created": created, "expires": expires, "count": n, "size": size})
	}
	if e = rows.Err(); e != nil {
		a.serverError(w, e)
		return
	}
	respond(w, 200, out)
}