// pithos-hosted: s3d on Silent Mode with each person's own Sia account. // // One HTTP listener (behind nginx/TLS) serves four kinds of traffic: // // /hosted/v1/... sign-in, sign-up gate, access rules, moving a data // folder in and out (Bearer session token) // /api/... the owner's own Pithos API, forwarded to their // per-account control server (Bearer session token) // /p///.. public files, /l/ share links (no auth) // anything else S3, path-style. Owner keys pass straight through to // their s3d (which checks the signature itself); keys // minted for other people are verified and limited here, // then re-signed with the owner's key. import http from 'node:http'; import fs from 'node:fs'; import path from 'node:path'; import crypto from 'node:crypto'; import { pipeline } from 'node:stream/promises'; import { readConfig } from '../core/config.js'; import { objectPath } from '../core/s3.js'; import { createAuth, accountId, challengeText, verifyEd25519 } from './auth.js'; import { createGate, bnsMessage, depositMemo } from './gate.js'; import { createSupervisor, httpError } from './supervisor.js'; import { parseAuth, verify, SigError } from './sigv4.js'; import { validateRule, classify, decideGrant, publicRuleFor, live, forOwner } from './rules.js'; import { readJson, writeJson } from './store.js'; // Owner API calls the gateway forwards. Everything that would act on the // server itself (binary, s3d.yml, Sia login, account switching) is the // gateway's business, and object routes are refused so nothing can bypass // the client-side encryption. const FORWARD = [ ['GET', /^\/api\/status$/], ['GET', /^\/api\/logs$/], ['GET', /^\/api\/stats$/], ['POST', /^\/api\/flush$/], ['GET', /^\/api\/autoflush$/], ['POST', /^\/api\/daemon\/(start|restart)$/], ['GET', /^\/api\/users$/], ['POST', /^\/api\/users$/], ['DELETE', /^\/api\/users\/[^/]+$/], ['GET', /^\/api\/keys$/], ['POST', /^\/api\/keys$/], ['DELETE', /^\/api\/keys\/[^/]+$/], ['GET', /^\/api\/account$/], ['PUT', /^\/api\/account\/label$/], ]; const KEY_CHANGES = /^\/api\/(users|keys)/; export async function createHosted(cfg, log = console.log) { fs.mkdirSync(cfg.root, { recursive: true }); const auth = createAuth(cfg.root); const gate = createGate(cfg.root, cfg.gate || {}); const sup = createSupervisor(cfg, log); const grantsFile = path.join(cfg.root, 'grants.json'); const linksFile = path.join(cfg.root, 'links.json'); const grants = readJson(grantsFile, {}); // accessKeyId -> { id, rule } const links = readJson(linksFile, {}); // token -> { id, rule } const rulesFile = (id) => path.join(sup.dir(id), 'rules.json'); const rulesOf = (id) => readJson(rulesFile(id), []); function saveRules(id, rules) { writeJson(rulesFile(id), rules); for (const [k, v] of Object.entries(grants)) if (v.id === id) delete grants[k]; for (const [k, v] of Object.entries(links)) if (v.id === id) delete links[k]; for (const r of rules) { if (r.type === 'grant') grants[r.accessKeyId] = { id, rule: r.id }; if (r.type === 'link') links[r.token] = { id, rule: r.id }; } writeJson(grantsFile, grants); writeJson(linksFile, links); } const ruleById = (id, ruleId) => rulesOf(id).find((r) => r.id === ruleId) || null; // ---- traffic meter ------------------------------------------------------- const month = () => new Date().toISOString().slice(0, 7); const usage = new Map(); // id -> { month, down, up } function usageOf(id) { let u = usage.get(id); if (!u) { u = readJson(path.join(sup.dir(id), 'usage.json'), null) || { month: month(), down: 0, up: 0 }; usage.set(id, u); } if (u.month !== month()) { u.month = month(); u.down = 0; u.up = 0; } return u; } const meterTimer = setInterval(() => { for (const [id, u] of usage) writeJson(path.join(sup.dir(id), 'usage.json'), u); }, 30_000); meterTimer.unref?.(); function checkCap(id) { if (cfg.downloadCapBytes && usageOf(id).down >= cfg.downloadCapBytes) { throw httpError(429, `this account has used its ${fmtBytes(cfg.downloadCapBytes)} of downloads this month`); } } // ---- helpers --------------------------------------------------------------- const clientIp = (req) => (cfg.trustProxy && String(req.headers['x-forwarded-for'] || '').split(',')[0].trim()) || req.socket.remoteAddress || ''; function bearer(req) { const m = /^Bearer\s+(\S+)$/.exec(req.headers.authorization || ''); const id = m && auth.lookup(m[1]); if (!id) throw httpError(401, 'sign in first'); return { id, token: m[1] }; } async function forwardApi(req, res, url, id) { const a = sup.account(id); if (url.pathname === '/api/status' && a.state !== 'active') { return json(res, 200, hostedStatus(a, { daemon: { state: 'stopped' }, registration: { registered: false } })); } const inst = await sup.ensure(id); const release = sup.hold(inst); try { const target = new URL(url.pathname + url.search, inst.pithos.url); const up = await new Promise((resolve, reject) => { const r = http.request(target, { method: req.method, headers: { cookie: inst.cookie, 'x-pithos': '1', 'content-type': req.headers['content-type'] || 'application/json' }, }, resolve); r.on('error', reject); req.pipe(r); }); const text = await readAll(up); if (req.method !== 'GET' && KEY_CHANGES.test(url.pathname)) await sup.indexKeys(inst); let body = text; if (up.statusCode === 200 && url.pathname === '/api/status') body = JSON.stringify(hostedStatus(a, JSON.parse(text))); if (up.statusCode === 200 && url.pathname === '/api/account') { const acc = JSON.parse(text); body = JSON.stringify({ indexerUrl: acc.indexerUrl, registered: acc.registered, label: acc.label, archives: [], hosted: true }); } res.writeHead(up.statusCode, { 'content-type': 'application/json; charset=utf-8', 'cache-control': 'no-store' }); res.end(body); } finally { release(); } } // Server paths stay on the server. function hostedStatus(a, st) { return { ...st, host: 'hosted', config: st.config ? { apiAddress: null, adminAddress: null, file: null, directory: null, https: null, exists: true } : undefined, hosted: { id: a.id, state: a.state, publicUrl: cfg.publicUrl, usage: usageOf(a.id), downloadCapBytes: cfg.downloadCapBytes || null }, }; } // ---- data folder in and out ------------------------------------------------ const incomingDir = (id) => path.join(sup.dir(id), 'incoming'); const incomingManifest = (id) => path.join(sup.dir(id), 'incoming.json'); function safeRel(rel) { const r = String(rel || '').replace(/\\/g, '/'); if (!r || r.startsWith('/') || r.split('/').some((s) => s === '..' || s === '' || s === '.') || /^[A-Za-z]:/.test(r)) throw httpError(400, 'bad file path'); return r; } async function receiveFile(req, id, rel) { const a = sup.account(id); if (a.state !== 'empty') throw httpError(409, 'drives are already hosted here; move them back to your computer first'); const file = path.join(incomingDir(id), safeRel(rel)); fs.mkdirSync(path.dirname(file), { recursive: true }); const hash = crypto.createHash('sha256'); let size = 0; const man = readJson(incomingManifest(id), {}); const total = Object.values(man).reduce((n, f) => n + f.size, 0); const limit = cfg.maxDataBytes || 4 * 1024 ** 3; req.on('data', (c) => { hash.update(c); size += c.length; if (total + size > limit) req.destroy(httpError(413, 'the data folder is larger than this server accepts')); }); await pipeline(req, fs.createWriteStream(file)); man[safeRel(rel)] = { size, sha256: hash.digest('hex') }; writeJson(incomingManifest(id), man); return man[safeRel(rel)]; } async function commit(id, files) { const a = sup.account(id); if (a.state !== 'empty') throw httpError(409, 'drives are already hosted here'); const man = readJson(incomingManifest(id), {}); if (!Array.isArray(files) || !files.length) throw httpError(400, 'no files listed'); for (const f of files) { const got = man[safeRel(f.path)]; if (!got || got.sha256 !== f.sha256 || got.size !== f.size) throw httpError(400, `${f.path} did not arrive intact; send it again`); } if (Object.keys(man).length !== files.length) throw httpError(400, 'the server has files the list does not mention; start the move again'); if (!fs.existsSync(path.join(incomingDir(id), 's3d.db'))) throw httpError(400, 'this is not an s3d data folder (no s3d.db)'); await sup.close(id, 'new data folder'); fs.rmSync(sup.dataDir(id), { recursive: true, force: true }); fs.renameSync(incomingDir(id), sup.dataDir(id)); fs.rmSync(incomingManifest(id), { force: true }); sup.setState(id, 'active'); let inst; try { inst = await sup.ensure(id); } catch (e) { sup.setState(id, 'empty'); throw httpError(502, `s3d could not start on the moved folder: ${e.message}`); } const reg = await inst.cli.registration().catch(() => ({ registered: false })); const keys = await sup.indexKeys(inst); return { registered: reg.registered, indexerUrl: reg.indexerUrl || null, users: [...new Set(keys.map((k) => k.user))] }; } function walk(base) { const out = []; const rec = (d) => { for (const e of fs.readdirSync(d, { withFileTypes: true })) { const p = path.join(d, e.name); if (e.isDirectory()) rec(p); else if (e.isFile()) out.push(p); } }; rec(base); return out.map((p) => { const data = fs.readFileSync(p); return { path: path.relative(base, p).split(path.sep).join('/'), size: data.length, sha256: crypto.createHash('sha256').update(data).digest('hex') }; }); } async function checkout(id, force) { const a = sup.account(id); if (a.state === 'moving-out') return { files: walk(sup.dataDir(id)) }; if (a.state !== 'active') throw httpError(409, 'nothing is hosted here'); const inst = await sup.ensure(id); const flushed = await sup.flushQuietly(inst, 10 * 60_000); if (!flushed && !force) throw httpError(409, 'some uploads could not be sent to Sia yet; try again later, or move anyway and they will finish on your computer'); await sup.close(id, 'moving to the owner\'s computer'); sup.setState(id, 'moving-out'); return { files: walk(sup.dataDir(id)) }; } function release(id) { const a = sup.account(id); if (a.state !== 'moving-out') throw httpError(409, 'start the move first'); fs.rmSync(sup.dataDir(id), { recursive: true, force: true }); fs.mkdirSync(sup.dataDir(id), { recursive: true }); saveRules(id, []); sup.setState(id, 'empty'); return { ok: true }; } // ---- S3 ---------------------------------------------------------------------- function splitS3Path(rawPath) { const parts = rawPath.split('/').slice(1); const dec = (s) => { try { return decodeURIComponent(s); } catch { return s; } }; const bucket = dec(parts[0] || ''); const key = parts.length > 1 ? parts.slice(1).map(dec).join('/') : null; return { bucket, key }; } async function s3(req, res, url) { const a = parseAuth(req, url); if (!a) return s3Error(res, 403, 'AccessDenied', 'sign requests with a key; public files are at /p///'); const ownerId = sup.ownerOfKey(a.akid); if (ownerId) { checkReadCap(req, ownerId); const inst = await sup.ensure(ownerId); return rawProxy(req, res, inst); } const g = grants[a.akid]; const rule = g && ruleById(g.id, g.rule); if (!rule) return s3Error(res, 403, 'InvalidAccessKeyId', 'The AWS Access Key Id you provided does not exist in our records.'); verify(req, url, a, rule.secretKey); const raw = req.url.split('?')[0]; const { bucket, key } = splitS3Path(raw); const query = Object.fromEntries(url.searchParams); const { op } = classify(req.method, query, key, req.headers); const d = decideGrant(rule, bucket, op, key, query.prefix || ''); if (!d.allow) return s3Error(res, 403, 'AccessDenied', d.reason); const payload = req.headers['x-amz-content-sha256'] || 'UNSIGNED-PAYLOAD'; if (payload === 'STREAMING-AWS4-HMAC-SHA256-PAYLOAD') { return s3Error(res, 501, 'NotImplemented', 'signed streaming uploads are not supported with shared keys; turn off payload signing in your S3 client'); } checkReadCap(req, g.id); const inst = await sup.ensure(g.id); const client = await sup.bucketClient(inst, bucket); if (!client) return s3Error(res, 404, 'NoSuchBucket', 'The specified bucket does not exist'); if (d.needsAbsent) { const head = await client.request('HEAD', objectPath(bucket, key)); head.resume(); if (head.statusCode === 200) return s3Error(res, 403, 'AccessDenied', 'this key may add files but not replace them'); } const pass = {}; for (const [k, v] of Object.entries(req.headers)) { if (/^(content-type|content-length|content-md5|content-encoding|content-disposition|cache-control|range|if-(match|none-match|modified-since|unmodified-since))$/.test(k) || (k.startsWith('x-amz-') && !/^x-amz-(date|content-sha256|security-token)$/.test(k))) pass[k] = v; } const release = sup.hold(inst); try { const up = await client.request(req.method, key == null ? objectPath(bucket) : objectPath(bucket, key), { query, headers: pass, body: req, payloadHash: payload }); relay(up, res, g.id, req.method); await done(res); } finally { release(); } } function checkReadCap(req, id) { if (req.method === 'GET') checkCap(id); } async function rawProxy(req, res, inst) { const c = readConfig(sup.configFile(inst.id)); const [host, port] = c.apiAddress.split(':'); const release = sup.hold(inst); const u = usageOf(inst.id); try { const up = await new Promise((resolve, reject) => { const headers = { ...req.headers }; delete headers['x-forwarded-for']; delete headers['x-real-ip']; delete headers.connection; const r = http.request({ host, port: Number(port), method: req.method, path: req.url, headers }, resolve); r.on('error', reject); req.on('data', (chunk) => { u.up += chunk.length; }); req.pipe(r); }); relay(up, res, inst.id, req.method); await done(res); } finally { release(); } } function relay(up, res, id, method) { const u = usageOf(id); res.writeHead(up.statusCode, up.headers); if (method === 'GET') up.on('data', (c) => { u.down += c.length; }); up.pipe(res); } // ---- public files and links ---------------------------------------------------- async function serveObject(req, res, id, bucket, key, { inline = true } = {}) { checkCap(id); const inst = await sup.ensure(id); const client = await sup.bucketClient(inst, bucket); if (!client) throw httpError(404, 'not found'); const release = sup.hold(inst); try { const up = await client.getObject(bucket, key, { range: req.headers.range }); if (up.statusCode >= 300 && up.statusCode !== 304) { up.resume(); throw httpError(up.statusCode === 403 ? 404 : up.statusCode, 'not found'); } const headers = {}; for (const h of ['content-type', 'content-length', 'content-range', 'accept-ranges', 'etag', 'last-modified']) if (up.headers[h]) headers[h] = up.headers[h]; const name = key.split('/').pop() || 'download'; headers['content-disposition'] = `${inline ? 'inline' : 'attachment'}; filename*=UTF-8''${encodeURIComponent(name)}`; headers['x-content-type-options'] = 'nosniff'; headers['content-security-policy'] = "sandbox; default-src 'none'; img-src 'self'; media-src 'self'; style-src 'unsafe-inline'"; headers['cache-control'] = 'public, max-age=60'; const u = usageOf(id); res.writeHead(up.statusCode, headers); if (req.method === 'HEAD') { up.resume(); return res.end(); } up.on('data', (c) => { u.down += c.length; }); up.pipe(res); await done(res); } finally { release(); } } async function servePublic(req, res, url) { const m = /^\/p\/([0-9a-f]{32})\/([^/]+)\/?(.*)$/.exec(url.pathname); if (!m) throw httpError(404, 'not found'); const [, id, bucket] = m; const key = decodeURIComponent(m[3]); const rule = publicRuleFor(rulesOf(id), bucket, key); if (!rule) throw httpError(404, 'not found'); if (key === '' || key.endsWith('/')) { if (rule.rights !== 'read-list') throw httpError(404, 'not found'); const inst = await sup.ensure(id); const client = await sup.bucketClient(inst, bucket); if (!client) throw httpError(404, 'not found'); const page = await client.listObjects(bucket, { prefix: key, token: url.searchParams.get('token') || undefined }); return json(res, 200, { folders: page.folders.map((f) => f.slice(key.length)), files: page.objects.filter((o) => o.key !== key).map((o) => ({ name: o.key.slice(key.length), size: o.size, modified: o.modified })), next: page.truncated ? page.nextToken : null, }); } return serveObject(req, res, id, bucket, key); } async function serveLink(req, res, url) { const token = url.pathname.slice(3); const l = links[token]; const rule = l && ruleById(l.id, l.rule); if (!rule || !live(rule)) throw httpError(410, 'this link has expired'); return serveObject(req, res, l.id, rule.bucket, rule.prefix, { inline: !url.searchParams.has('download') }); } // ---- routing ------------------------------------------------------------------- async function handle(req, res) { const url = new URL(req.url, cfg.publicUrl); const isS3 = /^AWS4-HMAC-SHA256 /.test(req.headers.authorization || '') || url.searchParams.has('X-Amz-Algorithm'); if (isS3) return s3(req, res, url); const p = url.pathname; if (p.startsWith('/p/') && (req.method === 'GET' || req.method === 'HEAD')) return servePublic(req, res, url); if (p.startsWith('/l/') && (req.method === 'GET' || req.method === 'HEAD')) return serveLink(req, res, url); if (p === '/hosted/v1/info' && req.method === 'GET') { return json(res, 200, { publicUrl: cfg.publicUrl, gate: gate.describe(), downloadCapBytes: cfg.downloadCapBytes || null, s3Endpoint: cfg.publicUrl }); } if (p === '/hosted/v1/challenge' && req.method === 'GET') { const nonce = auth.challenge(); return json(res, 200, { nonce, message: challengeText(cfg.publicUrl, nonce) }); } if (p === '/hosted/v1/session' && req.method === 'POST') { const body = await jsonBody(req); if (!auth.takeNonce(body.nonce)) throw httpError(401, 'the sign-in request expired; try again'); if (!verifyEd25519(body.pubkey, challengeText(cfg.publicUrl, body.nonce), body.signature)) throw httpError(401, 'the signature does not match'); const id = accountId(body.pubkey); let a = sup.account(id); if (!a) { if (!body.signup) { return json(res, 404, { error: 'no hosted account for this key yet', needsSignup: true, id, gate: gate.describe(), bnsMessage: bnsMessage(id), depositMemo: depositMemo(id) }); } const g = await gate.admit(body.signup, id, clientIp(req)); a = sup.createAccount({ id, pubkey: body.pubkey.toLowerCase(), gate: g }); log(`[${id.slice(0, 8)}] new account via ${g.type}`); } return json(res, 200, { token: auth.issue(id), id, state: a.state }); } if (p.startsWith('/hosted/v1/') || p.startsWith('/api/')) { const { id, token } = bearer(req); if (p.startsWith('/api/')) { if (!FORWARD.some(([m, re]) => m === req.method && re.test(p))) throw httpError(403, 'not available on a hosted account'); return forwardApi(req, res, url, id); } const a = sup.account(id); const q = url.searchParams; switch (`${req.method} ${p}`) { case 'GET /hosted/v1/me': return json(res, 200, { id, state: a.state, gate: { type: a.gate?.type, name: a.gate?.name || null }, createdAt: a.createdAt, usage: usageOf(id), downloadCapBytes: cfg.downloadCapBytes || null, publicUrl: cfg.publicUrl }); case 'DELETE /hosted/v1/session': auth.revoke(token); return json(res, 200, { ok: true }); case 'GET /hosted/v1/rules': return json(res, 200, forOwner(rulesOf(id))); case 'POST /hosted/v1/rules': { const inst = await sup.ensure(id); const rule = validateRule(await jsonBody(req), await sup.mapBuckets(inst)); const rules = rulesOf(id).filter((r) => live(r)); if (rules.length >= 500) throw httpError(400, 'too many rules on one account'); rules.push(rule); saveRules(id, rules); return json(res, 200, rule); } case 'DELETE /hosted/v1/rules': { const rules = rulesOf(id); const next = rules.filter((r) => r.id !== q.get('id')); if (next.length === rules.length) throw httpError(404, 'no such rule'); saveRules(id, next); return json(res, 200, { ok: true }); } case 'PUT /hosted/v1/data/file': return json(res, 200, await receiveFile(req, id, q.get('path'))); case 'POST /hosted/v1/data/commit': return json(res, 200, await commit(id, (await jsonBody(req)).files)); case 'POST /hosted/v1/data/abort': if (a.state !== 'empty') throw httpError(409, 'nothing to abort'); fs.rmSync(incomingDir(id), { recursive: true, force: true }); fs.rmSync(incomingManifest(id), { force: true }); return json(res, 200, { ok: true }); case 'POST /hosted/v1/data/checkout': return json(res, 200, await checkout(id, q.get('force') === '1')); case 'GET /hosted/v1/data/file': { if (a.state !== 'moving-out') throw httpError(409, 'start the move first'); const file = path.join(sup.dataDir(id), safeRel(q.get('path'))); if (!fs.existsSync(file)) throw httpError(404, 'no such file'); res.writeHead(200, { 'content-type': 'application/octet-stream', 'content-length': fs.statSync(file).size }); await pipeline(fs.createReadStream(file), res); return; } case 'POST /hosted/v1/data/cancel-checkout': if (a.state !== 'moving-out') throw httpError(409, 'no move in progress'); sup.setState(id, 'active'); return json(res, 200, { ok: true }); case 'POST /hosted/v1/data/release': return json(res, 200, release(id)); default: throw httpError(404, 'no such endpoint'); } } throw httpError(404, 'not found'); } const server = http.createServer((req, res) => { handle(req, res).catch((err) => { const status = err instanceof SigError ? err.status : err.status || 500; if (status >= 500) log('[hosted]', err.stack || err.message); if (res.headersSent) return res.destroy(); const isS3 = /^AWS4-HMAC-SHA256 /.test(req.headers.authorization || ''); if (isS3) return s3Error(res, status, err.code || (status === 404 ? 'NoSuchKey' : 'InternalError'), err.message); json(res, status, { error: status >= 500 && !err.status ? 'internal error' : err.message }); }); }); server.requestTimeout = 0; // large uploads and downloads stream for a long time const [host, port] = cfg.listen.split(':'); await new Promise((resolve, reject) => { server.once('error', reject); server.listen(Number(port), host, resolve); }); log(`pithos-hosted on ${cfg.listen} for ${cfg.publicUrl}`); return { server, supervisor: sup, async close() { clearInterval(meterTimer); for (const [id, u] of usage) writeJson(path.join(sup.dir(id), 'usage.json'), u); await new Promise((r) => server.close(r)); await sup.closeAll(); }, }; } function json(res, status, body) { const data = JSON.stringify(body); res.writeHead(status, { 'content-type': 'application/json; charset=utf-8', 'cache-control': 'no-store' }); res.end(data); } function s3Error(res, status, code, message) { const esc = (s) => String(s).replace(/[<>&'"]/g, (c) => `&#${c.charCodeAt(0)};`); res.writeHead(status, { 'content-type': 'application/xml' }); res.end(`${esc(code)}${esc(message)}`); } function readAll(stream) { return new Promise((resolve, reject) => { const parts = []; stream.on('data', (c) => parts.push(c)); stream.on('end', () => resolve(Buffer.concat(parts).toString('utf8'))); stream.on('error', reject); }); } async function jsonBody(req) { let size = 0; const parts = []; for await (const c of req) { size += c.length; if (size > 1024 * 1024) throw httpError(413, 'request too large'); parts.push(c); } try { return JSON.parse(Buffer.concat(parts).toString('utf8') || '{}'); } catch { throw httpError(400, 'invalid JSON'); } } function done(res) { return new Promise((resolve) => { if (res.writableFinished) return resolve(); res.on('finish', resolve); res.on('close', resolve); }); } function fmtBytes(n) { const u = ['B', 'KiB', 'MiB', 'GiB', 'TiB']; let i = 0; while (n >= 1024 && i < u.length - 1) { n /= 1024; i++; } return `${n.toFixed(n < 10 && i ? 1 : 0)} ${u[i]}`; }