diff --git a/src/backend/clients/email/EmailClient.ts b/src/backend/clients/email/EmailClient.ts index 414238fea..0e64d265b 100644 --- a/src/backend/clients/email/EmailClient.ts +++ b/src/backend/clients/email/EmailClient.ts @@ -28,10 +28,15 @@ import { type EmailTemplateName, } from './templates'; -/** Attachment shape passed through to the underlying transport. */ +/** + * Attachment shape passed through to the underlying transport. Give either + * `content` (held in memory) or `path` (a local file the transport streams on + * its own, re-read for every message that carries it). + */ export interface EmailAttachment { filename: string; - content: Buffer | string; + content?: Buffer | string; + path?: string; contentType?: string; encoding?: string; } diff --git a/src/backend/drivers/util/fileInput.ts b/src/backend/drivers/util/fileInput.ts index d32b8dd39..8344d4f64 100644 --- a/src/backend/drivers/util/fileInput.ts +++ b/src/backend/drivers/util/fileInput.ts @@ -18,11 +18,12 @@ */ import { posix as pathPosix } from 'node:path'; +import { Readable } from 'node:stream'; import type { Actor } from '../../core/actor.js'; import { HttpError } from '../../core/http/HttpError.js'; import type { FSService } from '../../services/fs/FSService.js'; import { expandTildePath, resolveNode } from '../../services/fs/resolveNode.js'; -import { hasNoBackingS3Object } from '../../stores/fs/FSEntry.js'; +import { hasNoBackingS3Object, type FSEntry } from '../../stores/fs/FSEntry.js'; import type { FSEntryStore } from '../../stores/fs/FSEntryStore.js'; import type { S3ObjectStore } from '../../stores/fs/S3ObjectStore.js'; import { mimeFromName } from '../../util/fileSigning.js'; @@ -38,6 +39,8 @@ import { secureFetch } from '../../util/secureHttp.js'; * (`/alice/music/sample.mp3`) • an object with `{ path?, uid?, uuid? }` * * This helper collapses those shapes into `{ buffer, filename, mimeType }`. + * Drivers that can hand a stream to their downstream (a mail transport, an + * upload) use {@link openFileInputStream} instead and never hold the file. */ export interface LoadedFile { @@ -57,10 +60,29 @@ export interface LoadedFile { } | null; } +/** + * A Puter FS reference as drivers receive it: a path string or `{ path?, uid?, + * uuid? }`. + */ +export type FileInputRef = + | string + | { path?: string; uid?: string; uuid?: string }; + +export interface OpenedFileInput { + body: Readable; + /** Object size when the store reports it; null when it does not. */ + contentLength: number | null; + filename: string; + mimeType: string; + fsEntry: NonNullable; +} + +type FileInputStores = { fsEntry: FSEntryStore; s3Object: S3ObjectStore }; + const DATA_URL_PATTERN = /^data:([^;,]+)?(?:;([^,]*))?,(.*)$/s; export async function loadFileInput( - stores: { fsEntry: FSEntryStore; s3Object: S3ObjectStore }, + stores: FileInputStores, fsService: FSService, actor: Actor, input: unknown, @@ -71,11 +93,7 @@ export async function loadFileInput( legacyCode: 'bad_request', }); } - if (!Number.isFinite(Number(actor?.user?.id ?? NaN))) { - throw new HttpError(401, 'Unauthorized', { - legacyCode: 'unauthorized', - }); - } + requireActorUser(actor); // Data URL — decode base64/plain inline. if (typeof input === 'string' && input.startsWith('data:')) { @@ -131,30 +149,47 @@ export async function loadFileInput( } // Path string or object reference → resolve into FSEntry, then S3 read. - const username = actor?.user?.username; + const opened = await openFileInputStream( + stores, + fsService, + actor, + input as FileInputRef, + options, + ); + const buffer = await collectStream(opened.body, options.maxBytes); + return { + buffer, + filename: opened.filename, + mimeType: opened.mimeType, + fsEntry: opened.fsEntry, + }; +} + +/** + * Resolve a Puter FS reference to its entry, with the checks every driver read + * needs: it must be a file (not a directory, symlink or shortcut) and `actor` + * must be allowed to read it. + */ +export async function resolveFileInputEntry( + stores: FileInputStores, + fsService: FSService, + actor: Actor, + input: FileInputRef, +): Promise { + requireActorUser(actor); + const username = actor.user?.username; const expandPath = (path: string | undefined) => path !== undefined ? expandTildePath(path, username) : undefined; const ref: { path?: string; uid?: string; uuid?: string } = typeof input === 'string' ? { path: expandPath(input) } - : (() => { - const record = input as Record; - return { - path: expandPath( - typeof record.path === 'string' - ? record.path - : undefined, - ), - uid: - typeof record.uid === 'string' - ? record.uid - : undefined, - uuid: - typeof record.uuid === 'string' - ? record.uuid - : undefined, - }; - })(); + : { + path: expandPath( + typeof input.path === 'string' ? input.path : undefined, + ), + uid: typeof input.uid === 'string' ? input.uid : undefined, + uuid: typeof input.uuid === 'string' ? input.uuid : undefined, + }; const entry = await resolveNode(stores.fsEntry, ref, { required: true }); if (!entry) @@ -173,43 +208,72 @@ export async function loadFileInput( // ACL gate: resolveNode does global UID/UUID/ID/path lookups, no // namespace check. Without this check, an attacker controlling // `path`/`uid`/`uuid` (e.g. AI chat `puter_path` content parts) could - // exfiltrate any user's file. Must run before the S3 read below. + // exfiltrate any user's file. Must run before any read of the object. await fsService.checkFSAccess(entry, actor, 'read'); + return entry; +} + +/** + * Open a Puter FS reference as a stream of its bytes. Nothing is buffered: the + * caller pipes `body` wherever it goes and owns its lifetime. A known size over + * `maxBytes` is refused (413) before a byte is read; a size that is only + * discovered while reading is the caller's to enforce. + */ +export async function openFileInputStream( + stores: FileInputStores, + fsService: FSService, + actor: Actor, + input: FileInputRef, + options: { maxBytes?: number } = {}, +): Promise { + const entry = await resolveFileInputEntry(stores, fsService, actor, input); + const fsEntry = { + uuid: entry.uuid, + path: entry.path, + bucket: entry.bucket, + bucketRegion: entry.bucketRegion, + size: entry.size, + sqlId: entry.id, + }; // Empty files (created via `touch`) have no backing S3 object — // getObjectStream would throw NoSuchKey, so return empty content. if (hasNoBackingS3Object(entry)) { return { - buffer: Buffer.alloc(0), + body: Readable.from([]), + contentLength: 0, filename: entry.name, mimeType: mimeFromName(entry.name) ?? 'application/octet-stream', - fsEntry: { - uuid: entry.uuid, - path: entry.path, - bucket: entry.bucket, - bucketRegion: entry.bucketRegion, - size: entry.size, - sqlId: entry.id, - }, + fsEntry, }; } - const objectKey = entry.uuid; const { body, contentType, contentLength } = await stores.s3Object.getObjectStream( { bucket: stores.s3Object.resolveBucket(entry.bucket), - objectKey, + objectKey: entry.uuid, }, stores.s3Object.resolveRegion(entry.bucketRegion), ); if (contentLength && options.maxBytes && contentLength > options.maxBytes) { body.destroy(); - throw new HttpError( - 413, - `File exceeds max size (${options.maxBytes} bytes)`, - { legacyCode: 'storage_limit_reached' }, - ); + throw tooLarge(options.maxBytes); } + return { + body, + contentLength: contentLength ?? null, + filename: entry.name, + mimeType: + contentType ?? + mimeFromName(entry.name) ?? + 'application/octet-stream', + fsEntry, + }; +} +async function collectStream( + body: Readable, + maxBytes: number | undefined, +): Promise { const chunks: Buffer[] = []; let total = 0; for await (const chunk of body) { @@ -217,33 +281,27 @@ export async function loadFileInput( ? chunk : Buffer.from(chunk as Uint8Array); total += buf.byteLength; - if (options.maxBytes && total > options.maxBytes) { + if (maxBytes && total > maxBytes) { body.destroy(); - throw new HttpError( - 413, - `File exceeds max size (${options.maxBytes} bytes)`, - { legacyCode: 'storage_limit_reached' }, - ); + throw tooLarge(maxBytes); } chunks.push(buf); } - const buffer = Buffer.concat(chunks, total); - const resolvedMime = - contentType ?? mimeFromName(entry.name) ?? 'application/octet-stream'; + return Buffer.concat(chunks, total); +} - return { - buffer, - filename: entry.name, - mimeType: resolvedMime, - fsEntry: { - uuid: entry.uuid, - path: entry.path, - bucket: entry.bucket, - bucketRegion: entry.bucketRegion, - size: entry.size, - sqlId: entry.id, - }, - }; +function requireActorUser(actor: Actor): void { + if (!Number.isFinite(Number(actor?.user?.id ?? NaN))) { + throw new HttpError(401, 'Unauthorized', { + legacyCode: 'unauthorized', + }); + } +} + +function tooLarge(maxBytes: number): HttpError { + return new HttpError(413, `File exceeds max size (${maxBytes} bytes)`, { + legacyCode: 'storage_limit_reached', + }); } function assertMax(buffer: Buffer, maxBytes?: number): void { diff --git a/src/puter-js/src/modules/Email.js b/src/puter-js/src/modules/Email.js index 9b288ca02..6142e2137 100644 --- a/src/puter-js/src/modules/Email.js +++ b/src/puter-js/src/modules/Email.js @@ -4,7 +4,9 @@ import * as utils from '../lib/utils.js'; /** * One attachment: either inline base64 `content`, or a Puter FS reference * (`path`/`uid`) read server-side with the caller's — falling back to the - * authorizing worker's — file permissions. + * authorizing worker's — file permissions. FS references are streamed from + * storage and never travel through the request, so prefer them for anything + * larger than a few hundred kilobytes. * * @typedef {Object} EmailAttachment * @property {string} [filename] Required with `content`; defaults to the file's name for FS refs. @@ -15,7 +17,7 @@ import * as utils from '../lib/utils.js'; */ /** - * The options form of `send()`. + * The options form of `sendTransactional()`. * * @typedef {Object} EmailSendOptions * @property {string | string[]} to Recipient address(es). @@ -32,7 +34,7 @@ import * as utils from '../lib/utils.js'; */ /** - * What one `send()` resolves to. + * What one `sendTransactional()` resolves to. * * @typedef {Object} EmailSendResult * @property {string | null} messageId First transport message id reported for this send, when available. @@ -43,67 +45,98 @@ import * as utils from '../lib/utils.js'; */ /** - * Restricted outbound email (the `puter-email` driver interface). + * The call shapes shared by `sendTransactional()` and its legacy alias. + * + * @typedef {{ + * (to: string | string[], subject: string, body: string): Promise, + * (options: EmailSendOptions): Promise, + * }} EmailSendMethod + */ + +/** + * `body` is positional-call sugar for `text`. + * + * @param {Record} args + * @returns {Record} + */ +const preprocessSendArgs = (args) => { + if ( + args.body !== undefined && + args.text === undefined && + args.html === undefined + ) { + args.text = args.body; + } + delete args.body; + return args; +}; + +/** + * Transactional email from your app (the `puter-email` driver interface). * * Every send must be authorized by a worker: either the worker calls - * directly (`me.puter.email.send(...)`), or a user calls with their own - * token and passes the worker's token as `emailAccessToken` — the caller is - * the one billed and rate-limited. In a worker handler: + * directly (`me.puter.email.sendTransactional(...)`), or a user calls with + * their own token and passes the worker's token as `emailAccessToken` — the + * caller is the one billed and rate-limited. In a worker handler: * * router.post('/notify', async ({ request, user }) => { * const { to, subject, text } = await request.json(); - * return await user.puter.email.send({ + * return await user.puter.email.sendTransactional({ * to, subject, text, * emailAccessToken: me.puter.authToken, * // Inline or Puter-FS attachments: * attachments: [ * { filename, content, contentType }, // content = base64 - * { path: '~/Documents/report.pdf' }, // read server-side + * { path: '~/Documents/report.pdf' }, // streamed server-side * ], * }); * }); * - * Positional form: `await puter.email.send(to, subject, body)`. + * Positional form: `await puter.email.sendTransactional(to, subject, body)`. * - * Every mail automatically gets an unsubscribe / report-abuse footer. - * Unsubscribing is per app: a recipient who opts out stops hearing from - * the app they opted out of, and still hears from the other apps the same - * account runs. Opted-out recipients are dropped from that app's future - * sends — they come back in the result's `suppressed` array — and a send - * whose `to` list is entirely opted out is rejected. + * Mail goes out from a Puter-controlled address, labelled with the app's + * title. Every mail automatically gets an unsubscribe / report-abuse + * footer. Unsubscribing is per app: a recipient who opts out stops hearing + * from the app they opted out of, and still hears from the other apps the + * same account runs. Opted-out recipients are dropped from that app's + * future sends — they come back in the result's `suppressed` array — and a + * send whose `to` list is entirely opted out is rejected. * * Each recipient gets a private delivery. A recipient whose delivery * fails comes back in the result's `failed` array (everyone else got * their copy — retry with just those addresses); the call only rejects * when no recipient could be delivered. + * + * `send()` is the previous name of this method and still works; it will be + * removed once existing apps have moved to `sendTransactional()`. */ export class EmailModule extends PuterModule { /** - * Sends one email. The positional form is shorthand for a plain-text body; - * everything else (html, cc/bcc, attachments, `emailAccessToken`) goes - * through the options form. + * Sends one transactional email. The positional form is shorthand for a + * plain-text body; everything else (html, cc/bcc, attachments, + * `emailAccessToken`) goes through the options form. * - * @type {{ - * (to: string | string[], subject: string, body: string): Promise, - * (options: EmailSendOptions): Promise, - * }} + * @type {EmailSendMethod} + */ + sendTransactional = utils.makeDriverMethod({ + iface: 'puter-email', + method: 'sendTransactional', + argNames: ['to', 'subject', 'body'], + preprocess: preprocessSendArgs, + }); + + /** + * Legacy name for {@link EmailModule.sendTransactional}; same arguments, + * same result. + * + * @deprecated Use `sendTransactional()`. + * @type {EmailSendMethod} */ send = utils.makeDriverMethod({ iface: 'puter-email', method: 'send', argNames: ['to', 'subject', 'body'], - preprocess: (args) => { - // `body` is positional-call sugar for `text`. - if ( - args.body !== undefined && - args.text === undefined && - args.html === undefined - ) { - args.text = args.body; - } - delete args.body; - return args; - }, + preprocess: preprocessSendArgs, }); } diff --git a/src/puter-js/tests/api/suites/system.suite.ts b/src/puter-js/tests/api/suites/system.suite.ts index 5ae37df52..50f0289c4 100644 --- a/src/puter-js/tests/api/suites/system.suite.ts +++ b/src/puter-js/tests/api/suites/system.suite.ts @@ -171,6 +171,40 @@ export default suite('system', { ); }, + 'email.sendTransactional passes a driver error through unchanged': async ( + t, + ) => { + const error = (await t.assert.rejects(() => + t.puter.email.sendTransactional( + 'nobody@example.com', + 'subject', + 'body', + ), + )) as { code?: string; message?: string }; + t.assert.equal(error.code, 'not_found'); + t.assert.equal( + error.message, + 'Driver not found: puter-email:(no default)', + ); + }, + + // `send` is the pre-rename alias: same wire shape, same error path. + 'email.send and email.sendTransactional reject identically': async (t) => { + const options = { + to: 'nobody@example.com', + subject: 'subject', + text: 'body', + }; + const viaAlias = (await t.assert.rejects(() => + t.puter.email.send(options), + )) as { code?: string; message?: string }; + const viaNew = (await t.assert.rejects(() => + t.puter.email.sendTransactional(options), + )) as { code?: string; message?: string }; + t.assert.equal(viaAlias.code, viaNew.code); + t.assert.equal(viaAlias.message, viaNew.message); + }, + 'email.send reports failures to a positional error callback': async (t) => { let reported: { code?: string } | null = null; await t.assert.rejects(() =>