huskies/server/src/chat/transport/matrix/pull.rs

//! `pull <channel>` gateway chat command — fetch a signed release manifest
//! from a configured channel, verify it, download and verify the artifact,
//! and install it atomically into the gateway's artifact store.
//!
//! Usage (gateway mode only): `{bot} pull <channel>`
//!
//! Pipeline:
//! 1. Look up `<channel>` in `projects.toml`'s `[release_channels.<name>]`
//!    table. Missing channel, missing `base_url`, or missing `pubkey` is a
//!    loud chat error — there is no unsigned-pull mode.
//! 2. `GET {base_url}/manifest.json` (with `Authorization: Bearer <token>`
//!    when a `bearer_token` is configured) and verify its Ed25519 signature
//!    against the channel's pinned `pubkey`.
//! 3. Refuse the manifest if its signed timestamp is not newer than the
//!    currently installed artifact's (rollback/replay protection).
//! 4. `GET {base_url}/{manifest.artifact}` and verify its sha256 against the
//!    manifest.
//! 5. Only once every check passes: atomically install the artifact plus its
//!    `.hash` and `.manifest.json` sidecars into `~/.huskies/artifacts/` —
//!    the same location `release` publishes to, so `upgrade all` picks it up
//!    unchanged.
//!
//! Any failure at any step installs nothing and reports a specific chat error.

use crate::service::gateway::config::ReleaseChannelConfig;
use crate::service::gateway::release_manifest as verify;
use std::collections::BTreeMap;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;

/// Parse a `pull <channel>` command from a raw message body.
///
/// Returns the channel name when the stripped message is `pull <name>`.
/// Returns `None` for `pull` with no argument or any other command.
pub fn extract_pull_command(message: &str, bot_name: &str, bot_user_id: &str) -> Option<String> {
    let stripped = crate::chat::util::strip_bot_mention(message, bot_name, bot_user_id);
    let trimmed = stripped
        .trim()
        .trim_start_matches(|c: char| !c.is_alphanumeric());

    let (cmd, rest) = match trimmed.split_once(char::is_whitespace) {
        Some((c, r)) => (c, r.trim()),
        None => (trimmed, ""),
    };

    if !cmd.eq_ignore_ascii_case("pull") || rest.is_empty() {
        return None;
    }
    Some(rest.split_whitespace().next().unwrap_or(rest).to_string())
}

/// Fetch, verify, and install the named release channel's latest artifact.
///
/// Returns a Markdown status message suitable for posting straight to chat.
pub async fn handle_pull(
    channel: &str,
    channels_store: &Arc<RwLock<BTreeMap<String, ReleaseChannelConfig>>>,
) -> String {
    let cfg = {
        let channels = channels_store.read().await;
        match channels.get(channel).cloned() {
            Some(c) => c,
            None => {
                let available: Vec<&String> = channels.keys().collect();
                return if available.is_empty() {
                    format!(
                        "No release channels are configured. Add one under \
                         `[release_channels.{channel}]` in projects.toml with `base_url` and `pubkey`."
                    )
                } else {
                    format!(
                        "Unknown channel `{channel}`. Configured channels: {}",
                        available
                            .iter()
                            .map(|s| s.as_str())
                            .collect::<Vec<_>>()
                            .join(", ")
                    )
                };
            }
        }
    };

    let Some(base_url) = cfg.base_url.filter(|u| !u.is_empty()) else {
        return format!(
            "Channel `{channel}` has no `base_url` configured — cannot pull. \
             There is no unsigned-pull mode."
        );
    };
    let Some(pubkey) = cfg.pubkey.filter(|p| !p.is_empty()) else {
        return format!(
            "Channel `{channel}` has no pinned `pubkey` configured — cannot pull. \
             There is no unsigned-pull mode."
        );
    };

    run_pull(
        channel,
        &base_url,
        &pubkey,
        cfg.bearer_token.as_deref(),
        &crate::http::artifacts_dir(),
        crate::http::SLED_ARTIFACT_NAME,
    )
    .await
}

/// Run the pull pipeline against an already-resolved channel config.
///
/// Split from [`handle_pull`] so tests can point `artifacts_dir` at a temp
/// directory instead of the real `~/.huskies/artifacts/`.
#[allow(clippy::too_many_arguments)]
async fn run_pull(
    channel: &str,
    base_url: &str,
    pubkey: &str,
    bearer_token: Option<&str>,
    artifacts_dir: &Path,
    artifact_name: &str,
) -> String {
    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(60))
        .build()
        .unwrap_or_default();

    let signed =
        match crate::service::gateway::io::fetch_release_manifest(&client, base_url, bearer_token)
            .await
        {
            Ok(s) => s,
            Err(e) => {
                return format!("Pull from `{channel}` failed: could not fetch manifest — {e}");
            }
        };

    if let Err(e) = verify::verify_manifest_signature(&signed, pubkey) {
        return format!("Pull from `{channel}` failed: {e}. Nothing was installed.");
    }

    if signed.manifest.channel != channel {
        return format!(
            "Pull from `{channel}` failed: manifest is signed for channel `{}`, not `{channel}`. \
             Nothing was installed.",
            signed.manifest.channel
        );
    }

    let installed =
        crate::service::gateway::io::read_installed_manifest(artifacts_dir, artifact_name);
    if let Err(e) = verify::check_rollback(&signed.manifest, installed.as_ref()) {
        return format!("Pull from `{channel}` refused: {e}");
    }

    let bytes = match crate::service::gateway::io::download_channel_artifact(
        &client,
        base_url,
        &signed.manifest.artifact,
        bearer_token,
    )
    .await
    {
        Ok(b) => b,
        Err(e) => {
            return format!(
                "Pull from `{channel}` failed: could not download artifact — {e}. Nothing was installed."
            );
        }
    };

    if let Err(e) = verify::verify_artifact_sha256(&bytes, &signed.manifest.sha256) {
        return format!("Pull from `{channel}` failed: {e}. Nothing was installed.");
    }

    if let Err(e) = crate::service::gateway::io::install_pulled_artifact(
        artifacts_dir,
        artifact_name,
        &bytes,
        &signed.manifest,
    ) {
        return format!("Pull from `{channel}` failed while installing: {e}");
    }

    let size_mb = bytes.len() / (1024 * 1024);
    format!(
        "Pulled **{}** from `{channel}` ({size_mb} MB, verified signature + sha256).\n\
         Say `upgrade all` to roll it out to the fleet.",
        signed.manifest.version
    )
}

// ── Tests ──────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use bft_json_crdt::keypair::{make_keypair, sign};
    use release_manifest::{ReleaseManifest, SignedManifest};

    fn make_store(
        entries: Vec<(&str, ReleaseChannelConfig)>,
    ) -> Arc<RwLock<BTreeMap<String, ReleaseChannelConfig>>> {
        let mut map = BTreeMap::new();
        for (name, cfg) in entries {
            map.insert(name.to_string(), cfg);
        }
        Arc::new(RwLock::new(map))
    }

    // ── extract_pull_command ─────────────────────────────────────────────────

    #[test]
    fn extract_pull_basic() {
        let cmd = extract_pull_command("Timmy pull stable", "Timmy", "@timmy:home");
        assert_eq!(cmd, Some("stable".to_string()));
    }

    #[test]
    fn extract_pull_case_insensitive() {
        let cmd = extract_pull_command("Timmy PULL stable", "Timmy", "@timmy:home");
        assert_eq!(cmd, Some("stable".to_string()));
    }

    #[test]
    fn extract_pull_no_arg_is_none() {
        assert_eq!(
            extract_pull_command("Timmy pull", "Timmy", "@timmy:home"),
            None
        );
    }

    #[test]
    fn extract_non_pull_returns_none() {
        assert_eq!(
            extract_pull_command("Timmy status", "Timmy", "@timmy:home"),
            None
        );
    }

    #[test]
    fn extract_pull_full_user_id() {
        let cmd = extract_pull_command("@timmy:home pull nightly", "Timmy", "@timmy:home");
        assert_eq!(cmd, Some("nightly".to_string()));
    }

    // ── handle_pull config validation ────────────────────────────────────────

    #[tokio::test]
    async fn pull_unknown_channel_reports_error() {
        let store = make_store(vec![]);
        let msg = handle_pull("stable", &store).await;
        assert!(
            msg.contains("No release channels"),
            "empty store should say no channels: {msg}"
        );
    }

    #[tokio::test]
    async fn pull_unknown_channel_among_configured_lists_them() {
        let store = make_store(vec![(
            "beta",
            ReleaseChannelConfig {
                base_url: Some("http://example.com".into()),
                pubkey: Some("ab".repeat(32)),
                bearer_token: None,
            },
        )]);
        let msg = handle_pull("stable", &store).await;
        assert!(msg.contains("Unknown channel"));
        assert!(msg.contains("beta"));
    }

    #[tokio::test]
    async fn pull_missing_base_url_is_actionable_error() {
        let store = make_store(vec![(
            "stable",
            ReleaseChannelConfig {
                base_url: None,
                pubkey: Some("ab".repeat(32)),
                bearer_token: None,
            },
        )]);
        let msg = handle_pull("stable", &store).await;
        assert!(msg.contains("no `base_url`"));
        assert!(msg.contains("no unsigned-pull mode"));
    }

    #[tokio::test]
    async fn pull_missing_pubkey_is_actionable_error() {
        let store = make_store(vec![(
            "stable",
            ReleaseChannelConfig {
                base_url: Some("http://example.com".into()),
                pubkey: None,
                bearer_token: None,
            },
        )]);
        let msg = handle_pull("stable", &store).await;
        assert!(msg.contains("no pinned `pubkey`"));
        assert!(msg.contains("no unsigned-pull mode"));
    }

    // ── run_pull end-to-end against a local mock channel ─────────────────────

    /// Spawn a minimal HTTP server serving fixed responses for
    /// `/manifest.json` and `/<artifact>` so `run_pull` can be exercised
    /// end-to-end without a real network.
    ///
    /// `run_pull` issues its manifest fetch and artifact download
    /// sequentially over one shared `reqwest::Client`, whose HTTP/1.1 pool
    /// assumes keep-alive by default. Every response here closes its
    /// connection after a single request, so we send `Connection: close` to
    /// stop the client from ever trying to reuse a connection this server
    /// has already torn down (a real source of intermittent "cannot reach"
    /// fetch errors that raced ahead of the sha256 check). We also block on
    /// a readiness signal fired from inside the accept loop so the returned
    /// port is guaranteed to have an active acceptor before any caller
    /// connects to it.
    async fn spawn_mock_channel(manifest_body: Vec<u8>, artifact_body: Vec<u8>) -> u16 {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
        tokio::spawn(async move {
            let _ = ready_tx.send(());
            loop {
                let Ok((mut stream, _)) = listener.accept().await else {
                    break;
                };
                let manifest_body = manifest_body.clone();
                let artifact_body = artifact_body.clone();
                tokio::spawn(async move {
                    use tokio::io::{AsyncReadExt, AsyncWriteExt};
                    let mut buf = [0u8; 4096];
                    let n = stream.read(&mut buf).await.unwrap_or(0);
                    let req = String::from_utf8_lossy(&buf[..n]).to_string();
                    let path = req.split_whitespace().nth(1).unwrap_or("");
                    let body = if path.ends_with("manifest.json") {
                        manifest_body
                    } else {
                        artifact_body
                    };
                    let response = format!(
                        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                        body.len()
                    );
                    let _ = stream.write_all(response.as_bytes()).await;
                    let _ = stream.write_all(&body).await;
                    let _ = stream.shutdown().await;
                });
            }
        });
        let _ = ready_rx.await;
        port
    }

    fn signed(
        kp: &bft_json_crdt::keypair::Ed25519KeyPair,
        manifest: ReleaseManifest,
    ) -> SignedManifest {
        let sig = sign(kp, &manifest.canonical_bytes());
        SignedManifest {
            manifest,
            signature: sig.to_bytes().iter().map(|b| format!("{b:02x}")).collect(),
        }
    }

    #[tokio::test]
    async fn pull_accepts_valid_signature_and_matching_sha256() {
        let kp = make_keypair();
        let pubkey = crate::node_identity::public_key_hex(&kp);

        use sha2::{Digest, Sha256};
        let artifact_body = b"a genuinely valid binary".to_vec();
        let mut hasher = Sha256::new();
        hasher.update(&artifact_body);
        let sha256: String = hasher
            .finalize()
            .iter()
            .map(|b| format!("{b:02x}"))
            .collect();

        let manifest = ReleaseManifest {
            artifact: "huskies-linux-arm64".to_string(),
            sha256,
            version: "abc1234".to_string(),
            channel: "stable".to_string(),
            timestamp: 1,
        };
        let signed_manifest = signed(&kp, manifest);
        let manifest_body = serde_json::to_vec(&signed_manifest).unwrap();

        let port = spawn_mock_channel(manifest_body, artifact_body).await;
        let dir = tempfile::tempdir().unwrap();

        let msg = run_pull(
            "stable",
            &format!("http://127.0.0.1:{port}"),
            &pubkey,
            None,
            dir.path(),
            "huskies-linux-arm64",
        )
        .await;
        assert!(msg.contains("Pulled"), "expected success message: {msg}");
        assert!(msg.contains("abc1234"));
        assert!(dir.path().join("huskies-linux-arm64").exists());
        assert!(dir.path().join("huskies-linux-arm64.hash").exists());
    }

    #[tokio::test]
    async fn pull_rejects_manifest_from_wrong_key() {
        let kp = make_keypair();
        let wrong_kp = make_keypair();
        let wrong_pubkey = crate::node_identity::public_key_hex(&wrong_kp);

        let manifest = ReleaseManifest {
            artifact: "huskies-linux-arm64".to_string(),
            sha256: "a".repeat(64),
            version: "abc1234".to_string(),
            channel: "stable".to_string(),
            timestamp: 1,
        };
        let signed_manifest = signed(&kp, manifest);
        let manifest_body = serde_json::to_vec(&signed_manifest).unwrap();

        let port = spawn_mock_channel(manifest_body, b"binary".to_vec()).await;
        let dir = tempfile::tempdir().unwrap();

        let msg = run_pull(
            "stable",
            &format!("http://127.0.0.1:{port}"),
            &wrong_pubkey,
            None,
            dir.path(),
            "huskies-linux-arm64",
        )
        .await;
        assert!(
            msg.contains("signature"),
            "expected signature failure: {msg}"
        );
        assert!(msg.contains("Nothing was installed"));
        assert!(!dir.path().join("huskies-linux-arm64").exists());
    }

    #[tokio::test]
    async fn pull_rejects_tampered_artifact_sha256_mismatch() {
        let kp = make_keypair();
        let pubkey = crate::node_identity::public_key_hex(&kp);

        use sha2::{Digest, Sha256};
        let mut hasher = Sha256::new();
        hasher.update(b"the real bytes");
        let real_sha256: String = hasher
            .finalize()
            .iter()
            .map(|b| format!("{b:02x}"))
            .collect();

        let manifest = ReleaseManifest {
            artifact: "huskies-linux-arm64".to_string(),
            sha256: real_sha256,
            version: "abc1234".to_string(),
            channel: "stable".to_string(),
            timestamp: 1,
        };
        let signed_manifest = signed(&kp, manifest);
        let manifest_body = serde_json::to_vec(&signed_manifest).unwrap();

        // Serve different bytes than what was signed — sha256 mismatch.
        let port = spawn_mock_channel(manifest_body, b"tampered bytes".to_vec()).await;
        let dir = tempfile::tempdir().unwrap();

        let msg = run_pull(
            "stable",
            &format!("http://127.0.0.1:{port}"),
            &pubkey,
            None,
            dir.path(),
            "huskies-linux-arm64",
        )
        .await;
        assert!(
            msg.contains("sha256 mismatch"),
            "expected sha256 failure: {msg}"
        );
        assert!(
            !msg.contains("could not fetch manifest")
                && !msg.contains("could not download artifact"),
            "manifest fetch and artifact download must succeed so the sha256 check is reached, \
             not preempted by a fetch/download error: {msg}"
        );
        assert!(msg.contains("Nothing was installed"));
        assert!(!dir.path().join("huskies-linux-arm64").exists());
    }

    /// Stress variant of [`pull_rejects_tampered_artifact_sha256_mismatch`]:
    /// repeats the same scenario against many independent mock channels to
    /// catch any reintroduction of the fetch-ordering race under load.
    #[tokio::test]
    async fn pull_rejects_tampered_artifact_sha256_mismatch_stress() {
        for _ in 0..50 {
            let kp = make_keypair();
            let pubkey = crate::node_identity::public_key_hex(&kp);

            use sha2::{Digest, Sha256};
            let mut hasher = Sha256::new();
            hasher.update(b"the real bytes");
            let real_sha256: String = hasher
                .finalize()
                .iter()
                .map(|b| format!("{b:02x}"))
                .collect();

            let manifest = ReleaseManifest {
                artifact: "huskies-linux-arm64".to_string(),
                sha256: real_sha256,
                version: "abc1234".to_string(),
                channel: "stable".to_string(),
                timestamp: 1,
            };
            let signed_manifest = signed(&kp, manifest);
            let manifest_body = serde_json::to_vec(&signed_manifest).unwrap();

            let port = spawn_mock_channel(manifest_body, b"tampered bytes".to_vec()).await;
            let dir = tempfile::tempdir().unwrap();

            let msg = run_pull(
                "stable",
                &format!("http://127.0.0.1:{port}"),
                &pubkey,
                None,
                dir.path(),
                "huskies-linux-arm64",
            )
            .await;
            assert!(
                msg.contains("sha256 mismatch"),
                "expected sha256 failure on every iteration: {msg}"
            );
            assert!(!dir.path().join("huskies-linux-arm64").exists());
        }
    }

    #[tokio::test]
    async fn pull_missing_manifest_endpoint_is_loud_error() {
        let dir = tempfile::tempdir().unwrap();
        let msg = run_pull(
            "stable",
            "http://127.0.0.1:1", // nothing listening
            &"ab".repeat(32),
            None,
            dir.path(),
            "huskies-linux-arm64",
        )
        .await;
        assert!(msg.contains("could not fetch manifest"));
    }

    #[tokio::test]
    async fn pull_refuses_manifest_older_than_installed() {
        let kp = make_keypair();
        let pubkey = crate::node_identity::public_key_hex(&kp);
        let dir = tempfile::tempdir().unwrap();

        // Pre-install a manifest with a later timestamp than the one about to be pulled.
        let installed = ReleaseManifest {
            artifact: "huskies-linux-arm64".to_string(),
            sha256: "a".repeat(64),
            version: "newer0000".to_string(),
            channel: "stable".to_string(),
            timestamp: 100,
        };
        crate::service::gateway::io::install_pulled_artifact(
            dir.path(),
            "huskies-linux-arm64",
            b"already installed",
            &installed,
        )
        .unwrap();

        let older_manifest = ReleaseManifest {
            artifact: "huskies-linux-arm64".to_string(),
            sha256: "b".repeat(64),
            version: "older0000".to_string(),
            channel: "stable".to_string(),
            timestamp: 50,
        };
        let signed_manifest = signed(&kp, older_manifest);
        let manifest_body = serde_json::to_vec(&signed_manifest).unwrap();
        let port = spawn_mock_channel(manifest_body, b"stale binary".to_vec()).await;

        let msg = run_pull(
            "stable",
            &format!("http://127.0.0.1:{port}"),
            &pubkey,
            None,
            dir.path(),
            "huskies-linux-arm64",
        )
        .await;
        assert!(msg.contains("refused"), "expected rollback refusal: {msg}");
        // The previously installed artifact must remain untouched.
        assert_eq!(
            std::fs::read(dir.path().join("huskies-linux-arm64")).unwrap(),
            b"already installed"
        );
    }
}