PDS pipethrough optimizations (bluesky-social#2770)

* Micro optimization in request proxying

* Request NSID parsing optimization

* DID document parsing optimization

* remove un-necessary call to next()

* Allow HandlerPipeThrough to be used with streams

* Refactor pipethrough to work with streams

* Expose "unicastLookup" DNS lookup and "isUnicastIp" utilities

* Use a hardened, HTTP2 compatible, client to perform proxied requests

* changeset

* tidy

* Properly handle compressed streams

* tidy

* update @types/node

* refactor

* Improved error management

* Expose parseContentEncoding() util

* use pipeline from nodejs

* Avoid decoding in read-after-write (if possible)

* Various fixes

* Return Buffer instance from streamToBytes

* fixes

* Add omit() utility

* tidy

* lint

* typo

* Use Buffer instead of ArrayBuffer form pipe through handler result

* optimization

* tidy

* refactor

* increase highWaterMark

* remove un-necessary type check

* Use undici.request where more relevant

* Improve soc in fetch utils

* feedback

* fidy

* tidy

* test refactor

* safer fetch

* changeset

* expose and re-use extractUrl util

* small optimizations

* tidy

* optimization

* build branch

---------

Co-authored-by: dholms <[email protected]>

.changeset/calm-shoes-own.md

@@ -0,0 +1,5 @@
+---
+"@atproto/xrpc": patch
+---
+
+Add NotAcceptable response type

.changeset/cold-zebras-peel.md

@@ -0,0 +1,5 @@
+---
+"@atproto/common-web": patch
+---
+
+Add omit() utility

.changeset/dry-laws-jam.md

@@ -0,0 +1,5 @@
+---
+"@atproto-labs/fetch-node": patch
+---
+
+Prevent bypass of ssrf ip verification

.changeset/flat-news-collect.md

@@ -0,0 +1,5 @@
+---
+"@atproto/xrpc-server": patch
+---
+
+Allow HandlerPipeThrough to be used with streams

.changeset/healthy-cats-unite.md

@@ -0,0 +1,5 @@
+---
+"@atproto/pds": patch
+---
+
+Use a hardened, HTTP2 compatible, client to perform proxied requests

.changeset/honest-coats-add.md

@@ -0,0 +1,5 @@
+---
+"@atproto/xrpc-server": minor
+---
+
+Use Buffer instead of ArrayBuffer in pipe through handler result

.changeset/khaki-lemons-joke.md

@@ -0,0 +1,5 @@
+---
+"@atproto/common-web": patch
+---
+
+DID document parsing optimization

.changeset/ninety-bees-beg.md

@@ -0,0 +1,5 @@
+---
+"@atproto/pds": patch
+---
+
+Enable keep-alive for hardened "safe" fetch() agent

.changeset/quick-eyes-destroy.md

@@ -0,0 +1,5 @@
+---
+"@atproto/xrpc-server": patch
+---
+
+Add ErrorOptions support to XRPCError and sub-classes

.changeset/rude-bees-design.md

@@ -0,0 +1,6 @@
+---
+"@atproto/common": patch
+---
+
+add streamToNodeBuffer utility to convert Uint8Array (async) iterables to Buffer
+

.changeset/shy-sloths-doubt.md

@@ -0,0 +1,5 @@
+---
+"@atproto/pds": patch
+---
+
+Improve performances of request proxying by avoiding un-necessary content decoding & buffering.

.changeset/silly-carrots-build.md

@@ -0,0 +1,5 @@
+---
+"@atproto/xrpc-server": patch
+---
+
+Expose parseContentEncoding() util

.changeset/silver-boxes-teach.md

@@ -0,0 +1,5 @@
+---
+"@atproto/pds": patch
+---
+
+Add SSRF protection to request proxying

.changeset/slimy-bananas-promise.md

@@ -0,0 +1,5 @@
+---
+"@atproto/xrpc-server": patch
+---
+
+Properly decompress requests with more than one content encoding

.changeset/stale-bears-talk.md

@@ -0,0 +1,5 @@
+---
+"@atproto-labs/fetch": patch
+---
+
+Expose extractUrl utility

.changeset/sweet-dolls-chew.md

@@ -0,0 +1,5 @@
+---
+"@atproto/xrpc-server": patch
+---
+
+Improve detection, and logging, of 500 server errors.

.changeset/tidy-frogs-peel.md

@@ -0,0 +1,6 @@
+---
+"@atproto-labs/fetch": patch
+---
+
+Add redirectCheckRequestTransform utility to prevent request redirects
+

.changeset/tough-rabbits-wait.md

@@ -0,0 +1,5 @@
+---
+"@atproto-labs/fetch-node": patch
+---
+
+Expose IP filtering utilities

.changeset/warm-cougars-cheat.md

@@ -0,0 +1,5 @@
+---
+"@atproto/xrpc-server": patch
+---
+
+Optimize extraction of NSID from url

.changeset/weak-pants-beg.md

@@ -0,0 +1,5 @@
+---
+"@atproto/oauth-provider": patch
+---
+
+Improve code re-use

.changeset/wet-flowers-kneel.md

@@ -0,0 +1,5 @@
+---
+"@atproto-labs/fetch": patch
+---
+
+Allow customizing fetch logging function

.github/workflows/build-and-push-pds-ghcr.yaml

@@ -3,7 +3,7 @@ on:
   push:
     branches:
       - main
-      - divy/fix-pds-calls-no-body
+      - msieben/micro-optimizations
 env:
   REGISTRY: ghcr.io
   USERNAME: ${{ github.actor }}

package.json

@@ -36,7 +36,7 @@
     "@swc/core": "^1.3.42",
     "@swc/jest": "^0.2.24",
     "@types/jest": "^28.1.4",
-    "@types/node": "^18.19.24",
+    "@types/node": "^18.19.50",
     "@typescript-eslint/eslint-plugin": "^7.4.0",
     "@typescript-eslint/parser": "^7.4.0",
     "dotenv": "^16.0.3",

packages/common-web/src/check.ts

@@ -17,6 +17,11 @@ export const is = <T>(obj: unknown, def: Checkable<T>): obj is T => {
   return def.safeParse(obj).success
 }
 
+export const create =
+  <T>(def: Checkable<T>) =>
+  (v: unknown): v is T =>
+    def.safeParse(v).success
+
 export const assure = <T>(def: Checkable<T>, obj: unknown): T => {
   return def.parse(obj)
 }

packages/common-web/src/did-doc.ts

@@ -17,11 +17,16 @@ export const getDid = (doc: DidDocument): string => {
 
 export const getHandle = (doc: DidDocument): string | undefined => {
   const aka = doc.alsoKnownAs
-  if (!aka) return undefined
-  const found = aka.find((name) => name.startsWith('at://'))
-  if (!found) return undefined
-  // strip off at:// prefix
-  return found.slice(5)
+  if (aka) {
+    for (let i = 0; i < aka.length; i++) {
+      const alias = aka[i]
+      if (alias.startsWith('at://')) {
+        // strip off "at://" prefix
+        return alias.slice(5)
+      }
+    }
+  }
+  return undefined
 }
 
 // @NOTE we parse to type/publicKeyMultibase to avoid the dependency on @atproto/crypto
@@ -35,20 +40,20 @@ export const getVerificationMaterial = (
   doc: DidDocument,
   keyId: string,
 ): { type: string; publicKeyMultibase: string } | undefined => {
-  const did = getDid(doc)
-  let keys = doc.verificationMethod
-  if (!keys) return undefined
-  if (typeof keys !== 'object') return undefined
-  if (!Array.isArray(keys)) {
-    keys = [keys]
+  // /!\ Hot path
+
+  const key = findItemById(doc, 'verificationMethod', `#${keyId}`)
+  if (!key) {
+    return undefined
+  }
+
+  if (!key.publicKeyMultibase) {
+    return undefined
   }
-  const found = keys.find(
-    (key) => key.id === `#${keyId}` || key.id === `${did}#${keyId}`,
-  )
-  if (!found?.publicKeyMultibase) return undefined
+
   return {
-    type: found.type,
-    publicKeyMultibase: found.publicKeyMultibase,
+    type: key.type,
+    publicKeyMultibase: key.publicKeyMultibase,
   }
 }
 
@@ -83,41 +88,68 @@ export const getServiceEndpoint = (
   doc: DidDocument,
   opts: { id: string; type?: string },
 ) => {
-  const did = getDid(doc)
-  let services = doc.service
-  if (!services) return undefined
-  if (typeof services !== 'object') return undefined
-  if (!Array.isArray(services)) {
-    services = [services]
+  // /!\ Hot path
+
+  const service = findItemById(doc, 'service', opts.id)
+  if (!service) {
+    return undefined
   }
-  const found = services.find(
-    (service) => service.id === opts.id || service.id === `${did}${opts.id}`,
-  )
-  if (!found) return undefined
-  if (opts.type && found.type !== opts.type) {
+
+  if (opts.type && service.type !== opts.type) {
     return undefined
   }
-  if (typeof found.serviceEndpoint !== 'string') {
+
+  if (typeof service.serviceEndpoint !== 'string') {
     return undefined
   }
-  return validateUrl(found.serviceEndpoint)
+
+  return validateUrl(service.serviceEndpoint)
+}
+
+function findItemById<
+  D extends DidDocument,
+  T extends 'verificationMethod' | 'service',
+>(doc: D, type: T, id: string): NonNullable<D[T]>[number] | undefined
+function findItemById(
+  doc: DidDocument,
+  type: 'verificationMethod' | 'service',
+  id: string,
+) {
+  // /!\ Hot path
+
+  const items = doc[type]
+  if (items) {
+    for (let i = 0; i < items.length; i++) {
+      const item = items[i]
+      const itemId = item.id
+
+      if (
+        itemId[0] === '#'
+          ? itemId === id
+          : // Optimized version of: itemId === `${doc.id}${id}`
+            itemId.length === doc.id.length + id.length &&
+            itemId[doc.id.length] === '#' &&
+            itemId.endsWith(id) &&
+            itemId.startsWith(doc.id) // <== We could probably skip this check
+      ) {
+        return item
+      }
+    }
+  }
+  return undefined
 }
 
 // Check protocol and hostname to prevent potential SSRF
 const validateUrl = (urlStr: string): string | undefined => {
-  let url
-  try {
-    url = new URL(urlStr)
-  } catch {
+  if (!urlStr.startsWith('http://') && !urlStr.startsWith('https://')) {
     return undefined
   }
-  if (!['http:', 'https:'].includes(url.protocol)) {
-    return undefined
-  } else if (!url.hostname) {
+
+  if (!URL.canParse(urlStr)) {
     return undefined
-  } else {
-    return urlStr
   }
+
+  return urlStr
 }
 
 // Types

packages/common-web/src/util.ts

@@ -9,6 +9,21 @@ export const noUndefinedVals = <T>(
   return obj as Record<string, T>
 }
 
+export function omit<
+  T extends undefined | Record<string, unknown>,
+  K extends keyof NonNullable<T>,
+>(obj: T, keys: readonly K[]): T extends undefined ? undefined : Omit<T, K>
+export function omit(
+  obj: Record<string, unknown>,
+  keys: readonly string[],
+): undefined | Record<string, unknown> {
+  if (!obj) return obj
+
+  return Object.fromEntries(
+    Object.entries(obj).filter((entry) => !keys.includes(entry[0])),
+  )
+}
+
 export const jitter = (maxMs: number) => {
   return Math.round((Math.random() - 0.5) * maxMs * 2)
 }