websocket server proxy

lib/guard/index.ts

@@ -0,0 +1,29 @@
+import * as jose from 'jose'
+
+const alg = 'ES256'
+// const options = {
+//     issuer: "urn:example:issuer",
+//     audience: "urn:example:audience",
+// };
+const keys = await jose.generateKeyPair(alg, { extractable: true })
+const keys2 = await jose.generateKeyPair(alg, { extractable: true })
+// const keys3 = await jose.generateKeyPair(alg, { extractable: true });
+
+const jwk1 = await jose.exportJWK(keys.publicKey).then((jwk) => {
+  jwk.kid = 'key1'
+
+  return jwk
+})
+
+const jwk2 = await jose.exportJWK(keys2.publicKey).then((jwk) => {
+  jwk.kid = 'key2'
+
+  return jwk
+})
+
+// const jwk3 = await jose.exportJWK(keys3.publicKey).then((jwk) => {
+//     jwk.kid = "key3";
+//     return jwk;
+// });
+
+export const jwks = jose.createLocalJWKSet({ keys: [jwk1, jwk2] })

lib/guard/keys.ts

@@ -0,0 +1,7 @@
+import * as jose from 'jose'
+
+export const alg = 'ES256'
+
+export const serverKeys = await jose.generateKeyPair(alg, { extractable: true })
+export const userKeys1 = await jose.generateKeyPair(alg, { extractable: true })
+export const userKeys3 = await jose.generateKeyPair(alg, { extractable: true })

lib/guard/t1.ts

@@ -0,0 +1,28 @@
+import * as jose from 'jose'
+
+const alg = 'ES256'
+
+const keys = await jose.generateKeyPair(alg, { extractable: true })
+const keys2 = await jose.generateKeyPair(alg, { extractable: true })
+
+console.log('keys\n', await jose.exportPKCS8(keys.privateKey))
+console.log('keys2\n', await jose.exportPKCS8(keys2.privateKey))
+
+const jwt = await new jose.GeneralSign(
+  new TextEncoder().encode('Hello, World!'),
+)
+  .addSignature(keys.privateKey)
+  .setProtectedHeader({ alg, b64: true })
+  .addSignature(keys2.privateKey)
+  .setProtectedHeader({ alg, b64: true })
+  .sign()
+
+console.log('jwt', jwt)
+
+const { payload, protectedHeader } = await jose.generalVerify(
+  jwt,
+  keys2.publicKey,
+)
+
+console.log(protectedHeader)
+console.log(new TextDecoder().decode(payload))

lib/guard/t2.ts

@@ -0,0 +1,51 @@
+import * as jose from 'jose'
+
+const alg = 'ES256'
+const options = {
+  issuer: 'urn:example:issuer',
+  audience: 'urn:example:audience',
+}
+
+const keys = await jose.generateKeyPair(alg, { extractable: true })
+const keys2 = await jose.generateKeyPair(alg, { extractable: true })
+// const keys3 = await jose.generateKeyPair(alg, { extractable: true });
+
+const jwk1 = await jose.exportJWK(keys.publicKey).then((jwk) => {
+  jwk.kid = 'key1'
+
+  return jwk
+})
+
+const jwk2 = await jose.exportJWK(keys2.publicKey).then((jwk) => {
+  jwk.kid = 'key2'
+
+  return jwk
+})
+
+// const jwk3 = await jose.exportJWK(keys3.publicKey).then((jwk) => {
+//     jwk.kid = "key3";
+//     return jwk;
+// });
+
+const jwks = jose.createLocalJWKSet({ keys: [jwk1, jwk2] })
+
+console.log('keys\n', await jose.exportPKCS8(keys.privateKey))
+console.log('keys2\n', await jose.exportPKCS8(keys2.privateKey))
+
+const jwt = await new jose.SignJWT({
+  foo: 'bar',
+})
+  .setIssuer(options.issuer)
+  .setAudience(options.audience)
+  .setProtectedHeader({ alg, kid: 'key1' })
+  .setExpirationTime('10m')
+  .setIssuedAt()
+  .sign(keys.privateKey)
+
+console.log('jwt', jwt)
+
+const { payload, protectedHeader } = await jose
+  .jwtVerify(jwt, jwks, options)
+
+console.log(protectedHeader)
+console.log(payload)

lib/ws/server.ts

@@ -9,43 +9,56 @@ import type { ServerOptions, WebSocket } from 'ws'
 const logger = consola.withTag('ws/server')
 
 type BufferLike =
-    | string
-    | Buffer
-    | DataView
-    | number
-    | ArrayBufferView
-    | Uint8Array
-    | ArrayBuffer
-    | SharedArrayBuffer
-    | readonly any[]
-    | readonly number[]
-    | { valueOf(): ArrayBuffer }
-    | { valueOf(): SharedArrayBuffer }
-    | { valueOf(): Uint8Array }
-    | { valueOf(): readonly number[] }
-    | { valueOf(): string }
-    | { [Symbol.toPrimitive](hint: string): string }
+  | string
+  | Buffer
+  | DataView
+  | number
+  | ArrayBufferView
+  | Uint8Array
+  | ArrayBuffer
+  | SharedArrayBuffer
+  | readonly any[]
+  | readonly number[]
+  | { valueOf(): ArrayBuffer }
+  | { valueOf(): SharedArrayBuffer }
+  | { valueOf(): Uint8Array }
+  | { valueOf(): readonly number[] }
+  | { valueOf(): string }
+  | { [Symbol.toPrimitive](hint: string): string }
 
 export class WebSocketProxy<
-T extends typeof WebSocket.WebSocket = typeof WebSocket.WebSocket,
-U extends typeof IncomingMessage = typeof IncomingMessage,
+  T extends typeof WebSocket.WebSocket = typeof WebSocket.WebSocket,
+  U extends typeof IncomingMessage = typeof IncomingMessage,
 > extends WebSocketServer {
   constructor(
-    options: ServerOptions<T, U>,
+    options?: ServerOptions<T, U>,
     callback?: () => void,
   ) {
     super(options, callback)
+
     logger.info('Created')
-    const wrappedMethods = {
-      on: this.customOn.bind(this),
-      send: this.customSend.bind(this),
-    }
+    // const wrappedMethods = {
+    //   on: this.customOn.bind(this),
+    //   // send: this.customSend.bind(this),
+    // }
+
+    return this.getProxy(this)
+  }
+
+  private getWrappedMethod<S>(target: S, prop: string) {
+    const methods = {
+      on: this.customOn.bind(target),
+      send: this.customSend.bind(target),
+    } as Record<string, (...args: any[]) => void>
+
+    return methods[prop]
+  }
 
-    return new Proxy(this, {
+  private getProxy<S extends object>(t: S) {
+    return new Proxy(t, {
       get: (target, prop, receiver) => {
-        logger.info('123', prop)
-        if (prop === 'on' || prop === 'send') {
-          return wrappedMethods[prop]
+        if (prop === 'on') {
+          return this.getWrappedMethod(target, prop)
         }
 
         return Reflect.get(target, prop, receiver)
@@ -57,11 +70,18 @@ U extends typeof IncomingMessage = typeof IncomingMessage,
     event: string,
     listener: (this: WebSocketServer, ...args: any[]) => void,
   ) {
-    console.log('event', event)
-
     this.on(event, async (...args: any[]) => {
+      if (event === 'connection') {
+        const arg = args as [socket: InstanceType<T>, request: InstanceType<U>]
+        const ws = this.getProxy(arg[0])
+        arg[0] = ws
+
+        listener.call(this, ...arg)
+      }
+
       if (event === 'message') {
         await sleep(100)
+        logger.info('Receiving', event, JSON.parse(args[0] as string))
         logger.debug('Receiving', event, JSON.parse(args[0] as string))
         const [data, isBinary] = args as [BufferLike, boolean]
         listener.call(this, data, isBinary)
@@ -75,6 +95,7 @@ U extends typeof IncomingMessage = typeof IncomingMessage,
 
   private async customSend(event: string | symbol, data: BufferLike) {
     await sleep(100)
+    logger.info('Sending', event)
     logger.debug('Sending', event)
     this.emit(event, data)
   }

package.json

@@ -66,6 +66,7 @@
     "dotenv": "^16.4.5",
     "fast-glob": "^3.3.2",
     "h3": "^1.11.1",
+    "jose": "^5.2.3",
     "listhen": "^1.7.2",
     "nanoid": "^5.0.6",
     "redis": "^4.6.13",

scripts/quicktype.ts

@@ -27,19 +27,6 @@ await Promise.all(files.map(async (file) => {
   })
 }))
 
-// inputData.addInput(jsonInput)
-// const filesRendered = await quicktypeMultiFile({
-//   inputData,
-//   lang,
-//   outputFilename: 'index',
-//   rendererOptions: {
-//     'just-types': true,
-//     'prefer-types': true,
-//     'prefer-unions': true,
-//     'declare-unions': true,
-//   },
-// })
-
 const filesRendered = await quicktypeMultipleJSONSchema(lang, data, {
   rendererOptions: {
     'just-types': true,

src/server/routers/data.ts

@@ -38,8 +38,8 @@ export const dataRouter = rootRouter({
       )
 
       logger.info(`Job ${job.id} added:`, JSON.stringify(data, null, 2))
-      const returnvalue = await job.waitUntilFinished(queueEvents)
-      logger.success(`Job ${job.id} result:`, returnvalue)
+      // const returnvalue = await job.waitUntilFinished(queueEvents)
+      // logger.success(`Job ${job.id} result:`, returnvalue)
 
       return redis.data.insertOne(id, data)
     }),
@@ -55,8 +55,8 @@ export const dataRouter = rootRouter({
         const job = await bullmq.getJob(jobId)
         if (
           job
-            && job.returnvalue
-            && job.returnvalue.status % n === 0
+          && job.returnvalue
+          && job.returnvalue.status % n === 0
         ) {
           emit.next(job.returnvalue)
         }

yarn.lock

@@ -1104,6 +1104,7 @@ __metadata:
     esno: "npm:^4.0.0"
     fast-glob: "npm:^3.3.2"
     h3: "npm:^1.11.1"
+    jose: "npm:^5.2.3"
     json-schema-migrate: "npm:^2.0.0"
     lint-staged: "npm:^15.2.2"
     listhen: "npm:^1.7.2"
@@ -3645,6 +3646,13 @@ __metadata:
   languageName: node
   linkType: hard
 
+"jose@npm:^5.2.3":
+  version: 5.2.3
+  resolution: "jose@npm:5.2.3"
+  checksum: 10/33056092ed6c1539eee6c0c75bd3664a39c639e02f21ec5cb32ddd231b9345f239e7bffe3f4741c903c711e237ceafb816ed634bd9cd5e4fb61b5a1f067e2c40
+  languageName: node
+  linkType: hard
+
 "js-base64@npm:^3.7.5":
   version: 3.7.7
   resolution: "js-base64@npm:3.7.7"