Merge pull request #198 from theta42/feat-no-inbound-relay
feat(multi-site): no-inbound relay automation via theta-proxy's existing API tokens
This commit is contained in:
@@ -29,7 +29,17 @@ class SiteSpoke extends Model {
|
||||
siteSlug: { type: 'string' },
|
||||
pushToken: { type: 'string', isRequired: true },
|
||||
created_on: { type: 'integer' },
|
||||
last_seen_on: { type: 'integer' }
|
||||
last_seen_on: { type: 'integer' },
|
||||
// No-inbound relay (MULTI_SITE_SPEC.md): a spoke with no public IP of
|
||||
// its own reports its WG mesh IP + the public hostname it wants
|
||||
// reached at; the master then best-effort creates a matching relay
|
||||
// route on its own theta-proxy (utils/proxy_client.js). relayNote
|
||||
// records what happened for visibility in the UI -- this automation
|
||||
// is optional/best-effort, never a join requirement.
|
||||
noInbound: { type: 'boolean', default: false },
|
||||
meshIp: { type: 'string' },
|
||||
publicHost: { type: 'string' },
|
||||
relayNote: { type: 'string' }
|
||||
};
|
||||
|
||||
toPublic() {
|
||||
|
||||
+1
-1
@@ -11,7 +11,7 @@
|
||||
"scripts": {
|
||||
"start": "node ./bin/www",
|
||||
"dev": "npx nodemon --ignore public/ ./bin/www",
|
||||
"test": "NODE_ENV=test jest tests/groups.test.js tests/subtypes.test.js tests/site_join.test.js tests/site_config.test.js tests/site_replicate.test.js --forceExit"
|
||||
"test": "NODE_ENV=test jest tests/groups.test.js tests/subtypes.test.js tests/site_join.test.js tests/site_config.test.js tests/site_replicate.test.js tests/proxy_client.test.js --forceExit"
|
||||
},
|
||||
"jest": {
|
||||
"testEnvironment": "node",
|
||||
|
||||
@@ -120,27 +120,43 @@ router.post('/spokes', async (req, res, next) => {
|
||||
const key = await SiteJoinKey.authenticate(rawKey);
|
||||
if (!key) return res.status(401).json({ status: 'error', message: 'invalid or revoked site join key' });
|
||||
|
||||
const { endpoint, siteSlug } = req.body || {};
|
||||
const { endpoint, siteSlug, noInbound, meshIp, publicHost } = req.body || {};
|
||||
if (!endpoint || !/^https?:\/\//.test(endpoint)) {
|
||||
return res.status(400).json({ status: 'error', message: 'a valid http(s) endpoint is required' });
|
||||
}
|
||||
|
||||
const now = Math.floor(Date.now() / 1000);
|
||||
let spoke = (await SiteSpoke.list({ where: { endpoint } }))[0];
|
||||
const patch = { siteSlug: siteSlug || (spoke && spoke.siteSlug) || null, last_seen_on: now, noInbound: !!noInbound, meshIp: meshIp || '', publicHost: publicHost || '' };
|
||||
if (spoke) {
|
||||
await spoke.update({ siteSlug: siteSlug || spoke.siteSlug, last_seen_on: now });
|
||||
await spoke.update(patch);
|
||||
} else {
|
||||
spoke = await SiteSpoke.create({
|
||||
id: crypto.randomUUID(),
|
||||
endpoint,
|
||||
siteSlug: siteSlug || null,
|
||||
pushToken: SiteSpoke.generatePushToken(),
|
||||
created_on: now,
|
||||
last_seen_on: now
|
||||
...patch
|
||||
});
|
||||
}
|
||||
logAudit('spoke_registered', { endpoint, siteSlug: spoke.siteSlug });
|
||||
res.json({ status: 'ok', pushToken: spoke.pushToken });
|
||||
|
||||
// No-inbound relay automation: best-effort, never blocks registration.
|
||||
// See utils/proxy_client.js for why this reuses theta-proxy's existing
|
||||
// API token system rather than a new credential type.
|
||||
let relayNote = 'not applicable (spoke has inbound access)';
|
||||
if (noInbound) {
|
||||
if (meshIp && publicHost) {
|
||||
const proxyClient = require('../utils/proxy_client');
|
||||
const result = await proxyClient.ensureRelayRoute({ host: publicHost, ip: meshIp, targetPort: 3001 });
|
||||
relayNote = result.note;
|
||||
} else {
|
||||
relayNote = 'skipped: noInbound set but meshIp/publicHost missing';
|
||||
}
|
||||
await spoke.update({ relayNote });
|
||||
}
|
||||
|
||||
logAudit('spoke_registered', { endpoint, siteSlug: spoke.siteSlug, noInbound: !!noInbound, relayNote });
|
||||
res.json({ status: 'ok', pushToken: spoke.pushToken, relay: { note: relayNote } });
|
||||
} catch (e) { next(e); }
|
||||
});
|
||||
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
'use strict';
|
||||
|
||||
let mockBaoStore = new Map();
|
||||
jest.mock('@simpleworkjs/bao-conf', () => ({
|
||||
get: jest.fn(async (path) => mockBaoStore.get(path) || null),
|
||||
set: jest.fn(async (path, value) => { mockBaoStore.set(path, value); })
|
||||
}));
|
||||
|
||||
describe('proxy_client', () => {
|
||||
let proxyClient;
|
||||
let originalFetch;
|
||||
let mockFetchImpl;
|
||||
let calls;
|
||||
|
||||
beforeEach(() => {
|
||||
jest.resetModules();
|
||||
mockBaoStore = new Map();
|
||||
calls = [];
|
||||
mockFetchImpl = async () => ({ ok: true, status: 404 });
|
||||
originalFetch = global.fetch;
|
||||
global.fetch = (...args) => { calls.push(args); return mockFetchImpl(...args); };
|
||||
proxyClient = require('../utils/proxy_client');
|
||||
proxyClient._reset();
|
||||
delete process.env.PROXY_INTERNAL_URL;
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
global.fetch = originalFetch;
|
||||
});
|
||||
|
||||
test('skips cleanly when required fields are missing', async () => {
|
||||
const result = await proxyClient.ensureRelayRoute({ host: '', ip: '', targetPort: 0 });
|
||||
expect(result.note).toMatch(/required/);
|
||||
expect(calls.length).toBe(0);
|
||||
});
|
||||
|
||||
test('skips cleanly when PROXY_INTERNAL_URL is not configured', async () => {
|
||||
const result = await proxyClient.ensureRelayRoute({ host: 'sso-a.example.com', ip: '172.24.5.1', targetPort: 3001 });
|
||||
expect(result.note).toMatch(/PROXY_INTERNAL_URL/);
|
||||
expect(calls.length).toBe(0);
|
||||
});
|
||||
|
||||
test('skips cleanly when no token is stored in OpenBao', async () => {
|
||||
process.env.PROXY_INTERNAL_URL = 'https://proxy.internal';
|
||||
const result = await proxyClient.ensureRelayRoute({ host: 'sso-a.example.com', ip: '172.24.5.1', targetPort: 3001 });
|
||||
expect(result.note).toMatch(/no proxy API token/);
|
||||
expect(calls.length).toBe(0);
|
||||
});
|
||||
|
||||
test('creates the route when the host does not already exist', async () => {
|
||||
process.env.PROXY_INTERNAL_URL = 'https://proxy.internal';
|
||||
mockBaoStore.set('integrations/theta-proxy', { token: 'prx_test_token' });
|
||||
mockFetchImpl = async (url, opts) => {
|
||||
if (opts.method === undefined) return { ok: true, status: 404 }; // GET lookup
|
||||
if (opts.method === 'POST') return { ok: true, status: 200 };
|
||||
throw new Error('unexpected method ' + opts.method);
|
||||
};
|
||||
|
||||
const result = await proxyClient.ensureRelayRoute({ host: 'sso-a.example.com', ip: '172.24.5.1', targetPort: 3001 });
|
||||
expect(result.note).toBe('created');
|
||||
|
||||
const postCall = calls.find((c) => c[1].method === 'POST');
|
||||
expect(postCall[0]).toBe('https://proxy.internal/api/host');
|
||||
expect(postCall[1].headers.Authorization).toBe('Bearer prx_test_token');
|
||||
expect(JSON.parse(postCall[1].body)).toEqual({ host: 'sso-a.example.com', ip: '172.24.5.1', targetPort: 3001 });
|
||||
});
|
||||
|
||||
test('updates the route when it exists but points somewhere else', async () => {
|
||||
process.env.PROXY_INTERNAL_URL = 'https://proxy.internal';
|
||||
mockBaoStore.set('integrations/theta-proxy', { token: 'prx_test_token' });
|
||||
mockFetchImpl = async (url, opts) => {
|
||||
if (!opts.method) return { ok: true, status: 200, json: async () => ({ results: { ip: '172.24.9.9', targetPort: 3001 } }) };
|
||||
if (opts.method === 'PUT') return { ok: true, status: 200 };
|
||||
throw new Error('unexpected method ' + opts.method);
|
||||
};
|
||||
|
||||
const result = await proxyClient.ensureRelayRoute({ host: 'sso-a.example.com', ip: '172.24.5.1', targetPort: 3001 });
|
||||
expect(result.note).toBe('updated');
|
||||
});
|
||||
|
||||
test('is a no-op when the route already matches', async () => {
|
||||
process.env.PROXY_INTERNAL_URL = 'https://proxy.internal';
|
||||
mockBaoStore.set('integrations/theta-proxy', { token: 'prx_test_token' });
|
||||
mockFetchImpl = async () => ({ ok: true, status: 200, json: async () => ({ results: { ip: '172.24.5.1', targetPort: 3001 } }) });
|
||||
|
||||
const result = await proxyClient.ensureRelayRoute({ host: 'sso-a.example.com', ip: '172.24.5.1', targetPort: 3001 });
|
||||
expect(result.note).toBe('already up to date');
|
||||
});
|
||||
|
||||
test('reports a network failure without throwing', async () => {
|
||||
process.env.PROXY_INTERNAL_URL = 'https://proxy.internal';
|
||||
mockBaoStore.set('integrations/theta-proxy', { token: 'prx_test_token' });
|
||||
mockFetchImpl = async () => { throw new Error('connection refused'); };
|
||||
|
||||
const result = await proxyClient.ensureRelayRoute({ host: 'sso-a.example.com', ip: '172.24.5.1', targetPort: 3001 });
|
||||
expect(result.note).toMatch(/failed: connection refused/);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,105 @@
|
||||
'use strict';
|
||||
|
||||
// Service-to-service client for theta-proxy's Host management API --
|
||||
// MULTI_SITE_SPEC.md's "no-inbound relay automation" (a master creating a
|
||||
// relay route so a spoke with zero inbound path of its own is reachable).
|
||||
//
|
||||
// Deliberately does NOT invent a new credential type. theta-proxy already
|
||||
// has a self-service API token system (models/api_token.js, `prx_<id>_<secret>`
|
||||
// bearer tokens that authenticate as their creator's user + group snapshot --
|
||||
// same pattern this app and jump-host both already have their own copy of).
|
||||
// The "service-to-service auth" gap was never "no credential type exists" --
|
||||
// it's that nothing wired one of these tokens into an actual inter-service
|
||||
// call. This is that wiring, using the credential type that was already
|
||||
// there. The token itself is operator-provisioned (minted on theta-proxy by
|
||||
// an admin with Host-management rights) and stored in OpenBao, same as the
|
||||
// agent-signing key in agent_keys.js.
|
||||
|
||||
const baoConf = require('@simpleworkjs/bao-conf');
|
||||
|
||||
const PATH = 'integrations/theta-proxy'; // baoConf adds the secret/data prefix
|
||||
const REQUEST_TIMEOUT_MS = 10000;
|
||||
|
||||
let cachedToken = null;
|
||||
|
||||
async function loadToken() {
|
||||
if (cachedToken) return cachedToken;
|
||||
let stored;
|
||||
try {
|
||||
stored = await baoConf.get(PATH);
|
||||
} catch (err) {
|
||||
console.error(`[proxy_client] could not read ${PATH} from OpenBao: ${err.message}`);
|
||||
return null;
|
||||
}
|
||||
if (!stored || !stored.token) return null;
|
||||
cachedToken = stored.token;
|
||||
return cachedToken;
|
||||
}
|
||||
|
||||
function proxyBaseUrl() {
|
||||
// Not OpenBao -- this is where the proxy's admin API lives, not a secret.
|
||||
// No safe default: relaying to a guessed host would be worse than
|
||||
// refusing, so this must be explicitly configured.
|
||||
return process.env.PROXY_INTERNAL_URL || '';
|
||||
}
|
||||
|
||||
// Creates the relay Host route if missing, updates its target IP if it
|
||||
// already exists and points somewhere else. Idempotent -- safe to call
|
||||
// again for the same host on every spoke resync.
|
||||
//
|
||||
// Returns { note } describing what happened (created/updated/skipped/failed)
|
||||
// rather than throwing on a missing token or base URL -- callers (spoke
|
||||
// registration) must never fail the whole registration just because this
|
||||
// automation isn't configured yet; it's an enhancement layered on top of a
|
||||
// working join, not a requirement of one.
|
||||
async function ensureRelayRoute({ host, ip, targetPort }) {
|
||||
if (!host || !ip || !targetPort) {
|
||||
return { note: 'skipped: host, ip, and targetPort are all required' };
|
||||
}
|
||||
const base = proxyBaseUrl();
|
||||
if (!base) {
|
||||
return { note: 'skipped: PROXY_INTERNAL_URL not configured' };
|
||||
}
|
||||
const token = await loadToken();
|
||||
if (!token) {
|
||||
return { note: `skipped: no proxy API token at OpenBao ${PATH} -- mint one on theta-proxy and store it there` };
|
||||
}
|
||||
|
||||
const headers = { Authorization: 'Bearer ' + token, 'Content-Type': 'application/json' };
|
||||
const controller = new AbortController();
|
||||
const timer = setTimeout(() => controller.abort(), REQUEST_TIMEOUT_MS);
|
||||
try {
|
||||
const existing = await fetch(base.replace(/\/+$/, '') + '/api/host/' + encodeURIComponent(host), {
|
||||
headers, signal: controller.signal
|
||||
});
|
||||
|
||||
if (existing.status === 200) {
|
||||
// GET /api/host/:item wraps the record in { item, results }, not
|
||||
// flat -- confirmed against a real running proxy (this check
|
||||
// silently always "updated" instead of no-op'ing until fixed).
|
||||
const body = await existing.json();
|
||||
const current = body.results || body;
|
||||
if (current.ip === ip && Number(current.targetPort) === Number(targetPort)) {
|
||||
return { note: 'already up to date' };
|
||||
}
|
||||
const put = await fetch(base.replace(/\/+$/, '') + '/api/host/' + encodeURIComponent(host), {
|
||||
method: 'PUT', headers, body: JSON.stringify({ ip, targetPort }), signal: controller.signal
|
||||
});
|
||||
return put.ok ? { note: 'updated' } : { note: `update failed: HTTP ${put.status}` };
|
||||
}
|
||||
|
||||
const create = await fetch(base.replace(/\/+$/, '') + '/api/host', {
|
||||
method: 'POST', headers, body: JSON.stringify({ host, ip, targetPort }), signal: controller.signal
|
||||
});
|
||||
return create.ok ? { note: 'created' } : { note: `create failed: HTTP ${create.status}` };
|
||||
} catch (err) {
|
||||
return { note: `failed: ${err.message}` };
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
// Test seam.
|
||||
function _reset() { cachedToken = null; }
|
||||
|
||||
module.exports = { ensureRelayRoute, _reset, PATH };
|
||||
Reference in New Issue
Block a user