1mod apps;
2mod cache;
3mod core;
4mod deploys;
5mod files;
6mod host;
7mod index;
8mod mcp;
9mod observability;
10mod relay;
11mod shale;
12mod storage;
13mod telemetry;
14mod users;
15mod youtube;
16
17use axum::{
18 Router,
19 body::Bytes,
20 extract::{Request, State},
21 http::{HeaderMap, Method, StatusCode},
22 response::{IntoResponse, Response, Sse, sse::Event},
23 routing::any,
24};
25use serde_json::{Value, json};
26use std::{
27 collections::HashMap,
28 path::PathBuf,
29 sync::{Arc, Mutex},
30 time::Duration,
31};
32use subtle::ConstantTimeEq;
33use tokio::sync::{Semaphore, watch};
34use tokio_stream::{StreamExt, wrappers::WatchStream};
35use tower_http::services::{ServeDir, ServeFile};
36
37type Result<T> = std::result::Result<T, Error>;
38
39#[derive(Clone, Debug)]
40struct Error {
41 status: u16,
42 message: String,
43}
44impl Error {
45 fn new(status: u16, message: impl Into<String>) -> Self {
46 Self {
47 status,
48 message: message.into(),
49 }
50 }
51}
52impl<E: std::error::Error> From<E> for Error {
53 fn from(error: E) -> Self {
54 Self::new(500, error.to_string())
55 }
56}
57impl IntoResponse for Error {
58 fn into_response(self) -> Response {
59 if self.status >= 500 {
60 eprintln!("{}: {}", self.status, self.message);
61 }
62 (
63 StatusCode::from_u16(self.status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
64 self.message,
65 )
66 .into_response()
67 }
68}
69
70struct Document {
71 value: Value,
72 bytes: Bytes,
73}
74impl Document {
75 fn new(value: Value) -> Self {
76 let bytes = Bytes::from(serde_json::to_vec(&value).unwrap());
77 Self { value, bytes }
78 }
79 fn response(&self) -> Response {
80 ([("content-type", "application/json")], self.bytes.clone()).into_response()
81 }
82}
83
84struct App {
85 mcp: mcp::Store,
86 relay: relay::Broker,
87 shale: shale::Backend,
88 http: reqwest::Client,
89 internal: Option<(url::Url, reqwest::Client)>,
90 cache: cache::Cache,
91 nomad_slots: Semaphore,
92 specs: Mutex<HashMap<String, Value>>,
93 repo: PathBuf,
94 data: PathBuf,
95 live: watch::Sender<Bytes>,
96 usage: Mutex<HashMap<String, Value>>,
97 vm_usage: Mutex<HashMap<String, Value>>,
98 seed_samples: Mutex<Vec<Value>>,
99 youtube: std::sync::OnceLock<youtube::Worker>,
100 heavy: Semaphore,
101 file_changes: tokio::sync::Mutex<()>,
102 index: Option<Arc<index::Index>>,
103}
104impl App {
105 fn request(&self, method: Method, address: &str) -> Result<reqwest::RequestBuilder> {
106 let url = url::Url::parse(address)?;
107 let client = match &self.internal {
108 Some((base, client)) if url.origin() == base.origin() => client,
109 _ => &self.http,
110 };
111 Ok(client.request(method, url))
112 }
113}
114
115fn env(name: &str, default: &str) -> String {
116 std::env::var(name).unwrap_or_else(|_| default.into())
117}
118fn array(value: &Value) -> &[Value] {
119 value.as_array().map(Vec::as_slice).unwrap_or_default()
120}
121fn string(value: &Value) -> &str {
122 value.as_str().unwrap_or_default()
123}
124fn number(value: &Value) -> f64 {
125 value
126 .as_f64()
127 .or_else(|| value.as_str().and_then(|s| s.parse().ok()))
128 .unwrap_or(0.0)
129}
130fn now() -> f64 {
131 std::time::SystemTime::now()
132 .duration_since(std::time::UNIX_EPOCH)
133 .unwrap()
134 .as_secs_f64()
135}
136fn encoded(value: &str) -> String {
137 url::form_urlencoded::byte_serialize(value.as_bytes()).collect()
138}
139fn params(items: &[(&str, String)]) -> String {
140 url::form_urlencoded::Serializer::new(String::new())
141 .extend_pairs(items.iter().map(|(k, v)| (*k, v)))
142 .finish()
143}
144
145fn user(headers: &HeaderMap) -> Result<Value> {
146 let name = headers.get("User-Name").and_then(|v| v.to_str().ok()).filter(|v| !v.is_empty()).ok_or_else(|| Error::new(401, "No signed-in user came with this request. Open the dashboard through its sign-in page."))?;
147 let groups: Vec<&str> = headers
148 .get("User-Groups")
149 .and_then(|v| v.to_str().ok())
150 .unwrap_or_default()
151 .split(|c: char| c == ',' || c.is_whitespace())
152 .filter(|s| !s.is_empty())
153 .map(|s| s.strip_prefix("role:").unwrap_or(s))
154 .collect();
155 let preview = headers
156 .get("cookie")
157 .and_then(|v| v.to_str().ok())
158 .unwrap_or_default()
159 .split(';')
160 .find_map(|s| s.trim().strip_prefix("view-as="));
161 let viewing = preview.is_some() && groups.contains(&"infra-admin");
162 let groups = if viewing {
163 preview
164 .unwrap()
165 .split(',')
166 .filter(|s| !s.is_empty() && *s != "infra-admin")
167 .collect()
168 } else {
169 groups
170 };
171 let sections: Vec<&str> = [
172 ("launcher", None),
173 ("admin", Some("infra-admin")),
174 ("metrics", Some("metrics")),
175 ("media", Some("media-manage")),
176 ("vms", Some("vm")),
177 ]
178 .into_iter()
179 .filter(|(_, group)| can_open(&groups, *group))
180 .map(|(section, _)| section)
181 .collect();
182 Ok(json!({"name":name,"groups":groups,"sections":sections,"viewing":viewing}))
183}
184fn can_open(groups: &[&str], access: Option<&str>) -> bool {
185 access.is_none() || groups.contains(&"infra-admin") || groups.contains(&access.unwrap())
186}
187fn need(user: &Value, section: &str) -> Result<()> {
188 if array(&user["sections"]).iter().any(|s| s == section) {
189 Ok(())
190 } else {
191 Err(Error::new(
192 403,
193 "Your account can't open this section. Ask an admin to add you to its group.",
194 ))
195 }
196}
197
198async fn command(program: &str, args: &[&str], input: Option<&[u8]>) -> Result<Vec<u8>> {
199 use std::process::Stdio;
200 use tokio::io::AsyncWriteExt;
201 let mut child = tokio::process::Command::new(program)
202 .args(args)
203 .stdin(if input.is_some() {
204 Stdio::piped()
205 } else {
206 Stdio::null()
207 })
208 .stdout(Stdio::piped())
209 .stderr(Stdio::piped())
210 .kill_on_drop(true)
211 .spawn()?;
212 if let Some(input) = input {
213 child.stdin.take().unwrap().write_all(input).await?;
214 }
215 let output = tokio::time::timeout(Duration::from_secs(120), child.wait_with_output())
216 .await
217 .map_err(|_| {
218 Error::new(
219 504,
220 "The operation is taking too long. Check its logs, then retry.",
221 )
222 })??;
223 if !output.status.success() {
224 return Err(Error::new(
225 502,
226 String::from_utf8_lossy(&output.stderr).trim(),
227 ));
228 }
229 Ok(output.stdout)
230}
231
232async fn api(State(app): State<Arc<App>>, request: Request) -> Result<Response> {
233 let path = request
234 .uri()
235 .path()
236 .trim_start_matches("/api/")
237 .trim_end_matches('/')
238 .to_owned();
239 let method = request.method().clone();
240 let query: HashMap<String, String> =
241 url::form_urlencoded::parse(request.uri().query().unwrap_or_default().as_bytes())
242 .into_owned()
243 .collect();
244 let parts: Vec<&str> = path.split('/').collect();
245 if parts.first() == Some(&"icons") && method == Method::GET {
246 return core::icon(&app, &parts, &query).await;
247 }
248 if parts.starts_with(&["account", "pictures"]) && method == Method::GET {
249 return users::picture(&app, &parts).await;
250 }
251 let me = user(request.headers())?;
252 let headers = request.headers().clone();
253 let section = match parts.first().copied().unwrap_or_default() {
254 "host" | "metrics" | "live" => Some("metrics"),
255 "services" if parts.len() == 1 => Some("metrics"),
256 "storage" if parts.len() == 1 => Some("metrics"),
257 "services" | "traces" | "storage" | "users" | "deploys" | "paper-clover" => Some("admin"),
258 "media" | "seedbox" | "youtube" => Some("media"),
259 "vms" => Some("vms"),
260 _ => None,
261 };
262 if let Some(section) = section {
263 need(&me, section)?;
264 }
265 if let ["deploys", "runs", id] = parts.as_slice() {
266 return deploys::run_stream(app, id).await;
267 }
268 if parts.first() == Some(&"account") {
269 return users::account(app, request, &parts[1..], &me).await;
270 }
271 if path == "live" && method == Method::GET {
272 let stream = WatchStream::new(app.live.subscribe())
273 .filter(|data| !data.is_empty())
274 .map(|data| {
275 Ok::<_, std::convert::Infallible>(
276 Event::default().data(String::from_utf8_lossy(&data)),
277 )
278 });
279 return Ok(Sse::new(stream).into_response());
280 }
281 let body = axum::body::to_bytes(request.into_body(), 8 * 1024 * 1024).await?;
282 let value = if body.is_empty() {
283 Value::Null
284 } else {
285 serde_json::from_slice(&body)
286 .map_err(|_| Error::new(400, "The request didn't match what this route expects."))?
287 };
288 match parts[0] {
289 "mcp" => mcp::manage(app, &method, &parts[1..], &me, value, &headers).await,
290 "users" => users::route(app, &method, &parts[1..], &me, value).await,
291 "vms" | "seedbox" | "paper-clover" => {
292 apps::route(app, &method, &parts, &query, value).await
293 }
294 "youtube" => youtube::route(app, &method, &parts[1..], value).await,
295 "media" => files::route(app, true, &method, &parts[1..], &query, value).await,
296 "storage" if parts.get(1) == Some(&"files") => {
297 if parts.len() == 3 && ["map", "largest"].contains(&parts[2]) {
298 index::route(app, parts[2], &query).await
299 } else {
300 files::route(app, false, &method, &parts[2..], &query, value).await
301 }
302 }
303 "storage" => storage::route(app, &method, &parts[1..], &query, value).await,
304 "deploys" => deploys::route(app, &method, &parts[1..], &query).await,
305 _ => core::route(app, &method, &parts, &query, &me, value).await,
306 }
307}
308
309#[tokio::main(worker_threads = 4)]
310async fn main() -> std::result::Result<(), Box<dyn std::error::Error>> {
311 let address: std::net::IpAddr = env("STUDIO_LISTEN_ADDRESS", "127.0.0.1").parse()?;
312 let proof = match std::env::var_os("STUDIO_PROXY_TOKEN_FILE") {
313 Some(file) => {
314 let token = std::fs::read_to_string(file)?.trim().to_owned();
315 if token.len() != 64 || !token.bytes().all(|b| b.is_ascii_hexdigit()) {
316 return Err(std::io::Error::other(
317 "The dashboard proxy token must contain 64 hexadecimal characters.",
318 )
319 .into());
320 }
321 Some(Arc::<str>::from(token))
322 }
323 None if address.is_loopback() => None,
324 None => {
325 return Err(std::io::Error::other(
326 "Set STUDIO_PROXY_TOKEN_FILE before listening on a non-loopback address.",
327 )
328 .into());
329 }
330 };
331 let certificates = if let Ok(path) =
332 std::env::var("STUDIO_CA_BUNDLE").or_else(|_| std::env::var("NODE_EXTRA_CA_CERTS"))
333 {
334 reqwest::Certificate::from_pem_bundle(&std::fs::read(path)?)?
335 } else {
336 Vec::new()
337 };
338 let client = || {
339 let mut http = reqwest::Client::builder().timeout(Duration::from_secs(15));
340 for certificate in &certificates {
341 http = http.add_root_certificate(certificate.clone());
342 }
343 http
344 };
345 let internal = std::env::var("STUDIO_INTERNAL_URL")
346 .ok()
347 .map(
348 |address| -> std::result::Result<_, Box<dyn std::error::Error>> {
349 let base = url::Url::parse(&address)?;
350 if base.scheme() != "https"
351 || base.host_str().is_none()
352 || !base.username().is_empty()
353 || base.password().is_some()
354 || base.path() != "/"
355 || base.query().is_some()
356 || base.fragment().is_some()
357 {
358 return Err(std::io::Error::other(
359 "STUDIO_INTERNAL_URL must be an HTTPS origin.",
360 )
361 .into());
362 }
363 let token = proof.as_ref().ok_or_else(|| {
364 std::io::Error::other("STUDIO_INTERNAL_URL requires STUDIO_PROXY_TOKEN_FILE.")
365 })?;
366 let mut value = axum::http::HeaderValue::from_str(token)?;
367 value.set_sensitive(true);
368 let mut headers = HeaderMap::new();
369 headers.insert("Studio-Proxy-Token", value);
370 Ok((
371 base,
372 client()
373 .default_headers(headers)
374 .redirect(reqwest::redirect::Policy::none())
375 .build()?,
376 ))
377 },
378 )
379 .transpose()?;
380 let (live, _) = watch::channel(Bytes::new());
381 let index = std::env::var("STUDIO_INDEX_POOL").ok().map(|pool| {
382 Arc::new(index::Index::new(
383 pool,
384 env("STUDIO_INDEX_DIR", ".cache/index").into(),
385 ))
386 });
387 let origin = env(
388 "STUDIO_PUBLIC_ORIGIN",
389 &format!("https://snowglobe.{}", env("STUDIO_DOMAIN", "studio.test")),
390 );
391 let app = Arc::new(App {
392 mcp: mcp::Store::new(&PathBuf::from(env("STUDIO_DATA_DIR", "data")), &origin)
393 .map_err(|error| std::io::Error::other(error.message))?,
394 relay: relay::Broker::default(),
395 shale: shale::Backend::new(
396 &env(
397 "STUDIO_SHALE_URL",
398 &format!("https://shale.{}", env("STUDIO_DOMAIN", "studio.test")),
399 ),
400 client(),
401 )
402 .map_err(|error| std::io::Error::other(error.message))?,
403 http: client().build()?,
404 internal,
405 cache: cache::Cache::default(),
406 nomad_slots: Semaphore::new(4),
407 specs: Mutex::new(HashMap::new()),
408 repo: env("STUDIO_REPO", "..").into(),
409 data: env("STUDIO_DATA_DIR", "data").into(),
410 live,
411 usage: Mutex::new(HashMap::new()),
412 vm_usage: Mutex::new(HashMap::new()),
413 seed_samples: Mutex::new(Vec::new()),
414 youtube: std::sync::OnceLock::new(),
415 heavy: Semaphore::new(4),
416 file_changes: tokio::sync::Mutex::new(()),
417 index,
418 });
419 if let Some(index) = app.index.clone() {
420 let state = app.clone();
421 tokio::spawn(async move {
422 index.start(state).await;
423 });
424 }
425 if std::env::var("STUDIO_YT_STATE").is_ok() {
426 for parts in [&[][..], &["library"][..]] {
427 let state = app.clone();
428 tokio::spawn(async move {
429 if let Err(error) = youtube::route(state, &Method::GET, parts, Value::Null).await {
430 eprintln!("youtube: {}", error.message);
431 }
432 });
433 }
434 }
435 core::start(app.clone());
436 telemetry::start(app.clone());
437 let dist = env("STUDIO_WEB_DIR", "dist");
438 let router = Router::new()
439 .route("/api/{*path}", any(api))
440 .route("/oauth/{*path}", any(mcp::oauth))
441 .route("/.well-known/{*path}", any(mcp::oauth))
442 .nest_service("/assets", ServeDir::new(format!("{dist}/assets")))
443 .with_state(app.clone())
444 .merge(observability::router(app.clone()))
445 .merge(shale::router(app.clone()))
446 .merge(relay::router(app))
447 .fallback_service(
448 ServeDir::new(&dist).fallback(ServeFile::new(format!("{dist}/index.html"))),
449 )
450 .layer(axum::middleware::from_fn(
451 move |mut request: Request, next: axum::middleware::Next| {
452 let proof = proof.clone();
453 async move {
454 if mcp::public(request.uri().path()) {
455 request.headers_mut().remove("Studio-Proxy-Token");
456 request.headers_mut().remove("User-Name");
457 request.headers_mut().remove("User-Groups");
458 } else if let Some(proof) = proof {
459 let supplied = request
460 .headers()
461 .get("Studio-Proxy-Token")
462 .map(|value| value.as_bytes())
463 .unwrap_or_default();
464 if supplied.ct_eq(proof.as_bytes()).unwrap_u8() == 0 {
465 return Error::new(403, "Open the dashboard through its sign-in page.")
466 .into_response();
467 }
468 request.headers_mut().remove("Studio-Proxy-Token");
469 }
470 let asset = request.uri().path().starts_with("/assets/");
471 let document = !asset && !request.uri().path().starts_with("/api/");
472 if document {
473 request.headers_mut().remove("if-modified-since");
474 request.headers_mut().remove("if-none-match");
475 }
476 let mut response = next.run(request).await;
477 if asset && response.status().is_success() {
478 response.headers_mut().insert(
479 "cache-control",
480 "public, max-age=31536000, immutable".parse().unwrap(),
481 );
482 } else if document {
483 response
484 .headers_mut()
485 .insert("cache-control", "no-store".parse().unwrap());
486 }
487 response
488 }
489 },
490 ));
491 let listener = tokio::net::TcpListener::bind(std::net::SocketAddr::new(
492 address,
493 env("PORT", "7070").parse()?,
494 ))
495 .await?;
496 println!("dashboard on {}", listener.local_addr()?);
497 let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
498 let (shutdown, waiting) = tokio::sync::oneshot::channel::<()>();
499 let mut server = Box::pin(std::future::IntoFuture::into_future(
500 axum::serve(listener, router).with_graceful_shutdown(async {
501 let _ = waiting.await;
502 }),
503 ));
504 tokio::select! {
505 result = &mut server => { result?; return Ok(()); },
506 _ = terminate.recv() => {},
507 _ = tokio::signal::ctrl_c() => {},
508 }
509 let _ = shutdown.send(());
510 if let Ok(result) = tokio::time::timeout(Duration::from_secs(5), server).await {
511 result?;
512 }
513 Ok(())
514}
515
516#[cfg(test)]
517mod tests {
518 use super::*;
519 #[test]
520 fn only_an_admin_can_preview_groups_and_preview_cannot_grant_admin() {
521 let mut headers = HeaderMap::new();
522 assert_eq!(user(&headers).unwrap_err().status, 401);
523 headers.insert("User-Name", "snow".parse().unwrap());
524 headers.insert("User-Groups", "role:metrics,media-manage".parse().unwrap());
525 headers.insert("cookie", "view-as=infra-admin".parse().unwrap());
526 let me = user(&headers).unwrap();
527 assert_eq!(me["viewing"], false);
528 assert!(need(&me, "admin").is_err());
529 assert!(need(&me, "metrics").is_ok());
530 headers.insert("User-Groups", "infra-admin".parse().unwrap());
531 let me = user(&headers).unwrap();
532 assert_eq!(me["viewing"], true);
533 assert!(need(&me, "admin").is_err());
534 headers.insert("cookie", "view-as=metrics".parse().unwrap());
535 let me = user(&headers).unwrap();
536 assert!(need(&me, "metrics").is_ok());
537 assert!(need(&me, "media").is_err());
538 }
539}