1use crate::*;
2use tokio::io::{AsyncReadExt, AsyncWriteExt};
3
4#[derive(serde::Deserialize)]
5struct Limits { request: usize, response: usize }
6static LIMITS: std::sync::LazyLock<Limits> = std::sync::LazyLock::new(|| serde_json::from_str(include_str!("host-protocol.json")).expect("host protocol limits"));
7
8static SLOTS: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(4);
9
10pub async fn sample(app: Arc<App>) -> Result<Arc<Document>> {
11 app.cache
12 .get("host-sample".into(), Duration::from_secs(1), || async {
13 call(json!({"operation":"host.sample"})).await
14 })
15 .await
16}
17
18pub async fn call(request: Value) -> Result<Value> {
19 let _slot = tokio::time::timeout(Duration::from_secs(70), SLOTS.acquire())
20 .await
21 .map_err(|_| {
22 Error::new(
23 504,
24 "The host operation is taking too long. Check its logs, then retry.",
25 )
26 })??;
27 open(request).await.map(|(value, _)| value)
28}
29
30// Streams authenticate the same Unix peer and use the same framed JSON header.
31pub async fn open(request: Value) -> Result<(Value, tokio::net::UnixStream)> {
32 let timeout = if matches!(
33 request["operation"].as_str(),
34 Some("vm.create" | "vm.media" | "vm.preset")
35 ) {
36 910
37 } else {
38 70
39 };
40 tokio::time::timeout(Duration::from_secs(timeout), async {
41 let mut socket = tokio::net::UnixStream::connect(env(
42 "STUDIO_HOST_SOCKET",
43 "/run/studio-host/host.sock",
44 ))
45 .await?;
46 #[cfg(target_os = "linux")]
47 if socket.peer_cred()?.uid() != 0 {
48 return Err(Error::new(
49 502,
50 "The host service identity couldn't be verified.",
51 ));
52 }
53 let mut message = serde_json::to_vec(&request)?;
54 message.push(b'\n');
55 if message.len() > LIMITS.request {
56 return Err(Error::new(
57 400,
58 "The host request is too large. Narrow the selection.",
59 ));
60 }
61 socket.write_all(&message).await?;
62 let value = response(&mut socket).await?;
63 Ok((value, socket))
64 })
65 .await
66 .map_err(|_| Error::new(504, "The host connection timed out. Try again."))?
67}
68
69pub async fn response(socket: &mut tokio::net::UnixStream) -> Result<Value> {
70 let length = socket.read_u32().await? as usize;
71 if length > LIMITS.response {
72 return Err(Error::new(
73 502,
74 "The host response is too large. Narrow the selection.",
75 ));
76 }
77 let mut bytes = vec![0; length];
78 socket.read_exact(&mut bytes).await?;
79 let mut response: Value = serde_json::from_slice(&bytes)?;
80 if let Some(message) = response["error"].as_str() {
81 return Err(Error::new(
82 response["status"]
83 .as_u64()
84 .filter(|s| (400..=599).contains(s))
85 .unwrap_or(502) as u16,
86 message,
87 ));
88 }
89 response
90 .as_object_mut()
91 .and_then(|v| v.remove("value"))
92 .ok_or_else(|| {
93 Error::new(
94 502,
95 "The host response is incomplete. Check its logs, then retry.",
96 )
97 })
98}