diff --git a/README.md b/README.md index 1e7fe6a..6b58d0d 100644 --- a/README.md +++ b/README.md @@ -232,6 +232,7 @@ redémarrages (volume `/data`). Variables d'environnement du conteneur : | `LOKI_DICTATE` | `rust` : la dictée vocale (`/api/transcribe`, `/api/dictate/*`) et son serveur de transcription sont tenus par la façade Rust (expérimental, migration en cours) | Go | | `LOKI_SETTINGS` | `rust` : les réglages partagés — préférences d'affichage, prompt système, intensité du raisonnement, adresse publique — sont servis par la façade Rust (expérimental, migration en cours) | Go | | `LOKI_FILES` | `rust` : les fichiers de la discussion — dépôt, panneau Fichiers, archive .zip, téléchargement, images et vignettes — sont servis par la façade Rust (expérimental, migration en cours) | Go | +| `LOKI_PUSH` | `rust` : les notifications Web Push (clé VAPID, abonnements, envoi « réponse prête ») sont tenues par la façade Rust (expérimental, migration en cours) | Go | **Accès par nom de domaine.** Sans clé de pilotage, l'API n'accepte que les hôtes locaux (IP, `localhost`, nom sans point, `.local`/`.lan`…) et refuse diff --git a/docs/superpowers/specs/2026-10-15-rust-migration-lot12-notifications-design.md b/docs/superpowers/specs/2026-10-15-rust-migration-lot12-notifications-design.md new file mode 100644 index 0000000..c9fdd09 --- /dev/null +++ b/docs/superpowers/specs/2026-10-15-rust-migration-lot12-notifications-design.md @@ -0,0 +1,47 @@ +# Lot 12 — notifications Web Push tenues par la façade + +## Contexte + +Les notifications « réponse prête » vivaient dans le Go : `push.go`, +`web_push.go` et webpush-go. Lot 12 : la façade les tient quand +`LOKI_PUSH=rust` (`push.rs`, expérimental ; sans la variable rien ne change). + +| Route | Rôle | +|---|---| +| `GET /api/push/key` | clé publique VAPID, générée au premier appel puis stable | +| `POST /api/push/subscribe` | enregistre un abonnement (idempotent par endpoint, clés mises à jour) | +| `POST /api/push/unsubscribe` | le retire | +| `POST /internal/facade/v1/push/send` | (le Go, jeton interne) pousse une notification à tous les abonnés, en tâche de fond | + +Les envois du Go passent tous par `sendPushToAll` : fin de tour du chat Go, +fin de tour tenu par la façade (`/internal/v1/turn/done`), fin de tâche. Ils +sont délégués à la façade. Clés et abonnements restent en base aux mêmes +clés (`push_vapid_priv`, `push_vapid_pub`, `push_subs`). + +Chiffrement RFC 8291 (`aes128gcm`), enregistrement complété à 4096 octets +comme webpush-go. Signature VAPID RFC 8292 (JWT ES256, 12 h, sujet +« mailto: »). TTL de 120 s, urgence haute. Un abonnement refusé pour de bon +(404/410) est purgé. La cryptographie passe par ring, déjà présent via +rustls. Une façade construite sans TLS répond que les notifications sont +indisponibles. + +## Écarts assumés + +- Messages des journaux d'envoi au format des crates Rust. + +## Parité + +- `internal/loki/parity_push_test.go` tire du Go : + - les durées du corps de notification ; + - onze appels de route (clé, abonnements valides et refusés, mise à jour + des clés, désabonnements), avec la liste en base après chacun. + + `tests/push.rs` rejoue le tout par la façade. La liste en base est + comparée à l'octet près, et la clé reste la même d'un appel à l'autre. +- Le chiffrement reproduit l'exemple de l'annexe A de la RFC 8291 à l'octet + près. +- Un envoi de bout en bout vers un faux service de push vérifie : + - les en-têtes ; + - la signature VAPID, contrôlée avec la clé rangée ; + - le corps, déchiffré côté navigateur avec la clé de l'abonnement ; + - l'abonnement en 410, purgé. diff --git a/internal/loki/facade_chat.go b/internal/loki/facade_chat.go index 411aee0..7e1b5d9 100644 --- a/internal/loki/facade_chat.go +++ b/internal/loki/facade_chat.go @@ -143,7 +143,7 @@ var facadeChatRoutes = map[string]bool{ // withFacadeChat remet la requête à la façade quand elle tient la // conversation — ou les routes des modèles (facade_models.go), la dictée // (facade_dictate.go), les réglages partagés (facade_settings.go), les fichiers -// de la discussion (facade_files.go). +// de la discussion (facade_files.go), les notifications (facade_push.go). // Posé APRÈS requireWebAuth : seule une requête authentifiée ici part, avec le // jeton interne qui en tient lieu là-bas. func withFacadeChat(path string, h http.HandlerFunc) http.HandlerFunc { @@ -158,6 +158,8 @@ func withFacadeChat(path string, h http.HandlerFunc) http.HandlerFunc { held, unavailable = facadeSettings, "réglages indisponibles : " case facadeFilesRoutes[path]: held, unavailable = facadeFiles, "fichiers indisponibles : " + case facadePushRoutes[path]: + held, unavailable = facadePush, "notifications indisponibles : " default: return h } diff --git a/internal/loki/facade_chat_test.go b/internal/loki/facade_chat_test.go index d45bf30..bdc8e1d 100644 --- a/internal/loki/facade_chat_test.go +++ b/internal/loki/facade_chat_test.go @@ -484,3 +484,30 @@ func TestFacadeFichiers(t *testing.T) { t.Fatalf(".part gardé sans la variable : %v", err) } } + +// LOKI_PUSH=rust : les routes des notifications partent à la façade, et les +// envois de ce process (fin de tour, fin de tâche) lui sont délégués. +func TestFacadeNotifications(t *testing.T) { + testHome(t) + f := demarrerFacade(t) + t.Setenv(facadeChatEnv, "") + t.Setenv(facadePushEnv, "rust") + local := func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "go") } + for p := range facadePushRoutes { + w := httptest.NewRecorder() + withFacadeChat(p, local)(w, httptest.NewRequest("POST", p, strings.NewReader("{}"))) + if w.Body.String() != f.reply { + t.Fatalf("%s : réponse %q", p, w.Body.String()) + } + } + sendPushToAll("Loki", "Réponse prête · 3s") + got := f.recu() + if len(got) != len(facadePushRoutes)+1 || got[len(got)-1] != `POST /internal/facade/v1/push/send jeton {"body":"Réponse prête · 3s","title":"Loki"}` { + t.Fatalf("reçu par la façade : %v", got) + } + t.Setenv(facadePushEnv, "") + sendPushToAll("Loki", "x") // aucun abonné : rien à envoyer, rien de délégué + if n := len(f.recu()); n != len(facadePushRoutes)+1 { + t.Fatalf("envoi délégué sans la variable (%d appels)", n) + } +} diff --git a/internal/loki/facade_push.go b/internal/loki/facade_push.go new file mode 100644 index 0000000..014c967 --- /dev/null +++ b/internal/loki/facade_push.go @@ -0,0 +1,43 @@ +package loki + +import ( + "fmt" + "os" + "strings" +) + +// facade_push.go — notifications Web Push tenues par la façade Rust +// (LOKI_PUSH=rust, lot 12 de la migration ; expérimental, le Go reste le +// défaut). +// +// La façade sert la clé VAPID et les (dés)abonnements, et pousse elle-même +// les notifications : les envois de ce process (fin de tour, fin de tâche) lui +// sont délégués par /internal/facade/v1/push/send. Clés et abonnements restent +// en base aux mêmes clés : hasPushSubs les lit toujours ici. + +const facadePushEnv = "LOKI_PUSH" + +// facadePush : la façade tient-elle les notifications ? +func facadePush() bool { + return strings.EqualFold(strings.TrimSpace(os.Getenv(facadePushEnv)), "rust") && + facadeAddr() != "" && internalToken() != "" +} + +// facadePushRoutes : les routes que la façade sert alors (push.rs). +var facadePushRoutes = map[string]bool{ + "/api/push/key": true, + "/api/push/subscribe": true, + "/api/push/unsubscribe": true, +} + +// facadePushSend délègue un envoi à la façade ; false quand elle ne tient pas +// les notifications (l'appelant envoie alors lui-même). +func facadePushSend(title, body string) bool { + if !facadePush() { + return false + } + if err := facadePost("/internal/facade/v1/push/send", map[string]any{"title": title, "body": body}); err != nil { + fmt.Printf("[push] envoi délégué à la façade : %v\n", err) + } + return true +} diff --git a/internal/loki/parity_push_test.go b/internal/loki/parity_push_test.go new file mode 100644 index 0000000..976b9c6 --- /dev/null +++ b/internal/loki/parity_push_test.go @@ -0,0 +1,109 @@ +package loki + +import ( + "encoding/json" + "net/http/httptest" + "os" + "path/filepath" + "regexp" + "strings" + "testing" + "time" +) + +// parity_push_test.go — journal de référence des notifications Web Push (lot +// 12 de la migration Rust) : clé VAPID (générée une fois, stable), abonnement, +// désabonnement, et le texte du corps de notification. Rangé dans +// rust/loki-rs/tests/parity_push/push.json ; tests/push.rs le rejoue par la +// façade. Les clés tirées au hasard sont masquées (seule leur forme compte). +// +// LOKI_PARITY_UPDATE=1 go test ./internal/loki -run TestPariteNotifications + +const parityPushFile = "../../rust/loki-rs/tests/parity_push/push.json" + +type parityPushCall struct { + Method string `json:"method"` + Path string `json:"path"` + Body string `json:"body,omitempty"` + Status int `json:"status"` + Out any `json:"out"` + // Après l'appel : les abonnements enregistrés (JSON brut, tel qu'en base). + Subs string `json:"subs"` +} + +type parityPushFixture struct { + Durations map[string]string `json:"durations"` + Calls []parityPushCall `json:"calls"` +} + +func TestPariteNotifications(t *testing.T) { + testHome(t) + var fx parityPushFixture + fx.Durations = map[string]string{} + for _, ms := range []int64{0, 400, 600, 42_000, 185_000, 3_600_000, 4_330_000, -5_000} { + fx.Durations[strings.TrimSpace(time.Duration(ms*int64(time.Millisecond)).String())] = fmtDurFR(time.Duration(ms) * time.Millisecond) + } + keyRe := regexp.MustCompile(`"key":"[A-Za-z0-9_-]{87}"`) + call := func(method, path, body string) parityPushCall { + w := httptest.NewRecorder() + r := httptest.NewRequest(method, path, strings.NewReader(body)) + switch path { + case "/api/push/key": + handlePushKey(w, r) + case "/api/push/subscribe": + handlePushSubscribe(w, r) + case "/api/push/unsubscribe": + handlePushUnsubscribe(w, r) + default: + t.Fatalf("route inattendue %s", path) + } + out := keyRe.ReplaceAllString(w.Body.String(), `"key":"{{CLE}}"`) + c := parityPushCall{Method: method, Path: path, Body: body, Status: w.Code, Subs: string(getBytes(bkState, stPushSubs))} + if err := json.Unmarshal([]byte(out), &c.Out); err != nil { + t.Fatalf("%s : %v (%s)", path, err, w.Body.String()) + } + return c + } + for _, c := range []struct{ m, p, b string }{ + {"GET", "/api/push/key", ""}, + {"POST", "/api/push/subscribe", `{"endpoint":"https://push.exemple.fr/a","keys":{"p256dh":"BAAA","auth":"x1"}}`}, + {"POST", "/api/push/subscribe", `{"endpoint":"https://push.exemple.fr/b","keys":{"p256dh":"BBBB","auth":"x2"},"expirationTime":null}`}, + {"POST", "/api/push/subscribe", `{"endpoint":"https://push.exemple.fr/a","keys":{"p256dh":"BCCC","auth":"x3"}}`}, + {"POST", "/api/push/subscribe", `{"keys":{"p256dh":"B","auth":"y"}}`}, + {"POST", "/api/push/subscribe", `pas du json`}, + {"POST", "/api/push/unsubscribe", `{"endpoint":"https://push.exemple.fr/inconnu"}`}, + {"POST", "/api/push/unsubscribe", `{"endpoint":"https://push.exemple.fr/a"}`}, + {"POST", "/api/push/unsubscribe", `pas du json`}, + {"POST", "/api/push/unsubscribe", `{"endpoint":"https://push.exemple.fr/b"}`}, + {"GET", "/api/push/key", ""}, + } { + fx.Calls = append(fx.Calls, call(c.m, c.p, c.b)) + } + // La paire VAPID est stable : la clé servie est la clé rangée. + if pub := getStr(bkState, stPushVAPIDPub); len(pub) != 87 || len(getStr(bkState, stPushVAPIDPriv)) != 43 { + t.Fatalf("paire VAPID inattendue : %q", pub) + } + + got, err := parityJSON(fx) + if err != nil { + t.Fatal(err) + } + if os.Getenv("LOKI_PARITY_UPDATE") == "1" { + if err := os.MkdirAll(filepath.Dir(parityPushFile), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(parityPushFile, got, 0o644); err != nil { + t.Fatal(err) + } + return + } + want, err := os.ReadFile(parityPushFile) + if err != nil { + t.Fatalf("journal de référence absent (%v) : LOKI_PARITY_UPDATE=1 pour le créer", err) + } + if string(got) != string(want) { + t.Errorf("les notifications ne produisent plus le journal rangé dans %s.\n"+ + "Si c'est voulu : LOKI_PARITY_UPDATE=1 go test ./internal/loki -run TestPariteNotifications,\n"+ + "puis cargo test --test push dit si le port Rust suit.\n--- obtenu ---\n%s", parityPushFile, got) + } +} diff --git a/internal/loki/push.go b/internal/loki/push.go index 5f5a69a..5763283 100644 --- a/internal/loki/push.go +++ b/internal/loki/push.go @@ -150,6 +150,10 @@ func hasPushSubs() bool { return len(loadSubs()) > 0 } // abonnement rejeté définitivement (404/410) est purgé pour ne pas s'accumuler. // Appelé depuis la goroutine de génération, à turn_done — jamais bloquant pour l'UI. func sendPushToAll(title, body string) { + // Notifications tenues par la façade (LOKI_PUSH=rust) : c'est elle qui pousse. + if facadePushSend(title, body) { + return + } subs := loadSubs() if len(subs) == 0 { fmt.Printf("[push] fin de tour : aucun abonné enregistré (rien à envoyer)\n") diff --git a/internal/loki/web_server.go b/internal/loki/web_server.go index d5538c3..aaef0c7 100644 --- a/internal/loki/web_server.go +++ b/internal/loki/web_server.go @@ -223,8 +223,9 @@ func newWebMux() *http.ServeMux { // conversation (LOKI_CHAT=rust, facade_chat.go), celles des modèles quand // elle les tient (LOKI_MODELS=rust, facade_models.go), celles de la dictée // (LOKI_DICTATE=rust, facade_dictate.go), des réglages partagés - // (LOKI_SETTINGS=rust, facade_settings.go) et des fichiers de la - // discussion (LOKI_FILES=rust, facade_files.go) ; sans effet sinon. + // (LOKI_SETTINGS=rust, facade_settings.go), des fichiers de la discussion + // (LOKI_FILES=rust, facade_files.go) et des notifications (LOKI_PUSH=rust, + // facade_push.go) ; sans effet sinon. api := func(path string, h http.HandlerFunc) { mux.HandleFunc(path, requireWebAuth(withFacadeChat(path, h))) } api("/api/ping", handlePing) api("/api/status", handleStatus) diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 0bbcbed..b07e29f 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -792,6 +792,7 @@ dependencies = [ "libc", "regex", "reqwest", + "ring", "rustls", "serde", "serde_json", diff --git a/rust/loki-rs/Cargo.toml b/rust/loki-rs/Cargo.toml index 81ab438..6e3dd4c 100644 --- a/rust/loki-rs/Cargo.toml +++ b/rust/loki-rs/Cargo.toml @@ -18,6 +18,7 @@ hyper-util = { version = "0.1.21", features = ["tokio", "server", "client-legacy regex = "1.13.1" url = "2.5.8" zip = { version = "2.4.2", default-features = false, features = ["deflate-flate2", "flate2"] } +ring = { version = "0.17.14", optional = true } reqwest = { version = "0.13.5", default-features = false, features = ["rustls-no-provider", "stream", "http2"] } rustls = { version = "0.23.46", default-features = false, features = ["ring", "std", "tls12"], optional = true } serde = { version = "1.0.229", features = ["derive"] } @@ -32,7 +33,7 @@ tokio-tungstenite = "0.30.0" # Crypto TLS (ring) pour les API externes en HTTPS. Activée par défaut ; le # contrôle croisé Linux sous Windows s'en passe (pas d'en-têtes libc Linux). default = ["tls"] -tls = ["dep:rustls"] +tls = ["dep:rustls", "dep:ring"] [target.'cfg(unix)'.dependencies] libc = "0.2" diff --git a/rust/loki-rs/src/api.rs b/rust/loki-rs/src/api.rs index c812991..0e34e26 100644 --- a/rust/loki-rs/src/api.rs +++ b/rust/loki-rs/src/api.rs @@ -36,6 +36,8 @@ pub struct Ctx { pub settings: bool, /// Fichiers de la discussion tenus par la façade (`LOKI_FILES=rust`). pub files: Option>, + /// Notifications Web Push tenues par la façade (`LOKI_PUSH=rust`). + pub push: Option>, /// Jeton de l'API interne : une requête qui le porte vient du Go. pub internal_token: String, } @@ -49,6 +51,7 @@ impl Ctx { || self.dictate.is_some() || self.settings || self.files.is_some() + || self.push.is_some() } /// La requête vient du Go (jeton interne valide) : déjà authentifiée par diff --git a/rust/loki-rs/src/lib.rs b/rust/loki-rs/src/lib.rs index e3dcf35..f3d2198 100644 --- a/rust/loki-rs/src/lib.rs +++ b/rust/loki-rs/src/lib.rs @@ -27,6 +27,7 @@ pub mod oai; pub mod paths; pub mod presets; pub mod proxy; +pub mod push; pub mod settings; pub mod store; pub mod sysinfo; diff --git a/rust/loki-rs/src/main.rs b/rust/loki-rs/src/main.rs index ec19d76..ee6d1c4 100644 --- a/rust/loki-rs/src/main.rs +++ b/rust/loki-rs/src/main.rs @@ -130,6 +130,9 @@ fn serve() -> ExitCode { // LOKI_FILES=rust : les fichiers de la discussion (dépôt, panneau // Fichiers, archive, téléchargement, images). let files_rust = held_by_facade("LOKI_FILES"); + // LOKI_PUSH=rust : les notifications Web Push (le Go lui délègue ses + // envois de fin de tour et de fin de tâche). + let push_rust = held_by_facade("LOKI_PUSH"); let mut env = vec![(auth::INTERNAL_ENV.to_string(), token.clone())]; if chat_rust { env.push(("LOKI_CHAT".into(), "rust".into())); @@ -146,7 +149,10 @@ fn serve() -> ExitCode { if files_rust { env.push(("LOKI_FILES".into(), "rust".into())); } - if chat_rust || models_rust || dictate_rust || settings_rust || files_rust { + if push_rust { + env.push(("LOKI_PUSH".into(), "rust".into())); + } + if chat_rust || models_rust || dictate_rust || settings_rust || files_rust || push_rust { env.push(("LOKI_FACADE_ADDR".into(), format!("127.0.0.1:{web_port}"))); } let sup = child::Supervisor::spawn(go_bin, go_port, env); @@ -216,6 +222,7 @@ fn serve() -> ExitCode { dictate, settings: settings_rust, files, + push: push_rust.then(|| std::sync::Arc::new(loki_rs::push::Push::new())), }; 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/proxy.rs b/rust/loki-rs/src/proxy.rs index 5197da7..bd38044 100644 --- a/rust/loki-rs/src/proxy.rs +++ b/rust/loki-rs/src/proxy.rs @@ -73,6 +73,8 @@ pub struct Config { pub settings: bool, /// Fichiers de la discussion tenus par la façade (`LOKI_FILES=rust`). pub files: Option>, + /// Notifications Web Push tenues par la façade (`LOKI_PUSH=rust`). + pub push: Option>, } struct Proxy { @@ -104,6 +106,7 @@ pub async fn serve( dictate: cfg.dictate, settings: cfg.settings, files: cfg.files, + push: cfg.push, internal_token: cfg.internal_token.clone(), }, }); @@ -159,6 +162,12 @@ impl Proxy { { return chatapi::serve_internal(chat, req).await; } + if from_go + && let Some(p) = &self.ctx.push + && path == crate::push::INTERNAL_SEND + { + return crate::push::serve_internal(self.ctx.store.clone(), p.clone(), req).await; + } if from_go && let Some(d) = &self.ctx.dictate && path == crate::dictate::INTERNAL_STOP @@ -185,6 +194,11 @@ impl Proxy { { return chatapi::serve(&self.ctx, chat, req, from_go).await; } + if let Some(p) = &self.ctx.push + && crate::push::handles(path) + { + return crate::push::serve(&self.ctx, p, req, from_go).await; + } if let Some(f) = &self.ctx.files && crate::convfiles::handles(path) { diff --git a/rust/loki-rs/src/push.rs b/rust/loki-rs/src/push.rs new file mode 100644 index 0000000..4e0403d --- /dev/null +++ b/rust/loki-rs/src/push.rs @@ -0,0 +1,608 @@ +//! Notifications Web Push — port de `push.go` et `web_push.go` (lot 12 de la +//! migration), tenues par la façade quand `LOKI_PUSH=rust` (expérimental ; le +//! Go par défaut) : +//! +//! - `GET /api/push/key` : clé publique VAPID (générée une fois, stable) ; +//! - `POST /api/push/subscribe` : enregistre un abonnement du navigateur ; +//! - `POST /api/push/unsubscribe` : le retire ; +//! - `POST /internal/facade/v1/push/send` (le Go, jeton interne) : pousse +//! « réponse prête » à tous les abonnés — le Go y délègue ses envois de fin +//! de tour et de fin de tâche. +//! +//! Le chiffrement est celui de RFC 8291 (`aes128gcm`, enregistrement complété +//! à 4096 octets comme webpush-go), signé VAPID (JWT ES256, 12 h). Mêmes clés +//! en base que le Go (`push_vapid_priv`, `push_vapid_pub`, `push_subs`) : un +//! abonnement pris d'un côté vaut de l'autre. +//! +//! Écart assumé : messages des journaux d'envoi au format des crates Rust. + +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use hyper::body::Incoming; +use hyper::{Request, Response, StatusCode}; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; + +use crate::api::Ctx; +use crate::auth::json as send_json; +use crate::proxy::Body; +use crate::store::{BK_STATE, Store}; + +const ST_PRIV: &str = "push_vapid_priv"; +const ST_PUB: &str = "push_vapid_pub"; +const ST_SUBS: &str = "push_subs"; +/// Taille d'enregistrement (`MaxRecordSize` de webpush-go). +pub const RECORD_SIZE: usize = 4096; +/// Sujet VAPID : l'adresse nue, « mailto: » ajouté à la signature. +#[cfg_attr(not(feature = "tls"), allow(dead_code))] +const SUBSCRIBER: &str = "mail@nathaninline.com"; +/// Périmé après 2 min : une notification tardive n'a pas de sens. +const TTL: u32 = 120; + +/// La route interne d'envoi. +pub const INTERNAL_SEND: &str = "/internal/facade/v1/push/send"; + +/// Routes servies ici. +pub fn handles(path: &str) -> bool { + matches!( + path, + "/api/push/key" | "/api/push/subscribe" | "/api/push/unsubscribe" + ) +} + +/// Clés d'un abonnement (`webpush.Keys`). +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] +pub struct Keys { + #[serde(default)] + pub auth: String, + #[serde(default)] + pub p256dh: String, +} + +/// Un abonnement du navigateur (`pushSub`). +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] +pub struct PushSub { + #[serde(default)] + pub endpoint: String, + #[serde(default)] + pub keys: Keys, +} + +/// `fmtDurFR` : « 42s », « 3 mn 05s », « 1 h 12 mn ». +pub fn fmt_dur_fr(ms: i64) -> String { + // d.Round(time.Second) : à la seconde la plus proche, moitié vers le haut. + let secs = ((ms.max(0) + 500) / 1000).max(0); + let (h, m, s) = (secs / 3600, (secs % 3600) / 60, secs % 60); + if h > 0 { + format!("{h} h {m:02} mn") + } else if m > 0 { + format!("{m} mn {s:02}s") + } else { + format!("{s}s") + } +} + +// --------------------------------------------------------------------------- +// Chiffrement (RFC 8291) et signature (RFC 8292) +// --------------------------------------------------------------------------- + +#[cfg(feature = "tls")] +mod crypto { + use super::{PushSub, RECORD_SIZE, SUBSCRIBER}; + use base64::Engine; + use base64::engine::general_purpose::{STANDARD, URL_SAFE, URL_SAFE_NO_PAD}; + use ring::rand::SystemRandom; + use ring::signature::{ECDSA_P256_SHA256_FIXED_SIGNING, EcdsaKeyPair, KeyPair}; + use ring::{aead, agreement, hkdf}; + use serde_json::json; + + /// `decodeSubscriptionKey` : base64 standard, sinon URL, remplissage ajouté. + pub fn decode_sub_key(key: &str) -> Result, String> { + let mut k = key.to_string(); + let rem = k.len() % 4; + if rem != 0 { + k.push_str(&"=".repeat(4 - rem)); + } + STANDARD + .decode(&k) + .or_else(|_| URL_SAFE.decode(&k)) + .map_err(|e| e.to_string()) + } + + /// `decodeVapidKey` : base64 URL avec ou sans remplissage. + fn decode_vapid_key(key: &str) -> Result, String> { + URL_SAFE + .decode(key) + .or_else(|_| URL_SAFE_NO_PAD.decode(key)) + .map_err(|e| e.to_string()) + } + + struct Len(usize); + + impl hkdf::KeyType for Len { + fn len(&self) -> usize { + self.0 + } + } + + fn hkdf_expand(salt: &[u8], ikm: &[u8], info: &[u8], len: usize) -> Result, String> { + let prk = hkdf::Salt::new(hkdf::HKDF_SHA256, salt).extract(ikm); + let parts = [info]; + let okm = prk + .expand(&parts, Len(len)) + .map_err(|_| "hkdf".to_string())?; + let mut out = vec![0u8; len]; + okm.fill(&mut out).map_err(|_| "hkdf".to_string())?; + Ok(out) + } + + /// Le corps chiffré d'une notification : en-tête `aes128gcm` (sel, taille + /// d'enregistrement, clé publique éphémère) puis le message complété de + /// `0x02` et de zéros jusqu'à remplir l'enregistrement, comme webpush-go. + /// + /// `local_public` et `shared` : clé publique éphémère et secret ECDH — + /// tirés au hasard en production, imposés par les tests. `pad` : complété + /// jusqu'à remplir l'enregistrement (webpush-go), ou pas (exemple de la + /// RFC 8291). + #[allow(clippy::too_many_arguments)] + pub fn encrypt_with( + message: &[u8], + auth: &[u8], + ua_public: &[u8], + salt: &[u8; 16], + local_public: &[u8], + shared: &[u8], + record_size: usize, + pad: bool, + ) -> Result, String> { + let mut info = b"WebPush: info\0".to_vec(); + info.extend_from_slice(ua_public); + info.extend_from_slice(local_public); + let ikm = hkdf_expand(auth, shared, &info, 32)?; + let cek = hkdf_expand(salt, &ikm, b"Content-Encoding: aes128gcm\0", 16)?; + let nonce = hkdf_expand(salt, &ikm, b"Content-Encoding: nonce\0", 12)?; + let mut header = salt.to_vec(); + header.extend_from_slice(&(record_size as u32).to_be_bytes()); + header.push(local_public.len() as u8); + header.extend_from_slice(local_public); + let record_len = record_size - 16; + let mut data = message.to_vec(); + data.push(0x02); + let target = record_len.saturating_sub(header.len()); + if data.len() > target { + return Err("payload has exceeded the maximum length".into()); + } + if pad { + data.resize(target, 0); + } + let key = aead::LessSafeKey::new( + aead::UnboundKey::new(&aead::AES_128_GCM, &cek).map_err(|_| "aes".to_string())?, + ); + let nonce = + aead::Nonce::try_assume_unique_for_key(&nonce).map_err(|_| "nonce".to_string())?; + key.seal_in_place_append_tag(nonce, aead::Aad::empty(), &mut data) + .map_err(|_| "chiffrement".to_string())?; + header.extend_from_slice(&data); + Ok(header) + } + + /// [`encrypt_with`] avec un sel et une clé éphémère tirés au hasard. + pub fn encrypt(message: &[u8], sub: &PushSub) -> Result, String> { + let auth = decode_sub_key(&sub.keys.auth)?; + let ua = decode_sub_key(&sub.keys.p256dh)?; + let rng = SystemRandom::new(); + let mut salt = [0u8; 16]; + ring::rand::SecureRandom::fill(&rng, &mut salt).map_err(|_| "aléa".to_string())?; + let local = agreement::EphemeralPrivateKey::generate(&agreement::ECDH_P256, &rng) + .map_err(|_| "clé éphémère".to_string())?; + let local_pub = local + .compute_public_key() + .map_err(|_| "clé éphémère".to_string())? + .as_ref() + .to_vec(); + let peer = agreement::UnparsedPublicKey::new(&agreement::ECDH_P256, ua.clone()); + let shared = agreement::agree_ephemeral(local, &peer, |s| s.to_vec()).map_err(|_| { + "Unmarshal Error: Public key is not a valid point on the curve".to_string() + })?; + encrypt_with( + message, + &auth, + &ua, + &salt, + &local_pub, + &shared, + RECORD_SIZE, + true, + ) + } + + /// `getVAPIDAuthorizationHeader` : « vapid t=, k= ». + pub fn vapid_header( + endpoint: &str, + public: &str, + private: &str, + now: u64, + ) -> Result { + let u = reqwest::Url::parse(endpoint).map_err(|e| e.to_string())?; + let host = match u.port() { + Some(p) => format!("{}:{p}", u.host_str().unwrap_or("")), + None => u.host_str().unwrap_or("").to_string(), + }; + let sub = if SUBSCRIBER.starts_with("https:") { + SUBSCRIBER.to_string() + } else { + format!("mailto:{SUBSCRIBER}") + }; + // json.Marshal d'une MapClaims : clés triées. + let claims = json!({ "aud": format!("{}://{host}", u.scheme()), "exp": now + 12 * 3600, "sub": sub }); + let signing_input = format!( + "{}.{}", + URL_SAFE_NO_PAD.encode(br#"{"alg":"ES256","typ":"JWT"}"#), + URL_SAFE_NO_PAD.encode(claims.to_string()) + ); + let priv_raw = decode_vapid_key(private)?; + let pub_raw = decode_vapid_key(public)?; + let rng = SystemRandom::new(); + let kp = EcdsaKeyPair::from_private_key_and_public_key( + &ECDSA_P256_SHA256_FIXED_SIGNING, + &priv_raw, + &pub_raw, + &rng, + ) + .map_err(|e| format!("clé VAPID illisible : {e}"))?; + let sig = kp + .sign(&rng, signing_input.as_bytes()) + .map_err(|_| "signature VAPID".to_string())?; + Ok(format!( + "vapid t={signing_input}.{}, k={}", + URL_SAFE_NO_PAD.encode(sig.as_ref()), + URL_SAFE_NO_PAD.encode(&pub_raw) + )) + } + + /// `GenerateVAPIDKeys` : (privée 32 octets, publique 65 octets), base64 URL + /// sans remplissage. + pub fn generate_vapid_keys() -> Result<(String, String), String> { + let rng = SystemRandom::new(); + let doc = EcdsaKeyPair::generate_pkcs8(&ECDSA_P256_SHA256_FIXED_SIGNING, &rng) + .map_err(|_| "génération VAPID".to_string())?; + let kp = EcdsaKeyPair::from_pkcs8(&ECDSA_P256_SHA256_FIXED_SIGNING, doc.as_ref(), &rng) + .map_err(|_| "génération VAPID".to_string())?; + let public = kp.public_key().as_ref().to_vec(); + // ECPrivateKey dans le PKCS#8 : « 04 20 » puis les 32 octets du scalaire. + let d = doc.as_ref(); + let i = d + .windows(2) + .position(|w| w == [0x04, 0x20]) + .ok_or("génération VAPID")?; + let private = d.get(i + 2..i + 34).ok_or("génération VAPID")?.to_vec(); + Ok(( + URL_SAFE_NO_PAD.encode(private), + URL_SAFE_NO_PAD.encode(public), + )) + } +} + +#[cfg(not(feature = "tls"))] +mod crypto { + //! Sans `tls` (pas de ring) : pas de notifications. + use super::PushSub; + const NO: &str = "notifications indisponibles (façade construite sans TLS)"; + pub fn decode_sub_key(_: &str) -> Result, String> { + Err(NO.into()) + } + pub fn encrypt(_: &[u8], _: &PushSub) -> Result, String> { + Err(NO.into()) + } + pub fn vapid_header(_: &str, _: &str, _: &str, _: u64) -> Result { + Err(NO.into()) + } + pub fn generate_vapid_keys() -> Result<(String, String), String> { + Err(NO.into()) + } +} + +pub use crypto::*; + +// --------------------------------------------------------------------------- +// Service +// --------------------------------------------------------------------------- + +/// Les notifications tenues par la façade. +pub struct Push { + client: reqwest::Client, + /// Génération paresseuse des clés : une seule paire, même sous la course. + vapid: tokio::sync::Mutex<()>, + /// Lecture-modification-écriture de la liste des abonnements. + subs: tokio::sync::Mutex<()>, +} + +impl Default for Push { + fn default() -> Self { + Self::new() + } +} + +/// Hôte d'un endpoint pour les journaux (jamais le jeton entier). +fn endpoint_host(endpoint: &str) -> String { + reqwest::Url::parse(endpoint) + .ok() + .and_then(|u| { + u.host_str().map(|h| match u.port() { + Some(p) => format!("{h}:{p}"), + None => h.to_string(), + }) + }) + .filter(|h| !h.is_empty()) + .unwrap_or_else(|| "?".into()) +} + +impl Push { + pub fn new() -> Self { + crate::chat::engine::install_crypto(); + Push { + client: reqwest::Client::builder() + .timeout(Duration::from_secs(30)) + .build() + .unwrap_or_default(), + vapid: tokio::sync::Mutex::new(()), + subs: tokio::sync::Mutex::new(()), + } + } + + /// `vapidKeys` : la paire, générée et rangée au premier appel. + pub async fn vapid_keys(&self, store: &Store) -> Result<(String, String), String> { + let _g = self.vapid.lock().await; + let private = store.get_str(BK_STATE, ST_PRIV).await; + let public = store.get_str(BK_STATE, ST_PUB).await; + if !private.is_empty() && !public.is_empty() { + return Ok((private, public)); + } + let (private, public) = generate_vapid_keys()?; + store + .put(BK_STATE, ST_PRIV, private.clone().into_bytes()) + .await?; + store + .put(BK_STATE, ST_PUB, public.clone().into_bytes()) + .await?; + Ok((private, public)) + } + + async fn load_subs(store: &Store) -> Vec { + store + .get(BK_STATE, ST_SUBS) + .await + .ok() + .flatten() + .and_then(|b| { + serde_json::from_slice::>>(&b) + .ok() + .flatten() + }) + .unwrap_or_default() + } + + async fn save_subs(store: &Store, subs: &[PushSub]) -> Result<(), String> { + let raw = serde_json::to_vec(subs).map_err(|e| e.to_string())?; + store.put(BK_STATE, ST_SUBS, raw).await + } + + /// `addSub` : idempotent par endpoint (les clés sont mises à jour). + pub async fn add_sub(&self, store: &Store, s: PushSub) -> Result { + let _g = self.subs.lock().await; + let mut subs = Self::load_subs(store).await; + match subs.iter_mut().find(|x| x.endpoint == s.endpoint) { + Some(x) => *x = s, + None => subs.push(s), + } + Self::save_subs(store, &subs).await?; + Ok(subs.len()) + } + + /// `removeSub`. + pub async fn remove_sub(&self, store: &Store, endpoint: &str) { + if endpoint.is_empty() { + return; + } + let _g = self.subs.lock().await; + let mut subs = Self::load_subs(store).await; + subs.retain(|x| x.endpoint != endpoint); + let _ = Self::save_subs(store, &subs).await; + } + + /// `sendPushToAll` : best-effort ; un abonnement rejeté pour de bon + /// (404/410) est purgé. + pub async fn send_all(&self, store: &Store, title: &str, body: &str) { + let subs = Self::load_subs(store).await; + if subs.is_empty() { + println!("[push] fin de tour : aucun abonné enregistré (rien à envoyer)"); + return; + } + let (private, public) = match self.vapid_keys(store).await { + Ok(k) => k, + Err(e) => { + println!("[push] clés VAPID indisponibles : {e}"); + return; + } + }; + #[derive(Serialize)] + struct Payload<'a> { + title: &'a str, + body: &'a str, + tag: &'a str, + } + let msg = serde_json::to_vec(&Payload { + title, + body, + tag: "loki-turn", + }) + .unwrap_or_default(); + println!("[push] envoi à {} abonné(s)…", subs.len()); + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |d| d.as_secs()); + for s in subs { + let host = endpoint_host(&s.endpoint); + let sent = async { + let body = encrypt(&msg, &s)?; + let auth = vapid_header(&s.endpoint, &public, &private, now)?; + self.client + .post(&s.endpoint) + .header("Content-Encoding", "aes128gcm") + .header("Content-Type", "application/octet-stream") + .header("TTL", TTL.to_string()) + .header("Urgency", "high") + .header("Authorization", auth) + .body(body) + .send() + .await + .map_err(|e| e.to_string()) + } + .await; + let resp = match sent { + Ok(r) => r, + Err(e) => { + println!("[push] échec envoi ({host}) : {e}"); + continue; + } + }; + let code = resp.status().as_u16(); + if code >= 300 { + let text = resp.text().await.unwrap_or_default(); + let text: String = text.chars().take(512).collect(); + println!("[push] refus {code} ({host}) : {}", text.trim()); + } else { + println!("[push] accepté {code} ({host})"); + } + if code == 404 || code == 410 { + self.remove_sub(store, &s.endpoint).await; + } + } + } +} + +// --------------------------------------------------------------------------- +// Routes +// --------------------------------------------------------------------------- + +async fn body_json(body: Incoming) -> Option { + use http_body_util::BodyExt; + let b = http_body_util::Limited::new(body, 1 << 20) + .collect() + .await + .ok()? + .to_bytes(); + serde_json::from_slice(&b).ok() +} + +/// `from_go` : remise par le Go (tunnel), déjà authentifiée par lui. +pub async fn serve( + ctx: &Ctx, + push: &std::sync::Arc, + req: Request, + from_go: bool, +) -> Response { + let (parts, body) = req.into_parts(); + if !from_go && let Err(refus) = ctx.guard(&parts.headers).await { + return *refus; + } + let fail = |code: StatusCode, e: String| send_json(code, &json!({ "ok": false, "error": e })); + match parts.uri.path() { + "/api/push/key" => match push.vapid_keys(&ctx.store).await { + Ok((_, public)) => send_json(StatusCode::OK, &json!({ "ok": true, "key": public })), + Err(e) => fail(StatusCode::INTERNAL_SERVER_ERROR, e), + }, + "/api/push/subscribe" => { + let s: PushSub = match body_json(body) + .await + .and_then(|v| serde_json::from_value::(v).ok()) + { + Some(s) if !s.endpoint.is_empty() => s, + _ => return fail(StatusCode::BAD_REQUEST, "abonnement invalide".into()), + }; + let host = endpoint_host(&s.endpoint); + match push.add_sub(&ctx.store, s).await { + Ok(n) => { + println!("[push] abonnement enregistré ({host}) — {n} au total"); + send_json(StatusCode::OK, &json!({ "ok": true })) + } + Err(e) => { + println!("[push] abonnement REFUSÉ (stockage) : {e}"); + fail(StatusCode::INTERNAL_SERVER_ERROR, e) + } + } + } + "/api/push/unsubscribe" => { + let ep = body_json(body) + .await + .and_then(|v| { + v.get("endpoint") + .and_then(Value::as_str) + .map(str::to_string) + }) + .unwrap_or_default(); + push.remove_sub(&ctx.store, &ep).await; + send_json(StatusCode::OK, &json!({ "ok": true })) + } + _ => send_json( + StatusCode::NOT_FOUND, + &json!({ "error": "404 page not found" }), + ), + } +} + +/// `POST /internal/facade/v1/push/send {title, body}` : envoi en tâche de +/// fond (le Go n'attend pas les services de push). +pub async fn serve_internal( + store: Store, + push: std::sync::Arc, + req: Request, +) -> Response { + let v = body_json(req.into_body()).await.unwrap_or(Value::Null); + let title = v + .get("title") + .and_then(Value::as_str) + .unwrap_or("Loki") + .to_string(); + let body = v + .get("body") + .and_then(Value::as_str) + .unwrap_or("") + .to_string(); + tokio::spawn(async move { push.send_all(&store, &title, &body).await }); + use http_body_util::BodyExt; + let mut r = Response::new( + http_body_util::Full::new(bytes::Bytes::new()) + .map_err(|n| match n {}) + .boxed(), + ); + *r.status_mut() = StatusCode::NO_CONTENT; + r +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn durees() { + assert_eq!(fmt_dur_fr(42_000), "42s"); + assert_eq!(fmt_dur_fr(185_000), "3 mn 05s"); + assert_eq!(fmt_dur_fr(4_330_000), "1 h 12 mn"); + assert_eq!(fmt_dur_fr(600), "1s"); + assert_eq!(fmt_dur_fr(-5), "0s"); + } + + #[cfg(feature = "tls")] + #[test] + fn cles_vapid() { + let (p, q) = generate_vapid_keys().unwrap(); + assert_eq!((p.len(), q.len()), (43, 87)); + let h = vapid_header("https://push.exemple.fr:8443/x/y", &q, &p, 1_700_000_000).unwrap(); + assert!( + h.starts_with("vapid t=eyJhbGciOiJFUzI1NiIsInR5cCI6IkpXVCJ9."), + "{h}" + ); + assert!(h.ends_with(&format!(", k={q}"))); + } +} diff --git a/rust/loki-rs/tests/common/fakego.rs b/rust/loki-rs/tests/common/fakego.rs index 766ba26..08a3b24 100644 --- a/rust/loki-rs/tests/common/fakego.rs +++ b/rust/loki-rs/tests/common/fakego.rs @@ -473,6 +473,7 @@ pub struct Opts { pub dictate: Option>, pub settings: bool, pub files: Option>, + pub push: Option>, } /// Comme [`start`], avec la conversation tenue par la façade @@ -566,6 +567,7 @@ pub async fn start_with( dictate: opts.dictate, settings: opts.settings, files: opts.files, + push: opts.push, }; tokio::spawn(loki_rs::proxy::serve(fl, cfg, async move { let _ = rx.await; diff --git a/rust/loki-rs/tests/lot2.rs b/rust/loki-rs/tests/lot2.rs index d52cce2..826f9a8 100644 --- a/rust/loki-rs/tests/lot2.rs +++ b/rust/loki-rs/tests/lot2.rs @@ -98,6 +98,7 @@ async fn banc(cle: Option<&str>) -> Banc { dictate: None, settings: false, files: None, + push: None, }; tokio::spawn(loki_rs::proxy::serve(l, cfg, async move { let _ = rx.await; diff --git a/rust/loki-rs/tests/parity_push/push.json b/rust/loki-rs/tests/parity_push/push.json new file mode 100644 index 0000000..3d7577b --- /dev/null +++ b/rust/loki-rs/tests/parity_push/push.json @@ -0,0 +1,126 @@ +{ + "durations": { + "-5s": "0s", + "0s": "0s", + "1h0m0s": "1 h 00 mn", + "1h12m10s": "1 h 12 mn", + "3m5s": "3 mn 05s", + "400ms": "0s", + "42s": "42s", + "600ms": "1s" + }, + "calls": [ + { + "method": "GET", + "path": "/api/push/key", + "status": 200, + "out": { + "key": "{{CLE}}", + "ok": true + }, + "subs": "" + }, + { + "method": "POST", + "path": "/api/push/subscribe", + "body": "{\"endpoint\":\"https://push.exemple.fr/a\",\"keys\":{\"p256dh\":\"BAAA\",\"auth\":\"x1\"}}", + "status": 200, + "out": { + "ok": true + }, + "subs": "[{\"endpoint\":\"https://push.exemple.fr/a\",\"keys\":{\"auth\":\"x1\",\"p256dh\":\"BAAA\"}}]" + }, + { + "method": "POST", + "path": "/api/push/subscribe", + "body": "{\"endpoint\":\"https://push.exemple.fr/b\",\"keys\":{\"p256dh\":\"BBBB\",\"auth\":\"x2\"},\"expirationTime\":null}", + "status": 200, + "out": { + "ok": true + }, + "subs": "[{\"endpoint\":\"https://push.exemple.fr/a\",\"keys\":{\"auth\":\"x1\",\"p256dh\":\"BAAA\"}},{\"endpoint\":\"https://push.exemple.fr/b\",\"keys\":{\"auth\":\"x2\",\"p256dh\":\"BBBB\"}}]" + }, + { + "method": "POST", + "path": "/api/push/subscribe", + "body": "{\"endpoint\":\"https://push.exemple.fr/a\",\"keys\":{\"p256dh\":\"BCCC\",\"auth\":\"x3\"}}", + "status": 200, + "out": { + "ok": true + }, + "subs": "[{\"endpoint\":\"https://push.exemple.fr/a\",\"keys\":{\"auth\":\"x3\",\"p256dh\":\"BCCC\"}},{\"endpoint\":\"https://push.exemple.fr/b\",\"keys\":{\"auth\":\"x2\",\"p256dh\":\"BBBB\"}}]" + }, + { + "method": "POST", + "path": "/api/push/subscribe", + "body": "{\"keys\":{\"p256dh\":\"B\",\"auth\":\"y\"}}", + "status": 400, + "out": { + "error": "abonnement invalide", + "ok": false + }, + "subs": "[{\"endpoint\":\"https://push.exemple.fr/a\",\"keys\":{\"auth\":\"x3\",\"p256dh\":\"BCCC\"}},{\"endpoint\":\"https://push.exemple.fr/b\",\"keys\":{\"auth\":\"x2\",\"p256dh\":\"BBBB\"}}]" + }, + { + "method": "POST", + "path": "/api/push/subscribe", + "body": "pas du json", + "status": 400, + "out": { + "error": "abonnement invalide", + "ok": false + }, + "subs": "[{\"endpoint\":\"https://push.exemple.fr/a\",\"keys\":{\"auth\":\"x3\",\"p256dh\":\"BCCC\"}},{\"endpoint\":\"https://push.exemple.fr/b\",\"keys\":{\"auth\":\"x2\",\"p256dh\":\"BBBB\"}}]" + }, + { + "method": "POST", + "path": "/api/push/unsubscribe", + "body": "{\"endpoint\":\"https://push.exemple.fr/inconnu\"}", + "status": 200, + "out": { + "ok": true + }, + "subs": "[{\"endpoint\":\"https://push.exemple.fr/a\",\"keys\":{\"auth\":\"x3\",\"p256dh\":\"BCCC\"}},{\"endpoint\":\"https://push.exemple.fr/b\",\"keys\":{\"auth\":\"x2\",\"p256dh\":\"BBBB\"}}]" + }, + { + "method": "POST", + "path": "/api/push/unsubscribe", + "body": "{\"endpoint\":\"https://push.exemple.fr/a\"}", + "status": 200, + "out": { + "ok": true + }, + "subs": "[{\"endpoint\":\"https://push.exemple.fr/b\",\"keys\":{\"auth\":\"x2\",\"p256dh\":\"BBBB\"}}]" + }, + { + "method": "POST", + "path": "/api/push/unsubscribe", + "body": "pas du json", + "status": 200, + "out": { + "ok": true + }, + "subs": "[{\"endpoint\":\"https://push.exemple.fr/b\",\"keys\":{\"auth\":\"x2\",\"p256dh\":\"BBBB\"}}]" + }, + { + "method": "POST", + "path": "/api/push/unsubscribe", + "body": "{\"endpoint\":\"https://push.exemple.fr/b\"}", + "status": 200, + "out": { + "ok": true + }, + "subs": "[]" + }, + { + "method": "GET", + "path": "/api/push/key", + "status": 200, + "out": { + "key": "{{CLE}}", + "ok": true + }, + "subs": "[]" + } + ] +} diff --git a/rust/loki-rs/tests/proxy.rs b/rust/loki-rs/tests/proxy.rs index 71d207b..347d2aa 100644 --- a/rust/loki-rs/tests/proxy.rs +++ b/rust/loki-rs/tests/proxy.rs @@ -104,6 +104,7 @@ async fn facade(backend: SocketAddr) -> (SocketAddr, oneshot::Sender<()>) { dictate: None, settings: false, files: None, + push: None, }; tokio::spawn(loki_rs::proxy::serve(l, cfg, async move { let _ = rx.await; diff --git a/rust/loki-rs/tests/push.rs b/rust/loki-rs/tests/push.rs new file mode 100644 index 0000000..98e088c --- /dev/null +++ b/rust/loki-rs/tests/push.rs @@ -0,0 +1,316 @@ +//! Lot 12 : notifications Web Push tenues par la façade (`LOKI_PUSH=rust`). +//! +//! - le journal tiré du Go (`internal/loki/parity_push_test.go` → +//! `tests/parity_push/push.json`) : durées, clé, abonnements — la liste en +//! base comparée à l'octet près ; +//! - le chiffrement contre l'exemple de la RFC 8291 (annexe A) ; +//! - un envoi de bout en bout vers un faux service de push : en-têtes, +//! signature VAPID vérifiée, corps déchiffré côté navigateur, abonnement +//! purgé sur 410. +#![cfg(feature = "tls")] + +mod common; + +use std::convert::Infallible; +use std::net::SocketAddr; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use base64::Engine; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use bytes::Bytes; +use common::fakego; +use http_body_util::{BodyExt, Full}; +use hyper::body::Incoming; +use hyper::{Request, Response, StatusCode, service::service_fn}; +use hyper_util::rt::TokioIo; +use loki_rs::push; +use ring::{aead, agreement, hkdf, signature}; +use serde_json::Value; + +fn fixture() -> Value { + serde_json::from_str(include_str!("parity_push/push.json")).expect("journal") +} + +/// `time.ParseDuration`, réduit à ce que le journal emploie. +fn go_ms(s: &str) -> i64 { + let (neg, mut s) = match s.strip_prefix('-') { + Some(r) => (true, r), + None => (false, s), + }; + let mut ms = 0f64; + while !s.is_empty() { + let n = s.find(|c: char| !c.is_ascii_digit() && c != '.').unwrap(); + let v: f64 = s[..n].parse().unwrap(); + let rest = &s[n..]; + let (mul, len) = if rest.starts_with("ms") { + (1.0, 2) + } else if rest.starts_with('h') { + (3_600_000.0, 1) + } else if rest.starts_with('m') { + (60_000.0, 1) + } else { + (1000.0, 1) + }; + ms += v * mul; + s = &rest[len..]; + } + if neg { -(ms as i64) } else { ms as i64 } +} + +#[test] +fn parite_des_durees() { + for (d, want) in fixture()["durations"].as_object().unwrap() { + assert_eq!(push::fmt_dur_fr(go_ms(d)), want.as_str().unwrap(), "{d}"); + } +} + +#[test] +fn exemple_de_la_rfc_8291() { + let b = |s: &str| URL_SAFE_NO_PAD.decode(s).unwrap(); + let body = push::encrypt_with( + b"When I grow up, I want to be a watermelon", + &b("BTBZMqHH6r4Tts7J_aSIgg"), + &b("BCVxsr7N_eNgVRqvHtD0zTZsEc6-VV-JvLexhqUzORcxaOzi6-AYWXvTBHm4bjyPjs7Vd8pZGH6SRpkNtoIAiw4"), + b("DGv6ra1nlYgDCS1FRnbzlw").as_slice().try_into().unwrap(), + &b("BP4z9KsN6nGRTbVYI_c7VJSPQTBtkgcy27mlmlMoZIIgDll6e3vCYLocInmYWAmS6TlzAC8wEqKK6PBru3jl7A8"), + &b("kyrL1jIIOHEzg3sM2ZWRHDRB62YACZhhSlknJ672kSs"), + 4096, + false, + ) + .unwrap(); + assert_eq!( + URL_SAFE_NO_PAD.encode(body), + "DGv6ra1nlYgDCS1FRnbzlwAAEABBBP4z9KsN6nGRTbVYI_c7VJSPQTBtkgcy27mlmlMoZIIgDll6e3vCYLocInmYWAmS6TlzAC8wEqKK6PBru3jl7A_yl95bQpu6cVPTpK4Mqgkf1CXztLVBSt2Ks3oZwbuwXPXLWyouBWLVWGNWQexSgSxsj_Qulcy4a-fN" + ); +} + +#[tokio::test] +async fn parite_des_routes() { + let fx = fixture(); + let tmp = tempfile::tempdir().unwrap(); + let opts = fakego::Opts { + push: Some(Arc::new(push::Push::new())), + ..Default::default() + }; + let (facade, go, _stop, _) = fakego::start_with(tmp.path().to_path_buf(), opts).await; + let key_re = regex::Regex::new(r#""key":"[A-Za-z0-9_-]{87}""#).unwrap(); + let mut first_key = String::new(); + for (i, c) in fx["calls"].as_array().unwrap().iter().enumerate() { + let (m, p) = (c["method"].as_str().unwrap(), c["path"].as_str().unwrap()); + let body = c.get("body").and_then(Value::as_str).unwrap_or(""); + let (code, b) = fakego::send(facade, m, p, &[], body).await; + let txt = String::from_utf8_lossy(&b).into_owned(); + if let Some(k) = key_re.find(&txt) { + // Clé stable d'un appel à l'autre. + if first_key.is_empty() { + first_key = k.as_str().to_string(); + } + assert_eq!(k.as_str(), first_key, "clé VAPID changée"); + } + let got: Value = + serde_json::from_str(&key_re.replace_all(&txt, r#""key":"{{CLE}}""#)).unwrap(); + let ctx = format!("appel {i} : {m} {p} {body}"); + assert_eq!(code.as_u16() as u64, c["status"].as_u64().unwrap(), "{ctx}"); + assert_eq!(got, c["out"], "{ctx}"); + assert_eq!( + go.get("state", "push_subs").unwrap_or_default(), + c["subs"].as_str().unwrap(), + "{ctx} (abonnements en base)" + ); + } + assert_eq!(go.get("state", "push_vapid_priv").unwrap().len(), 43); +} + +/// Faux service de push : répond 201 au premier endpoint, 410 au second, et +/// garde chaque requête (chemin, en-têtes, corps). +#[derive(Default)] +struct FakePush { + seen: Mutex>, +} + +async fn fake_push() -> (SocketAddr, Arc) { + let st = Arc::new(FakePush::default()); + let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = l.local_addr().unwrap(); + let s2 = st.clone(); + tokio::spawn(async move { + while let Ok((s, _)) = l.accept().await { + let st = s2.clone(); + tokio::spawn(async move { + let svc = service_fn(move |req: Request| { + let st = st.clone(); + async move { + let path = req.uri().path().to_string(); + let (parts, body) = req.into_parts(); + let b = body.collect().await.unwrap().to_bytes(); + st.seen + .lock() + .unwrap() + .push((path.clone(), parts.headers, b)); + let code = if path.ends_with("/perime") { 410 } else { 201 }; + let mut r = Response::new(Full::new(Bytes::from_static(b""))); + *r.status_mut() = StatusCode::from_u16(code).unwrap(); + Ok::<_, Infallible>(r) + } + }); + let _ = hyper::server::conn::http1::Builder::new() + .serve_connection(TokioIo::new(s), svc) + .await; + }); + } + }); + (addr, st) +} + +struct Len(usize); +impl hkdf::KeyType for Len { + fn len(&self) -> usize { + self.0 + } +} + +fn expand(salt: &[u8], ikm: &[u8], info: &[u8], n: usize) -> Vec { + let prk = hkdf::Salt::new(hkdf::HKDF_SHA256, salt).extract(ikm); + let parts = [info]; + let okm = prk.expand(&parts, Len(n)).unwrap(); + let mut out = vec![0u8; n]; + okm.fill(&mut out).unwrap(); + out +} + +#[tokio::test] +async fn envoi_de_bout_en_bout() { + let (addr, svc) = fake_push().await; + let tmp = tempfile::tempdir().unwrap(); + let p = Arc::new(push::Push::new()); + let opts = fakego::Opts { + push: Some(p.clone()), + ..Default::default() + }; + let (facade, go, _stop, _) = fakego::start_with(tmp.path().to_path_buf(), opts).await; + // Côté navigateur : une clé ECDH (à usage unique ici) et le secret d'auth. + let rng = ring::rand::SystemRandom::new(); + let ua_priv = agreement::EphemeralPrivateKey::generate(&agreement::ECDH_P256, &rng).unwrap(); + let ua_pub = ua_priv.compute_public_key().unwrap().as_ref().to_vec(); + let auth = b"seize octets !!!".to_vec(); + let subs = format!( + r#"[{{"endpoint":"http://{addr}/push/ok","keys":{{"auth":"{}","p256dh":"{}"}}}},{{"endpoint":"http://{addr}/push/perime","keys":{{"auth":"{}","p256dh":"{}"}}}}]"#, + URL_SAFE_NO_PAD.encode(&auth), + URL_SAFE_NO_PAD.encode(&ua_pub), + URL_SAFE_NO_PAD.encode(&auth), + URL_SAFE_NO_PAD.encode(&ua_pub), + ); + go.set("state", "push_subs", &subs); + let (code, _) = fakego::send( + facade, + "POST", + push::INTERNAL_SEND, + &[("x-loki-internal", fakego::JETON)], + r#"{"title":"Loki","body":"Réponse prête · 42s"}"#, + ) + .await; + assert_eq!(code, StatusCode::NO_CONTENT); + for _ in 0..200 { + if svc.seen.lock().unwrap().len() >= 2 + && go + .get("state", "push_subs") + .is_some_and(|s| !s.contains("perime")) + { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + let seen = svc.seen.lock().unwrap().clone(); + assert_eq!(seen.len(), 2); + // Abonnement périmé (410) purgé, l'autre gardé. + let left = go.get("state", "push_subs").unwrap(); + assert!( + left.contains("/push/ok") && !left.contains("perime"), + "{left}" + ); + + let (path, h, body) = seen.iter().find(|s| s.0 == "/push/ok").unwrap().clone(); + assert_eq!(path, "/push/ok"); + assert_eq!(h["content-encoding"], "aes128gcm"); + assert_eq!(h["content-type"], "application/octet-stream"); + assert_eq!(h["ttl"], "120"); + assert_eq!(h["urgency"], "high"); + // VAPID : JWT ES256 signé par la clé rangée, audience = origine du service. + let vapid_pub = go.get("state", "push_vapid_pub").unwrap(); + let auth_h = h["authorization"].to_str().unwrap().to_string(); + let rest = auth_h.strip_prefix("vapid t=").unwrap(); + let (jwt, k) = rest.split_once(", k=").unwrap(); + assert_eq!(k, vapid_pub); + let mut parts = jwt.split('.'); + let (h64, c64, s64) = ( + parts.next().unwrap(), + parts.next().unwrap(), + parts.next().unwrap(), + ); + let claims: Value = serde_json::from_slice(&URL_SAFE_NO_PAD.decode(c64).unwrap()).unwrap(); + assert_eq!(claims["aud"], format!("http://{addr}")); + assert_eq!(claims["sub"], "mailto:mail@nathaninline.com"); + signature::UnparsedPublicKey::new( + &signature::ECDSA_P256_SHA256_FIXED, + URL_SAFE_NO_PAD.decode(&vapid_pub).unwrap(), + ) + .verify( + format!("{h64}.{c64}").as_bytes(), + &URL_SAFE_NO_PAD.decode(s64).unwrap(), + ) + .expect("signature VAPID invalide"); + + // Déchiffrement côté navigateur (RFC 8291). + assert_eq!( + body.len(), + push::RECORD_SIZE, + "enregistrement complété à 4096 octets" + ); + let salt = &body[..16]; + assert_eq!(u32::from_be_bytes(body[16..20].try_into().unwrap()), 4096); + let id_len = body[20] as usize; + let as_pub = body[21..21 + id_len].to_vec(); + let cipher = body[21 + id_len..].to_vec(); + let peer = agreement::UnparsedPublicKey::new(&agreement::ECDH_P256, as_pub.clone()); + let shared = agreement::agree_ephemeral(ua_priv, &peer, |s| s.to_vec()).unwrap(); + let mut info = b"WebPush: info\0".to_vec(); + info.extend_from_slice(&ua_pub); + info.extend_from_slice(&as_pub); + let ikm = expand(&auth, &shared, &info, 32); + let cek = expand(salt, &ikm, b"Content-Encoding: aes128gcm\0", 16); + let nonce = expand(salt, &ikm, b"Content-Encoding: nonce\0", 12); + let key = aead::LessSafeKey::new(aead::UnboundKey::new(&aead::AES_128_GCM, &cek).unwrap()); + let mut buf = cipher; + let plain = key + .open_in_place( + aead::Nonce::try_assume_unique_for_key(&nonce).unwrap(), + aead::Aad::empty(), + &mut buf, + ) + .expect("déchiffrement"); + let end = plain.iter().rposition(|b| *b == 0x02).unwrap(); + assert!(plain[end + 1..].iter().all(|b| *b == 0)); + let msg: Value = serde_json::from_slice(&plain[..end]).unwrap(); + assert_eq!( + msg, + serde_json::json!({ "title": "Loki", "body": "Réponse prête · 42s", "tag": "loki-turn" }) + ); +} + +#[tokio::test] +async fn sans_la_variable_les_notifications_restent_au_go() { + let tmp = tempfile::tempdir().unwrap(); + let (facade, _go, _stop) = fakego::start(tmp.path().to_path_buf()).await; + let (_, _, b) = fakego::get(facade, "/api/push/key", &[]).await; + assert_eq!(&b[..], b"go:/api/push/key"); + let (code, _) = fakego::send( + facade, + "POST", + push::INTERNAL_SEND, + &[("x-loki-internal", fakego::JETON)], + "{}", + ) + .await; + assert_eq!(code, StatusCode::NOT_FOUND); +}