use assert_cmd::Command;
use predicates::prelude::*;
fn asphyxia() -> Command {
let mut cmd = Command::cargo_bin("asphyxia").expect("binary `asphyxia` should be built");
cmd.env("ASPHYXIA_CONFIG", "/nonexistent/asphyxia-test-config.toml");
cmd
}
#[test]
fn help_describes_the_tool() {
asphyxia()
.arg("--help")
.assert()
.success()
.stdout(predicate::str::contains("network scanner"));
}
#[test]
fn version_flag_prints_version() {
asphyxia()
.arg("--version")
.assert()
.success()
.stdout(predicate::str::contains(env!("CARGO_PKG_VERSION")));
}
#[test]
fn port_scan_requires_a_target() {
asphyxia()
.args(["ps", "-s", "80"])
.assert()
.failure()
.stderr(predicate::str::contains("--host"))
.stderr(predicate::str::contains("--stdin"));
}
#[test]
fn port_scan_host_and_stdin_are_mutually_exclusive() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "--stdin", "-s", "80"])
.assert()
.failure()
.stderr(predicate::str::contains("cannot be used with"));
}
#[test]
fn port_scan_all_ports_conflicts_with_specific() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "--all-ports", "-s", "80"])
.assert()
.failure()
.stderr(predicate::str::contains("cannot be used with"));
}
#[test]
fn port_scan_without_range_or_specific_prints_guidance() {
asphyxia()
.args(["ps", "-t", "127.0.0.1"])
.assert()
.success()
.stderr(predicate::str::contains("Please specify either -r, -s"));
}
#[test]
fn port_scan_top_ports_conflicts_with_specific() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "--top-ports", "100", "-s", "80"])
.assert()
.failure()
.stderr(predicate::str::contains("cannot be used with"));
}
#[test]
fn port_scan_top_ports_rejects_overflow() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "--top-ports", "5000"])
.assert()
.success()
.stderr(predicate::str::contains("at most 1000"));
}
#[test]
fn port_scan_named_set_rejects_unknown() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "--ports", "bogus"])
.assert()
.success()
.stderr(predicate::str::contains("Unknown port set"));
}
#[test]
fn port_scan_stdin_reads_plain_targets() {
asphyxia()
.args(["ps", "--stdin", "-s", "1", "-o", "json"])
.write_stdin("127.0.0.1\n")
.assert()
.success()
.stdout(predicate::eq("[]\n"));
}
#[test]
fn port_scan_stdin_reads_jsonl_from_address_scan() {
asphyxia()
.args(["ps", "--stdin", "-s", "1", "-o", "json"])
.write_stdin("{\"ip\":\"127.0.0.1\",\"proto\":\"tcp\",\"status\":\"up\"}\n")
.assert()
.success()
.stdout(predicate::eq("[]\n"));
}
#[test]
fn port_scan_stdin_with_no_targets_prints_guidance() {
asphyxia()
.args(["ps", "--stdin", "-s", "1"])
.write_stdin("")
.assert()
.success()
.stderr(predicate::str::contains("No targets read from stdin"));
}
#[test]
fn port_scan_rejects_invalid_ports() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "22,abc,443"])
.assert()
.success()
.stderr(predicate::str::contains("Invalid port number: abc"));
}
#[test]
fn port_scan_reports_unresolvable_host() {
asphyxia()
.args(["ps", "-t", "this-host-does-not-exist.invalid", "-s", "80"])
.assert()
.success()
.stderr(predicate::str::contains("Could not resolve host"));
}
#[test]
fn address_scan_without_args_prints_guidance() {
asphyxia()
.arg("as")
.assert()
.success()
.stderr(predicate::str::contains(
"Please specify either -s, -t, or -r",
));
}
#[test]
fn address_scan_rejects_invalid_ip() {
asphyxia()
.args(["as", "-t", "not-an-ip"])
.assert()
.success()
.stderr(predicate::str::contains("Invalid IP address: not-an-ip"));
}
#[test]
fn address_scan_rejects_invalid_subnet() {
asphyxia()
.args(["as", "-s", "192.168.1.0/33"])
.assert()
.success()
.stderr(predicate::str::contains("Invalid subnet format"));
}
#[test]
fn concurrency_flag_is_accepted() {
asphyxia()
.args(["as", "-s", "192.168.1.0/33", "--concurrency", "64"])
.assert()
.success()
.stderr(predicate::str::contains("Invalid subnet format"));
}
#[test]
fn retries_flag_is_accepted() {
asphyxia()
.args([
"ps",
"-t",
"127.0.0.1",
"-s",
"1",
"--retries",
"2",
"-o",
"json",
])
.assert()
.success()
.stdout(predicate::eq("[]\n"));
}
#[test]
fn retries_flag_rejects_non_numeric() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1", "--retries", "lots"])
.assert()
.failure();
}
#[test]
fn timing_profile_out_of_range_is_rejected() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1", "-T", "9"])
.assert()
.failure();
}
#[test]
fn timing_profile_is_accepted() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1", "-T", "4", "-o", "json"])
.assert()
.success()
.stdout(predicate::eq("[]\n"));
}
#[test]
fn rate_flag_is_accepted() {
asphyxia()
.args([
"ps",
"-t",
"127.0.0.1",
"-s",
"1",
"--rate",
"1000",
"-o",
"json",
])
.assert()
.success()
.stdout(predicate::eq("[]\n"));
}
#[test]
fn rate_limit_paces_multiple_probes() {
use std::time::Instant;
let start = Instant::now();
asphyxia()
.args([
"ps",
"-t",
"127.0.0.1",
"-s",
"1,2,3,4,5",
"--rate",
"20",
"-o",
"json",
])
.assert()
.success();
let elapsed = start.elapsed();
assert!(
elapsed >= std::time::Duration::from_millis(150),
"rate limit should pace the scan, took {:?}",
elapsed
);
}
#[test]
fn concurrency_flag_rejects_non_numeric() {
asphyxia()
.args(["as", "-s", "192.168.1.0/24", "--concurrency", "lots"])
.assert()
.failure();
}
#[test]
fn port_scan_json_with_no_open_ports_emits_empty_array() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1", "-o", "json"])
.assert()
.success()
.stdout(predicate::eq("[]\n"));
}
#[test]
fn port_scan_jsonl_with_no_open_ports_emits_nothing() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1", "-o", "jsonl"])
.assert()
.success()
.stdout(predicate::eq(""));
}
#[test]
fn port_scan_csv_emits_header_even_with_no_open_ports() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1", "-o", "csv"])
.assert()
.success()
.stdout(predicate::eq("ip,port,proto,scan,status,latency_ms\n"));
}
#[test]
fn port_scan_grep_with_no_open_ports_emits_nothing() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1", "-o", "grep"])
.assert()
.success()
.stdout(predicate::eq(""));
}
#[test]
fn target_file_conflicts_with_host() {
asphyxia()
.args([
"ps",
"-t",
"127.0.0.1",
"--target-file",
"targets.txt",
"-s",
"80",
])
.assert()
.failure()
.stderr(predicate::str::contains("cannot be used with"));
}
#[test]
fn target_file_reads_targets_from_disk() {
let dir = std::env::temp_dir();
let path = dir.join(format!("asphyxia-targets-{}.txt", std::process::id()));
std::fs::write(&path, "127.0.0.1\n127.0.0.2/31\n").unwrap();
asphyxia()
.args(["ps", "-i", path.to_str().unwrap(), "-s", "1", "-o", "json"])
.assert()
.success()
.stdout(predicate::eq("[]\n"));
let _ = std::fs::remove_file(&path);
}
#[test]
fn target_file_missing_reports_a_clear_error() {
asphyxia()
.args([
"ps",
"--target-file",
"/no/such/asphyxia-targets.txt",
"-s",
"1",
])
.assert()
.success()
.stderr(predicate::str::contains("Could not read target file"));
}
#[test]
fn config_supplies_defaults_that_flags_override() {
let dir = std::env::temp_dir();
let path = dir.join(format!("asphyxia-config-{}.toml", std::process::id()));
std::fs::write(&path, "output = \"json\"\ntimeout = 250\n").unwrap();
asphyxia()
.env("ASPHYXIA_CONFIG", &path)
.args(["ps", "-t", "127.0.0.1", "-s", "1"])
.assert()
.success()
.stdout(predicate::eq("[]\n"));
asphyxia()
.env("ASPHYXIA_CONFIG", &path)
.args(["ps", "-t", "127.0.0.1", "-s", "1", "-o", "csv"])
.assert()
.success()
.stdout(predicate::eq("ip,port,proto,scan,status,latency_ms\n"));
let _ = std::fs::remove_file(&path);
}
#[test]
fn service_detection_reports_service_and_banner() {
use std::io::Write;
use std::net::TcpListener;
use std::thread;
let listener = TcpListener::bind("127.0.0.1:0").expect("bind");
let port = listener.local_addr().unwrap().port();
let handle = thread::spawn(move || {
for _ in 0..2 {
match listener.accept() {
Ok((mut sock, _)) => {
let _ = sock.write_all(b"SSH-2.0-OpenSSH_9.6\r\n");
}
Err(_) => break,
}
}
});
asphyxia()
.args([
"ps",
"-t",
"127.0.0.1",
"-s",
&port.to_string(),
"--sV",
"-o",
"jsonl",
])
.assert()
.success()
.stdout(predicate::str::contains("\"service\":\"ssh\""))
.stdout(predicate::str::contains("SSH-2.0-OpenSSH_9.6"));
let _ = handle.join();
}
#[test]
fn syn_conflicts_with_udp() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1", "--syn", "--udp"])
.assert()
.failure()
.stderr(predicate::str::contains("cannot be used with"));
}
#[test]
fn syn_scan_runs_or_falls_back_cleanly() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1", "--syn", "-o", "json"])
.assert()
.success()
.stdout(predicate::eq("[]\n"));
}
#[test]
fn ack_scan_reports_firewall_state_and_never_open_or_closed() {
let assertion = asphyxia()
.args([
"ps",
"-t",
"127.0.0.1",
"-s",
"1",
"--scan",
"ack",
"--timeout",
"300",
"-o",
"jsonl",
])
.assert()
.success();
let stdout = String::from_utf8_lossy(&assertion.get_output().stdout).into_owned();
for line in stdout.lines() {
assert!(
line.contains("\"scan\":\"ack\""),
"an ACK result must record the probe that produced it: {line}"
);
assert!(
line.contains("\"status\":\"unfiltered\"") || line.contains("\"status\":\"filtered\""),
"an ACK result must be unfiltered or filtered: {line}"
);
}
assert!(
!stdout.contains("\"status\":\"open\"") && !stdout.contains("\"status\":\"closed\""),
"an ACK scan must not report open/closed: {stdout}"
);
}
#[test]
fn ack_scan_falls_back_visibly_for_an_ipv6_target() {
asphyxia()
.args([
"ps",
"-t",
"::1",
"-s",
"1",
"--scan",
"ack",
"--timeout",
"300",
"-o",
"json",
])
.assert()
.success()
.stderr(predicate::str::contains("falling back to connect scan"));
}
#[test]
fn ack_scan_rejects_a_udp_pairing() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1", "--scan", "ack", "--udp"])
.assert()
.failure()
.stderr(predicate::str::contains("cannot be used with"));
}
#[test]
fn udp_scan_reports_proto_udp_in_json() {
asphyxia()
.args([
"ps",
"-t",
"127.0.0.1",
"-s",
"9",
"--udp",
"--timeout",
"300",
"-o",
"jsonl",
])
.assert()
.success()
.stdout(predicate::str::contains("\"proto\":\"tcp\"").not());
}
#[test]
fn resume_writes_a_checkpoint_file() {
let dir = std::env::temp_dir();
let path = dir.join(format!("asphyxia-resume-out-{}.json", std::process::id()));
asphyxia()
.args([
"ps",
"-t",
"127.0.0.1",
"-s",
"1",
"--resume",
path.to_str().unwrap(),
"-o",
"json",
])
.assert()
.success();
let state = std::fs::read_to_string(&path).expect("checkpoint file should exist");
assert!(state.contains("\"job_count\":1"));
assert!(state.contains("\"done\":[true]"));
let _ = std::fs::remove_file(&path);
}
#[test]
fn resume_skips_completed_work_and_reuses_recorded_findings() {
let dir = std::env::temp_dir();
let path = dir.join(format!("asphyxia-resume-in-{}.json", std::process::id()));
let state = r#"{"proto":"tcp","targets":["127.0.0.1"],"ports":[1],"job_count":1,"done":[true],"findings":[{"host":0,"port":1,"latency_ms":7,"status":"open"}]}"#;
std::fs::write(&path, state).unwrap();
asphyxia()
.args([
"ps",
"-t",
"127.0.0.1",
"-s",
"1",
"--resume",
path.to_str().unwrap(),
"-o",
"jsonl",
])
.assert()
.success()
.stdout(predicate::str::contains("\"port\":1"))
.stdout(predicate::str::contains("\"status\":\"open\""));
let _ = std::fs::remove_file(&path);
}
#[test]
fn nmap_args_requires_the_nmap_flag() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "80", "--nmap-args", "-A"])
.assert()
.failure()
.stderr(predicate::str::contains("nmap"));
}
#[test]
fn nmap_flag_with_no_open_ports_does_not_invoke_nmap() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1", "--nmap"])
.assert()
.success()
.stdout(predicate::str::contains("No open ports found"));
}
#[test]
fn exclude_ports_removing_the_only_port_prints_guidance() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1", "--exclude-ports", "1"])
.assert()
.success()
.stderr(predicate::str::contains("No ports left to scan"));
}
#[test]
fn exclude_ports_rejects_invalid_list() {
asphyxia()
.args([
"ps",
"-t",
"127.0.0.1",
"-s",
"1,2",
"--exclude-ports",
"nope",
])
.assert()
.success()
.stderr(predicate::str::contains("Invalid port number: nope"));
}
#[test]
fn address_scan_exclude_rejects_invalid_spec() {
asphyxia()
.args(["as", "-s", "10.0.0.0/30", "--exclude", "not-an-ip"])
.assert()
.success()
.stderr(predicate::str::contains("Invalid exclude address"));
}
#[test]
fn address_scan_exclude_filters_hosts_from_a_pn_scan() {
asphyxia()
.args([
"as",
"-r",
"10.20.30.1",
"10.20.30.4",
"--Pn",
"--exclude",
"10.20.30.2",
"-o",
"grep",
])
.assert()
.success()
.stdout(predicate::str::contains("10.20.30.1"))
.stdout(predicate::str::contains("10.20.30.4"))
.stdout(predicate::str::contains("10.20.30.2").not());
}
#[test]
fn address_scan_exclude_file_is_read() {
let dir = std::env::temp_dir();
let path = dir.join(format!("asphyxia-exclude-{}.txt", std::process::id()));
std::fs::write(&path, "# comment\n10.20.31.3\n\n").unwrap();
asphyxia()
.args([
"as",
"-r",
"10.20.31.1",
"10.20.31.4",
"--Pn",
"--exclude-file",
path.to_str().unwrap(),
"-o",
"grep",
])
.assert()
.success()
.stdout(predicate::str::contains("10.20.31.3").not())
.stdout(predicate::str::contains("10.20.31.1"));
let _ = std::fs::remove_file(&path);
}
#[test]
fn output_file_writes_machine_output_and_keeps_stdout_clean() {
let dir = std::env::temp_dir();
let path = dir.join(format!("asphyxia-cli-out-{}.csv", std::process::id()));
let path_str = path.to_str().unwrap();
asphyxia()
.args([
"ps",
"-t",
"127.0.0.1",
"-s",
"1",
"-o",
"csv",
"--output-file",
path_str,
])
.assert()
.success()
.stdout(predicate::eq(""));
let contents = std::fs::read_to_string(&path).expect("output file should exist");
assert!(contents.starts_with("ip,port,proto,scan,status,latency_ms\n"));
let _ = std::fs::remove_file(&path);
}
#[test]
fn address_scan_json_with_no_hosts_emits_empty_array() {
asphyxia()
.args([
"as",
"-t",
"192.168.255.255",
"--timeout",
"200",
"-o",
"json",
])
.assert()
.success()
.stdout(predicate::eq("[]\n"));
}
#[test]
fn address_scan_pn_marks_target_up_without_probing() {
asphyxia()
.args(["as", "-t", "192.168.255.255", "--Pn", "-o", "jsonl"])
.assert()
.success()
.stdout(predicate::str::contains("\"ip\":\"192.168.255.255\""))
.stdout(predicate::str::contains("\"status\":\"up\""));
}
#[test]
fn address_scan_pn_marks_whole_range_up() {
asphyxia()
.args(["as", "-r", "10.10.10.1", "10.10.10.4", "--Pn", "-o", "json"])
.assert()
.success()
.stdout(predicate::str::contains("10.10.10.1"))
.stdout(predicate::str::contains("10.10.10.4"));
}
#[test]
fn output_flag_rejects_unknown_format() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1", "-o", "bogus"])
.assert()
.failure();
}
#[test]
fn text_output_remains_the_default() {
asphyxia()
.args(["ps", "-t", "127.0.0.1", "-s", "1"])
.assert()
.success()
.stdout(predicate::str::contains("Game Over"));
}
#[test]
fn interface_flag_is_documented_in_help() {
asphyxia()
.args(["ps", "--help"])
.assert()
.success()
.stdout(predicate::str::contains("--interface"));
asphyxia()
.args(["as", "--help"])
.assert()
.success()
.stdout(predicate::str::contains("--interface"));
}
#[test]
fn unknown_interface_fails_before_scanning() {
asphyxia()
.args([
"ps",
"-t",
"127.0.0.1",
"-s",
"80",
"-e",
"definitely-not-an-iface-9999",
])
.assert()
.failure()
.stderr(predicate::str::contains("interface"));
}
#[test]
fn config_interface_is_validated_too() {
let dir = std::env::temp_dir();
let path = dir.join(format!("asphyxia-iface-config-{}.toml", std::process::id()));
std::fs::write(&path, "interface = \"definitely-not-an-iface-9999\"\n").unwrap();
asphyxia()
.env("ASPHYXIA_CONFIG", &path)
.args(["ps", "-t", "127.0.0.1", "-s", "80"])
.assert()
.failure()
.stderr(predicate::str::contains("interface"));
let _ = std::fs::remove_file(&path);
}