Write rate limits

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

@@ -3,6 +3,7 @@ import { InvalidRequestError, AuthRequiredError } from '@atproto/xrpc-server'
 import { prepareCreate, prepareDelete, prepareUpdate } from '../../../../repo'
 import { Server } from '../../../../lexicon'
 import {
+  HandlerInput,
   isCreate,
   isUpdate,
   isDelete,
@@ -15,9 +16,36 @@ import {
 import AppContext from '../../../../context'
 import { ConcurrentWriteError } from '../../../../services/repo'
 
+const ratelimitPoints = ({ input }: { input: HandlerInput }) => {
+  let points = 0
+  for (const op of input.body.writes) {
+    if (isCreate(op)) {
+      points += 3
+    } else if (isUpdate(op)) {
+      points += 2
+    } else {
+      points += 1
+    }
+  }
+  return points
+}
+
 export default function (server: Server, ctx: AppContext) {
   server.com.atproto.repo.applyWrites({
     auth: ctx.accessVerifierCheckTakedown,
+    rateLimit: [
+      {
+        name: 'repo-write-hour',
+        calcKey: ({ auth }) => auth.credentials.did,
+        calcPoints: ratelimitPoints,
+      },
+      {
+        name: 'repo-write-day',
+        calcKey: ({ auth }) => auth.credentials.did,
+        calcPoints: ratelimitPoints,
+      },
+    ],
+
     handler: async ({ input, auth }) => {
       const tx = input.body
       const { repo, validate, swapCommit } = tx

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

@@ -16,6 +16,18 @@ import { ConcurrentWriteError } from '../../../../services/repo'
 export default function (server: Server, ctx: AppContext) {
   server.com.atproto.repo.createRecord({
     auth: ctx.accessVerifierCheckTakedown,
+    rateLimit: [
+      {
+        name: 'repo-write-hour',
+        calcKey: ({ auth }) => auth.credentials.did,
+        calcPoints: () => 3,
+      },
+      {
+        name: 'repo-write-day',
+        calcKey: ({ auth }) => auth.credentials.did,
+        calcPoints: () => 3,
+      },
+    ],
     handler: async ({ input, auth }) => {
       const { repo, collection, rkey, record, swapCommit, validate } =
         input.body

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

@@ -9,6 +9,18 @@ import { ConcurrentWriteError } from '../../../../services/repo'
 export default function (server: Server, ctx: AppContext) {
   server.com.atproto.repo.deleteRecord({
     auth: ctx.accessVerifierCheckTakedown,
+    rateLimit: [
+      {
+        name: 'repo-write-hour',
+        calcKey: ({ auth }) => auth.credentials.did,
+        calcPoints: () => 1,
+      },
+      {
+        name: 'repo-write-day',
+        calcKey: ({ auth }) => auth.credentials.did,
+        calcPoints: () => 1,
+      },
+    ],
     handler: async ({ input, auth }) => {
       const { repo, collection, rkey, swapCommit, swapRecord } = input.body
       const did = await ctx.services.account(ctx.db).getDidForActor(repo)

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

@@ -16,6 +16,18 @@ import { ConcurrentWriteError } from '../../../../services/repo'
 export default function (server: Server, ctx: AppContext) {
   server.com.atproto.repo.putRecord({
     auth: ctx.accessVerifierCheckTakedown,
+    rateLimit: [
+      {
+        name: 'repo-write-hour',
+        calcKey: ({ auth }) => auth.credentials.did,
+        calcPoints: () => 2,
+      },
+      {
+        name: 'repo-write-day',
+        calcKey: ({ auth }) => auth.credentials.did,
+        calcPoints: () => 2,
+      },
+    ],
     handler: async ({ auth, input }) => {
       const {
         repo,

packages/pds/src/index.ts

@@ -20,7 +20,7 @@ import {
   RateLimiterOpts,
   Options as XrpcServerOptions,
 } from '@atproto/xrpc-server'
-import { MINUTE } from '@atproto/common'
+import { DAY, HOUR, MINUTE } from '@atproto/common'
 import * as appviewConsumers from './app-view/event-stream/consumers'
 import inProcessAppView from './app-view/api'
 import API from './api'
@@ -137,6 +137,7 @@ export class PDS {
     )
 
     const app = express()
+    app.set('trust proxy', true)
     app.use(cors())
     app.use(loggerMiddleware)
     app.use(compression())
@@ -285,6 +286,18 @@ export class PDS {
             points: 3000,
           },
         ],
+        shared: [
+          {
+            name: 'repo-write-hour',
+            durationMs: HOUR,
+            points: 5000, // creates=3, puts=2, deletes=1
+          },
+          {
+            name: 'repo-write-day',
+            durationMs: DAY,
+            points: 35000, // creates=3, puts=2, deletes=1
+          },
+        ],
       }
     }
 

packages/xrpc-server/src/index.ts

@@ -5,4 +5,4 @@ export * from './stream'
 export * from './rate-limiter'
 
 export type { ServerTiming } from './util'
-export { getReqIp, serverTimingHeader, ServerTimer } from './util'
+export { serverTimingHeader, ServerTimer } from './util'

packages/xrpc-server/src/rate-limiter.ts

@@ -14,7 +14,6 @@ import {
   RateLimiterStatus,
   XRPCReqContext,
 } from './types'
-import { getReqIp } from './util'
 
 export type RateLimiterOpts = {
   keyPrefix: string
@@ -155,5 +154,5 @@ export const getTightestLimit = (
   return lowest
 }
 
-const defaultKey: CalcKeyFn = (ctx: XRPCReqContext) => getReqIp(ctx.req)
+const defaultKey: CalcKeyFn = (ctx: XRPCReqContext) => ctx.req.ip
 const defaultPoints: CalcPointsFn = () => 1

packages/xrpc-server/src/util.ts

@@ -268,10 +268,6 @@ function decodeBodyStream(
   return stream
 }
 
-export const getReqIp = (req: express.Request): string => {
-  return req.ips.at(-1) ?? req.ip
-}
-
 export function serverTimingHeader(timings: ServerTiming[]) {
   return timings
     .map((timing) => {