Pipe/stream (#3)

* Experimental: added /stream/:key path

* Stream tokens stored as LIFO. Excess tokens trimmed. Unnecessary await omitted.

* pipe

* stream path is rate-limited if not requested by middleware

* STREAM_COUNT => MAX_STREAM_COUNT

Author: SomajitDey

api/helper.js

@@ -12,11 +12,24 @@ const sigLen = parseInt(process.env.SIG_LEN);
 const hashLen = parseInt(process.env.HASH_LEN);
 const ttl = parseInt(process.env.TTL);
 const cacheTtl = parseInt(process.env.CACHE_TTL);
+const streamTimeout = parseInt(process.env.STREAM_TIMEOUT);
+const maxStreamCount = parseInt(process.env.MAX_STREAM_COUNT);
+
 const dbKeyPrefix = {
                 manyToOne: "m2o:",
                 oneToMany: "o2m:",
                 oneToOne: "o2o:",
-                cache: "cache:"
+                cache: "cache:",
+                stream: {
+                  public: {
+                    send: "pipePubSend:",
+                    receive: "pipePubRecv:"
+                  },
+                  private: {
+                    send: "pipePrivSend:",
+                    receive: "pipePrivRecv:"
+                  }
+                }
             }
 // Redis client for user database
 const redisData = new Redis({
@@ -44,9 +57,9 @@ function sign(str){
     return createHmac('md5', secret).update(str).digest('base64url').substring(0,sigLen);
 }
 
-//Brief: Return random base64url string of given length
+// Brief: Return random base64url string of given length
 function randStr(len = hashLen){
-  const byteSize = Math.ceil(len*6/8);
+  const byteSize = Math.ceil(len*6/8); // base64 char is 6 bit, byte is 8
   const buff = Buffer.alloc(byteSize);
   getRandomValues(buff);
   return buff.toString('base64url').substring(0,len);
@@ -56,14 +69,15 @@ export function id(){
     return sign('id');
 }
 
-export function validate(key){
+export function validate(key, silent=true){
     const sig = key.substring(0, sigLen);
     const hash = key.substring(sigLen);
     if (sig === sign(hash + 'public')){
         return 'public';
     } else if (sig === sign(hash + 'private')){
         return 'private';
     } else {
+        if (!silent) throw new Error('Invalid Key');
         return false;
     }
 }
@@ -199,3 +213,22 @@ export async function oneToOneTTL(privateKey, key){
     ])
     return {ttl: bool ? ttl : 0};
 }
+
+// Tokens are stored in LIFO stacks. Old and unused tokens are trimmed.
+export async function streamToken(privateOrPublicKey, receive=true){
+  const type = validate(privateOrPublicKey, false);
+  const typeComplement = (type == 'private') ? 'public' : 'private';
+  const publicKey = genPublicKey(privateOrPublicKey);
+  const mode = receive ? "receive" : "send";
+  const modeComplement = receive ? "send" : "receive";
+  const existing = await redisData.lpop(dbKeyPrefix.stream[typeComplement][modeComplement] + publicKey);
+  if (existing) return existing;
+  const token = randStr();
+  const dbKey = dbKeyPrefix.stream[type][mode] + publicKey;
+  Promise.all([
+    redisData.lpush(dbKey, token),
+    redisData.expire(dbKey, streamTimeout),
+    redisData.ltrim(dbKey, 0, maxStreamCount - 1)
+  ])
+  return token;
+}

api/index.js

@@ -15,7 +15,7 @@ const fastify = Fastify({
 fastify.addHook('onRequest', (request, reply, done) => {
     reply.headers({
         'Access-Control-Allow-Origin': '*',
-        'Access-Control-Allow-Methods': 'GET,POST,DELETE'
+        'Access-Control-Allow-Methods': 'GET,POST,PATCH,DELETE'
     })
     done();
 })
@@ -76,6 +76,7 @@ fastify.post('/public/:publicKey', async (request, reply) => {
 
             await fetch(webhook, {
                 method: "POST",
+                redirect: "follow",
                 headers: { "Content-type": "application/json" },
                 body: data,
                 signal: AbortSignal.timeout(webhookTimeout)
@@ -290,6 +291,36 @@ fastify.get('/private/:privateKey/:key', async (request, reply) => {
     }    
 })
 
+const streamHandler = async (request, reply) => {
+  const { key } = request.params;
+  try {
+    let recvBool;
+    switch (request.method) {
+      case 'POST': // Using fallthrough! POST and PUT cases run the same code.
+      case 'PUT':
+        recvBool = false;
+        break;
+      case 'GET':
+        recvBool = true;
+        break;
+      default:
+        throw new Error('Unsupported Method');
+    }
+    const token = await helper.streamToken(key, recvBool);
+    reply.redirect('https://ppng.io/' + token, 307);
+  } catch (err) {
+    if (err.message == 'Invalid Key') {
+      callUnauthorized(reply, 'Provided key is invalid');
+    } else if (err.message == 'Unsupported Method') {
+      callBadRequest(reply, 'Unsupported method');
+    } else {
+      callInternalServerError(reply, err);
+    }
+  }
+}
+
+fastify.all('/stream/:key', streamHandler);
+
 export default async function handler(req, res) {
   await fastify.ready();
   fastify.server.emit('request', req, res);

api/test.js

@@ -14,6 +14,12 @@ console.log('This should show public: ' + helper.validate(key.public));
 console.log('This should show private: ' + helper.validate(key.private));
 console.log('This should show false: ' + helper.validate('random'));
 
+console.log('Stream token private POST:', await helper.streamToken(key.private, false));
+console.log('Stream token public GET:', await helper.streamToken(key.public));
+console.log('Stream token public POST:', await helper.streamToken(key.public, false));
+console.log('Stream token public POST:', await helper.streamToken(key.public, false));
+console.log('Stream token private GET:', await helper.streamToken(key.private, true));
+
 await helper.publicProduce(key.public, 'dataA hi"');
 await helper.publicProduce(key.public, 'dataB hi"');
 await helper.publicProduce(key.public, 'dataC hi"');

example.env

@@ -14,6 +14,8 @@ RATELIMIT=20 # Max number of requests within following time period
 RATELIMIT_WINDOW=300
 CACHE_TTL=86400 # TTL for cache in seconds
 WEBHOOK_TIMEOUT=4000 # Request timeout for POST to webhook, in milliseconds
+STREAM_TIMEOUT=60 # TTL of stream/pipe tokens in seconds
+MAX_STREAM_COUNT=5 # Max number of stream/pipe tokens in memory
 
 # Redis credentials to be used for rate-limiting
 KV_REST_API_URL=

middleware.js

@@ -12,6 +12,12 @@ import { next } from '@vercel/edge';
 import { Ratelimit } from '@upstash/ratelimit';
 import { Redis } from '@upstash/redis';
 
+export const config = {
+  matcher: ['/public/:path*', '/private/:path*', '/keys/:path*', '/pipe/:path*', '/stream/:path*']
+};
+
+const middlewareSig = process.env.SECRET; // Secret known to middleware only
+
 const cache = new Map(); // must be outside of your serverless function handler
 
 const ratelimit = new Ratelimit({
@@ -26,17 +32,40 @@ const ratelimit = new Ratelimit({
   enableProtection: true
 })
 
+// Forwards requests at path /pipe/* to /stream/* in index.js after trimming the request body and headers
+// and returns the response. This is done in middleware.js because, unlike index.js, middleware.js doesn't have
+// problems with the `Expect: 100-continue` header sent by `curl -T- <url>`. middleware.js also doesn't read the
+// request body. index.js also has a set bodyLimit which is incompatible with payloads at /pipe/* of arbitrary size.
+// /stream/ path is exposed only to middleware.
+// For requests not at path /pipe/*, returns null.
+async function pipeToStream(request) {
+  const pipeUrl = new URL(request.url);
+  if (!pipeUrl.pathname.startsWith('/pipe/')) return null;
+  const streamUrl = pipeUrl.href.replace('/pipe/', '/stream/');
+  return fetch(streamUrl, { method: request.method, headers: { 'x-middleware' : middlewareSig }, redirect: 'manual' });
+}
+
 export default async function middleware(request) {
-  // You could alternatively limit based on user ID or similar
+  const fromMiddleware = request.headers.get('x-middleware') === middlewareSig;
+
+  // You could alternatively rate-limit based on user ID or similar
   const ip = ipAddress(request) || '127.0.0.1';
-  const { success, reset } = await ratelimit.limit(ip);
+  // For requests from middleware, no rate-limiting is necessary
+  const { success, reset } = fromMiddleware ? { success: true, reset: 0 } : await ratelimit.limit(ip);
 
-  return success ? next() : Response.json(
+  if (success) {
+    const streamResponse = await pipeToStream(request);
+    // For non-pipe requests, streamResponse is null.
+    return streamResponse ?? next();
+  }
+  else {
+    return Response.json(
     { message: `Try after ${(reset - Date.now())/1000} seconds`, error: "Too Many Requests", statusCode: 429 },
     {
       status: 429,
       statusText: "Too Many Requests",
       headers: {"Access-Control-Allow-Origin":"*"}
     },
   )
+  }
 }

package.json

@@ -11,6 +11,7 @@
   },
   "type": "module",
   "scripts": {
-    "test": "node --env-file=.env api/test.js"
+    "test": "node --env-file=.env api/test.js",
+    "lint": "semistandard --fix"
   }
 }

vercel.json

@@ -19,7 +19,7 @@
   ],
   "rewrites": [
     {
-      "source": "/(public|private|keys)/:path*",
+      "source": "/(public|private|keys|stream)/:path*",
       "destination": "/api"
     },
     {