feat(server): /v1/fetch endpoint backed by handler-owned browsers

px-camoufox/Cargo.toml

@@ -18,10 +18,12 @@ futures       = { workspace = true }
 px-core       = { workspace = true }
 px-errors     = { workspace = true }
 px-harvester  = { workspace = true }
+px-pipeline   = { workspace = true }
 serde         = { workspace = true }
 serde_json    = { workspace = true }
 tokio         = { workspace = true, features = ["sync", "time", "macros", "rt", "rt-multi-thread", "process", "net", "io-util"] }
 tracing       = { workspace = true }
+url           = { workspace = true }
 
 [lib]
 name = "px_camoufox"

px-camoufox/src/infrastructure/camoufox_fetcher.rs

@@ -0,0 +1,131 @@
+//! `Fetcher` impl on `CamoufoxPool`. Runs a single HTTP request from
+//! inside a fresh Camoufox session by navigating to the target URL's
+//! origin (so cookies/JS context are issued), then executing a
+//! `fetch()` from the page context. Captures status, headers, body.
+
+use crate::infrastructure::camoufox_pool::CamoufoxPool;
+use async_trait::async_trait;
+use fantoccini::ClientBuilder;
+use px_errors::AppError;
+use px_pipeline::{FetchRequest, FetchResponse, Fetcher};
+use serde_json::{Map, Value};
+use std::collections::HashMap;
+use std::time::{Duration, Instant};
+use tokio::time::sleep;
+
+#[async_trait]
+impl Fetcher for CamoufoxPool {
+    async fn fetch(&self, req: FetchRequest) -> Result<FetchResponse, AppError> {
+        let navigate_timeout = self.config.navigate_timeout;
+        let request_timeout = Duration::from_millis(req.timeout_ms);
+        self.with_session(None, async move |endpoint, caps| {
+            run_fetch(&endpoint, caps, &req, navigate_timeout, request_timeout).await
+        })
+        .await
+    }
+}
+
+async fn run_fetch(
+    endpoint: &str,
+    caps: Map<String, Value>,
+    req: &FetchRequest,
+    navigate_timeout: Duration,
+    request_timeout: Duration,
+) -> Result<FetchResponse, AppError> {
+    let started = Instant::now();
+    let client = ClientBuilder::native()
+        .capabilities(caps)
+        .connect(endpoint)
+        .await
+        .map_err(|e| AppError::InternalError(format!("webdriver connect: {e}")))?;
+
+    let origin = origin_of(&req.url)?;
+    let nav = client.goto(&origin);
+    if tokio::time::timeout(navigate_timeout, nav).await.is_err() {
+        let _ = client.close().await;
+        return Err(AppError::InternalError("navigate timeout".into()));
+    }
+    // Give Cloudflare / PerimeterX a beat to set their cookies.
+    sleep(Duration::from_millis(1_500)).await;
+
+    let script = build_fetch_script(req)?;
+    let exec = client.execute_async(&script, vec![]);
+    let raw = match tokio::time::timeout(request_timeout, exec).await {
+        Ok(Ok(v)) => v,
+        Ok(Err(e)) => {
+            let _ = client.close().await;
+            return Err(AppError::InternalError(format!("fetch eval: {e}")));
+        }
+        Err(_) => {
+            let _ = client.close().await;
+            return Err(AppError::InternalError("fetch timeout".into()));
+        }
+    };
+    let _ = client.close().await;
+
+    let status = raw
+        .get("status")
+        .and_then(Value::as_u64)
+        .ok_or_else(|| AppError::InternalError("fetch result missing status".into()))?
+        as u16;
+    let body = raw
+        .get("body")
+        .and_then(Value::as_str)
+        .unwrap_or("")
+        .to_string();
+    let headers: HashMap<String, String> = raw
+        .get("headers")
+        .and_then(Value::as_object)
+        .map(|m| {
+            m.iter()
+                .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
+                .collect()
+        })
+        .unwrap_or_default();
+
+    Ok(FetchResponse {
+        status,
+        headers,
+        body,
+        duration_ms: started.elapsed().as_millis() as u64,
+    })
+}
+
+fn origin_of(url: &str) -> Result<String, AppError> {
+    let parsed =
+        url::Url::parse(url).map_err(|e| AppError::BadRequest(format!("invalid url: {e}")))?;
+    Ok(format!(
+        "{}://{}{}",
+        parsed.scheme(),
+        parsed.host_str().unwrap_or(""),
+        parsed.port().map(|p| format!(":{p}")).unwrap_or_default()
+    ))
+}
+
+fn build_fetch_script(req: &FetchRequest) -> Result<String, AppError> {
+    let method = req.method().to_string();
+    let headers_json = serde_json::to_string(&req.headers)
+        .map_err(|e| AppError::InternalError(format!("encode headers: {e}")))?;
+    let body_json = serde_json::to_string(req.body.as_deref().unwrap_or(""))
+        .map_err(|e| AppError::InternalError(format!("encode body: {e}")))?;
+    let url_json = serde_json::to_string(&req.url)
+        .map_err(|e| AppError::InternalError(format!("encode url: {e}")))?;
+    let method_json = serde_json::to_string(&method)
+        .map_err(|e| AppError::InternalError(format!("encode method: {e}")))?;
+    Ok(format!(
+        r#"
+        const cb = arguments[arguments.length - 1];
+        const opts = {{ method: {method_json}, headers: {headers_json}, credentials: 'include' }};
+        const body = {body_json};
+        if (body !== '' && {method_json} !== 'GET') opts.body = body;
+        fetch({url_json}, opts)
+          .then(async (r) => {{
+            const text = await r.text();
+            const hdrs = {{}};
+            r.headers.forEach((v, k) => {{ hdrs[k] = v; }});
+            cb({{ status: r.status, headers: hdrs, body: text }});
+          }})
+          .catch((e) => cb({{ status: 0, headers: {{}}, body: String(e) }}));
+        "#
+    ))
+}

px-camoufox/src/infrastructure/camoufox_pool.rs

@@ -1,19 +1,19 @@
 use crate::domain::config::CamoufoxConfig;
+use crate::infrastructure::caps::{build_capabilities, pick_free_port, wait_for_geckodriver};
 use async_trait::async_trait;
 use fantoccini::ClientBuilder;
 use px_errors::AppError;
 use px_harvester::{HarvestRequest, HarvestResult, HarvestedCookie, Harvester};
-use serde_json::{Map, Value, json};
+use serde_json::{Map, Value};
 use std::sync::Arc;
 use std::time::Duration;
-use tokio::net::TcpListener;
 use tokio::process::Command;
 use tokio::sync::Semaphore;
 use tokio::time::sleep;
 
 pub struct CamoufoxPool {
-    config: CamoufoxConfig,
-    permits: Arc<Semaphore>,
+    pub(crate) config: CamoufoxConfig,
+    pub(crate) permits: Arc<Semaphore>,
 }
 
 impl CamoufoxPool {
@@ -25,97 +25,23 @@ impl CamoufoxPool {
         Ok(Self { config, permits })
     }
 
-    async fn pick_free_port() -> Result<u16, AppError> {
-        let listener = TcpListener::bind("127.0.0.1:0")
-            .await
-            .map_err(|e| AppError::InternalError(format!("bind ephemeral: {e}")))?;
-        let port = listener
-            .local_addr()
-            .map_err(|e| AppError::InternalError(format!("local_addr: {e}")))?
-            .port();
-        drop(listener);
-        Ok(port)
-    }
-
-    async fn wait_for_geckodriver(port: u16, timeout: Duration) -> Result<(), AppError> {
-        let deadline = tokio::time::Instant::now() + timeout;
-        loop {
-            if let Ok(resp) = get_status(port).await
-                && resp.contains("\"ready\":true")
-            {
-                return Ok(());
-            }
-            if tokio::time::Instant::now() >= deadline {
-                return Err(AppError::InternalError(format!(
-                    "geckodriver did not become ready on port {port} within {timeout:?}"
-                )));
-            }
-            sleep(Duration::from_millis(150)).await;
-        }
-    }
-
-    fn build_capabilities(&self, req: &HarvestRequest) -> Map<String, Value> {
-        let mut prefs = Map::new();
-        prefs.insert(
-            "intl.accept_languages".into(),
-            json!(self.config.locale.clone()),
-        );
-        prefs.insert("dom.webnotifications.enabled".into(), json!(false));
-        prefs.insert("media.peerconnection.enabled".into(), json!(false));
-
-        let mut firefox_options = Map::new();
-        let mut args: Vec<String> = Vec::new();
-        if self.config.headless {
-            args.push("-headless".into());
-        }
-        firefox_options.insert("args".into(), json!(args));
-        firefox_options.insert("prefs".into(), Value::Object(prefs));
-        firefox_options.insert(
-            "binary".into(),
-            json!(self.config.camoufox_bin.to_string_lossy()),
-        );
-
-        let mut caps = Map::new();
-        caps.insert("browserName".into(), json!("firefox"));
-        caps.insert("moz:firefoxOptions".into(), Value::Object(firefox_options));
-        if let Some(proxy_url) = &req.proxy {
-            caps.insert(
-                "proxy".into(),
-                json!({ "proxyType": "manual", "httpProxy": proxy_url, "sslProxy": proxy_url }),
-            );
-        }
-        caps
-    }
-}
-
-async fn get_status(port: u16) -> Result<String, AppError> {
-    use tokio::io::{AsyncReadExt, AsyncWriteExt};
-    let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", port))
-        .await
-        .map_err(|e| AppError::InternalError(format!("connect: {e}")))?;
-    let req =
-        format!("GET /status HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\nConnection: close\r\n\r\n");
-    stream
-        .write_all(req.as_bytes())
-        .await
-        .map_err(|e| AppError::InternalError(format!("write: {e}")))?;
-    let mut buf = Vec::with_capacity(2048);
-    stream
-        .read_to_end(&mut buf)
-        .await
-        .map_err(|e| AppError::InternalError(format!("read: {e}")))?;
-    Ok(String::from_utf8_lossy(&buf).to_string())
-}
-
-#[async_trait]
-impl Harvester for CamoufoxPool {
-    async fn harvest(&self, req: HarvestRequest) -> Result<HarvestResult, AppError> {
+    /// Spawn geckodriver + Camoufox, hand the resulting webdriver
+    /// endpoint to `body`, kill the child no matter what. Both the
+    /// `Harvester` and `Fetcher` impls share this lifecycle.
+    pub(crate) async fn with_session<F, R>(
+        &self,
+        proxy: Option<&str>,
+        body: F,
+    ) -> Result<R, AppError>
+    where
+        F: AsyncFnOnce(String, Map<String, Value>) -> Result<R, AppError>,
+    {
         let _permit = self
             .permits
             .acquire()
             .await
             .map_err(|e| AppError::InternalError(format!("semaphore: {e}")))?;
-        let port = Self::pick_free_port().await?;
+        let port = pick_free_port().await?;
         let mut child = Command::new(&self.config.geckodriver_bin)
             .arg("--port")
             .arg(port.to_string())
@@ -126,18 +52,28 @@ impl Harvester for CamoufoxPool {
             .kill_on_drop(true)
             .spawn()
             .map_err(|e| AppError::InternalError(format!("spawn geckodriver: {e}")))?;
-        Self::wait_for_geckodriver(port, Duration::from_secs(15)).await?;
-
-        let caps = self.build_capabilities(&req);
+        wait_for_geckodriver(port, Duration::from_secs(15)).await?;
+        let caps = build_capabilities(&self.config, proxy);
         let endpoint = format!("http://127.0.0.1:{port}");
-        let outcome = run_session(&endpoint, caps, &req, self.config.navigate_timeout).await;
-
+        let outcome = body(endpoint, caps).await;
         let _ = child.kill().await;
         outcome
     }
 }
 
-async fn run_session(
+#[async_trait]
+impl Harvester for CamoufoxPool {
+    async fn harvest(&self, req: HarvestRequest) -> Result<HarvestResult, AppError> {
+        let navigate_timeout = self.config.navigate_timeout;
+        let proxy = req.proxy.clone();
+        self.with_session(proxy.as_deref(), async move |endpoint, caps| {
+            harvest_session(&endpoint, caps, &req, navigate_timeout).await
+        })
+        .await
+    }
+}
+
+async fn harvest_session(
     endpoint: &str,
     caps: Map<String, Value>,
     req: &HarvestRequest,