//! Host side of a microVM's only route out: an HTTP `CONNECT` proxy on a Unix
//! socket, one per VM.
//!
//! # Why the guest has no network card
//!
//! It could have had one. A TAP device plus NAT is what the Firecracker
//! write-ups do, and it was measured against this before being rejected:
//!
//! - `ip tuntap add` is **denied to the daemon user** (needs `CAP_NET_ADMIN`), so
//!   TAP would need root to pre-provision devices at setup time — the same
//!   privilege detour the loop-mounted rootfs already forced.
//! - tank's `FORWARD` policy is `DROP` with Docker and Tailscale chains, so rules
//!   would have to be *inserted* at position 1; appended ones die silently.
//! - a leaked TAP device is a new class of host litter to reap.
//!
//! Against that, `CONNECT` needs no privilege at all, and it is better on the
//! merits: the client hands us the **hostname**, so resolution happens here and
//! the guest needs no DNS or `resolv.conf`; the allow-list is by name rather than
//! by address; and nothing in the guest can reach the network except through this
//! function. That is what the isolation plan's egress restriction actually asked
//! for, and it is strictly tighter than the mission container's present full
//! egress on `clawmates_edge`.
//!
//! The design rests on one measured fact: **`claude` honours `HTTPS_PROXY`**.
//! With the proxy pointed at a closed port, `claude -p` fails with
//! `ConnectionRefused` instead of answering. (That could only be measured in a
//! container — inside a VM the CLI collapses every failure into `Execution
//! error`.)
//!
//! # Shape
//!
//! Firecracker's convention for a guest-initiated connection is that the **host**
//! listens on `<uds_path>_<port>`. The guest's agent pumps bytes from
//! `127.0.0.1:3128` to vsock port 9002 and parses nothing, so all policy is here
//! and a compromised guest cannot argue with it.

use std::path::{Path, PathBuf};
use std::sync::Arc;

use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::net::{TcpStream, UnixListener, UnixStream};

/// Port the guest dials. Must match `fcagent`'s `EGRESS_PORT`.
pub const EGRESS_PORT: u32 = 9002;

/// Hosts a mission may reach, unless `CLAWMATES_FC_EGRESS_ALLOW` overrides it.
///
/// Narrow on purpose: this is the whole egress surface of a mission, and the
/// point of the exercise is that it is smaller than "the internet". An entry
/// beginning with `.` matches subdomains.
const DEFAULT_ALLOW: &[&str] = &[
    // The model API. Without it there is no agent.
    "api.anthropic.com",
    ".anthropic.com",
    // Our forge: clone, push, PRs.
    "git.redclaw.dev",
];

/// Parse the allow-list once per VM.
///
/// An empty `CLAWMATES_FC_EGRESS_ALLOW` means **deny everything**, not "fall back
/// to the default": an operator who blanked it asked for no egress, and quietly
/// restoring the default would hand a mission the network they just took away.
fn allow_list() -> Vec<String> {
    match std::env::var("CLAWMATES_FC_EGRESS_ALLOW") {
        Ok(raw) => raw
            .split(',')
            .map(|s| s.trim().to_ascii_lowercase())
            .filter(|s| !s.is_empty())
            .collect(),
        Err(_) => DEFAULT_ALLOW.iter().map(|s| s.to_string()).collect(),
    }
}

/// Is `host` allowed?
///
/// Case-insensitive, port already stripped. A leading `.` in an entry matches
/// that domain and its subdomains; anything else must match exactly. Deliberately
/// not a substring test — `api.anthropic.com.evil.test` contains the allowed name
/// and must not pass.
fn host_allowed(host: &str, allow: &[String]) -> bool {
    let host = host.trim().trim_end_matches('.').to_ascii_lowercase();
    if host.is_empty() {
        return false;
    }
    // A hostname is letters, digits, dots and hyphens — nothing else. This is
    // load-bearing, not hygiene: `evil.test/api.anthropic.com` ends with an
    // allowed suffix and would otherwise PASS the match below. A unit test found
    // it. Rejecting the character class also refuses IP literals, so an address
    // cannot be used to sidestep a list written in names.
    if !host
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '-')
    {
        return false;
    }
    allow.iter().any(|a| match a.strip_prefix('.') {
        Some(domain) => host == domain || host.ends_with(&format!(".{domain}")),
        None => host == *a,
    })
}

/// Split `host:port` from a CONNECT target.
///
/// Only 443 is allowed. Permitting arbitrary ports would turn the proxy into a
/// general-purpose tunnel to anything the allow-list happens to name, which is a
/// different and much larger promise than "the agent can reach its API".
fn parse_target(target: &str) -> Result<(String, u16), String> {
    let (host, port) = target
        .rsplit_once(':')
        .ok_or_else(|| format!("CONNECT target {target:?} has no port"))?;
    let port: u16 = port
        .trim()
        .parse()
        .map_err(|_| format!("CONNECT target {target:?} has a non-numeric port"))?;
    if port != 443 {
        return Err(format!("port {port} is not permitted (only 443)"));
    }
    // Strip IPv6 brackets so the allow-list sees the same text either way.
    let host = host.trim().trim_start_matches('[').trim_end_matches(']');
    Ok((host.to_string(), port))
}

/// What happened to one connection. Returned so the caller can log it and the
/// selftest can assert on it.
#[derive(Debug, PartialEq, Eq)]
pub enum Verdict {
    Allowed(String),
    Denied(String),
    Malformed(String),
}

/// One header line, with a cap.
///
/// `read_line` has no limit, and a guest that never sends a newline would make
/// the host allocate until it died. Read byte-wise instead — the reads come out
/// of the BufReader, so this is cheap for lines this size, and it keeps ONE
/// reader over the connection, which matters (see `serve`).
async fn read_line_capped(reader: &mut BufReader<UnixStream>, cap: usize) -> Result<String, String> {
    let mut out = Vec::new();
    loop {
        match reader.read_u8().await {
            Ok(b'\n') => break,
            Ok(b) => out.push(b),
            // EOF mid-line: return what we have and let the caller judge it.
            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
            Err(e) => return Err(format!("read: {e}")),
        }
        if out.len() > cap {
            return Err(format!("a request line longer than {cap} bytes"));
        }
    }
    Ok(String::from_utf8_lossy(&out)
        .trim_end_matches('\r')
        .to_string())
}

/// Serve one tunnelled connection.
async fn serve(stream: UnixStream, allow: Arc<Vec<String>>) -> Verdict {
    // ONE reader for the whole request. Wrapping the stream a second time would
    // discard whatever the first reader had already buffered — including the
    // first bytes of the TLS handshake — and the tunnel would come up looking
    // fine and then stall on a corrupt stream.
    let mut reader = BufReader::new(stream);

    let line = match read_line_capped(&mut reader, 8 * 1024).await {
        Ok(l) if !l.trim().is_empty() => l,
        Ok(_) => return Verdict::Malformed("no request line".into()),
        Err(e) => return Verdict::Malformed(e),
    };

    let mut parts = line.split_whitespace();
    let method = parts.next().unwrap_or_default().to_ascii_uppercase();
    let target = parts.next().unwrap_or_default().to_string();

    if method != "CONNECT" {
        // Plain HTTP would mean proxying a request we would then have to rewrite,
        // and everything a mission needs is TLS. Refused with a status, so the
        // client reports something better than a closed socket.
        let _ = reply(reader.get_mut(), 405, "only CONNECT is supported").await;
        return Verdict::Malformed(format!("method {method}"));
    }

    let (host, port) = match parse_target(&target) {
        Ok(v) => v,
        Err(e) => {
            let _ = reply(reader.get_mut(), 400, &e).await;
            return Verdict::Malformed(e);
        }
    };

    if !host_allowed(&host, &allow) {
        // 403 rather than a silent drop: a denial that looks like a network
        // timeout is indistinguishable from a hung agent, and this codebase has
        // paid for that confusion more than once.
        let _ = reply(
            reader.get_mut(),
            403,
            &format!("{host} is not on the egress allow-list"),
        )
        .await;
        return Verdict::Denied(host);
    }

    // Consume the remaining request headers: they belong to the CONNECT, not to
    // the tunnel.
    loop {
        match read_line_capped(&mut reader, 8 * 1024).await {
            Ok(h) if h.trim().is_empty() => break,
            Ok(_) => {}
            Err(e) => return Verdict::Malformed(e),
        }
    }

    let mut upstream = match TcpStream::connect((host.as_str(), port)).await {
        Ok(s) => s,
        Err(e) => {
            let _ = reply(reader.get_mut(), 502, &format!("connect {host}:{port}: {e}")).await;
            return Verdict::Denied(host);
        }
    };

    if reply(reader.get_mut(), 200, "Connection established")
        .await
        .is_err()
    {
        return Verdict::Denied(host);
    }

    // Anything already buffered past the headers is tunnel payload — a client
    // that pipelined its first TLS bytes would otherwise lose them.
    let pending = reader.buffer().to_vec();
    let mut stream = reader.into_inner();
    if !pending.is_empty() && upstream.write_all(&pending).await.is_err() {
        return Verdict::Denied(host);
    }

    // Bytes both ways until either side is done. Errors are not worth reporting:
    // a closed connection is the normal end of a tunnel.
    let _ = tokio::io::copy_bidirectional(&mut stream, &mut upstream).await;
    Verdict::Allowed(host)
}

async fn reply(s: &mut UnixStream, code: u16, text: &str) -> std::io::Result<()> {
    let reason = if code == 200 {
        "Connection established"
    } else {
        "Forbidden"
    };
    // The body carries the reason for a non-200 so it reaches the agent's own
    // error output, where whoever is reading a failed mission will see it.
    let body = if code == 200 { String::new() } else { format!("{text}\n") };
    let head = format!(
        "HTTP/1.1 {code} {reason}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
        body.len()
    );
    s.write_all(head.as_bytes()).await?;
    if !body.is_empty() {
        s.write_all(body.as_bytes()).await?;
    }
    s.flush().await
}

/// Start this VM's proxy. Returns the socket path and the task serving it.
///
/// Bound **before** firecracker starts, because a guest that dials before the
/// host is listening gets a connection refused it will not retry.
pub fn start(uds: &Path, vm_id: &str) -> Result<(PathBuf, tokio::task::JoinHandle<()>), String> {
    let path = PathBuf::from(format!("{}_{}", uds.display(), EGRESS_PORT));
    // Firecracker does not clean these up any more than it cleans up its own
    // socket, and a stale file makes bind fail with EADDRINUSE.
    let _ = std::fs::remove_file(&path);
    let listener =
        UnixListener::bind(&path).map_err(|e| format!("bind {}: {e}", path.display()))?;

    let allow = Arc::new(allow_list());
    eprintln!(
        "microvm {vm_id}: egress proxy on {} allowing {:?}",
        path.display(),
        allow
    );
    let vm = vm_id.to_string();
    let task = tokio::spawn(async move {
        loop {
            match listener.accept().await {
                Ok((s, _)) => {
                    let allow = allow.clone();
                    let vm = vm.clone();
                    tokio::spawn(async move {
                        match serve(s, allow).await {
                            // Logged at every outcome: this is the audit trail of
                            // everything a mission reached, and a denial that is
                            // not logged is a mystery hang later.
                            Verdict::Allowed(h) => eprintln!("microvm {vm}: egress -> {h}"),
                            Verdict::Denied(h) => {
                                eprintln!("microvm {vm}: egress DENIED {h}")
                            }
                            Verdict::Malformed(w) => {
                                eprintln!("microvm {vm}: egress malformed request ({w})")
                            }
                        }
                    });
                }
                Err(e) => {
                    eprintln!("microvm {vm}: egress accept failed: {e}");
                    return;
                }
            }
        }
    });
    Ok((path, task))
}

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

    fn allow() -> Vec<String> {
        DEFAULT_ALLOW.iter().map(|s| s.to_string()).collect()
    }

    #[test]
    fn the_model_api_and_the_forge_are_reachable() {
        for h in ["api.anthropic.com", "git.redclaw.dev", "API.Anthropic.COM"] {
            assert!(host_allowed(h, &allow()), "{h} must be allowed");
        }
    }

    /// The check is a match, never a substring test. A name that merely CONTAINS
    /// an allowed one is a different host controlled by someone else.
    #[test]
    fn a_lookalike_host_is_not_allowed() {
        for h in [
            "api.anthropic.com.evil.test",
            "notapi.anthropic.com.attacker.io",
            // These contain an allowed suffix but are not that host. The first
            // PASSED before the character-class check was added — a unit test
            // found it, not review.
            "evil.test/api.anthropic.com",
            "evil.test@api.anthropic.com",
            "api.anthropic.com:443",
            "git.redclaw.dev.evil.test",
            "example.com",
            "",
            "   ",
        ] {
            assert!(!host_allowed(h, &allow()), "{h} must NOT be allowed");
        }
    }

    /// A raw address must not sidestep a list written in names.
    #[test]
    fn an_ip_literal_is_not_allowed() {
        let a = vec![".anthropic.com".to_string()];
        assert!(!host_allowed("[::1]", &a));
        assert!(!host_allowed("2606:4700::1111", &a));
    }

    /// A trailing dot is the same host to a resolver, so it must be to us.
    #[test]
    fn a_trailing_dot_does_not_bypass_the_list() {
        assert!(host_allowed("api.anthropic.com.", &allow()));
    }

    /// A `.domain` entry covers subdomains, and only real subdomains.
    #[test]
    fn a_dot_prefixed_entry_matches_subdomains_only() {
        let a = vec![".example.com".to_string()];
        assert!(host_allowed("a.example.com", &a));
        assert!(host_allowed("example.com", &a));
        assert!(!host_allowed("notexample.com", &a));
        assert!(!host_allowed("example.com.evil.test", &a));
    }

    /// Blanking the allow-list means no egress. Falling back to the default
    /// would hand a mission the network an operator had just taken away.
    #[test]
    fn an_empty_allow_list_denies_everything() {
        let none: Vec<String> = vec![];
        assert!(!host_allowed("api.anthropic.com", &none));
    }

    /// Only 443. Anything else turns the proxy into a general-purpose tunnel to
    /// whatever the allow-list happens to name.
    #[test]
    fn only_https_is_tunnelled() {
        assert_eq!(parse_target("api.anthropic.com:443").unwrap().1, 443);
        for bad in [
            "api.anthropic.com:22",
            "api.anthropic.com:80",
            "api.anthropic.com",
            "api.anthropic.com:not-a-port",
        ] {
            assert!(parse_target(bad).is_err(), "{bad} must be refused");
        }
    }
}