1use std::net::SocketAddr;
35use std::path::PathBuf;
36use std::sync::Arc;
37
38use axum::body::{Body, Bytes};
39use axum::extract::{ws::WebSocketUpgrade, Path as AxumPath, State};
40use axum::http::{header, HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Uri};
41use axum::response::{IntoResponse, Response};
42use axum::routing::{any, post};
43use axum::Router;
44use base64::Engine as _;
45use clap::ValueEnum;
46use futures_util::{SinkExt, StreamExt};
47use serde::Deserialize;
48use tokio::sync::RwLock;
49use tower_http::services::ServeDir;
50
51use crate::client::CellosClient;
52use crate::config::Config;
53use crate::exit::{CtlError, CtlResult};
54
55const BUNDLE_DIR_RELATIVE: &str = "static";
60
61#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, ValueEnum)]
63pub enum BindMode {
64 #[default]
67 Auto,
68 Loopback,
70 Unix,
74}
75
76#[derive(Clone)]
86struct AppState {
87 upstream_base: Arc<String>,
89 upstream_bearer: Arc<Option<String>>,
91 session_token: Arc<String>,
93 session_cookie: Arc<RwLock<Option<String>>>,
96 exchange_consumed: Arc<std::sync::atomic::AtomicBool>,
101 bundle_dir: Arc<PathBuf>,
103}
104
105#[derive(Deserialize)]
106struct ExchangeRequest {
107 sess: String,
108}
109
110pub async fn run(cfg: &Config, open: bool, bind: BindMode) -> CtlResult<()> {
112 let _ = CellosClient::new(cfg)?;
117
118 let upstream_base = cfg.effective_server().trim_end_matches('/').to_string();
119 let upstream_bearer = cfg.effective_token();
120
121 let bundle_dir = resolve_bundle_dir()?;
122 if !bundle_dir.join("index.html").exists() {
123 return Err(CtlError::usage(format!(
124 "webui bundle not found at {}/index.html — run `npm --prefix web run build` first",
125 bundle_dir.display()
126 )));
127 }
128
129 let session_token = mint_session_token();
130
131 let upstream_log = upstream_base.clone();
132 let state = AppState {
133 upstream_base: Arc::new(upstream_base),
134 upstream_bearer: Arc::new(upstream_bearer),
135 session_token: Arc::new(session_token.clone()),
136 session_cookie: Arc::new(RwLock::new(None)),
137 exchange_consumed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
138 bundle_dir: Arc::new(bundle_dir.clone()),
139 };
140
141 let app = build_router(state);
142
143 let (want_tcp, _want_unix) = resolve_bind_plan(bind)?;
145
146 let tcp_listener = if want_tcp {
147 let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
148 Some(
149 tokio::net::TcpListener::bind(addr)
150 .await
151 .map_err(|e| CtlError::usage(format!("bind 127.0.0.1: {e}")))?,
152 )
153 } else {
154 None
155 };
156
157 #[cfg(unix)]
158 let unix_socket_path: Option<PathBuf> = if _want_unix {
159 Some(unix_socket_path_for_pid(std::process::id()))
160 } else {
161 None
162 };
163 #[cfg(not(unix))]
164 let unix_socket_path: Option<PathBuf> = None;
165
166 let browser_url = if let Some(l) = tcp_listener.as_ref() {
167 let local_addr = l
168 .local_addr()
169 .map_err(|e| CtlError::usage(format!("local_addr: {e}")))?;
170 Some(format!("http://{}/#sess={}", local_addr, session_token))
171 } else {
172 None
173 };
174
175 if let Some(url) = browser_url.as_ref() {
176 println!("cellctl webui: {}", url);
177 }
178 if let Some(p) = unix_socket_path.as_ref() {
179 println!("cellctl webui: unix://{}", p.display());
180 }
181 if browser_url.is_none() {
182 eprintln!(
184 "cellctl webui: --bind unix has no browser-reachable URL; \
185 use a forwarder (e.g. `socat TCP-LISTEN:0,reuseaddr,fork UNIX-CONNECT:{}`) \
186 or rerun with `--bind auto` for a loopback URL.",
187 unix_socket_path
188 .as_ref()
189 .map(|p| p.display().to_string())
190 .unwrap_or_else(|| "<socket>".to_string())
191 );
192 }
193 println!("upstream: {}", upstream_log);
197 println!("press Ctrl-C to stop");
198
199 if open {
200 if let Some(url) = browser_url.as_ref() {
201 if !is_safe_open_url(url) {
209 eprintln!(
210 "cellctl webui: refusing to open URL (failed loopback-http sanity check)"
211 );
212 } else if let Err(e) = opener::open(url) {
213 eprintln!("cellctl webui: could not launch browser: {e}");
214 }
215 } else {
216 eprintln!("cellctl webui: --open ignored: no loopback URL bound (use --bind auto)");
217 }
218 }
219
220 let (shutdown_tx, _shutdown_rx) = tokio::sync::broadcast::channel::<()>(1);
227 {
228 let shutdown_tx = shutdown_tx.clone();
229 tokio::spawn(async move {
230 wait_for_shutdown_signal().await;
231 eprintln!("shutting down");
232 let _ = shutdown_tx.send(());
233 });
234 }
235
236 let tcp_task = if let Some(listener) = tcp_listener {
238 let app = app.clone();
239 let mut rx = shutdown_tx.subscribe();
240 Some(tokio::spawn(async move {
241 axum::serve(listener, app)
242 .with_graceful_shutdown(async move {
243 let _ = rx.recv().await;
244 })
245 .await
246 }))
247 } else {
248 None
249 };
250
251 #[cfg(unix)]
253 let unix_task = if let Some(path) = unix_socket_path.clone() {
254 let app = app.clone();
255 let mut rx = shutdown_tx.subscribe();
256 let listener = bind_unix_listener(&path)?;
257 Some(tokio::spawn(async move {
258 serve_unix(listener, app, async move {
259 let _ = rx.recv().await;
260 })
261 .await
262 }))
263 } else {
264 None
265 };
266 #[cfg(not(unix))]
267 let unix_task: Option<tokio::task::JoinHandle<std::io::Result<()>>> = None;
268
269 let mut first_err: Option<String> = None;
271 if let Some(t) = tcp_task {
272 match t.await {
273 Ok(Ok(())) => {}
274 Ok(Err(e)) => {
275 let _ = first_err.get_or_insert_with(|| format!("tcp: {e}"));
276 }
277 Err(e) => {
278 let _ = first_err.get_or_insert_with(|| format!("tcp join: {e}"));
279 }
280 }
281 }
282 if let Some(t) = unix_task {
283 match t.await {
284 Ok(Ok(())) => {}
285 Ok(Err(e)) => {
286 let _ = first_err.get_or_insert_with(|| format!("unix: {e}"));
287 }
288 Err(e) => {
289 let _ = first_err.get_or_insert_with(|| format!("unix join: {e}"));
290 }
291 }
292 }
293
294 if let Some(p) = unix_socket_path.as_ref() {
297 let _ = std::fs::remove_file(p);
298 }
299
300 if let Some(e) = first_err {
301 return Err(CtlError::api(format!("webui server: {e}")));
302 }
303 Ok(())
304}
305
306fn resolve_bind_plan(bind: BindMode) -> CtlResult<(bool, bool)> {
311 #[cfg(unix)]
312 {
313 Ok(match bind {
314 BindMode::Auto => (true, true),
315 BindMode::Loopback => (true, false),
316 BindMode::Unix => (false, true),
317 })
318 }
319 #[cfg(not(unix))]
320 {
321 Ok(match bind {
322 BindMode::Auto => (true, false),
323 BindMode::Loopback => (true, false),
324 BindMode::Unix => {
325 return Err(CtlError::usage(
326 "--bind unix is not supported on Windows; use --bind loopback (the default)",
327 ));
328 }
329 })
330 }
331}
332
333#[cfg(unix)]
343fn unix_socket_path_for_pid(pid: u32) -> PathBuf {
344 let dir = if let Some(xdg) = std::env::var_os("XDG_RUNTIME_DIR") {
345 PathBuf::from(xdg)
346 } else {
347 let uid = unsafe { libc::getuid() };
350 PathBuf::from("/tmp").join(format!("cellctl-{uid}"))
351 };
352 dir.join(format!("cellctl-webui-{pid}.sock"))
353}
354
355#[cfg(unix)]
374fn bind_unix_listener(path: &std::path::Path) -> CtlResult<tokio::net::UnixListener> {
375 use std::os::unix::fs::PermissionsExt;
376
377 if let Some(parent) = path.parent() {
379 if !parent.exists() {
380 std::fs::create_dir_all(parent)
381 .map_err(|e| CtlError::usage(format!("mkdir {}: {e}", parent.display())))?;
382 }
383 let _ = std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700));
387 }
388
389 let _ = std::fs::remove_file(path);
391
392 let prev_umask = unsafe { libc::umask(0o077) };
398 let bind_result = tokio::net::UnixListener::bind(path);
399 unsafe {
400 let _ = libc::umask(prev_umask);
401 }
402 let listener =
403 bind_result.map_err(|e| CtlError::usage(format!("bind {}: {e}", path.display())))?;
404
405 let perms = std::fs::Permissions::from_mode(0o600);
410 std::fs::set_permissions(path, perms)
411 .map_err(|e| CtlError::usage(format!("chmod 0600 {}: {e}", path.display())))?;
412
413 Ok(listener)
414}
415
416async fn wait_for_shutdown_signal() {
419 #[cfg(unix)]
420 {
421 use tokio::signal::unix::{signal, SignalKind};
422 let mut sigint = match signal(SignalKind::interrupt()) {
423 Ok(s) => s,
424 Err(_) => {
425 let _ = tokio::signal::ctrl_c().await;
426 return;
427 }
428 };
429 let mut sigterm = match signal(SignalKind::terminate()) {
430 Ok(s) => s,
431 Err(_) => {
432 let _ = sigint.recv().await;
433 return;
434 }
435 };
436 tokio::select! {
437 _ = sigint.recv() => {},
438 _ = sigterm.recv() => {},
439 }
440 }
441 #[cfg(not(unix))]
442 {
443 let _ = tokio::signal::ctrl_c().await;
444 }
445}
446
447fn is_safe_open_url(s: &str) -> bool {
457 if s.chars().any(|c| c.is_control()) {
460 return false;
461 }
462 let Ok(u) = url::Url::parse(s) else {
463 return false;
464 };
465 let scheme = u.scheme();
466 if scheme != "http" && scheme != "https" {
467 return false;
468 }
469 let Some(host) = u.host_str() else {
470 return false;
471 };
472 matches!(host, "127.0.0.1" | "localhost" | "::1" | "[::1]")
475}
476
477#[cfg(unix)]
483async fn serve_unix(
484 listener: tokio::net::UnixListener,
485 app: Router,
486 shutdown: impl std::future::Future<Output = ()>,
487) -> std::io::Result<()> {
488 use std::convert::Infallible;
489 use tower::Service;
490
491 tokio::pin!(shutdown);
496
497 loop {
498 tokio::select! {
499 _ = &mut shutdown => return Ok(()),
500 accepted = listener.accept() => {
501 let (stream, _peer) = match accepted {
502 Ok(s) => s,
503 Err(e) => {
504 eprintln!("webui: unix accept error: {e}");
506 continue;
507 }
508 };
509 let app = app.clone();
510 tokio::spawn(async move {
511 let svc = hyper::service::service_fn(move |req: http::Request<hyper::body::Incoming>| {
515 let mut router = app.clone();
516 async move {
517 let resp: Response = match router.call(req.map(Body::new)).await {
518 Ok(r) => r,
519 Err(never) => match never {},
520 };
521 Ok::<_, Infallible>(resp)
522 }
523 });
524 let io = hyper_util::rt::TokioIo::new(stream);
525 let _ = hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new())
526 .serve_connection_with_upgrades(io, svc)
527 .await;
528 });
529 }
530 }
531 }
532}
533
534fn mint_session_token() -> String {
536 let buf: [u8; 32] = rand::random();
540 base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(buf)
541}
542
543fn resolve_bundle_dir() -> CtlResult<PathBuf> {
547 if let Ok(p) = std::env::var("CELLCTL_WEBUI_BUNDLE_DIR") {
548 return Ok(PathBuf::from(p));
549 }
550 let from_manifest = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(BUNDLE_DIR_RELATIVE);
553 Ok(from_manifest)
554}
555
556fn build_router(state: AppState) -> Router {
559 let bundle_dir = state.bundle_dir.as_ref().clone();
562 let serve_dir = ServeDir::new(&bundle_dir).append_index_html_on_directories(true);
563
564 Router::new()
565 .route("/auth/exchange", post(auth_exchange))
566 .route("/v1/{*rest}", any(proxy_v1))
567 .route("/ws/events", any(ws_events))
568 .fallback_service(serve_dir)
572 .layer(axum::middleware::from_fn(reject_non_get))
573 .with_state(state)
574}
575
576async fn reject_non_get(req: axum::http::Request<Body>, next: axum::middleware::Next) -> Response {
580 let method = req.method().clone();
581 let path = req.uri().path().to_string();
582
583 let is_auth_exchange = method == Method::POST && path == "/auth/exchange";
587 if method != Method::GET && !is_auth_exchange {
588 return method_not_allowed();
589 }
590 next.run(req).await
591}
592
593fn method_not_allowed() -> Response {
594 let mut resp = (StatusCode::METHOD_NOT_ALLOWED, "method not allowed\n").into_response();
595 resp.headers_mut()
596 .insert(header::ALLOW, HeaderValue::from_static("GET"));
597 resp
598}
599
600const COOKIE_MAX_AGE_SECS: u64 = 86_400;
604
605async fn auth_exchange(State(state): State<AppState>, headers: HeaderMap, body: Bytes) -> Response {
625 use std::sync::atomic::Ordering;
626 use subtle::ConstantTimeEq;
627
628 let ct_ok = headers
632 .get(header::CONTENT_TYPE)
633 .and_then(|v| v.to_str().ok())
634 .map(|s| {
635 let trimmed = s.split(';').next().unwrap_or("").trim();
637 trimmed.eq_ignore_ascii_case("application/json")
638 })
639 .unwrap_or(false);
640 if !ct_ok {
641 let mut resp = (
642 StatusCode::UNSUPPORTED_MEDIA_TYPE,
643 "Content-Type must be application/json\n",
644 )
645 .into_response();
646 resp.headers_mut().insert(
647 header::CONTENT_TYPE,
648 HeaderValue::from_static("text/plain; charset=utf-8"),
649 );
650 return resp;
651 }
652
653 let parsed: ExchangeRequest = match serde_json::from_slice(&body) {
656 Ok(p) => p,
657 Err(_) => return (StatusCode::BAD_REQUEST, "invalid json body\n").into_response(),
658 };
659
660 if state.exchange_consumed.load(Ordering::SeqCst) {
663 return (
664 StatusCode::UNAUTHORIZED,
665 "session token already exchanged\n",
666 )
667 .into_response();
668 }
669
670 let provided = parsed.sess.as_bytes();
675 let expected = state.session_token.as_bytes();
676 if provided.len() != expected.len() || provided.ct_eq(expected).unwrap_u8() == 0 {
677 return (StatusCode::UNAUTHORIZED, "bad session token\n").into_response();
678 }
679
680 if state
685 .exchange_consumed
686 .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
687 .is_err()
688 {
689 return (
690 StatusCode::UNAUTHORIZED,
691 "session token already exchanged\n",
692 )
693 .into_response();
694 }
695
696 let buf: [u8; 32] = rand::random();
700 let cookie_value = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(buf);
701
702 {
703 let mut slot = state.session_cookie.write().await;
704 *slot = Some(cookie_value.clone());
705 }
706
707 let cookie_header = format!(
710 "cellctl_session={}; HttpOnly; SameSite=Strict; Path=/; Max-Age={}",
711 cookie_value, COOKIE_MAX_AGE_SECS
712 );
713
714 let mut resp = (StatusCode::OK, "ok\n").into_response();
715 resp.headers_mut().insert(
716 header::SET_COOKIE,
717 HeaderValue::from_str(&cookie_header).unwrap(),
718 );
719 resp
720}
721
722async fn require_session_cookie(state: &AppState, headers: &HeaderMap) -> bool {
725 let expected = match state.session_cookie.read().await.clone() {
726 Some(v) => v,
727 None => return false,
728 };
729 let Some(cookie_hdr) = headers.get(header::COOKIE) else {
730 return false;
731 };
732 let Ok(cookie_str) = cookie_hdr.to_str() else {
733 return false;
734 };
735 for entry in cookie_str.split(';') {
737 let entry = entry.trim();
738 if let Some(v) = entry.strip_prefix("cellctl_session=") {
739 return v == expected;
740 }
741 }
742 false
743}
744
745async fn proxy_v1(
747 State(state): State<AppState>,
748 AxumPath(rest): AxumPath<String>,
749 uri: Uri,
750 headers: HeaderMap,
751) -> Response {
752 if !require_session_cookie(&state, &headers).await {
753 return unauthorized();
754 }
755
756 let query = uri.query().map(|q| format!("?{q}")).unwrap_or_default();
757 let upstream_url = format!("{}/v1/{}{}", state.upstream_base, rest, query);
758
759 let client = match reqwest::Client::builder().build() {
760 Ok(c) => c,
761 Err(e) => return upstream_error(format!("client: {e}")),
762 };
763
764 let mut req = client.get(&upstream_url);
765 if let Some(tok) = state.upstream_bearer.as_ref() {
766 req = req.header(reqwest::header::AUTHORIZATION, format!("Bearer {tok}"));
767 }
768
769 let upstream_resp = match req.send().await {
770 Ok(r) => r,
771 Err(e) => return upstream_error(format!("send: {e}")),
772 };
773
774 let status = upstream_resp.status();
775 let resp_headers = upstream_resp.headers().clone();
776 let body_bytes = match upstream_resp.bytes().await {
777 Ok(b) => b,
778 Err(e) => return upstream_error(format!("read body: {e}")),
779 };
780
781 let mut out = Response::new(Body::from(body_bytes));
782 *out.status_mut() = StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY);
783 if let Some(ct) = resp_headers.get(reqwest::header::CONTENT_TYPE) {
785 if let Ok(v) = HeaderValue::from_bytes(ct.as_bytes()) {
786 out.headers_mut().insert(header::CONTENT_TYPE, v);
787 }
788 }
789 out
790}
791
792async fn ws_events(
803 State(state): State<AppState>,
804 uri: Uri,
805 headers: HeaderMap,
806 ws: WebSocketUpgrade,
807) -> Response {
808 if !require_session_cookie(&state, &headers).await {
809 return unauthorized();
810 }
811
812 let query = uri.query().map(|q| format!("?{q}")).unwrap_or_default();
813 let ws_url = {
814 let base = state.upstream_base.as_str();
816 let ws_base = if let Some(rest) = base.strip_prefix("https://") {
817 format!("wss://{rest}")
818 } else if let Some(rest) = base.strip_prefix("http://") {
819 format!("ws://{rest}")
820 } else {
821 base.to_string()
822 };
823 format!("{ws_base}/ws/events{query}")
824 };
825 let bearer = state.upstream_bearer.as_ref().clone();
826 let subprotocols: Option<String> = headers
839 .get(axum::http::header::SEC_WEBSOCKET_PROTOCOL)
840 .and_then(|v| v.to_str().ok())
841 .map(|s| s.to_string());
842
843 ws.on_upgrade(move |client_ws| async move {
844 let (mut client_tx, mut client_rx) = client_ws.split();
845
846 let mut request =
849 match tokio_tungstenite::tungstenite::client::IntoClientRequest::into_client_request(
850 ws_url.as_str(),
851 ) {
852 Ok(r) => r,
853 Err(_) => return,
854 };
855 if let Some(tok) = bearer {
856 if let Ok(v) = tokio_tungstenite::tungstenite::http::HeaderValue::from_str(&format!(
857 "Bearer {tok}"
858 )) {
859 request.headers_mut().insert(
860 tokio_tungstenite::tungstenite::http::header::AUTHORIZATION,
861 v,
862 );
863 }
864 }
865 if let Some(proto) = subprotocols.as_ref() {
866 if let Ok(v) = tokio_tungstenite::tungstenite::http::HeaderValue::from_str(proto) {
867 request.headers_mut().insert(
868 tokio_tungstenite::tungstenite::http::header::SEC_WEBSOCKET_PROTOCOL,
869 v,
870 );
871 }
872 }
873 let (upstream_ws, _) = match tokio_tungstenite::connect_async(request).await {
874 Ok(p) => p,
875 Err(_) => return,
876 };
877 let (mut up_tx, mut up_rx) = upstream_ws.split();
878
879 loop {
880 tokio::select! {
881 msg = client_rx.next() => match msg {
882 Some(Ok(axum::extract::ws::Message::Text(t))) => {
883 let _ = up_tx
888 .send(tokio_tungstenite::tungstenite::Message::Text(
889 t.as_str().into(),
890 ))
891 .await;
892 }
893 Some(Ok(axum::extract::ws::Message::Binary(b))) => {
894 let _ = up_tx
895 .send(tokio_tungstenite::tungstenite::Message::Binary(b))
896 .await;
897 }
898 Some(Ok(axum::extract::ws::Message::Ping(p))) => {
899 let _ = up_tx
900 .send(tokio_tungstenite::tungstenite::Message::Ping(p))
901 .await;
902 }
903 Some(Ok(axum::extract::ws::Message::Pong(p))) => {
904 let _ = up_tx
905 .send(tokio_tungstenite::tungstenite::Message::Pong(p))
906 .await;
907 }
908 Some(Ok(axum::extract::ws::Message::Close(_))) | None => break,
909 Some(Err(_)) => break,
910 },
911 msg = up_rx.next() => match msg {
912 Some(Ok(tokio_tungstenite::tungstenite::Message::Text(t))) => {
913 let _ = client_tx
914 .send(axum::extract::ws::Message::Text(t.as_str().into()))
915 .await;
916 }
917 Some(Ok(tokio_tungstenite::tungstenite::Message::Binary(b))) => {
918 let _ = client_tx
919 .send(axum::extract::ws::Message::Binary(b))
920 .await;
921 }
922 Some(Ok(tokio_tungstenite::tungstenite::Message::Ping(p))) => {
923 let _ = client_tx
924 .send(axum::extract::ws::Message::Ping(p))
925 .await;
926 }
927 Some(Ok(tokio_tungstenite::tungstenite::Message::Pong(p))) => {
928 let _ = client_tx
929 .send(axum::extract::ws::Message::Pong(p))
930 .await;
931 }
932 Some(Ok(tokio_tungstenite::tungstenite::Message::Close(_))) | None => break,
933 Some(Ok(tokio_tungstenite::tungstenite::Message::Frame(_))) => {}
934 Some(Err(_)) => break,
935 },
936 }
937 }
938 })
939}
940
941fn unauthorized() -> Response {
942 (StatusCode::UNAUTHORIZED, "missing session cookie\n").into_response()
943}
944
945fn upstream_error(msg: String) -> Response {
946 (StatusCode::BAD_GATEWAY, format!("upstream: {msg}\n")).into_response()
947}
948
949#[allow(dead_code)]
955const _HEADER_NAME_KEEP: fn() = || {
956 let _: HeaderName = HeaderName::from_static("x-keep");
959};
960
961#[cfg(test)]
966mod tests {
967 use super::*;
968 use axum::body::to_bytes;
969 use axum::http::Request;
970 use tower::ServiceExt; fn test_state(bundle_dir: PathBuf) -> AppState {
973 AppState {
974 upstream_base: Arc::new("http://127.0.0.1:0".to_string()),
975 upstream_bearer: Arc::new(None),
976 session_token: Arc::new("test-token".to_string()),
977 session_cookie: Arc::new(RwLock::new(None)),
978 exchange_consumed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
979 bundle_dir: Arc::new(bundle_dir),
980 }
981 }
982
983 fn ensure_bundle_dir() -> PathBuf {
986 let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
987 let real = manifest_dir.join("static");
988 if real.join("index.html").exists() {
989 return real;
990 }
991 let tmp = std::env::temp_dir().join(format!("cellctl-webui-test-{}", std::process::id()));
993 std::fs::create_dir_all(&tmp).expect("mkdir tmp bundle");
994 std::fs::write(
995 tmp.join("index.html"),
996 "<!doctype html><title>cellctl webui</title>",
997 )
998 .expect("write index.html");
999 tmp
1000 }
1001
1002 #[tokio::test]
1003 async fn serves_index_at_root() {
1004 let bundle = ensure_bundle_dir();
1005 let app = build_router(test_state(bundle));
1006
1007 let resp = app
1008 .oneshot(
1009 Request::builder()
1010 .method("GET")
1011 .uri("/")
1012 .body(Body::empty())
1013 .unwrap(),
1014 )
1015 .await
1016 .unwrap();
1017 assert_eq!(resp.status(), StatusCode::OK);
1018 let body = to_bytes(resp.into_body(), 64 * 1024).await.unwrap();
1019 let body_str = std::str::from_utf8(&body).unwrap();
1020 assert!(
1021 body_str.to_ascii_lowercase().contains("<!doctype html")
1022 || body_str.to_ascii_lowercase().contains("<html"),
1023 "expected HTML at /, got: {body_str:.200}"
1024 );
1025 }
1026
1027 #[tokio::test]
1028 async fn non_get_returns_405() {
1029 let bundle = ensure_bundle_dir();
1030 let app = build_router(test_state(bundle));
1031
1032 let resp = app
1033 .oneshot(
1034 Request::builder()
1035 .method("DELETE")
1036 .uri("/v1/formations")
1037 .body(Body::empty())
1038 .unwrap(),
1039 )
1040 .await
1041 .unwrap();
1042 assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
1043 assert_eq!(
1044 resp.headers().get(header::ALLOW).map(|v| v.as_bytes()),
1045 Some(b"GET" as &[u8]),
1046 );
1047 }
1048
1049 #[tokio::test]
1050 async fn put_to_v1_returns_405() {
1051 let bundle = ensure_bundle_dir();
1052 let app = build_router(test_state(bundle));
1053
1054 let resp = app
1055 .oneshot(
1056 Request::builder()
1057 .method("PUT")
1058 .uri("/v1/formations/foo")
1059 .body(Body::from("body"))
1060 .unwrap(),
1061 )
1062 .await
1063 .unwrap();
1064 assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
1065 }
1066
1067 #[tokio::test]
1068 async fn proxy_without_cookie_returns_401() {
1069 let bundle = ensure_bundle_dir();
1070 let app = build_router(test_state(bundle));
1071
1072 let resp = app
1073 .oneshot(
1074 Request::builder()
1075 .method("GET")
1076 .uri("/v1/formations")
1077 .body(Body::empty())
1078 .unwrap(),
1079 )
1080 .await
1081 .unwrap();
1082 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
1083 }
1084
1085 #[tokio::test]
1086 async fn auth_exchange_with_wrong_token_returns_401() {
1087 let bundle = ensure_bundle_dir();
1088 let app = build_router(test_state(bundle));
1089
1090 let resp = app
1091 .oneshot(
1092 Request::builder()
1093 .method("POST")
1094 .uri("/auth/exchange")
1095 .header(header::CONTENT_TYPE, "application/json")
1096 .body(Body::from(r#"{"sess":"wrong"}"#))
1097 .unwrap(),
1098 )
1099 .await
1100 .unwrap();
1101 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
1102 }
1103
1104 #[tokio::test]
1105 async fn auth_exchange_with_right_token_sets_cookie() {
1106 let bundle = ensure_bundle_dir();
1107 let state = test_state(bundle);
1108 let app = build_router(state.clone());
1109
1110 let resp = app
1111 .oneshot(
1112 Request::builder()
1113 .method("POST")
1114 .uri("/auth/exchange")
1115 .header(header::CONTENT_TYPE, "application/json")
1116 .body(Body::from(r#"{"sess":"test-token"}"#))
1117 .unwrap(),
1118 )
1119 .await
1120 .unwrap();
1121 assert_eq!(resp.status(), StatusCode::OK);
1122 let cookie = resp
1123 .headers()
1124 .get(header::SET_COOKIE)
1125 .expect("Set-Cookie header present")
1126 .to_str()
1127 .unwrap()
1128 .to_string();
1129 assert!(cookie.starts_with("cellctl_session="));
1130 assert!(cookie.contains("HttpOnly"));
1131 assert!(cookie.contains("SameSite=Strict"));
1132
1133 let stored = state.session_cookie.read().await.clone();
1135 assert!(stored.is_some());
1136 let stored = stored.unwrap();
1137 assert!(cookie.contains(&format!("cellctl_session={stored}")));
1138 }
1139
1140 #[tokio::test]
1141 async fn proxy_with_valid_cookie_attempts_upstream() {
1142 let bundle = ensure_bundle_dir();
1147 let state = test_state(bundle);
1148 let app = build_router(state.clone());
1149
1150 let exch = app
1151 .clone()
1152 .oneshot(
1153 Request::builder()
1154 .method("POST")
1155 .uri("/auth/exchange")
1156 .header(header::CONTENT_TYPE, "application/json")
1157 .body(Body::from(r#"{"sess":"test-token"}"#))
1158 .unwrap(),
1159 )
1160 .await
1161 .unwrap();
1162 assert_eq!(exch.status(), StatusCode::OK);
1163 let cookie_hdr = exch
1164 .headers()
1165 .get(header::SET_COOKIE)
1166 .unwrap()
1167 .to_str()
1168 .unwrap()
1169 .to_string();
1170 let cookie_pair = cookie_hdr.split(';').next().unwrap().trim().to_string();
1172
1173 let resp = app
1174 .oneshot(
1175 Request::builder()
1176 .method("GET")
1177 .uri("/v1/formations")
1178 .header(header::COOKIE, cookie_pair)
1179 .body(Body::empty())
1180 .unwrap(),
1181 )
1182 .await
1183 .unwrap();
1184 assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
1186 }
1187
1188 #[tokio::test]
1200 async fn proxy_only_emits_get_to_upstream() {
1201 use std::sync::Mutex;
1202 use tokio::io::{AsyncReadExt, AsyncWriteExt};
1203 use tokio::net::TcpListener;
1204
1205 let mock = TcpListener::bind("127.0.0.1:0").await.expect("bind mock");
1207 let mock_addr = mock.local_addr().expect("mock addr");
1208 let methods_seen: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
1209 let methods_for_task = methods_seen.clone();
1210
1211 tokio::spawn(async move {
1218 loop {
1219 let (mut stream, _) = match mock.accept().await {
1220 Ok(p) => p,
1221 Err(_) => return,
1222 };
1223 let methods = methods_for_task.clone();
1224 tokio::spawn(async move {
1225 let mut buf = [0u8; 4096];
1226 let n = match stream.read(&mut buf).await {
1227 Ok(n) => n,
1228 Err(_) => return,
1229 };
1230 let head = String::from_utf8_lossy(&buf[..n]).to_string();
1231 if let Some(first_line) = head.lines().next() {
1232 if let Some(method) = first_line.split_whitespace().next() {
1233 methods.lock().unwrap().push(method.to_string());
1234 }
1235 }
1236 let _ = stream
1237 .write_all(
1238 b"HTTP/1.1 200 OK\r\n\
1239 Content-Type: application/json\r\n\
1240 Content-Length: 2\r\n\
1241 Connection: close\r\n\
1242 \r\n\
1243 {}",
1244 )
1245 .await;
1246 let _ = stream.shutdown().await;
1247 });
1248 }
1249 });
1250
1251 let bundle = ensure_bundle_dir();
1253 let state = AppState {
1254 upstream_base: Arc::new(format!("http://{}", mock_addr)),
1255 upstream_bearer: Arc::new(Some("test-bearer".to_string())),
1256 session_token: Arc::new("test-token".to_string()),
1257 session_cookie: Arc::new(RwLock::new(None)),
1258 exchange_consumed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
1259 bundle_dir: Arc::new(bundle),
1260 };
1261 let app = build_router(state.clone());
1262
1263 let exch = app
1267 .clone()
1268 .oneshot(
1269 Request::builder()
1270 .method("POST")
1271 .uri("/auth/exchange")
1272 .header(header::CONTENT_TYPE, "application/json")
1273 .body(Body::from(r#"{"sess":"test-token"}"#))
1274 .unwrap(),
1275 )
1276 .await
1277 .unwrap();
1278 assert_eq!(exch.status(), StatusCode::OK);
1279 let cookie_pair = exch
1280 .headers()
1281 .get(header::SET_COOKIE)
1282 .unwrap()
1283 .to_str()
1284 .unwrap()
1285 .split(';')
1286 .next()
1287 .unwrap()
1288 .trim()
1289 .to_string();
1290
1291 let non_get_methods = ["POST", "PUT", "DELETE", "PATCH"];
1294 for m in non_get_methods {
1295 let resp = app
1296 .clone()
1297 .oneshot(
1298 Request::builder()
1299 .method(m)
1300 .uri("/v1/formations")
1301 .header(header::COOKIE, cookie_pair.clone())
1302 .body(Body::from("payload that must never reach upstream"))
1303 .unwrap(),
1304 )
1305 .await
1306 .unwrap();
1307 assert_eq!(
1308 resp.status(),
1309 StatusCode::METHOD_NOT_ALLOWED,
1310 "inbound {m} should be 405"
1311 );
1312 assert_eq!(
1313 resp.headers().get(header::ALLOW).map(|v| v.as_bytes()),
1314 Some(b"GET" as &[u8]),
1315 "405 response for {m} must carry `Allow: GET`"
1316 );
1317 }
1318
1319 let get_resp = app
1321 .clone()
1322 .oneshot(
1323 Request::builder()
1324 .method("GET")
1325 .uri("/v1/formations")
1326 .header(header::COOKIE, cookie_pair.clone())
1327 .body(Body::empty())
1328 .unwrap(),
1329 )
1330 .await
1331 .unwrap();
1332 assert_ne!(
1334 get_resp.status(),
1335 StatusCode::METHOD_NOT_ALLOWED,
1336 "GET must pass the 405 middleware"
1337 );
1338
1339 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1341
1342 let observed = methods_seen.lock().unwrap().clone();
1344 let non_get_count = observed.iter().filter(|m| m.as_str() != "GET").count();
1347 assert_eq!(
1348 non_get_count, 0,
1349 "upstream saw non-GET method(s): {observed:?}"
1350 );
1351 assert!(
1352 observed.iter().any(|m| m == "GET"),
1353 "expected at least one GET to reach upstream, saw {observed:?}"
1354 );
1355 }
1356}