nexo_pairing/
plugin_http.rs

1//! Phase 81.33.b.real Stage 2 — daemon-side plugin HTTP proxy.
2//!
3//! Plugins declare `[plugin.http] mount_prefix = "/<prefix>"` in
4//! their `nexo-plugin.toml`. The daemon builds a
5//! [`PluginHttpRouter`] from every declaring plugin's manifest at
6//! boot, and the HTTP handler checks the router on every incoming
7//! request BEFORE the legacy hardcoded `/whatsapp/*` etc. paths.
8//!
9//! A match forwards the request to the plugin's subprocess via
10//! broker JSON-RPC (`plugin.<id>.http.request`), reads the reply,
11//! and writes it back to the TCP stream.
12//!
13//! ## Wire format
14//!
15//! See [`crate::plugin_http`] crate docs for the JSON shape.
16//! Briefly:
17//!
18//! - request payload:
19//!   `{ method, path, query, headers: [[k,v],…], body_base64 }`
20//! - reply payload:
21//!   `{ status, headers: [[k,v],…], body_base64 }`
22//!
23//! ## Limitations
24//!
25//! - No streaming responses (SSE, chunked transfer). Plugin must
26//!   buffer the full response before replying.
27//! - No WebSocket upgrades. WebSocket endpoints stay on
28//!   `[plugin.http_server]` (plugin binds its own port).
29//! - Body bytes are base64-encoded JSON; OK for HTML pages (≤100KB
30//!   typical) but wasteful for large uploads.
31
32use std::sync::Arc;
33use std::time::Duration;
34
35use base64::Engine;
36use nexo_broker::{AnyBroker, BrokerHandle, Message};
37use serde::{Deserialize, Serialize};
38use serde_json::json;
39
40const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
41
42/// Plugin-side reply parsed from the broker RPC.
43#[derive(Debug, Clone, Serialize, Deserialize)]
44pub struct PluginHttpResponse {
45    /// HTTP status code (e.g. 200, 404, 503). Plugin must supply
46    /// — daemon does NOT default.
47    pub status: u16,
48    /// Optional response headers. `Content-Type` recommended;
49    /// daemon does NOT default-fill so the plugin owns the
50    /// response shape completely.
51    #[serde(default)]
52    pub headers: Vec<(String, String)>,
53    /// Base64-encoded body bytes. Empty string = no body.
54    #[serde(default)]
55    pub body_base64: String,
56}
57
58impl PluginHttpResponse {
59    /// Decode the response body. Returns empty vec on bad base64
60    /// (logged at warn level).
61    pub fn decoded_body(&self) -> Vec<u8> {
62        if self.body_base64.is_empty() {
63            return Vec::new();
64        }
65        base64::engine::general_purpose::STANDARD
66            .decode(self.body_base64.as_bytes())
67            .unwrap_or_default()
68    }
69
70    /// Pick a header value (case-insensitive). Returns first
71    /// match.
72    pub fn header(&self, name: &str) -> Option<&str> {
73        self.headers
74            .iter()
75            .find(|(k, _)| k.eq_ignore_ascii_case(name))
76            .map(|(_, v)| v.as_str())
77    }
78}
79
80/// Daemon-side route entry: which plugin owns which prefix +
81/// per-plugin timeout.
82#[derive(Debug, Clone)]
83struct Route {
84    mount_prefix: String,
85    plugin_id: String,
86    timeout: Duration,
87}
88
89/// Daemon-reserved path prefixes. Plugin registrations whose
90/// `mount_prefix` collides with any of these are rejected so a
91/// malicious or buggy plugin cannot hijack health/metrics/admin
92/// surfaces.
93///
94/// Order: longest-first to match the router's matching order.
95/// Comparison is exact-equality OR prefix-of-plugin: a plugin
96/// asking for `/health/foo` would still shadow the daemon's
97/// `/health` endpoint, so we reject prefixes that contain
98/// a reserved path as their leading segment too.
99pub const RESERVED_PREFIXES: &[&str] = &[
100    "/health",
101    "/metrics",
102    "/pair",
103    "/admin",
104    "/.well-known",
105];
106
107/// Longest-prefix-first matcher. Insertion preserves declaration
108/// order until [`Self::sort`] is called; once sorted, prefix
109/// matching is deterministic for ambiguous prefixes (e.g.
110/// `/api/v1` wins over `/api`).
111#[derive(Debug, Clone, Default)]
112pub struct PluginHttpRouter {
113    routes: Vec<Route>,
114}
115
116impl PluginHttpRouter {
117    pub fn new() -> Self {
118        Self::default()
119    }
120
121    /// Register a plugin's mount prefix. Duplicates are accepted
122    /// (last-write-wins per `plugin_id`), so a plugin that
123    /// re-registers (hot-spawn restart) replaces its own entry.
124    ///
125    /// Rejects registrations colliding with [`RESERVED_PREFIXES`]
126    /// — daemon-internal paths (`/health`, `/metrics`, `/pair`,
127    /// `/admin`, `/.well-known`) cannot be hijacked. Caller
128    /// surfaces a warn-level log for rejected registrations so
129    /// operators see the cause; the plugin's broker handler
130    /// stays unhooked from those routes.
131    pub fn register(
132        &mut self,
133        plugin_id: &str,
134        mount_prefix: &str,
135        timeout: Option<Duration>,
136    ) -> Result<(), RouteRegistrationError> {
137        for reserved in RESERVED_PREFIXES {
138            if mount_prefix == *reserved
139                || mount_prefix.starts_with(&format!("{reserved}/"))
140            {
141                return Err(RouteRegistrationError::Reserved {
142                    requested: mount_prefix.to_string(),
143                    reserved: (*reserved).to_string(),
144                });
145            }
146        }
147        // Drop any existing entry for the same plugin_id (live
148        // restart replaces).
149        self.routes.retain(|r| r.plugin_id != plugin_id);
150        self.routes.push(Route {
151            mount_prefix: mount_prefix.to_string(),
152            plugin_id: plugin_id.to_string(),
153            timeout: timeout.unwrap_or(DEFAULT_TIMEOUT),
154        });
155        self.sort();
156        Ok(())
157    }
158
159    /// Longest-prefix-first ordering. Required for deterministic
160    /// matching when prefixes nest (`/api/v1` MUST win over
161    /// `/api` when both are registered).
162    fn sort(&mut self) {
163        self.routes.sort_by(|a, b| {
164            b.mount_prefix
165                .len()
166                .cmp(&a.mount_prefix.len())
167                .then_with(|| a.plugin_id.cmp(&b.plugin_id))
168        });
169    }
170
171    /// Match a request path. Returns `(plugin_id, timeout)` if a
172    /// route covers it; the caller forwards via
173    /// [`forward_request`].
174    pub fn match_path(&self, path: &str) -> Option<(&str, Duration)> {
175        self.routes
176            .iter()
177            .find(|r| path.starts_with(&r.mount_prefix))
178            .map(|r| (r.plugin_id.as_str(), r.timeout))
179    }
180
181    pub fn is_empty(&self) -> bool {
182        self.routes.is_empty()
183    }
184}
185
186/// Forward a daemon-received HTTP request to the plugin via
187/// broker JSON-RPC.
188///
189/// The caller has already parsed `method + full_path` from the
190/// stream and snapshot any required headers / body bytes. This
191/// helper does NOT touch the TCP stream — the caller writes the
192/// response back.
193///
194/// On broker failure (timeout, transport error, malformed reply),
195/// returns a typed error so the caller can render a 502/504.
196pub async fn forward_request(
197    broker: &AnyBroker,
198    plugin_id: &str,
199    method: &str,
200    path: &str,
201    query: &str,
202    headers: &[(String, String)],
203    body: &[u8],
204    timeout: Duration,
205) -> Result<PluginHttpResponse, PluginHttpForwardError> {
206    let topic = format!("plugin.{plugin_id}.http.request");
207    let body_b64 = base64::engine::general_purpose::STANDARD.encode(body);
208    let payload = json!({
209        "method": method,
210        "path": path,
211        "query": query,
212        "headers": headers,
213        "body_base64": body_b64,
214    });
215    let msg = Message::new(topic.clone(), payload);
216    let reply = broker
217        .request(&topic, msg, timeout)
218        .await
219        .map_err(|e| PluginHttpForwardError::Broker(e.to_string()))?;
220    serde_json::from_value::<PluginHttpResponse>(reply.payload).map_err(|e| {
221        PluginHttpForwardError::ParseReply(format!(
222            "plugin {plugin_id} returned malformed http reply: {e}"
223        ))
224    })
225}
226
227/// Route registration error. Returned by
228/// [`PluginHttpRouter::register`] when the requested
229/// `mount_prefix` violates a reserved-prefix or validation rule.
230#[derive(Debug, thiserror::Error)]
231pub enum RouteRegistrationError {
232    #[error("mount_prefix `{requested}` collides with daemon-reserved prefix `{reserved}`")]
233    Reserved {
234        requested: String,
235        reserved: String,
236    },
237}
238
239/// Typed forwarder errors. Caller renders the right status
240/// code:
241/// - `Broker` → 504 Gateway Timeout (or 502 for non-timeout).
242/// - `ParseReply` → 502 Bad Gateway (plugin contract violation).
243#[derive(Debug, thiserror::Error)]
244pub enum PluginHttpForwardError {
245    #[error("broker error: {0}")]
246    Broker(String),
247    #[error("plugin reply parse error: {0}")]
248    ParseReply(String),
249}
250
251/// Build a router from a slice of `(plugin_id, manifest)`. Skips
252/// plugins whose manifest does not declare `[plugin.http]`.
253pub fn build_router_from_handles(
254    handles: &std::collections::BTreeMap<String, Arc<dyn DynPluginManifest>>,
255) -> PluginHttpRouter {
256    let mut router = PluginHttpRouter::new();
257    for (plugin_id, handle) in handles.iter() {
258        if let Some(section) = handle.http_section() {
259            let _ = router.register(plugin_id, &section.mount_prefix, section.timeout);
260        }
261    }
262    router
263}
264
265/// Type-erased view of a plugin's manifest sufficient for router
266/// construction. The real `NexoPlugin` trait lives in nexo-core;
267/// pulling it into this crate would create a circular dep. The
268/// caller (daemon) walks `wire.plugin_handles` and wraps each
269/// handle in a small impl of this trait OR — simpler — passes a
270/// `Vec<(plugin_id, mount_prefix, timeout)>` directly to
271/// [`PluginHttpRouter::register`] in a loop.
272///
273/// We provide the trait as documentation for the contract; the
274/// daemon uses the direct `register()` loop in practice.
275pub trait DynPluginManifest: Send + Sync {
276    fn http_section(&self) -> Option<HttpSectionView>;
277}
278
279/// View struct mirroring
280/// [`nexo_plugin_manifest::http::PluginHttpSection`] but without
281/// the cross-crate dep (router crate already depends on
282/// nexo-plugin-manifest so this is mostly cosmetic — kept here so
283/// downstream alternative impls can avoid the dep if needed).
284#[derive(Debug, Clone)]
285pub struct HttpSectionView {
286    pub mount_prefix: String,
287    pub timeout: Option<Duration>,
288}
289
290#[cfg(test)]
291mod tests {
292    use super::*;
293
294    #[test]
295    fn match_path_uses_longest_prefix_first() {
296        let mut r = PluginHttpRouter::new();
297        r.register("plugin_short", "/api", None).unwrap();
298        r.register("plugin_long", "/api/v1", None).unwrap();
299        let (id, _) = r.match_path("/api/v1/users").expect("matches");
300        assert_eq!(id, "plugin_long");
301    }
302
303    #[test]
304    fn match_path_falls_back_to_shorter_prefix() {
305        let mut r = PluginHttpRouter::new();
306        r.register("plugin_short", "/api", None).unwrap();
307        r.register("plugin_long", "/api/v1", None).unwrap();
308        let (id, _) = r.match_path("/api/v2/users").expect("matches");
309        assert_eq!(id, "plugin_short");
310    }
311
312    #[test]
313    fn match_path_returns_none_for_no_match() {
314        let mut r = PluginHttpRouter::new();
315        r.register("plugin", "/whatsapp", None).unwrap();
316        assert!(r.match_path("/foo").is_none());
317        assert!(r.match_path("/").is_none());
318    }
319
320    #[test]
321    fn register_replaces_existing_entry_for_same_plugin() {
322        let mut r = PluginHttpRouter::new();
323        r.register("plugin", "/old", None).unwrap();
324        r.register("plugin", "/new", None).unwrap();
325        assert!(r.match_path("/old").is_none());
326        assert!(r.match_path("/new").is_some());
327    }
328
329    #[test]
330    fn register_applies_custom_timeout() {
331        let mut r = PluginHttpRouter::new();
332        r.register("plugin", "/slow", Some(Duration::from_secs(120))).unwrap();
333        let (_, t) = r.match_path("/slow/foo").unwrap();
334        assert_eq!(t, Duration::from_secs(120));
335    }
336
337    #[test]
338    fn register_default_timeout_when_none() {
339        let mut r = PluginHttpRouter::new();
340        r.register("plugin", "/fast", None).unwrap();
341        let (_, t) = r.match_path("/fast/foo").unwrap();
342        assert_eq!(t, DEFAULT_TIMEOUT);
343    }
344
345    #[test]
346    fn register_rejects_reserved_prefixes() {
347        let mut r = PluginHttpRouter::new();
348        for reserved in RESERVED_PREFIXES {
349            let result = r.register("evil_plugin", reserved, None);
350            assert!(
351                matches!(result, Err(RouteRegistrationError::Reserved { .. })),
352                "expected reservation rejection for `{reserved}`, got {result:?}",
353            );
354        }
355    }
356
357    #[test]
358    fn register_rejects_subpath_of_reserved() {
359        let mut r = PluginHttpRouter::new();
360        // `/health/foo` shadows the daemon's `/health` endpoint
361        // if matched first — must be rejected.
362        let result = r.register("evil_plugin", "/health/foo", None);
363        assert!(matches!(result, Err(RouteRegistrationError::Reserved { .. })));
364    }
365
366    #[test]
367    fn register_accepts_prefixes_that_only_share_substring_with_reserved() {
368        let mut r = PluginHttpRouter::new();
369        // `/healthy` is NOT a subpath of `/health` — `/health/` is
370        // the prefix that would shadow; `/healthy` is a sibling.
371        assert!(r.register("plugin", "/healthy", None).is_ok());
372        assert!(r.register("plugin2", "/metrics-aggregator", None).is_ok());
373    }
374
375    #[test]
376    fn plugin_http_response_decodes_body() {
377        let response = PluginHttpResponse {
378            status: 200,
379            headers: vec![("Content-Type".into(), "text/html".into())],
380            body_base64: base64::engine::general_purpose::STANDARD.encode("<html/>"),
381        };
382        assert_eq!(response.decoded_body(), b"<html/>");
383        assert_eq!(response.header("content-type"), Some("text/html"));
384        assert_eq!(response.header("CONTENT-TYPE"), Some("text/html"));
385    }
386
387    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
388    async fn forward_request_returns_broker_error_when_no_subscriber() {
389        let broker = AnyBroker::Local(nexo_broker::LocalBroker::new());
390        let result = forward_request(
391            &broker,
392            "plugin",
393            "GET",
394            "/whatsapp/pair",
395            "",
396            &[],
397            &[],
398            Duration::from_millis(100),
399        )
400        .await;
401        assert!(matches!(result, Err(PluginHttpForwardError::Broker(_))));
402    }
403}
nexo_pairing/plugin_http.rs

nexo_pairing/
plugin_http.rs
