serveBlob: only serve blobs referenced by feeds we replicate

serveBlob handed any held blob to anyone who knew its hash, on a public
port with require_opt_in:false. Combined with ssb-blobs sympathy that
meant publicly serving 4.6GB of strangers content we never reviewed.

Maintains a set of blob ids referenced by the local log (small, ~360KB,
rescanned every 5 min) and 404s anything outside it. Fails open until the
first scan completes so a read error cannot 404 the whole site.

Verified: a blob present on disk but referenced by no message returns
blobs.has=true over RPC while the viewer 404s it.

index.js

@@ -44,6 +44,43 @@ exports.version = require('./package').version
 
 exports.init = function (sbot, config) {
   var conf = config.viewer || {}
+
+  // --- known-blob gate -----------------------------------------------------
+  // This node used to cache blobs on behalf of strangers (ssb-blobs `sympathy`
+  // defaults to 3), and serveBlob would hand any of them to anyone who knew the
+  // hash. sympathy is 0 now and the cache has been pruned, so everything we hold
+  // is referenced by a feed we replicate. This keeps that true even if a stray
+  // blob ever lands.
+  //
+  // The local log is small (~360KB), so a regex scan is cheap. It is refreshed
+  // periodically rather than kept live: being a few minutes stale can only delay
+  // a legitimate image, never serve one that is not ours.
+  var knownBlobs = null   // null = not loaded yet -> fail open, never 404 everything
+  var logOffsetPath = path.join(config.path, 'flume', 'log.offset')
+
+  function refreshKnownBlobs() {
+    fs.readFile(logOffsetPath, function (err, buf) {
+      if (err) {
+        console.error('[viewer] blob gate: could not read log.offset:', err.message)
+        return   // keep whatever set we already had
+      }
+      var found = buf.toString('binary').match(/&[A-Za-z0-9+/]{43}=\.sha256/g) || []
+      var next = Object.create(null)
+      found.forEach(function (id) { next[id] = true })
+      if (knownBlobs === null) {
+        console.log('[viewer] blob gate active:', Object.keys(next).length, 'known blobs')
+      }
+      knownBlobs = next
+    })
+  }
+  refreshKnownBlobs()
+  setInterval(refreshKnownBlobs, 5 * 60 * 1000).unref()
+
+  // serveBlob is a free function outside this closure, so hand the check over.
+  sbot.isKnownBlob = function (id) {
+    if (knownBlobs === null) return true
+    return !!knownBlobs[id]
+  }
   var port = conf.port || 8807
   var host = conf.host || config.host || '::'
 
@@ -474,6 +511,12 @@ function serveBlob(req, res, sbot, id, query) {
   var etag = id + (unbox || '')
 
   if (req.headers['if-none-match'] === etag) return respond(res, 304)
+  // Only serve blobs referenced by a feed we replicate. Anything else is not
+  // ours to hand out, so treat it as absent rather than confirm we hold it.
+  if (typeof sbot.isKnownBlob === 'function' && !sbot.isKnownBlob(id)) {
+    return respond(res, 404, 'Not found')
+  }
+
   sbot.blobs.has(id, function (err, has) {
     if (err) {
       if (/^invalid/.test(err.message)) return respond(res, 400, err.message)

prune-blobs.js

@@ -0,0 +1,163 @@
+#!/usr/bin/env node
+//
+// prune-blobs.js — delete cached blobs that none of our own feeds reference.
+//
+// CONTEXT
+// -------
+// ssb-blobs `sympathy` (default 3) made this node fetch and cache blobs on
+// behalf of any peer up to 3 network hops away. With port 8008 open to the
+// world, that turned it into a free CDN: 31,283 blobs / 4.6 GB, of which only
+// 267 are referenced by any message we hold.
+//
+// Once sympathy is 0 the inflow stops, but the existing hoard stays on disk --
+// and ssb-viewer will serve any of it by hash to anyone who asks. This removes
+// it.
+//
+// SAFETY
+// ------
+//   * Dry run by default. Pass --delete to actually remove anything.
+//   * Every deleted id is written to a manifest first. Blobs are content-
+//     addressed, so anything deleted in error can be re-requested with
+//     `ssb-server blobs.want <id>` if a peer still has it.
+//   * The keep set is built from the raw log, so it does not depend on sbot
+//     being up or on any index being in sync.
+//
+// Blobs are stored by multiblob at blobs/<alg>/<hex[0:2]>/<hex[2:]>, where hex
+// is the base64 blob hash decoded to hex (see multiblob/index.js toPath).
+
+var fs = require('fs')
+var path = require('path')
+
+var DELETE = process.argv.indexOf('--delete') !== -1
+var HOME = process.env.HOME || '/home/cyberian'
+var LOG_OFFSET = path.join(HOME, '.ssb/flume/log.offset')
+var BLOB_DIR = path.join(HOME, '.ssb/blobs/sha256')
+var PUSH_DIR = path.join(HOME, '.ssb/blobs_push')
+var MANIFEST = path.join(HOME, 'logs/pruned-blobs.txt')
+
+var BLOB_RE = /&[A-Za-z0-9+/]{43}=\.sha256/g
+// sha256 of the empty string; ssb-blobs special-cases it and never stores it.
+var EMPTY_HASH = '&47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZG3hSuFU=.sha256'
+
+function idToHex (id) {
+  return Buffer.from(id.slice(1).replace(/\.sha256$/, ''), 'base64').toString('hex')
+}
+function hexToId (hex) {
+  return '&' + Buffer.from(hex, 'hex').toString('base64') + '.sha256'
+}
+function human (bytes) {
+  var u = ['B', 'KB', 'MB', 'GB']
+  var i = 0
+  while (bytes >= 1024 && i < u.length - 1) { bytes /= 1024; i++ }
+  return bytes.toFixed(1) + ' ' + u[i]
+}
+
+// ---- keep set -------------------------------------------------------------
+var keep = Object.create(null)
+keep[EMPTY_HASH] = true
+
+var log = fs.readFileSync(LOG_OFFSET).toString('binary')
+var refs = log.match(BLOB_RE) || []
+refs.forEach(function (r) { keep[r] = true })
+var fromLog = Object.keys(keep).length - 1
+
+// Anything queued for push to the network is ours by definition. The leveldb
+// is locked by the running sbot, so scrape the ids out of its files instead of
+// opening it.
+var fromPush = 0
+try {
+  fs.readdirSync(PUSH_DIR).forEach(function (f) {
+    if (!/\.(log|ldb)$/.test(f)) return
+    var buf = fs.readFileSync(path.join(PUSH_DIR, f)).toString('binary')
+    ;(buf.match(BLOB_RE) || []).forEach(function (r) {
+      if (!keep[r]) { keep[r] = true; fromPush++ }
+    })
+  })
+} catch (e) {
+  console.error('warning: could not scan blobs_push:', e.message)
+}
+
+// ---- walk the blob store --------------------------------------------------
+var kept = { n: 0, bytes: 0 }
+var doomed = []
+var orphans = []
+
+fs.readdirSync(BLOB_DIR).forEach(function (sub) {
+  var subdir = path.join(BLOB_DIR, sub)
+  var st
+  try { st = fs.statSync(subdir) } catch (e) { return }
+  if (!st.isDirectory()) return
+
+  fs.readdirSync(subdir).forEach(function (name) {
+    var full = path.join(subdir, name)
+    var size
+    try { size = fs.statSync(full).size } catch (e) { return }
+
+    var id
+    try {
+      id = hexToId(sub + name)
+    } catch (e) {
+      orphans.push(full)
+      return
+    }
+
+    if (keep[id]) { kept.n++; kept.bytes += size }
+    else doomed.push({ id: id, path: full, size: size })
+  })
+})
+
+var doomedBytes = doomed.reduce(function (a, b) { return a + b.size }, 0)
+
+console.log('keep set:      ' + (fromLog + fromPush) +
+            ' (' + fromLog + ' referenced in the log, ' + fromPush + ' from blobs_push)')
+console.log('blobs on disk: ' + (kept.n + doomed.length))
+console.log('  keeping:     ' + kept.n + '  (' + human(kept.bytes) + ')')
+console.log('  removing:    ' + doomed.length + '  (' + human(doomedBytes) + ')')
+if (orphans.length) console.log('  unparseable: ' + orphans.length + ' (left alone)')
+
+// A keep-set entry we do not actually have on disk is fine -- blob-wanter has a
+// standing want for it. Just report so the numbers reconcile.
+var missing = Object.keys(keep).filter(function (id) {
+  if (id === EMPTY_HASH) return false
+  var hex = idToHex(id)
+  return !fs.existsSync(path.join(BLOB_DIR, hex.substring(0, 2), hex.substring(2)))
+})
+if (missing.length) {
+  console.log('  note: ' + missing.length + ' referenced blob(s) are not on disk; ' +
+              'blob-wanter holds standing wants for them')
+}
+
+if (!DELETE) {
+  console.log('\nDRY RUN — nothing deleted. Re-run with --delete to apply.')
+  process.exit(0)
+}
+
+// ---- delete ---------------------------------------------------------------
+// Manifest first, so a crash mid-delete still leaves a record of what went.
+fs.mkdirSync(path.dirname(MANIFEST), { recursive: true })
+var out = fs.createWriteStream(MANIFEST, { flags: 'a' })
+out.write('# pruned ' + new Date().toISOString() + ' — ' + doomed.length +
+          ' blobs, ' + human(doomedBytes) + '\n')
+out.write('# re-request any of these with: ssb-server blobs.want <id>\n')
+doomed.forEach(function (d) { out.write(d.id + '\t' + d.size + '\n') })
+out.end()
+
+var removed = 0, failed = 0, freed = 0
+doomed.forEach(function (d) {
+  try { fs.unlinkSync(d.path); removed++; freed += d.size }
+  catch (e) { failed++ }
+})
+
+// Clean up the now-empty two-char shard directories.
+fs.readdirSync(BLOB_DIR).forEach(function (sub) {
+  var subdir = path.join(BLOB_DIR, sub)
+  try {
+    if (fs.statSync(subdir).isDirectory() && fs.readdirSync(subdir).length === 0) {
+      fs.rmdirSync(subdir)
+    }
+  } catch (e) { /* not empty, or vanished */ }
+})
+
+console.log('\nremoved ' + removed + ' blobs, freed ' + human(freed) +
+            (failed ? ', ' + failed + ' failed' : ''))
+console.log('manifest: ' + MANIFEST)