From 24206957bb3a74bdde270cbd2e7ce4a3cee2cdbf Mon Sep 17 00:00:00 2001 From: hermes-gadget Date: Thu, 1 Oct 2026 00:39:40 +0100 Subject: [PATCH] feat(owner): land owner-controlled packet sharing (rescued VPS WIP) (#101) * feat(owner): add owner-controlled packet sharing * docs: record B3 packet-sharing handoff * docs(api): regenerate openapi for packet-sharing routes * test(db): extend migration registry expectation to 052-054 --------- Co-authored-by: gadgethd <111318106+gadgethd@users.noreply.github.com> --- HANDOFF.md | 45 +- backend/package.json | 1 + backend/src/api/contracts.test.ts | 19 + backend/src/api/contracts.ts | 82 +++- backend/src/api/routes.ts | 2 + backend/src/api/routes/owner.ts | 123 ++++++ .../src/api/routes/ownerPacketSharing.test.ts | 229 ++++++++++ backend/src/db/dataLifecycle.test.ts | 1 + backend/src/db/dataLifecycle.ts | 7 + backend/src/db/index.ts | 27 ++ .../migrations/052_owner_packet_sharing.sql | 66 +++ ...wner_packet_share_destination_metadata.sql | 28 ++ .../054_owner_packet_share_default_off.sql | 5 + .../src/db/packetBatch.integration.test.ts | 3 + backend/src/index.ts | 18 +- backend/src/metrics.ts | 13 + backend/src/mqtt/client.ts | 8 +- backend/src/owner/packetShareConfig.test.ts | 113 +++++ backend/src/owner/packetShareConfig.ts | 197 +++++++++ backend/src/owner/packetShareCrypto.ts | 60 +++ .../src/owner/packetShareForwarder.test.ts | 99 +++++ backend/src/owner/packetShareForwarder.ts | 270 ++++++++++++ backend/src/owner/packetShareRepository.ts | 265 ++++++++++++ backend/src/tools/encryptPacketShareSecret.ts | 15 + docs/openapi.yaml | 403 ++++++++++++++++++ 25 files changed, 2065 insertions(+), 34 deletions(-) create mode 100644 backend/src/api/routes/ownerPacketSharing.test.ts create mode 100644 backend/src/db/migrations/052_owner_packet_sharing.sql create mode 100644 backend/src/db/migrations/053_owner_packet_share_destination_metadata.sql create mode 100644 backend/src/db/migrations/054_owner_packet_share_default_off.sql create mode 100644 backend/src/owner/packetShareConfig.test.ts create mode 100644 backend/src/owner/packetShareConfig.ts create mode 100644 backend/src/owner/packetShareCrypto.ts create mode 100644 backend/src/owner/packetShareForwarder.test.ts create mode 100644 backend/src/owner/packetShareForwarder.ts create mode 100644 backend/src/owner/packetShareRepository.ts create mode 100644 backend/src/tools/encryptPacketShareSecret.ts diff --git a/HANDOFF.md b/HANDOFF.md index cfb05af..d7d30b9 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -1,40 +1,29 @@ -# Mission C5d handoff +# B3 packet-sharing handoff -Status: implemented; production rollout remains gated. -Branch: `fix/ukmesh-ops-pipeline` -Base: fetched `origin/main` at `2fa80fa`. -Push state: not pushed; coordinator relays this branch. +## Status -## Commits +Backend packet-sharing routes, contracts, persistence, forwarding, and tests are committed locally in `fbf9803` (`feat(owner): add owner-controlled packet sharing`). No migration was applied, no service was restarted, and nothing was pushed. -- `4c549bb` — `test(viewshed): inject post-commit side-effect failures` -- `1d4d594` — `fix(deploy): rollback on unhealthy or mismatched sites` -- `f48c652` — `fix(alerts): persist forwarding queue across restarts` -- `d93030c` — `perf(stats): add per-network aggregate cutover controls` +## WIP inventory -## Changes +The packet-sharing slice rescued from the main checkout contains migrations 052–054 for destination/rule/delivery tables, destination metadata, and default-off rules; owner GET/POST routes and contracts; broker-secret encryption; destination configuration and delivery-queue services; MQTT forwarding and worker wiring; and route/config/forwarder tests. The main checkout also has mixed public-feed, tagger, owner-portal, and unrelated backend work. -- #47: the fetched main already contained the Redis completion marker and replay path. Added fault injection at link admission, both notification publishes, and marker persistence to verify retry behavior through the already-calculated path. Per the brief, this new test was not run. -- #54: deployment now waits for a running, healthy container and a successful HTTP response, rejects empty bundle lists and mismatches, and rolls back the prior pin on any failure. Rollback checks the restored service and supports an empty prior pin. Added mocked deploy tests. -- #55: forwarding records are fsynced into the mounted queue before HTTP 202. Startup restores pending records; retries use bounded backoff and dead-letter files. Health JSON exposes queue age, pending/dead-letter counts, last success, and last error. Archive-only mode returns degraded status. Updated the alert runbook and its archive-only test expectation. -- #57: existing aggregate controls now accept comma-separated network names through the Compose-forwarded variables; shadow comparisons can run while reads stay off. Chart scans share one `asOf` value, cancel as a batch on query failure, and enforce duration, timeout, and returned-row budgets. True PostgreSQL rows-scanned counts are not exposed by the current query API; review query plans and DB scan telemetry during the gated shadow period. +The main checkout and its WIP remain untouched. The packet-sharing backend copy there becomes redundant after this change is merged and deployed. The mixed owner-portal UI edits were not included in the backend scope in this brief and need separate disposition if they are not handled elsewhere. -PR #99 and #100 have no functional overlap with the changed worker, deploy, alert receiver, or stats repository code. PR #100 edits other sections of `docs/operations.md`. Both PRs also add `HANDOFF.md`; combine their handoff sections when integrating those branches. +## Migration decision -## Tests and checks +Kept the supplied `052_owner_packet_sharing.sql`, `053_owner_packet_share_destination_metadata.sql`, and `054_owner_packet_share_default_off.sql` names. No duplicate prefix exists for these files in the target worktree, so no renumbering was needed. Prefix 055 is reserved by `055_message_tags.sql` on `fix/health-latency-bundle`; the separate untracked `055_public_feed.sql` in the main checkout was not copied. -- From `backend`: `node --import tsx --test src/workers/alertDeliveryQueue.test.ts src/workers/alert-receiver.test.ts src/stats/statsRepository.test.ts` — passed, 18/18. -- From `backend`: `npm run typecheck` — passed. -- From the worktree root: `bash -n scripts/deploy-website.sh scripts/test-deploy-website.sh scripts/test-alert-receiver.sh && bash scripts/test-deploy-website.sh` — passed. +## Tests + +- `cd backend && npm test` — passed, 341 tests. +- `cd backend && npm run typecheck` — passed. - `git diff --check` — passed. -- `viewshed-worker/tests/test_side_effect_markers.py` was not executed, as required by #47. `scripts/test-alert-receiver.sh` was not run because it starts a receiver process, prohibited by the host constraint. -No services or containers were started, no live DB writes or config edits were made, and no push was attempted. +The worktree had no installed dependencies. Tests reused the existing main-checkout `backend/node_modules` through a temporary symlink in this worktree; the symlink was removed afterward. No tests wrote to a live database. -## Gated items for Ben +## Gated items and open questions -- Backfill and catch up hourly rollups, set `STATS_AGGREGATE_SHADOW_ENABLED=ukmesh`, and review clean parity logs before setting `STATS_AGGREGATE_READS_ENABLED=ukmesh`. Keep the shadow window short and review DB scan plans/telemetry because the application result-row guard is not a scanned-row counter. -- Confirm `ALERT_FORWARD_URL` is valid in production and inspect `/healthz` after the approved config rollout. Archive-only operation now reports HTTP 503 health. -- The C5d `HANDOFF.md` must be combined with the handoff content being added by open PRs #99/#100. - -Open question: which database scan telemetry should be used to set a production rows-scanned threshold before additional networks switch to aggregates? +- Ben owns migration application and deployment. +- The target worktree does not contain the `OWNER_PACKET_SHARE_*` Compose passthroughs or `owner_packet_share_deliveries` in the retention target list described by the brief. No `.env` or config files were changed under the VPS constraint. Before deployment, Ben must confirm the runtime passes `OWNER_PACKET_SHARE_ENCRYPTION_KEY` to the backend and includes `owner_packet_share_deliveries` in lifecycle retention targets. Poll interval and batch size have code defaults. +- Confirm whether the packet-sharing owner-portal UI edits in the main checkout are tracked in another change; they were left untouched and are not in `fbf9803`. diff --git a/backend/package.json b/backend/package.json index fa3815d..a4b65c8 100644 --- a/backend/package.json +++ b/backend/package.json @@ -15,6 +15,7 @@ "db:lifecycle": "node dist/tools/dataLifecycle.js", "owner-auth:inventory": "node dist/tools/inventoryOwnerAuthorization.js", "owner-auth:revoke": "node dist/tools/revokeOwnerSessions.js", + "owner-packet-share:encrypt-secret": "node dist/tools/encryptPacketShareSecret.js", "analysis:cleanup-orphans": "node dist/tools/cleanupOrphanedAnalysisRuns.js", "bench:decode": "tsx src/tools/benchmarkPacketDecode.ts", "load:realtime": "tsx src/tools/loadRealtime.ts", diff --git a/backend/src/api/contracts.test.ts b/backend/src/api/contracts.test.ts index b7e5a8b..cfc1d18 100644 --- a/backend/src/api/contracts.test.ts +++ b/backend/src/api/contracts.test.ts @@ -57,3 +57,22 @@ test('slow path contracts cover status, mode query, and pending response', () => true, ); }); + +test('owner packet-sharing contracts cover bounded owner settings and destination metadata', () => { + const get = API_CONTRACTS.find((entry) => entry.method === 'GET' && entry.path === '/owner/packet-sharing'); + assert.equal(get?.access, 'owner'); + assert.deepEqual(get?.queryParameters?.[0]?.['schema'], { + type: 'string', + pattern: '^[0-9A-Fa-f]{64}$', + }); + assert.deepEqual((get?.responseSchema?.['required'] as string[] | undefined)?.sort(), [ + 'destinations', 'enabled', 'featureEnabled', 'nodeId', + ]); + + const post = API_CONTRACTS.find((entry) => entry.method === 'POST' && entry.path === '/owner/packet-sharing'); + assert.equal(post?.access, 'owner'); + assert.deepEqual((post?.requestSchema?.['required'] as string[] | undefined)?.sort(), [ + 'destinationIds', 'enabled', 'nodeId', + ]); + assert.equal(post?.additionalResponses?.['409']?.responseSchema['properties'] != null, true); +}); diff --git a/backend/src/api/contracts.ts b/backend/src/api/contracts.ts index 45287b0..4d5c76f 100644 --- a/backend/src/api/contracts.ts +++ b/backend/src/api/contracts.ts @@ -91,6 +91,34 @@ const FEED_MESSAGE_HISTORY_SCHEMA = { }, }; +const OWNER_PACKET_SHARING_SCHEMA = { + type: 'object', + required: ['nodeId', 'featureEnabled', 'enabled', 'destinations'], + properties: { + nodeId: { type: 'string', pattern: '^[0-9A-Fa-f]{64}$' }, + featureEnabled: { type: 'boolean' }, + enabled: { type: 'boolean' }, + destinations: { + type: 'array', + items: { + type: 'object', + required: ['id', 'name', 'websiteUrl', 'description', 'configured', 'selected', 'lastForwardedAt'], + properties: { + id: { type: 'string', pattern: '^[a-z0-9][a-z0-9_-]{0,63}$' }, + name: { type: 'string', minLength: 1, maxLength: 120 }, + websiteUrl: { type: ['string', 'null'], format: 'uri', maxLength: 2_048 }, + description: { type: ['string', 'null'], maxLength: 160 }, + configured: { type: 'boolean' }, + selected: { type: 'boolean' }, + lastForwardedAt: { type: ['string', 'null'], format: 'date-time' }, + }, + additionalProperties: false, + }, + }, + }, + additionalProperties: false, +}; + function humanize(path: string): string { return path .replace(/^\/v1\//, '') @@ -168,6 +196,7 @@ const OWNER_GET = [ '/owner/csrf', '/owner/live', '/owner/live-last-hop', + '/owner/packet-sharing', '/owner/session', ] as const; @@ -181,10 +210,11 @@ export const API_CONTRACTS: readonly ApiContract[] = [ '/owner/alert-rules/:id/test', '/owner/login', '/owner/logout', + '/owner/packet-sharing', ], 'owner'), ...contracts('DELETE', ['/coverage/planned/:planId'], 'public'), ...contracts('DELETE', ['/owner/alert-rules/:id'], 'owner'), -].map((contract) => { +].map((contract): ApiContract => { if (contract.path === '/path-beta/slow-mode') { return { ...contract, @@ -250,6 +280,56 @@ export const API_CONTRACTS: readonly ApiContract[] = [ }, }; } + if (contract.path === '/owner/packet-sharing' && contract.method === 'GET') { + return { + ...contract, + summary: 'Read owner packet-sharing settings', + responseSchema: OWNER_PACKET_SHARING_SCHEMA, + queryParameters: [{ + name: 'nodeId', + in: 'query', + required: false, + schema: { type: 'string', pattern: '^[0-9A-Fa-f]{64}$' }, + }], + }; + } + if (contract.path === '/owner/packet-sharing' && contract.method === 'POST') { + const destinationIdsSchema = { + type: 'array', + maxItems: 32, + items: { type: 'string', pattern: '^[a-z0-9][a-z0-9_-]{0,63}$' }, + }; + return { + ...contract, + summary: 'Update owner packet-sharing settings', + requestSchema: { + type: 'object', + required: ['nodeId', 'enabled', 'destinationIds'], + properties: { + nodeId: { type: 'string', pattern: '^[0-9A-Fa-f]{64}$' }, + enabled: { type: 'boolean' }, + destinationIds: destinationIdsSchema, + }, + additionalProperties: false, + }, + responseSchema: { + type: 'object', + required: ['nodeId', 'enabled', 'destinationIds'], + properties: { + nodeId: { type: 'string', pattern: '^[0-9A-Fa-f]{64}$' }, + enabled: { type: 'boolean' }, + destinationIds: destinationIdsSchema, + }, + additionalProperties: false, + }, + additionalResponses: { + '409': { + description: 'One or more selected destinations are not configured', + responseSchema: ERROR_SCHEMA, + }, + }, + }; + } return contract; }); diff --git a/backend/src/api/routes.ts b/backend/src/api/routes.ts index e9fd13b..d662833 100644 --- a/backend/src/api/routes.ts +++ b/backend/src/api/routes.ts @@ -30,6 +30,7 @@ import { analyticsQuery, pool, query, + withTransaction, } from '../db/index.js'; import { autoLinkOwnerNodeIds, @@ -212,6 +213,7 @@ registerOwnerRoutes(router, { getOwnerCredentialGeneration, invalidateOwnerNodeIdCache, query, + withTransaction, }); registerPathingRoutes(router, { pathBetaLimiter: PATH_BETA_LIMITER, diff --git a/backend/src/api/routes/owner.ts b/backend/src/api/routes/owner.ts index c31b91d..5b9d621 100644 --- a/backend/src/api/routes/owner.ts +++ b/backend/src/api/routes/owner.ts @@ -4,6 +4,12 @@ import { createOwnerService } from '../../owner/ownerService.js'; import type { OwnerSession } from '../../owner/ownerSession.js'; import { createCsrfToken, readCookie, requireDoubleSubmitCsrf } from '../../security/operatorAuth.js'; import { resolveWebhookTarget } from '../../security/outboundWebhook.js'; +import { + destinationConfigured, + publicDestinationView, +} from '../../owner/packetShareConfig.js'; +import { hasPacketShareEncryptionKey } from '../../owner/packetShareCrypto.js'; +import { createPacketShareRepository, type PacketShareTransactionFn } from '../../owner/packetShareRepository.js'; import { deliverDueOwnerAlerts, queueOwnerTestDelivery, @@ -57,12 +63,14 @@ type OwnerRouteDeps = { getOwnerCredentialGeneration: GetOwnerCredentialGenerationFn; invalidateOwnerNodeIdCache: (mqttUsername: string) => void; query: QueryFn; + withTransaction?: PacketShareTransactionFn; }; export function registerOwnerRoutes(router: Router, deps: OwnerRouteDeps): void { const repository = createOwnerRepository({ query: deps.query, }); + const packetShareRepository = createPacketShareRepository(deps.query, deps.withTransaction); const service = createOwnerService({ ownerLiveCacheTtlMs: deps.ownerLiveCacheTtlMs, @@ -252,6 +260,121 @@ export function registerOwnerRoutes(router: Router, deps: OwnerRouteDeps): void } }); + router.get('/owner/packet-sharing', async (req, res) => { + let requestedNodeId: string | undefined; + try { + requestedNodeId = parseBoundedString(req.query['nodeId'], { + name: 'nodeId', + maxLength: 64, + pattern: /^[0-9a-fA-F]{64}$/, + })?.toUpperCase(); + } catch { + res.status(400).json({ error: 'Invalid packet sharing settings' }); + return; + } + try { + const ownedNodeIds = await deps.requireOwnerSession(req, res); + if (!ownedNodeIds) return; + const session = deps.getOwnerSession(req)!; + const ownedNodeIdSet = new Set(ownedNodeIds.map((nodeId) => nodeId.toUpperCase())); + const nodeId = requestedNodeId ?? ownedNodeIds[0]?.toUpperCase(); + if (!nodeId || !ownedNodeIdSet.has(nodeId)) { + res.status(403).json({ error: 'Node is not owned by this session' }); + return; + } + const [destinations, selectedIds, lastForwardedAt] = await Promise.all([ + packetShareRepository.listDestinations(), + packetShareRepository.listOwnerRules(session.mqttUsername, nodeId), + packetShareRepository.listLastForwardedAt(session.mqttUsername, nodeId), + ]); + const selected = new Set(selectedIds); + const encryptionKeyAvailable = hasPacketShareEncryptionKey(); + res.setHeader('Cache-Control', 'no-store'); + res.json({ + nodeId, + featureEnabled: true, + enabled: selectedIds.length > 0, + destinations: destinations.map((destination) => publicDestinationView( + destination, + selected.has(destination.destination_id), + encryptionKeyAvailable, + lastForwardedAt.get(destination.destination_id) ?? null, + )), + }); + } catch (error) { + console.error('[api] GET /owner/packet-sharing', (error as Error).message); + res.status(500).json({ error: 'Internal server error' }); + } + }); + + router.post('/owner/packet-sharing', csrfProtection, async (req, res) => { + try { + const ownedNodeIds = await deps.requireOwnerSession(req, res); + if (!ownedNodeIds) return; + const session = deps.getOwnerSession(req)!; + const rawBody = req.body as unknown; + if (!rawBody || typeof rawBody !== 'object' || Array.isArray(rawBody)) { + res.status(400).json({ error: 'Invalid packet sharing settings' }); + return; + } + const body = rawBody as { + nodeId?: unknown; + enabled?: unknown; + destinationIds?: unknown; + }; + if (Object.keys(rawBody).some((key) => !['nodeId', 'enabled', 'destinationIds'].includes(key))) { + res.status(400).json({ error: 'Invalid packet sharing settings' }); + return; + } + const nodeId = typeof body.nodeId === 'string' ? body.nodeId.trim().toUpperCase() : ''; + const ownedNodeIdSet = new Set(ownedNodeIds.map((ownedNodeId) => ownedNodeId.toUpperCase())); + const rawDestinationIds = body.destinationIds; + const enabled = body.enabled; + if (!Array.isArray(rawDestinationIds) + || rawDestinationIds.length > 32 + || !rawDestinationIds.every((value) => typeof value === 'string') + || typeof enabled !== 'boolean' + ) { + res.status(400).json({ error: 'Invalid packet sharing settings' }); + return; + } + const destinationIds = Array.from(new Set(rawDestinationIds.map((value) => value.trim().toLowerCase()))); + if (!/^[0-9A-F]{64}$/.test(nodeId) + || destinationIds.some((destinationId) => !/^[a-z0-9][a-z0-9_-]{0,63}$/.test(destinationId))) { + res.status(400).json({ error: 'Invalid packet sharing settings' }); + return; + } + if (!ownedNodeIdSet.has(nodeId)) { + res.status(403).json({ error: 'Node is not owned by this session' }); + return; + } + const destinations = await packetShareRepository.listDestinations(); + const byId = new Map(destinations.map((destination) => [destination.destination_id, destination])); + const encryptionKeyAvailable = hasPacketShareEncryptionKey(); + if (enabled && destinationIds.some((destinationId) => { + const destination = byId.get(destinationId); + return !destination || !destinationConfigured(destination, encryptionKeyAvailable); + })) { + res.status(409).json({ error: 'One or more selected destinations are not configured' }); + return; + } + const persistedDestinationIds = await packetShareRepository.replaceOwnerRules( + session.mqttUsername, + nodeId, + destinationIds, + enabled, + ); + res.status(200).json({ + nodeId, + enabled: enabled && persistedDestinationIds.length > 0, + destinationIds: persistedDestinationIds, + }); + } catch (error) { + console.error('[api] POST /owner/packet-sharing', (error as Error).message); + res.status(500).json({ error: 'Internal server error' }); + } + }); + router.get('/owner/alert-rules', async (req, res) => { try { const ownedNodeIds = await deps.requireOwnerSession(req, res); diff --git a/backend/src/api/routes/ownerPacketSharing.test.ts b/backend/src/api/routes/ownerPacketSharing.test.ts new file mode 100644 index 0000000..9df24d1 --- /dev/null +++ b/backend/src/api/routes/ownerPacketSharing.test.ts @@ -0,0 +1,229 @@ +import assert from 'node:assert/strict'; +import { once } from 'node:events'; +import type { AddressInfo } from 'node:net'; +import test from 'node:test'; +import express, { Router, type RequestHandler } from 'express'; +import type { QueryResultRow } from 'pg'; +import { registerOwnerRoutes } from './owner.js'; + +const nodeId = 'A'.repeat(64); +const destination = { + destination_id: 'remote-site', + display_name: 'Remote site', + website_url: 'https://remote.example.test', + description: 'A remote packet site', + broker_url: 'mqtts://remote.example.test:8883', + topic_template: 'replicated/{network}/{observerId}/packets', + broker_username: null, + broker_password_ciphertext: null, + enabled: true, +}; +const passThroughLimiter: RequestHandler = (_req, _res, next) => next(); + +function ownerRouteDeps( + query: (sql: string, params?: unknown[]) => Promise<{ rows: QueryResultRow[] }>, + mqttUsername = 'node1', + requireOwnerSession: (req: import('express').Request, res: import('express').Response) => Promise = async () => [nodeId], +) { + return { + ownerCookieName: 'meshcore_owner_session', + ownerLiveCacheTtlMs: 1_000, + ownerLiveCache: new Map(), + ownerDashboardCacheTtlMs: 1_000, + ownerLastHopCacheTtlMs: 1_000, + ownerSessionTtlMs: 60_000, + mqttUsernameMaxLen: 128, + mqttPasswordMaxLen: 128, + ownerLoginLimiter: passThroughLimiter, + hasControlChars: () => false, + verifyMqttCredentials: async () => true, + resolveOwnerNodeIds: async () => [nodeId], + autoLinkOwnerNodeIds: async () => [nodeId], + buildOwnerDashboard: async () => ({ nodes: [] }), + encryptOwnerSession: () => 'session', + isSecureRequest: () => false, + getOwnerSession: () => ({ v: 3 as const, mqttUsername, exp: Date.now() + 60_000, gen: 0 }), + requireOwnerSession, + getOwnerCredentialGeneration: async () => 0, + invalidateOwnerNodeIdCache: () => undefined, + query, + withTransaction: async (work: (runQuery: typeof query) => Promise) => work(query), + }; +} + +async function withOwnerApi( + query: (sql: string, params?: unknown[]) => Promise<{ rows: QueryResultRow[] }>, + check: (baseUrl: string, writes: string[]) => Promise, + selectedDestinationRows: QueryResultRow[] = [{ destination_id: 'remote-site' }], + requireOwnerSession?: (req: import('express').Request, res: import('express').Response) => Promise, +) { + const writes: string[] = []; + const router = Router(); + registerOwnerRoutes(router, ownerRouteDeps(async (sql, params) => { + if (sql.startsWith('DELETE FROM owner_packet_share_rules') + || sql.startsWith('UPDATE owner_packet_share_deliveries') + || sql.startsWith('INSERT INTO owner_packet_share_rules')) { + writes.push(`${sql}\n${JSON.stringify(params ?? [])}`); + return sql.startsWith('INSERT INTO owner_packet_share_rules') + ? { rows: [{ destination_id: 'remote-site' }] } + : { rows: [] }; + } + if (sql.includes('FROM owner_packet_share_destinations')) { + return { rows: [destination] }; + } + if (sql.includes('FROM owner_packet_share_rules')) { + return { rows: selectedDestinationRows }; + } + return query(sql, params); + }, 'node1', requireOwnerSession)); + const app = express(); + app.use(express.json()); + app.use('/api', router); + const server = app.listen(0, '127.0.0.1'); + await once(server, 'listening'); + try { + await check(`http://127.0.0.1:${(server.address() as AddressInfo).port}`, writes); + } finally { + server.closeAllConnections(); + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +} + +test('owner packet-sharing settings start with every destination off', async () => { + await withOwnerApi(async () => ({ rows: [] }), async (baseUrl) => { + const response = await fetch(`${baseUrl}/api/owner/packet-sharing?nodeId=${nodeId}`); + assert.equal(response.status, 200); + const body = await response.json(); + assert.equal(body.enabled, false); + assert.equal(body.destinations[0]?.selected, false); + }, []); +}); + +test('owner packet-sharing settings expose destinations and persist a selected destination', async () => { + const lastForwardedAt = '2026-09-16T22:45:00.000Z'; + await withOwnerApi(async (sql) => ( + sql.includes('FROM owner_packet_share_deliveries') + ? { rows: [{ destination_id: 'remote-site', last_forwarded_at: lastForwardedAt }] } + : { rows: [] } + ), async (baseUrl, writes) => { + const getResponse = await fetch(`${baseUrl}/api/owner/packet-sharing?nodeId=${nodeId}`); + assert.equal(getResponse.status, 200); + assert.equal(getResponse.headers.get('cache-control'), 'no-store'); + assert.deepEqual(await getResponse.json(), { + nodeId, + featureEnabled: true, + enabled: true, + destinations: [{ + id: 'remote-site', + name: 'Remote site', + websiteUrl: 'https://remote.example.test', + description: 'A remote packet site', + configured: true, + selected: true, + lastForwardedAt, + }], + }); + + const postResponse = await fetch(`${baseUrl}/api/owner/packet-sharing`, { + method: 'POST', + headers: { + 'content-type': 'application/json', + cookie: 'meshcore_owner_csrf=test-token', + 'x-csrf-token': 'test-token', + }, + body: JSON.stringify({ nodeId: nodeId.toLowerCase(), enabled: true, destinationIds: ['remote-site'] }), + }); + assert.equal(postResponse.status, 200); + assert.deepEqual(await postResponse.json(), { nodeId, enabled: true, destinationIds: ['remote-site'] }); + assert.equal(writes.length, 3); + + const disableResponse = await fetch(`${baseUrl}/api/owner/packet-sharing`, { + method: 'POST', + headers: { + 'content-type': 'application/json', + cookie: 'meshcore_owner_csrf=test-token', + 'x-csrf-token': 'test-token', + }, + body: JSON.stringify({ nodeId, enabled: false, destinationIds: [] }), + }); + assert.equal(disableResponse.status, 200); + assert.deepEqual(await disableResponse.json(), { nodeId, enabled: false, destinationIds: [] }); + assert.equal(writes.length, 5); + }); +}); + +test('owner packet-sharing rejects a destination that is not in the operator registry', async () => { + await withOwnerApi(async () => ({ rows: [] }), async (baseUrl) => { + const response = await fetch(`${baseUrl}/api/owner/packet-sharing`, { + method: 'POST', + headers: { + 'content-type': 'application/json', + cookie: 'meshcore_owner_csrf=test-token', + 'x-csrf-token': 'test-token', + }, + body: JSON.stringify({ nodeId, enabled: true, destinationIds: ['unknown-site'] }), + }); + assert.equal(response.status, 409); + assert.deepEqual(await response.json(), { error: 'One or more selected destinations are not configured' }); + }); +}); + +test('owner packet-sharing settings are available to every authenticated owner account', async () => { + const router = Router(); + registerOwnerRoutes(router, ownerRouteDeps(async () => ({ rows: [] }), 'node2')); + const app = express(); + app.use(express.json()); + app.use('/api', router); + const server = app.listen(0, '127.0.0.1'); + await once(server, 'listening'); + try { + const response = await fetch(`http://127.0.0.1:${(server.address() as AddressInfo).port}/api/owner/packet-sharing`); + assert.equal(response.status, 200); + assert.deepEqual(await response.json(), { + nodeId, + featureEnabled: true, + enabled: false, + destinations: [], + }); + } finally { + server.closeAllConnections(); + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +}); + +test('owner packet-sharing rejects nodes outside the authenticated session', async () => { + await withOwnerApi(async () => ({ rows: [] }), async (baseUrl, writes) => { + const response = await fetch(`${baseUrl}/api/owner/packet-sharing?nodeId=${'B'.repeat(64)}`); + assert.equal(response.status, 403); + assert.deepEqual(await response.json(), { error: 'Node is not owned by this session' }); + assert.equal(writes.length, 0); + }); +}); + +test('owner packet-sharing requires an authenticated session', async () => { + await withOwnerApi(async () => ({ rows: [] }), async (baseUrl) => { + const response = await fetch(`${baseUrl}/api/owner/packet-sharing`); + assert.equal(response.status, 401); + assert.deepEqual(await response.json(), { error: 'Unauthorized' }); + }, [{ destination_id: 'remote-site' }], async (_req, res) => { + res.status(401).json({ error: 'Unauthorized' }); + return null; + }); +}); + +test('owner packet-sharing rejects malformed settings without database writes', async () => { + await withOwnerApi(async () => ({ rows: [] }), async (baseUrl, writes) => { + const response = await fetch(`${baseUrl}/api/owner/packet-sharing`, { + method: 'POST', + headers: { + 'content-type': 'application/json', + cookie: 'meshcore_owner_csrf=test-token', + 'x-csrf-token': 'test-token', + }, + body: JSON.stringify({ nodeId, enabled: true, destinationIds: ['remote-site'], unexpected: true }), + }); + assert.equal(response.status, 400); + assert.deepEqual(await response.json(), { error: 'Invalid packet sharing settings' }); + assert.equal(writes.length, 0); + }); +}); diff --git a/backend/src/db/dataLifecycle.test.ts b/backend/src/db/dataLifecycle.test.ts index fb5bcac..f9a61d6 100644 --- a/backend/src/db/dataLifecycle.test.ts +++ b/backend/src/db/dataLifecycle.test.ts @@ -50,6 +50,7 @@ function receiptEnvironment(restoreVerifiedAt = '2026-07-28T12:00:00.000Z') { test('data lifecycle targets are closed and destructive gates fail closed', () => { assert.equal(lifecyclePolicy('packets').retention, '30 days'); + assert.equal(lifecyclePolicy('owner_packet_share_deliveries').retention, '30 days'); assert.equal(lifecyclePolicy('node_neighbor_samples').retention, '7 days'); assert.equal(compressionPolicy('packets').compressAfter, '14 days'); assert.equal(compressionPolicy('packet_paths').compressAfter, '14 days'); diff --git a/backend/src/db/dataLifecycle.ts b/backend/src/db/dataLifecycle.ts index 9822fc0..d968008 100644 --- a/backend/src/db/dataLifecycle.ts +++ b/backend/src/db/dataLifecycle.ts @@ -106,6 +106,13 @@ export const DATA_LIFECYCLE_POLICIES: readonly DataLifecyclePolicy[] = Object.fr kind: 'row-table', featureImpact: ['delivered owner-alert audit rows older than 90 days are removed'], }, + { + table: 'owner_packet_share_deliveries', + timestampColumn: 'created_at', + retention: '30 days', + kind: 'row-table', + featureImpact: ['terminal owner packet-sharing queue and audit rows older than 30 days are removed'], + }, { table: 'observer_registration_requests', timestampColumn: 'updated_at', diff --git a/backend/src/db/index.ts b/backend/src/db/index.ts index cc0339d..bdb5878 100644 --- a/backend/src/db/index.ts +++ b/backend/src/db/index.ts @@ -157,6 +157,33 @@ export async function query( return queryPool(pool, 'oltp', text, params, signal); } +export type DatabaseQueryFn = ( + text: string, + params?: unknown[], +) => Promise<{ rows: T[] }>; + +export async function withTransaction(work: (transactionQuery: DatabaseQueryFn) => Promise): Promise { + const client = await pool.connect(); + const transactionQuery: DatabaseQueryFn = async ( + text: string, + params?: unknown[], + ) => { + const result = await client.query(text, params); + return { rows: result.rows }; + }; + try { + await client.query('BEGIN'); + const result = await work(transactionQuery); + await client.query('COMMIT'); + return result; + } catch (error) { + await client.query('ROLLBACK').catch(() => undefined); + throw error; + } finally { + client.release(); + } +} + export async function namedQuery( name: string, text: string, diff --git a/backend/src/db/migrations/052_owner_packet_sharing.sql b/backend/src/db/migrations/052_owner_packet_sharing.sql new file mode 100644 index 0000000..d68571b --- /dev/null +++ b/backend/src/db/migrations/052_owner_packet_sharing.sql @@ -0,0 +1,66 @@ +-- Owner-controlled packet replication. Destination rows are operator-managed; +-- owners can only select destinations that already exist here. + +CREATE TABLE IF NOT EXISTS owner_packet_share_destinations ( + destination_id TEXT PRIMARY KEY, + display_name TEXT NOT NULL CHECK (char_length(display_name) BETWEEN 1 AND 120), + broker_url TEXT, + topic_template TEXT, + broker_username TEXT, + broker_password_ciphertext TEXT, + enabled BOOLEAN NOT NULL DEFAULT FALSE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CHECK (destination_id ~ '^[a-z0-9][a-z0-9_-]{0,63}$'), + CHECK (broker_url IS NULL OR char_length(broker_url) <= 2_048), + CHECK (topic_template IS NULL OR char_length(topic_template) BETWEEN 1 AND 512), + CHECK (broker_username IS NULL OR char_length(broker_username) BETWEEN 1 AND 128), + CHECK (broker_password_ciphertext IS NULL OR char_length(broker_password_ciphertext) BETWEEN 1 AND 2_048) +); + +CREATE TABLE IF NOT EXISTS owner_packet_share_rules ( + owner_username TEXT NOT NULL, + node_id TEXT NOT NULL CHECK (node_id ~ '^[0-9A-F]{64}$'), + destination_id TEXT NOT NULL REFERENCES owner_packet_share_destinations(destination_id) ON DELETE RESTRICT, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (owner_username, node_id, destination_id) +); + +CREATE INDEX IF NOT EXISTS owner_packet_share_rules_node_idx + ON owner_packet_share_rules (node_id, enabled, destination_id); + +CREATE TABLE IF NOT EXISTS owner_packet_share_deliveries ( + id BIGSERIAL PRIMARY KEY, + owner_username TEXT NOT NULL, + node_id TEXT NOT NULL CHECK (node_id ~ '^[0-9A-F]{64}$'), + destination_id TEXT NOT NULL REFERENCES owner_packet_share_destinations(destination_id) ON DELETE RESTRICT, + event_key TEXT NOT NULL UNIQUE, + packet_hash TEXT NOT NULL, + source_topic TEXT NOT NULL, + destination_topic TEXT NOT NULL, + payload JSONB NOT NULL, + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'delivering', 'succeeded', 'failed', 'dead_lettered')), + attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0 AND attempts <= 5), + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + last_attempt_at TIMESTAMPTZ, + delivered_at TIMESTAMPTZ, + last_error TEXT, + claim_token TEXT, + claim_expires_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS owner_packet_share_deliveries_due_idx + ON owner_packet_share_deliveries (next_attempt_at, id) + WHERE status IN ('pending', 'failed') AND attempts < 5; + +CREATE INDEX IF NOT EXISTS owner_packet_share_deliveries_destination_idx + ON owner_packet_share_deliveries (destination_id, created_at DESC); + +CREATE INDEX IF NOT EXISTS owner_packet_share_deliveries_retention_idx + ON owner_packet_share_deliveries (created_at, id) + WHERE status IN ('succeeded', 'dead_lettered'); diff --git a/backend/src/db/migrations/053_owner_packet_share_destination_metadata.sql b/backend/src/db/migrations/053_owner_packet_share_destination_metadata.sql new file mode 100644 index 0000000..646ad79 --- /dev/null +++ b/backend/src/db/migrations/053_owner_packet_share_destination_metadata.sql @@ -0,0 +1,28 @@ +ALTER TABLE owner_packet_share_destinations + ADD COLUMN IF NOT EXISTS website_url TEXT, + ADD COLUMN IF NOT EXISTS description TEXT; + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conrelid = 'owner_packet_share_destinations'::regclass + AND conname = 'owner_packet_share_destinations_website_url_length' + ) THEN + ALTER TABLE owner_packet_share_destinations + ADD CONSTRAINT owner_packet_share_destinations_website_url_length + CHECK (website_url IS NULL OR char_length(website_url) BETWEEN 1 AND 2_048); + END IF; + + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conrelid = 'owner_packet_share_destinations'::regclass + AND conname = 'owner_packet_share_destinations_description_length' + ) THEN + ALTER TABLE owner_packet_share_destinations + ADD CONSTRAINT owner_packet_share_destinations_description_length + CHECK (description IS NULL OR char_length(description) BETWEEN 1 AND 160); + END IF; +END $$; diff --git a/backend/src/db/migrations/054_owner_packet_share_default_off.sql b/backend/src/db/migrations/054_owner_packet_share_default_off.sql new file mode 100644 index 0000000..1c4ac8d --- /dev/null +++ b/backend/src/db/migrations/054_owner_packet_share_default_off.sql @@ -0,0 +1,5 @@ +-- Packet-sharing rules require an explicit owner selection. +-- Existing rows are intentionally preserved so active test selections keep working. + +ALTER TABLE owner_packet_share_rules + ALTER COLUMN enabled SET DEFAULT FALSE; diff --git a/backend/src/db/packetBatch.integration.test.ts b/backend/src/db/packetBatch.integration.test.ts index 37431a6..ec49e6c 100644 --- a/backend/src/db/packetBatch.integration.test.ts +++ b/backend/src/db/packetBatch.integration.test.ts @@ -103,6 +103,9 @@ test('base schema and every migration apply to a brand-new database', { assert.deepEqual(executed, [ '050_packet_paths_and_retention.sql', '051_reset_readiness.sql', + '052_owner_packet_sharing.sql', + '053_owner_packet_share_destination_metadata.sql', + '054_owner_packet_share_default_off.sql', '056_async_privacy_rematerialization.sql', ]); assert.deepEqual(await runMigrations(freshPool), []); diff --git a/backend/src/index.ts b/backend/src/index.ts index e760eda..c467321 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -50,6 +50,8 @@ import { createGlobalApiLimiter } from './api/bootstrap/limiters.js'; import { createHopReachCompatibilityRoutes } from './api/hopreachCompatibility.js'; import { readAclFile } from './mqtt/aclManager.js'; import { parseOwnerGrantConfig } from './owner/ownerGrantConfig.js'; +import { createPacketShareForwarder } from './owner/packetShareForwarder.js'; +import { createPacketShareRepository } from './owner/packetShareRepository.js'; import { buildOwnerInventoryBaseline, loadOwnerInventoryBaseline, @@ -74,6 +76,9 @@ const SHUTDOWN_DEADLINE_MS = Math.min( ); const NODE_IDENTITY_REFRESH_INTERVAL_MS = 30 * 60_000; const lifecycle = new LifecycleCoordinator(SHUTDOWN_DEADLINE_MS); +const packetShareForwarder = createPacketShareForwarder({ + repository: createPacketShareRepository(query), +}); let shutdownExitCode = 0; let forceCloseHttpConnections = () => {}; let nodeIdentityRefreshTimer: NodeJS.Timeout | null = null; @@ -113,6 +118,11 @@ lifecycle.register({ stage: 30, close: closeQueuePublisher, }); +lifecycle.register({ + name: 'owner-packet-share-forwarder', + stage: 30, + close: packetShareForwarder.stop, +}); lifecycle.register({ name: 'operator-operations', stage: 30, @@ -203,8 +213,13 @@ async function main() { } // 3. Wire up MQTT → WS broadcast - onPacket((packet) => { + onPacket((packet, upstreamEnvelope) => { broadcastPacket(packet); + if (upstreamEnvelope) { + void packetShareForwarder.enqueuePacket(packet, upstreamEnvelope).catch((error: unknown) => { + console.error('[packet-share] packet handling failed:', error instanceof Error ? error.message : error); + }); + } if (packet.path?.length && packet.rxNodeId) { void queueLinkJob( packet.packetHash, @@ -404,6 +419,7 @@ async function main() { }); startMetricsServer(METRICS_PORT); + packetShareForwarder.start(); if (MQTT_INGEST_ENABLED) { void startMqttClient().catch((err: unknown) => { console.error('[mqtt] failed to start client:', err instanceof Error ? err.message : err); diff --git a/backend/src/metrics.ts b/backend/src/metrics.ts index cc55a30..9a0c844 100644 --- a/backend/src/metrics.ts +++ b/backend/src/metrics.ts @@ -129,6 +129,19 @@ export const mqttIngestActive = new Gauge({ registers: [metricsRegistry], }); +export const packetShareOutcomesTotal = new Counter({ + name: 'meshcore_owner_packet_share_outcomes_total', + help: 'Owner packet sharing decisions and delivery outcomes.', + labelNames: ['outcome'] as const, + registers: [metricsRegistry], +}); + +export const packetShareQueueDepth = new Gauge({ + name: 'meshcore_owner_packet_share_queue_depth', + help: 'Owner packet sharing deliveries currently claimed for processing.', + registers: [metricsRegistry], +}); + export const dbQueriesTotal = new Counter({ name: 'meshcore_db_queries_total', help: 'Database queries by pool and bounded outcome.', diff --git a/backend/src/mqtt/client.ts b/backend/src/mqtt/client.ts index 5f1e04d..7d9cfd1 100644 --- a/backend/src/mqtt/client.ts +++ b/backend/src/mqtt/client.ts @@ -21,7 +21,7 @@ import { mqttMessagesTotal, } from '../metrics.js'; -type PacketCallback = (packet: LivePacket) => void; +type PacketCallback = (packet: LivePacket, upstreamEnvelope?: Record) => void; type NodeCallback = (nodeId: string, meta?: { network?: string; observerId?: string }) => void; type NodeUpsertCallback = (node: Record) => void; @@ -62,8 +62,8 @@ export function onPacket(cb: PacketCallback) { subscribers.push(cb); } export function onNodeSeen(cb: NodeCallback) { nodeSubscribers.push(cb); } export function onNodeUpsert(cb: NodeUpsertCallback) { upsertSubscribers.push(cb); } -function emit(packet: LivePacket) { - for (const cb of subscribers) cb(packet); +function emit(packet: LivePacket, upstreamEnvelope?: Record) { + for (const cb of subscribers) cb(packet, upstreamEnvelope); } // Debounced node update batching — 500 ms window @@ -885,7 +885,7 @@ async function handleMessage(topic: string, rawPayload: Buffer): Promise { livePacket.isPrivate = visibility.isPrivate; livePacket.visibilityOk = visibility.visibilityOk; try { - emit(livePacket); + emit(livePacket, json); } catch (error) { console.error( '[mqtt] packet subscriber failed after durable ingest:', diff --git a/backend/src/owner/packetShareConfig.test.ts b/backend/src/owner/packetShareConfig.test.ts new file mode 100644 index 0000000..2377750 --- /dev/null +++ b/backend/src/owner/packetShareConfig.test.ts @@ -0,0 +1,113 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + buildPacketSharePayload, + destinationConfigured, + isValidPacketShareWebsiteUrl, + isForwardablePacket, + renderPacketShareTopic, +} from './packetShareConfig.js'; +import { decryptPacketShareSecret, encryptPacketShareSecret } from './packetShareCrypto.js'; +import type { LivePacket } from '../types/index.js'; + +const packet: LivePacket = { + id: 'event-id', + packetHash: 'B'.repeat(16), + rxNodeId: 'A'.repeat(64), + topic: `meshcore/LHR/${'A'.repeat(64)}/packets`, + iata: 'LHR', + network: 'ukmesh', + packetType: 5, + visibilityOk: true, + ts: Date.parse('2026-09-16T12:00:00Z'), +}; + +test('packet share topic rendering is bounded and supports only known values', () => { + assert.equal( + renderPacketShareTopic('replicated/{network}/{iata}/{observerId}/packets', packet), + `replicated/ukmesh/LHR/${'A'.repeat(64)}/packets`, + ); + assert.throws(() => renderPacketShareTopic('replicated/+/packets', packet)); + assert.throws(() => renderPacketShareTopic('replicated//packets', packet)); +}); + +test('forward payload matches the flat mctomqtt packet shape and excludes decoded content', () => { + const payload = buildPacketSharePayload(packet, { + hash: 'upstream-hash', + origin: 'Test observer', + type: 'PACKET', + direction: 'rx', + len: '2', + packet_type: '5', + route: 'F', + payload_len: '1', + raw: '0102', + origin_id: packet.rxNodeId, + SNR: '12.2', + RSSI: '-30', + score: '1000', + timestamp: '2026-09-16T12:00:00.000Z', + time: '12:00:00', + date: '16/9/2026', + body: 'should not be copied', + secret: 'should not be copied', + }); + assert.deepEqual(payload, { + hash: 'upstream-hash', + origin: 'Test observer', + type: 'PACKET', + direction: 'rx', + len: '2', + packet_type: '5', + route: 'F', + payload_len: '1', + raw: '0102', + origin_id: packet.rxNodeId, + SNR: '12.2', + RSSI: '-30', + score: '1000', + timestamp: '2026-09-16T12:00:00.000Z', + time: '12:00:00', + date: '16/09/2026', + }); + assert.equal('body' in payload, false); + assert.equal('secret' in payload, false); +}); + +test('only public validated packets are forwardable', () => { + assert.equal(isForwardablePacket(packet), true); + assert.equal(isForwardablePacket({ ...packet, visibilityOk: false }), false); + assert.equal(isForwardablePacket({ ...packet, network: 'test' }), false); + assert.equal(isForwardablePacket({ ...packet, rxNodeId: 'short' }), false); +}); + +test('destination configuration requires complete broker details', () => { + const base = { + enabled: true, + broker_url: 'mqtts://remote.example.test:8883', + topic_template: 'replicated/{network}/{observerId}/packets', + broker_username: null, + broker_password_ciphertext: null, + }; + assert.equal(destinationConfigured(base), true); + assert.equal(destinationConfigured({ ...base, broker_url: null }), false); + assert.equal(destinationConfigured({ ...base, broker_username: 'user' }), false); + assert.equal(destinationConfigured({ ...base, broker_password_ciphertext: 'v1.invalid' }, false), false); + assert.equal(destinationConfigured({ ...base, broker_username: 'user', broker_password_ciphertext: 'v1.invalid' }), false); + assert.equal(destinationConfigured({ ...base, broker_url: 'https://remote.example.test' }), false); + assert.equal(destinationConfigured({ ...base, broker_url: 'mqtts://user:pass@remote.example.test:8883' }), false); + assert.equal(destinationConfigured({ ...base, topic_template: 'replicated/{unknown}/packets' }), false); +}); + +test('destination metadata accepts only safe HTTPS website links', () => { + assert.equal(isValidPacketShareWebsiteUrl('https://remote.example.test'), true); + assert.equal(isValidPacketShareWebsiteUrl('http://remote.example.test'), false); + assert.equal(isValidPacketShareWebsiteUrl('https://user:pass@remote.example.test'), false); + assert.equal(isValidPacketShareWebsiteUrl(null), false); +}); + +test('packet share secrets round-trip with authenticated encryption', () => { + const encrypted = encryptPacketShareSecret('broker-password', 'test-encryption-key'); + assert.equal(decryptPacketShareSecret(encrypted, 'test-encryption-key'), 'broker-password'); + assert.throws(() => decryptPacketShareSecret(encrypted, 'wrong-encryption-key')); +}); diff --git a/backend/src/owner/packetShareConfig.ts b/backend/src/owner/packetShareConfig.ts new file mode 100644 index 0000000..5dad676 --- /dev/null +++ b/backend/src/owner/packetShareConfig.ts @@ -0,0 +1,197 @@ +import type { LivePacket } from '../types/index.js'; +import { isPacketShareSecretCiphertext } from './packetShareCrypto.js'; + +const MAX_TOPIC_LENGTH = 512; +const MAX_PAYLOAD_BYTES = 64 * 1024; +const PACKET_SHARE_PROTOCOLS = new Set(['mqtt:', 'mqtts:', 'ws:', 'wss:']); +const PACKET_SHARE_TOPIC_PLACEHOLDERS = /\{(network|iata|observerId|packetHash)\}/g; + +export type PacketShareDestinationRow = { + destination_id: string; + display_name: string; + website_url?: string | null; + description?: string | null; + broker_url: string | null; + topic_template: string | null; + broker_username: string | null; + broker_password_ciphertext: string | null; + enabled: boolean; +}; + +export type PacketShareDestinationView = { + id: string; + name: string; + websiteUrl: string | null; + description: string | null; + configured: boolean; + selected: boolean; + lastForwardedAt: string | null; +}; + +export type PacketShareTarget = PacketShareDestinationRow & { + owner_username: string; + node_id: string; +}; + +export type PacketShareForwardPayload = Record; + +function packetTopicValue(packet: LivePacket): Record { + return { + network: String(packet.network ?? 'ukmesh').trim().toLowerCase() || 'ukmesh', + iata: String(packet.iata ?? 'UNK').trim().toUpperCase() || 'UNK', + observerId: String(packet.rxNodeId ?? '').trim().toUpperCase(), + packetHash: packet.packetHash, + }; +} + +export function isValidPacketShareBrokerUrl(rawUrl: string | null): boolean { + if (!rawUrl?.trim()) return false; + try { + const url = new URL(rawUrl.trim()); + return PACKET_SHARE_PROTOCOLS.has(url.protocol) + && Boolean(url.hostname) + && !url.username + && !url.password; + } catch { + return false; + } +} + +export function isValidPacketShareWebsiteUrl(rawUrl: string | null | undefined): boolean { + if (!rawUrl?.trim()) return false; + try { + const url = new URL(rawUrl.trim()); + return url.protocol === 'https:' + && Boolean(url.hostname) + && !url.username + && !url.password; + } catch { + return false; + } +} + +export function isValidPacketShareTopicTemplate(template: string | null): boolean { + if (!template || !template.trim() || template.length > MAX_TOPIC_LENGTH) return false; + if (/[\u0000-\u001f\u007f]/.test(template) || template.includes('#') || template.includes('+')) return false; + const withPlaceholdersReplaced = template.replace(PACKET_SHARE_TOPIC_PLACEHOLDERS, 'value'); + if (withPlaceholdersReplaced.includes('{') || withPlaceholdersReplaced.includes('}')) return false; + return withPlaceholdersReplaced.split('/').every((part) => part.length > 0); +} + +export function destinationConfigured( + destination: Pick, + encryptionKeyAvailable = true, +): boolean { + if (!destination.enabled + || !isValidPacketShareBrokerUrl(destination.broker_url) + || !isValidPacketShareTopicTemplate(destination.topic_template)) return false; + const username = destination.broker_username?.trim() || ''; + const password = destination.broker_password_ciphertext?.trim() || ''; + if (Boolean(username) !== Boolean(password)) return false; + if (password && (!encryptionKeyAvailable || !isPacketShareSecretCiphertext(password))) return false; + return true; +} + +export function renderPacketShareTopic(template: string, packet: LivePacket): string { + if (!isValidPacketShareTopicTemplate(template)) { + throw new Error('invalid packet share destination topic'); + } + const values = packetTopicValue(packet); + const rendered = template.replace(/\{(network|iata|observerId|packetHash)\}/g, (_, key: keyof typeof values) => values[key]); + if ( + rendered.length < 1 + || rendered.length > MAX_TOPIC_LENGTH + || rendered.includes('#') + || rendered.includes('+') + || /[\u0000-\u001f\u007f]/.test(rendered) + || rendered.split('/').some((part) => part.length < 1) + ) { + throw new Error('invalid packet share destination topic'); + } + return rendered; +} + +function copyEnvelopeValue(target: PacketShareForwardPayload, source: Record, key: string): void { + const value = source[key]; + if (typeof value === 'string' || typeof value === 'number' || typeof value === 'boolean') { + target[key] = String(value); + } +} + +function padDatePart(value: number): string { + return String(value).padStart(2, '0'); +} + +function packetDateParts(packet: LivePacket, timestamp: unknown): { time: string; date: string } { + const candidate = typeof timestamp === 'string' || typeof timestamp === 'number' + ? new Date(timestamp) + : new Date(packet.ts); + const date = Number.isFinite(candidate.getTime()) ? candidate : new Date(packet.ts); + if (!Number.isFinite(date.getTime())) return { time: '', date: '' }; + return { + time: `${padDatePart(date.getUTCHours())}:${padDatePart(date.getUTCMinutes())}:${padDatePart(date.getUTCSeconds())}`, + date: `${padDatePart(date.getUTCDate())}/${padDatePart(date.getUTCMonth() + 1)}/${date.getUTCFullYear()}`, + }; +} + +/** Build a bounded flat mctomqtt-compatible packet without forwarding decoded content. */ +export function buildPacketSharePayload( + packet: LivePacket, + upstreamEnvelope: Record, +): PacketShareForwardPayload { + const envelope: PacketShareForwardPayload = {}; + for (const key of [ + 'raw', 'hash', 'packet_type', 'SNR', 'RSSI', 'score', 'route', 'len', + 'payload_len', 'direction', 'origin', 'origin_id', 'timestamp', 'type', 'time', 'date', + ]) { + copyEnvelopeValue(envelope, upstreamEnvelope, key); + } + + if (!('timestamp' in envelope)) envelope.timestamp = new Date(packet.ts).toISOString(); + if (!('hash' in envelope)) envelope.hash = packet.packetHash; + if (!('type' in envelope)) envelope.type = 'PACKET'; + if (!('direction' in envelope)) envelope.direction = 'rx'; + if (!('origin_id' in envelope) && packet.rxNodeId) envelope.origin_id = packet.rxNodeId; + if (!('packet_type' in envelope) && packet.packetType != null) envelope.packet_type = String(packet.packetType); + if (!('len' in envelope) && typeof envelope.raw === 'string') envelope.len = String(Math.floor(envelope.raw.length / 2)); + + const dateParts = packetDateParts(packet, envelope.timestamp); + if (dateParts.time) envelope.time = dateParts.time; + if (dateParts.date) envelope.date = dateParts.date; + + if (Buffer.byteLength(JSON.stringify(envelope), 'utf8') > MAX_PAYLOAD_BYTES) { + throw new Error('packet share payload exceeds configured limit'); + } + return envelope; +} + +export function isForwardablePacket(packet: LivePacket): boolean { + return packet.visibilityOk === true + && packet.network !== 'test' + && typeof packet.rxNodeId === 'string' + && /^[0-9A-F]{64}$/.test(packet.rxNodeId) + && /^[0-9A-F]{16,128}$/.test(packet.packetHash); +} + +export function packetShareEventKey(ownerUsername: string, destinationId: string, packet: LivePacket): string { + return `${ownerUsername}:${destinationId}:${packet.rxNodeId}:${packet.packetHash}`; +} + +export function publicDestinationView( + destination: PacketShareDestinationRow, + selected: boolean, + encryptionKeyAvailable: boolean, + lastForwardedAt: string | null = null, +): PacketShareDestinationView { + return { + id: destination.destination_id, + name: destination.display_name, + websiteUrl: isValidPacketShareWebsiteUrl(destination.website_url) + ? destination.website_url!.trim() + : null, + description: destination.description?.trim() || null, + configured: destinationConfigured(destination, encryptionKeyAvailable), + selected, + lastForwardedAt, + }; +} diff --git a/backend/src/owner/packetShareCrypto.ts b/backend/src/owner/packetShareCrypto.ts new file mode 100644 index 0000000..6c88ec8 --- /dev/null +++ b/backend/src/owner/packetShareCrypto.ts @@ -0,0 +1,60 @@ +import { createCipheriv, createDecipheriv, createHash, randomBytes } from 'node:crypto'; + +const CIPHERTEXT_VERSION = 'v1'; + +function keyFromSecret(secret: string): Buffer { + const normalized = secret.trim(); + if (normalized.length < 16) throw new Error('OWNER_PACKET_SHARE_ENCRYPTION_KEY is too short'); + return createHash('sha256').update(normalized).digest(); +} + +function configuredKey(): Buffer { + const secret = String(process.env['OWNER_PACKET_SHARE_ENCRYPTION_KEY'] ?? ''); + if (!secret) throw new Error('OWNER_PACKET_SHARE_ENCRYPTION_KEY is not set'); + return keyFromSecret(secret); +} + +/** Encrypt a destination broker password for storage in the primary database. */ +export function encryptPacketShareSecret(secret: string, keySecret?: string): string { + if (!secret) throw new Error('packet share secret is empty'); + const iv = randomBytes(12); + const cipher = createCipheriv('aes-256-gcm', keySecret ? keyFromSecret(keySecret) : configuredKey(), iv); + const ciphertext = Buffer.concat([cipher.update(secret, 'utf8'), cipher.final()]); + const tag = cipher.getAuthTag(); + return [ + CIPHERTEXT_VERSION, + iv.toString('base64url'), + tag.toString('base64url'), + ciphertext.toString('base64url'), + ].join('.'); +} + +/** Decrypt a destination broker password; malformed or unauthenticated data fails closed. */ +export function decryptPacketShareSecret(ciphertext: string, keySecret?: string): string { + const [version, ivB64, tagB64, bodyB64] = ciphertext.split('.'); + if (version !== CIPHERTEXT_VERSION || !ivB64 || !tagB64 || !bodyB64) { + throw new Error('invalid packet share secret format'); + } + const iv = Buffer.from(ivB64, 'base64url'); + const tag = Buffer.from(tagB64, 'base64url'); + const body = Buffer.from(bodyB64, 'base64url'); + if (iv.length !== 12 || tag.length !== 16 || body.length < 1) { + throw new Error('invalid packet share secret encoding'); + } + const decipher = createDecipheriv('aes-256-gcm', keySecret ? keyFromSecret(keySecret) : configuredKey(), iv); + decipher.setAuthTag(tag); + return Buffer.concat([decipher.update(body), decipher.final()]).toString('utf8'); +} + +export function isPacketShareSecretCiphertext(ciphertext: string): boolean { + const [version, ivB64, tagB64, bodyB64] = ciphertext.split('.'); + if (version !== CIPHERTEXT_VERSION || !ivB64 || !tagB64 || !bodyB64) return false; + if (![ivB64, tagB64, bodyB64].every((value) => /^[A-Za-z0-9_-]+$/.test(value))) return false; + return Buffer.from(ivB64, 'base64url').length === 12 + && Buffer.from(tagB64, 'base64url').length === 16 + && Buffer.from(bodyB64, 'base64url').length > 0; +} + +export function hasPacketShareEncryptionKey(): boolean { + return String(process.env['OWNER_PACKET_SHARE_ENCRYPTION_KEY'] ?? '').trim().length >= 16; +} diff --git a/backend/src/owner/packetShareForwarder.test.ts b/backend/src/owner/packetShareForwarder.test.ts new file mode 100644 index 0000000..25b2abb --- /dev/null +++ b/backend/src/owner/packetShareForwarder.test.ts @@ -0,0 +1,99 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import type { LivePacket } from '../types/index.js'; +import type { PacketSharePublisher } from './packetShareForwarder.js'; +import { createPacketShareForwarder } from './packetShareForwarder.js'; +import type { PacketShareRepository } from './packetShareRepository.js'; + +const packet: LivePacket = { + id: 'packet-event', + packetHash: 'B'.repeat(16), + rxNodeId: 'A'.repeat(64), + topic: `meshcore/LHR/${'A'.repeat(64)}/packets`, + iata: 'LHR', + network: 'ukmesh', + packetType: 5, + visibilityOk: true, + ts: Date.parse('2026-09-16T12:00:00Z'), +}; + +test('forwarder queues one selected owner packet and publishes it through the MQTT seam', async () => { + const destination = { + destination_id: 'remote-site', + display_name: 'Remote site', + broker_url: 'mqtts://remote.example.test:8883', + topic_template: 'replicated/{network}/{observerId}/packets', + broker_username: null, + broker_password_ciphertext: null, + enabled: true, + }; + const enqueued: Array> = []; + const published: Array<{ topic: string; payload: string }> = []; + let claimed = false; + const repository = { + activeTargets: async () => [{ ...destination, owner_username: 'node1', node_id: packet.rxNodeId! }], + enqueueDelivery: async (input: Record) => { + enqueued.push(input); + return true; + }, + reclaimExpiredClaims: async () => 0, + claimDue: async (_limit: number, _claimToken: string) => { + if (claimed || enqueued.length < 1) return []; + claimed = true; + const row = enqueued[0]!; + return [{ + id: '1', + owner_username: 'node1', + node_id: packet.rxNodeId!, + destination_id: 'remote-site', + event_key: String(row['eventKey']), + packet_hash: packet.packetHash, + source_topic: packet.topic, + destination_topic: String(row['destinationTopic']), + payload: row['payload'] as Record, + attempts: 1, + }]; + }, + getDestination: async () => destination, + markSucceeded: async () => undefined, + markFailed: async () => undefined, + } as unknown as PacketShareRepository; + const publisher: PacketSharePublisher = { + publish: async (topic, payload) => { published.push({ topic, payload }); }, + close: async () => undefined, + }; + const forwarder = createPacketShareForwarder({ + repository, + publisherFactory: () => publisher, + }); + + await forwarder.enqueuePacket(packet, { raw: '0102', packet_type: 5, body: 'private decoded text' }); + await forwarder.processDue(); + await forwarder.stop(); + + assert.equal(enqueued.length, 1); + assert.equal(published.length, 1); + assert.equal(published[0]?.topic, `replicated/ukmesh/${'A'.repeat(64)}/packets`); + const outbound = JSON.parse(published[0]!.payload) as Record; + assert.equal(enqueued[0]?.['eventKey'], `node1:remote-site:${packet.rxNodeId}:${packet.packetHash}`); + assert.equal(outbound['raw'], '0102'); + assert.equal(outbound['packet_type'], '5'); + assert.equal(outbound['timestamp'], '2026-09-16T12:00:00.000Z'); + assert.equal(outbound['time'], '12:00:00'); + assert.equal(outbound['date'], '16/09/2026'); + assert.equal('relay' in outbound, false); + assert.equal('packet' in outbound, false); + assert.equal('version' in outbound, false); + assert.equal('body' in outbound, false); +}); + +test('forwarder does not queue a privacy-rejected packet', async () => { + let targets = 0; + const repository = { + activeTargets: async () => { targets += 1; return []; }, + } as unknown as PacketShareRepository; + const forwarder = createPacketShareForwarder({ repository }); + await forwarder.enqueuePacket({ ...packet, visibilityOk: false }, { raw: '0102' }); + await forwarder.stop(); + assert.equal(targets, 0); +}); diff --git a/backend/src/owner/packetShareForwarder.ts b/backend/src/owner/packetShareForwarder.ts new file mode 100644 index 0000000..6251233 --- /dev/null +++ b/backend/src/owner/packetShareForwarder.ts @@ -0,0 +1,270 @@ +import mqtt, { type MqttClient } from 'mqtt'; +import { randomUUID } from 'node:crypto'; +import type { LivePacket } from '../types/index.js'; +import { + buildPacketSharePayload, + destinationConfigured, + isForwardablePacket, + isValidPacketShareBrokerUrl, + packetShareEventKey, + renderPacketShareTopic, + type PacketShareDestinationRow, +} from './packetShareConfig.js'; +import { decryptPacketShareSecret, hasPacketShareEncryptionKey } from './packetShareCrypto.js'; +import type { PacketShareDeliveryRow, PacketShareRepository } from './packetShareRepository.js'; +import { packetShareOutcomesTotal, packetShareQueueDepth } from '../metrics.js'; + +const DEFAULT_POLL_INTERVAL_MS = 5_000; +const DEFAULT_BATCH_SIZE = 25; +const PUBLISH_TIMEOUT_MS = 10_000; + +export type PacketSharePublisher = { + publish: (topic: string, payload: string) => Promise; + close: () => Promise; +}; + +export type PacketSharePublisherFactory = ( + destination: PacketShareDestinationRow, +) => PacketSharePublisher; + +function positiveIntegerSetting(name: string, fallback: number, max: number): number { + const parsed = Number(process.env[name]); + return Number.isFinite(parsed) && parsed > 0 ? Math.min(max, Math.floor(parsed)) : fallback; +} + +function destinationUrl(destination: PacketShareDestinationRow): URL { + if (!destination.broker_url) throw new Error('destination broker URL is missing'); + if (!isValidPacketShareBrokerUrl(destination.broker_url)) { + throw new Error('destination broker URL must use mqtt, mqtts, ws, or wss without credentials'); + } + const url = new URL(destination.broker_url.trim()); + return url; +} + +function destinationFingerprint(destination: PacketShareDestinationRow): string { + return JSON.stringify([ + destination.broker_url, + destination.topic_template, + destination.broker_username, + destination.broker_password_ciphertext, + ]); +} + +function connectMqttPublisher(destination: PacketShareDestinationRow): PacketSharePublisher { + let client: MqttClient | null = null; + let connecting: Promise | null = null; + + const closeClient = async (force = true): Promise => { + const current = client; + client = null; + connecting = null; + if (current) await current.endAsync(force).catch(() => undefined); + }; + + const connect = async (): Promise => { + if (client?.connected) return client; + if (connecting) return connecting; + const url = destinationUrl(destination); + const encryptedPassword = destination.broker_password_ciphertext?.trim() || ''; + const password = encryptedPassword + ? decryptPacketShareSecret(encryptedPassword) + : undefined; + const created = mqtt.connect(url.toString(), { + clientId: `meshcore-share-${randomUUID().replaceAll('-', '').slice(0, 24)}`, + username: destination.broker_username?.trim() || undefined, + password, + clean: true, + reconnectPeriod: 0, + connectTimeout: PUBLISH_TIMEOUT_MS, + }); + client = created; + created.on('error', (error) => { + console.error(`[packet-share] MQTT destination ${destination.destination_id} error:`, error.message); + }); + connecting = new Promise((resolve, reject) => { + let settled = false; + let timer: NodeJS.Timeout; + const cleanup = () => { + clearTimeout(timer); + created.removeListener('connect', onConnect); + created.removeListener('close', onClose); + }; + const finish = (error?: Error) => { + if (settled) return; + settled = true; + cleanup(); + if (error) { + if (client === created) client = null; + created.end(true); + reject(error); + } else { + resolve(created); + } + }; + const onConnect = () => finish(); + const onClose = () => finish(new Error('destination MQTT connection closed before connect')); + timer = setTimeout(() => finish(new Error('destination MQTT connection timed out')), PUBLISH_TIMEOUT_MS); + created.once('connect', onConnect); + created.once('close', onClose); + }).finally(() => { + connecting = null; + }); + return connecting; + }; + + return { + async publish(topic: string, payload: string): Promise { + const current = await connect(); + await current.publishAsync(topic, payload, { qos: 1 }); + if (!current.connected) { + await closeClient(); + throw new Error('destination MQTT connection closed after publish'); + } + }, + close: closeClient, + }; +} + +export type PacketShareForwarderDeps = { + repository: PacketShareRepository; + publisherFactory?: PacketSharePublisherFactory; + pollIntervalMs?: number; + batchSize?: number; +}; + +export function createPacketShareForwarder(deps: PacketShareForwarderDeps) { + const publisherFactory = deps.publisherFactory ?? connectMqttPublisher; + const pollIntervalMs = deps.pollIntervalMs ?? positiveIntegerSetting( + 'OWNER_PACKET_SHARE_POLL_INTERVAL_MS', DEFAULT_POLL_INTERVAL_MS, 60_000, + ); + const batchSize = deps.batchSize ?? positiveIntegerSetting( + 'OWNER_PACKET_SHARE_BATCH_SIZE', DEFAULT_BATCH_SIZE, 100, + ); + const publishers = new Map(); + let pollTimer: NodeJS.Timeout | null = null; + let processing: Promise | null = null; + let stopping = false; + + async function publisherFor(destination: PacketShareDestinationRow): Promise { + const fingerprint = destinationFingerprint(destination); + const existing = publishers.get(destination.destination_id); + if (existing?.fingerprint === fingerprint) return existing.publisher; + if (existing) { + publishers.delete(destination.destination_id); + await existing.publisher.close().catch(() => undefined); + } + const created = publisherFactory(destination); + publishers.set(destination.destination_id, { publisher: created, fingerprint }); + return created; + } + + async function enqueuePacket(packet: LivePacket, upstreamEnvelope: Record): Promise { + if (!isForwardablePacket(packet)) { + packetShareOutcomesTotal.inc({ outcome: 'skipped' }); + return; + } + const targets = await deps.repository.activeTargets(packet.rxNodeId!); + if (targets.length < 1) return; + + for (const target of targets) { + if (!destinationConfigured(target, hasPacketShareEncryptionKey())) { + packetShareOutcomesTotal.inc({ outcome: 'unconfigured' }); + continue; + } + let destinationTopic: string; + try { + destinationTopic = renderPacketShareTopic(target.topic_template!, packet); + } catch (error) { + packetShareOutcomesTotal.inc({ outcome: 'invalid_destination' }); + console.error('[packet-share] invalid destination topic:', error instanceof Error ? error.message : error); + continue; + } + try { + const eventKey = packetShareEventKey(target.owner_username, target.destination_id, packet); + const payload = buildPacketSharePayload(packet, upstreamEnvelope); + const inserted = await deps.repository.enqueueDelivery({ + ownerUsername: target.owner_username, + nodeId: packet.rxNodeId!, + destinationId: target.destination_id, + eventKey, + packet, + destinationTopic, + payload: payload as unknown as Record, + }); + packetShareOutcomesTotal.inc({ outcome: inserted ? 'queued' : 'duplicate' }); + } catch (error) { + packetShareOutcomesTotal.inc({ outcome: 'queue_error' }); + console.error('[packet-share] queue error:', error instanceof Error ? error.message : error); + } + } + void processDue(); + } + + async function deliver(row: PacketShareDeliveryRow, claimToken: string): Promise { + const destination = await deps.repository.getDestination(row.destination_id); + if (!destination || !destinationConfigured(destination, hasPacketShareEncryptionKey())) { + await deps.repository.markFailed(row.id, claimToken, row.attempts, 'destination is missing or incomplete'); + packetShareOutcomesTotal.inc({ outcome: 'delivery_error' }); + return; + } + try { + const publisher = await publisherFor(destination); + await publisher.publish(row.destination_topic, JSON.stringify(row.payload)); + await deps.repository.markSucceeded(row.id, claimToken); + packetShareOutcomesTotal.inc({ outcome: 'delivered' }); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + const entry = publishers.get(row.destination_id); + if (entry) { + publishers.delete(row.destination_id); + await entry.publisher.close().catch(() => undefined); + } + await deps.repository.markFailed(row.id, claimToken, row.attempts, message); + packetShareOutcomesTotal.inc({ outcome: 'delivery_error' }); + } + } + + async function processDue(): Promise { + if (stopping) return; + if (processing) { + await processing; + return; + } + processing = (async () => { + try { + await deps.repository.reclaimExpiredClaims(); + const claimToken = randomUUID(); + const rows = await deps.repository.claimDue(batchSize, claimToken); + packetShareQueueDepth.set(rows.length); + await Promise.all(rows.map((row) => deliver(row, claimToken))); + } catch (error) { + packetShareOutcomesTotal.inc({ outcome: 'worker_error' }); + console.error('[packet-share] worker error:', error instanceof Error ? error.message : error); + } finally { + packetShareQueueDepth.set(0); + processing = null; + } + })(); + await processing; + } + + function start(): void { + if (pollTimer || stopping) return; + stopping = false; + pollTimer = setInterval(() => void processDue(), pollIntervalMs); + pollTimer.unref(); + void processDue(); + } + + async function stop(): Promise { + stopping = true; + if (pollTimer) clearInterval(pollTimer); + pollTimer = null; + if (processing) await processing.catch(() => undefined); + const current = [...publishers.values()].map(({ publisher }) => publisher); + publishers.clear(); + await Promise.all(current.map((publisher) => publisher.close().catch(() => undefined))); + } + + return { enqueuePacket, processDue, start, stop }; +} diff --git a/backend/src/owner/packetShareRepository.ts b/backend/src/owner/packetShareRepository.ts new file mode 100644 index 0000000..d9c7c6c --- /dev/null +++ b/backend/src/owner/packetShareRepository.ts @@ -0,0 +1,265 @@ +import type { DatabaseQueryFn } from '../db/index.js'; +import type { LivePacket } from '../types/index.js'; +import type { PacketShareDestinationRow, PacketShareTarget } from './packetShareConfig.js'; + +export type PacketShareQueryFn = DatabaseQueryFn; +export type PacketShareTransactionFn = (work: (query: PacketShareQueryFn) => Promise) => Promise; + +export type PacketShareDeliveryRow = { + id: string; + owner_username: string; + node_id: string; + destination_id: string; + event_key: string; + packet_hash: string; + source_topic: string; + destination_topic: string; + payload: Record; + attempts: number; +}; + +export function createPacketShareRepository( + query: PacketShareQueryFn, + transaction?: PacketShareTransactionFn, +) { + async function listDestinations(): Promise { + const result = await query( + `SELECT destination_id, display_name, website_url, description, + broker_url, topic_template, + broker_username, broker_password_ciphertext, enabled + FROM owner_packet_share_destinations + ORDER BY display_name, destination_id`, + ); + return result.rows; + } + + async function getDestination(destinationId: string): Promise { + const result = await query( + `SELECT destination_id, display_name, website_url, description, + broker_url, topic_template, + broker_username, broker_password_ciphertext, enabled + FROM owner_packet_share_destinations + WHERE destination_id = $1`, + [destinationId], + ); + return result.rows[0] ?? null; + } + + async function listOwnerRules(ownerUsername: string, nodeId: string): Promise { + const result = await query<{ destination_id: string }>( + `SELECT destination_id + FROM owner_packet_share_rules + WHERE owner_username = $1 AND node_id = $2 AND enabled = TRUE + ORDER BY destination_id`, + [ownerUsername, nodeId], + ); + return result.rows.map((row) => row.destination_id); + } + + async function listLastForwardedAt(ownerUsername: string, nodeId: string): Promise> { + const result = await query<{ destination_id: string; last_forwarded_at: Date | string }>( + `SELECT destination_id, MAX(delivered_at) AS last_forwarded_at + FROM owner_packet_share_deliveries + WHERE owner_username = $1 + AND node_id = $2 + AND status = 'succeeded' + AND delivered_at IS NOT NULL + GROUP BY destination_id`, + [ownerUsername, nodeId], + ); + return new Map(result.rows.map((row) => [row.destination_id, new Date(row.last_forwarded_at).toISOString()])); + } + + async function activeTargets(nodeId: string): Promise { + const result = await query( + `SELECT rules.owner_username, rules.node_id, + destination.destination_id, destination.display_name, + destination.website_url, destination.description, + destination.broker_url, destination.topic_template, + destination.broker_username, destination.broker_password_ciphertext, + destination.enabled + FROM owner_packet_share_rules rules + JOIN owner_packet_share_destinations destination + ON destination.destination_id = rules.destination_id + WHERE rules.node_id = $1 + AND rules.enabled = TRUE + AND destination.enabled = TRUE + ORDER BY destination.destination_id`, + [nodeId], + ); + return result.rows; + } + + async function replaceOwnerRules( + ownerUsername: string, + nodeId: string, + destinationIds: string[], + enabled: boolean, + ): Promise { + const update = async (runQuery: PacketShareQueryFn): Promise => { + await runQuery( + `DELETE FROM owner_packet_share_rules + WHERE owner_username = $1 AND node_id = $2`, + [ownerUsername, nodeId], + ); + await runQuery( + `UPDATE owner_packet_share_deliveries + SET status = 'dead_lettered', + last_error = 'sharing rule disabled', + claim_token = NULL, + claim_expires_at = NULL, + updated_at = NOW() + WHERE owner_username = $1 + AND node_id = $2 + AND status IN ('pending', 'failed', 'delivering') + AND ($4::boolean = FALSE OR destination_id <> ALL($3::text[]))`, + [ownerUsername, nodeId, destinationIds, enabled], + ); + if (!enabled || destinationIds.length < 1) return []; + const result = await runQuery<{ destination_id: string }>( + `INSERT INTO owner_packet_share_rules (owner_username, node_id, destination_id, enabled) + SELECT $1, $2, destination_id, TRUE + FROM owner_packet_share_destinations + WHERE destination_id = ANY($3::text[]) + AND enabled = TRUE + RETURNING destination_id`, + [ownerUsername, nodeId, destinationIds], + ); + return result.rows.map((row) => row.destination_id).sort(); + }; + if (transaction) { + return transaction(update); + } + return update(query); + } + + async function enqueueDelivery(input: { + ownerUsername: string; + nodeId: string; + destinationId: string; + eventKey: string; + packet: LivePacket; + destinationTopic: string; + payload: Record; + }): Promise { + const result = await query( + `INSERT INTO owner_packet_share_deliveries ( + owner_username, node_id, destination_id, event_key, + packet_hash, source_topic, destination_topic, payload + ) + SELECT $1, $2, $3, $4, $5, $6, $7, $8::jsonb + WHERE EXISTS ( + SELECT 1 + FROM owner_packet_share_rules rules + JOIN owner_packet_share_destinations destination + ON destination.destination_id = rules.destination_id + WHERE rules.owner_username = $1 + AND rules.node_id = $2 + AND rules.destination_id = $3 + AND rules.enabled = TRUE + AND destination.enabled = TRUE + ) + ON CONFLICT (event_key) DO NOTHING + RETURNING id`, + [ + input.ownerUsername, + input.nodeId, + input.destinationId, + input.eventKey, + input.packet.packetHash, + input.packet.topic, + input.destinationTopic, + JSON.stringify(input.payload), + ], + ); + return result.rows.length > 0; + } + + async function reclaimExpiredClaims(now = new Date()): Promise { + const result = await query<{ count: string }>( + `WITH reclaimed AS ( + UPDATE owner_packet_share_deliveries + SET status = CASE WHEN attempts >= 5 THEN 'dead_lettered' ELSE 'failed' END, + next_attempt_at = NOW(), + claim_token = NULL, + claim_expires_at = NULL, + updated_at = NOW() + WHERE status = 'delivering' + AND claim_expires_at < $1 + RETURNING id + ) SELECT COUNT(*)::text AS count FROM reclaimed`, + [now], + ); + return Number(result.rows[0]?.count ?? 0); + } + + async function claimDue(limit: number, claimToken: string): Promise { + const result = await query( + `WITH due AS ( + SELECT id + FROM owner_packet_share_deliveries + WHERE status IN ('pending', 'failed') + AND attempts < 5 + AND next_attempt_at <= NOW() + ORDER BY next_attempt_at, id + FOR UPDATE SKIP LOCKED + LIMIT $1 + ) + UPDATE owner_packet_share_deliveries delivery + SET status = 'delivering', + attempts = delivery.attempts + 1, + last_attempt_at = NOW(), + claim_token = $2, + claim_expires_at = NOW() + INTERVAL '30 seconds', + updated_at = NOW() + FROM due + WHERE delivery.id = due.id + RETURNING delivery.id::text, delivery.owner_username, delivery.node_id, + delivery.destination_id, delivery.event_key, delivery.packet_hash, + delivery.source_topic, delivery.destination_topic, + delivery.payload, delivery.attempts`, + [limit, claimToken], + ); + return result.rows; + } + + async function markSucceeded(id: string, claimToken: string): Promise { + await query( + `UPDATE owner_packet_share_deliveries + SET status = 'succeeded', delivered_at = NOW(), last_error = NULL, + claim_token = NULL, claim_expires_at = NULL, updated_at = NOW() + WHERE id = $1 AND claim_token = $2`, + [id, claimToken], + ); + } + + async function markFailed(id: string, claimToken: string, attempts: number, error: string): Promise { + const terminal = attempts >= 5; + const delaySeconds = Math.min(300, 2 ** Math.max(0, attempts - 1)); + await query( + `UPDATE owner_packet_share_deliveries + SET status = $3, + next_attempt_at = NOW() + ($4::text || ' seconds')::interval, + last_error = $5, claim_token = NULL, claim_expires_at = NULL, + updated_at = NOW() + WHERE id = $1 AND claim_token = $2`, + [id, claimToken, terminal ? 'dead_lettered' : 'failed', String(delaySeconds), error.slice(0, 500)], + ); + } + + return { + listDestinations, + getDestination, + listOwnerRules, + listLastForwardedAt, + activeTargets, + replaceOwnerRules, + enqueueDelivery, + reclaimExpiredClaims, + claimDue, + markSucceeded, + markFailed, + }; +} + +export type PacketShareRepository = ReturnType; diff --git a/backend/src/tools/encryptPacketShareSecret.ts b/backend/src/tools/encryptPacketShareSecret.ts new file mode 100644 index 0000000..644cd1f --- /dev/null +++ b/backend/src/tools/encryptPacketShareSecret.ts @@ -0,0 +1,15 @@ +import { encryptPacketShareSecret } from '../owner/packetShareCrypto.js'; + +const chunks: Buffer[] = []; +process.stdin.on('data', (chunk: Buffer | string) => { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); +}); +process.stdin.on('end', () => { + try { + const secret = Buffer.concat(chunks).toString('utf8').replace(/\r?\n$/, ''); + process.stdout.write(`${encryptPacketShareSecret(secret)}\n`); + } catch (error) { + console.error(error instanceof Error ? error.message : error); + process.exitCode = 1; + } +}); diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 50ed4b1..b68b273 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -3419,6 +3419,409 @@ ] } }, + "/api/owner/packet-sharing": { + "get": { + "summary": "Read owner packet-sharing settings", + "operationId": "get-owner-packet-sharing", + "tags": [ + "owner" + ], + "x-access": "owner", + "parameters": [ + { + "name": "nodeId", + "in": "query", + "required": false, + "schema": { + "type": "string", + "pattern": "^[0-9A-Fa-f]{64}$" + } + } + ], + "responses": { + "200": { + "description": "Successful response", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "nodeId", + "featureEnabled", + "enabled", + "destinations" + ], + "properties": { + "nodeId": { + "type": "string", + "pattern": "^[0-9A-Fa-f]{64}$" + }, + "featureEnabled": { + "type": "boolean" + }, + "enabled": { + "type": "boolean" + }, + "destinations": { + "type": "array", + "items": { + "type": "object", + "required": [ + "id", + "name", + "websiteUrl", + "description", + "configured", + "selected", + "lastForwardedAt" + ], + "properties": { + "id": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$" + }, + "name": { + "type": "string", + "minLength": 1, + "maxLength": 120 + }, + "websiteUrl": { + "type": [ + "string", + "null" + ], + "format": "uri", + "maxLength": 2048 + }, + "description": { + "type": [ + "string", + "null" + ], + "maxLength": 160 + }, + "configured": { + "type": "boolean" + }, + "selected": { + "type": "boolean" + }, + "lastForwardedAt": { + "type": [ + "string", + "null" + ], + "format": "date-time" + } + }, + "additionalProperties": false + } + } + }, + "additionalProperties": false + } + } + } + }, + "400": { + "description": "Invalid or out-of-bounds request", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "401": { + "description": "Authentication required", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "403": { + "description": "Authentication or CSRF validation failed", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "429": { + "description": "Rate limit exceeded", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "500": { + "description": "Internal server error", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + } + }, + "security": [ + { + "ownerSession": [] + } + ] + }, + "post": { + "summary": "Update owner packet-sharing settings", + "operationId": "post-owner-packet-sharing", + "tags": [ + "owner" + ], + "x-access": "owner", + "parameters": [], + "responses": { + "200": { + "description": "Successful response", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "nodeId", + "enabled", + "destinationIds" + ], + "properties": { + "nodeId": { + "type": "string", + "pattern": "^[0-9A-Fa-f]{64}$" + }, + "enabled": { + "type": "boolean" + }, + "destinationIds": { + "type": "array", + "maxItems": 32, + "items": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$" + } + } + }, + "additionalProperties": false + } + } + } + }, + "400": { + "description": "Invalid or out-of-bounds request", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "401": { + "description": "Authentication required", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "403": { + "description": "Authentication or CSRF validation failed", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "409": { + "description": "One or more selected destinations are not configured", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "429": { + "description": "Rate limit exceeded", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + }, + "500": { + "description": "Internal server error", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + } + }, + "additionalProperties": false + } + } + } + } + }, + "security": [ + { + "ownerSession": [] + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "nodeId", + "enabled", + "destinationIds" + ], + "properties": { + "nodeId": { + "type": "string", + "pattern": "^[0-9A-Fa-f]{64}$" + }, + "enabled": { + "type": "boolean" + }, + "destinationIds": { + "type": "array", + "maxItems": 32, + "items": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$" + } + } + }, + "additionalProperties": false + } + } + } + } + } + }, "/api/owner/session": { "get": { "summary": "Read owner session",