use anyhow::Context;
use lanflare_core::config::AppConfig;
use lanflare_core::crypto::cert::TlsIdentity;
use lanflare_core::discovery::interface::{local_interfaces, InterfaceFilter};
use lanflare_core::discovery::mdns::{start_mdns_engine, MdnsConfig, MdnsDiscoveryEvent, MdnsNodeInfo};
use lanflare_core::discovery::multicast::{
start_multicast_engine, MulticastConfig, MulticastDiscoveryEvent, MulticastNodeInfo,
};
use lanflare_core::discovery::registry::{DiscoveredPeer, PeerRegistry};
use lanflare_proto::discovery::{DeviceType, ProtocolType};
use std::path::PathBuf;
use std::time::Duration;
use tokio::sync::{mpsc, oneshot};
use crate::tui::render_peer_table;
pub async fn status(config: &AppConfig, identity: &TlsIdentity) -> anyhow::Result<()> {
println!("LanFlare Node Status:");
println!(" Alias: {}", config.alias);
println!(" Port: {}", config.port);
println!(" Download Dir: {}", config.download_dir.display());
println!(" Auto Accept: {}", config.auto_accept);
println!(
" PIN Protect: {}",
if config.pin.is_some() { "Enabled" } else { "Disabled" }
);
println!(" Fingerprint: {}", identity.fingerprint);
println!();
println!("Local Network Interfaces:");
let filter = InterfaceFilter::default();
match local_interfaces(&filter) {
Ok(interfaces) => {
if interfaces.v4.is_empty() && interfaces.v6.is_empty() {
println!(" (No active non-loopback network interfaces found)");
} else {
for iface in &interfaces.v4 {
println!(" - IPv4: {} ({})", iface.address, iface.name);
}
for iface in &interfaces.v6 {
println!(" - IPv6: index {} ({})", iface.index, iface.name);
}
}
}
Err(err) => {
println!(" Failed to enumerate network interfaces: {err}");
}
}
Ok(())
}
#[derive(Clone, Debug)]
pub struct ScanOptions {
pub duration_secs: u64,
pub multicast: bool,
pub mdns: bool,
pub subnet: bool,
pub fallback_subnet: bool,
}
impl Default for ScanOptions {
fn default() -> Self {
Self {
duration_secs: 3,
multicast: true,
mdns: true,
subnet: false,
fallback_subnet: true,
}
}
}
pub async fn scan(
config: &AppConfig,
identity: &TlsIdentity,
options: &ScanOptions,
) -> anyhow::Result<Vec<DiscoveredPeer>> {
println!(
"Scanning local network for LanFlare / LocalSend peers (timeout: {}s)...",
options.duration_secs
);
let registry = PeerRegistry::new();
let (mc_stop_tx, mc_stop_rx) = oneshot::channel();
let (mc_event_tx, mut mc_event_rx) = mpsc::channel::<MulticastDiscoveryEvent>(100);
let mc_handle = if options.multicast {
let mc_config = MulticastConfig {
port: config.port,
node_info: MulticastNodeInfo {
alias: config.alias.clone(),
version: "2.0".to_string(),
device_model: Some("LanFlare Node".to_string()),
device_type: Some(DeviceType::Desktop),
fingerprint: identity.fingerprint.clone(),
port: config.port,
protocol: ProtocolType::Https,
download: false,
},
..MulticastConfig::default()
};
match start_multicast_engine(mc_config, mc_event_tx, mc_stop_rx).await {
Ok(handle) => {
let h_clone = handle.clone();
tokio::spawn(async move {
h_clone.announce_burst().await;
});
Some(handle)
}
Err(e) => {
tracing::warn!("Failed to start multicast engine: {e}");
None
}
}
} else {
None
};
let (mdns_stop_tx, mdns_stop_rx) = oneshot::channel();
let (mdns_event_tx, mut mdns_event_rx) = mpsc::channel::<MdnsDiscoveryEvent>(100);
let mdns_handle = if options.mdns {
let mdns_config = MdnsConfig {
node_info: MdnsNodeInfo {
alias: config.alias.clone(),
instance_name: String::new(),
version: "2.0".to_string(),
device_model: Some("LanFlare Node".to_string()),
device_type: Some(DeviceType::Desktop),
fingerprint: identity.fingerprint.clone(),
port: config.port,
protocol: ProtocolType::Https,
download: false,
},
enable_responder: true,
enable_browser: true,
..MdnsConfig::default()
};
match start_mdns_engine(mdns_config, mdns_event_tx, mdns_stop_rx).await {
Ok(handle) => Some(handle),
Err(e) => {
tracing::warn!("Failed to start mDNS engine: {e}");
None
}
}
} else {
None
};
let registry_clone = registry.clone();
let collector_task = tokio::spawn(async move {
loop {
tokio::select! {
Some(event) = mc_event_rx.recv() => {
if let MulticastDiscoveryEvent::PeerDiscovered { remote_ip, scope_id, message } = event {
registry_clone.handle_multicast(remote_ip, scope_id, &message).await;
}
}
Some(event) = mdns_event_rx.recv() => {
match event {
MdnsDiscoveryEvent::PeerDiscovered(peer) => {
registry_clone.handle_mdns(&peer).await;
}
MdnsDiscoveryEvent::PeerRemoved { fullname, .. } => {
tracing::debug!("mDNS service removed: {fullname}");
}
}
}
else => break,
}
}
});
let registry_subnet = registry.clone();
let identity_subnet = identity.clone();
let port_subnet = config.port;
let should_run_subnet = options.subnet || options.fallback_subnet;
let immediate_subnet = options.subnet;
let subnet_task = if should_run_subnet {
Some(tokio::spawn(async move {
if !immediate_subnet {
tokio::time::sleep(Duration::from_millis(500)).await;
if !registry_subnet.get_all().await.is_empty() {
return;
}
}
let filter = InterfaceFilter::default();
if let Ok(interfaces) = local_interfaces(&filter) {
let valid_ifaces: Vec<_> = interfaces
.v4
.into_iter()
.filter(|iface| is_sweepable_interface(&iface.name, &iface.address))
.collect();
if !valid_ifaces.is_empty() {
let descs: Vec<String> = valid_ifaces
.iter()
.map(|i| format!("{}/24 via {}", i.address, i.name))
.collect();
println!("Probing subnets in parallel: {}...", descs.join(", "));
let scan_tasks = valid_ifaces.into_iter().map(|iface| {
let reg = registry_subnet.clone();
let ident = identity_subnet.clone();
async move {
lanflare_core::discovery::scanner::scan_subnet(
iface.address,
port_subnet,
&ident,
®,
)
.await
}
});
futures_util::future::join_all(scan_tasks).await;
}
}
}))
} else {
None
};
let start = std::time::Instant::now();
let scan_timeout = Duration::from_secs(options.duration_secs);
loop {
tokio::time::sleep(Duration::from_millis(100)).await;
if !registry.get_all().await.is_empty() {
tokio::time::sleep(Duration::from_millis(250)).await;
break;
}
if let Some(ref st) = subnet_task {
if st.is_finished() && (immediate_subnet || start.elapsed() >= scan_timeout) {
break;
}
} else if start.elapsed() >= scan_timeout {
break;
}
if start.elapsed() >= scan_timeout.max(Duration::from_secs(5)) {
break;
}
}
if let Some(st) = subnet_task {
let _ = st.await;
}
let _ = mc_stop_tx.send(());
let _ = mdns_stop_tx.send(());
if let Some(h) = mc_handle {
h.stop().await;
}
if let Some(h) = mdns_handle {
h.stop().await;
}
collector_task.abort();
let peers = registry.get_all().await;
print!("{}", render_peer_table(&peers));
Ok(peers)
}
fn is_sweepable_interface(name: &str, ip: &std::net::Ipv4Addr) -> bool {
if ip.is_loopback() {
return false;
}
let octets = ip.octets();
if octets[0] == 100 && (octets[1] & 0xC0) == 64 {
return false;
}
let lower = name.to_ascii_lowercase();
if lower.starts_with("tailscale")
|| lower.starts_with("docker")
|| lower.starts_with("vboxnet")
|| lower.starts_with("veth")
|| lower.starts_with("br-")
{
return false;
}
true
}
use std::io::Write;
use std::net::IpAddr;
use lanflare_core::client::transfer::{
OutgoingFile, TransferOptions, TransferOrchestrator, TransferProgress,
};
use lanflare_core::client::{LsClient, LsClientConfig};
use lanflare_core::server::{start_server, PrepareUploadDecision, ServerTransferEvent};
use lanflare_proto::discovery::RegisterDtoV2;
use crate::tui::{shorten_fingerprint, TransferProgressTracker};
pub fn prompt_user_input(prompt: &str) -> anyhow::Result<String> {
print!("{prompt}");
std::io::stdout().flush().context("Failed to flush stdout")?;
let mut input = String::new();
std::io::stdin()
.read_line(&mut input)
.context("Failed to read user input from stdin")?;
Ok(input.trim().to_string())
}
pub fn match_target_peer<'a>(
query: &str,
peers: &'a [DiscoveredPeer],
) -> Option<&'a DiscoveredPeer> {
let q = query.trim();
if q.is_empty() {
return None;
}
let q_upper = q.to_ascii_uppercase();
if let Some(p) = peers.iter().find(|p| p.fingerprint == q_upper || p.fingerprint.starts_with(&q_upper)) {
return Some(p);
}
if let Ok(ip) = q.parse::<IpAddr>()
&& let Some(p) = peers.iter().find(|p| p.endpoints.iter().any(|ep| ep.ip == ip))
{
return Some(p);
}
if let Some(p) = peers.iter().find(|p| p.alias.eq_ignore_ascii_case(q)) {
return Some(p);
}
peers.iter().find(|p| p.alias.to_ascii_lowercase().contains(&q.to_ascii_lowercase()))
}
pub async fn discover_or_select_target(
config: &AppConfig,
identity: &TlsIdentity,
query: Option<&str>,
) -> anyhow::Result<DiscoveredPeer> {
if let Some(q) = query {
let trimmed = q.trim();
if let Ok(ip) = trimmed.parse::<IpAddr>() {
println!("Probing direct IP address {ip}...");
let probe_config = LsClientConfig {
timeout: Duration::from_secs(3),
connect_timeout: Duration::from_secs(2),
tls_identity: Some(identity.clone()),
target_fingerprint: None,
};
if let Ok(probe_client) = LsClient::new(probe_config) {
for proto in [ProtocolType::Https, ProtocolType::Http] {
let proto_str = match proto {
ProtocolType::Https => "https",
ProtocolType::Http => "http",
};
let base_url = match ip {
IpAddr::V4(v4) => format!("{proto_str}://{v4}:{}", config.port),
IpAddr::V6(v6) => format!("{proto_str}://[{v6}]:{}", config.port),
};
if let Ok(info) = probe_client.get_info(&base_url).await {
let peer_port = info.port.unwrap_or(config.port);
let peer_proto = info.protocol.unwrap_or(proto);
let ep = lanflare_core::discovery::registry::PeerEndpoint::new(
ip,
None,
lanflare_core::discovery::DiscoverySource::ManualScan,
0,
);
return Ok(DiscoveredPeer {
fingerprint: info.fingerprint,
alias: info.alias,
version: info.version,
device_model: info.device_model,
device_type: info.device_type,
port: peer_port,
protocol: peer_proto,
download: info.download,
endpoints: vec![ep],
discovery_sources: std::collections::HashSet::new(),
first_seen_ms: 0,
last_seen_ms: 0,
metadata: std::collections::HashMap::new(),
});
}
}
}
let ep = lanflare_core::discovery::registry::PeerEndpoint::new(
ip,
None,
lanflare_core::discovery::DiscoverySource::ManualScan,
0,
);
return Ok(DiscoveredPeer {
fingerprint: String::new(),
alias: format!("Direct Peer ({ip})"),
version: "2.0".to_string(),
device_model: None,
device_type: None,
port: config.port,
protocol: ProtocolType::Https,
download: false,
endpoints: vec![ep],
discovery_sources: std::collections::HashSet::new(),
first_seen_ms: 0,
last_seen_ms: 0,
metadata: std::collections::HashMap::new(),
});
}
}
println!("Scanning network to find target peer...");
let scan_opts = ScanOptions {
duration_secs: 2,
multicast: true,
mdns: true,
subnet: false,
fallback_subnet: true,
};
let peers = scan(config, identity, &scan_opts).await?;
if peers.is_empty() {
if let Some(q) = query {
anyhow::bail!("No peers found matching query '{q}'. Use explicit IP if known.");
} else {
anyhow::bail!("No peers discovered on local network.");
}
}
if let Some(q) = query {
if let Some(matched) = match_target_peer(q, &peers) {
return Ok(matched.clone());
}
anyhow::bail!("No discovered peer matched query '{q}'.");
}
if peers.len() == 1 {
let p = &peers[0];
println!("Single peer found: {} ({})", p.alias, p.primary_ip().map(|ip| ip.to_string()).unwrap_or_default());
return Ok(p.clone());
}
println!("\nSelect a peer to send files to:");
for (idx, p) in peers.iter().enumerate() {
let ip_str = p.primary_ip().map(|ip| ip.to_string()).unwrap_or_else(|| "-".to_string());
let fp_str = shorten_fingerprint(&p.fingerprint);
println!(" [{}] {} | IP: {} | Port: {} | FP: {}", idx + 1, p.alias, ip_str, p.port, fp_str);
}
loop {
let choice_str = prompt_user_input(&format!("Enter peer number (1-{}): ", peers.len()))?;
if let Ok(choice) = choice_str.parse::<usize>()
&& (1..=peers.len()).contains(&choice)
{
return Ok(peers[choice - 1].clone());
}
println!("Invalid choice. Please enter a number between 1 and {}.", peers.len());
}
}
#[derive(Clone, Debug)]
pub struct SendOptions {
pub files: Vec<PathBuf>,
pub to: Option<String>,
pub pin: Option<String>,
}
pub async fn send(
config: &AppConfig,
identity: &TlsIdentity,
options: &SendOptions,
) -> anyhow::Result<()> {
if options.files.is_empty() {
anyhow::bail!("No files specified to send");
}
let mut outgoing_files = Vec::with_capacity(options.files.len());
let mut total_bytes = 0u64;
for f in &options.files {
if !f.exists() {
anyhow::bail!("File not found: {}", f.display());
}
println!("Preparing file: {}...", f.display());
let outgoing = OutgoingFile::from_path(f)
.await
.with_context(|| format!("Failed to read file metadata for: {}", f.display()))?
.with_computed_sha256()
.await
.with_context(|| format!("Failed to compute SHA-256 for: {}", f.display()))?;
total_bytes += outgoing.size;
outgoing_files.push(outgoing);
}
println!(
"Prepared {} file(s) ({:.2} MB total).",
outgoing_files.len(),
total_bytes as f64 / (1024.0 * 1024.0)
);
let target = discover_or_select_target(config, identity, options.to.as_deref()).await?;
let base_url = target
.primary_base_url()
.context("Target peer does not have a valid IP endpoint")?;
println!("Target resolved: {} at {}", target.alias, base_url);
let target_fp = if target.fingerprint.is_empty() {
None
} else {
Some(target.fingerprint.clone())
};
let client_config = LsClientConfig {
timeout: Duration::from_secs(60),
connect_timeout: Duration::from_secs(5),
tls_identity: Some(identity.clone()),
target_fingerprint: target_fp,
};
let client = LsClient::new(client_config)
.context("Failed to initialize LocalSend client with TLS identity")?;
let sender_info = RegisterDtoV2 {
alias: config.alias.clone(),
version: "2.0".to_string(),
device_model: Some("LanFlare CLI".to_string()),
device_type: Some(DeviceType::Desktop),
fingerprint: identity.fingerprint.clone(),
port: config.port,
protocol: ProtocolType::Https,
download: false,
};
let orchestrator = TransferOrchestrator::new(client, sender_info);
let (progress_tx, mut progress_rx) = mpsc::channel::<TransferProgress>(100);
let mut tracker = TransferProgressTracker::new(true);
let transfer_opts = TransferOptions {
max_concurrency: 4,
pin: options.pin.clone().or_else(|| config.pin.clone()),
cancel_token: None,
progress_interval: Duration::from_millis(50),
..Default::default()
};
let orchestrator_task = tokio::spawn(async move {
orchestrator
.transfer(&base_url, outgoing_files, transfer_opts, Some(progress_tx))
.await
});
while let Some(ev) = progress_rx.recv().await {
tracker.handle_event(&ev);
}
match orchestrator_task.await {
Ok(Ok(())) => {
println!("All files transferred successfully!");
Ok(())
}
Ok(Err(err)) => {
anyhow::bail!("Transfer failed: {err}");
}
Err(join_err) => {
anyhow::bail!("Transfer task panicked or aborted: {join_err}");
}
}
}
#[derive(Clone, Debug)]
pub struct ReceiveOptions {
pub dir: Option<PathBuf>,
pub port: Option<u16>,
pub auto_accept: bool,
pub pin: Option<String>,
}
pub async fn receive(
config: &AppConfig,
identity: &TlsIdentity,
options: &ReceiveOptions,
) -> anyhow::Result<()> {
let dest_dir = options
.dir
.as_ref()
.unwrap_or(&config.download_dir);
tokio::fs::create_dir_all(dest_dir)
.await
.with_context(|| format!("Failed to create download directory: {}", dest_dir.display()))?;
let port = options.port.unwrap_or(config.port);
let auto_accept = options.auto_accept || config.auto_accept;
let pin = options.pin.clone().or_else(|| config.pin.clone());
let server_config = AppConfig {
alias: config.alias.clone(),
port,
download_dir: dest_dir.clone(),
auto_accept,
pin: pin.clone(),
};
let registry = PeerRegistry::new();
let (event_tx, mut event_rx) = mpsc::channel::<ServerTransferEvent>(100);
let server_handle = start_server(
server_config,
Some(identity.clone()),
registry.clone(),
Some(event_tx),
)
.await
.with_context(|| format!("Failed to start receive server on port {port}"))?;
let bound_port = server_handle.bound_port();
println!("Receiver active and listening on port {bound_port}");
println!("Saving files to: {}", dest_dir.display());
println!("Auto-accept transfers: {}", if auto_accept { "Enabled" } else { "Disabled (interactive prompt)" });
if let Some(ref p) = pin {
println!("PIN protection enabled: required PIN is '{p}'");
}
println!("Certificate Fingerprint: {}", identity.fingerprint);
println!("Press Ctrl+C to terminate receiver daemon.\n");
while let Some(event) = event_rx.recv().await {
match event {
ServerTransferEvent::DeviceRegistered { ip, info } => {
println!("Device registered: {} ({}) at {}", info.alias, info.version, ip);
}
ServerTransferEvent::PrepareUpload {
session_id,
sender_ip,
sender_info,
files,
decision_tx,
..
} => {
println!("\n[Incoming Transfer Request]");
println!(" From: {} ({})", sender_info.alias, sender_ip);
println!(" Session ID: {session_id}");
println!(" Files ({}) :", files.len());
let mut total_size = 0u64;
for (id, file) in &files {
total_size += file.size;
println!(
" - {} ({:.2} MB) [ID: {}]",
file.file_name,
file.size as f64 / (1024.0 * 1024.0),
id
);
}
println!(" Total Size: {:.2} MB", total_size as f64 / (1024.0 * 1024.0));
if auto_accept {
println!("Auto-accepting incoming transfer...");
let all_ids = files.keys().cloned().collect();
let _ = decision_tx.send(PrepareUploadDecision::Accept(all_ids));
} else {
let ans = prompt_user_input("Accept this transfer? [Y/n]: ")
.unwrap_or_else(|_| "n".to_string());
if ans.is_empty() || ans.eq_ignore_ascii_case("y") || ans.eq_ignore_ascii_case("yes") {
let all_ids = files.keys().cloned().collect();
let _ = decision_tx.send(PrepareUploadDecision::Accept(all_ids));
println!("Accepted transfer.");
} else {
let _ = decision_tx.send(PrepareUploadDecision::Decline);
println!("Declined transfer.");
}
}
}
ServerTransferEvent::FileStarted { file_name, total_bytes, .. } => {
println!(
"Receiving file: {} ({:.2} MB)...",
file_name,
total_bytes as f64 / (1024.0 * 1024.0)
);
}
ServerTransferEvent::FileCompleted { result, .. } => {
println!(
"File saved successfully: {} (SHA-256: {}) ✓",
result.target_path.display(),
result.sha256
);
}
ServerTransferEvent::FileFailed { file_id, error, .. } => {
eprintln!("File transfer failed [ID: {file_id}]: {error}");
}
ServerTransferEvent::SessionEnded { session_id, reason } => {
println!("Transfer session ended [ID: {session_id}]: {reason:?}\n");
}
}
}
server_handle.wait().await;
Ok(())
}