feat: carry sealed Oblivious HTTP requests over Nostr relays, off by default

dokploy.env.example

@@ -21,3 +21,6 @@ SHIELDED_DATA_DIR=/data
 # SHIELDED_OHTTP_RATE=20
 # SHIELDED_PUBLIC_ORIGIN=https://shielded.mutinynet.arkade.sh
 # SHIELDED_ESPLORA_URL=https://mempool.mutinynet.arkade.sh/api
+# The gateway also listens for sealed requests on Nostr relays: wss:// URLs separated by commas, six public relays when unset. SHIELDED_OHTTP_HTTP=1 adds POST /api/rollup/ohttp for a separate HTTP relay. Rules for choosing relays: docs/DOKPLOY.md.
+# SHIELDED_NOSTR_RELAYS=
+# SHIELDED_OHTTP_HTTP=1

packages/protocol/src/nostr-carrier.ts

@@ -0,0 +1,57 @@
+import {concatBytes} from '@noble/hashes/utils.js';
+import {base64} from '@scure/base';
+import {finalizeEvent,generateSecretKey,getEventHash,verifyEvent,type Event,type VerifiedEvent} from 'nostr-tools/pure';
+
+export const NOSTR_REQUEST_KIND=21470;
+export const NOSTR_ANSWER_KIND=21471;
+export const NOSTR_CHUNK_BYTES=32<<10;
+export const NOSTR_MAX_CHUNKS=32;
+/** Probed relays; none sits behind Cloudflare. Frozen, so a caller's sort() cannot reorder the default. */
+export const NOSTR_RELAYS:readonly string[]=Object.freeze(['wss://nos.lol','wss://relay.snort.social','wss://offchain.pub','wss://nostr.mom','wss://nostr.oxtr.dev','wss://nostr-pub.wellorder.net']);
+type RequestRef={id:string;pubkey:string};
+
+/** Rounded to the nearest 30 s, so created_at is not the exact send time and a clock 45 s slow still passes strfry's 60 s age limit. */
+export const roundedSeconds=(nowMs:number)=>Math.round(nowMs/30_000)*30;
+const tag=(event:Event,name:string,value?:string)=>event.tags.find(t=>t[0]===name&&(value===undefined||t[1]===value));
+const whole=(text?:string)=>/^\d+$/.test(text??'')?Number(text):NaN;
+const decode=(text:string)=>{try{const bytes=base64.decode(text);return bytes.length>0&&bytes.length<=NOSTR_CHUNK_BYTES?bytes:undefined;}catch{return undefined;}};
+// verifyEvent throws on null, primitives and events frozen before their first check, and caches its verdict on the event.
+const verified=(event:Event)=>{try{return verifyEvent(event);}catch{return false;}};
+/** A request's one-time key carries no authority and relays check signatures (NIP-01), so a request is checked by its id: a hash, where Schnorr costs about 2 ms. */
+export const idMatches=(event:Event)=>{try{return getEventHash(event)===event.id;}catch{return false;}};
+const answerEvent=(request:RequestRef,extra:string[][],content:string,secret:Uint8Array,nowMs:number)=>finalizeEvent({kind:NOSTR_ANSWER_KIND,tags:[['e',request.id],['p',request.pubkey],...extra],content,created_at:roundedSeconds(nowMs)},secret);
+
+export function requestEvent(sealed:Uint8Array,gateway:string,nowMs:number):VerifiedEvent {
+ if(sealed.length>NOSTR_CHUNK_BYTES)throw new RangeError(`A sealed request is at most ${NOSTR_CHUNK_BYTES} bytes.`);
+ return finalizeEvent({kind:NOSTR_REQUEST_KIND,tags:[['p',gateway]],content:base64.encode(sealed),created_at:roundedSeconds(nowMs)},generateSecretKey());
+}
+export function readRequest(event:Event,gateway:string):Uint8Array|undefined {
+ if(!idMatches(event)||event.kind!==NOSTR_REQUEST_KIND||!tag(event,'p',gateway))return undefined;
+ return decode(event.content);
+}
+export function answerEvents(sealed:Uint8Array,request:RequestRef,secret:Uint8Array,nowMs:number):VerifiedEvent[] {
+ const count=Math.ceil(sealed.length/NOSTR_CHUNK_BYTES);
+ if(count<1||count>NOSTR_MAX_CHUNKS)throw new RangeError(`A sealed answer is 1 to ${NOSTR_MAX_CHUNKS} chunks of ${NOSTR_CHUNK_BYTES} bytes.`);
+ return Array.from({length:count},(_,i)=>answerEvent(request,[['chunk',String(i),String(count)]],base64.encode(sealed.subarray(i*NOSTR_CHUNK_BYTES,(i+1)*NOSTR_CHUNK_BYTES)),secret,nowMs));
+}
+/** retryAfter is whole seconds; readAnswer ignores a status event whose delay is anything else. */
+export const statusEvent=(status:429|503,retryAfter:number,request:RequestRef,secret:Uint8Array,nowMs:number):VerifiedEvent=>answerEvent(request,[['status',String(status)],['retry-after',String(retryAfter)]],'',secret,nowMs);
+
+export function readAnswer(events:readonly Event[],requestId:string,gateway:string):{kind:'sealed';bytes:Uint8Array}|{kind:'status';status:number;retryAfter:number}|undefined {
+ const mine=events.filter(event=>verified(event)&&event.kind===NOSTR_ANSWER_KIND&&event.pubkey===gateway&&tag(event,'e',requestId));
+ for(const event of mine){
+  const status=whole(tag(event,'status')?.[1]),retryAfter=whole(tag(event,'retry-after')?.[1]);
+  if((status===429||status===503)&&Number.isFinite(retryAfter))return {kind:'status',status,retryAfter};
+ }
+ const parts=new Map<number,string>();let count=0;
+ for(const event of mine){
+  const slot=tag(event,'chunk'),index=whole(slot?.[1]),total=whole(slot?.[2]);
+  if(!(index<total&&total<=NOSTR_MAX_CHUNKS))continue;
+  if(count&&count!==total||parts.has(index)&&parts.get(index)!==event.content)return undefined;
+  count=total;parts.set(index,event.content);
+ }
+ if(!count||parts.size<count)return undefined;
+ const chunks:Uint8Array[]=[];
+ for(let at=0;at<count;at++){const chunk=decode(parts.get(at)!);if(!chunk)return undefined;chunks.push(chunk);}
+ return {kind:'sealed',bytes:concatBytes(...chunks)};
+}

src/rollup/nostr-carrier.ts

@@ -0,0 +1,79 @@
+import {AbstractSimplePool} from 'nostr-tools/abstract-pool';
+import type {Filter} from 'nostr-tools/filter';
+import type {SimplePool} from 'nostr-tools/pool';
+import {getPublicKey,type Event} from 'nostr-tools/pure';
+import {normalizeURL} from 'nostr-tools/utils';
+import {answerEvents,idMatches,NOSTR_RELAYS,NOSTR_REQUEST_KIND,readRequest,roundedSeconds,statusEvent} from '../../packages/protocol/src/nostr-carrier.ts';
+import type {OhttpOpen} from './ohttp-gateway.ts';
+
+export type CarrierPool=Pick<SimplePool,'subscribeMany'|'publish'|'destroy'>;
+// The listener works a relay at a time: subscribeMany remembers every event id it sees, and each id pins the whole relay message it was cut from.
+type Relay={onnotice:(message:string)=>void;subscribe(filters:Filter[],params:{onevent:(event:Event)=>void;onclose:(reason:string)=>void}):{close(reason?:string):void};publish(event:Event):Promise<string>};
+export type RelayPool={ensureRelay(url:string):Promise<Relay>;destroy():void};
+
+// The most body whose sealed answer fits 8 chunks: 63 Padmé steps of 4 KiB (a 64th plus 32 bytes of nonce and tag passes 256 KiB), less 8 bytes of BHTTP framing and 16 396 for the copied headers at their largest, as Node's fetch refuses 16 KiB of header names and values.
+// ponytail: a flat 30 s retry, so a relay that keeps refusing us is asked every 30 s; back off if an operator minds.
+const ANSWER_BYTES=63*4096-8-16_396,REMEMBERED=10_000,STATUSES_PER_SECOND=5,RESUBSCRIBE_MS=30_000;
+
+export function nostrRelaysFromEnv(value:string|undefined):readonly string[] {
+ if(!value?.trim())return NOSTR_RELAYS;
+ const relays=value.split(',').map(relay=>relay.trim());
+ if(!relays.every(relay=>URL.canParse(relay)&&new URL(relay).protocol==='wss:'))throw new Error('SHIELDED_NOSTR_RELAYS is a comma-separated list of wss:// relay URLs.');
+ return [...new Set(relays.map(normalizeURL))];
+}
+
+// The pool's own reconnect is left off: it reopens at since=lastEmitted+1, so one future-dated event deafens a relay, and it never retries a connection that failed first.
+// The pool checks a request's id as readRequest does, never its signature: receive() dedupes by id before readRequest runs.
+export function startNostrCarrier({open,secret,relays,pool=new AbstractSimplePool({verifyEvent:idMatches,maxWaitForConnection:3000}),now=Date.now,maxAnswer=ANSWER_BYTES}:{open:OhttpOpen;secret:Uint8Array;relays:readonly string[];pool?:RelayPool;now?:()=>number;maxAnswer?:number}):{close():void} {
+ const gateway=getPublicKey(secret),tally=new Map<string,number>(),handled=new Map<string,Set<string>>(),live=new Map<string,{close():void}>(),waiting=new Map<string,ReturnType<typeof setTimeout>>();
+ let since=now(),closed=false,statusAt=0,statuses=0;
+ // A throw into the library makes it print the event being handled, so neither counting nor a handler may throw.
+ const note=(kind:string)=>{try{tally.set(kind,(tally.get(kind)??0)+1);if(now()-since<60_000)return;console.log(`[nostr] ${[...tally].map(([name,count])=>`${name} ${count}`).join(', ')}`);tally.clear();since=now();}catch{}};
+ const safely=<A extends unknown[]>(fn:(...args:A)=>void)=>(...args:A)=>{try{fn(...args);}catch{note('failed');}};
+ const relayOf=async(url:string)=>{const relay=await pool.ensureRelay(url);relay.onnotice=()=>{};return relay;};
+ // A flood of shed requests would make the gateway sign and publish a status for each, so they are capped per second (a fixed window on the monotonic clock, like admit()).
+ const statusRoom=()=>{const at=performance.now();if(at-statusAt>=1000){statusAt=at;statuses=0;}return ++statuses<=STATUSES_PER_SECOND;};
+ // A throw can follow the upstream call, so it never becomes a status: a client reads one as "never ran" and may resend a POST.
+ const run=async(event:Event,sealed:Uint8Array,targets:Set<string>)=>{
+  try{
+   const outcome=await open(sealed,{maxAnswer});
+   if(closed)return;
+   if(outcome.kind==='refused'){note('refused');return;}
+   if(outcome.kind==='busy'&&!statusRoom()){note('status-capped');return;}
+   const request={id:event.id,pubkey:event.pubkey},at=now(),answers=outcome.kind==='sealed'?answerEvents(outcome.bytes,request,secret,at):[statusEvent(outcome.status,outcome.retryAfter,request,secret,at)];
+   note(outcome.kind==='sealed'?'answered':'busy');
+   await Promise.all([...targets].map(async url=>{
+    const relay=await relayOf(url).catch(()=>undefined);
+    await Promise.all(answers.map(answer=>(relay?.publish(answer)??Promise.reject()).catch(()=>note('unpublished'))));
+   }));
+  }catch{note('failed');}
+ };
+ // A copy stops here, not at open()'s replay check: it adds its relay to the answer's targets, and open() records an enc only once it has opened, so a copy before that would be admitted again.
+ const receive=(url:string,event:Event)=>{
+  if(closed)return;
+  if(handled.has(event.id)){handled.get(event.id)!.add(url);note('duplicate');return;}
+  const sealed=readRequest(event,gateway);
+  if(!sealed){note('dropped');return;}
+  const targets=new Set([url]);handled.set(event.id,targets);
+  if(handled.size>REMEMBERED)handled.delete(handled.keys().next().value!);
+  void run(event,sealed,targets);
+ };
+ const lost=(url:string)=>{if(closed)return;note('lost');waiting.set(url,setTimeout(()=>void listen(url),RESUBSCRIBE_MS));};
+ const listen=async(url:string)=>{
+  waiting.delete(url);
+  try{
+   // No connect timeout on purpose: the pool's own leaves the half-open socket behind, and undici bounds a connect itself.
+   const relay=await relayOf(url);
+   // limit:0 asks for new events only, so a restart reruns none a relay stored (strfry keeps ephemeral events 300 s); since bounds a relay that ignores it.
+   if(!closed)live.set(url,relay.subscribe([{kinds:[NOSTR_REQUEST_KIND],'#p':[gateway],limit:0,since:roundedSeconds(now())-60}],{onevent:safely(event=>receive(url,event)),onclose:safely(()=>lost(url))}));
+  }catch{lost(url);}
+ };
+ for(const url of new Set(relays))void listen(url);
+ return {close(){
+  closed=true;
+  for(const timer of waiting.values())clearTimeout(timer);
+  waiting.clear();
+  for(const sub of live.values())sub.close();
+  live.clear();pool.destroy();
+ }};
+}