1use std::path::PathBuf;
2use std::sync::Arc;
3
4use crossbeam_channel::Sender;
5use koan_core::audio::viz::VizSnapshot;
6use koan_core::auth::{self, parse_duration_secs};
7use koan_core::config::Config;
8use koan_core::db::pool::Pool;
9use koan_core::player::commands::PlayerCommand;
10use koan_core::player::state::SharedPlayerState;
11
12use super::KoanSchema;
13use crate::auth::AuthUser;
14use crate::auth::middleware::{AuthState, auth_middleware};
15use crate::auth::routes::{AuthRouteState, RateLimiter, auth_router};
16
17fn require_server_url() {
24 let why = match Config::cached().server_url() {
25 None => "server.url is not set. Set it, or KOAN_SERVER__URL, to the address clients reach this server at, e.g. https://koan.example.com".to_owned(),
26 Some(url) if !url.starts_with("https://") && !url.starts_with("http://") => {
27 format!("server.url is {url:?}, which is not an http:// or https:// address")
28 }
29 Some(_) => return,
30 };
31 log::error!("{why}");
32 eprintln!("koan: {why}");
33 std::process::exit(1);
34}
35
36pub fn cmd_serve(
37 port: Option<u16>,
38 bind: Option<std::net::IpAddr>,
39 subsonic_port: Option<u16>,
40 playground: bool,
41) {
42 use koan_core::player::Player;
43
44 require_server_url();
45 if let Err(e) = koan_core::db::connection::Database::open_default() {
49 log::error!("cannot open the database: {e}");
50 eprintln!("koan: cannot open the database: {e}");
51 std::process::exit(1);
52 }
53 let db_path = koan_core::config::db_path();
54 let pool = Arc::new(Pool::new(db_path.clone()));
55 koan_core::db::pool::on_outdated(|| OUTDATED.cancel());
56
57 let (state, _timeline, _viz, cmd_tx) = Player::spawn();
58
59 let watched = db_path.clone();
62 let jobs = super::jobs::JobRegistry::default();
63 let record = super::jobs::record_watch(&jobs);
64 koan_core::helpers::spawn_library_watch(db_path, move |scan| {
65 record(&scan);
66 if let koan_core::helpers::WatchScan::Finished(_) = scan {
69 crate::clients::fulfil_from(&watched);
70 if let Ok(db) = koan_core::db::connection::Database::open_existing(&watched) {
71 crate::clients::changed_if_library_moved(&db.conn);
72 }
73 }
74 });
75
76 let started = ApiServer::start_with(
77 ApiServerOpts {
78 state,
79 cmd_tx,
80 pool,
81 port,
82 bind,
83 subsonic_port,
84 playground,
85 viz: None, profile: Profile::Headless,
87 hosts: Vec::new(),
88 },
89 jobs,
90 );
91 match started {
92 Ok(server) => server.wait(),
93 Err(e) => {
94 eprintln!("koan: {}", e);
95 std::process::exit(1);
96 }
97 }
98}
99
100const REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
107
108const MAX_INFLIGHT_QUERIES: usize = 64;
112
113const MAX_QUERY_BODY: usize = 2 << 20;
117
118fn load_perimeter<S>(router: axum::Router<S>) -> axum::Router<S>
123where
124 S: Clone + Send + Sync + 'static,
125{
126 router
127 .layer(tower_http::catch_panic::CatchPanicLayer::new())
130 .layer(tower_http::limit::RequestBodyLimitLayer::new(
131 MAX_QUERY_BODY,
132 ))
133 .layer(tower_http::timeout::TimeoutLayer::with_status_code(
134 axum::http::StatusCode::REQUEST_TIMEOUT,
135 REQUEST_TIMEOUT,
136 ))
137 .layer(
141 tower::ServiceBuilder::new()
142 .layer(axum::error_handling::HandleErrorLayer::new(
143 |err: tower::BoxError| async move {
144 if err.is::<tower::load_shed::error::Overloaded>() {
145 (
146 axum::http::StatusCode::SERVICE_UNAVAILABLE,
147 "server at capacity",
148 )
149 } else {
150 (
151 axum::http::StatusCode::INTERNAL_SERVER_ERROR,
152 "internal error",
153 )
154 }
155 },
156 ))
157 .load_shed()
158 .concurrency_limit(MAX_INFLIGHT_QUERIES),
159 )
160}
161
162#[derive(Clone, Copy, Debug, PartialEq, Eq)]
165pub enum Profile {
166 Headless,
169 Tui,
171 Embedded,
176}
177
178pub struct ApiServerOpts {
180 pub state: Arc<SharedPlayerState>,
181 pub cmd_tx: Sender<PlayerCommand>,
182 pub pool: Arc<Pool>,
185 pub port: Option<u16>,
186 pub bind: Option<std::net::IpAddr>,
187 pub subsonic_port: Option<u16>,
188 pub playground: bool,
189 pub viz: Option<Arc<VizSnapshot>>,
190 pub profile: Profile,
191 pub hosts: Vec<String>,
194}
195
196pub struct ApiServer {
199 addr: std::net::SocketAddr,
200 host_key: Option<String>,
203 stop: tokio_util::sync::CancellationToken,
204 thread: Option<std::thread::JoinHandle<()>>,
205}
206
207impl ApiServer {
208 pub fn start(opts: ApiServerOpts) -> Result<Self, String> {
218 Self::start_with(opts, Default::default())
219 }
220
221 pub(crate) fn start_with(
222 opts: ApiServerOpts,
223 jobs: super::jobs::JobRegistry,
224 ) -> Result<Self, String> {
225 let ApiServerOpts {
226 state,
227 cmd_tx,
228 pool,
229 port,
230 bind,
231 subsonic_port,
232 playground,
233 viz,
234 profile,
235 hosts,
236 } = opts;
237 use axum::routing::{get, post};
238
239 let cfg = Config::cached();
240 let embedded = profile == Profile::Embedded;
241 let port = port.unwrap_or(cfg.graphql.port);
242 let setup_wizard = cfg.graphql.setup_wizard && profile == Profile::Headless;
246 let bind = bind.unwrap_or(cfg.graphql.bind);
247 let subsonic_port = if embedded {
249 None
250 } else {
251 subsonic_port.or(cfg.subsonic.port)
252 };
253 let subsonic_enabled = embedded || cfg.subsonic.enabled;
254 let playground_enabled = !embedded && (playground || cfg.graphql.playground);
255 let auth_enabled = cfg.graphql.auth_enabled;
256 let demo = cfg.server.demo;
257
258 if embedded && !auth_enabled {
259 return Err(
260 "graphql.auth_enabled is off, which would make anyone who reaches \
261 this port an admin"
262 .into(),
263 );
264 }
265 if demo && embedded {
266 return Err("server.demo is for a headless server".into());
267 }
268 if demo && !auth_enabled {
270 return Err("server.demo needs graphql.auth_enabled".into());
271 }
272
273 let (private_pem, public_pem) = if auth_enabled {
280 let kp = auth::load_or_generate_keypair().map_err(|e| {
281 format!(
282 "auth_enabled = true but the keypair could not be loaded or created: {}",
283 e
284 )
285 })?;
286 if kp.0.is_empty() || kp.1.is_empty() {
289 return Err("auth_enabled = true but the keypair files are empty. \
290 Run `koan auth regenerate-keys`."
291 .into());
292 }
293 kp
294 } else {
295 auth::load_or_generate_keypair().unwrap_or_default()
298 };
299
300 let access_ttl = parse_duration_secs(&cfg.graphql.access_token_ttl).unwrap_or(900);
301 let refresh_ttl = parse_duration_secs(&cfg.graphql.refresh_token_ttl).unwrap_or(2_592_000);
302
303 let introspection_key = if playground_enabled && auth_enabled {
307 Some(Arc::new(auth::random_token().map_err(|e| {
308 format!("failed to generate introspection key: {}", e)
309 })?))
310 } else {
311 None
312 };
313
314 let proxy_auth = if embedded {
315 None
316 } else {
317 crate::ui::ProxyAuth::from_config(
318 &cfg.graphql.proxy_auth_header,
319 &cfg.graphql.proxy_auth_from,
320 )?
321 };
322
323 if bind.is_unspecified() && port != 0 {
328 let loopbacks: &[std::net::IpAddr] = match bind {
330 std::net::IpAddr::V4(_) => &[std::net::Ipv4Addr::LOCALHOST.into()],
331 std::net::IpAddr::V6(_) => &[
332 std::net::Ipv4Addr::LOCALHOST.into(),
333 std::net::Ipv6Addr::LOCALHOST.into(),
334 ],
335 };
336 for loopback in loopbacks {
337 std::net::TcpListener::bind((*loopback, port))
338 .map_err(|e| format!("port {port} is already in use on this machine — {e}"))?;
339 }
340 }
341 let gql_listener = std::net::TcpListener::bind((bind, port))
342 .and_then(|l| l.set_nonblocking(true).map(|()| l))
343 .map_err(|e| {
344 format!("failed to bind GraphQL port {port} — {e} (another instance running?)")
345 })?;
346 let addr = gql_listener
347 .local_addr()
348 .map_err(|e| format!("failed to read the GraphQL listener's address — {e}"))?;
349
350 let sub_listener = subsonic_port
353 .filter(|p| subsonic_enabled && *p != port)
354 .and_then(|sub_port| {
355 match std::net::TcpListener::bind((bind, sub_port))
356 .and_then(|l| l.set_nonblocking(true).map(|()| l))
357 {
358 Ok(l) => Some(l),
359 Err(e) => {
360 log::warn!(
361 "Dedicated Subsonic port {sub_port} unavailable — {e}. Mounted on GraphQL port only."
362 );
363 None
364 }
365 }
366 });
367
368 let rt = tokio::runtime::Runtime::new()
369 .map_err(|e| format!("could not start the API server's runtime — {e}"))?;
370 let (gql_listener, sub_listener) = {
371 let _in = rt.enter();
372 let gql = tokio::net::TcpListener::from_std(gql_listener)
373 .map_err(|e| format!("failed to listen on {addr} — {e}"))?;
374 let sub = sub_listener.and_then(|l| match tokio::net::TcpListener::from_std(l) {
375 Ok(l) => Some(l),
376 Err(e) => {
377 log::warn!("Dedicated Subsonic listener unusable — {e}");
378 None
379 }
380 });
381 (gql, sub)
382 };
383
384 let host_key = embedded
385 .then(auth::random_token)
386 .transpose()
387 .map_err(|e| format!("failed to generate the host key: {e}"))?;
388
389 if demo {
390 let db = pool.get().map_err(|e| format!("demo: {e}"))?;
391 crate::demo::start(&db.conn).map_err(|e| format!("demo: {e}"))?;
392 }
393
394 let _ = std::fs::remove_file(auth::keypair_dir().join("subsonic.key"));
397
398 crate::auth::set_signing_keys(Arc::new(private_pem.clone()), Arc::new(public_pem.clone()));
399
400 let auth_state = AuthState {
401 public_pem: Arc::new(public_pem.clone()),
402 auth_enabled,
403 introspection_key: introspection_key.clone(),
404 pool: pool.clone(),
405 };
406
407 let users = Arc::new(crate::auth::password::PasswordVerifier::new(pool.clone()));
409 let auth_route_state = AuthRouteState {
410 users: users.clone(),
411 pool: pool.clone(),
412 private_pem: Arc::new(private_pem),
413 public_pem: Arc::new(public_pem),
414 access_ttl_secs: access_ttl,
415 refresh_ttl_secs: refresh_ttl,
416 cookie_secure: cfg.graphql.cookie_secure,
417 login_limiter: Arc::new(RateLimiter::default()),
418 };
419
420 koan_core::scrobbling::start(pool.path().to_path_buf());
422 let _ = crate::push::pusher();
425
426 let stop = tokio_util::sync::CancellationToken::new();
428 let mcp_routes = (!embedded).then(|| {
431 crate::mcp::router(
432 state.clone(),
433 cmd_tx.clone(),
434 auth_state.clone(),
435 cfg.server_url(),
436 profile == Profile::Headless,
437 stop.clone(),
438 )
439 });
440 let schema = super::build_schema_sharing(state, cmd_tx, pool.clone(), viz, jobs.clone());
441
442 if auth_enabled {
443 log::info!(
444 "Auth enabled (Ed25519 JWT, access TTL {}s, refresh TTL {}s)",
445 access_ttl,
446 refresh_ttl
447 );
448 let no_accounts = pool
449 .get()
450 .ok()
451 .and_then(|db| koan_core::db::queries::auth::has_users(&db.conn).ok())
452 == Some(false);
453 if no_accounts && setup_wizard {
454 log::warn!(
455 "No accounts yet: whoever opens /setup first becomes the admin. \
456 Set graphql.setup_wizard = false to make it with `koan auth setup` instead."
457 );
458 }
459 } else {
460 log::info!("Auth disabled — all requests treated as admin");
461 }
462
463 let mut allowed_hosts = cfg.graphql.allowed_hosts.clone();
464 allowed_hosts.extend(hosts);
465 let forwarded_from = if embedded {
468 crate::auth::routes::ForwardedFrom::Loopback
469 } else {
470 crate::auth::routes::ForwardedFrom::Private
471 };
472 let browser_policy = Arc::new(BrowserPolicy {
473 origins: cfg.graphql.cors_origins.clone(),
474 hosts: allowed_hosts,
475 });
476
477 if cfg.graphql.cors_origins.is_empty() {
478 log::info!("CORS: no origins configured — browsers get no cross-origin access");
479 }
480
481 let serving = stop.clone();
482 let serving_key = host_key.clone();
483 let thread = std::thread::Builder::new()
484 .name("koan-api".into())
485 .spawn(move || {
486 let stop = serving;
487 rt.block_on(async {
488 if !embedded {
489 let stop = stop.clone();
490 tokio::spawn(async move {
491 shutdown_signal().await;
492 stop.cancel();
493 });
494 }
495
496 let query_route =
501 load_perimeter(axum::Router::new().route("/graphql", post(graphql_handler)));
502
503 let gql_app = axum::Router::new()
504 .merge(query_route)
505 .route("/graphql/ws", get(graphql_ws_handler))
506 .layer(axum::Extension(stop.clone()))
507 .layer(axum::middleware::from_fn_with_state(
508 auth_state.clone(),
509 auth_middleware,
510 ))
511 .layer(axum::middleware::from_fn_with_state(
514 browser_policy.clone(),
515 browser_guard,
516 ))
517 .with_state(schema);
518
519 let covers = Arc::new(crate::covers::Covers::in_config_dir());
523 let ui_routes = (!embedded).then(|| crate::ui::router(
524 pool.clone(),
525 auth_route_state.clone(),
526 auth_enabled,
527 setup_wizard,
528 covers.clone(),
529 cfg.server_url(),
530 cfg.mcp.redirect_hosts.clone(),
531 proxy_auth,
532 jobs,
533 )
534 .layer(tower_http::catch_panic::CatchPanicLayer::new()));
536 let ui_routes = ui_routes.map(|ui| {
537 if demo {
538 ui.layer(axum::middleware::from_fn(crate::demo::guard))
539 } else {
540 ui
541 }
542 });
543
544 let auth_app = auth_router(auth_route_state);
546
547 let origins: Vec<axum::http::HeaderValue> = cfg
550 .graphql
551 .cors_origins
552 .iter()
553 .filter_map(|o| o.parse().ok())
554 .collect();
555 let cors = tower_http::cors::CorsLayer::new()
556 .allow_origin(origins)
557 .allow_methods([
558 axum::http::Method::GET,
559 axum::http::Method::POST,
560 axum::http::Method::OPTIONS,
561 ])
562 .allow_headers([
563 axum::http::header::AUTHORIZATION,
564 axum::http::header::CONTENT_TYPE,
565 axum::http::HeaderName::from_static("x-introspection-key"),
566 ])
567 .allow_credentials(true);
568
569 let share_url = match embedded {
575 true => crate::ServerUrl::Live,
576 false => crate::ServerUrl::Fixed(cfg.server_url()),
577 };
578 let share_routes =
579 crate::share::router(pool.clone(), share_url, covers.clone())
580 .layer(tower_http::catch_panic::CatchPanicLayer::new());
581 let subsonic_merged = subsonic_enabled
588 .then(|| {
589 crate::subsonic::subsonic_router(
590 pool,
591 covers,
592 users,
593 profile,
594 stop.clone(),
595 serving_key,
596 )
597 })
598 .map(|sub| {
599 if demo {
600 sub.layer(axum::middleware::from_fn(crate::demo::guard))
601 } else {
602 sub
603 }
604 });
605 let subsonic_dedicated = subsonic_merged.clone();
606
607 let mut app = match embedded {
611 true => share_routes,
612 false => auth_app.merge(gql_app).merge(share_routes),
613 };
614 if let Some(ui) = ui_routes {
615 app = app.merge(ui);
616 }
617 if let Some(mcp) = mcp_routes {
618 app = app.merge(mcp);
619 }
620 if let Some(sub) = subsonic_merged {
621 app = app.merge(sub);
622 }
623 if playground_enabled {
624 app = app.route(
625 "/graphql",
626 get(graphql_playground).with_state(introspection_key.clone()),
627 );
628 }
629 let app = app
633 .layer(cors)
634 .layer(axum::middleware::from_fn_with_state(
635 browser_policy.clone(),
636 host_guard,
637 ))
638 .layer(axum::Extension(forwarded_from));
639
640 log::info!("GraphQL API on http://{addr}/graphql");
641 if subsonic_enabled {
642 log::info!("Subsonic REST on http://{addr}/rest/");
643 }
644 if playground_enabled {
645 let playground_url = match introspection_key {
646 Some(ref key) => {
647 format!("http://{addr}/graphql?introspection-key={key}")
648 }
649 None => format!("http://{addr}/graphql"),
650 };
651 log::info!("GraphiQL: {}", playground_url);
652 #[cfg(target_os = "macos")]
654 let _ = std::process::Command::new("open").arg(&playground_url).spawn();
655 #[cfg(target_os = "linux")]
656 let _ = std::process::Command::new("xdg-open")
657 .arg(&playground_url)
658 .spawn();
659 }
660 let gql_server = axum::serve(
661 gql_listener,
662 app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
663 )
664 .with_graceful_shutdown(stop.clone().cancelled_owned());
665
666 if let Some((sub_listener, sub_app)) = sub_listener.zip(subsonic_dedicated) {
667 if let Ok(a) = sub_listener.local_addr() {
668 log::info!("Subsonic REST also on http://{a}/rest/ (dedicated port)");
669 }
670 let sub_server = axum::serve(
673 sub_listener,
674 sub_app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
675 )
676 .with_graceful_shutdown(stop.clone().cancelled_owned());
677
678 tokio::select! {
679 r = gql_server => { if let Err(e) = r { log::error!("GraphQL server error: {e}"); } },
680 r = sub_server => { if let Err(e) = r { log::error!("Subsonic server error: {e}"); } },
681 _ = drain_deadline(&stop) => log::info!("shutting down with connections still open"),
682 }
683 return;
684 }
685
686 tokio::select! {
687 r = gql_server => { if let Err(e) = r { log::error!("GraphQL server error: {e}"); } },
688 _ = drain_deadline(&stop) => log::info!("shutting down with connections still open"),
689 }
690 });
691 rt.shutdown_timeout(DRAIN);
694 })
695 .map_err(|e| format!("could not start the API server thread — {e}"))?;
696
697 Ok(Self {
698 addr,
699 host_key,
700 stop,
701 thread: Some(thread),
702 })
703 }
704
705 pub fn host_key(&self) -> Option<&str> {
707 self.host_key.as_deref()
708 }
709
710 pub fn addr(&self) -> std::net::SocketAddr {
712 self.addr
713 }
714
715 pub fn stop(mut self) {
718 self.stop.cancel();
719 self.join();
720 }
721
722 pub fn wait(mut self) {
725 self.join();
726 }
727
728 fn join(&mut self) {
729 if let Some(thread) = self.thread.take()
730 && thread.join().is_err()
731 {
732 log::error!("API server thread panicked");
733 }
734 }
735}
736
737impl Drop for ApiServer {
738 fn drop(&mut self) {
739 self.stop.cancel();
740 }
741}
742
743pub fn start_api_background(
746 state: Arc<SharedPlayerState>,
747 cmd_tx: Sender<PlayerCommand>,
748 db_path: PathBuf,
749 port: Option<u16>,
750 bind: Option<std::net::IpAddr>,
751 subsonic_port: Option<u16>,
752 playground: bool,
753) {
754 let started = ApiServer::start(ApiServerOpts {
755 state,
756 cmd_tx,
757 pool: Arc::new(Pool::new(db_path)),
758 port,
759 bind,
760 subsonic_port,
761 playground,
762 viz: None,
763 profile: Profile::Tui,
764 hosts: Vec::new(),
765 });
766 match started {
767 Ok(server) => server.wait(),
768 Err(e) => log::error!("API server not started: {}", e),
769 }
770}
771
772pub(crate) struct BrowserPolicy {
781 origins: Vec<String>,
782 hosts: Vec<String>,
783}
784
785impl BrowserPolicy {
786 fn host_allowed(&self, host: &str) -> bool {
787 if self.hosts.iter().any(|h| h.eq_ignore_ascii_case(host)) {
788 return true;
789 }
790 let bare = strip_port(host);
791 if self.hosts.iter().any(|h| h.eq_ignore_ascii_case(bare)) {
792 return true;
793 }
794 bare.eq_ignore_ascii_case("localhost") || bare.parse::<std::net::IpAddr>().is_ok()
797 }
798
799 fn origin_allowed(&self, origin: &str, host: Option<&str>) -> bool {
802 if self.origins.iter().any(|o| o == origin) {
803 return true;
804 }
805 match (origin.split_once("://"), host) {
806 (Some((_, authority)), Some(host)) => authority.eq_ignore_ascii_case(host),
807 _ => false,
808 }
809 }
810}
811
812fn strip_port(host: &str) -> &str {
814 if let Some(rest) = host.strip_prefix('[') {
815 return rest.split(']').next().unwrap_or(rest);
816 }
817 match host.rsplit_once(':') {
818 Some((h, port)) if !port.is_empty() && port.bytes().all(|b| b.is_ascii_digit()) => h,
819 _ => host,
820 }
821}
822
823fn header_str(request: &axum::extract::Request, name: axum::http::HeaderName) -> Option<&str> {
824 request.headers().get(name).and_then(|v| v.to_str().ok())
825}
826
827async fn host_guard(
829 axum::extract::State(policy): axum::extract::State<Arc<BrowserPolicy>>,
830 request: axum::extract::Request,
831 next: axum::middleware::Next,
832) -> axum::response::Response {
833 use axum::response::IntoResponse;
834
835 let host = header_str(&request, axum::http::header::HOST)
838 .map(str::to_owned)
839 .or_else(|| request.uri().host().map(str::to_owned));
840
841 if let Some(ref host) = host
842 && !policy.host_allowed(host)
843 {
844 log::warn!("rejected request for unrecognised Host: {}", host);
845 return (axum::http::StatusCode::FORBIDDEN, "host not allowed").into_response();
846 }
847
848 next.run(request).await
849}
850
851async fn browser_guard(
860 axum::extract::State(policy): axum::extract::State<Arc<BrowserPolicy>>,
861 request: axum::extract::Request,
862 next: axum::middleware::Next,
863) -> axum::response::Response {
864 use axum::response::IntoResponse;
865
866 let host = header_str(&request, axum::http::header::HOST).map(str::to_owned);
867 if let Some(origin) = header_str(&request, axum::http::header::ORIGIN)
869 && !policy.origin_allowed(origin, host.as_deref())
870 {
871 log::warn!(
872 "rejected GraphQL request from disallowed Origin: {}",
873 origin
874 );
875 return (axum::http::StatusCode::FORBIDDEN, "origin not allowed").into_response();
876 }
877
878 if request.method() == axum::http::Method::POST && !is_graphql_content_type(&request) {
879 return (
880 axum::http::StatusCode::UNSUPPORTED_MEDIA_TYPE,
881 "content type must be application/json or application/graphql",
882 )
883 .into_response();
884 }
885
886 next.run(request).await
887}
888
889fn is_graphql_content_type(request: &axum::extract::Request) -> bool {
890 header_str(request, axum::http::header::CONTENT_TYPE).is_some_and(|ct| {
891 let ct = ct.trim().to_ascii_lowercase();
892 ct.starts_with("application/json") || ct.starts_with("application/graphql")
893 })
894}
895
896static OUTDATED: std::sync::LazyLock<tokio_util::sync::CancellationToken> =
900 std::sync::LazyLock::new(tokio_util::sync::CancellationToken::new);
901
902async fn shutdown_signal() {
907 #[cfg(unix)]
908 {
909 use tokio::signal::unix::{SignalKind, signal};
910 let mut terminate = signal(SignalKind::terminate()).expect("failed to listen for SIGTERM");
911 tokio::select! {
912 r = tokio::signal::ctrl_c() => r.expect("failed to listen for ctrl+c"),
913 _ = terminate.recv() => {}
914 _ = OUTDATED.cancelled() => {}
915 }
916 }
917 #[cfg(not(unix))]
918 tokio::select! {
919 r = tokio::signal::ctrl_c() => r.expect("failed to listen for ctrl+c"),
920 _ = OUTDATED.cancelled() => {}
921 }
922}
923
924const DRAIN: std::time::Duration = std::time::Duration::from_secs(10);
926
927async fn drain_deadline(stop: &tokio_util::sync::CancellationToken) {
931 stop.cancelled().await;
932 tokio::time::sleep(DRAIN).await;
933}
934
935async fn graphql_handler(
936 axum::Extension(user): axum::Extension<AuthUser>,
937 axum::extract::State(schema): axum::extract::State<KoanSchema>,
938 req: async_graphql_axum::GraphQLRequest,
939) -> async_graphql_axum::GraphQLResponse {
940 let mut request = req.into_inner();
941 request = request.data(user);
944 schema.execute(request).await.into()
945}
946
947async fn graphql_ws_handler(
952 axum::Extension(user): axum::Extension<AuthUser>,
953 lease: Option<axum::Extension<crate::auth::Lease>>,
954 axum::Extension(stop): axum::Extension<tokio_util::sync::CancellationToken>,
955 axum::extract::State(schema): axum::extract::State<KoanSchema>,
956 protocol: async_graphql_axum::GraphQLProtocol,
957 websocket: axum::extract::WebSocketUpgrade,
958) -> axum::response::Response {
959 use axum::extract::ws::{CloseFrame, Message, close_code};
960 use futures_util::{SinkExt, StreamExt};
961 websocket
962 .protocols(async_graphql::http::ALL_WEBSOCKET_PROTOCOLS)
963 .on_upgrade(move |socket| async move {
964 let (mut sink, stream) = socket.split();
965 let serve = async_graphql_axum::GraphQLWebSocket::new_with_pair(
966 &mut sink, stream, schema, protocol,
967 )
968 .on_connection_init(move |_| async move {
969 let mut data = async_graphql::Data::default();
970 data.insert(user);
971 Ok(data)
972 })
973 .serve();
974 let ended = async {
975 match lease {
976 Some(axum::Extension(lease)) => lease.ended().await,
977 None => std::future::pending().await,
978 }
979 };
980 let close = tokio::select! {
981 _ = serve => None,
982 _ = ended => Some((close_code::NORMAL, "sign in again")),
983 _ = stop.cancelled() => Some((close_code::AWAY, "server stopping")),
984 };
985 if let Some((code, reason)) = close {
986 let _ = sink
987 .send(Message::Close(Some(CloseFrame {
988 code,
989 reason: reason.into(),
990 })))
991 .await;
992 }
993 })
994}
995
996async fn graphql_playground(
997 axum::extract::Query(params): axum::extract::Query<std::collections::HashMap<String, String>>,
998 axum::extract::State(key): axum::extract::State<Option<Arc<String>>>,
999) -> axum::response::Response {
1000 use axum::response::IntoResponse;
1001
1002 if let Some(ref expected) = key {
1004 let provided = params.get("introspection-key");
1005 if provided.map(|k| k.as_str()) != Some(expected.as_str()) {
1006 return (
1007 axum::http::StatusCode::FORBIDDEN,
1008 "invalid or missing introspection-key",
1009 )
1010 .into_response();
1011 }
1012 }
1013
1014 let mut source = async_graphql::http::GraphiQLSource::build().endpoint("/graphql");
1017 if let Some(ref k) = key {
1018 source = source.header("X-Introspection-Key", k.as_str());
1019 }
1020
1021 axum::response::Html(source.finish()).into_response()
1022}
1023
1024pub fn cmd_serve_daemon(
1026 port: Option<u16>,
1027 bind: Option<std::net::IpAddr>,
1028 subsonic_port: Option<u16>,
1029 playground: bool,
1030) {
1031 use std::fs;
1032 use std::process::Command;
1033
1034 require_server_url();
1035
1036 let cfg = Config::cached();
1037 let port_val = port.unwrap_or(cfg.graphql.port);
1038 let bind_val = bind.unwrap_or(cfg.graphql.bind);
1039
1040 let exe = std::env::current_exe().expect("failed to get current exe path");
1041 let mut cmd = Command::new(exe);
1042
1043 cmd.arg("--headless");
1044 cmd.arg("--port").arg(port_val.to_string());
1045 cmd.arg("--bind").arg(bind_val.to_string());
1046 if let Some(sp) = subsonic_port {
1047 cmd.arg("--subsonic").arg(sp.to_string());
1048 }
1049 if playground || cfg.graphql.playground {
1050 cmd.arg("--playground");
1051 }
1052
1053 cmd.stdin(std::process::Stdio::null());
1054 cmd.stdout(std::process::Stdio::null());
1055 cmd.stderr(std::process::Stdio::null());
1056
1057 let mut child = cmd.spawn().expect("failed to spawn daemon process");
1058 let pid = child.id();
1059
1060 let pid_path = koan_core::config::config_dir().join("koan-serve.pid");
1061 fs::write(&pid_path, pid.to_string()).ok();
1062
1063 std::thread::spawn(move || {
1064 let _ = child.wait();
1065 });
1066
1067 eprintln!("kōan daemon started (pid {}) on port {}", pid, port_val);
1068 if let Some(sp) = subsonic_port {
1069 eprintln!(" Subsonic REST on port {}", sp);
1070 }
1071 eprintln!(" PID file: {}", pid_path.display());
1072}
1073
1074pub async fn execute_in_process(
1083 schema: &KoanSchema,
1084 query: &str,
1085 variables: Option<serde_json::Value>,
1086 caller: AuthUser,
1087) -> serde_json::Value {
1088 let mut request = async_graphql::Request::new(query).data(caller);
1089 if let Some(serde_json::Value::Object(map)) = variables {
1090 let mut gql_vars = async_graphql::Variables::default();
1091 for (k, v) in map {
1092 gql_vars.insert(
1093 async_graphql::Name::new(&k),
1094 async_graphql::Value::from_json(v).unwrap_or(async_graphql::Value::Null),
1095 );
1096 }
1097 request = request.variables(gql_vars);
1098 }
1099 let response = schema.execute(request).await;
1100 serde_json::to_value(&response).unwrap_or(serde_json::Value::Null)
1101}
1102
1103#[cfg(test)]
1108mod tests {
1109 use super::*;
1110 use axum::body::Body;
1111 use axum::http::{Request as HttpRequest, StatusCode};
1112 use axum::routing::{get, post};
1113 use tower::ServiceExt as _;
1114
1115 fn policy() -> Arc<BrowserPolicy> {
1116 Arc::new(BrowserPolicy {
1117 origins: vec!["https://music.example.com".into()],
1118 hosts: vec!["koan.local".into()],
1119 })
1120 }
1121
1122 async fn ok() -> &'static str {
1123 "ok"
1124 }
1125
1126 fn routes() -> axum::Router<Arc<BrowserPolicy>> {
1127 axum::Router::new()
1128 .route("/graphql", post(ok).get(ok))
1129 .route("/graphql/ws", get(ok))
1130 }
1131
1132 async fn run_host(req: HttpRequest<Body>) -> StatusCode {
1133 let app = routes()
1134 .layer(axum::middleware::from_fn_with_state(policy(), host_guard))
1135 .with_state(policy());
1136 app.oneshot(req).await.unwrap().status()
1137 }
1138
1139 async fn run_browser(req: HttpRequest<Body>) -> StatusCode {
1140 let app = routes()
1141 .layer(axum::middleware::from_fn_with_state(
1142 policy(),
1143 browser_guard,
1144 ))
1145 .with_state(policy());
1146 app.oneshot(req).await.unwrap().status()
1147 }
1148
1149 fn json_post(uri: &str) -> axum::http::request::Builder {
1150 HttpRequest::post(uri).header(axum::http::header::CONTENT_TYPE, "application/json")
1151 }
1152
1153 #[test]
1156 fn host_policy_accepts_loopback_literals_and_configured_names() {
1157 let p = policy();
1158 assert!(p.host_allowed("localhost:4000"));
1159 assert!(p.host_allowed("127.0.0.1:4000"));
1160 assert!(p.host_allowed("192.168.1.20:4000"));
1161 assert!(p.host_allowed("[::1]:4000"));
1162 assert!(p.host_allowed("koan.local"));
1163 assert!(p.host_allowed("koan.local:4000"));
1164 }
1165
1166 #[test]
1167 fn host_policy_rejects_attacker_controlled_names() {
1168 let p = policy();
1169 assert!(!p.host_allowed("evil.com"));
1170 assert!(!p.host_allowed("rebind.evil.com:4000"));
1171 assert!(!p.host_allowed("koan.local.evil.com"));
1172 }
1173
1174 #[tokio::test]
1175 async fn host_guard_rejects_foreign_host() {
1176 let req = json_post("/graphql")
1177 .header(axum::http::header::HOST, "rebind.evil.com")
1178 .body(Body::empty())
1179 .unwrap();
1180 assert_eq!(run_host(req).await, StatusCode::FORBIDDEN);
1181 }
1182
1183 #[tokio::test]
1184 async fn host_guard_allows_known_host_and_missing_host() {
1185 let req = json_post("/graphql")
1186 .header(axum::http::header::HOST, "127.0.0.1:4000")
1187 .body(Body::empty())
1188 .unwrap();
1189 assert_eq!(run_host(req).await, StatusCode::OK);
1190
1191 let req = json_post("/graphql").body(Body::empty()).unwrap();
1192 assert_eq!(run_host(req).await, StatusCode::OK);
1193 }
1194
1195 #[tokio::test]
1198 async fn ws_upgrade_from_foreign_origin_is_rejected() {
1199 let req = HttpRequest::get("/graphql/ws")
1200 .header(axum::http::header::HOST, "127.0.0.1:4000")
1201 .header(axum::http::header::ORIGIN, "https://evil.com")
1202 .body(Body::empty())
1203 .unwrap();
1204 assert_eq!(run_browser(req).await, StatusCode::FORBIDDEN);
1205 }
1206
1207 #[tokio::test]
1208 async fn ws_upgrade_without_origin_is_allowed() {
1209 let req = HttpRequest::get("/graphql/ws")
1210 .header(axum::http::header::HOST, "127.0.0.1:4000")
1211 .body(Body::empty())
1212 .unwrap();
1213 assert_eq!(run_browser(req).await, StatusCode::OK);
1214 }
1215
1216 #[tokio::test]
1217 async fn configured_and_same_origin_are_allowed() {
1218 let req = HttpRequest::get("/graphql/ws")
1219 .header(axum::http::header::HOST, "127.0.0.1:4000")
1220 .header(axum::http::header::ORIGIN, "https://music.example.com")
1221 .body(Body::empty())
1222 .unwrap();
1223 assert_eq!(run_browser(req).await, StatusCode::OK);
1224
1225 let req = json_post("/graphql")
1227 .header(axum::http::header::HOST, "127.0.0.1:4000")
1228 .header(axum::http::header::ORIGIN, "http://127.0.0.1:4000")
1229 .body(Body::empty())
1230 .unwrap();
1231 assert_eq!(run_browser(req).await, StatusCode::OK);
1232 }
1233
1234 #[tokio::test]
1237 async fn text_plain_post_is_rejected() {
1238 let req = HttpRequest::post("/graphql")
1239 .header(axum::http::header::CONTENT_TYPE, "text/plain")
1240 .body(Body::from(r#"{"query":"mutation{clearQueue{ok}}"}"#))
1241 .unwrap();
1242 assert_eq!(run_browser(req).await, StatusCode::UNSUPPORTED_MEDIA_TYPE);
1243 }
1244
1245 #[tokio::test]
1246 async fn post_without_content_type_is_rejected() {
1247 let req = HttpRequest::post("/graphql").body(Body::empty()).unwrap();
1248 assert_eq!(run_browser(req).await, StatusCode::UNSUPPORTED_MEDIA_TYPE);
1249 }
1250
1251 #[tokio::test]
1254 async fn load_perimeter_refuses_an_oversized_query_body() {
1255 async fn parse(_: async_graphql_axum::GraphQLRequest) -> StatusCode {
1256 StatusCode::OK
1257 }
1258 let app = load_perimeter(axum::Router::new().route("/graphql", post(parse)));
1259 let body = |padding: usize| {
1262 let chunks = [
1263 axum::body::Bytes::from_static(br#"{"query":"{__typename}""#),
1264 axum::body::Bytes::from(vec![b' '; padding]),
1265 axum::body::Bytes::from_static(b"}"),
1266 ];
1267 Body::from_stream(tokio_stream::iter(chunks.map(Ok::<_, std::io::Error>)))
1268 };
1269 let req = json_post("/graphql").body(body(1 << 10)).unwrap();
1270 assert_eq!(
1271 app.clone().oneshot(req).await.unwrap().status(),
1272 StatusCode::OK
1273 );
1274 let req = json_post("/graphql").body(body(3 << 20)).unwrap();
1275 assert_ne!(app.oneshot(req).await.unwrap().status(), StatusCode::OK);
1276 }
1277
1278 #[tokio::test]
1279 async fn load_perimeter_passes_requests_and_turns_panics_into_500s() {
1280 async fn boom() -> &'static str {
1281 panic!("resolver exploded");
1282 }
1283
1284 let app = load_perimeter(
1285 axum::Router::new()
1286 .route("/graphql", post(ok))
1287 .route("/boom", post(boom)),
1288 );
1289
1290 let req = json_post("/graphql").body(Body::empty()).unwrap();
1291 assert_eq!(
1292 app.clone().oneshot(req).await.unwrap().status(),
1293 StatusCode::OK
1294 );
1295
1296 let req = json_post("/boom").body(Body::empty()).unwrap();
1298 assert_eq!(
1299 app.oneshot(req).await.unwrap().status(),
1300 StatusCode::INTERNAL_SERVER_ERROR
1301 );
1302 }
1303
1304 #[tokio::test]
1305 async fn json_post_is_accepted() {
1306 let req = json_post("/graphql").body(Body::empty()).unwrap();
1307 assert_eq!(run_browser(req).await, StatusCode::OK);
1308
1309 let req = HttpRequest::post("/graphql")
1310 .header(
1311 axum::http::header::CONTENT_TYPE,
1312 "application/json; charset=utf-8",
1313 )
1314 .body(Body::empty())
1315 .unwrap();
1316 assert_eq!(run_browser(req).await, StatusCode::OK);
1317 }
1318
1319 #[test]
1323 fn an_embedded_server_ends_its_links_on_stop_and_starts_again() {
1324 use koan_core::auth::Role;
1325 use koan_core::db::connection::Database;
1326 use koan_core::player::commands::CommandChannel;
1327
1328 let _own = koan_core::config::SWITCHING
1329 .lock()
1330 .unwrap_or_else(|e| e.into_inner());
1331 let dir = tempfile::tempdir().unwrap();
1332 koan_core::config::set_config_dir(dir.path());
1333 let db_path = dir.path().join("koan.db");
1334 let db = Database::open(&db_path).unwrap();
1335 koan_core::db::schema::create_tables(&db.conn).unwrap();
1336 koan_core::db::queries::auth::create_user(&db.conn, "sharer", "hunter22", Role::Admin)
1337 .unwrap();
1338
1339 let start = || {
1340 ApiServer::start(ApiServerOpts {
1341 state: SharedPlayerState::new(),
1342 cmd_tx: CommandChannel::new().tx,
1343 pool: Arc::new(Pool::new(db_path.clone())),
1344 port: Some(0),
1345 bind: Some(std::net::Ipv4Addr::LOCALHOST.into()),
1346 subsonic_port: None,
1347 playground: false,
1348 viz: None,
1349 profile: Profile::Embedded,
1350 hosts: Vec::new(),
1351 })
1352 .unwrap()
1353 };
1354 let ping = |addr: std::net::SocketAddr| {
1356 let body = reqwest::blocking::get(format!(
1357 "http://{addr}/rest/ping?u=sharer&p=hunter22&v=1.16.1&c=test&f=json"
1358 ))
1359 .unwrap()
1360 .text()
1361 .unwrap();
1362 assert!(body.contains(r#""status":"ok""#), "{body}");
1363 };
1364 let linked = || {
1365 crate::clients::registry()
1366 .list(Some("sharer"))
1367 .iter()
1368 .any(|c| c.device == "embedded-restart")
1369 };
1370
1371 let server = start();
1372 let addr = server.addr();
1373 ping(addr);
1374 let (mut socket, _) = tungstenite::connect(format!(
1375 "ws://{addr}/rest/koanLink?u=sharer&p=hunter22&v=1.16.1&c=test&device=embedded-restart"
1376 ))
1377 .unwrap();
1378 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
1379 while !linked() {
1380 assert!(
1381 std::time::Instant::now() < deadline,
1382 "the device never linked"
1383 );
1384 std::thread::sleep(std::time::Duration::from_millis(20));
1385 }
1386
1387 let stopping = std::time::Instant::now();
1388 server.stop();
1389 assert!(stopping.elapsed() < DRAIN, "stop waited out the drain");
1390 assert!(!linked(), "the link outlived the server");
1391 let closed = loop {
1393 match socket.read() {
1394 Ok(tungstenite::Message::Close(frame)) => break frame,
1395 Ok(_) => continue,
1396 Err(e) => panic!("no close frame: {e}"),
1397 }
1398 };
1399 assert_eq!(
1400 closed.map(|f| f.code),
1401 Some(tungstenite::protocol::frame::coding::CloseCode::Away)
1402 );
1403
1404 let again = start();
1405 ping(again.addr());
1406 again.stop();
1407 }
1408
1409 #[test]
1412 fn an_embedded_server_refuses_a_port_held_on_loopback() {
1413 use koan_core::player::commands::CommandChannel;
1414
1415 let _own = koan_core::config::SWITCHING
1416 .lock()
1417 .unwrap_or_else(|e| e.into_inner());
1418 let dir = tempfile::tempdir().unwrap();
1419 koan_core::config::set_config_dir(dir.path());
1420 let held = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
1421 let port = held.local_addr().unwrap().port();
1422 let started = ApiServer::start(ApiServerOpts {
1423 state: SharedPlayerState::new(),
1424 cmd_tx: CommandChannel::new().tx,
1425 pool: Arc::new(Pool::new(dir.path().join("koan.db"))),
1426 port: Some(port),
1427 bind: Some(std::net::Ipv4Addr::UNSPECIFIED.into()),
1428 subsonic_port: None,
1429 playground: false,
1430 viz: None,
1431 profile: Profile::Embedded,
1432 hosts: Vec::new(),
1433 });
1434 let Err(why) = started else {
1435 panic!("started beside another listener on port {port}");
1436 };
1437 assert!(why.contains("already in use"), "{why}");
1438 }
1439}