service_broker/
lib.rs

1// Copyright 2024 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use anyhow::{Context, Result, bail, format_err};
6use async_utils::async_once::Once;
7use cm_types::Name;
8use fidl::AsHandleRef;
9use fidl::endpoints::ServerEnd;
10use fidl_fuchsia_data as fdata;
11use fidl_fuchsia_io as fio;
12use fidl_fuchsia_process as fprocess;
13use fidl_fuchsia_process_lifecycle as fpl;
14use fuchsia_component::directory::AsRefDirectory;
15use fuchsia_component::server::{ServiceFs, ServiceObj, ServiceObjTrait};
16use fuchsia_fs::directory::{WatchEvent, Watcher};
17use futures::prelude::*;
18use std::borrow::Cow;
19use std::path::Path;
20use std::sync::Arc;
21
22/// Safely extracts a string from a VFS event without allocating
23/// unless ownership is explicitly required.
24fn extract_event_filename<'a>(path: &'a Path) -> Option<Cow<'a, str>> {
25    let s = path.to_str()?;
26    if s == "." {
27        // Return a borrowed reference, no allocation
28        Some(Cow::Borrowed("."))
29    } else {
30        // We can still borrow here, deferring .to_owned() until the caller
31        // proves they need it.
32        Some(Cow::Borrowed(s))
33    }
34}
35
36async fn wait_for_first_instance(svc: &fio::DirectoryProxy) -> Result<String> {
37    const INPUT_SERVICE: &str = "input";
38    let (service_dir, request) = fidl::endpoints::create_proxy::<fio::DirectoryMarker>();
39    svc.as_ref_directory().open(
40        INPUT_SERVICE,
41        fio::Flags::PROTOCOL_DIRECTORY | fio::PERM_READABLE,
42        request.into(),
43    )?;
44    let watcher = Watcher::new(&service_dir).await.context("failed to create watcher")?;
45
46    let mut stream =
47        watcher.map(|result| result.context("failed to get watcher event")).try_filter_map(|msg| {
48            futures::future::ok(match msg.event {
49                WatchEvent::EXISTING | WatchEvent::ADD_FILE => {
50                    let filename = extract_event_filename(msg.filename.as_path())
51                        .expect("filename must be valid utf8");
52
53                    if filename.as_ref() == "." { None } else { Some(filename.into_owned()) }
54                }
55                _ => None,
56            })
57        });
58
59    let first = stream.try_next().await?.ok_or_else(|| {
60        format_err!("Watcher stream closed unexpectedly before finding an instance")
61    })?;
62
63    Ok(format!("{INPUT_SERVICE}/{first}"))
64}
65
66async fn connect_request(
67    svc: &fio::DirectoryProxy,
68    request: zx::Channel,
69    protocol_name: &Name,
70    instance_dir: &str,
71) {
72    let target_path = format!("{instance_dir}/{}", protocol_name.as_str());
73
74    if let Err(e) = svc.as_ref_directory().open(&target_path, fio::Flags::PROTOCOL_SERVICE, request)
75    {
76        log::error!("[service-broker] Failed to forward connection to {target_path}: {e}");
77    }
78}
79
80async fn first_instance_to_protocol<'a>(
81    svc: fio::DirectoryProxy,
82    fs: &mut ServiceFs<ServiceObj<'a, ()>>,
83    protocol_name: Name,
84    scope: &'a fuchsia_async::Scope,
85) -> Result<()> {
86    let cached_instance: Arc<Once<String>> = Arc::new(Once::new());
87    let svc_arc = Arc::new(svc);
88
89    fs.dir("svc").add_service_at("output", move |request: zx::Channel| {
90        let svc = Arc::clone(&svc_arc);
91        let protocol_name = protocol_name.clone();
92        let cached_instance = Arc::clone(&cached_instance);
93
94        scope.spawn(async move {
95            // Safely initializes the path once. All subsequent connection requests
96            // will instantly resolve the string without hitting the filesystem.
97            let init_future = async || wait_for_first_instance(&svc).await;
98
99            match cached_instance.get_or_try_init(init_future).await {
100                Ok(instance_dir) => {
101                    connect_request(&svc, request, &protocol_name, instance_dir).await;
102                }
103                Err(e) => {
104                    log::error!(
105                        "[service-broker] Failed to resolve first instance: {e}, {protocol_name}"
106                    );
107                }
108            }
109        });
110
111        Some(())
112    });
113
114    Ok(())
115}
116
117async fn first_instance_to_default<T: ServiceObjTrait>(
118    svc: fio::DirectoryProxy,
119    fs: &mut ServiceFs<T>,
120) -> Result<()> {
121    // TODO(surajmalhotra): Do this wait every time we get a connection request to handle cases
122    // where the instance goes away and comes back.
123    let instance_dir_path = wait_for_first_instance(&svc).await?;
124    let (instance_dir, request) = fidl::endpoints::create_proxy::<fio::DirectoryMarker>();
125    svc.as_ref_directory().open(
126        &instance_dir_path,
127        fio::Flags::PROTOCOL_DIRECTORY | fio::PERM_READABLE,
128        request.into(),
129    )?;
130
131    fs.dir("svc").dir("output").add_remote("default", instance_dir);
132    Ok(())
133}
134
135async fn filter_and_rename<T: ServiceObjTrait>(
136    _svc: fio::DirectoryProxy,
137    _fs: &mut ServiceFs<T>,
138    _filter: &Vec<String>,
139    _rename: &Vec<String>,
140) -> Result<()> {
141    bail!("filter_and_rename policy is not yet implemented");
142    // Add a bunch of directories which forward requests?
143}
144
145fn get_value<'a>(dict: &'a fdata::Dictionary, key: &str) -> Option<&'a fdata::DictionaryValue> {
146    match &dict.entries {
147        Some(entries) => {
148            for entry in entries {
149                if entry.key == key {
150                    return entry.value.as_ref().map(|val| &**val);
151                }
152            }
153            None
154        }
155        _ => None,
156    }
157}
158
159fn get_program_string<'a>(program: &'a fdata::Dictionary, key: &str) -> Result<&'a str> {
160    if let Some(fdata::DictionaryValue::Str(value)) = get_value(program, key) {
161        Ok(value)
162    } else {
163        Err(format_err!("{key} not found in program or is not a string"))
164    }
165}
166
167fn get_program_strvec<'a>(
168    program: &'a fdata::Dictionary,
169    key: &str,
170) -> Result<Option<&'a Vec<String>>> {
171    match get_value(program, key) {
172        Some(args_value) => match args_value {
173            fdata::DictionaryValue::StrVec(vec) => Ok(Some(vec)),
174            _ => Err(format_err!(
175                "Expected {key} in program to be vector of strings, found something else"
176            )),
177        },
178        None => Ok(None),
179    }
180}
181
182pub async fn main(
183    ns_entries: Vec<fprocess::NameInfo>,
184    directory_request: ServerEnd<fio::DirectoryMarker>,
185    lifecycle: ServerEnd<fpl::LifecycleMarker>,
186    program: Option<fdata::Dictionary>,
187) -> Result<()> {
188    drop(lifecycle);
189    if directory_request.as_handle_ref().is_invalid() {
190        bail!("No valid handle found for outgoing directory");
191    }
192    let Some(svc) = ns_entries.into_iter().find(|e| e.path == "/svc") else {
193        bail!("No /svc in namespace");
194    };
195    let Some(program) = program else {
196        bail!("No program section provided");
197    };
198    let scope = fuchsia_async::Scope::new();
199    let svc = svc.directory.into_proxy();
200    let mut fs = ServiceFs::new();
201    match get_program_string(&program, "policy")? {
202        "first_instance_to_protocol" => {
203            let protocol_name_str = get_program_string(&program, "protocol_name")?;
204
205            let protocol_name = Name::new(protocol_name_str).map_err(|e| {
206                format_err!("Invalid protocol_name '{protocol_name_str}' in program dict: {e}")
207            })?;
208
209            first_instance_to_protocol(svc, &mut fs, protocol_name, &scope).await
210        }
211        "first_instance_to_default" => first_instance_to_default(svc, &mut fs).await,
212        "filter_and_rename" => {
213            let empty = vec![];
214            let filter = get_program_strvec(&program, "filter")?.unwrap_or(&empty);
215            let rename = get_program_strvec(&program, "rename")?.unwrap_or(&empty);
216            filter_and_rename(svc, &mut fs, filter, rename).await
217        }
218        policy => Err(format_err!("Unsupported policy specified: {policy}")),
219    }?;
220
221    log::debug!("[service-broker] Initialized.");
222
223    fs.serve_connection(directory_request).context("failed to serve outgoing namespace")?;
224    fs.collect::<()>().await;
225    Ok(())
226}
227
228#[cfg(test)]
229mod tests {
230    use super::*;
231    use fidl::endpoints::{Proxy, create_endpoints, create_proxy};
232    use fidl_fuchsia_data as fdata;
233    use fuchsia_async as fasync;
234    use futures::StreamExt;
235
236    fn make_program_dict(entries: Vec<(&str, fdata::DictionaryValue)>) -> fdata::Dictionary {
237        let entries = entries
238            .into_iter()
239            .map(|(k, v)| fdata::DictionaryEntry { key: k.to_string(), value: Some(Box::new(v)) })
240            .collect();
241        fdata::Dictionary { entries: Some(entries), ..Default::default() }
242    }
243
244    #[test]
245    fn test_get_program_string() {
246        let dict = make_program_dict(vec![(
247            "policy",
248            fdata::DictionaryValue::Str("first_instance_to_protocol".to_string()),
249        )]);
250
251        assert_eq!(get_program_string(&dict, "policy").unwrap(), "first_instance_to_protocol");
252
253        let err = get_program_string(&dict, "missing_key").unwrap_err();
254        assert_eq!(err.to_string(), "missing_key not found in program or is not a string");
255    }
256
257    #[test]
258    fn test_get_program_strvec() {
259        let dict = make_program_dict(vec![(
260            "filter",
261            fdata::DictionaryValue::StrVec(vec!["fuchsia.foo.Bar".to_string()]),
262        )]);
263
264        let vec = get_program_strvec(&dict, "filter").unwrap().unwrap();
265        assert_eq!(vec.len(), 1);
266        assert_eq!(vec[0], "fuchsia.foo.Bar");
267
268        assert!(get_program_strvec(&dict, "rename").unwrap().is_none());
269    }
270
271    #[fuchsia::test]
272    async fn test_filter_and_rename_graceful_failure() {
273        let (dir_proxy, _server_end) = fidl::endpoints::create_proxy::<fio::DirectoryMarker>();
274        let mut fs = ServiceFs::<ServiceObj<'_, ()>>::new();
275
276        let result = filter_and_rename(dir_proxy, &mut fs, &vec![], &vec![]).await;
277        assert!(result.is_err());
278        assert_eq!(
279            result.unwrap_err().to_string(),
280            "filter_and_rename policy is not yet implemented"
281        );
282    }
283
284    #[fuchsia::test]
285    async fn test_broker_caching_and_routing_end_to_end() {
286        let (svc_dir, svc_server_end) = create_proxy::<fio::DirectoryMarker>();
287        let mut fake_svc_fs = ServiceFs::new();
288
289        fake_svc_fs.dir("input").dir("instance_123").add_service_at(
290            "my_protocol",
291            |req: zx::Channel| {
292                let _ = req.write(&[1], &mut []);
293                Some(())
294            },
295        );
296
297        fasync::Task::spawn(async move {
298            fake_svc_fs.serve_connection(svc_server_end).unwrap();
299            fake_svc_fs.collect::<()>().await;
300        })
301        .detach();
302
303        let ns_entries = vec![fprocess::NameInfo {
304            path: "/svc".to_string(),
305            directory: svc_dir.into_channel().unwrap().into_zx_channel().into(),
306        }];
307
308        let (out_dir, out_server_end) = create_proxy::<fio::DirectoryMarker>();
309        let (_, lifecycle_server_end) = create_endpoints::<fpl::LifecycleMarker>();
310
311        let program_dict = make_program_dict(vec![
312            ("policy", fdata::DictionaryValue::Str("first_instance_to_protocol".to_string())),
313            ("protocol_name", fdata::DictionaryValue::Str("my_protocol".to_string())),
314        ]);
315
316        fasync::Task::spawn(async move {
317            let res =
318                main(ns_entries, out_server_end, lifecycle_server_end, Some(program_dict)).await;
319            assert!(res.is_ok(), "Broker main task failed");
320        })
321        .detach();
322
323        let (client_end, server_end) = zx::Channel::create();
324
325        out_dir
326            .open("svc/output", fio::Flags::PROTOCOL_SERVICE, &fio::Options::default(), server_end)
327            .expect("Failed to send open request to broker");
328
329        let signals = fasync::OnSignals::new(&client_end, zx::Signals::CHANNEL_READABLE)
330            .await
331            .expect("Failed waiting for signal. Routing may have dropped the channel.");
332
333        assert!(signals.contains(zx::Signals::CHANNEL_READABLE));
334
335        let (client_end2, server_end2) = zx::Channel::create();
336        out_dir
337            .open("svc/output", fio::Flags::PROTOCOL_SERVICE, &fio::Options::default(), server_end2)
338            .unwrap();
339
340        let _ = fasync::OnSignals::new(&client_end2, zx::Signals::CHANNEL_READABLE).await.unwrap();
341    }
342}
service_broker/lib.rs

service_broker/
lib.rs
