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 = UnixListener::bind(&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
108fn apply_socket_permissions(path: &str, mode: u32, group: Option<&str>) -> io::Result<()> {
109    if let Some(group) = group {
110        let gid = resolve_group(group)?;
111        std::os::unix::fs::chown(path, None, Some(gid))?;
112    }
113    std::fs::set_permissions(path, std::fs::Permissions::from_mode(mode))
114}
115
116fn resolve_group(group: &str) -> io::Result<u32> {
117    if let Ok(gid) = group.parse::<u32>() {
118        return Ok(gid);
119    }
120    resolve_group_from(group, "/etc/group")
121}
122
123fn resolve_group_from(group: &str, group_file: impl AsRef<Path>) -> io::Result<u32> {
124    let body = std::fs::read_to_string(group_file)?;
125    for line in body.lines() {
126        let mut fields = line.split(':');
127        let Some(name) = fields.next() else {
128            continue;
129        };
130        if name != group {
131            continue;
132        }
133        let _passwd = fields.next();
134        let Some(gid) = fields.next() else {
135            break;
136        };
137        return gid.parse::<u32>().map_err(|err| {
138            io::Error::new(
139                io::ErrorKind::InvalidData,
140                format!("group `{group}` has invalid gid `{gid}`: {err}"),
141            )
142        });
143    }
144    Err(io::Error::new(
145        io::ErrorKind::NotFound,
146        format!("group `{group}` not found"),
147    ))
148}
149
150fn format_socket_mode(mode: u32) -> String {
151    format!("{mode:04o}")
152}
153
154async fn shutdown_signal() {
155    let mut int = signal(SignalKind::interrupt()).ok();
156    let mut quit = signal(SignalKind::quit()).ok();
157    let mut term = signal(SignalKind::terminate()).ok();
158
159    tokio::select! {
160        () = async {
161            if let Some(sig) = int.as_mut() {
162                sig.recv().await;
163            } else {
164                std::future::pending::<()>().await;
165            }
166        } => {}
167        () = async {
168            if let Some(sig) = quit.as_mut() {
169                sig.recv().await;
170            } else {
171                std::future::pending::<()>().await;
172            }
173        } => {}
174        () = async {
175            if let Some(sig) = term.as_mut() {
176                sig.recv().await;
177            } else {
178                std::future::pending::<()>().await;
179            }
180        } => {}
181    }
182}
183
184#[cfg(test)]
185#[allow(clippy::significant_drop_tightening)]
186mod tests {
187    use std::collections::BTreeMap;
188    use std::path::{Path, PathBuf};
189    use std::sync::Arc;
190    use std::time::Duration;
191
192    use async_trait::async_trait;
193    use basil::Client;
194    use basil_proto::KeyType;
195    use basil_proto::broker::v1::SealedRequest;
196    use basil_proto::broker::v1::StatusRequest;
197    use basil_proto::broker::v1::admin_service_client::AdminServiceClient;
198    use basil_proto::broker::v1::invocation_service_client::InvocationServiceClient;
199    use basil_proto::envoy::service::discovery::v3::DiscoveryRequest;
200    use basil_proto::envoy::service::secret::v3::secret_discovery_service_client::SecretDiscoveryServiceClient;
201    use basil_proto::spiffe::X509BundlesRequest;
202    use basil_proto::spiffe::spiffe_workload_api_client::SpiffeWorkloadApiClient;
203    use hyper_util::rt::TokioIo;
204    use tokio::net::UnixStream;
205    use tokio::sync::oneshot;
206    use tonic::Code;
207    use tonic::Request;
208    use tonic::metadata::MetadataValue;
209    use tonic::transport::{Channel, Endpoint, Uri};
210    use tower::service_fn;
211
212    use super::*;
213    use crate::backend::{Backend, BackendError, NewKey};
214    use crate::catalog::load;
215    use crate::manager::BackendManager;
216
217    struct DummyBackend;
218
219    #[async_trait]
220    impl Backend for DummyBackend {
221        fn kind(&self) -> &'static str {
222            "dummy"
223        }
224
225        async fn new_key(&self, key_type: KeyType) -> Result<NewKey, BackendError> {
226            let _ = key_type;
227            Err(BackendError::Unsupported("new_key"))
228        }
229
230        async fn public_key(&self, key_id: &str) -> Result<Vec<u8>, BackendError> {
231            let _ = key_id;
232            Err(BackendError::Unsupported("public_key"))
233        }
234
235        async fn sign(&self, key_id: &str, message: &[u8]) -> Result<Vec<u8>, BackendError> {
236            let _ = (key_id, message);
237            Err(BackendError::Unsupported("sign"))
238        }
239
240        async fn verify(
241            &self,
242            key_id: &str,
243            message: &[u8],
244            signature: &[u8],
245        ) -> Result<bool, BackendError> {
246            let _ = (key_id, message, signature);
247            Err(BackendError::Unsupported("verify"))
248        }
249    }
250
251    fn state() -> Arc<BrokerState> {
252        let catalog = r#"{
253          "schemaVersion": 1,
254          "backends": { "bao": { "kind": "vault", "addr": "https://127.0.0.1:8200" } },
255          "keys": {}
256        }"#;
257        let policy = r#"{
258          "roles": {},
259          "rules": [],
260          "config": { "names": { "users": {}, "groups": {} }, "memberships": {} }
261        }"#;
262        let (catalog, policy, config, warnings) = load(catalog, policy).expect("fixture loads");
263        assert!(warnings.is_empty());
264        let mut backends: BTreeMap<String, Box<dyn Backend>> = BTreeMap::new();
265        backends.insert("bao".to_string(), Box::new(DummyBackend));
266        let manager = BackendManager::new(catalog.clone(), backends).expect("manager builds");
267        Arc::new(BrokerState::new(catalog, policy, config, manager, "dummy"))
268    }
269
270    async fn spawn_server(socket: PathBuf) -> oneshot::Sender<()> {
271        let (tx, rx) = oneshot::channel();
272        let config = ServerConfig {
273            socket_path: socket.to_string_lossy().into_owned(),
274            socket_mode: DEFAULT_SOCKET_MODE,
275            socket_group: None,
276            invocation: InvocationRuntimeConfig::default(),
277        };
278        tokio::spawn(async move {
279            serve_with_shutdown(config, state(), async {
280                let _ = rx.await;
281            })
282            .await
283            .expect("server exits cleanly");
284        });
285        wait_for_socket(&socket).await;
286        tx
287    }
288
289    async fn wait_for_socket(socket: &Path) {
290        for _ in 0..100 {
291            if socket.exists() {
292                return;
293            }
294            tokio::time::sleep(Duration::from_millis(10)).await;
295        }
296        panic!("server socket did not appear: {}", socket.display());
297    }
298
299    fn socket_path(name: &str) -> PathBuf {
300        std::env::temp_dir().join(format!("basil-{name}-{}.sock", uuid::Uuid::new_v4()))
301    }
302
303    #[test]
304    fn group_resolution_accepts_numeric_gid_and_group_file_name() {
305        assert_eq!(resolve_group("4242").expect("numeric gid"), 4242);
306        let path = std::env::temp_dir().join(format!(
307            "basil-group-test-{}-{}",
308            std::process::id(),
309            uuid::Uuid::new_v4()
310        ));
311        std::fs::write(&path, "root:x:0:\nbasil-edge:x:9876:edge\n").expect("write group fixture");
312        assert_eq!(
313            resolve_group_from("basil-edge", &path).expect("named group"),
314            9876
315        );
316        let err = resolve_group_from("missing", &path).expect_err("missing group");
317        assert_eq!(err.kind(), std::io::ErrorKind::NotFound);
318        std::fs::remove_file(path).expect("remove group fixture");
319    }
320
321    #[tokio::test]
322    async fn configured_socket_mode_is_applied_before_serving() {
323        let socket = socket_path("mode");
324        let (tx, rx) = oneshot::channel();
325        let config = ServerConfig {
326            socket_path: socket.to_string_lossy().into_owned(),
327            socket_mode: 0o660,
328            socket_group: None,
329            invocation: InvocationRuntimeConfig::default(),
330        };
331        tokio::spawn(async move {
332            serve_with_shutdown(config, state(), async {
333                let _ = rx.await;
334            })
335            .await
336            .expect("server exits cleanly");
337        });
338        wait_for_socket(&socket).await;
339        let mode = std::fs::metadata(&socket)
340            .expect("stat socket")
341            .permissions()
342            .mode()
343            & 0o777;
344        assert_eq!(mode, 0o660);
345        let _ = tx.send(());
346    }
347
348    #[test]
349    fn active_runtime_sources_have_no_legacy_json_wire_symbols() {
350        let src = Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
351        let banned = [
352            ["json", "_codec"].concat(),
353            ["Client", "Request"].concat(),
354            ["Client", "Response"].concat(),
355            ["core", "::", "handler"].concat(),
356            ["core", "::", "server"].concat(),
357        ];
358        let mut stack = vec![src];
359        while let Some(path) = stack.pop() {
360            for entry in std::fs::read_dir(&path).expect("source directory readable") {
361                let entry = entry.expect("source entry readable");
362                let path = entry.path();
363                if path.is_dir() {
364                    stack.push(path);
365                    continue;
366                }
367                if path.extension().and_then(std::ffi::OsStr::to_str) != Some("rs") {
368                    continue;
369                }
370                let source = std::fs::read_to_string(&path).expect("source file readable");
371                for needle in &banned {
372                    assert!(
373                        !source.contains(needle),
374                        "legacy JSON wire symbol `{needle}` remains in {}",
375                        path.display()
376                    );
377                }
378            }
379        }
380    }
381
382    async fn uds_channel(path: &Path) -> Channel {
383        let path = path.to_path_buf();
384        Endpoint::try_from("http://[::]:50051")
385            .expect("endpoint")
386            .connect_with_connector(service_fn(move |_: Uri| {
387                let path = path.clone();
388                async move { UnixStream::connect(path).await.map(TokioIo::new) }
389            }))
390            .await
391            .expect("connect")
392    }
393
394    #[tokio::test]
395    async fn broker_grpc_serves_status_on_unix_socket() {
396        let socket = socket_path("broker-only");
397        let shutdown = spawn_server(socket.clone()).await;
398        {
399            let mut client = Client::connect(socket.to_str().expect("utf8 path"))
400                .await
401                .expect("broker client connects");
402            let status = client.status().await.expect("status");
403            assert_eq!(status.backend, "dummy");
404            assert_eq!(status.protocol, 1);
405        }
406        let _ = shutdown.send(());
407    }
408
409    #[tokio::test]
410    async fn invocation_service_is_registered_but_disabled_by_default() {
411        let socket = socket_path("invocation-disabled");
412        let shutdown = spawn_server(socket.clone()).await;
413
414        let channel = uds_channel(&socket).await;
415        let mut invocation = InvocationServiceClient::new(channel);
416        let status = invocation
417            .invoke(SealedRequest::default())
418            .await
419            .expect_err("invocation is disabled by default");
420        assert_eq!(status.code(), Code::FailedPrecondition);
421
422        let mut broker = Client::connect(socket.to_str().expect("utf8 path"))
423            .await
424            .expect("broker client still connects");
425        let status = broker
426            .status()
427            .await
428            .expect("typed status remains available");
429        assert_eq!(status.protocol, 1);
430
431        let _ = shutdown.send(());
432    }
433    #[tokio::test]
434    async fn broker_and_spiffe_services_share_one_unix_socket() {
435        let socket = socket_path("broker-spiffe");
436        let shutdown = spawn_server(socket.clone()).await;
437
438        {
439            let mut broker = Client::connect(socket.to_str().expect("utf8 path"))
440                .await
441                .expect("broker client connects");
442            let status = broker.status().await.expect("status");
443            assert_eq!(status.backend, "dummy");
444            assert_eq!(status.protocol, 1);
445        }
446
447        {
448            let channel = uds_channel(&socket).await;
449            let mut broker = AdminServiceClient::new(channel.clone());
450            let mut request = Request::new(StatusRequest {});
451            request
452                .metadata_mut()
453                .insert("workload.spiffe.io", "true".parse().expect("metadata"));
454            let status = broker
455                .status(request)
456                .await
457                .expect("broker RPC ignores Workload API metadata")
458                .into_inner();
459            assert_eq!(status.backend, "dummy");
460            assert_eq!(status.protocol, 1);
461        }
462
463        {
464            let channel = uds_channel(&socket).await;
465            let mut spiffe = SpiffeWorkloadApiClient::new(channel);
466            let status = spiffe
467                .fetch_x509_bundles(X509BundlesRequest {})
468                .await
469                .expect_err("registered SPIFFE service rejects missing workload header");
470            assert_eq!(status.code(), Code::InvalidArgument);
471            assert_eq!(
472                status.message(),
473                "SPIFFE Workload API requests require workload.spiffe.io=true"
474            );
475        }
476
477        {
478            let channel = uds_channel(&socket).await;
479            let mut spiffe = SpiffeWorkloadApiClient::new(channel);
480            let mut request = Request::new(X509BundlesRequest {});
481            request
482                .metadata_mut()
483                .append("workload.spiffe.io", "true".parse().expect("metadata"));
484            request
485                .metadata_mut()
486                .append("workload.spiffe.io", "false".parse().expect("metadata"));
487            let status = spiffe
488                .fetch_x509_bundles(request)
489                .await
490                .expect_err("duplicate Workload API metadata is fail-closed");
491            assert_eq!(status.code(), Code::InvalidArgument);
492        }
493
494        {
495            let channel = uds_channel(&socket).await;
496            let mut spiffe = SpiffeWorkloadApiClient::new(channel);
497            let mut request = Request::new(X509BundlesRequest {});
498            request
499                .metadata_mut()
500                .insert_bin("workload.spiffe.io-bin", MetadataValue::from_bytes(b"true"));
501            let status = spiffe
502                .fetch_x509_bundles(request)
503                .await
504                .expect_err("binary Workload API metadata is fail-closed");
505            assert_eq!(status.code(), Code::InvalidArgument);
506        }
507
508        {
509            let channel = uds_channel(&socket).await;
510            let mut sds = SecretDiscoveryServiceClient::new(channel);
511            let status = sds
512                .fetch_secrets(DiscoveryRequest {
513                    version_info: String::new(),
514                    node: None,
515                    resource_names: vec!["default".to_string()],
516                    type_url: crate::sds::SECRET_TYPE_URL.to_string(),
517                    response_nonce: String::new(),
518                    error_detail: None,
519                })
520                .await
521                .expect_err("registered SDS service has no configured resources");
522            assert_eq!(status.code(), Code::NotFound);
523        }
524        let _ = shutdown.send(());
525    }
526}
basil_core/transport/grpc_server.rs

basil_core/transport/
grpc_server.rs
