scope/http/

ghola.rs

//! Ghola sidecar HTTP client.
//!
//! Forwards all HTTP requests to a locally running
//! [Ghola](https://github.com/robot-accomplice/ghola) sidecar
//! (`127.0.0.1:18789`). When stealth mode is enabled, the sidecar
//! applies temporal drift and ghost signing to every outgoing request.
//!
//! The sidecar is an external Go binary. When nothing answers on that
//! address, `http::shared::build` starts it with [`start_sidecar`]
//! (`<binary> --serve`, the binary found by `ghola_install::find`). The
//! sidecar it started is kept here until [`stop_spawned_sidecar`] stops it;
//! the caller decides when (the CLI asks at exit).

use super::{HttpClient, Request, Response};
use crate::error::ScopeError;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::net::TcpStream;
use std::path::Path;
use std::process::{Child, Command, Stdio};
use std::sync::Mutex;
use std::time::{Duration, Instant};

pub(crate) const SIDECAR_ADDR: &str = "127.0.0.1:18789";
const SIDECAR_URL: &str = "http://127.0.0.1:18789";
/// How long a started sidecar has to answer.
pub(crate) const SPAWN_TIMEOUT: Duration = Duration::from_secs(5);
const POLL_INTERVAL: Duration = Duration::from_millis(100);

/// The sidecar this process started and has not stopped.
static SPAWNED: Mutex<Option<Child>> = Mutex::new(None);

#[derive(Serialize)]
struct BridgeRequest {
    url: String,
    method: String,
    headers: HashMap<String, String>,
    body: String,
    drift: bool,
    ghost: bool,
    retries: i32,
    buffer_size: u32,
}

#[derive(Deserialize)]
struct BridgeResponse {
    status_code: u16,
    headers: HashMap<String, String>,
    body: String,
    #[serde(default)]
    error: String,
}

/// HTTP client that forwards requests to the Ghola sidecar bridge
/// running on `127.0.0.1:18789`. When `stealth` is `true`, the bridge
/// applies temporal drift and ghost signing to every request.
pub struct GholaHttpClient {
    client: reqwest::Client,
    stealth: bool,
    buffer_size: u32,
    base_url: String,
}

impl GholaHttpClient {
    /// Creates a new client that talks to an already-running sidecar.
    pub fn new(stealth: bool, buffer_size: u32) -> Result<Self, ScopeError> {
        let client = reqwest::Client::builder()
            .timeout(Duration::from_secs(30))
            .build()
            .map_err(|e| {
                ScopeError::Network(format!("failed to build ghola bridge client: {e}"))
            })?;
        Ok(Self {
            client,
            stealth,
            buffer_size,
            base_url: SIDECAR_URL.to_string(),
        })
    }

    /// Creates a client pointing at a custom URL (for testing).
    #[cfg(test)]
    pub fn with_base_url(
        stealth: bool,
        buffer_size: u32,
        base_url: &str,
    ) -> Result<Self, ScopeError> {
        let client = reqwest::Client::builder()
            .timeout(Duration::from_secs(5))
            .build()
            .map_err(|e| {
                ScopeError::Network(format!("failed to build ghola bridge client: {e}"))
            })?;
        Ok(Self {
            client,
            stealth,
            buffer_size,
            base_url: base_url.to_string(),
        })
    }
}

#[async_trait]
impl HttpClient for GholaHttpClient {
    async fn send(&self, request: Request) -> Result<Response, ScopeError> {
        let bridge_req = BridgeRequest {
            url: request.url,
            method: request.method,
            headers: request.headers,
            body: request.body.unwrap_or_default(),
            drift: self.stealth,
            ghost: self.stealth,
            retries: 0,
            buffer_size: self.buffer_size,
        };

        let resp = self
            .client
            .post(&self.base_url)
            .json(&bridge_req)
            .send()
            .await
            .map_err(|e| ScopeError::Network(format!("failed to reach ghola sidecar: {e}")))?;

        let bridge_resp: BridgeResponse = resp
            .json()
            .await
            .map_err(|e| ScopeError::Network(format!("invalid sidecar response: {e}")))?;

        if !bridge_resp.error.is_empty() {
            return Err(ScopeError::Network(format!(
                "sidecar error: {}",
                bridge_resp.error
            )));
        }

        Ok(Response {
            status_code: bridge_resp.status_code,
            headers: bridge_resp.headers,
            body: bridge_resp.body,
        })
    }
}

pub(crate) fn is_bridge_running() -> bool {
    SIDECAR_ADDR
        .parse()
        .ok()
        .and_then(|addr| TcpStream::connect_timeout(&addr, Duration::from_millis(200)).ok())
        .is_some()
}

fn spawned() -> std::sync::MutexGuard<'static, Option<Child>> {
    SPAWNED.lock().unwrap_or_else(|e| e.into_inner())
}

/// Starts `binary --serve` and waits up to `timeout` for the sidecar to
/// answer on the sidecar's address (`127.0.0.1:18789`). Returns its PID. The sidecar is kept until
/// [`stop_spawned_sidecar`]; it is not stopped when this process exits.
///
/// It blocks the calling thread while it waits.
///
/// # Errors
///
/// Returns `IoError` when the binary cannot run; `Network` when it exits
/// or does not answer in time (it is stopped then).
pub fn start_sidecar(binary: &Path, timeout: Duration) -> Result<u32, ScopeError> {
    let mut child = Command::new(binary)
        .arg("--serve")
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()
        .map_err(|e| {
            ScopeError::io_context(e, format!("failed to start {} --serve", binary.display()))
        })?;
    let start = Instant::now();
    loop {
        if is_bridge_running() {
            let pid = child.id();
            *spawned() = Some(child);
            return Ok(pid);
        }
        if let Ok(Some(status)) = child.try_wait() {
            return Err(ScopeError::Network(format!(
                "{} --serve exited ({status}) before it answered on {SIDECAR_ADDR}",
                binary.display()
            )));
        }
        if start.elapsed() >= timeout {
            let _ = child.kill();
            let _ = child.wait();
            return Err(ScopeError::Network(format!(
                "the ghola sidecar did not answer on {SIDECAR_ADDR} within {timeout:?}"
            )));
        }
        std::thread::sleep(POLL_INTERVAL);
    }
}

/// The PID of the sidecar this process started, while it runs.
pub fn spawned_sidecar() -> Option<u32> {
    let mut guard = spawned();
    match guard.as_mut().map(Child::try_wait) {
        Some(Ok(None)) => guard.as_ref().map(Child::id),
        _ => {
            *guard = None;
            None
        }
    }
}

/// Stops the sidecar this process started. Returns its PID, or `None`
/// when this process started none.
pub fn stop_spawned_sidecar() -> Option<u32> {
    let mut child = spawned().take()?;
    let _ = child.kill();
    let _ = child.wait();
    Some(child.id())
}

/// Keeps `child` as the sidecar this process started, so tests elsewhere
/// (the CLI's exit question) can stop it without a sidecar port.
#[cfg(any(test, feature = "test-support"))]
pub fn adopt_sidecar(child: Child) {
    *spawned() = Some(child);
}

// ============================================================================
// Unit Tests
// ============================================================================

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_bridge_request_serialization() {
        let req = BridgeRequest {
            url: "https://example.com".to_string(),
            method: "GET".to_string(),
            headers: HashMap::new(),
            body: String::new(),
            drift: true,
            ghost: false,
            retries: 3,
            buffer_size: 4096,
        };
        let json = serde_json::to_string(&req).unwrap();
        assert!(json.contains("\"drift\":true"));
        assert!(json.contains("\"ghost\":false"));
        assert!(json.contains("\"retries\":3"));
        assert!(json.contains("\"buffer_size\":4096"));
    }

    #[test]
    fn test_bridge_response_deserialization() {
        let json = r#"{"status_code":200,"headers":{},"body":"ok","error":""}"#;
        let resp: BridgeResponse = serde_json::from_str(json).unwrap();
        assert_eq!(resp.status_code, 200);
        assert_eq!(resp.body, "ok");
        assert!(resp.error.is_empty());
    }

    #[test]
    fn test_bridge_response_with_error() {
        let json = r#"{"status_code":0,"headers":{},"body":"","error":"connection refused"}"#;
        let resp: BridgeResponse = serde_json::from_str(json).unwrap();
        assert_eq!(resp.error, "connection refused");
    }

    #[test]
    fn test_bridge_response_missing_error_field() {
        let json = r#"{"status_code":200,"headers":{},"body":"data"}"#;
        let resp: BridgeResponse = serde_json::from_str(json).unwrap();
        assert!(resp.error.is_empty());
    }

    #[test]
    fn test_ghola_client_creation() {
        let client = GholaHttpClient::new(true, 4096);
        assert!(client.is_ok());
    }

    #[test]
    fn test_ghola_client_creation_stealth_off() {
        let client = GholaHttpClient::new(false, 4096);
        assert!(client.is_ok());
    }

    /// Serializes the tests that use the spawned sidecar or its port.
    static SIDECAR: Mutex<()> = Mutex::new(());

    fn sidecar_lock() -> std::sync::MutexGuard<'static, ()> {
        SIDECAR.lock().unwrap_or_else(|e| e.into_inner())
    }

    #[test]
    fn test_no_sidecar_is_spawned_by_default() {
        let _lock = sidecar_lock();
        assert_eq!(spawned_sidecar(), None);
        assert_eq!(stop_spawned_sidecar(), None);
    }

    #[test]
    fn test_start_sidecar_reports_a_missing_binary() {
        let _lock = sidecar_lock();
        let dir = tempfile::tempdir().unwrap();
        let err =
            start_sidecar(&dir.path().join("no-ghola"), Duration::from_millis(50)).unwrap_err();
        assert!(err.to_string().contains("--serve"), "{err}");
        assert_eq!(spawned_sidecar(), None);
    }

    /// A process that runs until it is stopped.
    fn long_running() -> Child {
        let mut command = if cfg!(windows) {
            let mut c = Command::new("ping");
            c.args(["-n", "60", "127.0.0.1"]);
            c
        } else {
            let mut c = Command::new("sleep");
            c.arg("60");
            c
        };
        command
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .unwrap()
    }

    #[test]
    fn test_an_adopted_sidecar_is_kept_until_stopped() {
        let _lock = sidecar_lock();
        let child = long_running();
        let pid = child.id();
        adopt_sidecar(child);
        assert_eq!(spawned_sidecar(), Some(pid));
        assert_eq!(stop_spawned_sidecar(), Some(pid));
        assert_eq!(spawned_sidecar(), None);
        assert_eq!(stop_spawned_sidecar(), None);
    }

    #[test]
    fn test_a_sidecar_that_exited_is_forgotten() {
        let _lock = sidecar_lock();
        let mut child = long_running();
        child.kill().unwrap();
        child.wait().unwrap();
        adopt_sidecar(child);
        assert_eq!(spawned_sidecar(), None);
        assert_eq!(stop_spawned_sidecar(), None);
    }

    /// [`start_sidecar`], retried while Linux says the fake script is
    /// busy: another test thread that forks while the script is still open
    /// for writing holds it (ETXTBSY) for a moment.
    #[cfg(unix)]
    fn start_retrying(ghola: &Path, timeout: Duration) -> Result<u32, ScopeError> {
        for _ in 0..50 {
            match start_sidecar(ghola, timeout) {
                Err(e) if e.to_string().contains("Text file busy") => {
                    std::thread::sleep(Duration::from_millis(20));
                }
                other => return other,
            }
        }
        start_sidecar(ghola, timeout)
    }

    /// An executable `sh` script that stands in for `ghola --serve`.
    #[cfg(unix)]
    fn fake_ghola(dir: &Path, body: &str) -> std::path::PathBuf {
        use std::os::unix::fs::PermissionsExt;
        let path = dir.join("ghola");
        std::fs::write(&path, format!("#!/bin/sh\n{body}\n")).unwrap();
        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap();
        path
    }

    #[cfg(unix)]
    #[test]
    fn test_start_sidecar_reports_a_sidecar_that_exits() {
        let _lock = sidecar_lock();
        if is_bridge_running() {
            eprintln!("skipped: something answers on {SIDECAR_ADDR}");
            return;
        }
        let dir = tempfile::tempdir().unwrap();
        let ghola = fake_ghola(dir.path(), "[ \"$1\" = --serve ] && exit 3");
        let err = start_retrying(&ghola, Duration::from_secs(10)).unwrap_err();
        assert!(err.to_string().contains("exited"), "{err}");
        assert_eq!(spawned_sidecar(), None);
    }

    #[cfg(unix)]
    #[test]
    fn test_start_sidecar_stops_a_sidecar_that_never_answers() {
        let _lock = sidecar_lock();
        if is_bridge_running() {
            eprintln!("skipped: something answers on {SIDECAR_ADDR}");
            return;
        }
        let dir = tempfile::tempdir().unwrap();
        let ghola = fake_ghola(dir.path(), "exec sleep 60");
        let started = Instant::now();
        let err = start_retrying(&ghola, Duration::from_millis(300)).unwrap_err();
        assert!(err.to_string().contains("did not answer"), "{err}");
        assert!(started.elapsed() < Duration::from_secs(5));
        assert_eq!(spawned_sidecar(), None);
    }

    #[cfg(unix)]
    #[test]
    fn test_start_sidecar_keeps_a_sidecar_that_answers() {
        let _lock = sidecar_lock();
        // The sidecar's port, held here: the fake answers through it.
        let Ok(_port) = std::net::TcpListener::bind(SIDECAR_ADDR) else {
            eprintln!("skipped: {SIDECAR_ADDR} is in use");
            return;
        };
        let dir = tempfile::tempdir().unwrap();
        let ghola = fake_ghola(dir.path(), "exec sleep 60");
        let pid = start_retrying(&ghola, Duration::from_secs(5)).unwrap();
        assert_eq!(spawned_sidecar(), Some(pid));
        assert_eq!(stop_spawned_sidecar(), Some(pid));
        assert_eq!(spawned_sidecar(), None);
    }

    #[test]
    fn test_is_bridge_running_returns_bool() {
        // Smoke test: the result depends on the host; the probe must not panic.
        let _ = is_bridge_running();
    }

    #[test]
    fn test_bridge_request_full_serialization() {
        let mut headers = HashMap::new();
        headers.insert("Authorization".to_string(), "Bearer tk".to_string());
        let req = BridgeRequest {
            url: "https://api.test.com/v1".to_string(),
            method: "POST".to_string(),
            headers,
            body: r#"{"data":1}"#.to_string(),
            drift: false,
            ghost: true,
            retries: 0,
            buffer_size: 8192,
        };
        let json = serde_json::to_string(&req).unwrap();
        assert!(json.contains("\"method\":\"POST\""));
        assert!(json.contains("\"ghost\":true"));
        assert!(json.contains("\"drift\":false"));
        assert!(json.contains("\"retries\":0"));
        assert!(json.contains("\"buffer_size\":8192"));
        assert!(json.contains("Authorization"));
    }

    #[test]
    fn test_bridge_response_roundtrip() {
        let mut headers = HashMap::new();
        headers.insert("content-type".to_string(), "application/json".to_string());
        let json = serde_json::json!({
            "status_code": 201,
            "headers": headers,
            "body": r#"{"id":42}"#,
            "error": ""
        });
        let resp: BridgeResponse = serde_json::from_value(json).unwrap();
        assert_eq!(resp.status_code, 201);
        assert_eq!(resp.body, r#"{"id":42}"#);
        assert!(resp.error.is_empty());
        assert_eq!(
            resp.headers.get("content-type").map(String::as_str),
            Some("application/json")
        );
    }

    #[tokio::test]
    async fn test_send_to_mock_sidecar() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/")
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(
                r#"{"status_code":200,"headers":{},"body":"{\"result\":\"ok\"}","error":""}"#,
            )
            .create_async()
            .await;

        let ghola = GholaHttpClient::with_base_url(true, 4096, &server.url()).unwrap();

        let req = Request::get("https://api.example.com/data");
        let resp = ghola.send(req).await.unwrap();
        assert_eq!(resp.status_code, 200);
        assert_eq!(resp.body, r#"{"result":"ok"}"#);
        mock.assert_async().await;
    }

    #[test]
    fn test_sidecar_constants() {
        assert_eq!(SIDECAR_ADDR, "127.0.0.1:18789");
        assert_eq!(SIDECAR_URL, "http://127.0.0.1:18789");
        assert_eq!(SPAWN_TIMEOUT, Duration::from_secs(5));
        assert_eq!(POLL_INTERVAL, Duration::from_millis(100));
    }

    #[tokio::test]
    async fn test_send_success_via_mock() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/")
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"status_code":200,"headers":{"x-test":"yes"},"body":"{\"data\":42}","error":""}"#)
            .create_async()
            .await;

        let client = GholaHttpClient::with_base_url(true, 4096, &server.url()).unwrap();
        let req = Request::get("https://api.example.com/v1");
        let resp = client.send(req).await.unwrap();

        assert_eq!(resp.status_code, 200);
        assert!(resp.is_success());
        assert_eq!(resp.body, r#"{"data":42}"#);
        assert_eq!(resp.headers.get("x-test").map(String::as_str), Some("yes"));
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn test_send_with_stealth_off() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/")
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"status_code":200,"headers":{},"body":"ok","error":""}"#)
            .create_async()
            .await;

        let client = GholaHttpClient::with_base_url(false, 4096, &server.url()).unwrap();
        let req = Request::post_json("https://api.example.com", r#"{"q":1}"#);
        let resp = client.send(req).await.unwrap();

        assert_eq!(resp.status_code, 200);
        assert_eq!(resp.body, "ok");
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn test_send_bridge_error_response() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/")
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"status_code":0,"headers":{},"body":"","error":"upstream timeout"}"#)
            .create_async()
            .await;

        let client = GholaHttpClient::with_base_url(true, 4096, &server.url()).unwrap();
        let req = Request::get("https://api.example.com");
        let result = client.send(req).await;

        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("upstream timeout"));
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn test_send_invalid_json_response() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/")
            .with_status(200)
            .with_body("not valid json at all")
            .create_async()
            .await;

        let client = GholaHttpClient::with_base_url(true, 4096, &server.url()).unwrap();
        let req = Request::get("https://api.example.com");
        let result = client.send(req).await;

        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("invalid sidecar response")
        );
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn test_send_non_success_bridge_status() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/")
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"status_code":429,"headers":{},"body":"rate limited","error":""}"#)
            .create_async()
            .await;

        let client = GholaHttpClient::with_base_url(false, 4096, &server.url()).unwrap();
        let req = Request::get("https://api.example.com");
        let resp = client.send(req).await.unwrap();

        assert_eq!(resp.status_code, 429);
        assert!(!resp.is_success());
        assert_eq!(resp.body, "rate limited");
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn test_send_with_custom_headers() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/")
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"status_code":200,"headers":{},"body":"{}","error":""}"#)
            .create_async()
            .await;

        let client = GholaHttpClient::with_base_url(true, 4096, &server.url()).unwrap();
        let req = Request::get("https://api.example.com")
            .with_header("Authorization", "Bearer token")
            .with_header("X-Chain", "ethereum");
        let resp = client.send(req).await.unwrap();

        assert!(resp.is_success());
        mock.assert_async().await;
    }

    #[tokio::test]
    async fn test_send_connection_refused() {
        let client = GholaHttpClient::with_base_url(true, 4096, "http://127.0.0.1:1").unwrap();
        let req = Request::get("https://api.example.com");
        let result = client.send(req).await;

        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("failed to reach ghola sidecar")
        );
    }

    #[tokio::test]
    async fn test_send_includes_buffer_size() {
        let mut server = mockito::Server::new_async().await;
        let mock = server
            .mock("POST", "/")
            .match_body(mockito::Matcher::PartialJsonString(
                r#"{"buffer_size":8192}"#.to_string(),
            ))
            .with_status(200)
            .with_header("content-type", "application/json")
            .with_body(r#"{"status_code":200,"headers":{},"body":"ok","error":""}"#)
            .create_async()
            .await;

        let client = GholaHttpClient::with_base_url(true, 8192, &server.url()).unwrap();
        let req = Request::get("https://api.example.com");
        let resp = client.send(req).await.unwrap();

        assert_eq!(resp.status_code, 200);
        mock.assert_async().await;
    }

    #[test]
    fn test_custom_buffer_size_stored() {
        let client = GholaHttpClient::new(true, 16384).unwrap();
        assert_eq!(client.buffer_size, 16384);
    }

    #[test]
    fn test_default_buffer_size() {
        let client = GholaHttpClient::new(false, 4096).unwrap();
        assert_eq!(client.buffer_size, 4096);
    }
}