Merge pull request #1742 from bluesky-social/multi-pds-auth-tests

Misc tests and fixes for entryway PDS

packages/dev-env/src/network-no-appview.ts

@@ -28,7 +28,7 @@ export class TestNetworkNoAppView {
       ...params.pds,
     })
 
-    mockNetworkUtilities(pds)
+    mockNetworkUtilities([pds])
 
     return new TestNetworkNoAppView(plc, pds)
   }

packages/dev-env/src/network.ts

@@ -54,7 +54,7 @@ export class TestNetwork extends TestNetworkNoAppView {
       ...params.pds,
     })
 
-    mockNetworkUtilities(pds, bsky)
+    mockNetworkUtilities([pds], bsky)
 
     return new TestNetwork(plc, pds, bsky)
   }

packages/dev-env/src/util.ts

@@ -2,15 +2,17 @@ import { IdResolver } from '@atproto/identity'
 import { TestPds } from './pds'
 import { TestBsky } from './bsky'
 
-export const mockNetworkUtilities = (pds: TestPds, bsky?: TestBsky) => {
-  mockResolvers(pds.ctx.idResolver, pds)
+export const mockNetworkUtilities = (pdses: TestPds[], bsky?: TestBsky) => {
+  for (const pds of pdses) {
+    mockResolvers(pds.ctx.idResolver, pdses)
+  }
   if (bsky) {
-    mockResolvers(bsky.ctx.idResolver, pds)
-    mockResolvers(bsky.indexer.ctx.idResolver, pds)
+    mockResolvers(bsky.ctx.idResolver, pdses)
+    mockResolvers(bsky.indexer.ctx.idResolver, pdses)
   }
 }
 
-export const mockResolvers = (idResolver: IdResolver, pds: TestPds) => {
+export const mockResolvers = (idResolver: IdResolver, pdses: TestPds[]) => {
   // Map pds public url to its local url when resolving from plc
   const origResolveDid = idResolver.did.resolveNoCache
   idResolver.did.resolveNoCache = async (did: string) => {
@@ -20,29 +22,37 @@ export const mockResolvers = (idResolver: IdResolver, pds: TestPds) => {
     ) as ReturnType<typeof origResolveDid>)
     const service = result?.service?.find((svc) => svc.id === '#atproto_pds')
     if (typeof service?.serviceEndpoint === 'string') {
-      service.serviceEndpoint = service.serviceEndpoint.replace(
-        pds.ctx.cfg.service.publicUrl,
-        `http://localhost:${pds.port}`,
-      )
+      for (const pds of pdses) {
+        service.serviceEndpoint = service.serviceEndpoint.replace(
+          pds.ctx.cfg.service.publicUrl,
+          `http://localhost:${pds.port}`,
+        )
+      }
     }
     return result
   }
 
   const origResolveHandleDns = idResolver.handle.resolveDns
   idResolver.handle.resolve = async (handle: string) => {
-    const isPdsHandle = pds.ctx.cfg.identity.serviceHandleDomains.some(
-      (domain) => handle.endsWith(domain),
-    )
-    if (!isPdsHandle) {
+    const eligiblePdses = pdses.filter((pds) => {
+      return pds.ctx.cfg.identity.serviceHandleDomains.some((domain) =>
+        handle.endsWith(domain),
+      )
+    })
+
+    if (!eligiblePdses.length) {
       return origResolveHandleDns.call(idResolver.handle, handle)
     }
 
-    const url = `${pds.url}/.well-known/atproto-did`
-    try {
-      const res = await fetch(url, { headers: { host: handle } })
-      return await res.text()
-    } catch (err) {
-      return undefined
+    for (const pds of eligiblePdses) {
+      const url = `${pds.url}/.well-known/atproto-did`
+      try {
+        const res = await fetch(url, { headers: { host: handle } })
+        if (res.status !== 200) continue
+        return await res.text()
+      } catch {
+        // ignore
+      }
     }
   }
 }

packages/pds/src/api/com/atproto/repo/uploadBlob.ts

@@ -18,7 +18,10 @@ export default function (server: Server, ctx: AppContext) {
         async (agent) => {
           const result = await agent.api.com.atproto.repo.uploadBlob(
             await streamToBytes(input.body), // @TODO proxy streaming
-            authPassthru(req, true),
+            {
+              ...authPassthru(req),
+              encoding: input.encoding,
+            },
           )
           return resultPassthru(result)
         },

packages/pds/src/api/com/atproto/server/createAccount.ts

@@ -1,4 +1,3 @@
-import { randomInt } from 'node:crypto'
 import { InvalidRequestError } from '@atproto/xrpc-server'
 import disposable from 'disposable-email'
 import * as plc from '@did-plc/lib'
@@ -263,7 +262,18 @@ const getDidAndPlcOp = async (
 // @TODO this implementation is a stub
 const assignPds = async (ctx: AppContext) => {
   const pdses = await ctx.db.db.selectFrom('pds').selectAll().execute()
-  if (!pdses.length) return
-  const pds = pdses.at(randomInt(pdses.length))
-  return pds
+  const idx = randomIndexByWeight(pdses.map((pds) => pds.weight))
+  if (idx === -1) return
+  return pdses.at(idx)
+}
+
+const randomIndexByWeight = (weights) => {
+  let sum = 0
+  const cumulative = weights.map((weight) => {
+    sum += weight
+    return sum
+  })
+  if (!sum) return -1
+  const rand = Math.random() * sum
+  return cumulative.findIndex((item) => item >= rand)
 }

packages/pds/src/api/proxy.ts

@@ -1,15 +1,19 @@
 import * as express from 'express'
 import AtpAgent from '@atproto/api'
 import { Headers, XRPCError } from '@atproto/xrpc'
-import { InvalidRequestError, UpstreamFailureError } from '@atproto/xrpc-server'
+import {
+  AuthRequiredError,
+  InvalidRequestError,
+  UpstreamFailureError,
+} from '@atproto/xrpc-server'
 import AppContext from '../context'
 
 export const proxy = async <T>(
   ctx: AppContext,
   pdsDid: string | null | undefined,
   fn: (agent: AtpAgent) => Promise<T>,
 ): Promise<T | null> => {
-  if (isThisPds(ctx, pdsDid)) {
+  if (isThisPds(ctx, pdsDid) || !ctx.cfg.service.isEntryway) {
     return null // skip proxying
   }
   const accountService = ctx.services.account(ctx.db)
@@ -18,7 +22,11 @@ export const proxy = async <T>(
     throw new UpstreamFailureError('unknown pds')
   }
   // @TODO reuse agents
-  const agent = new AtpAgent({ service: `https://${pds.host}` })
+  const service = new URL(`https://${pds.host}`)
+  if (service.hostname === 'localhost') {
+    service.protocol = 'http:'
+  }
+  const agent = new AtpAgent({ service })
   try {
     return await fn(agent)
   } catch (err) {
@@ -46,14 +54,19 @@ export const isThisPds = (
 }
 
 // @NOTE on the identity service this serves a 400 w/ ExpiredToken to prompt a refresh flow from the client.
-// but on our other PDSes the same case should be a 403 w/ AccountNotFound, assuming their access token verifies.
+// the analogous case on our non-entryway PDSes would be a 403 w/ AccountNotFound when the user can auth but doesn't have an account.
 export const ensureThisPds = (ctx: AppContext, pdsDid: string | null) => {
   if (!isThisPds(ctx, pdsDid)) {
-    // instruct client to refresh token during potential account migration
-    throw new InvalidRequestError(
-      'Token audience is out of date',
-      'ExpiredToken',
-    )
+    if (ctx.cfg.service.isEntryway) {
+      // instruct client to refresh token during potential account migration
+      throw new InvalidRequestError(
+        'Token audience is out of date',
+        'ExpiredToken',
+      )
+    } else {
+      // this shouldn't really happen, since we validate the token's audience on our non-entryway PDSes.
+      throw new AuthRequiredError('Bad token audience')
+    }
   }
 }
 

packages/pds/src/auth.ts

@@ -1,11 +1,20 @@
 import * as assert from 'node:assert'
-import { KeyObject, createPrivateKey, createSecretKey } from 'node:crypto'
+import {
+  KeyObject,
+  createPrivateKey,
+  createPublicKey,
+  createSecretKey,
+} from 'node:crypto'
 import express from 'express'
 import KeyEncoder from 'key-encoder'
 import * as ui8 from 'uint8arrays'
 import * as jose from 'jose'
 import * as crypto from '@atproto/crypto'
-import { AuthRequiredError, InvalidRequestError } from '@atproto/xrpc-server'
+import {
+  AuthRequiredError,
+  ForbiddenError,
+  InvalidRequestError,
+} from '@atproto/xrpc-server'
 import AppContext from './context'
 import { softDeleted } from './db/util'
 
@@ -17,6 +26,7 @@ const HMACSHA256_JWT = 'HS256'
 export type ServerAuthOpts = {
   jwtSecret: string
   jwtSigningKey?: crypto.Secp256k1Keypair
+  jwtVerifyKeyHex?: string
   adminPass: string
   moderatorPass?: string
   triagePass?: string
@@ -41,19 +51,22 @@ export type RefreshToken = AuthToken & { jti: string; aud: string }
 export class ServerAuth {
   private _signingSecret: KeyObject
   private _signingKey?: KeyObject
+  private _verifyKey?: KeyObject
   private _adminPass: string
   private _moderatorPass?: string
   private _triagePass?: string
 
   constructor(opts: {
     signingSecret: KeyObject
     signingKey?: KeyObject
+    verifyKey?: KeyObject
     adminPass: string
     moderatorPass?: string
     triagePass?: string
   }) {
     this._signingSecret = opts.signingSecret
     this._signingKey = opts.signingKey
+    this._verifyKey = opts.verifyKey
     this._adminPass = opts.adminPass
     this._moderatorPass = opts.moderatorPass
     this._triagePass = opts.triagePass
@@ -64,12 +77,16 @@ export class ServerAuth {
     const signingKey = opts.jwtSigningKey
       ? await createPrivateKeyObject(opts.jwtSigningKey)
       : undefined
+    const verifyKey = opts.jwtVerifyKeyHex
+      ? await createPublicKeyObject(opts.jwtVerifyKeyHex)
+      : signingKey
     const adminPass = opts.adminPass
     const moderatorPass = opts.moderatorPass
     const triagePass = opts.triagePass
     return new ServerAuth({
       signingSecret,
       signingKey,
+      verifyKey,
       adminPass,
       moderatorPass,
       triagePass,
@@ -200,8 +217,8 @@ export class ServerAuth {
     const header = jose.decodeProtectedHeader(token)
     let result: jose.JWTVerifyResult
     try {
-      if (header.alg === SECP256K1_JWT && this._signingKey) {
-        const key = await this._signingKey
+      if (header.alg === SECP256K1_JWT && this._verifyKey) {
+        const key = await this._verifyKey
         result = await jose.jwtVerify(token, key, options)
       } else {
         const key = this._signingSecret
@@ -244,37 +261,50 @@ export const parseBasicAuth = (
 }
 
 export const accessVerifier =
-  (auth: ServerAuth) =>
+  (auth: ServerAuth, { cfg }: AppContext) =>
   async (ctx: { req: express.Request; res: express.Response }) => {
     const creds = await auth.getCredentialsOrThrow(ctx.req, [
       AuthScope.Access,
       AuthScope.AppPass,
     ])
+    if (!cfg.service.isEntryway && creds.audience !== cfg.service.did) {
+      throw new AuthRequiredError('Bad token audience')
+    }
     return {
       credentials: creds,
       artifacts: auth.getToken(ctx.req),
     }
   }
 
 export const accessVerifierNotAppPassword =
-  (auth: ServerAuth) =>
+  (auth: ServerAuth, { cfg }: AppContext) =>
   async (ctx: { req: express.Request; res: express.Response }) => {
     const creds = await auth.getCredentialsOrThrow(ctx.req, [AuthScope.Access])
+    if (!cfg.service.isEntryway && creds.audience !== cfg.service.did) {
+      throw new AuthRequiredError('Bad token audience')
+    }
     return {
       credentials: creds,
       artifacts: auth.getToken(ctx.req),
     }
   }
 
 export const accessVerifierCheckTakedown =
-  (auth: ServerAuth, { db, services }: AppContext) =>
+  (auth: ServerAuth, { db, services, cfg }: AppContext) =>
   async (ctx: { req: express.Request; res: express.Response }) => {
     const creds = await auth.getCredentialsOrThrow(ctx.req, [
       AuthScope.Access,
       AuthScope.AppPass,
     ])
+    if (!cfg.service.isEntryway && creds.audience !== cfg.service.did) {
+      throw new AuthRequiredError('Bad token audience')
+    }
     const actor = await services.account(db).getAccount(creds.did, true)
-    if (!actor || softDeleted(actor)) {
+    if (!actor) {
+      // will be turned into ExpiredToken for the client if proxied by entryway
+      throw new ForbiddenError('Account not found', 'AccountNotFound')
+    }
+    if (softDeleted(actor)) {
       throw new AuthRequiredError(
         'Account has been taken down',
         'AccountTakedown',
@@ -286,8 +316,8 @@ export const accessVerifierCheckTakedown =
     }
   }
 
-export const accessOrRoleVerifier = (auth: ServerAuth) => {
-  const verifyAccess = accessVerifier(auth)
+export const accessOrRoleVerifier = (auth: ServerAuth, ctx: AppContext) => {
+  const verifyAccess = accessVerifier(auth, ctx)
   const verifyRole = roleVerifier(auth)
   return async (ctx: { req: express.Request; res: express.Response }) => {
     // For non-admin tokens, we don't want to consider alternative verifiers and let it fail if it fails
@@ -313,8 +343,11 @@ export const accessOrRoleVerifier = (auth: ServerAuth) => {
   }
 }
 
-export const optionalAccessOrRoleVerifier = (auth: ServerAuth) => {
-  const verifyAccess = accessVerifier(auth)
+export const optionalAccessOrRoleVerifier = (
+  auth: ServerAuth,
+  ctx: AppContext,
+) => {
+  const verifyAccess = accessVerifier(auth, ctx)
   return async (ctx: { req: express.Request; res: express.Response }) => {
     try {
       return await verifyAccess(ctx)
@@ -392,4 +425,11 @@ const createPrivateKeyObject = async (
   return createPrivateKey({ format: 'pem', key })
 }
 
+const createPublicKeyObject = async (
+  publicKeyHex: string,
+): Promise<KeyObject> => {
+  const key = keyEncoder.encodePublic(publicKeyHex, 'raw', 'pem')
+  return createPublicKey({ format: 'pem', key })
+}
+
 const keyEncoder = new KeyEncoder('secp256k1')

packages/pds/src/config/config.ts

@@ -13,12 +13,14 @@ export const envToCfg = (env: ServerEnvironment): ServerConfig => {
     hostname === 'localhost'
       ? `http://localhost:${port}`
       : `https://${hostname}`
-  const did = env.serviceDid ?? `did:web:${hostname}`
+  const publicHostname = new URL(publicUrl).host
+  const did = env.serviceDid ?? `did:web:${encodeURIComponent(publicHostname)}`
   const serviceCfg: ServerConfig['service'] = {
     port,
     hostname,
     publicUrl,
     did,
+    isEntryway: env.isEntryway !== false, // defaults true
     version: env.version, // default?
     privacyPolicyUrl: env.privacyPolicyUrl,
     termsOfServiceUrl: env.termsOfServiceUrl,
@@ -210,6 +212,7 @@ export type ServiceConfig = {
   hostname: string
   publicUrl: string
   did: string
+  isEntryway: boolean
   version?: string
   privacyPolicyUrl?: string
   termsOfServiceUrl?: string