Lot 4 (socle) : store accessible à Rust, routeur /api côté façade

Le Go expose sa base à la façade par l'API interne : lecture, écriture,
suppression d'une clé et lecture d'un bucket entier. Il reste l'unique
propriétaire du fichier bbolt ; les écritures passent par putBytes/update
et invalident son cache comme une écriture locale. Base illisible : 503,
pour que Rust distingue « rien » de « pas pu lire ».

Côté Rust : client du store (seul point de passage, que la migration
finale de la base ne touchera qu'ici), routeur /api avec contexte partagé
et garde d'authentification réutilisable ; /api/ping y emménage. Modules
réservés aux aides portées ensuite (système, état du moteur, presets,
chemins, GGUF).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
MichaelandClaude Opus 5.5 committed 2026-10-14 17:40:29 +02:00
1 parent a672760f3c
commit bea29bc6a9
14 files changed
+401 -22

No files matched your search

+98
View File
@@ -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).
+57
View File
@@ -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)
}
}
+73
View File
@@ -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<String, Box<Response<Body>>> {
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<Incoming>) -> Response<Body> {
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<Body> {
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<Bytes, String> {
use http_body_util::{BodyExt, Limited};
Limited::new(body, max)
.collect()
.await
.map(|c| c.to_bytes())
.map_err(|e| e.to_string())
}
+1
View File
@@ -0,0 +1 @@
//! Lot 4 : module en cours de portage.
+1
View File
@@ -0,0 +1 @@
//! Lot 4 : module en cours de portage.
+7
View File
@@ -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;
+1
View File
@@ -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");
+1
View File
@@ -0,0 +1 @@
//! Lot 4 : module en cours de portage.
+1
View File
@@ -0,0 +1 @@
//! Lot 4 : module en cours de portage.
+16 -22
View File
@@ -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<Bytes, hyper::Error>;
@@ -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<HttpConnector, Incoming>,
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<Body> {
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<Incoming>) -> Response<Body> {
let upgrade = upgrade_protocol(req.headers());
+136
View File
@@ -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<HttpConnector, Full<Bytes>>,
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<u8>,
) -> 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<Option<Vec<u8>>, 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<u8>) -> 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<HashMap<String, String>, 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<String, String> {
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");
}
}
+1
View File
@@ -0,0 +1 @@
//! Lot 4 : module en cours de portage.
+4
View File
@@ -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;
+4
View File
@@ -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;