| 1 | import { getNodeByName, statusCodeForMissingPerm, VfsNode } from './vfs' |
| 2 | import Koa from 'koa' |
| 3 | import { |
| 4 | HTTP_CONFLICT, HTTP_FOOL, HTTP_INSUFFICIENT_STORAGE, HTTP_RANGE_NOT_SATISFIABLE, HTTP_NO_CONTENT, HTTP_SERVER_ERROR, |
| 5 | HTTP_PRECONDITION_FAILED, HTTP_LENGTH_REQUIRED, MTIME_CHECK, |
| 6 | } from './const' |
| 7 | import { basename, dirname, extname, join } from 'path' |
| 8 | import fs from 'fs' |
| 9 | import { |
| 10 | isValidFileName, loadFileAttr, pendingPromise, storeFileAttr, try_, createStreamLimiter, pathEncode, |
| 11 | enforceFinal, Timeout, waitFor, |
| 12 | } from './misc' |
| 13 | import { defineConfig } from './config' |
| 14 | import { getDiskSpaceSync } from './util-os' |
| 15 | import { disconnect, updateConnection, updateConnectionForCtx } from './connections' |
| 16 | import { roundSpeed } from './throttler' |
| 17 | import { getCurrentUsername } from './auth' |
| 18 | import { setCommentFor } from './comments' |
| 19 | import _ from 'lodash' |
| 20 | import events from './events' |
| 21 | import { rm, rename, utimes } from 'fs/promises' |
| 22 | import { expiringCache } from './expiringCache' |
| 23 | import { onProcessExit } from './first' |
| 24 | |
| 25 | export const deleteUnfinishedUploadsAfter = defineConfig<undefined|number>('delete_unfinished_uploads_after', 86_400) |
| 26 | export const minAvailableMb = defineConfig('min_available_mb', 100) |
| 27 | export const dontOverwriteUploading = defineConfig('dont_overwrite_uploading', true) |
| 28 | |
| 29 | const waitingToBeDeleted: Record<string, { |
| 30 | timeout: Timeout, // pending action |
| 31 | expires: number, // when |
| 32 | mtime?: any |
| 33 | }> = {} |
| 34 | onProcessExit(() => { |
| 35 | if (!Object.keys(waitingToBeDeleted).length) return |
| 36 | console.log("Removing unfinished uploads") |
| 37 | for (const path in waitingToBeDeleted) |
| 38 | try { fs.rmSync(path, { force: true }) } |
| 39 | catch {} |
| 40 | }) |
| 41 | |
| 42 | const ATTR_UPLOADER = 'uploader' |
| 43 | |
| 44 | export function getUploadMeta(path: string) { |
| 45 | return loadFileAttr(path, ATTR_UPLOADER) |
| 46 | } |
| 47 | |
| 48 | function setUploadMeta(path: string, ctx: Koa.Context) { |
| 49 | return storeFileAttr(path, ATTR_UPLOADER, { |
| 50 | username: getCurrentUsername(ctx) || undefined, |
| 51 | ip: ctx.ip, |
| 52 | }) |
| 53 | } |
| 54 | |
| 55 | export function getUploadTempFor(fullPath: string) { |
| 56 | return join(dirname(fullPath), 'hfs$upload-' + basename(fullPath).slice(-200)) |
| 57 | } |
| 58 | |
| 59 | const diskSpaceCache = expiringCache<ReturnType<typeof getDiskSpaceSync>>(3_000) // invalidate shortly |
| 60 | const uploadingFiles = new Map<string, { ctx: Koa.Context, size: number, got: number }>() |
| 61 | // initially sync for formidable; still sync to avoid async races and PUT piping gaps |
| 62 | export function uploadWriter(base: VfsNode, baseUri: string, filename: string, ctx: Koa.Context) { |
| 63 | if (!filename || !isValidFileName(filename) || !filename) |
| 64 | return fail(HTTP_FOOL) |
| 65 | if (statusCodeForMissingPerm(base, 'can_upload', ctx)) |
| 66 | return fail() |
| 67 | const fullPath = join(base.source!, filename) |
| 68 | const already = uploadingFiles.get(fullPath) // this can be checked so early because this function is sync |
| 69 | if (already) // if it's the same client, we tell to retry later |
| 70 | return fail(HTTP_CONFLICT, ctx.query.id && ctx.query.id === already.ctx.query.id ? 'retry' : 'already uploading') |
| 71 | const dir = dirname(fullPath) |
| 72 | // enforce minAvailableMb |
| 73 | const min = minAvailableMb.get() * (1 << 20) |
| 74 | const {simulate} = ctx.query // `simulate` is used to get the same error but with an empty body, so that the request is processed quickly |
| 75 | const contentLength = Number(simulate ?? ctx.headers["content-length"] |
| 76 | ?? ctx.headers['x-expected-entity-length']) // some webdav clients send this; it's not equivalent to content-length, but can get the job done in some cases |
| 77 | const isPartial = ctx.query.partial !== undefined // while the presence of "partial" conveys that the upload is split... |
| 78 | const stillToWrite = Math.max(contentLength, Number(ctx.query.partial) || 0) // ...the number is used to tell how much space we need (fullSize - offset) |
| 79 | if (isNaN(stillToWrite)) { |
| 80 | if (min) |
| 81 | return fail(HTTP_LENGTH_REQUIRED) |
| 82 | } |
| 83 | else |
| 84 | try { |
| 85 | // refer to the source of the closest node that actually belongs to the vfs, so that cache is more effective |
| 86 | let closestVfsNode = base // if base=root, there's no parent and no original |
| 87 | while (closestVfsNode?.parent && !closestVfsNode.original) |
| 88 | closestVfsNode = closestVfsNode.parent! // if it's not original, it surely has a parent |
| 89 | const dirToCheck = closestVfsNode!.source! |
| 90 | const res = diskSpaceCache.try(dirToCheck, () => getDiskSpaceSync(dirToCheck)) |
| 91 | if (!res) throw 'miss' |
| 92 | const { free } = res |
| 93 | if (typeof free !== 'number' || isNaN(free)) |
| 94 | throw JSON.stringify(res) |
| 95 | const reservedSpace = _.sumBy(Array.from(uploadingFiles.values()), x => x.size - x.got) |
| 96 | if (stillToWrite > free - (min || 0) - reservedSpace) |
| 97 | return fail(HTTP_INSUFFICIENT_STORAGE) |
| 98 | } |
| 99 | catch(e: any) { // warn, but let it through |
| 100 | console.warn("Can't check disk size:", e.message || String(e)) |
| 101 | } |
| 102 | // optionally 'skip' |
| 103 | if (ctx.query.existing === 'skip' && fs.existsSync(fullPath)) |
| 104 | return fail(HTTP_CONFLICT, 'exists') |
| 105 | let overwriteRequestedButForbidden = false |
| 106 | const mtime = Number(ctx.query.mtime) || 0 |
| 107 | try { |
| 108 | // if upload creates a folder, then add meta to it too |
| 109 | if (!dir.endsWith(':\\') && fs.mkdirSync(dir, { recursive: true })) |
| 110 | setUploadMeta(dir, ctx) |
| 111 | // use temporary name while uploading |
| 112 | const tempName = getUploadTempFor(fullPath) |
| 113 | // try to catch errors early (sync): this is avoiding on chrome139 when uploading a big file (1GB) to get miss the error code and have to make a `simulate` |
| 114 | try { fs.accessSync(tempName, fs.constants.W_OK) } |
| 115 | catch { |
| 116 | fs.closeSync(fs.openSync(tempName, 'w')) |
| 117 | fs.unlinkSync(tempName) |
| 118 | } |
| 119 | const stats = try_(() => fs.statSync(tempName)) |
| 120 | const resumableSize = stats?.size || 0 |
| 121 | // checks for resume feature |
| 122 | const par = String(ctx.query.resume || '') |
| 123 | const resume = parseInt(par) || 0 |
| 124 | const strictResume = par.at(-1) === '!' |
| 125 | if (resume > resumableSize || resume < 0) |
| 126 | return fail(HTTP_RANGE_NOT_SATISFIABLE) |
| 127 | const resumeInfo = resumableSize && waitingToBeDeleted[tempName] |
| 128 | if (strictResume) // frontend asked to be notified about resumable uploads |
| 129 | if (resumableSize > resume && (!resumeInfo || resumeInfo.mtime === mtime)) { |
| 130 | ctx.set('x-size', String(resumableSize)) |
| 131 | if (!resumeInfo) // if unavailable, the client can request hashing |
| 132 | ctx.set(MTIME_CHECK, 'not-available') |
| 133 | return fail(HTTP_PRECONDITION_FAILED) |
| 134 | } |
| 135 | // append if resuming |
| 136 | if (!resume && stats) |
| 137 | fs.unlinkSync(tempName) |
| 138 | const writeStream = createStreamLimiter(isNaN(contentLength) ? Infinity : contentLength) |
| 139 | const fullSize = stillToWrite + resume |
| 140 | // allow plugins to mess with the write-stream, because the read-stream can be complicated in case of multipart |
| 141 | const obj = { ctx, writeStream, fullPath, tempName, resume, fullSize, uri: '' } |
| 142 | const resEvent = events.emit('uploadStart', obj) |
| 143 | ctx.state.uploadDestinationPath = tempName |
| 144 | if (resEvent?.isDefaultPrevented()) return |
| 145 | |
| 146 | const fileStream = fs.createWriteStream(tempName, resume ? { flags: 'r+', start: resume } : undefined) |
| 147 | writeStream.on('error', e => { |
| 148 | releaseFile() |
| 149 | console.debug(e) |
| 150 | }) |
| 151 | writeStream.pipe(fileStream) |
| 152 | Object.assign(obj, { fileStream }) |
| 153 | trackProgress() |
| 154 | cancelDeletion(tempName) |
| 155 | const tracked = { ctx, got: 0, size: stillToWrite } |
| 156 | uploadingFiles.set(fullPath, tracked) |
| 157 | console.debug('Upload started') |
| 158 | // the file stream doesn't have an event for data being written, so we use 'data' of its feeder, which happens before, so we postpone a bit, trying to have a fresher number |
| 159 | writeStream.on('data', () => setTimeout(() => tracked.got = bytesGot())) |
| 160 | |
| 161 | const lockMiddleware = pendingPromise<string>() // expose outside, to let know when all operations stopped |
| 162 | let errored: any |
| 163 | fileStream.on('error', (e: any) => { |
| 164 | console.warn('File error while uploading', filename, ':', e.message) |
| 165 | errored = e |
| 166 | fail(HTTP_SERVER_ERROR, e.code) // don't send e.message as it may contain a disk paths we don't want to leak |
| 167 | }) |
| 168 | writeStream.once('close', async () => { |
| 169 | try { |
| 170 | ctx.state.uploadSize = bytesGot() // in case content-length is not specified |
| 171 | await new Promise(res => fileStream.close(res)) // this only seems necessary on Windows |
| 172 | if (errored) |
| 173 | return |
| 174 | if (simulate) |
| 175 | return rm(tempName).catch(() => {}) |
| 176 | if (ctx.isAborted()) { // in the very unlikely case the connection is interrupted between last-byte and here, we still consider it unfinished, as the client had no way to know, and will resume, but it would get an error if we finish the process |
| 177 | const sec = deleteUnfinishedUploadsAfter.get() |
| 178 | return _.isNumber(sec) && delayedDelete(tempName, sec) |
| 179 | } |
| 180 | if (isPartial) // we are supposed to leave the unfinished upload as it is, with its temp name |
| 181 | return ctx.status = HTTP_NO_CONTENT // lockMiddleware contains an empty string, so we must take care of the status |
| 182 | let dest = fullPath // final destination, considering numbering if necessary |
| 183 | if (dontOverwriteUploading.get() && fs.existsSync(dest) && !await overwriteAnyway()) { |
| 184 | if (overwriteRequestedButForbidden) { |
| 185 | await rm(tempName).catch(e => console.warn(String(e))) |
| 186 | releaseFile() |
| 187 | return fail() // status code set by overwriteAnyway |
| 188 | } |
| 189 | const ext = extname(dest) |
| 190 | const base = dest.slice(0, -ext.length || Infinity) |
| 191 | let i = 1 |
| 192 | do dest = `${base} (${i++})${ext}` |
| 193 | while (fs.existsSync(dest)) |
| 194 | } |
| 195 | try { |
| 196 | const done = await waitFor( // an antivirus may lock the temp file to scan it |
| 197 | () => rename(tempName, dest).then(() => true, e => e?.code !== 'EBUSY' && Promise.reject(e)), |
| 198 | { timeout: 10_000 }) |
| 199 | if (!done) |
| 200 | throw 'EBUSY' |
| 201 | if (mtime) // so we use it to touch the file |
| 202 | await utimes(dest, Date.now() / 1000, mtime / 1000) |
| 203 | cancelDeletion(tempName) // not necessary, as deletion's failure is silent, but still |
| 204 | obj.fullPath = ctx.state.uploadDestinationPath = dest |
| 205 | void setUploadMeta(dest, ctx) |
| 206 | if (ctx.query.comment) |
| 207 | void setCommentFor(dest, String(ctx.query.comment)) |
| 208 | obj.uri = enforceFinal('/', baseUri) + pathEncode(basename(dest)) |
| 209 | events.emit('uploadFinished', obj) |
| 210 | console.debug("Upload finished", dest) |
| 211 | if (resEvent) for (const cb of resEvent) |
| 212 | if (_.isFunction(cb)) |
| 213 | cb(obj) |
| 214 | } |
| 215 | catch (err: any) { |
| 216 | void setUploadMeta(tempName, ctx) |
| 217 | console.error("Couldn't rename temp to", dest, String(err)) |
| 218 | } |
| 219 | } |
| 220 | finally { |
| 221 | releaseFile() |
| 222 | lockMiddleware.resolve(obj.uri) |
| 223 | } |
| 224 | }) |
| 225 | return Object.assign(obj.writeStream, { lockMiddleware }) |
| 226 | |
| 227 | function trackProgress() { |
| 228 | let lastGot = 0 |
| 229 | let lastGotTime = 0 |
| 230 | Object.assign(ctx.state, { opTotal: fullSize, opOffset: resume / fullSize, opProgress: 0 }) |
| 231 | const conn = updateConnectionForCtx(ctx) |
| 232 | if (!conn) return |
| 233 | if (writeStream.closed || writeStream.destroyed || writeStream.writableFinished) return |
| 234 | // tracking |
| 235 | const h = setInterval(() => { |
| 236 | const now = Date.now() |
| 237 | const got = bytesGot() |
| 238 | const inSpeedKb = roundSpeed((got - lastGot) / (now - lastGotTime)) |
| 239 | lastGot = got |
| 240 | lastGotTime = now |
| 241 | updateConnection(conn, { inSpeedKb, got }, { opProgress: (resume + got) / fullSize }) |
| 242 | }, 1000) |
| 243 | const stopTracking = () => clearInterval(h) |
| 244 | writeStream.once('close', stopTracking) |
| 245 | writeStream.once('error', stopTracking) |
| 246 | } |
| 247 | |
| 248 | function bytesGot() { |
| 249 | return fileStream.bytesWritten + fileStream.writableLength |
| 250 | } |
| 251 | } |
| 252 | catch (e: any) { |
| 253 | releaseFile() |
| 254 | return fail(HTTP_SERVER_ERROR, e.code || e.message || String(e)) |
| 255 | } |
| 256 | |
| 257 | async function overwriteAnyway() { |
| 258 | if (ctx.query.existing !== 'overwrite') return false |
| 259 | const n = await getNodeByName(filename, base) |
| 260 | if (n && !statusCodeForMissingPerm(n, 'can_delete', ctx)) return true |
| 261 | overwriteRequestedButForbidden = true |
| 262 | return false |
| 263 | } |
| 264 | |
| 265 | function delayedDelete(path: string, secs: number) { |
| 266 | clearTimeout(waitingToBeDeleted[path]?.timeout) |
| 267 | return waitingToBeDeleted[path] = { |
| 268 | mtime, |
| 269 | expires: Date.now() + secs * 1000, |
| 270 | timeout: setTimeout(() => { |
| 271 | delete waitingToBeDeleted[path] |
| 272 | rm(path).catch(() => {}) |
| 273 | }, secs * 1000) |
| 274 | } |
| 275 | } |
| 276 | |
| 277 | function cancelDeletion(path: string) { |
| 278 | clearTimeout(waitingToBeDeleted[path]?.timeout) |
| 279 | delete waitingToBeDeleted[path] |
| 280 | } |
| 281 | |
| 282 | function releaseFile() { |
| 283 | uploadingFiles.delete(fullPath) |
| 284 | } |
| 285 | |
| 286 | function fail(status=ctx.status, msg?: string) { |
| 287 | console.debug('Upload failed', status, msg||'') |
| 288 | ctx.status = status |
| 289 | if (msg) |
| 290 | ctx.body = msg |
| 291 | if (status >= 400 // with other codes Chrome will report ERR_CONNECTION_RESET |
| 292 | && !ctx.get('x-hfs-wait') // you can disable the following behavior |
| 293 | && !ctx.req.complete) // if request body is already complete, forcing a disconnect can interfere with follow-up requests on reused sockets. |
| 294 | setTimeout(() => disconnect(ctx), 200) // don't wait, if the upload is still in progress |
| 295 | } |
| 296 | } |
| 297 | |
| 298 | declare module "koa" { |
| 299 | interface DefaultState { |
| 300 | uploadDestinationPath?: string |
| 301 | uploadSize?: number |
| 302 | } |
| 303 | } |