microvms-core 1.3.0

// SPDX-License-Identifier: Apache-2.0
//! Core's answers to the shared case corpus (`verify/parity/cases/`, #272).
//!
//! Here rather than in microvms-domain or microvms-app because this is the one crate whose tests
//! reach every area: the domain can't see the app's `wrap_dockerfile`, `egress_posture_for` or
//! image names, and neither crate has a `tests/` directory. Each handler calls the function the
//! capability table names for core, with nothing between the case and the call but type
//! conversion; the rules for reading and judging a case are in `parity_corpus`.

mod parity_corpus;

use std::sync::Arc;

use futures_util::future::BoxFuture;
use microvms_app::testing::{CountingMinter, FakeControlPlane};
use microvms_core::agents::{Agent, AgentSpec, AgentVm};
use microvms_core::control::artifact::{
    BaseImage, WrapOptions, default_dockerfile, wrap_dockerfile,
};
use microvms_core::control::ensure::{EnsureImageRequest, prepare};
use microvms_core::control::{ControlPlane, DEFAULT_AGENT_PORT, SystemClock, egress_posture_for};
use microvms_core::cost::{
    CalendarDate, DEFAULT_ESTIMATE_LABEL, DEFAULT_REPORT_CYCLES, DEFAULT_RESIDENCY_CYCLES,
    DEFAULT_RUN_LABEL, DurationP, PlanUsage, RunUsage, compare_residency, estimate_run,
    pinned_rates, run_report,
};
use microvms_core::names::FileNameStore;
use microvms_core::prelude::*;
use microvms_core::sandbox::{RunRequest, Sandbox};
use microvms_core::session::{HttpBackend, HttpRequest, HttpResponse, OpenStream, Session};
use microvms_core::{Error, Region, SizeClass};
use parity_corpus::{Case, Run};
use serde_json::{Value, json};

const AREAS: [&str; 11] = [
    "forward",
    "control-plane",
    "credential-error",
    "image-name",
    "cost",
    "error",
    "egress",
    "names",
    "size-class",
    "wrap-dockerfile",
    "tunnel",
];

const BUCKET: &str = "parity-cases-bucket";
const BUILD_ROLE: &str = "arn:aws:iam::123456789012:role/build";

#[tokio::test]
async fn core_answers_the_shared_case_corpus() {
    let mut run = Run::plan("core", &AREAS, &[]);
    for case in run.cases() {
        let answer = match case.area.as_str() {
            "control-plane" => control_plane_case(&case).await,
            "credential-error" => credential_error(&case),
            "forward" => forward(&case).await,
            "image-name" => image_name(&case),
            "cost" => cost(&case),
            "error" => daemon_status(&case).await,
            "egress" => egress(&case).await,
            "names" => names(&case).await,
            "size-class" => size_class(&case),
            "wrap-dockerfile" => wrap(&case),
            "tunnel" => tunnel(&case).await,
            other => unreachable!("planned only the owned areas, not {other}"),
        };
        run.judge(&case, answer);
    }
    run.finish();
}

fn credential_error(case: &Case) -> Value {
    use std::process::{Command, Stdio};

    let fixture = parity_corpus::credentials::fixture_path();
    let child = Command::new(&fixture)
        .args(["credentials", "launch"])
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .spawn()
        .expect("build the credential fixture with mise run parity:fixtures");
    parity_corpus::credentials::run(
        child,
        "core",
        fixture,
        &["credentials", "core"],
        &case.input,
    )
}

async fn tunnel(case: &Case) -> Value {
    use microvms_core::session::serve::{ServeLimits, TunnelTarget, start_tunnel};
    use std::process::{Command, Stdio};
    use tokio::io::AsyncReadExt as _;

    let child = Command::new(parity_corpus::tunnel::fixture_path())
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::inherit())
        .spawn()
        .expect("build the fixture with mise run parity:fixtures");
    let mut peer = parity_corpus::tunnel::Peer::started(child, &case.input);
    let ready = &peer.ready;
    let identity = microvms_core::identity::TunnelIdentity::from_encoded_parts(
        ready["host_seed"].as_str().unwrap(),
        ready["vm_public_key"].as_str().unwrap(),
    )
    .unwrap();
    let session = Session::builder(ready["endpoint"].as_str().unwrap(), "parity-tunnel-token")
        .build()
        .unwrap();
    let serving = start_tunnel(
        "127.0.0.1:0".parse().unwrap(),
        TunnelTarget::for_session(
            &session,
            case.input_u64("guest_port") as u16,
            Some(identity),
        ),
        ServeLimits {
            max_connections: Some(1),
        },
    )
    .await
    .unwrap();
    let mut client = tokio::net::TcpStream::connect(serving.local_address())
        .await
        .unwrap();
    let mut bytes = Vec::new();
    // A refused verified tunnel resets the local socket; preserve that verdict in the case.
    let reset = match tokio::time::timeout(
        std::time::Duration::from_secs(20),
        client.read_to_end(&mut bytes),
    )
    .await
    .expect("tunnel completes")
    {
        Ok(_) => false,
        Err(error) if error.kind() == std::io::ErrorKind::ConnectionReset => true,
        Err(error) => panic!("the tunnel read failed: {error}"),
    };
    let report = serving.finished().await.unwrap();
    let verdict = peer.finish();
    json!({"received": String::from_utf8(bytes).unwrap(), "refused": report.report.refused,
        "confirmed": verdict["confirmed"], "reset": reset})
}

/// The facets every runner reports for a failure.
fn refusal(error: &Error) -> Value {
    json!({"error": {
        "code": error.code(),
        "wire_kind": error.wire_kind().map(|kind| kind.as_str()),
        "retryable": error.retryable(),
    }})
}

/// A control plane that makes no call: the fake has nothing queued, so one would panic
/// naming the operation instead of passing against AWS.
fn offline_plane() -> ControlPlane {
    ControlPlane::with_transport(
        Arc::new(FakeControlPlane::new()),
        Region::US_EAST_1,
        Arc::new(SystemClock::new()),
    )
}

fn size(case: &Case) -> SizeClass {
    let mib = u32::try_from(case.input_u64("size_mib")).expect("a size in range");
    SizeClass::from_baseline_mib(mib).expect("a documented size class")
}

fn image_name(case: &Case) -> Value {
    match case.capability.as_str() {
        // `ensure_image`'s local half, which names the image before any call.
        "ensure-image" => {
            let dockerfile =
                default_dockerfile(DEFAULT_AGENT_PORT, None, &BaseImage::al2023(), None);
            let mut request = EnsureImageRequest::new(
                case.input_str("name_prefix"),
                case.input_binary(),
                dockerfile,
                BUCKET,
                BUILD_ROLE,
            );
            request.size = size(case);
            match prepare(&offline_plane(), request) {
                Ok(prepared) => json!({"name": prepared.name}),
                Err(error) => refusal(&error),
            }
        }
        "agent-image-name" => {
            let specs = case
                .input_strings("agents")
                .iter()
                .map(|name| {
                    let agent = Agent::parse(name).unwrap_or_else(|| panic!("an agent: {name}"));
                    AgentSpec::new(agent)
                })
                .collect();
            let answer = AgentVm::new(Sandbox::with_control_plane(offline_plane()), specs)
                .and_then(|vm| {
                    let request = vm.image_request(case.input_binary(), BUILD_ROLE, size(case))?;
                    Ok(vm.image_name(&request))
                });
            match answer {
                Ok(name) => json!({"name": name}),
                Err(error) => refusal(&error),
            }
        }
        other => panic!("{}: no core handler for {other:?} in image-name", case.id),
    }
}

/// The case's cycle count, or `default` when it leaves one out: core has no optional arguments,
/// so its runner passes the constant a surface's caller gets by leaving the count out.
fn cycles(case: &Case, key: &str, default: u32) -> u32 {
    case.input.get(key).map_or(default, |_| {
        u32::try_from(case.input_u64(key)).expect("a cycle count in range")
    })
}

fn cost(case: &Case) -> Value {
    let (rates, today) = (pinned_rates(), CalendarDate::today_utc());
    let answer = match case.capability.as_str() {
        "estimate" => {
            let mut plan = PlanUsage {
                running_seconds: case.input_f64("running_seconds"),
                suspended_seconds: case.input_f64("suspended_seconds"),
                suspend_resume_cycles: cycles(case, "suspend_resume_cycles", DEFAULT_REPORT_CYCLES),
                ..PlanUsage::default()
            };
            // Core has no optional arguments, so its runner applies the defaults every
            // surface's caller gets by leaving `launched` and the label out: core's inference
            // and core's label.
            plan.launched = plan.infer_launched();
            estimate_run(size(case), &plan, &rates, today, DEFAULT_ESTIMATE_LABEL)
                .map(|report| report.to_json())
        }
        "run-report" => DurationP::measured_secs_f64(case.input_f64("running_seconds"))
            .and_then(|running| {
                let mut usage = RunUsage {
                    running: Some(running),
                    suspend_resume_cycles: cycles(
                        case,
                        "suspend_resume_cycles",
                        DEFAULT_REPORT_CYCLES,
                    ),
                    ..RunUsage::default()
                };
                usage.launched = usage.infer_launched();
                run_report(size(case), &usage, &rates, today, DEFAULT_RUN_LABEL)
            })
            .map(|report| report.to_json()),
        "compare-residency" => compare_residency(
            size(case),
            std::time::Duration::from_secs(case.input_u64("hold_seconds")),
            cycles(case, "cycles", DEFAULT_RESIDENCY_CYCLES),
            &rates,
            today,
        )
        .and_then(|comparison| {
            Ok(json!({
                "cycles": comparison.cycles(),
                "ratio": comparison.ratio().to_string(),
                "render": comparison.render()?,
            }))
        }),
        other => panic!("{}: no core handler for {other:?} in cost", case.id),
    };
    answer.unwrap_or_else(|error| refusal(&error))
}

/// A daemon that answers every request with the case's status and body.
struct ScriptedStatus {
    status: u16,
    body: Vec<u8>,
    upload: bool,
}

impl HttpBackend for ScriptedStatus {
    fn send(&self, request: HttpRequest) -> BoxFuture<'_, Result<HttpResponse, Error>> {
        let discovery = self.upload && request.path == "/v1/health";
        let response = HttpResponse {
            status: if discovery { 200 } else { self.status },
            headers: std::collections::HashMap::new(),
            body: if discovery {
                Vec::new()
            } else {
                self.body.clone()
            },
        };
        Box::pin(async move { Ok(response) })
    }

    fn open_stream(
        &self,
        request: HttpRequest,
        _idle_timeout: std::time::Duration,
    ) -> BoxFuture<'_, Result<OpenStream, Error>> {
        panic!(
            "no error case streams, but {} {} did",
            request.method, request.path
        )
    }
}

async fn daemon_status(case: &Case) -> Value {
    if case.input.get("response").is_some() {
        use futures_util::StreamExt as _;
        let upstream = parity_corpus::ForwardUpstream::for_case(case).await;
        let session = Session::direct(format!("http://{}", upstream.address), "agent-token")
            .expect("session");
        let result = if case.capability == "stream" {
            let handle = session.exec("x-1");
            let stream = handle.stream();
            futures_util::pin_mut!(stream);
            stream.next().await.expect("stream response").map(|_| ())
        } else if case.capability == "upload-file" {
            session
                .upload_file(case.input_str("path"), b"parity", None)
                .await
        } else {
            session.health().await.map(|_| ())
        };
        let mut answer = match result {
            Ok(()) => json!({"ok":true}),
            Err(error) => refusal(&error),
        };
        answer["upstream_requests"] = json!(upstream.request_count());
        return answer;
    }
    let backend = Arc::new(ScriptedStatus {
        status: u16::try_from(case.input_u64("status")).expect("a status"),
        body: case.input_str("body").as_bytes().to_vec(),
        upload: matches!(case.capability.as_str(), "upload-file" | "sync-directory"),
    });
    let session = Session::builder("http://127.0.0.1:9", "agent-token")
        .with_backend(backend)
        .build()
        .expect("the session builds");
    let result = match case.capability.as_str() {
        "health" => session.health().await.map(|_| ()),
        "upload-file" => {
            session
                .upload_file(case.input_str("path"), b"parity", None)
                .await
        }
        // No baseline, so the whole tree travels and its upload meets the status.
        "sync-directory" => {
            let dir = tempfile::tempdir().expect("a temp dir");
            std::fs::write(dir.path().join("upload.bin"), b"parity").expect("writes");
            session
                .sync_pass(
                    &microvms_core::workspace::DiskTree,
                    dir.path(),
                    None,
                    std::time::Duration::from_secs(60),
                )
                .await
                .map(|_| ())
        }
        other => panic!("{}: no core handler for {other:?} in error", case.id),
    };
    match result {
        Ok(()) => json!({"ok": true}),
        Err(error) => refusal(&error),
    }
}

async fn egress(case: &Case) -> Value {
    let egress = case.input_bool("egress");
    let connectors = case.input_strings("egress_network_connectors");
    let deny_egress = case.input_bool("deny_egress");
    match case.capability.as_str() {
        "launch" => {
            let mut request = RunRequest::new()
                .with_image("arn:aws:lambda:us-east-1:123456789012:microvm-image:img");
            request.egress = egress;
            request.egress_network_connectors = connectors;
            request.deny_egress = deny_egress;
            let mut sandbox = Sandbox::with_control_plane(offline_plane());
            match sandbox.run(request).await {
                Ok(_) => json!({"ok": true}),
                Err(error) => refusal(&error),
            }
        }
        "egress-posture-for" => match egress_posture_for(egress, &connectors, deny_egress, None) {
            Ok(posture) => json!({"posture": posture.as_str()}),
            Err(error) => refusal(&error),
        },
        other => panic!("{}: no core handler for {other:?} in egress", case.id),
    }
}

/// The case's record written where the CLI's registry keeps it, then adopted by name.
///
/// `Sandbox::from_name` is `names::resolve`, a plane for the record's region, then
/// `adopt_record`. The plane is the one step swapped, for the offline one, so a regression that
/// lets the record through fails on the fake's panic rather than resolving credentials and
/// calling AWS from the test host. The Python and TypeScript runners call `from_name` itself,
/// behind a proxy nothing listens on.
async fn names(case: &Case) -> Value {
    assert_eq!(
        case.capability, "from-name",
        "{}: a from-name case",
        case.id
    );
    let state = tempfile::tempdir().expect("a state directory");
    let home = case.input.get("home").and_then(Value::as_str);
    let store = match home {
        Some(home) => FileNameStore::default_location(&|key| {
            matches!(key, "HOME" | "USERPROFILE").then(|| home.to_owned())
        }),
        None => FileNameStore::in_state_root(state.path()),
    };
    let name = case.input_str("name");
    if home.is_none() {
        std::fs::create_dir_all(store.dir()).expect("the names directory");
        std::fs::write(store.path_of(name), case.input_str("record_text"))
            .expect("writes the record");
    }
    let region: Region = case.input_str("region").parse().expect("a region");
    let adopted = match microvms_core::names::resolve(&store, name, Some(&region)) {
        Ok(record) => Sandbox::adopt_record(offline_plane(), record).await,
        Err(error) => Err(error),
    };
    match adopted {
        Ok(_) => json!({"ok": true}),
        Err(error) => {
            let mut answer = refusal(&error);
            answer["message_mentions"] = case.message_mentions(&error.to_string());
            answer
        }
    }
}

fn size_class(case: &Case) -> Value {
    let cpus = case.input("cpus").as_f64();
    let memory = case
        .input("memory_mib")
        .as_u64()
        .map(|mib| u32::try_from(mib).expect("a memory figure in range"));
    match SizeClass::from_request(cpus, memory) {
        Ok(class) => json!({"baseline_mib": class.baseline_mib()}),
        Err(error) => refusal(&error),
    }
}

fn wrap(case: &Case) -> Value {
    match wrap_dockerfile(case.input_str("task"), &WrapOptions::default()) {
        Ok(text) => json!({"dockerfile": text}),
        Err(error) => refusal(&error),
    }
}

async fn forward(case: &Case) -> Value {
    use microvms_core::session::proxy::ProxyAuth;
    use microvms_core::session::serve::{ServeLimits, start_forward};
    let upstream = parity_corpus::ForwardUpstream::for_case(case).await;
    let auth = Arc::new(ProxyAuth::with_clock(
        Arc::new(CountingMinter::default()),
        DEFAULT_AGENT_PORT,
        Arc::new(SystemClock::new()),
    ));
    let forward = start_forward(
        format!(
            "{}:0",
            case.input
                .get("bind_ip")
                .and_then(Value::as_str)
                .unwrap_or("127.0.0.1")
        )
        .parse()
        .expect("local bind"),
        &format!("http://{}", upstream.address),
        upstream.address.port(),
        Some(Arc::clone(&auth)),
        ServeLimits {
            max_connections: Some(1),
        },
    )
    .await
    .expect("starts the forward");
    let address = forward.local_address();
    let request = parity_corpus::forward_request(case, address.port());
    let response =
        parity_corpus::forward_exchange(([127, 0, 0, 1], address.port()).into(), &request).await;
    let summary = forward
        .stop(Some(std::time::Duration::from_secs(10)))
        .await
        .expect("stops");
    // A refused request must stop before the proxy credential is requested. Accepted cases
    // prove this is a live minter and the actual forwarding path reaches the upstream.
    assert_eq!(
        auth.mint_count(),
        case.expect["upstream_requests"].as_u64().unwrap(),
        "{}: mints",
        case.id
    );
    let mut answer = parity_corpus::forward_answer(&response, case);
    answer["served"] = json!(summary.report.served);
    answer["refused"] = json!(summary.report.refused);
    answer["upstream_requests"] = json!(upstream.request_count());
    answer
}

async fn control_plane_case(case: &Case) -> Value {
    let script =
        microvms_app::testing::ParityControlPlane::from_json(&case.input("script").to_string())
            .unwrap();
    let plane = match case.input.get("retry") {
        Some(retry) => script.plane().with_retry(
            microvms_core::control::transport::RetryPolicy::new(
                retry["max_attempts"].as_u64().unwrap().try_into().unwrap(),
                microvms_core::duration::of_secs_f64(
                    retry["max_elapsed_seconds"].as_f64().unwrap(),
                )
                .unwrap(),
            )
            .unwrap(),
        ),
        None => script.plane(),
    };
    if case.input_strings("actions") == ["sweep"] {
        let options = microvms_core::control::SweepOptions {
            keep: case.input_strings("keep"),
            dry_run: case.input_bool("dry_run"),
        };
        let report = plane
            .sweep(case.input_str("prefix"), &options)
            .await
            .unwrap();
        return json!({"report": report, "calls": script.calls()});
    }
    if case.input_strings("actions") == ["list"] {
        let items = plane
            .list_microvm_summaries(&Default::default())
            .await
            .unwrap();
        // CLI remote listing also inventories images, so every surface replays both reads.
        plane.list_images().await.unwrap();
        return json!({"items": items.iter().map(|item| json!({
            "id": item.microvm_id, "state": item.state, "image_arn": item.image_arn,
            "started_at": item.started_at,
        })).collect::<Vec<_>>(), "calls": script.calls()});
    }
    let id = case.input_str("microvm_id");
    for action in case.input_strings("actions") {
        let result = async {
            if action == "suspend" {
                assert_eq!(plane.get_microvm(id).await?.state, "RUNNING");
                plane.suspend(id).await?;
                assert_eq!(plane.get_microvm(id).await?.state, "SUSPENDED");
                Ok(())
            } else {
                assert_eq!(action, "resume");
                plane.resume(id).await
            }
        }
        .await;
        if let Err(error) = result {
            let mut answer = refusal(&error);
            if case.input.get("message_mentions").is_some() {
                answer["message_mentions"] = case.message_mentions(&error.to_string());
            }
            if case.input.get("report_retry").is_some() {
                answer["retry"] = json!({"attempt_count": error.attempt_count(), "backoff_positive": !error.backoff().is_zero(), "throttled": error.throttled()});
            }
            answer["calls"] = json!(script.calls());
            return answer;
        }
    }
    panic!("a control-plane case must end with a refusal")
}