feat(push): M7 backend — opt-in push notifications via self-hosted ntfy
Additive, v1-only backend contract for the Android app's opt-in push (Part 1 of
M7; docs/android/PLAN.md §11). The app is a pure consumer — this lands the
endpoints, fan-out, and relay it needs.
- Schema: push_devices (per-device endpoint) + notification_subscriptions
(per-user opted-in streams), FK→users ON DELETE CASCADE.
- Stream catalog + event→stream mapping (config/notificationStreams.js): public
streams (news.post, server.status, idoc.warning, champ.start, governor.election)
drawn ONLY from the SSE PUBLIC_KINDS allowlist; personal owner-keyed streams
(vendor.sale, house.idoc, account.login). Full-state upserts (champ/city) fire
only on a real transition via an injectable tracker.
- Fan-out (utils/pushDispatch.js): content-free tickles ({ stream, ref }) POSTed
to each subscribed device; never throws. Two producers — shardIngest.ingest
(beside the SSE broadcast) and the create/publish-post path (news.post).
Personal events resolve to the owner via shardLinks. SSRF guard: endpoints must
be HTTPS, non-private, and on the NTFY_BASE_URL/NTFY_ALLOWED_ORIGINS allow-set —
enforced at registration and every publish.
- Routes under the role-agnostic self surface (never /admin): POST|GET
/auth/me/devices, DELETE /auth/me/devices/:id, GET
/auth/me/notifications/streams, GET|PUT /auth/me/notifications/subscriptions.
Swagger regenerated (4 paths, PushDevice/NotificationStreams/etc. schemas).
- ntfy service in docker-compose.yml: pinned image, declarative ./ntfy/server.yml,
no published host port, anonymous unguessable topics (no accounts) — zero
interactive setup. No publish token required (content-free design); optional
NTFY_PUBLISH_TOKEN honored.
- Tests: pushDispatch (mapping, PUBLIC_KINDS gate, owner-keying, SSRF guard,
content-free payload) + notifications route auth gate. Full suite green (247).
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
40
server/test/notificationsRoutes.test.js
Normal file
40
server/test/notificationsRoutes.test.js
Normal file
@@ -0,0 +1,40 @@
|
||||
// Point the DB at a closed port BEFORE requiring anything that builds the pool —
|
||||
// these cases reject at requireAuth (no session token) before any query runs.
|
||||
process.env.DB_HOST = '127.0.0.1'
|
||||
process.env.DB_PORT = '59999'
|
||||
|
||||
const { test, after } = require('node:test')
|
||||
const assert = require('node:assert/strict')
|
||||
|
||||
const { startApp } = require('./_helper')
|
||||
const authRouter = require('../src/router/v1/auth/auth.routes')
|
||||
const db = require('../src/utils/db')
|
||||
|
||||
after(() => db.close())
|
||||
|
||||
// The push-notification self surface (/auth/me/devices*, /auth/me/notifications/*)
|
||||
// must be mounted AND gated: an unauthenticated caller gets 401 on every route —
|
||||
// never 404 (route missing) and never 200 (gate bypassed).
|
||||
test('/auth/me push routes reject unauthenticated callers with 401', async () => {
|
||||
const app = await startApp((a) => a.use('/api/v1/auth', authRouter))
|
||||
try {
|
||||
const calls = [
|
||||
['GET', '/api/v1/auth/me/devices'],
|
||||
['POST', '/api/v1/auth/me/devices', { endpoint: 'https://ntfy.example.com/UPabc' }],
|
||||
['DELETE', '/api/v1/auth/me/devices/1'],
|
||||
['GET', '/api/v1/auth/me/notifications/streams'],
|
||||
['GET', '/api/v1/auth/me/notifications/subscriptions'],
|
||||
['PUT', '/api/v1/auth/me/notifications/subscriptions', { streams: ['news.post'] }],
|
||||
]
|
||||
for (const [method, path, body] of calls) {
|
||||
const res = await fetch(app.url + path, {
|
||||
method,
|
||||
headers: body ? { 'Content-Type': 'application/json' } : {},
|
||||
body: body ? JSON.stringify(body) : undefined,
|
||||
})
|
||||
assert.equal(res.status, 401, `${method} ${path} should be 401, got ${res.status}`)
|
||||
}
|
||||
} finally {
|
||||
await app.close()
|
||||
}
|
||||
})
|
||||
235
server/test/pushDispatch.test.js
Normal file
235
server/test/pushDispatch.test.js
Normal file
@@ -0,0 +1,235 @@
|
||||
// Point the DB at a closed port before requiring anything that builds the pool —
|
||||
// these tests inject fake models, so no real query should ever run.
|
||||
process.env.DB_HOST = '127.0.0.1'
|
||||
process.env.DB_PORT = '59999'
|
||||
|
||||
const { test, after } = require('node:test')
|
||||
const assert = require('node:assert/strict')
|
||||
|
||||
const { mapShardEvent, createTracker } = require('../src/config/notificationStreams')
|
||||
const pushDispatch = require('../src/utils/pushDispatch')
|
||||
const db = require('../src/utils/db')
|
||||
|
||||
after(() => db.close())
|
||||
|
||||
// Run body with env keys set, then restore prior values.
|
||||
function withEnv(vars, fn) {
|
||||
const prior = {}
|
||||
for (const [k, v] of Object.entries(vars)) {
|
||||
prior[k] = process.env[k]
|
||||
if (v === undefined) delete process.env[k]
|
||||
else process.env[k] = v
|
||||
}
|
||||
try {
|
||||
return fn()
|
||||
} finally {
|
||||
for (const [k, v] of Object.entries(prior)) {
|
||||
if (v === undefined) delete process.env[k]
|
||||
else process.env[k] = v
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── mapShardEvent ───────────────────────────────────────────────────────────
|
||||
|
||||
test('server.hello / shutdown / crashed map to the public server.status stream', () => {
|
||||
const t = createTracker()
|
||||
assert.deepEqual(mapShardEvent({ kind: 'server.hello', bootId: 'b1' }, t), [
|
||||
{ streamId: 'server.status', ref: 'up:b1' },
|
||||
])
|
||||
assert.deepEqual(mapShardEvent({ kind: 'server.shutdown' }, t), [{ streamId: 'server.status', ref: 'down' }])
|
||||
assert.deepEqual(mapShardEvent({ kind: 'server.crashed' }, t), [{ streamId: 'server.status', ref: 'down' }])
|
||||
})
|
||||
|
||||
test('house.decay INTO idoc yields the public idoc.warning AND the owner-keyed house.idoc', () => {
|
||||
const t = createTracker()
|
||||
const out = mapShardEvent({ kind: 'house.decay', to: 'IDOC', serial: '0x40', ownerAcct: 'bob' }, t)
|
||||
assert.deepEqual(out, [
|
||||
{ streamId: 'idoc.warning', ref: '0x40' },
|
||||
{ streamId: 'house.idoc', ref: '0x40', ownerAccount: 'bob' },
|
||||
])
|
||||
// A non-IDOC decay stage produces nothing.
|
||||
assert.deepEqual(mapShardEvent({ kind: 'house.decay', to: 'Fairly', serial: '0x41' }, t), [])
|
||||
})
|
||||
|
||||
test('champ.update fires champ.start only on the inactive→active transition', () => {
|
||||
const t = createTracker()
|
||||
// First sight active → start.
|
||||
assert.deepEqual(mapShardEvent({ kind: 'champ.update', serial: 'c1', active: true }, t), [
|
||||
{ streamId: 'champ.start', ref: 'c1' },
|
||||
])
|
||||
// Still active → no re-fire.
|
||||
assert.deepEqual(mapShardEvent({ kind: 'champ.update', serial: 'c1', active: true }, t), [])
|
||||
// Goes inactive, then active again → fires again.
|
||||
assert.deepEqual(mapShardEvent({ kind: 'champ.update', serial: 'c1', active: false }, t), [])
|
||||
assert.deepEqual(mapShardEvent({ kind: 'champ.update', serial: 'c1', active: true }, t), [
|
||||
{ streamId: 'champ.start', ref: 'c1' },
|
||||
])
|
||||
})
|
||||
|
||||
test('city.update fires governor.election only on a real governor change, never on first sight', () => {
|
||||
const t = createTracker()
|
||||
// First sight of the city → no election (could be a reconnect snapshot).
|
||||
assert.deepEqual(mapShardEvent({ kind: 'city.update', city: 'Britain', governor: { serial: '0x1' } }, t), [])
|
||||
// Same governor → nothing.
|
||||
assert.deepEqual(mapShardEvent({ kind: 'city.update', city: 'Britain', governor: { serial: '0x1' } }, t), [])
|
||||
// New governor → election.
|
||||
assert.deepEqual(mapShardEvent({ kind: 'city.update', city: 'Britain', governor: { serial: '0x2' } }, t), [
|
||||
{ streamId: 'governor.election', ref: 'Britain' },
|
||||
])
|
||||
})
|
||||
|
||||
test('personal streams are owner-keyed and sensitive kinds never yield a public target', () => {
|
||||
const t = createTracker()
|
||||
const sale = mapShardEvent({ kind: 'vendor.sale', ownerAcct: 'bob', t: 7 }, t)
|
||||
assert.deepEqual(sale, [{ streamId: 'vendor.sale', ref: '7', ownerAccount: 'bob' }])
|
||||
|
||||
const login = mapShardEvent({ kind: 'account.login.attempt', acct: 'bob', ip: '1.2.3.4', t: 9 }, t)
|
||||
assert.deepEqual(login, [{ streamId: 'account.login', ref: '9', ownerAccount: 'bob' }])
|
||||
|
||||
// Every personal target carries an ownerAccount (never a bare public push).
|
||||
for (const target of [...sale, ...login]) assert.ok(target.ownerAccount, 'personal target must be owner-keyed')
|
||||
|
||||
// A truly sensitive, unmapped kind produces nothing at all.
|
||||
assert.deepEqual(mapShardEvent({ kind: 'cheat.fastwalk', acct: 'bob' }, t), [])
|
||||
assert.deepEqual(mapShardEvent({ kind: 'admin.audit', actor: 'staff' }, t), [])
|
||||
})
|
||||
|
||||
// ── isAllowedEndpoint (SSRF guard) ──────────────────────────────────────────
|
||||
|
||||
test('isAllowedEndpoint pins the configured ntfy origin and rejects everything else', () => {
|
||||
withEnv({ NTFY_BASE_URL: 'https://ntfy.example.com', NTFY_ALLOWED_ORIGINS: undefined }, () => {
|
||||
assert.equal(pushDispatch.isAllowedEndpoint('https://ntfy.example.com/UPabc'), true)
|
||||
assert.equal(pushDispatch.isAllowedEndpoint('https://evil.example.com/x'), false) // wrong origin
|
||||
assert.equal(pushDispatch.isAllowedEndpoint('http://ntfy.example.com/x'), false) // not https
|
||||
assert.equal(pushDispatch.isAllowedEndpoint('https://127.0.0.1/x'), false) // loopback
|
||||
assert.equal(pushDispatch.isAllowedEndpoint('not a url'), false)
|
||||
})
|
||||
})
|
||||
|
||||
test('isAllowedEndpoint (no allow-set) permits any public https host but blocks private/loopback/http', () => {
|
||||
withEnv({ NTFY_BASE_URL: undefined, NTFY_ALLOWED_ORIGINS: undefined }, () => {
|
||||
assert.equal(pushDispatch.isAllowedEndpoint('https://relay.somehost.net/UPabc'), true)
|
||||
assert.equal(pushDispatch.isAllowedEndpoint('http://relay.somehost.net/x'), false)
|
||||
assert.equal(pushDispatch.isAllowedEndpoint('https://10.0.0.5/x'), false)
|
||||
assert.equal(pushDispatch.isAllowedEndpoint('https://192.168.1.10/x'), false)
|
||||
assert.equal(pushDispatch.isAllowedEndpoint('https://localhost/x'), false)
|
||||
})
|
||||
})
|
||||
|
||||
// ── publish ─────────────────────────────────────────────────────────────────
|
||||
|
||||
function captureFetch() {
|
||||
const calls = []
|
||||
return {
|
||||
calls,
|
||||
fetchImpl: async (url, opts) => {
|
||||
calls.push({ url, opts })
|
||||
return { ok: true, status: 200 }
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
test('publish sends a content-free tickle to public subscribers', async () => {
|
||||
await withEnv({ NTFY_BASE_URL: undefined, NTFY_ALLOWED_ORIGINS: undefined }, async () => {
|
||||
const { calls, fetchImpl } = captureFetch()
|
||||
const streamCalls = []
|
||||
const pushDevices = {
|
||||
endpointsForStream: async (s) => {
|
||||
streamCalls.push(s)
|
||||
return [{ endpoint: 'https://relay.test/UPa', transport: 'unifiedpush' }]
|
||||
},
|
||||
endpointsForUserStream: async () => [],
|
||||
}
|
||||
await pushDispatch.publish('news.post', { ref: '5' }, { pushDevices, fetchImpl })
|
||||
assert.deepEqual(streamCalls, ['news.post'])
|
||||
assert.equal(calls.length, 1)
|
||||
assert.equal(calls[0].url, 'https://relay.test/UPa')
|
||||
assert.equal(calls[0].opts.method, 'POST')
|
||||
assert.deepEqual(JSON.parse(calls[0].opts.body), { stream: 'news.post', ref: '5' })
|
||||
})
|
||||
})
|
||||
|
||||
test('publish (personal) targets only the owner’s subscribed devices', async () => {
|
||||
await withEnv({ NTFY_BASE_URL: undefined, NTFY_ALLOWED_ORIGINS: undefined }, async () => {
|
||||
const { calls, fetchImpl } = captureFetch()
|
||||
const userStreamCalls = []
|
||||
const pushDevices = {
|
||||
endpointsForStream: async () => {
|
||||
throw new Error('public path must not be used for a personal publish')
|
||||
},
|
||||
endpointsForUserStream: async (userId, s) => {
|
||||
userStreamCalls.push([userId, s])
|
||||
return [{ endpoint: 'https://relay.test/UPb' }]
|
||||
},
|
||||
}
|
||||
await pushDispatch.publish('vendor.sale', { ref: '9', ownerUserId: 42 }, { pushDevices, fetchImpl })
|
||||
assert.deepEqual(userStreamCalls, [[42, 'vendor.sale']])
|
||||
assert.equal(calls.length, 1)
|
||||
})
|
||||
})
|
||||
|
||||
test('publish skips endpoints that fail the SSRF guard', async () => {
|
||||
await withEnv({ NTFY_BASE_URL: undefined, NTFY_ALLOWED_ORIGINS: undefined }, async () => {
|
||||
const { calls, fetchImpl } = captureFetch()
|
||||
const pushDevices = {
|
||||
endpointsForStream: async () => [
|
||||
{ endpoint: 'http://relay.test/insecure' }, // not https → skipped
|
||||
{ endpoint: 'https://10.0.0.9/private' }, // private → skipped
|
||||
{ endpoint: 'https://relay.test/ok' }, // delivered
|
||||
],
|
||||
endpointsForUserStream: async () => [],
|
||||
}
|
||||
await pushDispatch.publish('server.status', { ref: 'down' }, { pushDevices, fetchImpl })
|
||||
assert.equal(calls.length, 1)
|
||||
assert.equal(calls[0].url, 'https://relay.test/ok')
|
||||
})
|
||||
})
|
||||
|
||||
// ── fromShardEvent (owner resolution) ───────────────────────────────────────
|
||||
|
||||
test('fromShardEvent resolves a personal event to the owning user, or drops it if unlinked', async () => {
|
||||
await withEnv({ NTFY_BASE_URL: undefined, NTFY_ALLOWED_ORIGINS: undefined }, async () => {
|
||||
const { calls, fetchImpl } = captureFetch()
|
||||
const userStreamCalls = []
|
||||
const shardLinks = { getByAccount: async (acct) => (acct === 'mine' ? { userId: 42 } : null) }
|
||||
const pushDevices = {
|
||||
endpointsForStream: async () => [],
|
||||
endpointsForUserStream: async (userId, s) => {
|
||||
userStreamCalls.push([userId, s])
|
||||
return [{ endpoint: 'https://relay.test/UPc' }]
|
||||
},
|
||||
}
|
||||
const deps = { shardLinks, pushDevices, fetchImpl, tracker: createTracker() }
|
||||
|
||||
await pushDispatch.fromShardEvent({ kind: 'vendor.sale', ownerAcct: 'mine', t: 1 }, deps)
|
||||
assert.deepEqual(userStreamCalls, [[42, 'vendor.sale']])
|
||||
assert.equal(calls.length, 1)
|
||||
|
||||
// Unlinked account → nobody to notify → no publish.
|
||||
userStreamCalls.length = 0
|
||||
calls.length = 0
|
||||
await pushDispatch.fromShardEvent({ kind: 'vendor.sale', ownerAcct: 'stranger', t: 2 }, deps)
|
||||
assert.equal(userStreamCalls.length, 0)
|
||||
assert.equal(calls.length, 0)
|
||||
})
|
||||
})
|
||||
|
||||
test('fromShardEvent fans a public shard event to the stream’s subscribers', async () => {
|
||||
await withEnv({ NTFY_BASE_URL: undefined, NTFY_ALLOWED_ORIGINS: undefined }, async () => {
|
||||
const { fetchImpl } = captureFetch()
|
||||
const publicCalls = []
|
||||
const pushDevices = {
|
||||
endpointsForStream: async (s) => {
|
||||
publicCalls.push(s)
|
||||
return []
|
||||
},
|
||||
endpointsForUserStream: async () => [],
|
||||
}
|
||||
await pushDispatch.fromShardEvent(
|
||||
{ kind: 'server.hello', bootId: 'b1' },
|
||||
{ pushDevices, fetchImpl, tracker: createTracker() },
|
||||
)
|
||||
assert.deepEqual(publicCalls, ['server.status'])
|
||||
})
|
||||
})
|
||||
@@ -22,6 +22,8 @@ function makeDeps() {
|
||||
},
|
||||
uoLinkConfig: { recordStatus: noop },
|
||||
broadcast: (ev) => { calls.broadcast.push(ev) },
|
||||
// No-op push fan-out so ingest() stays hermetic (no real relay/DB).
|
||||
pushDispatch: async () => {},
|
||||
log: { warn() {}, info() {}, error() {} },
|
||||
}
|
||||
}
|
||||
|
||||
@@ -33,6 +33,8 @@ function makeDeps() {
|
||||
shardLinks: { removeByAccount: async (account) => { calls.linkRemove.push(account) } },
|
||||
uoLinkConfig: { recordStatus: noop },
|
||||
broadcast: (ev) => { calls.broadcast.push(ev) },
|
||||
// No-op push fan-out so ingest() stays hermetic (no real relay/DB).
|
||||
pushDispatch: async () => {},
|
||||
log: { warn() {}, info() {}, error() {} },
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user