diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 0b95c1c..cc948ef 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -54,16 +54,68 @@ use nuncio_store::search::SearchEngine; use nuncio_store::vault::SecretManager; use std::pin::Pin; use std::sync::Arc; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use subtle::ConstantTimeEq; use thiserror::Error; use tokio::net::TcpListener; use tokio_stream::wrappers::errors::BroadcastStreamRecvError; use tokio_stream::wrappers::{BroadcastStream, ReceiverStream, TcpListenerStream}; use tokio_stream::{Stream, StreamExt}; +use tonic::service::interceptor::InterceptedService; use tonic::transport::Server; use tonic::{Request, Response, Status}; +// Resource bounds applied to the tonic `Server` builder in +// `serve_on_listener_with_overrides_and_shutdown`. These exist so a +// misbehaving or malicious client cannot exhaust the daemon by opening +// unbounded connections/streams, sending oversized frames, or holding a +// handler open forever. + +/// Caps in-flight requests per client connection. A single connection that +/// opens far more concurrent streams than any legitimate client needs (the +/// CLI and future native clients issue a handful of concurrent calls at +/// most) is throttled rather than allowed to consume worker capacity. +const GRPC_CONCURRENCY_LIMIT_PER_CONNECTION: usize = 32; + +/// Per-call deadline applied by the tonic transport to every RPC's request +/// future. This bounds unary handlers (a stuck backend call must not hang +/// the connection forever) while staying generous enough for the longest +/// legitimate unary call, `Mail.Sync`/`Calendar.Sync`/`Contacts.Sync` +/// against a large, slow upstream mailbox. +/// +/// Server-streaming RPCs (`System.Subscribe`) are unaffected in practice: +/// tonic's timeout wraps the handler's future, which for a streaming +/// response resolves as soon as the response headers/body stream are +/// constructed, not when the stream itself finishes emitting items -- so a +/// long-lived `Subscribe` connection is not cut off by this deadline. +const GRPC_REQUEST_TIMEOUT: Duration = Duration::from_secs(300); + +/// HTTP/2 concurrent stream cap per connection, independent of the +/// concurrency limiter above (this bounds protocol-level multiplexing; +/// `concurrency_limit_per_connection` bounds how many of those streams are +/// actively dispatched to a handler at once). +const GRPC_MAX_CONCURRENT_STREAMS: u32 = 64; + +/// Largest inbound gRPC message the daemon will decode, per service. +/// Deliberately larger than tonic's unconfigured 4 MiB default: `Mail` +/// requests can legitimately carry a full message body plus attachment +/// bytes inline (`SendMessageRequest.attachments`), so 4 MiB is too tight +/// for a real email with a modest attachment. 16 MiB comfortably covers +/// that case while still bounding the allocation a client can force with +/// one oversized frame far below "unbounded". +const GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES: usize = 16 * 1024 * 1024; + +/// TCP-level keepalive so a half-open connection (client crashed or network +/// dropped without a clean close) is detected and reclaimed instead of +/// holding a connection slot indefinitely. +const GRPC_TCP_KEEPALIVE: Duration = Duration::from_secs(60); + +/// HTTP/2 PING-based keepalive: send a ping after this much idle time... +const GRPC_HTTP2_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30); + +/// ...and close the connection if the pong doesn't arrive within this long. +const GRPC_HTTP2_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(20); + // The gRPC loopback address defaults/resolver live in `nuncio-proto` (see // `nuncio_proto::addr`) so that thin presentation-shell clients (e.g. // `nuncio-cli`) agree on the same default address without depending on this @@ -2625,7 +2677,11 @@ pub async fn serve_on_listener_with_overrides_and_shutdown( event_bus: event_bus.clone(), }; let system_interceptor = BearerAuthInterceptor::new(token.clone()); - let system_svc = SystemServer::with_interceptor(system_service, system_interceptor); + let system_svc = InterceptedService::new( + SystemServer::new(system_service) + .max_decoding_message_size(GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES), + system_interceptor, + ); let accounts_service = AccountsGrpcService { db: db.clone(), @@ -2635,7 +2691,11 @@ pub async fn serve_on_listener_with_overrides_and_shutdown( .unwrap_or_else(|| Arc::new(RealAccountConnectionTester)), }; let accounts_interceptor = BearerAuthInterceptor::new(token.clone()); - let accounts_svc = AccountsServer::with_interceptor(accounts_service, accounts_interceptor); + let accounts_svc = InterceptedService::new( + AccountsServer::new(accounts_service) + .max_decoding_message_size(GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES), + accounts_interceptor, + ); // Mail: mounted behind its own `BearerAuthInterceptor`, exactly like // `System` and `Accounts` above -- see the hard invariant documented on @@ -2648,7 +2708,11 @@ pub async fn serve_on_listener_with_overrides_and_shutdown( overrides, }; let mail_interceptor = BearerAuthInterceptor::new(token.clone()); - let mail_svc = MailServer::with_interceptor(mail_service, mail_interceptor); + let mail_svc = InterceptedService::new( + MailServer::new(mail_service) + .max_decoding_message_size(GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES), + mail_interceptor, + ); // Filters: mounted behind its own `BearerAuthInterceptor`, exactly like // `System`/`Accounts`/`Mail` above -- see the hard invariant documented @@ -2658,21 +2722,33 @@ pub async fn serve_on_listener_with_overrides_and_shutdown( filter_engine, }; let filters_interceptor = BearerAuthInterceptor::new(token.clone()); - let filters_svc = FiltersServer::with_interceptor(filters_service, filters_interceptor); + let filters_svc = InterceptedService::new( + FiltersServer::new(filters_service) + .max_decoding_message_size(GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES), + filters_interceptor, + ); // Export: mounted behind its own `BearerAuthInterceptor`, exactly like // every other service above -- see the hard invariant documented on // this function's doc comment. let export_service = ExportGrpcService { db: db.clone() }; let export_interceptor = BearerAuthInterceptor::new(token.clone()); - let export_svc = ExportServer::with_interceptor(export_service, export_interceptor); + let export_svc = InterceptedService::new( + ExportServer::new(export_service) + .max_decoding_message_size(GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES), + export_interceptor, + ); // Audit: mounted behind its own `BearerAuthInterceptor`, exactly like // every other service above -- see the hard invariant documented on // this function's doc comment. let audit_service = AuditGrpcService { db: db.clone() }; let audit_interceptor = BearerAuthInterceptor::new(token.clone()); - let audit_svc = AuditServer::with_interceptor(audit_service, audit_interceptor); + let audit_svc = InterceptedService::new( + AuditServer::new(audit_service) + .max_decoding_message_size(GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES), + audit_interceptor, + ); // Calendar: mounted behind its own `BearerAuthInterceptor`, exactly like // every other service above -- see the hard invariant documented on @@ -2683,7 +2759,11 @@ pub async fn serve_on_listener_with_overrides_and_shutdown( overrides: calendar_overrides, }; let calendar_interceptor = BearerAuthInterceptor::new(token.clone()); - let calendar_svc = CalendarServer::with_interceptor(calendar_service, calendar_interceptor); + let calendar_svc = InterceptedService::new( + CalendarServer::new(calendar_service) + .max_decoding_message_size(GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES), + calendar_interceptor, + ); // Contacts: mounted behind its own `BearerAuthInterceptor`, exactly like // every other service above -- see the hard invariant documented on @@ -2693,9 +2773,19 @@ pub async fn serve_on_listener_with_overrides_and_shutdown( overrides: contacts_overrides, }; let contacts_interceptor = BearerAuthInterceptor::new(token); - let contacts_svc = ContactsServer::with_interceptor(contacts_service, contacts_interceptor); + let contacts_svc = InterceptedService::new( + ContactsServer::new(contacts_service) + .max_decoding_message_size(GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES), + contacts_interceptor, + ); Server::builder() + .concurrency_limit_per_connection(GRPC_CONCURRENCY_LIMIT_PER_CONNECTION) + .timeout(GRPC_REQUEST_TIMEOUT) + .max_concurrent_streams(GRPC_MAX_CONCURRENT_STREAMS) + .tcp_keepalive(Some(GRPC_TCP_KEEPALIVE)) + .http2_keepalive_interval(Some(GRPC_HTTP2_KEEPALIVE_INTERVAL)) + .http2_keepalive_timeout(Some(GRPC_HTTP2_KEEPALIVE_TIMEOUT)) .add_service(system_svc) .add_service(accounts_svc) .add_service(mail_svc) @@ -4306,6 +4396,53 @@ mod tests { assert_eq!(err.code(), Code::InvalidArgument); } + /// Proves the server's configured `max_decoding_message_size` (see + /// `GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES`) is actually wired into the + /// `Server::builder()` chain and not just documented: a message body + /// that exceeds the configured limit is rejected with + /// `OutOfRange` at decode time, before the handler (and its field + /// validation, or the "no account configured" business-logic error) + /// ever runs. + #[tokio::test] + async fn send_message_rejects_body_larger_than_max_decoding_message_size() { + let event_bus = Arc::new(EventBus::new()); + let (addr, _handle, _dir) = spawn_test_server(event_bus, "correct-token").await; + + let mut client = MailClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + let mut req = valid_send_message_request(); + req.body_text = "x".repeat(GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES + 1); + let err = client + .send_message(authed_bearer_request(req)) + .await + .expect_err("oversized message must be rejected"); + assert_eq!(err.code(), Code::OutOfRange); + } + + /// Proves the raised limit is real, not a no-op: a body larger than + /// tonic's unconfigured 4 MiB default but still under the daemon's + /// configured `GRPC_MAX_DECODING_MESSAGE_SIZE_BYTES` decodes + /// successfully and reaches the handler -- surfacing the expected + /// "no account configured" business error rather than + /// `OutOfRange`. + #[tokio::test] + async fn send_message_accepts_body_above_default_but_within_configured_limit() { + let event_bus = Arc::new(EventBus::new()); + let (addr, _handle, _dir) = spawn_test_server(event_bus, "correct-token").await; + + let mut client = MailClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + let mut req = valid_send_message_request(); + req.body_text = "x".repeat(6 * 1024 * 1024); + let err = client + .send_message(authed_bearer_request(req)) + .await + .expect_err("no configured account must still be rejected"); + assert_eq!(err.code(), Code::Internal); + } + /// Proves `SendMessage` never fabricates success -- with no account /// configured at all, the daemon has nothing to send from, and this /// surfaces as `Status::internal` rather than a fabricated