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