Properly negotiate response content-encoding (bluesky-social#2852)

* Properly negotiate response content-encoding

* negotiate acceptable encoding and type before building responses

* remove un-necessary async

* typo

* Remove response content-encoding logic

* Avoid using chunked encoding when writing a buffer to the response

.changeset/lazy-trains-leave.md

@@ -0,0 +1,5 @@
+---
+"@atproto/oauth-provider": patch
+---
+
+Remove response content-encoding logic

packages/oauth/oauth-provider/src/assets/assets-middleware.ts

@@ -41,6 +41,6 @@ export function authorizeAssetsMiddleware(): Middleware {
       res.setHeader('Cache-Control', 'public, max-age=31536000, immutable')
     }
 
-    await writeStream(res, asset.createStream(), asset.type)
+    writeStream(res, asset.createStream(), { contentType: asset.type })
   }
 }

packages/oauth/oauth-provider/src/lib/http/middleware.ts

@@ -95,7 +95,7 @@ export function createFinalHandler(
     res.setHeader('Content-Security-Policy', "default-src 'none'")
     res.setHeader('X-Content-Type-Options', 'nosniff')
 
-    writeJson(res, payload, status)
+    writeJson(res, payload, { status })
   }
 }
 

packages/oauth/oauth-provider/src/lib/http/response.ts

@@ -1,6 +1,4 @@
-import { PassThrough, Readable, Transform } from 'node:stream'
-import { pipeline } from 'node:stream/promises'
-import { constants, createBrotliCompress, createGzip } from 'node:zlib'
+import { Readable, pipeline } from 'node:stream'
 
 import { Handler, ServerResponse } from './types.js'
 
@@ -26,108 +24,69 @@ export function writeRedirect(
   res.writeHead(status, { Location: url }).end()
 }
 
-function negotiateEncoding(accept?: string | string[]) {
-  if (accept?.includes('br')) return 'br'
-  if (accept?.includes('gzip')) return 'gzip'
-  return 'identity'
+export type WriteResponseOptions = {
+  status?: number
+  contentType?: string
 }
 
-function getEncoder(encoding: string): Transform {
-  switch (encoding) {
-    case 'br':
-      return createBrotliCompress({
-        // Default quality is too slow
-        params: { [constants.BROTLI_PARAM_QUALITY]: 5 },
-      })
-    case 'gzip':
-      return createGzip()
-    case 'identity':
-      return new PassThrough()
-    default:
-      throw new Error(`Unsupported encoding: ${encoding}`)
-  }
-}
-
-const ifString = (value: unknown): string | undefined =>
-  typeof value === 'string' ? value : undefined
-
-export async function writeStream(
+export function writeStream(
   res: ServerResponse,
   stream: Readable,
-  contentType = ifString((stream as any).headers?.['content-type']) ||
-    'application/octet-stream',
-  status = 200,
-): Promise<void> {
+  {
+    status = 200,
+    contentType = 'application/octet-stream',
+  }: WriteResponseOptions = {},
+): void {
   res.statusCode = status
   res.setHeader('content-type', contentType)
-  appendHeader(res, 'vary', 'accept-encoding')
-
-  const encoding = negotiateEncoding(res.req.headers['accept-encoding'])
-
-  res.setHeader('content-encoding', encoding)
-  res.setHeader('transfer-encoding', 'chunked')
 
   if (res.req.method === 'HEAD') {
     res.end()
     stream.destroy()
-    return
-  }
-
-  try {
-    await pipeline(stream, getEncoder(encoding), res)
-  } catch (err) {
-    // Prevent the socket from being left open in a bad state
-    res.socket?.destroy()
-
-    if (err != null && typeof err === 'object') {
-      // If an abort signal is used, we can consider this function's job successful
-      if ('name' in err && err.name === 'AbortError') return
-
-      // If the client closes the connection, we don't care about the error
-      if ('code' in err && err.code === 'ERR_STREAM_PREMATURE_CLOSE') return
-    }
-
-    throw err
+  } else {
+    pipeline([stream, res], (_err: Error | null) => {
+      // The error will be propagated through the streams
+    })
   }
 }
 
-export async function writeBuffer(
+export function writeBuffer(
   res: ServerResponse,
-  buffer: Buffer,
-  contentType?: string,
-  status = 200,
-): Promise<void> {
-  const stream = Readable.from([buffer])
-  return writeStream(res, stream, contentType, status)
+  chunk: string | Buffer,
+  {
+    status = 200,
+    contentType = 'application/octet-stream',
+  }: WriteResponseOptions = {},
+): void {
+  res.statusCode = status
+  res.setHeader('content-type', contentType)
+  res.end(chunk)
 }
 
-export async function writeJson(
+export function writeJson(
   res: ServerResponse,
   payload: unknown,
-  status = 200,
-  contentType = 'application/json',
-): Promise<void> {
+  { contentType = 'application/json', ...options }: WriteResponseOptions = {},
+): void {
   const buffer = Buffer.from(JSON.stringify(payload))
-  return writeBuffer(res, buffer, contentType, status)
+  writeBuffer(res, buffer, { ...options, contentType })
 }
 
-export function staticJsonHandler(
+export function staticJsonMiddleware(
   value: unknown,
-  contentType = 'application/json',
-  status = 200,
+  { contentType = 'application/json', ...options }: WriteResponseOptions = {},
 ): Handler<unknown> {
   const buffer = Buffer.from(JSON.stringify(value))
-  return function (req, res, next) {
-    void writeBuffer(res, buffer, contentType, status).catch(next)
+  const staticOptions: WriteResponseOptions = { ...options, contentType }
+  return function (req, res) {
+    writeBuffer(res, buffer, staticOptions)
   }
 }
 
-export async function writeHtml(
+export function writeHtml(
   res: ServerResponse,
   html: Buffer | string,
-  status = 200,
-  contentType = 'text/html',
-): Promise<void> {
-  const buffer = Buffer.isBuffer(html) ? html : Buffer.from(html)
-  return writeBuffer(res, buffer, contentType, status)
+  { contentType = 'text/html', ...options }: WriteResponseOptions = {},
+): void {
+  writeBuffer(res, html, { ...options, contentType })
 }

packages/oauth/oauth-provider/src/oauth-provider.ts

@@ -29,6 +29,8 @@ import {
   oauthTokenIdentificationSchema,
   oauthTokenRequestSchema,
 } from '@atproto/oauth-types'
+import { mediaType } from '@hapi/accept'
+import createHttpError from 'http-errors'
 import type { Redis, RedisOptions } from 'ioredis'
 import z, { ZodError } from 'zod'
 
@@ -75,7 +77,7 @@ import {
   combineMiddlewares,
   parseHttpRequest,
   setupCsrfToken,
-  staticJsonHandler,
+  staticJsonMiddleware,
   validateCsrfToken,
   validateFetchDest,
   validateFetchMode,
@@ -1028,7 +1030,7 @@ export class OAuthProvider extends OAuthVerifier {
           res.setHeader('Cache-Control', 'max-age=300')
           next()
         },
-        staticJsonHandler(json),
+        staticJsonMiddleware(json),
       ])
 
     /**
@@ -1056,9 +1058,18 @@ export class OAuthProvider extends OAuthVerifier {
         }
 
         try {
+          // Ensure we can agree on a content encoding & type before starting to
+          // build the JSON response.
+          if (!mediaType(req.headers['accept'], ['application/json'])) {
+            throw createHttpError(406, 'Unsupported media type')
+          }
+
           const result = await buildJson.call(this, req, res)
-          if (result !== undefined) writeJson(res, result, status)
-          else if (!res.headersSent) res.writeHead(status ?? 204).end()
+          if (result !== undefined) {
+            writeJson(res, result, { status })
+          } else if (!res.headersSent) {
+            res.writeHead(status ?? 204).end()
+          }
         } catch (err) {
           if (!res.headersSent) {
             if (err instanceof WWWAuthenticateError) {
@@ -1067,7 +1078,9 @@ export class OAuthProvider extends OAuthVerifier {
               res.appendHeader('Access-Control-Expose-Headers', name)
             }
 
-            writeJson(res, buildErrorPayload(err), buildErrorStatus(err))
+            const payload = buildErrorPayload(err)
+            const status = buildErrorStatus(err)
+            writeJson(res, payload, { status })
           } else {
             res.destroy()
           }
@@ -1096,11 +1109,6 @@ export class OAuthProvider extends OAuthVerifier {
           validateSameOrigin(req, res, issuerOrigin)
 
           await handler.call(this, req, res)
-
-          // Should never happen (fool proofing)
-          if (!res.headersSent) {
-            throw new Error('Navigation handler did not send a response')
-          }
         } catch (err) {
           onError?.(
             req,

packages/oauth/oauth-provider/src/output/send-web-page.ts

@@ -8,7 +8,7 @@ import {
   Html,
   js,
 } from '../lib/html/index.js'
-import { writeHtml } from '../lib/http/response.js'
+import { writeHtml, WriteResponseOptions } from '../lib/http/response.js'
 
 export function declareBackendData(name: string, data: unknown) {
   // The script tag is removed after the data is assigned to the global variable
@@ -18,9 +18,11 @@ export function declareBackendData(name: string, data: unknown) {
   return js`window[${name}]=${data};document.currentScript.remove();`
 }
 
+export type SendWebPageOptions = BuildDocumentOptions & WriteResponseOptions
+
 export async function sendWebPage(
   res: ServerResponse,
-  { status = 200, ...options }: BuildDocumentOptions & { status?: number },
+  options: SendWebPageOptions,
 ): Promise<void> {
   // @TODO: make these headers configurable (?)
   res.setHeader('Permissions-Policy', 'otp-credentials=*, document-domain=()')
@@ -53,7 +55,7 @@ export async function sendWebPage(
 
   const html = buildDocument(options)
 
-  return writeHtml(res, html.toString(), status)
+  return writeHtml(res, html.toString(), options)
 }
 
 function assetToHash(asset: Html | AssetRef): string {