liminal-server 0.28.2

use crate::agent_pass;
pub use agent_pass::base64url as encoded;

use super::{TestError, TestResult};
use liminal_server::config::types::AuthConfig;
use lys_core::keys::Ed25519Identity;
use std::collections::BTreeMap;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use zeroize::Zeroizing;
type Request = (String, BTreeMap<String, String>, Vec<u8>);
const ISSUER: &str = "https://federation.example";
#[path = "../support/agent_issuer.rs"]
pub mod agent_issuer;
type Importer = (
    Arc<liminal_server::link::LinkHost>,
    Arc<dyn liminal_server::auth_lys::session::LysAdmitter>,
);
fn token(
    key: &Ed25519Identity,
    holder: &str,
    expiry: u64,
    issued_at: u64,
) -> Result<String, TestError> {
    let rights = [serde_json::json!({
        "resource":{"kind":"liminal.link","id":"orders-link"},
        "actions":["link_import","link_export"],"mode":"outright","grant":"machine-grant"
    })];
    let signing = agent_pass::Issuer {
        url: ISSUER,
        key_id: "federation",
        identity: key,
    };
    // NET-019: vary exp for the first dial pass to exercise renewal before expiry.
    let token = agent_pass::mint(
        agent_pass::Options {
            holder,
            expires_at: expiry,
            issued_at,
        },
        &rights,
        &signing,
    )
    .map_err(|error| error.to_string())?;
    Ok(token.to_string())
}
fn read_request(reader: &mut BufReader<TcpStream>) -> Result<Request, TestError> {
    let mut request = String::new();
    reader.read_line(&mut request)?;
    let mut headers = BTreeMap::new();
    loop {
        let mut line = String::new();
        reader.read_line(&mut line)?;
        if line == "\r\n" {
            break;
        }
        let (name, value) = line.split_once(':').ok_or("malformed request header")?;
        headers.insert(name.to_ascii_lowercase(), value.trim().to_owned());
    }
    let mut body = vec![
        0;
        headers
            .get("content-length")
            .map_or(Ok(0), |n| n.parse::<usize>())?
    ];
    reader.read_exact(&mut body)?;
    Ok((request, headers, body))
}

fn reply(mut stream: TcpStream, status: u16, reason: &str, answer: &str) -> TestResult {
    write!(
        stream,
        "HTTP/1.1 {status} {reason}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{answer}",
        answer.len()
    )?;
    Ok(())
}

struct Issuer {
    listener: TcpListener,
    prefix: &'static str,
    machine: Ed25519Identity,
    count: Arc<AtomicUsize>,
    stop: Arc<AtomicBool>,
    used: Arc<AtomicUsize>,
    issuing: Arc<AtomicBool>,
    issued_answer: serde_json::Value,
    access_token: String,
    peer_pass: String,
    answer: serde_json::Value,
    renewed: serde_json::Value,
    status: u16,
    disconnect: bool,
    /// How many of the first pass asks answer 503 with no JSON body, as an
    /// unbound route behind a tunnel does (ACCESS-013 R5).
    unavailable: usize,
    /// Holds the first pass ask's answer until the test releases it.
    release: Option<std::sync::mpsc::Receiver<()>>,
}
/// The issuer's answers to a pass ask, chosen by what was asked before.
struct Answers<'a> {
    answer: &'a serde_json::Value,
    renewed: &'a serde_json::Value,
    issued_answer: &'a serde_json::Value,
    /// How many first asks answer 503 before any other answer.
    unavailable: usize,
}
impl Answers<'_> {
    /// A fixture started refusing issues once `issuing` is set: its first
    /// issue is the access token, later ones the renewed pass, as a fixture
    /// that always issued answers them (ACCESS-009 R2).
    fn pick(&self, status: u16, posted: usize, issuing: bool, issued: &mut usize) -> (u16, String) {
        if posted < self.unavailable {
            (503, "error code: 1033".to_owned())
        } else if status != 200 && issuing {
            let first = *issued == 0;
            *issued += 1;
            (
                200,
                if first {
                    self.issued_answer
                } else {
                    self.renewed
                }
                .to_string(),
            )
        } else if posted == 0 || status != 200 {
            (status, self.answer.to_string())
        } else {
            (status, self.renewed.to_string())
        }
    }
}
/// The fake Lys's clock, in milliseconds.
fn now_ms() -> Result<u64, TestError> {
    Ok(u64::try_from(
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)?
            .as_millis(),
    )?)
}
/// Holds the first pass ask until the test releases it, when it holds one.
fn held(posted: usize, release: Option<&std::sync::mpsc::Receiver<()>>) -> TestResult {
    if posted == 0
        && let Some(release) = release
    {
        release.recv()?;
    }
    Ok(())
}
impl Issuer {
    fn serve(self) -> TestResult {
        let Self {
            listener,
            prefix,
            machine,
            count,
            stop,
            used,
            issuing,
            issued_answer,
            access_token,
            peer_pass,
            answer,
            renewed,
            status,
            disconnect,
            unavailable,
            release,
        } = self;
        // Issues answered after a refusing fixture was switched to issue.
        let mut issued = 0_usize;

        let agents = format!("POST {prefix}/agents/");
        let mut nonces = BTreeMap::new();
        loop {
            let (stream, _) = listener.accept()?;
            if stop.load(Ordering::Acquire) {
                return Ok(());
            }
            if disconnect {
                continue;
            }
            let mut reader = BufReader::new(stream);
            let (request, headers, body) = read_request(&mut reader)?;
            let path = request
                .split_ascii_whitespace()
                .nth(1)
                .ok_or("request path absent")?;
            let (status, answer) =
                if !request.starts_with(&agents) {
                    (
                        404,
                        serde_json::json!({"refusal":"agent_dial_refused"}).to_string(),
                    )
                } else if path == format!("{prefix}/agents/seat/usage") {
                    let current = if count.load(Ordering::Acquire) == 1 {
                        &access_token
                    } else {
                        &peer_pass
                    };
                    assert!(
                        headers
                            .get("authorization")
                            .is_some_and(|value| value == &format!("Bearer {current}")),
                        "services presented another pass"
                    );
                    assert_eq!(
                        serde_json::from_slice::<serde_json::Value>(&body)?["event"],
                        "reading"
                    );
                    used.fetch_add(1, Ordering::Release);
                    (200, "{}".into())
                } else {
                    assert_eq!(path, format!("{prefix}{}", agent_issuer::PATH));
                    assert!(
                        !headers.keys().any(|name| name.starts_with("x-lys-machine-")
                            || name == "x-lys-server-epoch")
                    );
                    assert_eq!(
                        serde_json::from_slice::<serde_json::Value>(&body)?,
                        serde_json::json!({"audience":"liminal"})
                    );
                    let now = now_ms()?;
                    let checked = agent_issuer::verify(
                        &machine.public_key_bytes(),
                        headers.get("lys-agent-signature"),
                        path,
                        &body,
                        &mut nonces,
                        now,
                    );
                    let posted = count.fetch_add(1, Ordering::Release);
                    held(posted, release.as_ref())?;
                    if let Err(reason) = checked {
                        let answer =
                            serde_json::json!({"refusal":"AgentSignatureRefused", "reason":reason})
                                .to_string();
                        reply(reader.into_inner(), 401, "Refused", &answer)?;
                        continue;
                    }
                    let answers = Answers {
                        answer: &answer,
                        renewed: &renewed,
                        issued_answer: &issued_answer,
                        unavailable,
                    };
                    answers.pick(status, posted, issuing.load(Ordering::Acquire), &mut issued)
                };
            reply(reader.into_inner(), status, "Answer", &answer)?;
        }
    }
}

pub struct Fixture {
    pub(super) root: tempfile::TempDir,
    pub(super) address: std::net::SocketAddr,
    pub(super) prefix: &'static str,
    pub(super) posts: Arc<AtomicUsize>,
    pub(super) usages: Arc<AtomicUsize>,
    pub(super) stopped: Arc<AtomicBool>,
    /// Lys issues when set; a fixture started with a refusal starts unset
    /// and [`Fixture::issue`] flips it while the server runs (ACCESS-009 R2).
    issuing: Arc<AtomicBool>,
    pub(super) thread: Option<std::thread::JoinHandle<TestResult>>,
    pub(super) expiry: u64,
    pub(super) peer_pass: String,
    pub(super) first_pass: String,
}
impl Fixture {
    pub(super) fn start(refusal: Option<&str>) -> Result<Self, TestError> {
        Self::start_at(refusal, "")
    }
    pub(super) fn start_at(refusal: Option<&str>, prefix: &'static str) -> Result<Self, TestError> {
        Self::start_with(refusal, prefix, None, 0)
    }
    /// A Lys that issues, holding its first pass answer until the returned
    /// sender releases it (SRV-041 R4).
    pub(super) fn start_held() -> Result<(Self, std::sync::mpsc::Sender<()>), TestError> {
        let (release, held) = std::sync::mpsc::channel();
        Ok((Self::start_with(None, "", Some(held), 0)?, release))
    }
    /// A Lys whose first pass is stamped `ahead` seconds from now, as one
    /// built that long after the ask was sent, holding its answer until the
    /// returned sender releases it (ACCESS-013 R6).
    pub(super) fn start_late(ahead: u64) -> Result<(Self, std::sync::mpsc::Sender<()>), TestError> {
        let (release, held) = std::sync::mpsc::channel();
        let issued_at = now_ms()? / 1000 + ahead;
        Ok((Self::start_with(None, "", Some(held), issued_at)?, release))
    }
    fn start_with(
        refusal: Option<&str>,
        prefix: &'static str,
        release: Option<std::sync::mpsc::Receiver<()>>,
        issued_at: u64,
    ) -> Result<Self, TestError> {
        let root = tempfile::tempdir()?;
        let signer = Ed25519Identity::from_seed(&Zeroizing::new([9; 32]));
        let machine = Ed25519Identity::from_seed(&Zeroizing::new([1; 32]));
        std::fs::write(root.path().join("identity"), hex::encode([1; 32]))?;
        let keys = serde_json::json!({"keys":[{"kty":"OKP","crv":"Ed25519","kid":"federation",
            "x":encoded(&signer.public_key_bytes()),"alg":"EdDSA","use":"sig"}]});
        std::fs::write(root.path().join("keys"), serde_json::to_vec(&keys)?)?;
        let expiry = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)?
            .as_secs()
            + 3600;
        let access_token = token(&signer, agent_issuer::AGENT, expiry, issued_at)?;
        let first_pass = access_token.clone();
        let peer_pass = token(&signer, agent_issuer::AGENT, u64::MAX, 0)?;
        let renewed =
            serde_json::json!({"pass":peer_pass,"audience":"liminal","expires_at":u64::MAX});
        let issued_answer =
            serde_json::json!({"pass":access_token,"audience":"liminal","expires_at":expiry});
        let unavailable = match refusal {
            Some("unavailable_once") => 1,
            Some("unavailable") => usize::MAX,
            _ => 0,
        };
        let refusal = refusal.filter(|_| unavailable == 0);
        let answer = refusal.map_or_else(
            || issued_answer.clone(),
            |name| serde_json::json!({"error":name}),
        );
        let status = if refusal.is_some() { 403 } else { 200 };
        let issuing = Arc::new(AtomicBool::new(refusal.is_none()));
        let disconnect = refusal == Some("disconnect");
        let listener = TcpListener::bind("127.0.0.1:0")?;
        let address = listener.local_addr()?;
        let posts = Arc::new(AtomicUsize::new(0));
        let stopped = Arc::new(AtomicBool::new(false));
        let usages = Arc::new(AtomicUsize::new(0));
        let used = Arc::clone(&usages);
        let count = Arc::clone(&posts);
        let stop = Arc::clone(&stopped);
        let retained_peer = peer_pass.clone();
        let issuer = Issuer {
            listener,
            prefix,
            machine,
            count,
            stop,
            used,
            issuing: Arc::clone(&issuing),
            issued_answer,
            access_token,
            peer_pass,
            answer,
            renewed,
            status,
            disconnect,
            unavailable,
            release,
        };
        let thread = std::thread::spawn(move || issuer.serve());
        Ok(Self {
            root,
            address,
            prefix,
            posts,
            usages,
            stopped,
            issuing,
            thread: Some(thread),
            expiry,
            peer_pass: retained_peer,
            first_pass,
        })
    }
    pub(super) fn importer(&self) -> Result<Importer, TestError> {
        use liminal_server::auth_lys::{
            LysPassConsumer,
            lys_pass_adapter::{LysPassAdapter, MachineLinkRights},
            session::ConsumerAdmitter,
        };
        use liminal_server::link_handshake::{LocalMachine, MachineIdentity};
        let keys = std::fs::read_to_string(self.root.path().join("keys"))?;
        let adapter = Arc::new(
            LysPassAdapter::from_keys_json(&keys, ISSUER.into())?.with_live(Arc::new(OpenAsk)),
        );
        let token = token(
            &Ed25519Identity::from_seed(&Zeroizing::new([9; 32])),
            "importer",
            u64::MAX,
            0,
        )?;
        let rights = MachineLinkRights::new(
            &adapter,
            adapter.verify_offline(token.as_bytes(), "liminal", self.expiry)?,
        );
        let local = LocalMachine::new(Ed25519Identity::from_seed(&Zeroizing::new([2; 32])));
        let exporter = MachineIdentity::new(
            Ed25519Identity::from_seed(&Zeroizing::new([1; 32])).public_key_bytes(),
        );
        Ok((
            Arc::new(liminal_server::link::LinkHost::new(
                local,
                Box::new(rights),
                BTreeMap::from([(agent_issuer::AGENT.into(), exporter)]),
            )),
            Arc::new(ConsumerAdmitter::new(adapter)),
        ))
    }
    pub(super) fn auth(&self, extra: &str) -> Result<AuthConfig, TestError> {
        let keys = toml_path(&self.root.path().join("keys"))?;
        let identity = toml_path(&self.root.path().join("identity"))?;
        Ok(toml::from_str(&format!(
            "token = 'fixture'\n[lys]\nissuer = '{ISSUER}'\nnetwork_id = 'network'\nkeys_path = {keys}\napi_url = 'http://{}{}/'\n[lys.machine]\nidentity_path = {identity}\nagent_id = '{}'\naudience = 'liminal'\n{extra}",
            self.address,
            self.prefix,
            agent_issuer::AGENT,
        ))?)
    }
    /// Lys begins issuing: the next ask is answered with the access token.
    pub(super) fn issue(&self) {
        self.issuing.store(true, Ordering::Release);
    }
    pub(super) fn finish(mut self) -> TestResult {
        self.stopped.store(true, Ordering::Release);
        TcpStream::connect(self.address)?;
        self.thread
            .take()
            .ok_or("fixture thread missing")?
            .join()
            .map_err(|_| "fixture verification panicked")?
    }
}
impl Drop for Fixture {
    fn drop(&mut self) {
        if let Some(thread) = self.thread.take() {
            self.stopped.store(true, Ordering::Release);
            if let Err(error) = TcpStream::connect(self.address) {
                eprintln!("fixture shutdown: {error}");
            }
            match thread.join() {
                Ok(Ok(())) => {}
                result => eprintln!("fixture verification failed: {result:?}"),
            }
        }
    }
}

#[derive(Debug)]
struct OpenAsk;
impl liminal_server::auth_lys::lys_pass_adapter::LiveLysAsk for OpenAsk {
    fn ask(
        &self,
        _: &str,
        holder: &str,
        target: &lys_pass::Target,
        _: u64,
    ) -> Result<liminal_server::auth_lys::lys_pass_adapter::LiveAnswer, lys_pass::Error> {
        use liminal_server::auth_lys::lys_pass_adapter::LiveAnswer;
        if holder != agent_issuer::AGENT
            || target.kind != "liminal.link"
            || target.id != "orders-link"
            || target.action != "link_export"
        {
            return Ok(LiveAnswer::Refused {
                refusal: "fixture_open_refused".into(),
            });
        }
        Ok(LiveAnswer::Outright {
            grant: "grant-link".into(),
        })
    }
}

/// Encodes a fixture path as a quoted TOML string.
///
/// # Errors
/// The fixture path is not valid UTF-8.
pub fn toml_path(path: &std::path::Path) -> Result<toml::Value, TestError> {
    Ok(toml::Value::String(
        path.to_str().ok_or("fixture path is not UTF-8")?.to_owned(),
    ))
}