basil_core/transport/
grpc_server.rs

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

basil_core/transport/
grpc_server.rs
