exfiltrate 0.3.0

// SPDX-License-Identifier: MIT OR Apache-2.0
//! The credential exchange, against a real server.
//!
//! The unit tests in `exfiltrate_internal::auth` cover the derivation and the
//! comparison. What they cannot cover is the ordering and the refusals — that
//! the challenge arrives before the `Hello`, that a wrong proof ends the
//! connection rather than merely being noted, and that a command sent without a
//! credential never reaches a command at all. Those are properties of the
//! server's dispatch, so this drives one.
//!
//! Its own file, and its own process, because the server is a singleton and this
//! one has a token.
#![cfg(unix)]

use exfiltrate_internal::auth;
use exfiltrate_internal::rpc::{AuthChallenge, AuthProof, CommandInvocation, RPC};
use exfiltrate_internal::transport::{Address, Stream};
use exfiltrate_internal::wire::{BACKOFF_DURATION, InFlightMessage, ReadStatus, send_socket_rpc};
use std::time::{Duration, Instant};

const TOKEN: &str = "TEST0-TOKEN";

fn socket_path() -> std::path::PathBuf {
    std::env::temp_dir().join(format!("exfiltrate-auth-{}.sock", std::process::id()))
}

/// Starts the one server this process gets, and returns its address.
fn server() -> Address {
    use std::sync::OnceLock;
    static ADDRESS: OnceLock<Address> = OnceLock::new();
    ADDRESS
        .get_or_init(|| {
            let path = socket_path();
            let _ = std::fs::remove_file(&path);
            let text = format!("unix:{}", path.display());
            let mut config = exfiltrate::Config::default()
                .with_addr(text.as_str())
                .with_token(TOKEN)
                .with_app(exfiltrate::app_info!());
            config.instance_registry = false;
            exfiltrate::begin_with(config);
            Address::parse(&text).unwrap()
        })
        .clone()
}

/// A fresh connection, retried until the listen thread is up.
fn dial() -> Stream {
    let address = server();
    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        match exfiltrate_internal::transport::connect(&address) {
            Ok(stream) => return stream,
            Err(error) => {
                assert!(Instant::now() < deadline, "could not connect: {error}");
                std::thread::sleep(BACKOFF_DURATION);
            }
        }
    }
}

/// Reads frames until `want` matches, or `None` if the peer hung up or stalled.
fn wait_for<T>(
    stream: &mut Stream,
    in_flight: &mut InFlightMessage,
    mut want: impl FnMut(RPC) -> Option<T>,
) -> Option<T> {
    let deadline = Instant::now() + Duration::from_secs(20);
    while Instant::now() < deadline {
        match in_flight.read_stream(stream) {
            Ok(ReadStatus::Completed(frame)) => {
                let rpc: RPC = rmp_serde::from_slice(&frame).expect("undecodable frame");
                if let Some(found) = want(rpc) {
                    return Some(found);
                }
            }
            Ok(ReadStatus::Progress) => {}
            Ok(_) => std::thread::sleep(BACKOFF_DURATION),
            // The peer closed, which is itself an answer in some of these tests.
            Err(_) => return None,
        }
    }
    None
}

/// Says hello and collects the challenge that must precede the server's own.
fn hello(stream: &mut Stream, in_flight: &mut InFlightMessage) -> AuthChallenge {
    send_socket_rpc(
        RPC::Hello(exfiltrate::Config::default().build_info()),
        stream,
    )
    .unwrap();
    let mut challenge = None;
    let saw_hello = wait_for(stream, in_flight, |rpc| match rpc {
        RPC::AuthChallenge(offered) => {
            challenge = Some(offered);
            None
        }
        RPC::Hello(_) => Some(()),
        _ => None,
    });
    assert!(
        saw_hello.is_some(),
        "the server never answered the handshake"
    );
    challenge.expect("the challenge must arrive before the server's own Hello")
}

#[test]
fn the_right_token_is_admitted_and_then_commands_work() {
    let mut stream = dial();
    let mut in_flight = InFlightMessage::new();
    let challenge = hello(&mut stream, &mut in_flight);

    let proof = auth::derive(TOKEN, &challenge).unwrap();
    send_socket_rpc(RPC::AuthProof(AuthProof { proof }), &mut stream).unwrap();
    let result = wait_for(&mut stream, &mut in_flight, |rpc| match rpc {
        RPC::AuthResult(result) => Some(result),
        _ => None,
    })
    .expect("no answer to the proof");
    assert!(result.ok, "{}", result.message);

    send_socket_rpc(
        RPC::Command(CommandInvocation::new("list".to_string(), Vec::new(), 1)),
        &mut stream,
    )
    .unwrap();
    let response = wait_for(&mut stream, &mut in_flight, |rpc| match rpc {
        RPC::CommandResponse(response) if response.reply_id == 1 => Some(response),
        _ => None,
    })
    .expect("an admitted connection must be served");
    assert!(response.success, "{}", response.response);
}

#[test]
fn a_wrong_token_is_refused_and_the_connection_ends() {
    let mut stream = dial();
    let mut in_flight = InFlightMessage::new();
    let challenge = hello(&mut stream, &mut in_flight);

    let proof = auth::derive("WRONG-TOKEN", &challenge).unwrap();
    send_socket_rpc(RPC::AuthProof(AuthProof { proof }), &mut stream).unwrap();
    let result = wait_for(&mut stream, &mut in_flight, |rpc| match rpc {
        RPC::AuthResult(result) => Some(result),
        _ => None,
    })
    .expect("no answer to the proof");
    assert!(!result.ok);
    assert!(
        result.message.contains(auth::TOKEN_ENV),
        "{}",
        result.message
    );

    // Closing is what makes every guess pay for a fresh challenge and another
    // derivation rather than retrying cheaply on the same socket.
    send_socket_rpc(
        RPC::Command(CommandInvocation::new("list".to_string(), Vec::new(), 2)),
        &mut stream,
    )
    .ok();
    let served = wait_for(&mut stream, &mut in_flight, |rpc| match rpc {
        RPC::CommandResponse(response) if response.success => Some(()),
        _ => None,
    });
    assert!(
        served.is_none(),
        "a refused connection kept serving commands"
    );
}

#[test]
fn a_command_without_a_credential_is_refused_in_a_way_a_person_can_act_on() {
    let mut stream = dial();
    let mut in_flight = InFlightMessage::new();

    // No `Hello`, no proof — which is exactly how a client older than this
    // protocol presents. It must get an answer rather than silence.
    send_socket_rpc(
        RPC::Command(CommandInvocation::new("list".to_string(), Vec::new(), 3)),
        &mut stream,
    )
    .unwrap();
    let response = wait_for(&mut stream, &mut in_flight, |rpc| match rpc {
        RPC::CommandResponse(response) if response.reply_id == 3 => Some(response),
        _ => None,
    })
    .expect("an unauthenticated command must be answered, not ignored");
    assert!(!response.success);
    let message = response.response.to_string();
    assert!(message.contains(auth::TOKEN_ENV), "{message}");
}

#[test]
fn each_connection_gets_its_own_challenge() {
    let mut first = dial();
    let mut second = dial();
    let first_challenge = hello(&mut first, &mut InFlightMessage::new());
    let second_challenge = hello(&mut second, &mut InFlightMessage::new());
    assert_ne!(
        first_challenge.nonce, second_challenge.nonce,
        "a proof captured from one connection must be worthless on the next"
    );
}