Façade Rust loki-rs : proxy devant le Go et mise à jour au démarrage

Premier lot de la migration vers Rust. Le binaire loki-rs :

- serve : prend le port public (8090), lance `loki web 8091` en écoute
  locale et lui relaie tout — HTTP, flux SSE trame par trame, WebSocket.
  Les en-têtes de bout en bout passent intacts (Host, Authorization,
  Origin, Sec-Fetch-*, X-Forwarded-Proto : le Go en tire ses contrôles
  d'accès) ; X-Loki-Via entrant est effacé, un client ne doit jamais
  pouvoir se faire passer pour le tunnel. Go mort : relancé avec attente
  croissante, 502 lisible entre-temps.
- self-update : lit la release « rolling », installe loki et loki-rs dans
  $LOKI_HOME/bin/<build>/ après vérification SHA-256, bascule par
  réécriture atomique de release.json. Les versions s'ordonnent par numéro
  de run CI ; une ABI d'image différente n'est jamais installée. Toute
  erreur laisse l'état précédent, le démarrage n'est jamais bloqué.
  Le HTTPS passe par curl (présent dans l'image) : pas de pile TLS ici.
- which-bin : dossier d'où l'entrypoint prend les deux binaires, toujours
  ensemble.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

rust/loki-rs/tests/common/mod.rs

@@ -0,0 +1,76 @@
+//! Serveur HTTP minimal pour les tests : sert des fichiers en mémoire et
+//! journalise les chemins demandés.
+#![allow(dead_code)]
+
+use bytes::Bytes;
+use http_body_util::Full;
+use hyper::{Request, Response, StatusCode, body::Incoming, service::service_fn};
+use hyper_util::rt::TokioIo;
+use std::collections::HashMap;
+use std::net::SocketAddr;
+use std::sync::{Arc, Mutex};
+
+pub struct FileServer {
+    pub addr: SocketAddr,
+    pub hits: Arc<Mutex<Vec<String>>>,
+    _rt: tokio::runtime::Runtime,
+}
+
+impl FileServer {
+    pub fn url(&self) -> String {
+        format!("http://{}/rolling", self.addr)
+    }
+    pub fn hits(&self) -> Vec<String> {
+        self.hits.lock().unwrap().clone()
+    }
+}
+
+pub fn serve_files(files: HashMap<String, Vec<u8>>) -> FileServer {
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    let files = Arc::new(files);
+    let hits = Arc::new(Mutex::new(Vec::new()));
+    let listener = rt
+        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
+        .unwrap();
+    let addr = listener.local_addr().unwrap();
+    let h = hits.clone();
+    rt.spawn(async move {
+        loop {
+            let Ok((stream, _)) = listener.accept().await else {
+                return;
+            };
+            let files = files.clone();
+            let h = h.clone();
+            tokio::spawn(async move {
+                let svc = service_fn(move |req: Request<Incoming>| {
+                    let files = files.clone();
+                    let h = h.clone();
+                    async move {
+                        let path = req.uri().path().trim_start_matches("/rolling/").to_string();
+                        h.lock().unwrap().push(path.clone());
+                        let resp = match files.get(&path) {
+                            Some(b) => Response::new(Full::new(Bytes::from(b.clone()))),
+                            None => {
+                                let mut r = Response::new(Full::new(Bytes::from_static(b"404")));
+                                *r.status_mut() = StatusCode::NOT_FOUND;
+                                r
+                            }
+                        };
+                        Ok::<_, std::convert::Infallible>(resp)
+                    }
+                });
+                let _ = hyper::server::conn::http1::Builder::new()
+                    .serve_connection(TokioIo::new(stream), svc)
+                    .await;
+            });
+        }
+    });
+    FileServer {
+        addr,
+        hits,
+        _rt: rt,
+    }
+}

rust/loki-rs/tests/proxy.rs

@@ -0,0 +1,206 @@
+use bytes::Bytes;
+use futures_util::{SinkExt, StreamExt};
+use http_body_util::{BodyExt, Empty, Full, StreamBody, combinators::BoxBody};
+use hyper::body::{Frame, Incoming};
+use hyper::{Request, Response, StatusCode, service::service_fn};
+use hyper_util::client::legacy::Client;
+use hyper_util::rt::{TokioExecutor, TokioIo};
+use std::convert::Infallible;
+use std::net::SocketAddr;
+use std::time::{Duration, Instant};
+use tokio::net::TcpListener;
+use tokio::sync::oneshot;
+use tokio_tungstenite::tungstenite::Message;
+
+type B = BoxBody<Bytes, Infallible>;
+
+/// Backend factice : /echo renvoie les en-têtes reçus, /sse émet un événement,
+/// attend 500 ms, puis un second ; /post renvoie le corps reçu.
+async fn backend() -> SocketAddr {
+    let l = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = l.local_addr().unwrap();
+    tokio::spawn(async move {
+        loop {
+            let (s, _) = l.accept().await.unwrap();
+            tokio::spawn(async move {
+                let svc = service_fn(|req: Request<Incoming>| async move {
+                    let resp: Response<B> = match req.uri().path() {
+                        "/echo" => {
+                            let mut lines: Vec<String> = req
+                                .headers()
+                                .iter()
+                                .map(|(k, v)| format!("{}: {}", k, v.to_str().unwrap()))
+                                .collect();
+                            lines.push(format!("uri: {}", req.uri()));
+                            Response::new(Full::new(Bytes::from(lines.join("\n"))).boxed())
+                        }
+                        "/post" => {
+                            let body = req.into_body().collect().await.unwrap().to_bytes();
+                            Response::new(Full::new(body).boxed())
+                        }
+                        "/sse" => {
+                            let (tx, rx) = tokio::sync::mpsc::channel::<Bytes>(4);
+                            tokio::spawn(async move {
+                                tx.send(Bytes::from_static(b"data: un\n\n")).await.unwrap();
+                                tokio::time::sleep(Duration::from_millis(500)).await;
+                                let _ = tx.send(Bytes::from_static(b"data: deux\n\n")).await;
+                            });
+                            let stream = futures_util::stream::unfold(rx, |mut rx| async move {
+                                rx.recv()
+                                    .await
+                                    .map(|b| (Ok::<_, Infallible>(Frame::data(b)), rx))
+                            });
+                            let mut r = Response::new(BodyExt::boxed(StreamBody::new(stream)));
+                            r.headers_mut()
+                                .insert("content-type", "text/event-stream".parse().unwrap());
+                            r
+                        }
+                        _ => Response::new(Empty::new().boxed()),
+                    };
+                    Ok::<_, Infallible>(resp)
+                });
+                let _ = hyper::server::conn::http1::Builder::new()
+                    .serve_connection(TokioIo::new(s), svc)
+                    .await;
+            });
+        }
+    });
+    addr
+}
+
+async fn ws_backend() -> SocketAddr {
+    let l = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = l.local_addr().unwrap();
+    tokio::spawn(async move {
+        while let Ok((s, _)) = l.accept().await {
+            tokio::spawn(async move {
+                let mut ws = tokio_tungstenite::accept_async(s).await.unwrap();
+                while let Some(Ok(m)) = ws.next().await {
+                    if m.is_text() {
+                        ws.send(m).await.unwrap();
+                    }
+                }
+            });
+        }
+    });
+    addr
+}
+
+/// Lance la façade devant `backend`, rend son adresse.
+async fn facade(backend: SocketAddr) -> (SocketAddr, oneshot::Sender<()>) {
+    let l = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = l.local_addr().unwrap();
+    let (tx, rx) = oneshot::channel::<()>();
+    tokio::spawn(loki_rs::proxy::serve(l, backend, async move {
+        let _ = rx.await;
+    }));
+    (addr, tx)
+}
+
+fn client() -> Client<hyper_util::client::legacy::connect::HttpConnector, Full<Bytes>> {
+    Client::builder(TokioExecutor::new()).build_http()
+}
+
+#[tokio::test]
+async fn transmet_les_en_tetes_et_efface_x_loki_via() {
+    let (f, _stop) = facade(backend().await).await;
+    let req = Request::get(format!("http://{f}/echo?a=1"))
+        .header("host", "loki.exemple.fr")
+        .header("authorization", "Bearer secret")
+        .header("origin", "https://loki.exemple.fr")
+        .header("x-forwarded-proto", "https")
+        .header("sec-fetch-site", "same-origin")
+        .header("x-loki-via", "tunnel")
+        .body(Full::new(Bytes::new()))
+        .unwrap();
+    let resp = client().request(req).await.unwrap();
+    assert_eq!(resp.status(), StatusCode::OK);
+    let body = String::from_utf8(
+        resp.into_body()
+            .collect()
+            .await
+            .unwrap()
+            .to_bytes()
+            .to_vec(),
+    )
+    .unwrap();
+    for want in [
+        "host: loki.exemple.fr",
+        "authorization: Bearer secret",
+        "origin: https://loki.exemple.fr",
+        "x-forwarded-proto: https",
+        "sec-fetch-site: same-origin",
+        "uri: /echo?a=1",
+    ] {
+        assert!(
+            body.lines().any(|l| l == want),
+            "manque « {want} » dans :\n{body}"
+        );
+    }
+    assert!(
+        !body.contains("x-loki-via"),
+        "X-Loki-Via a traversé :\n{body}"
+    );
+}
+
+#[tokio::test]
+async fn relaie_le_corps() {
+    let (f, _stop) = facade(backend().await).await;
+    let req = Request::post(format!("http://{f}/post"))
+        .body(Full::new(Bytes::from_static(b"{\"x\":1}")))
+        .unwrap();
+    let resp = client().request(req).await.unwrap();
+    let body = resp.into_body().collect().await.unwrap().to_bytes();
+    assert_eq!(&body[..], b"{\"x\":1}");
+}
+
+#[tokio::test]
+async fn sse_sans_tampon() {
+    let (f, _stop) = facade(backend().await).await;
+    let t0 = Instant::now();
+    let req = Request::get(format!("http://{f}/sse"))
+        .body(Full::new(Bytes::new()))
+        .unwrap();
+    let mut body = client().request(req).await.unwrap().into_body();
+    let first = body.frame().await.unwrap().unwrap().into_data().unwrap();
+    let elapsed = t0.elapsed();
+    assert_eq!(&first[..], b"data: un\n\n");
+    assert!(
+        elapsed < Duration::from_millis(300),
+        "premier événement reçu après {elapsed:?}"
+    );
+    let second = body.frame().await.unwrap().unwrap().into_data().unwrap();
+    assert_eq!(&second[..], b"data: deux\n\n");
+}
+
+#[tokio::test]
+async fn websocket_relaye() {
+    let (f, _stop) = facade(ws_backend().await).await;
+    let (mut ws, resp) = tokio_tungstenite::connect_async(format!("ws://{f}/api/node/ws"))
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::SWITCHING_PROTOCOLS);
+    ws.send(Message::text("bonjour")).await.unwrap();
+    let got = tokio::time::timeout(Duration::from_secs(5), ws.next())
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap();
+    assert_eq!(got.into_text().unwrap().as_str(), "bonjour");
+}
+
+#[tokio::test]
+async fn backend_absent_donne_502() {
+    // Port réservé puis libéré : personne n'y écoute.
+    let dead = TcpListener::bind("127.0.0.1:0")
+        .await
+        .unwrap()
+        .local_addr()
+        .unwrap();
+    let (f, _stop) = facade(dead).await;
+    let req = Request::get(format!("http://{f}/"))
+        .body(Full::new(Bytes::new()))
+        .unwrap();
+    let resp = client().request(req).await.unwrap();
+    assert_eq!(resp.status(), StatusCode::BAD_GATEWAY);
+}

rust/loki-rs/tests/update.rs

@@ -0,0 +1,148 @@
+mod common;
+
+use loki_rs::update::{self, Outcome, UpdateConfig, sha256_file, state};
+use std::collections::HashMap;
+use std::path::Path;
+use std::time::Duration;
+
+const GO: &[u8] = b"binaire go";
+const RS: &[u8] = b"binaire rust";
+
+fn sha(b: &[u8]) -> String {
+    let t = tempfile::NamedTempFile::new().unwrap();
+    std::fs::write(t.path(), b).unwrap();
+    sha256_file(t.path()).unwrap()
+}
+
+fn release(
+    build: u64,
+    abi: u32,
+    go: &[u8],
+    rs: &[u8],
+    sums_go: Option<&str>,
+) -> HashMap<String, Vec<u8>> {
+    let sums = format!(
+        "{}  loki-linux\n{}  loki-rs-linux\n",
+        sums_go.map(str::to_string).unwrap_or_else(|| sha(go)),
+        sha(rs)
+    );
+    HashMap::from([
+        (
+            "manifest.json".into(),
+            format!(r#"{{"build":{build},"abi":{abi},"commit":"c{build}"}}"#).into_bytes(),
+        ),
+        ("SHA256SUMS".into(), sums.into_bytes()),
+        ("loki-linux".into(), go.to_vec()),
+        ("loki-rs-linux".into(), rs.to_vec()),
+    ])
+}
+
+fn cfg(url: String, data_bin: &Path) -> UpdateConfig {
+    UpdateConfig {
+        enabled: true,
+        base_url: url,
+        data_bin: data_bin.to_path_buf(),
+        image_build: 3,
+        image_abi: 1,
+        small_timeout: Duration::from_secs(10),
+        big_timeout: Duration::from_secs(10),
+    }
+}
+
+#[test]
+fn installe_puis_reconnait_a_jour() {
+    let srv = common::serve_files(release(7, 1, GO, RS, None));
+    let t = tempfile::tempdir().unwrap();
+    let c = cfg(srv.url(), t.path());
+
+    assert_eq!(update::run(&c), Outcome::Installed(7));
+    let dir = state::select_dir(t.path(), Path::new("/img"), 3, 1);
+    assert_eq!(dir, t.path().join("7"));
+    assert_eq!(std::fs::read(dir.join("loki")).unwrap(), GO);
+    assert_eq!(std::fs::read(dir.join("loki-rs")).unwrap(), RS);
+    assert!(!t.path().join(".tmp-download").exists());
+
+    assert_eq!(update::run(&c), Outcome::UpToDate(7));
+}
+
+#[test]
+fn nouveau_build_remplace_et_nettoie_l_ancien() {
+    let t = tempfile::tempdir().unwrap();
+    let srv = common::serve_files(release(7, 1, GO, RS, None));
+    assert_eq!(
+        update::run(&cfg(srv.url(), t.path())),
+        Outcome::Installed(7)
+    );
+    let srv2 = common::serve_files(release(9, 1, b"go 9", b"rs 9", None));
+    assert_eq!(
+        update::run(&cfg(srv2.url(), t.path())),
+        Outcome::Installed(9)
+    );
+    assert!(!t.path().join("7").exists());
+    assert_eq!(
+        state::select_dir(t.path(), Path::new("/img"), 3, 1),
+        t.path().join("9")
+    );
+}
+
+#[test]
+fn empreinte_fausse_ne_change_rien() {
+    let t = tempfile::tempdir().unwrap();
+    let srv = common::serve_files(release(7, 1, GO, RS, None));
+    assert_eq!(
+        update::run(&cfg(srv.url(), t.path())),
+        Outcome::Installed(7)
+    );
+
+    let bad = "0".repeat(64);
+    let srv2 = common::serve_files(release(9, 1, b"go 9", b"rs 9", Some(&bad)));
+    assert!(matches!(
+        update::run(&cfg(srv2.url(), t.path())),
+        Outcome::Failed(_)
+    ));
+    assert_eq!(
+        state::select_dir(t.path(), Path::new("/img"), 3, 1),
+        t.path().join("7")
+    );
+    assert!(!t.path().join("9").exists());
+    assert!(!t.path().join(".tmp-download").exists());
+}
+
+#[test]
+fn abi_differente_ne_telecharge_que_le_manifeste() {
+    let t = tempfile::tempdir().unwrap();
+    let srv = common::serve_files(release(7, 2, GO, RS, None));
+    assert_eq!(
+        update::run(&cfg(srv.url(), t.path())),
+        Outcome::AbiMismatch {
+            remote_abi: 2,
+            image_abi: 1
+        }
+    );
+    assert_eq!(srv.hits(), vec!["manifest.json".to_string()]);
+}
+
+#[test]
+fn image_plus_recente_que_la_release() {
+    let t = tempfile::tempdir().unwrap();
+    let srv = common::serve_files(release(3, 1, GO, RS, None));
+    assert_eq!(update::run(&cfg(srv.url(), t.path())), Outcome::UpToDate(3));
+}
+
+#[test]
+fn serveur_absent() {
+    let t = tempfile::tempdir().unwrap();
+    let mut c = cfg("http://127.0.0.1:9/rolling".into(), t.path());
+    c.small_timeout = Duration::from_secs(3);
+    assert!(matches!(update::run(&c), Outcome::Failed(_)));
+}
+
+#[test]
+fn desactive_sans_requete() {
+    let t = tempfile::tempdir().unwrap();
+    let srv = common::serve_files(release(7, 1, GO, RS, None));
+    let mut c = cfg(srv.url(), t.path());
+    c.enabled = false;
+    assert_eq!(update::run(&c), Outcome::Disabled);
+    assert!(srv.hits().is_empty());
+}