pss: HTTPS fallback for reverse direction (job_ack / job_deliver)
Bee 2.8's pushsync from homelab to prod consistently routes our trojan chunks to neighbour ff7adaed instead of prod_bee at ff896a93, even with both nodes directly peered. The chunk lands at a third-party node that can't decrypt it, so the gateway never receives provider ack/deliver envelopes and the chat times out. Workaround: gateway advertises a clientReplyUrl in job_notify. Provider POSTs the same signed envelopes there; gateway verifies the signature (same trust check as the PSS subscription) and dispatches via the same onEnvelope handler. PSS remains the fallback when no replyUrl is set (non-hosted clients). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
1 parent
2c8daee63b
commit
e06060aa16
6 files changed
+73
-6
No files matched your search
@@ -8,6 +8,7 @@ import { usersRoutes } from './users/routes';
|
||||
import { accountsRoutes } from './accounts/routes';
|
||||
import { apiKeysRoutes } from './apiKeys/routes';
|
||||
import { gatewayRoutes } from './gateway/routes';
|
||||
import { pssInboxRoutes } from './gateway/pssInbox';
|
||||
import { beeRoutes } from './bee/routes';
|
||||
import { getGatewayContext } from './gateway/context';
|
||||
|
||||
@@ -27,6 +28,7 @@ const app = new Elysia()
|
||||
.use(accountsRoutes)
|
||||
.use(apiKeysRoutes)
|
||||
.use(beeRoutes)
|
||||
.use(pssInboxRoutes)
|
||||
.use(gatewayRoutes);
|
||||
|
||||
// Start the HTTP server immediately so auth / accounts / api-keys routes are
|
||||
|
||||
@@ -56,6 +56,10 @@ const EnvSchema = z.object({
|
||||
|
||||
// Server
|
||||
HOSTED_GATEWAY_PORT: z.coerce.number().int().positive().default(8080),
|
||||
// Public base URL of this gateway, advertised to providers in job_notify
|
||||
// envelopes as `clientReplyUrl`. Providers POST ack/deliver here when PSS
|
||||
// reverse delivery is unreliable (e.g. Bee 2.8 push-routing skips us).
|
||||
PUBLIC_API_URL: z.string().url().optional(),
|
||||
LOG_LEVEL: z.enum(['fatal', 'error', 'warn', 'info', 'debug', 'trace']).default('info'),
|
||||
|
||||
// Operator PSS keypair file. All `prod_api` and `prod_worker` replicas
|
||||
|
||||
+17
-6
@@ -17,6 +17,7 @@ import { EciesCipher, jsonDecrypt } from '../t4t/lib/crypto';
|
||||
import { loadOrCreatePssKey } from '../t4t/lib/keys';
|
||||
import { ModelDiscovery } from '../t4t/modes/gateway/models';
|
||||
import type {
|
||||
Envelope,
|
||||
JobAckBody,
|
||||
JobDeliverBody,
|
||||
OpenAIChatResponse,
|
||||
@@ -66,6 +67,11 @@ export interface GatewayContext {
|
||||
clearFailureTimers: (routing: Hex) => void;
|
||||
rejectAndCleanup: (routing: Hex, err: Error) => void;
|
||||
scheduleAckTimeout: (routing: Hex, slot: FailureSlot) => void;
|
||||
/** Process an inbound signed envelope from any transport (PSS or HTTPS).
|
||||
* The caller is responsible for verifyEnvelope; the PSS subscription
|
||||
* layer already does this. The HTTPS inbox endpoint must do it before
|
||||
* invoking. */
|
||||
onEnvelope: (envelope: Envelope) => Promise<void>;
|
||||
}
|
||||
|
||||
let bootPromise: Promise<GatewayContext> | null = null;
|
||||
@@ -176,11 +182,9 @@ async function bootGatewayContext(): Promise<GatewayContext> {
|
||||
}, (ACK_WINDOW_SECONDS + FAILURE_GRACE_SECONDS) * 1000);
|
||||
}
|
||||
|
||||
// Subscribe to the operator's client topic — ALL tenants' jobs come back
|
||||
// here (because all jobs go on-chain as the operator wallet).
|
||||
pss.subscribe({
|
||||
topic: clientTopic(chain.address),
|
||||
onEnvelope: async (envelope) => {
|
||||
// Handler for any verified envelope addressed to this gateway, regardless
|
||||
// of which transport (PSS subscription or HTTPS /pss/inbox) brought it.
|
||||
const onEnvelope = async (envelope: Envelope): Promise<void> => {
|
||||
if (envelope.type === 'job_ack') {
|
||||
const body = envelope.body as JobAckBody;
|
||||
logger.info({ jobId: body.jobId, eta: body.estimatedCompletion }, 'ack received');
|
||||
@@ -231,7 +235,13 @@ async function bootGatewayContext(): Promise<GatewayContext> {
|
||||
clearFailureTimers(body.jobId);
|
||||
}
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
// Subscribe to the operator's client topic — ALL tenants' jobs come back
|
||||
// here (because all jobs go on-chain as the operator wallet).
|
||||
pss.subscribe({
|
||||
topic: clientTopic(chain.address),
|
||||
onEnvelope,
|
||||
});
|
||||
|
||||
logger.info(
|
||||
@@ -259,6 +269,7 @@ async function bootGatewayContext(): Promise<GatewayContext> {
|
||||
clearFailureTimers,
|
||||
rejectAndCleanup,
|
||||
scheduleAckTimeout,
|
||||
onEnvelope,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -138,6 +138,7 @@ export async function handleChat(args: HandleChatArgs): Promise<HandleChatResult
|
||||
deliveryDeadline,
|
||||
clientPssPubKey: ctx.pssPubKeyX,
|
||||
clientSwarmOverlay: ctx.selfOverlay,
|
||||
clientReplyUrl: env.PUBLIC_API_URL ? `${env.PUBLIC_API_URL.replace(/\/$/, '')}/pss/inbox` : undefined,
|
||||
},
|
||||
},
|
||||
(msg) => ctx.chain.wallet.signMessage({ account: ctx.chain.wallet.account!, message: msg }),
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
import Elysia, { t } from 'elysia';
|
||||
import { getGatewayContext } from './context';
|
||||
import { verifyEnvelope } from '../t4t/lib/envelope';
|
||||
import type { Envelope } from '../t4t/lib/types';
|
||||
import { logger } from '../logger';
|
||||
|
||||
/**
|
||||
* HTTPS fallback for the reverse-direction PSS path. The provider POSTs the
|
||||
* same signed envelopes (`job_ack`, `job_deliver`) it would have sent via PSS.
|
||||
* We verify the signature here — same trust check as the PSS subscription — and
|
||||
* hand off to the shared envelope handler.
|
||||
*
|
||||
* Lives outside the JWT-guarded routes because providers don't have JWTs.
|
||||
* Security model: anyone can POST, but only envelopes signed by a wallet the
|
||||
* gateway is actively waiting on (`pending.has(jobId)`) ever produce side
|
||||
* effects. Unsigned or unmatched envelopes are dropped.
|
||||
*/
|
||||
export const pssInboxRoutes = new Elysia().post(
|
||||
'/pss/inbox',
|
||||
async ({ body, set }) => {
|
||||
const env = body as Envelope;
|
||||
if (!(await verifyEnvelope(env))) {
|
||||
set.status = 400;
|
||||
logger.warn({ from: env.from, type: env.type }, 'pss/inbox: invalid envelope signature');
|
||||
return { error: 'invalid envelope signature' };
|
||||
}
|
||||
const ctx = await getGatewayContext();
|
||||
await ctx.onEnvelope(env);
|
||||
logger.info({ from: env.from, type: env.type }, 'pss/inbox: envelope dispatched');
|
||||
return { ok: true };
|
||||
},
|
||||
{
|
||||
body: t.Object({
|
||||
v: t.Number(),
|
||||
type: t.Union([t.Literal('job_notify'), t.Literal('job_ack'), t.Literal('job_deliver')]),
|
||||
from: t.String(),
|
||||
to: t.String(),
|
||||
ts: t.Number(),
|
||||
nonce: t.String(),
|
||||
sig: t.String(),
|
||||
body: t.Any(),
|
||||
}),
|
||||
},
|
||||
);
|
||||
@@ -36,6 +36,11 @@ export interface JobNotifyBody {
|
||||
clientPssPubKey: Hex
|
||||
/** Gateway's Bee overlay for PSS routing. */
|
||||
clientSwarmOverlay: Hex
|
||||
/** Optional HTTPS URL the provider may POST ack/deliver envelopes to,
|
||||
* for clients (like the hosted gateway) where reverse-direction PSS
|
||||
* delivery isn't reliable. The provider trusts this because the
|
||||
* envelope is signed by the gateway's on-chain wallet. */
|
||||
clientReplyUrl?: string
|
||||
}
|
||||
|
||||
export interface JobAckBody {
|
||||
|
||||
Reference in new issue
Block a user