diff --git a/CLAUDE.md b/CLAUDE.md index b721f34..4635ed7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -134,6 +134,7 @@ Environment variables consumed by the `picloud` binary: | `PICLOUD_DB_MAX_CONNECTIONS` | `32` | Postgres pool size. Matched to `PICLOUD_MAX_CONCURRENT_EXECUTIONS` so the data plane can't starve background workers. | | `PICLOUD_SESSION_TTL_HOURS` | `24` | Sliding-window admin session lifetime (idle timeout). | | `PICLOUD_SESSION_ABSOLUTE_TTL_HOURS` | `720` (30 days) | Absolute hard cap on an admin session's lifetime (audit 2026-07-11 C1). The sliding `touch` bump is clamped at `login_time + this`, so even a continuously-used or stolen-but-warm admin token self-expires. Mirrors the data-plane app-user session cap. | +| `PICLOUD_SMTP_BIND` | — (off) | A3 native SMTP ingress. When set (e.g. `0.0.0.0:2525`), a receive-only SMTP listener binds alongside the HTTP server: it resolves each `RCPT TO` mailbox to the app-owned email trigger that claims it (`email_trigger_details.inbound_address`, unique) and writes an `Email` outbox row the dispatcher fires — same tail as the HMAC webhook. **No AUTH/STARTTLS at the listener** (TLS terminates upstream — Caddy/relay); the recipient address + per-app isolation are the boundary. `DATA` is size-capped. Unset = the webhook path only. Distinct from the outbound relay `PICLOUD_SMTP_*` vars (`SmtpConfig`). | | `PICLOUD_INTERCEPTOR_TIMEOUT_MS` | `5000` (5s) | §9.4 M5 default per-interceptor wall-clock timeout when a `[[interceptors]]` marker sets no `timeout_ms`. Bounds ONE hook run so a runaway guard (`loop {}`) is denied (fail closed) rather than hanging the guarded write. The effective deadline is always tightened to at most the caller's remaining budget — a hook can never EXTEND its caller's deadline. | | `PICLOUD_OUTBOX_CLAIM_TIMEOUT_SEC` | `600` (10 min) | How long a dispatcher may hold an OUTBOX claim before the reclaim task returns the row to the queue. **Without this a crash or restart mid-dispatch stranded every in-flight row permanently** — the dispatcher claims a row and only clears the claim on success (delete) or failure (reschedule), and `claim_due` selects `claimed_at IS NULL`, so a process that died in between left rows nothing would ever pick up again. Every other claim-based store (queue, group queue, workflow steps) already had a reclaimer; the outbox was the gap, and since it is the universal trigger path, the loss covered kv/docs/files/cron/pubsub/email/`invoke_async`/dead-letter alike. The default is deliberately generous: a script may run 300s, so a claim older than twice that is abandoned rather than slow. A reclaim does **not** bump `attempt_count` (the handler never ran — same reasoning as the transient queue `release`). Runs on the existing `PICLOUD_QUEUE_RECLAIM_INTERVAL_MS` ticker. | | `PICLOUD_REALTIME_BROADCAST_CAPACITY` | `64` | Per-channel SSE broadcast buffer depth (a slow consumer sees oldest events dropped). | @@ -156,6 +157,6 @@ Environment variables consumed by the `picloud` binary: ## Out of MVP -This section captured the original MVP cut. Most of it has since **shipped in v1.1.x**: queue triggers, cron triggers, inbound email (`email:receive`, HMAC-webhook model), KV / docs / email / users / HTTP SDKs, function-to-function `invoke()`, and secrets are all live (blueprint §12 Phase 4 table). The **Workflows** track (DAG + nested workflows) has since **shipped** (migrations `0071`/`0072`). The **§9.4 service-interceptor** track has since shipped in full (migrations `0073`–`0075`; before/after phases, data-transform, all six services, per-interceptor timeout, resolve cache, `pic interceptors ls`). A read-only **metrics/observability dashboard** has since shipped (A2): `GET /api/v1/admin/apps/{id}/metrics?window=` aggregates the existing `execution_logs` table (counts, error rate, latency avg/p50/p95, an hourly series) via `ExecutionLogRepository::summarize_for_app` + `metrics_api`, surfaced as the dashboard's per-app **Metrics** tab — no hot-path instrumentation. **Still deferred:** a raw SMTP-listener ingress, and **multi-node cluster mode**. Don't pre-build for them — but don't make decisions that close the door on them either. +This section captured the original MVP cut. Most of it has since **shipped in v1.1.x**: queue triggers, cron triggers, inbound email (`email:receive`, HMAC-webhook model), KV / docs / email / users / HTTP SDKs, function-to-function `invoke()`, and secrets are all live (blueprint §12 Phase 4 table). The **Workflows** track (DAG + nested workflows) has since **shipped** (migrations `0071`/`0072`). The **§9.4 service-interceptor** track has since shipped in full (migrations `0073`–`0075`; before/after phases, data-transform, all six services, per-interceptor timeout, resolve cache, `pic interceptors ls`). A read-only **metrics/observability dashboard** has since shipped (A2): `GET /api/v1/admin/apps/{id}/metrics?window=` aggregates the existing `execution_logs` table (counts, error rate, latency avg/p50/p95, an hourly series) via `ExecutionLogRepository::summarize_for_app` + `metrics_api`, surfaced as the dashboard's per-app **Metrics** tab — no hot-path instrumentation. **Still deferred:** **multi-node cluster mode**. Don't pre-build for them — but don't make decisions that close the door on them either. **Pulled forward to Phase 3 (pre-v1.1):** admin auth, multi-app scoping. The general cross-app **export/import** sharing model stays at v1.3+; note that v1.2 §11.6 shipped a narrower form — **group-owned shared collections** (KV/docs/files/topics/queues) let apps in one subtree share data through the owning group, with the ancestor-chain walk as the isolation boundary. See blueprint §11.5 + design-doc §11.6. diff --git a/crates/manager-core/migrations/0076_email_inbound_address.sql b/crates/manager-core/migrations/0076_email_inbound_address.sql new file mode 100644 index 0000000..6cdb8c9 --- /dev/null +++ b/crates/manager-core/migrations/0076_email_inbound_address.sql @@ -0,0 +1,13 @@ +-- A3 raw SMTP ingress — an inbound email ADDRESS an email trigger listens on. +-- +-- The HMAC-webhook path (v1.1.7) addresses a trigger by its UUID in the URL. A +-- native SMTP listener instead receives `RCPT TO:
` and must resolve +-- that address to the bound trigger. `inbound_address` is the mailbox the +-- trigger claims; NULL for a webhook-only trigger. Case-insensitively unique +-- among app-owned triggers so an address resolves to exactly one script. +ALTER TABLE email_trigger_details + ADD COLUMN inbound_address TEXT; + +CREATE UNIQUE INDEX email_trigger_details_inbound_address_uidx + ON email_trigger_details (lower(inbound_address)) + WHERE inbound_address IS NOT NULL; diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index 49651fe..d602287 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -269,7 +269,8 @@ pub use trigger_repo::{ collection_matches, CreateDeadLetterTrigger, CreateDocsTrigger, CreateEmailTrigger, CreateFilesTrigger, CreateKvTrigger, CreatePubsubTrigger, DeadLetterTriggerMatch, DocsTriggerMatch, EmailInboundTarget, FilesTriggerMatch, KvTriggerMatch, PostgresTriggerRepo, - Trigger, TriggerDetails, TriggerDispatchMode, TriggerKind, TriggerRepo, TriggerRepoError, + SmtpInboundTarget, Trigger, TriggerDetails, TriggerDispatchMode, TriggerKind, TriggerRepo, + TriggerRepoError, }; pub use triggers_api::{triggers_router, TriggersApiError, TriggersState}; pub use users_admin_api::{app_users_router, AppUsersApiError, AppUsersState}; diff --git a/crates/manager-core/src/trigger_repo.rs b/crates/manager-core/src/trigger_repo.rs index 3358b0b..d3d11d1 100644 --- a/crates/manager-core/src/trigger_repo.rs +++ b/crates/manager-core/src/trigger_repo.rs @@ -316,6 +316,21 @@ pub struct CreateEmailTrigger { /// Envelope version of the sealed inbound secret (0 = legacy no-AAD, 1 = /// AAD-bound to the sealing owner). `0` when there is no secret. pub inbound_secret_version: i16, + /// A3: the mailbox this trigger receives at over the native SMTP listener. + /// `None` = webhook-only (addressed by trigger UUID in the URL). + pub inbound_address: Option, + pub registered_by_principal: AdminUserId, +} + +/// A3: the SMTP-ingress lookup result — the app-owned email trigger claiming an +/// `RCPT TO` address, plus what an `Email` outbox row needs. No secret: the +/// SMTP path has no HMAC (the mailbox address + per-app isolation are the +/// boundary), unlike the webhook path. +#[derive(Debug, Clone)] +pub struct SmtpInboundTarget { + pub trigger_id: TriggerId, + pub app_id: AppId, + pub script_id: ScriptId, pub registered_by_principal: AdminUserId, } @@ -475,6 +490,17 @@ pub trait TriggerRepo: Send + Sync { trigger_id: TriggerId, ) -> Result, TriggerRepoError>; + /// A3: resolve an inbound SMTP `RCPT TO` address to the app-owned email + /// trigger claiming it (case-insensitive). `None` when no enabled app + /// trigger claims the address. Defaults to `None` so non-Postgres impls + /// (tests) need not provide it. + async fn email_inbound_target_by_address( + &self, + _address: &str, + ) -> Result, TriggerRepoError> { + Ok(None) + } + async fn list_for_app(&self, app_id: AppId) -> Result, TriggerRepoError>; /// Group-owned trigger TEMPLATES (§11 tail) declared directly at `group_id`. @@ -1420,13 +1446,15 @@ impl TriggerRepo for PostgresTriggerRepo { sqlx::query( "INSERT INTO email_trigger_details \ - (trigger_id, inbound_secret_encrypted, inbound_secret_nonce, inbound_secret_version) \ - VALUES ($1, $2, $3, $4)", + (trigger_id, inbound_secret_encrypted, inbound_secret_nonce, \ + inbound_secret_version, inbound_address) \ + VALUES ($1, $2, $3, $4, $5)", ) .bind(parent.id) .bind(req.inbound_secret_encrypted.as_deref()) .bind(req.inbound_secret_nonce.as_deref()) .bind(req.inbound_secret_version) + .bind(req.inbound_address.as_deref()) .execute(&mut *tx) .await?; @@ -1493,6 +1521,33 @@ impl TriggerRepo for PostgresTriggerRepo { })) } + async fn email_inbound_target_by_address( + &self, + address: &str, + ) -> Result, TriggerRepoError> { + // `t.app_id IS NOT NULL` excludes group templates (only their + // materialized per-app copies are invocable); `t.enabled` skips inert + // triggers. The address is matched case-insensitively (mailboxes are). + let row: Option<(Uuid, Uuid, Uuid, Uuid)> = sqlx::query_as( + "SELECT t.id, t.app_id, t.script_id, t.registered_by_principal \ + FROM triggers t \ + JOIN email_trigger_details d ON d.trigger_id = t.id \ + WHERE lower(d.inbound_address) = lower($1) \ + AND t.kind = 'email' AND t.app_id IS NOT NULL AND t.enabled", + ) + .bind(address) + .fetch_optional(&self.pool) + .await?; + Ok( + row.map(|(tid, app_id, script_id, principal)| SmtpInboundTarget { + trigger_id: tid.into(), + app_id: app_id.into(), + script_id: script_id.into(), + registered_by_principal: principal.into(), + }), + ) + } + async fn list_for_app(&self, app_id: AppId) -> Result, TriggerRepoError> { let parents: Vec = sqlx::query_as( "SELECT id, app_id, script_id, name, kind, enabled, sealed, shared, materialized_from, \ diff --git a/crates/manager-core/src/triggers_api.rs b/crates/manager-core/src/triggers_api.rs index ddb3474..f2136c4 100644 --- a/crates/manager-core/src/triggers_api.rs +++ b/crates/manager-core/src/triggers_api.rs @@ -576,6 +576,10 @@ struct CreateEmailTriggerRequest { /// (or absent) means the trigger accepts unsigned POSTs. #[serde(default)] inbound_secret: Option, + /// A3: the mailbox this trigger receives at over the native SMTP listener + /// (`RCPT TO`). Absent = webhook-only. Case-insensitively unique. + #[serde(default)] + inbound_address: Option, } async fn create_email_trigger( @@ -628,6 +632,10 @@ async fn create_email_trigger( inbound_secret_encrypted, inbound_secret_nonce, inbound_secret_version: version, + inbound_address: input + .inbound_address + .map(|a| a.trim().to_string()) + .filter(|a| !a.is_empty()), registered_by_principal: principal.user_id, }; let created = s.triggers.create_email_trigger(app_id, req).await?; diff --git a/crates/manager-core/tests/expected_schema.txt b/crates/manager-core/tests/expected_schema.txt index 4398bea..3247e90 100644 --- a/crates/manager-core/tests/expected_schema.txt +++ b/crates/manager-core/tests/expected_schema.txt @@ -183,6 +183,7 @@ table: email_trigger_details inbound_secret_encrypted: bytea NULL inbound_secret_nonce: bytea NULL inbound_secret_version: smallint NOT NULL default=0 + inbound_address: text NULL table: execution_logs id: uuid NOT NULL default=gen_random_uuid() @@ -614,6 +615,7 @@ indexes on docs_trigger_details: docs_trigger_details_pkey: public.docs_trigger_details USING btree (trigger_id) indexes on email_trigger_details: + email_trigger_details_inbound_address_uidx: public.email_trigger_details USING btree (lower(inbound_address)) WHERE (inbound_address IS NOT NULL) email_trigger_details_pkey: public.email_trigger_details USING btree (trigger_id) indexes on execution_logs: @@ -1151,3 +1153,4 @@ constraints on workflows: 0073: interceptors 0074: interceptor phase 0075: interceptor timeout + 0076: email inbound address diff --git a/crates/picloud/Cargo.toml b/crates/picloud/Cargo.toml index 9c441c2..ae3140a 100644 --- a/crates/picloud/Cargo.toml +++ b/crates/picloud/Cargo.toml @@ -34,6 +34,7 @@ thiserror.workspace = true tracing.workspace = true tracing-subscriber.workspace = true figment.workspace = true +chrono.workspace = true [dev-dependencies] picloud-test-support.workspace = true diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index 582c677..8d18506 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -3,6 +3,8 @@ //! the listener. Tests use the same `build_app` against an //! ephemeral test database. +pub mod smtp; + use std::sync::Arc; use std::time::Duration; diff --git a/crates/picloud/src/main.rs b/crates/picloud/src/main.rs index c05afd6..0ae6ecf 100644 --- a/crates/picloud/src/main.rs +++ b/crates/picloud/src/main.rs @@ -8,12 +8,13 @@ use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; +use picloud::smtp::SmtpIngress; use picloud::{build_app, init_db, AuthDeps}; use picloud_manager_core::{ auth::{hash_password, validate_password_hash}, bootstrap_first_admin, migrations, seed_hello_world_if_fresh, AdminSessionRepository, - AdminUserRepository, HelloWorldOutcome, PostgresAppRepository, PostgresRouteRepository, - PostgresScriptRepository, + AdminUserRepository, HelloWorldOutcome, PostgresAppRepository, PostgresOutboxRepo, + PostgresRouteRepository, PostgresScriptRepository, PostgresTriggerRepo, }; use tracing_subscriber::EnvFilter; @@ -74,6 +75,25 @@ async fn run_server() -> anyhow::Result<()> { // so a delayed sweep can't extend session lifetimes. spawn_session_pruner(auth.sessions.clone()); + // A3: opt-in native SMTP ingress. Binds only when PICLOUD_SMTP_BIND is set; + // shares the pool, resolves each RCPT TO to an app-owned email trigger, and + // writes Email outbox rows the dispatcher fires (same tail as the webhook). + // Its own shutdown_signal() future resolves on the same ctrl-c/SIGTERM. + if let Ok(smtp_bind) = std::env::var("PICLOUD_SMTP_BIND") { + let smtp_addr: SocketAddr = smtp_bind + .parse() + .map_err(|e| anyhow::anyhow!("invalid PICLOUD_SMTP_BIND `{smtp_bind}`: {e}"))?; + let ingress = SmtpIngress::new( + Arc::new(PostgresTriggerRepo::new(pool.clone())), + Arc::new(PostgresOutboxRepo::new(pool.clone())), + ); + tokio::spawn(async move { + if let Err(e) = ingress.serve(smtp_addr, shutdown_signal()).await { + tracing::error!(?e, "smtp ingress failed"); + } + }); + } + let app = build_app(pool, auth, master_key).await?; let listener = tokio::net::TcpListener::bind(addr).await?; diff --git a/crates/picloud/src/smtp.rs b/crates/picloud/src/smtp.rs new file mode 100644 index 0000000..ae7a29a --- /dev/null +++ b/crates/picloud/src/smtp.rs @@ -0,0 +1,342 @@ +//! A3 — a minimal, receive-only native SMTP listener. +//! +//! The v1.1.7 inbound-email path is an HMAC-signed HTTP webhook. This adds a +//! first-party SMTP listener so an MX can point straight at PiCloud: it speaks +//! just enough SMTP to accept a message (`HELO`/`EHLO`, `MAIL`, `RCPT`, `DATA`, +//! `RSET`, `NOOP`, `QUIT`), resolves each `RCPT TO` mailbox to the app-owned +//! email trigger that claims it, and inserts an `Email` outbox row exactly as +//! the webhook does — so the dispatcher fires the bound script unchanged. +//! +//! **Security model.** No AUTH and no STARTTLS at the listener (TLS terminates +//! upstream — Caddy/stunnel or the sending relay); the recipient *address* plus +//! the existing per-app isolation are the boundary (an address resolves to +//! exactly one app-owned trigger). Opt-in: the listener binds only when +//! `PICLOUD_SMTP_BIND` is set. `DATA` is hard-capped so a single session can't +//! stream unbounded bytes. +//! +//! **Parsing.** A deliberately minimal RFC-5322 split — headers up to the first +//! blank line (`From`/`Subject`/`Message-ID` extracted), the rest as the text +//! body. No MIME/multipart decoding (a follow-up); an HTML-only or multipart +//! message still delivers, with the raw body as `text`. + +use std::net::SocketAddr; +use std::sync::Arc; + +use picloud_manager_core::{NewOutboxRow, OutboxRepo, OutboxSourceKind, TriggerRepo}; +use picloud_shared::TriggerEvent; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{TcpListener, TcpStream}; + +/// Default `DATA` ceiling (1 MiB) — matches the webhook body cap. +const DEFAULT_MAX_MESSAGE_BYTES: usize = 1024 * 1024; + +/// The SMTP ingress: resolves recipients and writes outbox rows. Cheap to +/// clone (two `Arc`s). +#[derive(Clone)] +pub struct SmtpIngress { + triggers: Arc, + outbox: Arc, + max_message_bytes: usize, +} + +impl SmtpIngress { + #[must_use] + pub fn new(triggers: Arc, outbox: Arc) -> Self { + Self { + triggers, + outbox, + max_message_bytes: DEFAULT_MAX_MESSAGE_BYTES, + } + } + + /// Is `address` a live mailbox (an enabled app-owned email trigger claims + /// it)? Used to accept/reject `RCPT TO` per the SMTP convention. + async fn mailbox_exists(&self, address: &str) -> bool { + matches!( + self.triggers.email_inbound_target_by_address(address).await, + Ok(Some(_)) + ) + } + + /// Deliver one received message to every matching recipient: resolve the + /// address → trigger, build the `Email` event, insert an outbox row (the + /// dispatcher fires the script). Returns the number of rows written. Unknown + /// mailboxes and per-recipient failures are skipped (logged), never fatal to + /// the others. + pub async fn deliver(&self, recipients: &[String], raw: &[u8]) -> usize { + let msg = ParsedMessage::parse(raw); + let mut delivered = 0; + for rcpt in recipients { + let target = match self.triggers.email_inbound_target_by_address(rcpt).await { + Ok(Some(t)) => t, + Ok(None) => continue, // unknown mailbox — silently drop + Err(e) => { + tracing::error!(?e, rcpt, "smtp: mailbox lookup failed"); + continue; + } + }; + let event = TriggerEvent::Email { + from: msg.from.clone(), + to: recipients.to_vec(), + cc: Vec::new(), + subject: msg.subject.clone(), + text: Some(msg.body.clone()), + html: None, + received_at: chrono::Utc::now(), + message_id: msg.message_id.clone(), + }; + let payload = match serde_json::to_value(&event) { + Ok(p) => p, + Err(e) => { + tracing::error!(?e, "smtp: serialize email event failed"); + continue; + } + }; + let row = NewOutboxRow { + app_id: target.app_id, + source_kind: OutboxSourceKind::Email, + trigger_id: Some(target.trigger_id), + script_id: Some(target.script_id), + reply_to: None, + payload, + origin_principal: Some(target.registered_by_principal), + trigger_depth: 1, + root_execution_id: None, + }; + match self.outbox.insert(row).await { + Ok(_) => delivered += 1, + Err(e) => tracing::error!(?e, rcpt, "smtp: outbox insert failed"), + } + } + delivered + } + + /// Bind `addr` and serve until `shutdown` resolves. Each connection is + /// handled on its own task; a connection error is logged, never fatal. + pub async fn serve( + self, + addr: SocketAddr, + shutdown: impl std::future::Future + Send, + ) -> std::io::Result<()> { + let listener = TcpListener::bind(addr).await?; + tracing::info!(%addr, "smtp ingress listening"); + tokio::pin!(shutdown); + loop { + tokio::select! { + () = &mut shutdown => break, + accepted = listener.accept() => { + match accepted { + Ok((stream, peer)) => { + let me = self.clone(); + tokio::spawn(async move { + if let Err(e) = me.handle_conn(stream).await { + tracing::debug!(?e, %peer, "smtp connection ended"); + } + }); + } + Err(e) => tracing::warn!(?e, "smtp accept failed"), + } + } + } + } + Ok(()) + } + + /// Drive one SMTP session to completion. + async fn handle_conn(&self, stream: TcpStream) -> std::io::Result<()> { + let (read, mut write) = stream.into_split(); + let mut reader = BufReader::new(read); + write.write_all(b"220 picloud ESMTP ready\r\n").await?; + + let mut recipients: Vec = Vec::new(); + let mut line = String::new(); + loop { + line.clear(); + if reader.read_line(&mut line).await? == 0 { + break; // peer closed + } + let cmd = line.trim_end_matches(['\r', '\n']); + let verb = cmd + .split_whitespace() + .next() + .unwrap_or("") + .to_ascii_uppercase(); + match verb.as_str() { + "HELO" | "EHLO" => write.write_all(b"250 picloud\r\n").await?, + "MAIL" | "NOOP" => write.write_all(b"250 OK\r\n").await?, + "RCPT" => match extract_address(cmd) { + Some(addr) if self.mailbox_exists(&addr).await => { + recipients.push(addr); + write.write_all(b"250 OK\r\n").await?; + } + _ => { + write.write_all(b"550 5.1.1 no such mailbox\r\n").await?; + } + }, + "DATA" => { + if recipients.is_empty() { + write + .write_all(b"554 5.5.1 no valid recipients\r\n") + .await?; + continue; + } + write + .write_all(b"354 End data with .\r\n") + .await?; + if let Some(raw) = read_data(&mut reader, self.max_message_bytes).await? { + let n = self.deliver(&recipients, &raw).await; + recipients.clear(); + write + .write_all(format!("250 OK queued {n}\r\n").as_bytes()) + .await?; + } else { + recipients.clear(); + write.write_all(b"552 5.3.4 message too large\r\n").await?; + } + } + "RSET" => { + recipients.clear(); + write.write_all(b"250 OK\r\n").await?; + } + "QUIT" => { + write.write_all(b"221 2.0.0 Bye\r\n").await?; + break; + } + _ => { + write + .write_all(b"502 5.5.2 command not recognized\r\n") + .await?; + } + } + } + Ok(()) + } +} + +/// Read the `DATA` payload: lines until a lone `.`, with dot-unstuffing (a line +/// starting `..` becomes `.`). Bounded by a cumulative byte cap — over it, we +/// drain to the terminator (to keep the socket in sync) and return `None` so the +/// caller replies 552. (TLS/relay upstream bounds any single unterminated line.) +async fn read_data( + reader: &mut R, + max_bytes: usize, +) -> std::io::Result>> { + let mut data: Vec = Vec::new(); + let mut total = 0usize; + let mut line = String::new(); + loop { + line.clear(); + let n = reader.read_line(&mut line).await?; + if n == 0 { + break; // connection dropped mid-DATA + } + total += n; + let trimmed = line.trim_end_matches(['\r', '\n']); + if trimmed == "." { + return Ok(Some(data)); + } + if total > max_bytes { + drain_to_terminator(reader).await?; + return Ok(None); + } + // Dot-unstuffing (RFC 5321 §4.5.2). + let content = trimmed.strip_prefix('.').unwrap_or(trimmed); + data.extend_from_slice(content.as_bytes()); + data.extend_from_slice(b"\r\n"); + } + Ok(Some(data)) +} + +/// Drain remaining `DATA` lines up to the `.` terminator after an overflow, so +/// the connection stays protocol-synchronized. Bounded by a line budget. +async fn drain_to_terminator(reader: &mut R) -> std::io::Result<()> { + let mut line = String::new(); + for _ in 0..1_000_000 { + line.clear(); + let n = reader.read_line(&mut line).await?; + if n == 0 || line.trim_end_matches(['\r', '\n']) == "." { + break; + } + } + Ok(()) +} + +/// Extract the address from a `RCPT TO:` (or `RCPT TO: addr`) command. +fn extract_address(cmd: &str) -> Option { + let after = cmd.split_once(':')?.1.trim(); + let inner = after + .strip_prefix('<') + .and_then(|s| s.strip_suffix('>')) + .unwrap_or(after); + let addr = inner.trim(); + (!addr.is_empty() && addr.contains('@')).then(|| addr.to_string()) +} + +/// The minimal parse of a received message. +struct ParsedMessage { + from: String, + subject: String, + message_id: Option, + body: String, +} + +impl ParsedMessage { + fn parse(raw: &[u8]) -> Self { + let text = String::from_utf8_lossy(raw); + // Headers up to the first blank line; the rest is the body. + let (headers, body) = match text.split_once("\r\n\r\n") { + Some((h, b)) => (h, b), + None => text.split_once("\n\n").unwrap_or((&text, "")), + }; + let mut from = String::new(); + let mut subject = String::new(); + let mut message_id = None; + for header_line in headers.lines() { + if let Some((name, value)) = header_line.split_once(':') { + let value = value.trim().to_string(); + match name.trim().to_ascii_lowercase().as_str() { + "from" => from = value, + "subject" => subject = value, + "message-id" => message_id = Some(value), + _ => {} + } + } + } + Self { + from, + subject, + message_id, + body: body.to_string(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn extract_address_handles_angle_brackets_and_bare() { + assert_eq!( + extract_address("RCPT TO:"), + Some("a@b.com".to_string()) + ); + assert_eq!( + extract_address("RCPT TO: c@d.com"), + Some("c@d.com".to_string()) + ); + assert_eq!(extract_address("RCPT TO:<>"), None); + assert_eq!(extract_address("RCPT TO:not-an-address"), None); + } + + #[test] + fn parse_extracts_headers_and_body() { + let raw = + b"From: alice@example.com\r\nSubject: Hi\r\nMessage-ID: <1@x>\r\n\r\nhello world\r\n"; + let m = ParsedMessage::parse(raw); + assert_eq!(m.from, "alice@example.com"); + assert_eq!(m.subject, "Hi"); + assert_eq!(m.message_id.as_deref(), Some("<1@x>")); + assert!(m.body.contains("hello world")); + } +} diff --git a/crates/picloud/tests/smtp_ingress.rs b/crates/picloud/tests/smtp_ingress.rs new file mode 100644 index 0000000..f7d8710 --- /dev/null +++ b/crates/picloud/tests/smtp_ingress.rs @@ -0,0 +1,105 @@ +//! A3 — the native SMTP ingress delivery core. Seeds an app-owned email trigger +//! with an inbound `RCPT TO` address, then drives `SmtpIngress::deliver` +//! directly (no socket — deterministic) and asserts it writes exactly one +//! `Email` outbox row for the known mailbox and none for an unknown one. The +//! dispatcher tail (outbox → script) is covered by the existing dispatcher +//! tests; this pins the resolve-address → outbox step the SMTP path adds. + +use std::sync::Arc; + +use picloud::smtp::SmtpIngress; +use picloud_manager_core::{PostgresOutboxRepo, PostgresTriggerRepo}; +use sqlx::PgPool; +use uuid::Uuid; + +mod common; + +async fn seed(pool: &PgPool, address: &str) -> Uuid { + let group: (Uuid,) = + sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id") + .bind(format!("smtp-g-{}", Uuid::new_v4())) + .fetch_one(pool) + .await + .expect("group"); + let app: (Uuid,) = + sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id") + .bind(format!("smtp-a-{}", Uuid::new_v4())) + .bind(group.0) + .fetch_one(pool) + .await + .expect("app"); + let admin: (Uuid,) = sqlx::query_as( + "INSERT INTO admin_users (username, password_hash) VALUES ($1, 'x') RETURNING id", + ) + .bind(format!("smtp-u-{}", Uuid::new_v4())) + .fetch_one(pool) + .await + .expect("admin"); + let script: (Uuid,) = sqlx::query_as( + "INSERT INTO scripts (name, source, app_id) VALUES ($1, 'x', $2) RETURNING id", + ) + .bind(format!("smtp-s-{}", Uuid::new_v4())) + .bind(app.0) + .fetch_one(pool) + .await + .expect("script"); + let trigger: (Uuid,) = sqlx::query_as( + "INSERT INTO triggers \ + (app_id, script_id, kind, enabled, dispatch_mode, retry_max_attempts, \ + retry_backoff, retry_base_ms, registered_by_principal, name) \ + VALUES ($1, $2, 'email', TRUE, 'async', 3, 'exponential', 1000, $3, $4) RETURNING id", + ) + .bind(app.0) + .bind(script.0) + .bind(admin.0) + .bind(format!("smtp-t-{}", Uuid::new_v4())) + .fetch_one(pool) + .await + .expect("trigger"); + sqlx::query("INSERT INTO email_trigger_details (trigger_id, inbound_address) VALUES ($1, $2)") + .bind(trigger.0) + .bind(address) + .execute(pool) + .await + .expect("email details"); + app.0 +} + +async fn email_outbox_count(pool: &PgPool, app_id: Uuid) -> i64 { + let row: (i64,) = + sqlx::query_as("SELECT count(*) FROM outbox WHERE app_id = $1 AND source_kind = 'email'") + .bind(app_id) + .fetch_one(pool) + .await + .expect("count"); + row.0 +} + +#[tokio::test] +async fn deliver_writes_one_outbox_row_for_a_known_mailbox() { + let Some(pool) = common::test_pool("smtp_ingress").await else { + return; + }; + let address = "hooks@app.test"; + let app_id = seed(&pool, address).await; + + let ingress = SmtpIngress::new( + Arc::new(PostgresTriggerRepo::new(pool.clone())), + Arc::new(PostgresOutboxRepo::new(pool.clone())), + ); + let raw = b"From: alice@example.com\r\nSubject: Hi there\r\n\r\nhello body\r\n"; + + // Known mailbox → one outbox row. + let delivered = ingress.deliver(&[address.to_string()], raw).await; + assert_eq!(delivered, 1, "a known mailbox must be delivered"); + assert_eq!(email_outbox_count(&pool, app_id).await, 1); + + // Unknown mailbox → nothing (the resolution IS the boundary). + let delivered = ingress.deliver(&["nobody@app.test".to_string()], raw).await; + assert_eq!(delivered, 0, "an unknown mailbox must not deliver"); + assert_eq!( + email_outbox_count(&pool, app_id).await, + 1, + "no extra outbox row for an unknown mailbox" + ); +}