basil_core/transport/
grpc_server.rs

1// SPDX-FileCopyrightText: 2026 OpenBasil Contributors
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! tonic server wiring for the broker gRPC API.
6
7use std::future::Future;
8use std::io;
9use std::os::unix::fs::PermissionsExt as _;
10use std::path::Path;
11use std::sync::Arc;
12
13use basil_proto::broker::v1::admin_service_server::AdminServiceServer;
14use basil_proto::broker::v1::aead_service_server::AeadServiceServer;
15use basil_proto::broker::v1::invocation_service_server::InvocationServiceServer;
16use basil_proto::broker::v1::minting_service_server::MintingServiceServer;
17use basil_proto::broker::v1::nats_service_server::NatsServiceServer;
18use basil_proto::broker::v1::secret_service_server::SecretServiceServer;
19use basil_proto::broker::v1::signing_service_server::SigningServiceServer;
20use basil_proto::envoy::service::secret::v3::secret_discovery_service_server::SecretDiscoveryServiceServer;
21use basil_proto::spiffe::spiffe_workload_api_server::SpiffeWorkloadApiServer;
22use tokio::net::UnixListener;
23use tokio::signal::unix::{SignalKind, signal};
24use tokio_stream::wrappers::UnixListenerStream;
25use tonic::transport::Server;
26use tracing::{info, warn};
27
28use crate::grpc::BrokerGrpc;
29use crate::sds::EnvoySdsGrpc;
30use crate::service::broker::InvocationRuntimeConfig;
31use crate::spiffe::SpiffeWorkloadGrpc;
32use crate::state::BrokerState;
33
34/// Default Unix socket mode: owner read/write only.
35pub const DEFAULT_SOCKET_MODE: u32 = 0o600;
36
37/// Runtime configuration for the gRPC listener.
38#[derive(Debug, Clone)]
39pub struct ServerConfig {
40    /// Path to bind the listening Unix socket at.
41    pub socket_path: String,
42    /// File mode to apply to the listening Unix socket after bind.
43    pub socket_mode: u32,
44    /// Group name or numeric gid to apply to the listening Unix socket.
45    pub socket_group: Option<String>,
46    /// Runtime settings for the sealed invocation service.
47    pub invocation: InvocationRuntimeConfig,
48}
49
50/// Bind a Unix socket and serve the broker gRPC services until shutdown.
51///
52/// Registers all broker services, the SPIFFE Workload API, and Envoy SDS on one
53/// tonic server.
54pub async fn run(config: ServerConfig, state: Arc<BrokerState>) -> std::io::Result<()> {
55    serve_with_shutdown(config, state, shutdown_signal()).await
56}
57
58async fn serve_with_shutdown(
59    config: ServerConfig,
60    state: Arc<BrokerState>,
61    shutdown: impl Future<Output = ()>,
62) -> std::io::Result<()> {
63    let path = config.socket_path;
64    if Path::new(&path).exists() {
65        std::fs::remove_file(&path)?;
66        warn!(%path, "removed stale socket");
67    }
68
69    let listener = bind_restricted(&path)?;
70    apply_socket_permissions(&path, config.socket_mode, config.socket_group.as_deref())?;
71
72    info!(
73        %path,
74        mode = %format_socket_mode(config.socket_mode),
75        group = ?config.socket_group,
76        backend = state.backend_label(),
77        "basil gRPC agent listening"
78    );
79    let incoming = UnixListenerStream::new(listener);
80    let broker = BrokerGrpc::new_with_invocation_config(state.clone(), config.invocation);
81
82    let server = Server::builder()
83        .add_service(InvocationServiceServer::new(broker.clone()))
84        .add_service(SigningServiceServer::new(broker.clone()))
85        .add_service(AeadServiceServer::new(broker.clone()))
86        .add_service(SecretServiceServer::new(broker.clone()))
87        .add_service(MintingServiceServer::new(broker.clone()))
88        .add_service(NatsServiceServer::new(broker.clone()))
89        .add_service(AdminServiceServer::new(broker));
90    let server = server
91        .add_service(SpiffeWorkloadApiServer::new(SpiffeWorkloadGrpc::new(
92            state.clone(),
93        )))
94        .add_service(SecretDiscoveryServiceServer::new(EnvoySdsGrpc::new(state)));
95    let result = server
96        .serve_with_incoming_shutdown(incoming, shutdown)
97        .await;
98
99    if let Err(e) = std::fs::remove_file(&path)
100        && e.kind() != std::io::ErrorKind::NotFound
101    {
102        warn!(?e, %path, "could not remove socket on shutdown");
103    }
104
105    result.map_err(std::io::Error::other)
106}
107
108/// Bind the listening Unix socket with the process umask tightened to `0o177`,
109/// so the socket node is created owner-only no matter how loose the inherited
110/// umask is. The listen backlog is live from `bind`, so the mode must be
111/// restrictive *at creation*; the later [`apply_socket_permissions`] can only
112/// widen it to the configured mode/group (never leaves a permissive window).
113fn bind_restricted(path: &str) -> io::Result<UnixListener> {
114    let inherited = rustix::process::umask(rustix::fs::Mode::from_raw_mode(0o177));
115    let listener = UnixListener::bind(path);
116    rustix::process::umask(inherited);
117    listener
118}
119
120fn apply_socket_permissions(path: &str, mode: u32, group: Option<&str>) -> io::Result<()> {
121    if let Some(group) = group {
122        let gid = resolve_group(group)?;
123        std::os::unix::fs::chown(path, None, Some(gid))?;
124    }
125    std::fs::set_permissions(path, std::fs::Permissions::from_mode(mode))
126}
127
128fn resolve_group(group: &str) -> io::Result<u32> {
129    if let Ok(gid) = group.parse::<u32>() {
130        return Ok(gid);
131    }
132    resolve_group_from(group, "/etc/group")
133}
134
135fn resolve_group_from(group: &str, group_file: impl AsRef<Path>) -> io::Result<u32> {
136    let body = std::fs::read_to_string(group_file)?;
137    for line in body.lines() {
138        let mut fields = line.split(':');
139        let Some(name) = fields.next() else {
140            continue;
141        };
142        if name != group {
143            continue;
144        }
145        let _passwd = fields.next();
146        let Some(gid) = fields.next() else {
147            break;
148        };
149        return gid.parse::<u32>().map_err(|err| {
150            io::Error::new(
151                io::ErrorKind::InvalidData,
152                format!("group `{group}` has invalid gid `{gid}`: {err}"),
153            )
154        });
155    }
156    Err(io::Error::new(
157        io::ErrorKind::NotFound,
158        format!("group `{group}` not found"),
159    ))
160}
161
162fn format_socket_mode(mode: u32) -> String {
163    format!("{mode:04o}")
164}
165
166async fn shutdown_signal() {
167    let mut int = signal(SignalKind::interrupt()).ok();
168    let mut quit = signal(SignalKind::quit()).ok();
169    let mut term = signal(SignalKind::terminate()).ok();
170
171    tokio::select! {
172        () = async {
173            if let Some(sig) = int.as_mut() {
174                sig.recv().await;
175            } else {
176                std::future::pending::<()>().await;
177            }
178        } => {}
179        () = async {
180            if let Some(sig) = quit.as_mut() {
181                sig.recv().await;
182            } else {
183                std::future::pending::<()>().await;
184            }
185        } => {}
186        () = async {
187            if let Some(sig) = term.as_mut() {
188                sig.recv().await;
189            } else {
190                std::future::pending::<()>().await;
191            }
192        } => {}
193    }
194}
195
196#[cfg(test)]
197#[allow(clippy::significant_drop_tightening)]
198mod tests {
199    use std::collections::BTreeMap;
200    use std::path::{Path, PathBuf};
201    use std::sync::Arc;
202    use std::time::Duration;
203
204    use async_trait::async_trait;
205    use basil::Client;
206    use basil_proto::KeyType;
207    use basil_proto::broker::v1::SealedRequest;
208    use basil_proto::broker::v1::StatusRequest;
209    use basil_proto::broker::v1::admin_service_client::AdminServiceClient;
210    use basil_proto::broker::v1::invocation_service_client::InvocationServiceClient;
211    use basil_proto::envoy::service::discovery::v3::DiscoveryRequest;
212    use basil_proto::envoy::service::secret::v3::secret_discovery_service_client::SecretDiscoveryServiceClient;
213    use basil_proto::spiffe::X509BundlesRequest;
214    use basil_proto::spiffe::spiffe_workload_api_client::SpiffeWorkloadApiClient;
215    use hyper_util::rt::TokioIo;
216    use tokio::net::UnixStream;
217    use tokio::sync::oneshot;
218    use tonic::Code;
219    use tonic::Request;
220    use tonic::metadata::MetadataValue;
221    use tonic::transport::{Channel, Endpoint, Uri};
222    use tower::service_fn;
223
224    use super::*;
225    use crate::backend::{Backend, BackendError, NewKey};
226    use crate::catalog::load;
227    use crate::manager::BackendManager;
228
229    struct DummyBackend;
230
231    #[async_trait]
232    impl Backend for DummyBackend {
233        fn kind(&self) -> &'static str {
234            "dummy"
235        }
236
237        async fn new_key(&self, key_type: KeyType) -> Result<NewKey, BackendError> {
238            let _ = key_type;
239            Err(BackendError::Unsupported("new_key"))
240        }
241
242        async fn public_key(&self, key_id: &str) -> Result<Vec<u8>, BackendError> {
243            let _ = key_id;
244            Err(BackendError::Unsupported("public_key"))
245        }
246
247        async fn sign(&self, key_id: &str, message: &[u8]) -> Result<Vec<u8>, BackendError> {
248            let _ = (key_id, message);
249            Err(BackendError::Unsupported("sign"))
250        }
251
252        async fn verify(
253            &self,
254            key_id: &str,
255            message: &[u8],
256            signature: &[u8],
257        ) -> Result<bool, BackendError> {
258            let _ = (key_id, message, signature);
259            Err(BackendError::Unsupported("verify"))
260        }
261    }
262
263    fn state() -> Arc<BrokerState> {
264        let catalog = r#"{
265          "schemaVersion": 1,
266          "backends": { "bao": { "kind": "vault", "addr": "https://127.0.0.1:8200" } },
267          "keys": {}
268        }"#;
269        // These tests exercise transport wiring over a real UDS, so the peer uid
270        // the kernel reports is this test process's own; register it as a policy
271        // subject so the `status` canary RPC (which requires a resolved subject)
272        // answers.
273        let uid = rustix::process::getuid().as_raw();
274        let policy = format!(
275            r#"{{
276              "schemaVersion": 2,
277              "subjects": {{ "test.peer": {{ "allOf": [ {{ "kind": "unix", "uid": {uid} }} ] }} }},
278              "roles": {{}},
279              "rules": [],
280              "config": {{
281                "names": {{ "users": {{ "{uid}": "test-peer" }}, "groups": {{}} }},
282                "memberships": {{ "{uid}": [{uid}] }}
283              }}
284            }}"#
285        );
286        let (catalog, policy, config, warnings) = load(catalog, &policy).expect("fixture loads");
287        assert!(warnings.is_empty());
288        let mut backends: BTreeMap<String, Box<dyn Backend>> = BTreeMap::new();
289        backends.insert("bao".to_string(), Box::new(DummyBackend));
290        let manager = BackendManager::new(catalog.clone(), backends).expect("manager builds");
291        Arc::new(BrokerState::new(catalog, policy, config, manager, "dummy"))
292    }
293
294    async fn spawn_server(socket: PathBuf) -> oneshot::Sender<()> {
295        let (tx, rx) = oneshot::channel();
296        let config = ServerConfig {
297            socket_path: socket.to_string_lossy().into_owned(),
298            socket_mode: DEFAULT_SOCKET_MODE,
299            socket_group: None,
300            invocation: InvocationRuntimeConfig::default(),
301        };
302        tokio::spawn(async move {
303            serve_with_shutdown(config, state(), async {
304                let _ = rx.await;
305            })
306            .await
307            .expect("server exits cleanly");
308        });
309        wait_for_socket(&socket).await;
310        tx
311    }
312
313    async fn wait_for_socket(socket: &Path) {
314        for _ in 0..100 {
315            if socket.exists() {
316                return;
317            }
318            tokio::time::sleep(Duration::from_millis(10)).await;
319        }
320        panic!("server socket did not appear: {}", socket.display());
321    }
322
323    fn socket_path(name: &str) -> PathBuf {
324        // Unix-domain socket paths must fit in sun_path: 104 bytes on macOS, 108
325        // on Linux. macOS's std::env::temp_dir() (/var/folders/...) is long enough
326        // that "basil-{name}-{uuid}.sock" overflowed the macOS limit; anchor at the
327        // short, always-writable /tmp so the full path stays well under it.
328        PathBuf::from("/tmp").join(format!("basil-{name}-{}.sock", uuid::Uuid::new_v4()))
329    }
330
331    #[tokio::test]
332    async fn socket_is_owner_only_at_bind_even_under_a_loose_umask() {
333        // Loosen the process umask: without the tightened bind the socket node
334        // would be group/world-accessible for the instant before the explicit
335        // chmod, with the listen backlog already live.
336        let inherited = rustix::process::umask(rustix::fs::Mode::empty());
337        let socket = socket_path("umask");
338        let listener = bind_restricted(&socket.to_string_lossy()).expect("binds");
339        rustix::process::umask(inherited);
340
341        let mode = std::fs::metadata(&socket)
342            .expect("socket metadata")
343            .permissions()
344            .mode()
345            & 0o777;
346        assert_eq!(mode, 0o600, "socket must be owner-only at bind");
347        drop(listener);
348        let _ = std::fs::remove_file(&socket);
349    }
350
351    #[test]
352    fn group_resolution_accepts_numeric_gid_and_group_file_name() {
353        assert_eq!(resolve_group("4242").expect("numeric gid"), 4242);
354        let path = std::env::temp_dir().join(format!(
355            "basil-group-test-{}-{}",
356            std::process::id(),
357            uuid::Uuid::new_v4()
358        ));
359        std::fs::write(&path, "root:x:0:\nbasil-edge:x:9876:edge\n").expect("write group fixture");
360        assert_eq!(
361            resolve_group_from("basil-edge", &path).expect("named group"),
362            9876
363        );
364        let err = resolve_group_from("missing", &path).expect_err("missing group");
365        assert_eq!(err.kind(), std::io::ErrorKind::NotFound);
366        std::fs::remove_file(path).expect("remove group fixture");
367    }
368
369    #[tokio::test]
370    async fn configured_socket_mode_is_applied_before_serving() {
371        let socket = socket_path("mode");
372        let (tx, rx) = oneshot::channel();
373        let config = ServerConfig {
374            socket_path: socket.to_string_lossy().into_owned(),
375            socket_mode: 0o660,
376            socket_group: None,
377            invocation: InvocationRuntimeConfig::default(),
378        };
379        tokio::spawn(async move {
380            serve_with_shutdown(config, state(), async {
381                let _ = rx.await;
382            })
383            .await
384            .expect("server exits cleanly");
385        });
386        wait_for_socket(&socket).await;
387        let mode = std::fs::metadata(&socket)
388            .expect("stat socket")
389            .permissions()
390            .mode()
391            & 0o777;
392        assert_eq!(mode, 0o660);
393        let _ = tx.send(());
394    }
395
396    #[test]
397    fn active_runtime_sources_have_no_legacy_json_wire_symbols() {
398        let src = Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
399        let banned = [
400            ["json", "_codec"].concat(),
401            ["Client", "Request"].concat(),
402            ["Client", "Response"].concat(),
403            ["core", "::", "handler"].concat(),
404            ["core", "::", "server"].concat(),
405        ];
406        let mut stack = vec![src];
407        while let Some(path) = stack.pop() {
408            for entry in std::fs::read_dir(&path).expect("source directory readable") {
409                let entry = entry.expect("source entry readable");
410                let path = entry.path();
411                if path.is_dir() {
412                    stack.push(path);
413                    continue;
414                }
415                if path.extension().and_then(std::ffi::OsStr::to_str) != Some("rs") {
416                    continue;
417                }
418                let source = std::fs::read_to_string(&path).expect("source file readable");
419                for needle in &banned {
420                    assert!(
421                        !source.contains(needle),
422                        "legacy JSON wire symbol `{needle}` remains in {}",
423                        path.display()
424                    );
425                }
426            }
427        }
428    }
429
430    async fn uds_channel(path: &Path) -> Channel {
431        let path = path.to_path_buf();
432        Endpoint::try_from("http://[::]:50051")
433            .expect("endpoint")
434            .connect_with_connector(service_fn(move |_: Uri| {
435                let path = path.clone();
436                async move { UnixStream::connect(path).await.map(TokioIo::new) }
437            }))
438            .await
439            .expect("connect")
440    }
441
442    #[tokio::test]
443    async fn broker_grpc_serves_status_on_unix_socket() {
444        let socket = socket_path("broker-only");
445        let shutdown = spawn_server(socket.clone()).await;
446        {
447            let mut client = Client::connect(socket.to_str().expect("utf8 path"))
448                .await
449                .expect("broker client connects");
450            let status = client.status().await.expect("status");
451            assert_eq!(status.backend, "dummy");
452            assert_eq!(status.protocol, 1);
453        }
454        let _ = shutdown.send(());
455    }
456
457    #[tokio::test]
458    async fn invocation_service_is_registered_but_disabled_by_default() {
459        let socket = socket_path("invocation-disabled");
460        let shutdown = spawn_server(socket.clone()).await;
461
462        let channel = uds_channel(&socket).await;
463        let mut invocation = InvocationServiceClient::new(channel);
464        let status = invocation
465            .invoke(SealedRequest::default())
466            .await
467            .expect_err("invocation is disabled by default");
468        assert_eq!(status.code(), Code::FailedPrecondition);
469
470        let mut broker = Client::connect(socket.to_str().expect("utf8 path"))
471            .await
472            .expect("broker client still connects");
473        let status = broker
474            .status()
475            .await
476            .expect("typed status remains available");
477        assert_eq!(status.protocol, 1);
478
479        let _ = shutdown.send(());
480    }
481    #[tokio::test]
482    async fn broker_and_spiffe_services_share_one_unix_socket() {
483        let socket = socket_path("broker-spiffe");
484        let shutdown = spawn_server(socket.clone()).await;
485
486        {
487            let mut broker = Client::connect(socket.to_str().expect("utf8 path"))
488                .await
489                .expect("broker client connects");
490            let status = broker.status().await.expect("status");
491            assert_eq!(status.backend, "dummy");
492            assert_eq!(status.protocol, 1);
493        }
494
495        {
496            let channel = uds_channel(&socket).await;
497            let mut broker = AdminServiceClient::new(channel.clone());
498            let mut request = Request::new(StatusRequest {});
499            request
500                .metadata_mut()
501                .insert("workload.spiffe.io", "true".parse().expect("metadata"));
502            let status = broker
503                .status(request)
504                .await
505                .expect("broker RPC ignores Workload API metadata")
506                .into_inner();
507            assert_eq!(status.backend, "dummy");
508            assert_eq!(status.protocol, 1);
509        }
510
511        {
512            let channel = uds_channel(&socket).await;
513            let mut spiffe = SpiffeWorkloadApiClient::new(channel);
514            let status = spiffe
515                .fetch_x509_bundles(X509BundlesRequest {})
516                .await
517                .expect_err("registered SPIFFE service rejects missing workload header");
518            assert_eq!(status.code(), Code::InvalidArgument);
519            assert_eq!(
520                status.message(),
521                "SPIFFE Workload API requests require workload.spiffe.io=true"
522            );
523        }
524
525        {
526            let channel = uds_channel(&socket).await;
527            let mut spiffe = SpiffeWorkloadApiClient::new(channel);
528            let mut request = Request::new(X509BundlesRequest {});
529            request
530                .metadata_mut()
531                .append("workload.spiffe.io", "true".parse().expect("metadata"));
532            request
533                .metadata_mut()
534                .append("workload.spiffe.io", "false".parse().expect("metadata"));
535            let status = spiffe
536                .fetch_x509_bundles(request)
537                .await
538                .expect_err("duplicate Workload API metadata is fail-closed");
539            assert_eq!(status.code(), Code::InvalidArgument);
540        }
541
542        {
543            let channel = uds_channel(&socket).await;
544            let mut spiffe = SpiffeWorkloadApiClient::new(channel);
545            let mut request = Request::new(X509BundlesRequest {});
546            request
547                .metadata_mut()
548                .insert_bin("workload.spiffe.io-bin", MetadataValue::from_bytes(b"true"));
549            let status = spiffe
550                .fetch_x509_bundles(request)
551                .await
552                .expect_err("binary Workload API metadata is fail-closed");
553            assert_eq!(status.code(), Code::InvalidArgument);
554        }
555
556        {
557            let channel = uds_channel(&socket).await;
558            let mut sds = SecretDiscoveryServiceClient::new(channel);
559            let status = sds
560                .fetch_secrets(DiscoveryRequest {
561                    version_info: String::new(),
562                    node: None,
563                    resource_names: vec!["default".to_string()],
564                    type_url: crate::sds::SECRET_TYPE_URL.to_string(),
565                    response_nonce: String::new(),
566                    error_detail: None,
567                })
568                .await
569                .expect_err("registered SDS service has no configured resources");
570            assert_eq!(status.code(), Code::NotFound);
571        }
572        let _ = shutdown.send(());
573    }
574}
basil_core/transport/grpc_server.rs

basil_core/transport/
grpc_server.rs
