Ozone delegates email sending to actor's pds (#2272)

* ozone delegates email sending to user's pds

* lexicon: add content field to mod email event

* test email sending via mod event

lexicons/com/atproto/admin/defs.json

@@ -585,6 +585,10 @@
           "type": "string",
           "description": "The subject line of the email sent to the user."
         },
+        "content": {
+          "type": "string",
+          "description": "The content of the email sent to the user."
+        },
         "comment": {
           "type": "string",
           "description": "Additional comment about the outgoing comm."

packages/api/src/client/lexicons.ts

@@ -897,6 +897,10 @@ export const schemaDict = {
             type: 'string',
             description: 'The subject line of the email sent to the user.',
           },
+          content: {
+            type: 'string',
+            description: 'The content of the email sent to the user.',
+          },
           comment: {
             type: 'string',
             description: 'Additional comment about the outgoing comm.',

packages/api/src/client/types/com/atproto/admin/defs.ts

@@ -704,6 +704,8 @@ export function validateModEventUnmute(v: unknown): ValidationResult {
 export interface ModEventEmail {
   /** The subject line of the email sent to the user. */
   subjectLine: string
+  /** The content of the email sent to the user. */
+  content?: string
   /** Additional comment about the outgoing comm. */
   comment?: string
   [k: string]: unknown

packages/bsky/src/lexicon/lexicons.ts

@@ -897,6 +897,10 @@ export const schemaDict = {
             type: 'string',
             description: 'The subject line of the email sent to the user.',
           },
+          content: {
+            type: 'string',
+            description: 'The content of the email sent to the user.',
+          },
           comment: {
             type: 'string',
             description: 'Additional comment about the outgoing comm.',

packages/bsky/src/lexicon/types/com/atproto/admin/defs.ts

@@ -704,6 +704,8 @@ export function validateModEventUnmute(v: unknown): ValidationResult {
 export interface ModEventEmail {
   /** The subject line of the email sent to the user. */
   subjectLine: string
+  /** The content of the email sent to the user. */
+  content?: string
   /** Additional comment about the outgoing comm. */
   comment?: string
   [k: string]: unknown

packages/dev-env/src/ozone.ts

@@ -34,6 +34,7 @@ export class TestOzone {
     const port = config.port || (await getPort())
     const url = `http://localhost:${port}`
     const env: ozone.OzoneEnvironment = {
+      devMode: true,
       version: '0.0.0',
       port,
       didPlcUrl: config.plcUrl,

packages/ozone/src/api/admin/emitModerationEvent.ts

@@ -2,12 +2,14 @@ import { AuthRequiredError, InvalidRequestError } from '@atproto/xrpc-server'
 import { Server } from '../../lexicon'
 import AppContext from '../../context'
 import {
+  isModEventEmail,
   isModEventLabel,
   isModEventReverseTakedown,
   isModEventTakedown,
 } from '../../lexicon/types/com/atproto/admin/defs'
 import { subjectFromInput } from '../../mod-service/subject'
 import { ModerationLangService } from '../../mod-service/lang'
+import { retryHttp } from '../../util'
 
 export default function (server: Server, ctx: AppContext) {
   server.com.atproto.admin.emitModerationEvent({
@@ -75,6 +77,23 @@ export default function (server: Server, ctx: AppContext) {
         }
       }
 
+      if (isModEventEmail(event) && event.content) {
+        // sending email prior to logging the event to avoid a long transaction below
+        if (!subject.isRepo()) {
+          throw new InvalidRequestError(
+            'Email can only be sent to a repo subject',
+          )
+        }
+        const { content, subjectLine } = event
+        await retryHttp(() =>
+          ctx.modService(db).sendEmail({
+            subject: subjectLine,
+            content,
+            recipientDid: subject.did,
+          }),
+        )
+      }
+
       const moderationEvent = await db.transaction(async (dbTxn) => {
         const moderationTxn = ctx.modService(dbTxn)
 

packages/ozone/src/config/config.ts

@@ -13,6 +13,7 @@ export const envToCfg = (env: OzoneEnvironment): OzoneConfig => {
     publicUrl: env.publicUrl,
     did: env.serverDid,
     version: env.version,
+    devMode: env.devMode,
   }
 
   assert(env.dbPostgresUrl)
@@ -71,6 +72,7 @@ export type ServiceConfig = {
   publicUrl: string
   did: string
   version?: string
+  devMode?: boolean
 }
 
 export type DatabaseConfig = {

packages/ozone/src/config/env.ts

@@ -1,8 +1,9 @@
-import { envInt, envList, envStr } from '@atproto/common'
+import { envBool, envInt, envList, envStr } from '@atproto/common'
 
 export const readEnv = (): OzoneEnvironment => {
   return {
     nodeEnv: envStr('NODE_ENV'),
+    devMode: envBool('OZONE_DEV_MODE'),
     version: envStr('OZONE_VERSION'),
     port: envInt('OZONE_PORT'),
     publicUrl: envStr('OZONE_PUBLIC_URL'),
@@ -27,6 +28,7 @@ export const readEnv = (): OzoneEnvironment => {
 
 export type OzoneEnvironment = {
   nodeEnv?: string
+  devMode?: boolean
   version?: string
   port?: number
   publicUrl?: string

packages/ozone/src/context.ts

@@ -58,31 +58,31 @@ export class AppContext {
         aud,
         keypair: signingKey,
       })
-    const appviewAuth = async () =>
-      cfg.appview.did ? createAuthHeaders(cfg.appview.did) : undefined
 
     const backgroundQueue = new BackgroundQueue(db)
     const eventPusher = new EventPusher(db, createAuthHeaders, {
       appview: cfg.appview,
       pds: cfg.pds ?? undefined,
     })
 
+    const idResolver = new IdResolver({
+      plcUrl: cfg.identity.plcUrl,
+    })
+
     const modService = ModerationService.creator(
+      cfg,
       backgroundQueue,
+      idResolver,
       eventPusher,
       appviewAgent,
-      appviewAuth,
+      createAuthHeaders,
       cfg.service.did,
       overrides?.imgInvalidator,
       cfg.cdn.paths,
     )
 
     const communicationTemplateService = CommunicationTemplateService.creator()
 
-    const idResolver = new IdResolver({
-      plcUrl: cfg.identity.plcUrl,
-    })
-
     const sequencer = new Sequencer(db)
 
     return new AppContext(

packages/ozone/src/daemon/context.ts

@@ -7,6 +7,7 @@ import { EventPusher } from './event-pusher'
 import { EventReverser } from './event-reverser'
 import { ModerationService, ModerationServiceCreator } from '../mod-service'
 import { BackgroundQueue } from '../background'
+import { IdResolver } from '@atproto/identity'
 
 export type DaemonContextOptions = {
   db: Database
@@ -39,21 +40,26 @@ export class DaemonContext {
         keypair: signingKey,
       })
 
-    const appviewAuth = async () =>
-      cfg.appview.did ? createAuthHeaders(cfg.appview.did) : undefined
-
     const eventPusher = new EventPusher(db, createAuthHeaders, {
       appview: cfg.appview,
       pds: cfg.pds ?? undefined,
     })
+
     const backgroundQueue = new BackgroundQueue(db)
+    const idResolver = new IdResolver({
+      plcUrl: cfg.identity.plcUrl,
+    })
+
     const modService = ModerationService.creator(
+      cfg,
       backgroundQueue,
+      idResolver,
       eventPusher,
       appviewAgent,
-      appviewAuth,
+      createAuthHeaders,
       cfg.service.did,
     )
+
     const eventReverser = new EventReverser(db, modService)
 
     return new DaemonContext({

packages/ozone/src/lexicon/lexicons.ts

@@ -897,6 +897,10 @@ export const schemaDict = {
             type: 'string',
             description: 'The subject line of the email sent to the user.',
           },
+          content: {
+            type: 'string',
+            description: 'The content of the email sent to the user.',
+          },
           comment: {
             type: 'string',
             description: 'Additional comment about the outgoing comm.',

packages/ozone/src/lexicon/types/com/atproto/admin/defs.ts

@@ -704,6 +704,8 @@ export function validateModEventUnmute(v: unknown): ValidationResult {
 export interface ModEventEmail {
   /** The subject line of the email sent to the user. */
   subjectLine: string
+  /** The content of the email sent to the user. */
+  content?: string
   /** Additional comment about the outgoing comm. */
   comment?: string
   [k: string]: unknown

packages/ozone/src/mod-service/index.ts

@@ -1,9 +1,13 @@
+import net from 'node:net'
+import { Insertable, sql } from 'kysely'
 import { CID } from 'multiformats/cid'
 import { AtUri, INVALID_HANDLE } from '@atproto/syntax'
 import { InvalidRequestError } from '@atproto/xrpc-server'
 import { addHoursToDate } from '@atproto/common'
+import { IdResolver } from '@atproto/identity'
+import AtpAgent from '@atproto/api'
 import { Database } from '../db'
-import { AppviewAuth, ModerationViews } from './views'
+import { AuthHeaders, ModerationViews } from './views'
 import { Main as StrongRef } from '../lexicon/types/com/atproto/repo/strongRef'
 import {
   isModEventComment,
@@ -30,9 +34,7 @@ import {
 } from './types'
 import { ModerationEvent } from '../db/schema/moderation_event'
 import { StatusKeyset, TimeIdKeyset, paginate } from '../db/pagination'
-import AtpAgent from '@atproto/api'
 import { Label } from '../lexicon/types/com/atproto/label/defs'
-import { Insertable, sql } from 'kysely'
 import {
   ModSubject,
   RecordSubject,
@@ -46,44 +48,53 @@ import { BackgroundQueue } from '../background'
 import { EventPusher } from '../daemon'
 import { ImageInvalidator } from '../image-invalidator'
 import { httpLogger as log } from '../logger'
+import { OzoneConfig } from '../config'
 
 export type ModerationServiceCreator = (db: Database) => ModerationService
 
 export class ModerationService {
   constructor(
     public db: Database,
+    public cfg: OzoneConfig,
     public backgroundQueue: BackgroundQueue,
+    public idResolver: IdResolver,
     public eventPusher: EventPusher,
     public appviewAgent: AtpAgent,
-    private appviewAuth: AppviewAuth,
+    private createAuthHeaders: (aud: string) => Promise<AuthHeaders>,
     public serverDid: string,
     public imgInvalidator?: ImageInvalidator,
     public cdnPaths?: string[],
   ) {}
 
   static creator(
+    cfg: OzoneConfig,
     backgroundQueue: BackgroundQueue,
+    idResolver: IdResolver,
     eventPusher: EventPusher,
     appviewAgent: AtpAgent,
-    appviewAuth: AppviewAuth,
+    createAuthHeaders: (aud: string) => Promise<AuthHeaders>,
     serverDid: string,
     imgInvalidator?: ImageInvalidator,
     cdnPaths?: string[],
   ) {
     return (db: Database) =>
       new ModerationService(
         db,
+        cfg,
         backgroundQueue,
+        idResolver,
         eventPusher,
         appviewAgent,
-        appviewAuth,
+        createAuthHeaders,
         serverDid,
         imgInvalidator,
         cdnPaths,
       )
   }
 
-  views = new ModerationViews(this.db, this.appviewAgent, this.appviewAuth)
+  views = new ModerationViews(this.db, this.appviewAgent, () =>
+    this.createAuthHeaders(this.cfg.appview.did),
+  )
 
   async getEvent(id: number): Promise<ModerationEventRow | undefined> {
     return await this.db.db
@@ -291,6 +302,9 @@ export class ModerationService {
 
     if (isModEventEmail(event)) {
       meta.subjectLine = event.subjectLine
+      if (event.content) {
+        meta.content = event.content
+      }
     }
 
     const subjectInfo = subject.info()
@@ -903,6 +917,49 @@ export class ModerationService {
       )
       .execute()
   }
+
+  async sendEmail(opts: {
+    content: string
+    recipientDid: string
+    subject: string
+  }) {
+    const { subject, content, recipientDid } = opts
+    const { pds } = await this.idResolver.did.resolveAtprotoData(recipientDid)
+    const url = new URL(pds)
+    if (!this.cfg.service.devMode && !isSafeUrl(url)) {
+      throw new InvalidRequestError('Invalid pds service in DID doc')
+    }
+    const agent = new AtpAgent({ service: url })
+    const { data: serverInfo } =
+      await agent.api.com.atproto.server.describeServer()
+    if (serverInfo.did !== `did:web:${url.hostname}`) {
+      // @TODO do bidirectional check once implemented. in the meantime,
+      // matching did to hostname we're talking to is pretty good.
+      throw new InvalidRequestError('Invalid pds service in DID doc')
+    }
+    const { data: delivery } = await agent.api.com.atproto.admin.sendEmail(
+      {
+        subject,
+        content,
+        recipientDid,
+        senderDid: this.cfg.service.did,
+      },
+      {
+        encoding: 'application/json',
+        ...(await this.createAuthHeaders(serverInfo.did)),
+      },
+    )
+    if (!delivery.sent) {
+      throw new InvalidRequestError('Email was accepted but not sent')
+    }
+  }
+}
+
+const isSafeUrl = (url: URL) => {
+  if (url.protocol !== 'https:') return false
+  if (!url.hostname || url.hostname === 'localhost') return false
+  if (net.isIP(url.hostname) === 0) return false
+  return true
 }
 
 const TAKEDOWNS = ['pds_takedown' as const, 'appview_takedown' as const]