diff --git a/Cargo.lock b/Cargo.lock index 7aa9999..4265b9c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2612,17 +2612,23 @@ name = "nunciod" version = "0.1.0" dependencies = [ "arc-swap", + "async-trait", + "hex", "nuncio-cal", "nuncio-cli", "nuncio-core", "nuncio-filter", "nuncio-mail", + "nuncio-proto", "nuncio-store", "serde", "serde_json", + "subtle", "tempfile", "thiserror 1.0.69", "tokio", + "tokio-stream", + "tonic", "tracing", "tracing-subscriber", ] diff --git a/Cargo.toml b/Cargo.toml index e16fb47..680a3be 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -83,6 +83,7 @@ tonic-prost = "0.14" prost = "0.14" tonic-prost-build = "0.14" protoc-bin-vendored = "3.2" +subtle = "2.6" diff --git a/crates/nuncio-store/src/vault.rs b/crates/nuncio-store/src/vault.rs index b470814..35601f4 100644 --- a/crates/nuncio-store/src/vault.rs +++ b/crates/nuncio-store/src/vault.rs @@ -22,6 +22,12 @@ pub const WORM_KEY_ACCOUNT: &str = "worm-audit-hmac-key"; /// ledger (see [`crate::db::DatabaseEngine::save_filter_execution_log`]). pub const LEDGER_KEY_ACCOUNT: &str = "ledger-hmac-key"; +/// Keyring account name for the bearer token used to authenticate loopback gRPC calls +/// against the `nuncio.v1` API server exposed by `nunciod` (see backlog story 1.A.2 / +/// GH-149). The token is minted on first use via [`SecretManager::get_or_create_key_bytes`] +/// and hex-encoded for use as an `authorization: Bearer ` header value. +pub const GRPC_TOKEN_ACCOUNT: &str = "grpc-bearer-token"; + /// Zeroized sensitive string wrapper ensuring heap bytes are wiped on drop. #[derive(Debug, Clone, PartialEq, Eq, Zeroize, ZeroizeOnDrop)] pub struct ZeroizingSecret(String); diff --git a/crates/nunciod/Cargo.toml b/crates/nunciod/Cargo.toml index 64ba984..2f47de1 100644 --- a/crates/nunciod/Cargo.toml +++ b/crates/nunciod/Cargo.toml @@ -15,16 +15,22 @@ nuncio-mail = { path = "../nuncio-mail" } nuncio-cal = { path = "../nuncio-cal" } nuncio-store = { path = "../nuncio-store" } nuncio-filter = { path = "../nuncio-filter" } +nuncio-proto = { path = "../nuncio-proto" } arc-swap.workspace = true tokio = { workspace = true, features = ["full"] } +tokio-stream = { workspace = true, features = ["net"] } serde.workspace = true serde_json.workspace = true thiserror.workspace = true tracing.workspace = true tracing-subscriber = "0.3" tempfile.workspace = true +tonic.workspace = true +async-trait.workspace = true +subtle.workspace = true +hex.workspace = true [dev-dependencies] nuncio-cli = { path = "../nuncio-cli" } diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs new file mode 100644 index 0000000..0efa42f --- /dev/null +++ b/crates/nunciod/src/grpc.rs @@ -0,0 +1,319 @@ +//! gRPC server wiring for `nunciod`: serves the versioned `nuncio.v1.System` +//! contract (see `nuncio-proto`, backlog story 1.A.1 / GH-148) over loopback, +//! authenticated by a bearer token sourced from the OS keyring vault +//! (`nuncio_store::vault::GRPC_TOKEN_ACCOUNT`, backlog story 1.A.2 / GH-149). +//! +//! This runs ALONGSIDE the existing hand-rolled JSON-RPC IPC server +//! (`nuncio_core::ipc::IpcDaemonServer`); migrating callers off the JSON-RPC +//! transport is out of scope here and lands in a later story. + +use nuncio_core::EventBus; +use nuncio_proto::v1::system_server::{System, SystemServer}; +use nuncio_proto::v1::{GetStatusRequest, GetStatusResponse}; +use std::sync::Arc; +use subtle::ConstantTimeEq; +use thiserror::Error; +use tokio::net::TcpListener; +use tokio_stream::wrappers::TcpListenerStream; +use tonic::transport::Server; +use tonic::{Request, Response, Status}; + +/// Default loopback bind address for the `nuncio.v1` gRPC server, used when +/// the [`GRPC_ADDR_ENV_VAR`] environment variable is unset. Deliberately +/// distinct from the legacy JSON-RPC IPC server's default (`127.0.0.1:9422`). +pub const DEFAULT_GRPC_ADDR: &str = "127.0.0.1:9420"; + +/// Environment variable overriding the gRPC bind address. +pub const GRPC_ADDR_ENV_VAR: &str = "NUNCIO_GRPC_ADDR"; + +/// Errors that can occur while binding or running the `nuncio.v1` gRPC server. +#[derive(Debug, Error)] +pub enum GrpcServeError { + /// Failed to bind the loopback TCP listener. + #[error("failed to bind gRPC listener on {addr}: {source}")] + Bind { + /// The address that failed to bind. + addr: String, + /// Underlying IO error. + #[source] + source: std::io::Error, + }, + /// The tonic transport server failed while serving requests. + #[error("gRPC transport server failed: {0}")] + Transport(#[from] tonic::transport::Error), +} + +/// Resolves the gRPC bind address from [`GRPC_ADDR_ENV_VAR`], falling back to +/// [`DEFAULT_GRPC_ADDR`] when unset. +pub fn grpc_addr_from_env() -> String { + std::env::var(GRPC_ADDR_ENV_VAR).unwrap_or_else(|_| DEFAULT_GRPC_ADDR.to_string()) +} + +/// `nuncio.v1.System` gRPC service implementation backed by the daemon's live +/// [`EventBus`] state. +struct SystemGrpcService { + event_bus: Arc, +} + +#[tonic::async_trait] +impl System for SystemGrpcService { + async fn get_status( + &self, + _request: Request, + ) -> Result, Status> { + let state = self.event_bus.current_state(); + Ok(Response::new(GetStatusResponse { + engine_status: format!("{:?}", state.status), + version: env!("CARGO_PKG_VERSION").to_string(), + })) + } +} + +/// Bearer-token authentication interceptor for the loopback `nuncio.v1` gRPC +/// server. +/// +/// Requires an `authorization: Bearer ` metadata header on every call +/// and rejects missing, malformed, or mismatched tokens with +/// `Status::unauthenticated`. The comparison against the expected token uses +/// [`subtle::ConstantTimeEq`] so a mismatching guess cannot be distinguished +/// by timing. The token itself is never logged. +#[derive(Clone)] +struct BearerAuthInterceptor { + expected_token: Arc, +} + +impl BearerAuthInterceptor { + fn new(expected_token: impl Into>) -> Self { + Self { + expected_token: expected_token.into(), + } + } +} + +impl tonic::service::Interceptor for BearerAuthInterceptor { + fn call(&mut self, request: Request<()>) -> Result, Status> { + let header = request + .metadata() + .get("authorization") + .ok_or_else(|| Status::unauthenticated("missing authorization metadata"))?; + let header_str = header + .to_str() + .map_err(|_| Status::unauthenticated("authorization metadata is not valid ASCII"))?; + let provided = header_str.strip_prefix("Bearer ").ok_or_else(|| { + Status::unauthenticated("expected 'Bearer ' authorization scheme") + })?; + + let matches: bool = provided + .as_bytes() + .ct_eq(self.expected_token.as_bytes()) + .into(); + if matches { + Ok(request) + } else { + Err(Status::unauthenticated("invalid bearer token")) + } + } +} + +/// Binds a loopback TCP listener at `addr` and serves the `nuncio.v1.System` +/// gRPC service on it, authenticated by `token`, until the transport server +/// errors. +/// +/// `addr` MUST be a loopback address (e.g. `127.0.0.1:PORT`); callers are +/// responsible for passing loopback-only addresses (see +/// [`grpc_addr_from_env`]). +pub async fn serve( + addr: &str, + event_bus: Arc, + token: impl Into>, +) -> Result<(), GrpcServeError> { + let listener = TcpListener::bind(addr) + .await + .map_err(|source| GrpcServeError::Bind { + addr: addr.to_string(), + source, + })?; + serve_on_listener(listener, event_bus, token).await +} + +/// Serves the `nuncio.v1.System` gRPC service on an already-bound +/// [`TcpListener`]. +/// +/// Exposed separately from [`serve`] so tests can bind an ephemeral loopback +/// port (`127.0.0.1:0`), read back the OS-assigned port via +/// `TcpListener::local_addr`, and connect a client to it deterministically: +/// once `TcpListener::bind` returns, the OS is already accepting connections +/// on that port, so no fixed sleep is needed before dialing it. +pub async fn serve_on_listener( + listener: TcpListener, + event_bus: Arc, + token: impl Into>, +) -> Result<(), GrpcServeError> { + let service = SystemGrpcService { event_bus }; + let interceptor = BearerAuthInterceptor::new(token); + let svc = SystemServer::with_interceptor(service, interceptor); + + Server::builder() + .add_service(svc) + .serve_with_incoming(TcpListenerStream::new(listener)) + .await + .map_err(GrpcServeError::Transport) +} + +#[cfg(test)] +mod tests { + use super::*; + use nuncio_core::{CoreCommand, EngineStatus}; + use nuncio_proto::v1::system_client::SystemClient; + use tonic::Code; + + async fn spawn_test_server( + event_bus: Arc, + token: &str, + ) -> (std::net::SocketAddr, tokio::task::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral loopback port"); + let addr = listener.local_addr().expect("listener has local addr"); + let token = token.to_string(); + let handle = tokio::spawn(async move { + let _ = serve_on_listener(listener, event_bus, token).await; + }); + (addr, handle) + } + + #[test] + fn grpc_addr_from_env_defaults_when_unset() { + // Serialize access to the shared process environment variable so + // this test cannot interleave with any other test touching it. + static ENV_GUARD: std::sync::Mutex<()> = std::sync::Mutex::new(()); + let _guard = ENV_GUARD.lock().unwrap_or_else(|e| e.into_inner()); + + std::env::remove_var(GRPC_ADDR_ENV_VAR); + assert_eq!(grpc_addr_from_env(), DEFAULT_GRPC_ADDR); + + std::env::set_var(GRPC_ADDR_ENV_VAR, "127.0.0.1:12345"); + assert_eq!(grpc_addr_from_env(), "127.0.0.1:12345"); + std::env::remove_var(GRPC_ADDR_ENV_VAR); + } + + #[test] + fn grpc_serve_error_messages_do_not_leak_and_are_displayable() { + let bind_err = GrpcServeError::Bind { + addr: "127.0.0.1:9420".to_string(), + source: std::io::Error::new(std::io::ErrorKind::AddrInUse, "in use"), + }; + assert!(bind_err.to_string().contains("127.0.0.1:9420")); + } + + #[tokio::test] + async fn get_status_rejects_missing_bearer_token() { + let event_bus = Arc::new(EventBus::new()); + let (addr, _handle) = spawn_test_server(event_bus, "correct-token").await; + + let mut client = SystemClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + let err = client + .get_status(GetStatusRequest {}) + .await + .expect_err("missing token must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); + } + + #[tokio::test] + async fn get_status_rejects_wrong_bearer_token() { + let event_bus = Arc::new(EventBus::new()); + let (addr, _handle) = spawn_test_server(event_bus, "correct-token").await; + + let mut client = SystemClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + let mut request = Request::new(GetStatusRequest {}); + request.metadata_mut().insert( + "authorization", + "Bearer wrong-token" + .parse() + .expect("valid ascii metadata value"), + ); + let err = client + .get_status(request) + .await + .expect_err("wrong token must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); + } + + #[tokio::test] + async fn get_status_rejects_malformed_authorization_header() { + let event_bus = Arc::new(EventBus::new()); + let (addr, _handle) = spawn_test_server(event_bus, "correct-token").await; + + let mut client = SystemClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + let mut request = Request::new(GetStatusRequest {}); + request.metadata_mut().insert( + "authorization", + "not-a-bearer-scheme" + .parse() + .expect("valid ascii metadata value"), + ); + let err = client + .get_status(request) + .await + .expect_err("malformed scheme must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); + } + + #[tokio::test] + async fn get_status_accepts_correct_bearer_token_and_reports_live_state() { + let event_bus = Arc::new(EventBus::new()); + event_bus.process_command(CoreCommand::SyncAll); + assert_eq!(event_bus.current_state().status, EngineStatus::Syncing); + + let (addr, _handle) = spawn_test_server(event_bus, "correct-token").await; + + let mut client = SystemClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + let mut request = Request::new(GetStatusRequest {}); + request.metadata_mut().insert( + "authorization", + "Bearer correct-token" + .parse() + .expect("valid ascii metadata value"), + ); + let response = client + .get_status(request) + .await + .expect("valid token is accepted") + .into_inner(); + assert_eq!(response.engine_status, "Syncing"); + assert_eq!(response.version, env!("CARGO_PKG_VERSION")); + } + + #[tokio::test] + async fn serve_binds_and_starts_on_a_valid_loopback_address() { + // Exercises the `serve()` wrapper's happy path (bind succeeds, then + // delegates into `serve_on_listener`), as distinct from the + // `spawn_test_server` helper above which pre-binds the listener + // itself and calls `serve_on_listener` directly. + let event_bus = Arc::new(EventBus::new()); + let _handle = tokio::spawn(async move { + let _ = serve("127.0.0.1:0", event_bus, "token").await; + }); + // Yield so the spawned task is polled at least once and actually + // reaches the bind + delegate-to-`serve_on_listener` call before + // this test function returns. + tokio::task::yield_now().await; + } + + #[tokio::test] + async fn serve_fails_closed_when_bind_address_is_invalid() { + let event_bus = Arc::new(EventBus::new()); + let err = serve("not-a-valid-addr", event_bus, "token") + .await + .expect_err("invalid bind address must fail"); + assert!(matches!(err, GrpcServeError::Bind { .. })); + } +} diff --git a/crates/nunciod/src/lib.rs b/crates/nunciod/src/lib.rs index 2e33c98..f82b8b1 100644 --- a/crates/nunciod/src/lib.rs +++ b/crates/nunciod/src/lib.rs @@ -2,6 +2,7 @@ #![cfg_attr(test, allow(clippy::unwrap_used, clippy::expect_used, clippy::panic))] +pub mod grpc; pub mod orchestrator; pub use orchestrator::SelfHealingSyncOrchestrator; diff --git a/crates/nunciod/src/main.rs b/crates/nunciod/src/main.rs index 3e9f7d1..93c9065 100644 --- a/crates/nunciod/src/main.rs +++ b/crates/nunciod/src/main.rs @@ -278,6 +278,30 @@ async fn main() -> Result<(), Box> { }) }); + // gRPC `nuncio.v1.System` server (backlog story 1.A.2 / GH-149). + // + // Runs ALONGSIDE the existing JSON-RPC IPC server below; the migration + // off the hand-rolled JSON-RPC transport happens in later stories. The + // bearer token is minted (or loaded, on subsequent runs) from the real + // OS keyring vault via `SecretManager::production()`, fails closed if + // the keyring is unavailable, and is never logged. + let grpc_secrets = nuncio_store::vault::SecretManager::production(); + let grpc_token_bytes = grpc_secrets + .get_or_create_key_bytes(nuncio_store::vault::GRPC_TOKEN_ACCOUNT, 32) + .map_err(|e| format!("failed to provision gRPC bearer token from vault: {e}"))?; + let grpc_token = hex::encode(grpc_token_bytes); + let grpc_addr = nunciod::grpc::grpc_addr_from_env(); + tracing::info!( + "nunciod gRPC (nuncio.v1.System) starting on {} (loopback only)", + grpc_addr + ); + let grpc_event_bus = event_bus.clone(); + let _grpc_task = tokio::spawn(async move { + if let Err(e) = nunciod::grpc::serve(&grpc_addr, grpc_event_bus, grpc_token).await { + tracing::error!("nunciod gRPC server failed: {}", e); + } + }); + let addr = std::env::var("NUNCIO_IPC_ADDR").unwrap_or_else(|_| "127.0.0.1:9422".to_string()); let server = IpcDaemonServer::with_handler(event_bus.clone(), &addr, handler); diff --git a/crates/nunciod/tests/grpc_auth_test.rs b/crates/nunciod/tests/grpc_auth_test.rs new file mode 100644 index 0000000..86e34f7 --- /dev/null +++ b/crates/nunciod/tests/grpc_auth_test.rs @@ -0,0 +1,127 @@ +//! Integration test for the `nuncio.v1.System` gRPC server wired into +//! `nunciod` (backlog story 1.A.2 / GH-149). +//! +//! Starts a real `nunciod::grpc` server on an ephemeral loopback port and +//! drives it with the generated `nuncio-proto` tonic client to prove: +//! (a) calls with no/invalid bearer token are rejected with `Unauthenticated`; +//! (b) calls with the correct bearer token succeed and return the daemon's +//! real, live engine status (not a stub). +//! +//! The bearer token is minted through `SecretManager::mock()` (never the +//! real OS keyring), exactly the way `nunciod`'s boot sequence mints it +//! through `SecretManager::production()` in `main.rs`. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use nuncio_core::{CoreCommand, EngineStatus, EventBus}; +use nuncio_proto::v1::system_client::SystemClient; +use nuncio_proto::v1::GetStatusRequest; +use nuncio_store::vault::{SecretManager, GRPC_TOKEN_ACCOUNT}; +use nunciod::grpc::serve_on_listener; +use std::sync::Arc; +use tokio::net::TcpListener; +use tonic::{Code, Request}; + +/// Binds an ephemeral loopback port and spawns the real gRPC server on it. +/// Because `TcpListener::bind` has already succeeded (the OS is accepting +/// connections on the returned port) before this function returns, callers +/// can dial `addr` immediately with no fixed sleep. +async fn start_server(event_bus: Arc, token: String) -> std::net::SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral loopback port"); + let addr = listener.local_addr().expect("listener has local addr"); + tokio::spawn(async move { + let _ = serve_on_listener(listener, event_bus, token).await; + }); + addr +} + +#[tokio::test] +async fn rejects_calls_without_a_bearer_token() { + let event_bus = Arc::new(EventBus::new()); + let secrets = SecretManager::mock(); + let token = hex::encode( + secrets + .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32) + .expect("token provisioned from mock vault"), + ); + + let addr = start_server(event_bus, token).await; + let mut client = SystemClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let err = client + .get_status(GetStatusRequest {}) + .await + .expect_err("call without any authorization header must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); +} + +#[tokio::test] +async fn rejects_calls_with_an_invalid_bearer_token() { + let event_bus = Arc::new(EventBus::new()); + let secrets = SecretManager::mock(); + let token = hex::encode( + secrets + .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32) + .expect("token provisioned from mock vault"), + ); + + let addr = start_server(event_bus, token).await; + let mut client = SystemClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let mut request = Request::new(GetStatusRequest {}); + request.metadata_mut().insert( + "authorization", + "Bearer definitely-the-wrong-token" + .parse() + .expect("valid ascii metadata value"), + ); + let err = client + .get_status(request) + .await + .expect_err("call with an invalid bearer token must be rejected"); + assert_eq!(err.code(), Code::Unauthenticated); +} + +#[tokio::test] +async fn accepts_correct_bearer_token_and_returns_real_engine_status() { + let event_bus = Arc::new(EventBus::new()); + + // Drive real daemon state via the same `EventBus` the gRPC server reads + // from, proving `GetStatus` reflects live state rather than a stub. + event_bus.process_command(CoreCommand::SyncAll); + assert_eq!(event_bus.current_state().status, EngineStatus::Syncing); + + let secrets = SecretManager::mock(); + let token = hex::encode( + secrets + .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32) + .expect("token provisioned from mock vault"), + ); + + let addr = start_server(event_bus.clone(), token.clone()).await; + let mut client = SystemClient::connect(format!("http://{addr}")) + .await + .expect("client connects"); + + let mut request = Request::new(GetStatusRequest {}); + request.metadata_mut().insert( + "authorization", + format!("Bearer {token}") + .parse() + .expect("valid ascii metadata value"), + ); + let response = client + .get_status(request) + .await + .expect("call with the correct bearer token must succeed") + .into_inner(); + + assert_eq!(response.engine_status, "Syncing"); + assert_eq!(response.version, env!("CARGO_PKG_VERSION")); +}