diff --git a/internal/loki/web_internal.go b/internal/loki/web_internal.go index ed30060..6a239ad 100644 --- a/internal/loki/web_internal.go +++ b/internal/loki/web_internal.go @@ -2,9 +2,12 @@ package loki import ( "crypto/subtle" + "io" "net/http" "os" "strings" + + bolt "go.etcd.io/bbolt" ) // API interne entre la façade Rust (rust/loki-rs) et ce serveur. @@ -56,12 +59,107 @@ func withInternalAPI(next http.Handler) http.Handler { switch r.URL.Path { case "/internal/v1/auth": handleInternalAuth(w, r) + case "/internal/v1/store": + handleInternalStore(w, r) + case "/internal/v1/store/all": + handleInternalStoreAll(w, r) default: http.NotFound(w, r) } }) } +// --- Store ---------------------------------------------------------------- +// +// Accès générique à la base pour les routes déjà portées en Rust (lot 4) : le +// Go reste l'unique propriétaire du fichier bbolt jusqu'au lot qui le +// migrera ; Rust lit et écrit À TRAVERS lui. Les écritures passent par +// putBytes/update, donc invalident le cache de ce process comme une écriture +// locale. +// +// GET /internal/v1/store?bucket=B&key=K 200 + valeur brute | 404 absente +// PUT /internal/v1/store?bucket=B&key=K corps = valeur → 204 +// DELETE /internal/v1/store?bucket=B&key=K → 204 +// GET /internal/v1/store/all?bucket=B 200 + {"clé":"valeur",…} +// +// Base illisible : 503 — l'appelant distingue « rien » de « pas pu lire », +// comme getBytesErr. + +// internalStoreMaxPut borne une écriture : la plus grosse valeur légitime (le +// journal d'une discussion) reste très en dessous. +const internalStoreMaxPut = 64 << 20 + +func handleInternalStore(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + bucket, key := q.Get("bucket"), q.Get("key") + if bucket == "" || key == "" { + sendJSON(w, http.StatusBadRequest, map[string]any{"error": "bucket et key requis"}) + return + } + switch r.Method { + case http.MethodGet: + var val []byte + var found bool + err := view(bucket, func(b *bolt.Bucket) error { + if v := b.Get([]byte(key)); v != nil { + val, found = append([]byte(nil), v...), true + } + return nil + }) + if err != nil { + sendJSON(w, http.StatusServiceUnavailable, map[string]any{"error": err.Error()}) + return + } + if !found { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "application/octet-stream") + w.Write(val) + case http.MethodPut: + val, err := io.ReadAll(io.LimitReader(r.Body, internalStoreMaxPut+1)) + if err != nil || len(val) > internalStoreMaxPut { + sendJSON(w, http.StatusRequestEntityTooLarge, map[string]any{"error": "valeur trop grande ou illisible"}) + return + } + if err := putBytes(bucket, key, val); err != nil { + sendJSON(w, http.StatusServiceUnavailable, map[string]any{"error": err.Error()}) + return + } + w.WriteHeader(http.StatusNoContent) + case http.MethodDelete: + err := update(bucket, func(b *bolt.Bucket) error { return b.Delete([]byte(key)) }) + if err != nil { + sendJSON(w, http.StatusServiceUnavailable, map[string]any{"error": err.Error()}) + return + } + w.WriteHeader(http.StatusNoContent) + default: + w.Header().Set("Allow", "GET, PUT, DELETE") + sendJSON(w, http.StatusMethodNotAllowed, map[string]any{"error": "méthode non permise"}) + } +} + +func handleInternalStoreAll(w http.ResponseWriter, r *http.Request) { + bucket := r.URL.Query().Get("bucket") + if r.Method != http.MethodGet || bucket == "" { + sendJSON(w, http.StatusBadRequest, map[string]any{"error": "GET ?bucket= attendu"}) + return + } + m := map[string]string{} + err := view(bucket, func(b *bolt.Bucket) error { + return b.ForEach(func(k, v []byte) error { + m[string(k)] = string(v) + return nil + }) + }) + if err != nil { + sendJSON(w, http.StatusServiceUnavailable, map[string]any{"error": err.Error()}) + return + } + sendJSON(w, http.StatusOK, m) +} + // handleInternalAuth : de quoi authentifier une requête de pilotage côté Rust. // Base illisible → 503 : la façade ferme alors l'accès, elle n'ouvre jamais // sur une panne d'E/S (cf. readWebKeyErr). diff --git a/internal/loki/web_internal_test.go b/internal/loki/web_internal_test.go index 1c2a1a9..d3e8f31 100644 --- a/internal/loki/web_internal_test.go +++ b/internal/loki/web_internal_test.go @@ -2,6 +2,7 @@ package loki import ( "encoding/json" + "io" "net/http" "net/http/httptest" "strings" @@ -73,3 +74,59 @@ func TestAPIInterneLaisseLeResteAuMux(t *testing.T) { t.Fatalf("relais au mux : %q", w.Body.String()) } } + +func appelStore(t *testing.T, method, path string, body string) *httptest.ResponseRecorder { + t.Helper() + var rd io.Reader + if body != "" { + rd = strings.NewReader(body) + } + r := httptest.NewRequest(method, path, rd) + r.Header.Set(internalTokenHeader, jetonTest) + w := httptest.NewRecorder() + withInternalAPI(http.NotFoundHandler()).ServeHTTP(w, r) + return w +} + +func TestAPIInterneStore(t *testing.T) { + testHome(t) + t.Setenv(internalTokenEnv, jetonTest) + if w := appelStore(t, "GET", "/internal/v1/store?bucket=state&key=absente", ""); w.Code != 404 { + t.Fatalf("clé absente : %d", w.Code) + } + if w := appelStore(t, "PUT", "/internal/v1/store?bucket=state&key=k", "valeur\x00binaire"); w.Code != 204 { + t.Fatalf("put : %d %s", w.Code, w.Body.String()) + } + if got := getStr(bkState, "k"); got != "valeur\x00binaire" { + t.Fatalf("écrit : %q", got) + } + w := appelStore(t, "GET", "/internal/v1/store?bucket=state&key=k", "") + if w.Code != 200 || w.Body.String() != "valeur\x00binaire" { + t.Fatalf("get : %d %q", w.Code, w.Body.String()) + } + if err := putStr(bkConfig, "PORT", "8081"); err != nil { + t.Fatal(err) + } + w = appelStore(t, "GET", "/internal/v1/store/all?bucket=config", "") + var m map[string]string + if err := json.Unmarshal(w.Body.Bytes(), &m); err != nil || m["PORT"] != "8081" { + t.Fatalf("all : %d %s", w.Code, w.Body.String()) + } + // Écriture via l'API = écriture locale : le cache de config est à jour. + _ = ReadConfig() + if w := appelStore(t, "PUT", "/internal/v1/store?bucket=config&key=PORT", "9000"); w.Code != 204 { + t.Fatalf("put config : %d", w.Code) + } + if LLMPort() != 9000 { + t.Fatalf("cache non invalidé : port %d", LLMPort()) + } + if w := appelStore(t, "DELETE", "/internal/v1/store?bucket=state&key=k", ""); w.Code != 204 { + t.Fatalf("delete : %d", w.Code) + } + if getStr(bkState, "k") != "" { + t.Fatal("clé non supprimée") + } + if w := appelStore(t, "GET", "/internal/v1/store?bucket=state", ""); w.Code != 400 { + t.Fatalf("sans clé : %d", w.Code) + } +} diff --git a/rust/loki-rs/src/api.rs b/rust/loki-rs/src/api.rs new file mode 100644 index 0000000..d5237e4 --- /dev/null +++ b/rust/loki-rs/src/api.rs @@ -0,0 +1,73 @@ +//! Routes `/api/*` servies par Rust. Toute route listée dans [`handles`] est +//! traitée ici ; les autres partent au Go. +//! +//! Chaque route de pilotage passe par [`Ctx::guard`], port de +//! `requireWebAuth` (clé relue à chaque requête, fermeture sur erreur). + +use bytes::Bytes; +use hyper::body::Incoming; +use hyper::header::HeaderMap; +use hyper::{Method, Request, Response, StatusCode}; +use serde_json::json; + +use crate::auth::{self, HostEnv, InternalClient}; +use crate::engine; +use crate::proxy::Body; +use crate::store::Store; + +/// Ce dont les routes ont besoin, partagé par toutes les connexions. +pub struct Ctx { + pub store: Store, + pub internal: InternalClient, + pub host_env: HostEnv, + /// `$LOKI_HOME` et nom du service moteur (fichiers PID / journal). + pub engine: engine::Paths, +} + +impl Ctx { + /// `Ok` = requête autorisée (avec la version du Go, pour `/api/ping`) ; + /// sinon la réponse de refus. + pub async fn guard(&self, headers: &HeaderMap) -> Result>> { + let info = self.internal.auth().await; + if let Some(refus) = auth::authorize(headers, &info, &self.host_env) { + return Err(Box::new(refus)); + } + Ok(info.map(|i| i.version).unwrap_or_default()) + } +} + +/// Cette route est-elle servie par Rust ? +pub fn handles(_method: &Method, path: &str) -> bool { + matches!(path, "/api/ping") +} + +pub async fn serve(ctx: &Ctx, req: Request) -> Response { + let (parts, _body) = req.into_parts(); + let version = match ctx.guard(&parts.headers).await { + Ok(v) => v, + Err(refus) => return *refus, + }; + match parts.uri.path() { + // Vérifie la connectivité ET la clé (200 = bonne clé, 401 = mauvaise). + "/api/ping" => auth::json( + StatusCode::OK, + &json!({ "ok": true, "service": "loki", "version": version }), + ), + _ => not_found(), + } +} + +fn not_found() -> Response { + auth::json(StatusCode::NOT_FOUND, &json!({ "error": "route inconnue" })) +} + +/// Lecture bornée du corps d'une requête (routes POST portées). +#[allow(dead_code)] +pub async fn read_body(body: Incoming, max: usize) -> Result { + use http_body_util::{BodyExt, Limited}; + Limited::new(body, max) + .collect() + .await + .map(|c| c.to_bytes()) + .map_err(|e| e.to_string()) +} diff --git a/rust/loki-rs/src/engstatus.rs b/rust/loki-rs/src/engstatus.rs new file mode 100644 index 0000000..134237e --- /dev/null +++ b/rust/loki-rs/src/engstatus.rs @@ -0,0 +1 @@ +//! Lot 4 : module en cours de portage. diff --git a/rust/loki-rs/src/gguf.rs b/rust/loki-rs/src/gguf.rs new file mode 100644 index 0000000..134237e --- /dev/null +++ b/rust/loki-rs/src/gguf.rs @@ -0,0 +1 @@ +//! Lot 4 : module en cours de portage. diff --git a/rust/loki-rs/src/lib.rs b/rust/loki-rs/src/lib.rs index b34f1c6..ef434cf 100644 --- a/rust/loki-rs/src/lib.rs +++ b/rust/loki-rs/src/lib.rs @@ -7,9 +7,16 @@ //! - `auth` : authentification de l'API de pilotage (lot 2) ; //! - `update` : mise à jour des binaires au démarrage du conteneur. +pub mod api; pub mod auth; pub mod child; pub mod engine; +pub mod engstatus; +pub mod gguf; +pub mod paths; +pub mod presets; pub mod proxy; +pub mod store; +pub mod sysinfo; pub mod ui; pub mod update; diff --git a/rust/loki-rs/src/main.rs b/rust/loki-rs/src/main.rs index 979c44e..e134036 100644 --- a/rust/loki-rs/src/main.rs +++ b/rust/loki-rs/src/main.rs @@ -135,6 +135,7 @@ fn serve() -> ExitCode { backend, internal_token: token, host_env: auth::HostEnv::from_env(), + engine: engine::Paths::from_env(), }; let res = proxy::serve(listener, cfg, shutdown_signal()).await; eprintln!("[loki-rs] arrêt demandé — arrêt du serveur Go"); diff --git a/rust/loki-rs/src/paths.rs b/rust/loki-rs/src/paths.rs new file mode 100644 index 0000000..134237e --- /dev/null +++ b/rust/loki-rs/src/paths.rs @@ -0,0 +1 @@ +//! Lot 4 : module en cours de portage. diff --git a/rust/loki-rs/src/presets.rs b/rust/loki-rs/src/presets.rs new file mode 100644 index 0000000..134237e --- /dev/null +++ b/rust/loki-rs/src/presets.rs @@ -0,0 +1 @@ +//! Lot 4 : module en cours de portage. diff --git a/rust/loki-rs/src/proxy.rs b/rust/loki-rs/src/proxy.rs index 1ddacfb..f2e1b05 100644 --- a/rust/loki-rs/src/proxy.rs +++ b/rust/loki-rs/src/proxy.rs @@ -1,5 +1,5 @@ //! Façade HTTP : reçoit tout le trafic public. Les routes déjà réécrites en -//! Rust sont servies ici (lot 2 : UI statique, `/api/ping`) ; tout le reste +//! Rust sont servies ici (UI statique, routes de `api.rs`) ; tout le reste //! est relayé au serveur Go (`loki web` sur 127.0.0.1). //! //! Règles, héritées de `web_server.go` : @@ -27,7 +27,10 @@ use std::sync::Arc; use std::time::Duration; use tokio::net::TcpListener; -use crate::auth::{self, HostEnv, INTERNAL_HEADER, InternalClient}; +use crate::api; +use crate::auth::{HostEnv, INTERNAL_HEADER, InternalClient}; +use crate::engine; +use crate::store::Store; use crate::ui; pub type Body = BoxBody; @@ -53,13 +56,14 @@ pub struct Config { /// Jeton de l'API interne du Go (`LOKI_INTERNAL_TOKEN`). pub internal_token: String, pub host_env: HostEnv, + /// `$LOKI_HOME` et service moteur, pour les routes portées. + pub engine: engine::Paths, } struct Proxy { client: Client, backend: SocketAddr, - internal: InternalClient, - host_env: HostEnv, + ctx: api::Ctx, } /// Sert la façade sur `listener` jusqu'à ce que `shutdown` se termine. @@ -72,8 +76,12 @@ pub async fn serve( let proxy = Arc::new(Proxy { client, backend: cfg.backend, - internal: InternalClient::new(cfg.backend, cfg.internal_token), - host_env: cfg.host_env, + ctx: api::Ctx { + store: Store::new(cfg.backend, cfg.internal_token.clone()), + internal: InternalClient::new(cfg.backend, cfg.internal_token), + host_env: cfg.host_env, + engine: cfg.engine, + }, }); tokio::pin!(shutdown); loop { @@ -123,26 +131,12 @@ impl Proxy { if let Some(resp) = ui::serve(req.method(), path, req.headers()) { return resp; } - if path == "/api/ping" { - return self.ping(req.headers()).await; + if api::handles(req.method(), path) { + return api::serve(&self.ctx, req).await; } self.forward(req).await } - /// `/api/ping` : vérifie la connectivité ET la clé (200 = bonne clé, - /// 401 = mauvaise clé), comme `handlePing`. - async fn ping(&self, headers: &HeaderMap) -> Response { - let info = self.internal.auth().await; - if let Some(refus) = auth::authorize(headers, &info, &self.host_env) { - return refus; - } - let version = info.map(|i| i.version).unwrap_or_default(); - auth::json( - StatusCode::OK, - &serde_json::json!({ "ok": true, "service": "loki", "version": version }), - ) - } - /// Relais vers le Go (HTTP, SSE, WebSocket). async fn forward(&self, mut req: Request) -> Response { let upgrade = upgrade_protocol(req.headers()); diff --git a/rust/loki-rs/src/store.rs b/rust/loki-rs/src/store.rs new file mode 100644 index 0000000..055b09b --- /dev/null +++ b/rust/loki-rs/src/store.rs @@ -0,0 +1,136 @@ +//! Accès à la base de Loki À TRAVERS le Go (`/internal/v1/store`, voir +//! `web_internal.go`). Le Go reste l'unique propriétaire du fichier bbolt +//! jusqu'au lot qui le migrera ; ce module est l'unique point de passage +//! côté Rust, pour que cette migration ne touche qu'ici. + +use bytes::Bytes; +use http_body_util::{BodyExt, Full}; +use hyper::{Method, Request, StatusCode}; +use hyper_util::client::legacy::{Client, connect::HttpConnector}; +use hyper_util::rt::TokioExecutor; +use std::collections::HashMap; +use std::net::SocketAddr; +use std::time::Duration; + +use crate::auth::INTERNAL_HEADER; + +pub const BK_CONFIG: &str = "config"; +pub const BK_PREFS: &str = "prefs"; +pub const BK_STATE: &str = "state"; + +#[derive(Clone)] +pub struct Store { + client: Client>, + backend: SocketAddr, + token: String, +} + +impl Store { + pub fn new(backend: SocketAddr, token: String) -> Self { + Store { + client: Client::builder(TokioExecutor::new()).build_http(), + backend, + token, + } + } + + async fn call( + &self, + method: Method, + path: &str, + body: Vec, + ) -> Result<(StatusCode, Bytes), String> { + let req = Request::builder() + .method(method) + .uri(format!("http://{}{path}", self.backend)) + .header(INTERNAL_HEADER, &self.token) + .body(Full::new(Bytes::from(body))) + .map_err(|e| e.to_string())?; + let fut = async { + let resp = self.client.request(req).await.map_err(|e| e.to_string())?; + let status = resp.status(); + let body = resp + .into_body() + .collect() + .await + .map_err(|e| e.to_string())? + .to_bytes(); + Ok((status, body)) + }; + tokio::time::timeout(Duration::from_secs(10), fut) + .await + .map_err(|_| "store : délai dépassé".to_string())? + } + + /// Valeur brute ; `Ok(None)` = absente, `Err` = pas pu lire (à distinguer + /// pour les secrets : comme `getBytesErr`). + pub async fn get(&self, bucket: &str, key: &str) -> Result>, String> { + let path = format!("/internal/v1/store?bucket={}&key={}", enc(bucket), enc(key)); + match self.call(Method::GET, &path, Vec::new()).await? { + (StatusCode::OK, b) => Ok(Some(b.to_vec())), + (StatusCode::NOT_FOUND, _) => Ok(None), + (s, b) => Err(format!("store {s} : {}", String::from_utf8_lossy(&b))), + } + } + + /// Comme `getStr` du Go : erreur ou absence → "". + pub async fn get_str(&self, bucket: &str, key: &str) -> String { + match self.get(bucket, key).await { + Ok(Some(v)) => String::from_utf8_lossy(&v).into_owned(), + _ => String::new(), + } + } + + pub async fn put(&self, bucket: &str, key: &str, val: Vec) -> Result<(), String> { + let path = format!("/internal/v1/store?bucket={}&key={}", enc(bucket), enc(key)); + match self.call(Method::PUT, &path, val).await? { + (StatusCode::NO_CONTENT, _) => Ok(()), + (s, b) => Err(format!("store {s} : {}", String::from_utf8_lossy(&b))), + } + } + + pub async fn del(&self, bucket: &str, key: &str) -> Result<(), String> { + let path = format!("/internal/v1/store?bucket={}&key={}", enc(bucket), enc(key)); + match self.call(Method::DELETE, &path, Vec::new()).await? { + (StatusCode::NO_CONTENT, _) => Ok(()), + (s, b) => Err(format!("store {s} : {}", String::from_utf8_lossy(&b))), + } + } + + pub async fn all(&self, bucket: &str) -> Result, String> { + let path = format!("/internal/v1/store/all?bucket={}", enc(bucket)); + match self.call(Method::GET, &path, Vec::new()).await? { + (StatusCode::OK, b) => serde_json::from_slice(&b).map_err(|e| e.to_string()), + (s, b) => Err(format!("store {s} : {}", String::from_utf8_lossy(&b))), + } + } + + /// `ReadConfig()` du Go : la configuration du moteur ; base illisible → + /// configuration vide (comme `allKV`). + pub async fn config(&self) -> HashMap { + self.all(BK_CONFIG).await.unwrap_or_default() + } +} + +/// Encodage d'un paramètre de requête (RFC 3986, caractères non réservés +/// conservés). +pub fn enc(s: &str) -> String { + let mut out = String::with_capacity(s.len()); + for b in s.bytes() { + if b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.' | b'~') { + out.push(b as char); + } else { + out.push_str(&format!("%{b:02X}")); + } + } + out +} + +#[cfg(test)] +mod tests { + #[test] + fn encodage() { + assert_eq!(super::enc("sysprompt:a b/é"), "sysprompt%3Aa%20b%2F%C3%A9"); + assert_eq!(super::enc("PORT"), "PORT"); + } +} diff --git a/rust/loki-rs/src/sysinfo.rs b/rust/loki-rs/src/sysinfo.rs new file mode 100644 index 0000000..134237e --- /dev/null +++ b/rust/loki-rs/src/sysinfo.rs @@ -0,0 +1 @@ +//! Lot 4 : module en cours de portage. diff --git a/rust/loki-rs/tests/lot2.rs b/rust/loki-rs/tests/lot2.rs index 924f9ac..8977dd2 100644 --- a/rust/loki-rs/tests/lot2.rs +++ b/rust/loki-rs/tests/lot2.rs @@ -89,6 +89,10 @@ async fn banc(cle: Option<&str>) -> Banc { hostname: None, trusted: vec!["loki.exemple.fr".into()], }, + engine: loki_rs::engine::Paths { + home: std::env::temp_dir(), + service: "loki-engine".into(), + }, }; tokio::spawn(loki_rs::proxy::serve(l, cfg, async move { let _ = rx.await; diff --git a/rust/loki-rs/tests/proxy.rs b/rust/loki-rs/tests/proxy.rs index 6c2ab14..fe85af9 100644 --- a/rust/loki-rs/tests/proxy.rs +++ b/rust/loki-rs/tests/proxy.rs @@ -95,6 +95,10 @@ async fn facade(backend: SocketAddr) -> (SocketAddr, oneshot::Sender<()>) { backend, internal_token: "jeton-de-test".into(), host_env: Default::default(), + engine: loki_rs::engine::Paths { + home: std::env::temp_dir(), + service: "loki-engine".into(), + }, }; tokio::spawn(loki_rs::proxy::serve(l, cfg, async move { let _ = rx.await;