use std::collections::HashMap;
use std::net::{TcpListener, TcpStream, ToSocketAddrs};
use std::os::fd::AsRawFd;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Weak};
use std::time::{Duration, Instant};
use parking_lot::Mutex;
use crate::config::{Config, DEVICES_PORT};
use crate::remote::acks::{AckOutcome, Envelope};
use crate::remote::devices;
use crate::remote::link::{CommandSource, LinkCommand, LinkHello, LinkReport, LinkState, Local};
use crate::remote::proof::{self, Peer, Proven};
use crate::remote::wire::{self, Waker};
pub const SERVICE: &str = "_koan._tcp";
#[derive(Debug, serde::Serialize, serde::Deserialize)]
#[serde(tag = "type", rename_all = "camelCase")]
enum ProofFrame {
#[serde(rename = "nearbyAuth")]
Auth {
id: String,
nonce: String,
sig: String,
},
#[serde(rename = "nearbyProof")]
Proof {
verified: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
sig: Option<String>,
},
#[serde(rename = "nearbySigned")]
Signed {
seq: u64,
sig: String,
command: String,
},
#[serde(rename = "nearbySignedReport")]
SignedReport {
seq: u64,
sig: String,
report: String,
},
}
fn frame(f: &ProofFrame) -> Option<String> {
serde_json::to_string(f).ok()
}
struct Conn {
outbox: Vec<Envelope>,
waker: Arc<Waker>,
proven: Option<proof::Peer>,
addr: String,
since: i64,
}
static CONNS: Mutex<Option<HashMap<String, Conn>>> = Mutex::new(None);
#[derive(Debug, Clone, PartialEq)]
pub struct Session {
pub key: Option<u64>,
pub inbound: bool,
pub id: Option<String>,
pub addr: String,
pub proven: Option<Peer>,
pub since: i64,
}
struct Inbound {
session: Session,
ended: Arc<AtomicBool>,
waker: Arc<Waker>,
}
static INBOUND: Mutex<Vec<Inbound>> = Mutex::new(Vec::new());
pub fn sessions() -> Vec<Session> {
let mut out: Vec<Session> = INBOUND.lock().iter().map(|i| i.session.clone()).collect();
if let Some(conns) = CONNS.lock().as_ref() {
out.extend(conns.iter().map(|(id, c)| Session {
key: None,
inbound: false,
id: Some(id.clone()),
addr: c.addr.clone(),
proven: c.proven.clone(),
since: c.since,
}));
}
out.sort_by_key(|s| s.since);
out
}
struct Held {
addr: String,
id: Option<String>,
proven: bool,
}
static HELD: Mutex<Vec<Held>> = Mutex::new(Vec::new());
fn held_addr(ip: &std::net::IpAddr) -> bool {
let ip = ip.to_canonical().to_string();
HELD.lock().iter().any(|h| h.addr == ip)
}
fn held_id(id: &str) -> bool {
HELD.lock().iter().any(|h| h.id.as_deref() == Some(id))
}
pub fn end(key: u64) {
let Some(held) = INBOUND
.lock()
.iter()
.find(|i| i.session.key == Some(key))
.map(|i| Held {
addr: i.session.addr.clone(),
id: i.session.id.clone(),
proven: i.session.proven.is_some(),
})
else {
return;
};
for i in INBOUND
.lock()
.iter()
.filter(|i| i.session.addr == held.addr || (held.id.is_some() && i.session.id == held.id))
{
i.ended.store(true, Ordering::Relaxed);
i.waker.wake();
}
let id = held.id.clone();
{
let mut all = HELD.lock();
match all.iter_mut().find(|h| h.addr == held.addr) {
Some(h) if h.id.is_none() => *h = held,
Some(_) => {}
None => all.push(held),
}
}
if let Some(id) = id
&& let Some(conn) = CONNS.lock().as_ref().and_then(|c| c.get(&id))
{
conn.waker.wake();
}
devices::touch();
}
pub fn held() -> Vec<(String, Option<String>)> {
HELD.lock()
.iter()
.map(|h| (h.addr.clone(), h.id.clone().filter(|_| h.proven)))
.collect()
}
pub fn release(id: &str) {
release_where(|h| h.id.as_deref() == Some(id));
}
pub fn release_addr(addr: &str) {
release_where(|h| h.addr == addr);
}
fn release_where(matches: impl Fn(&Held) -> bool) {
let released: Vec<Held> = {
let mut held = HELD.lock();
let (out, kept) = held.drain(..).partition(|h| matches(h));
*held = kept;
out
};
if released.is_empty() {
return;
}
for id in released.iter().filter_map(|h| h.id.as_deref()) {
redial_device(id, "");
}
devices::touch();
}
fn refused(ip: &std::net::IpAddr) -> bool {
let ip = ip.to_canonical().to_string();
Config::cached().devices.refused.contains(&ip)
}
struct Listed(u64);
impl Listed {
fn new(addr: &std::net::SocketAddr, waker: &Arc<Waker>) -> (Self, Arc<AtomicBool>) {
static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
let key = NEXT.fetch_add(1, Ordering::Relaxed);
let ended = Arc::new(AtomicBool::new(false));
INBOUND.lock().push(Inbound {
session: Session {
key: Some(key),
inbound: true,
id: None,
addr: addr.ip().to_canonical().to_string(),
proven: None,
since: chrono::Utc::now().timestamp(),
},
ended: ended.clone(),
waker: waker.clone(),
});
devices::touch();
(Self(key), ended)
}
fn said(&self, id: &str, proven: Option<Peer>) {
if let Some(i) = INBOUND
.lock()
.iter_mut()
.find(|i| i.session.key == Some(self.0))
{
i.session.id = Some(id.to_owned());
i.session.proven = proven;
}
devices::touch();
}
}
impl Drop for Listed {
fn drop(&mut self) {
INBOUND.lock().retain(|i| i.session.key != Some(self.0));
devices::touch();
}
}
struct Running {
local: Local,
listener: Option<Arc<Stop>>,
dialers: HashMap<String, Dialer>,
off: bool,
}
struct Dialer {
stop: Arc<Stop>,
#[cfg_attr(not(target_vendor = "apple"), allow(dead_code))]
addr: Arc<Mutex<String>>,
redial: Arc<AtomicBool>,
}
impl Dialer {
fn redial(&self) {
self.redial.store(true, Ordering::Relaxed);
self.stop.waker.wake();
}
}
static RUNNING: Mutex<Option<Running>> = Mutex::new(None);
static PORT: Mutex<Option<u16>> = Mutex::new(None);
#[derive(Debug, Clone, PartialEq)]
pub struct Found {
pub name: String,
pub bonjour: Option<String>,
pub id: Option<String>,
pub platform: Option<String>,
pub server: Option<String>,
pub problem: Option<String>,
}
static FOUND: Mutex<Vec<(String, Found)>> = Mutex::new(Vec::new());
static BLOCKED: AtomicBool = AtomicBool::new(false);
pub fn found() -> Vec<Found> {
FOUND.lock().iter().map(|(_, f)| f.clone()).collect()
}
pub fn servers() -> Vec<(String, Vec<String>)> {
let mut out: Vec<(String, Vec<String>)> = Vec::new();
for (_, f) in FOUND.lock().iter() {
let Some(server) = f.server.as_deref().filter(|s| is_server(s)) else {
continue;
};
match out.iter_mut().find(|(s, _)| s == server) {
Some((_, names)) => names.push(f.name.clone()),
None => out.push((server.to_string(), vec![f.name.clone()])),
}
}
out
}
fn is_server(s: &str) -> bool {
url::Url::parse(s).is_ok_and(|u| {
matches!(u.scheme(), "http" | "https")
&& u.username().is_empty()
&& u.password().is_none()
&& u.query().is_none()
&& u.fragment().is_none()
&& match u.host() {
Some(url::Host::Domain(d)) => {
let d = d.trim_end_matches('.').to_ascii_lowercase();
d != "localhost" && !d.ends_with(".localhost")
}
Some(url::Host::Ipv4(ip)) => !ip.is_loopback() && !ip.is_unspecified(),
Some(url::Host::Ipv6(ip)) => !ip.is_loopback() && !ip.is_unspecified(),
None => false,
}
})
}
#[cfg(any(target_vendor = "apple", test))]
const TXT_ENTRY_MAX: usize = 255;
#[cfg(any(target_vendor = "apple", test))]
fn announceable(url: &str) -> Option<String> {
let url = url.trim().trim_end_matches('/');
(is_server(url) && "server=".len() + url.len() <= TXT_ENTRY_MAX).then(|| url.to_string())
}
#[cfg(target_vendor = "apple")]
fn announced_server() -> String {
let cfg = Config::cached();
if crate::helpers::remote_credential(&cfg).is_none() {
return String::new();
}
announceable(&cfg.remote.url).unwrap_or_default()
}
#[cfg(target_vendor = "apple")]
static ADVERT: Mutex<Option<bonjour::Advert>> = Mutex::new(None);
pub fn readvertise() {
#[cfg(target_vendor = "apple")]
{
let Some(port) = *PORT.lock() else { return };
let identity = match RUNNING.lock().as_ref() {
Some(r) if r.listener.is_some() => r.local.identity.clone(),
_ => return,
};
let mut advert = ADVERT.lock();
if advert.is_none() {
return;
}
*advert = None;
let server = announced_server();
*advert = bonjour::advertise(port, &identity, &server);
if advert.is_none() && !server.is_empty() {
log::warn!("nearby: cannot announce the server; announcing this device without it");
*advert = bonjour::advertise(port, &identity, "");
}
}
}
pub fn local_network_blocked() -> bool {
BLOCKED.load(Ordering::Relaxed)
}
fn set_blocked(blocked: bool) {
if BLOCKED.swap(blocked, Ordering::Relaxed) != blocked {
if blocked {
log::warn!("nearby: the local network is blocked for this app");
}
devices::touch();
}
}
fn note(key: &str, problem: Option<String>) {
let mut found = FOUND.lock();
if let Some((_, f)) = found.iter_mut().find(|(k, _)| k == key)
&& f.problem != problem
{
f.problem = problem;
drop(found);
devices::touch();
}
}
fn locally_refused(error: &str) -> bool {
let e = error.to_lowercase();
cfg!(any(target_os = "ios", target_os = "tvos"))
&& (e.contains("no route to host") || e.contains("network is unreachable"))
}
fn explain(error: &str) -> String {
let e = error.to_lowercase();
if locally_refused(error) {
"Blocked: allow Local Network for kōan in Settings".into()
} else if e.contains("no route to host") || e.contains("network is unreachable") {
"Not reachable from this network".into()
} else if e.contains("refused") {
"Not accepting connections. Is kōan open there, and discoverable?".into()
} else if e.contains("timed out") || e.contains("would block") {
"Not answering on this network".into()
} else if e.contains("lookup") || e.contains("nodename") || e.contains("not known") {
"Address not found".into()
} else {
format!("Cannot connect: {error}")
}
}
pub fn refresh() {
let listener = RUNNING.lock().as_mut().and_then(|r| r.listener.take());
if let Some(stop) = listener {
stop.stop();
}
reconfigure();
redial_all();
#[cfg(target_vendor = "apple")]
bonjour::restart();
}
pub fn dial_now() {
redial_all();
}
fn redial_all() {
if let Some(r) = RUNNING.lock().as_ref() {
for d in r.dialers.values() {
d.redial();
}
}
}
fn redial_device(id: &str, except: &str) {
let keys: Vec<String> = FOUND
.lock()
.iter()
.filter(|(k, f)| k != except && f.id.as_deref() == Some(id))
.map(|(k, _)| k.clone())
.collect();
if let Some(r) = RUNNING.lock().as_ref() {
for key in keys {
if let Some(d) = r.dialers.get(&key) {
d.redial();
}
}
}
}
struct Stop {
flag: AtomicBool,
waker: Arc<Waker>,
others: Mutex<Vec<Weak<Waker>>>,
}
impl Stop {
fn new() -> Option<Arc<Self>> {
Some(Arc::new(Self {
flag: AtomicBool::new(false),
waker: Waker::new().ok()?,
others: Mutex::new(Vec::new()),
}))
}
fn stop(&self) {
self.flag.store(true, Ordering::Relaxed);
self.waker.wake();
for w in self.others.lock().iter().filter_map(Weak::upgrade) {
w.wake();
}
}
fn also_wake(&self, waker: &Arc<Waker>) {
let mut others = self.others.lock();
others.retain(|w| w.strong_count() > 0);
others.push(Arc::downgrade(waker));
}
fn stopped(&self) -> bool {
self.flag.load(Ordering::Relaxed)
}
}
pub fn start(local: Local) {
*RUNNING.lock() = Some(Running {
local,
listener: None,
dialers: HashMap::new(),
off: false,
});
#[cfg(target_vendor = "apple")]
if !enabled() {
bonjour::pause();
}
reconfigure();
dial_remembered();
#[cfg(target_vendor = "apple")]
std::thread::Builder::new()
.name("koan-bonjour".into())
.spawn(bonjour::browse_forever)
.expect("failed to spawn the Bonjour thread");
}
fn enabled() -> bool {
Config::load().map(|c| c.devices.nearby).unwrap_or(true)
}
pub fn suspend() {
if let Some(r) = RUNNING.lock().as_mut() {
for (_, d) in r.dialers.drain() {
d.stop.stop();
}
}
FOUND.lock().clear();
devices::touch();
#[cfg(target_vendor = "apple")]
bonjour::pause();
}
pub fn wake() {
if !enabled() {
return;
}
reconfigure();
dial_remembered();
#[cfg(target_vendor = "apple")]
bonjour::restart();
}
fn dial_remembered() {
if !crate::quiet::awake() || !enabled() {
return;
}
if let Some(r) = RUNNING.lock().as_mut() {
for seen in devices::remembered_nearby() {
let key = format!("id:{}", seen.id);
if r.dialers.contains_key(&key) {
continue;
}
FOUND.lock().push((
key.clone(),
Found {
name: seen.name,
bonjour: None,
id: Some(seen.id),
platform: Some(seen.platform),
server: None,
problem: None,
},
));
spawn_dialer(r, key, seen.addr);
}
}
}
pub fn reconfigure() {
let cfg = Config::load().unwrap_or_default();
let mut running = RUNNING.lock();
let Some(r) = running.as_mut() else { return };
if !cfg.devices.nearby {
if let Some(stop) = r.listener.take() {
stop.stop();
}
for (_, d) in r.dialers.drain() {
d.stop.stop();
}
FOUND.lock().clear();
if !r.off {
log::info!("nearby: off (devices.nearby = false)");
r.off = true;
devices::touch();
#[cfg(target_vendor = "apple")]
bonjour::pause();
}
return;
}
let turned_on = std::mem::replace(&mut r.off, false);
match (
&r.listener,
cfg.devices.discoverable && crate::quiet::findable(),
) {
(None, true) => {
if let Some(stop) = Stop::new() {
let (local, port, s) = (r.local.clone(), cfg.devices.port, stop.clone());
std::thread::Builder::new()
.name("koan-nearby".into())
.spawn(move || listen(local, port, s))
.expect("failed to spawn the listener");
r.listener = Some(stop);
}
}
(Some(stop), false) => {
stop.stop();
r.listener = None;
}
_ => {}
}
let wanted: Vec<String> = cfg
.devices
.addresses
.iter()
.map(|a| a.trim().to_string())
.filter(|a| !a.is_empty())
.collect();
r.dialers.retain(|key, d| {
let keep = key.starts_with("bonjour:") || key.starts_with("id:") || wanted.contains(key);
if !keep {
d.stop.stop();
FOUND.lock().retain(|(k, _)| k != key);
}
keep
});
if !crate::quiet::awake() {
return;
}
for addr in wanted {
if !r.dialers.contains_key(&addr) {
spawn_dialer(r, addr.clone(), addr);
}
}
if turned_on {
log::info!("nearby: on");
drop(running);
dial_remembered();
#[cfg(target_vendor = "apple")]
bonjour::restart();
}
}
pub fn listening_port() -> Option<u16> {
*PORT.lock()
}
pub fn send(id: &str, cmd: LinkCommand) -> bool {
send_envelope(id, cmd.into())
}
pub fn send_acked(id: &str, cmd: LinkCommand, ack: u64) -> bool {
send_envelope(
id,
Envelope {
command: cmd,
ack: Some(ack),
},
)
}
fn send_envelope(id: &str, envelope: Envelope) -> bool {
release(id);
queue(id, devices::listed_owner(id), envelope)
}
fn queue(id: &str, listed: Option<Option<String>>, envelope: Envelope) -> bool {
let mut conns = CONNS.lock();
let Some(conn) = conns.as_mut().and_then(|c| c.get_mut(id)) else {
return false;
};
if !proven_as_listed(listed, conn.proven.as_ref()) {
log::info!("nearby: {id} has not proved it is that device; not sent here");
return false;
}
conn.outbox.push(envelope);
conn.waker.wake();
true
}
fn listen(local: Local, port: u16, stop: Arc<Stop>) {
while !stop.stopped() {
if let Err(e) = listen_once(&local, port, &stop) {
log::warn!("nearby: {e}; listening again");
let mut fds = [libc::pollfd {
fd: stop.waker_fd(),
events: libc::POLLIN,
revents: 0,
}];
unsafe { libc::poll(fds.as_mut_ptr(), 1, 1000) };
}
}
*PORT.lock() = None;
devices::touch();
log::info!("nearby: stopped listening");
}
fn bind(port: u16) -> std::io::Result<TcpListener> {
TcpListener::bind(("::", port)).or_else(|_| TcpListener::bind(("0.0.0.0", port)))
}
fn listen_once(local: &Local, port: u16, stop: &Arc<Stop>) -> Result<(), String> {
let mut listener = bind(port);
for _ in 0..20 {
if listener.is_ok() || stop.stopped() {
break;
}
std::thread::sleep(Duration::from_millis(100));
listener = bind(port);
}
let listener = listener
.or_else(|e| {
log::warn!("nearby: port {port} is taken ({e}); listening elsewhere");
bind(0)
})
.map_err(|e| format!("cannot listen: {e}"))?;
let port = listener.local_addr().map_err(|e| e.to_string())?.port();
listener.set_nonblocking(true).map_err(|e| e.to_string())?;
*PORT.lock() = Some(port);
devices::touch();
log::info!("nearby: listening on {port}");
#[cfg(target_vendor = "apple")]
let _advert = Announced::new(port, &local.identity);
while !stop.stopped() {
match listener.accept() {
Ok((_, addr)) if refused(&addr.ip()) || held_addr(&addr.ip()) => {
log::info!("nearby: {addr} is refused or held off; hung up");
}
Ok((stream, addr)) => {
let (local, stop) = (local.clone(), stop.clone());
let _ = std::thread::Builder::new()
.name("koan-nearby-peer".into())
.spawn(move || {
if let Err(e) = serve(stream, addr, &local, &stop) {
log::info!("nearby: {addr}: {e}");
}
});
}
Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
let mut fds = [
libc::pollfd {
fd: listener.as_raw_fd(),
events: libc::POLLIN,
revents: 0,
},
libc::pollfd {
fd: stop.waker_fd(),
events: libc::POLLIN,
revents: 0,
},
];
unsafe { libc::poll(fds.as_mut_ptr(), 2, -1) };
}
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
Err(e) => return Err(format!("accept: {e}")),
}
}
Ok(())
}
impl Stop {
fn waker_fd(&self) -> i32 {
self.waker.read_fd()
}
fn drain(&self) {
self.waker.drain();
}
}
fn serve(
stream: TcpStream,
addr: std::net::SocketAddr,
local: &Local,
stop: &Arc<Stop>,
) -> Result<(), String> {
stream.set_nonblocking(false).map_err(|e| e.to_string())?;
stream
.set_read_timeout(Some(Duration::from_secs(10)))
.map_err(|e| e.to_string())?;
let mut socket = tungstenite::accept(stream).map_err(|e| e.to_string())?;
let fd = socket.get_ref().as_raw_fd();
socket
.get_ref()
.set_nonblocking(true)
.map_err(|e| e.to_string())?;
let _ = socket.get_ref().set_nodelay(true);
let waker = Waker::new().map_err(|e| e.to_string())?;
wire::wake_on_engine_change(&waker);
stop.also_wake(&waker);
let mut session = Serving {
listed: Some(Listed::new(&addr, &waker)),
local,
stop,
greeted: false,
sent: None,
waker: waker.clone(),
levels: None,
answers: Default::default(),
nonce: proof::nonce(),
answered: false,
proven: None,
reports: None,
pending: Vec::new(),
};
wire::drive(&mut socket, fd, &waker, &mut session)
}
struct Serving<'a> {
local: &'a Local,
stop: &'a Arc<Stop>,
greeted: bool,
sent: Option<(LinkState, Instant)>,
waker: Arc<Waker>,
levels: Option<crate::remote::levels::Watch>,
answers: Arc<Mutex<Vec<LinkReport>>>,
nonce: Option<String>,
answered: bool,
proven: Option<(Proven, proof::Session)>,
reports: Option<proof::Session>,
pending: Vec<String>,
listed: Option<(Listed, Arc<AtomicBool>)>,
}
impl Serving<'_> {
fn proof(&mut self, frame_in: ProofFrame) {
match frame_in {
ProofFrame::Auth { id, nonce, sig } => {
if std::mem::replace(&mut self.answered, true) {
return;
}
let Some(listen_nonce) = self.nonce.clone() else {
return;
};
let me = &self.local.identity.device_id;
let proven = proof::verify_dial(me, &id, &listen_nonce, &nonce, &sig);
let verified = proven.is_some();
match &proven {
Some(p) => log::info!("nearby: {id} proved itself: {:?}", p.peer),
None => log::info!("nearby: {id} proved nothing; trusted as the network is"),
}
if let Some((listed, ended)) = &self.listed {
listed.said(&id, proven.as_ref().map(|p| p.peer.clone()));
if held_id(&id) {
ended.store(true, Ordering::Relaxed);
}
}
self.proven =
proven.map(|p| (p, proof::Session::new(me, &id, &listen_nonce, &nonce)));
let sig = proof::sign_listen(&id, me, &nonce, &listen_nonce, verified);
if sig.is_some() && proof::signs_reports(&listen_nonce, &nonce) {
self.reports = Some(proof::Session::reports(me, &id, &listen_nonce, &nonce));
self.sent = None;
}
self.pending
.extend(frame(&ProofFrame::Proof { verified, sig }));
}
ProofFrame::Signed { seq, sig, command } => {
let Some((by, session)) = self.proven.as_mut() else {
log::warn!("nearby: a signed command from a peer that proved nothing; refused");
return;
};
if !session.accept(by, seq, &sig, &command) {
log::warn!("nearby: a command not signed for this connection; refused");
return;
}
let envelope = match serde_json::from_str::<Envelope>(&command) {
Ok(envelope) => envelope,
Err(e) => return log::warn!("nearby: not a command ({e}): {command}"),
};
let peer = by.peer.clone();
self.run(envelope, &peer);
}
ProofFrame::Proof { .. } | ProofFrame::SignedReport { .. } => {}
}
}
fn run(&mut self, envelope: Envelope, peer: &Peer) {
let source_of = |cmd: &LinkCommand| match peer {
Peer::Own if cmd.relayable() => Some(CommandSource::Account),
Peer::Shared(_) if cmd.allowed_playback() => Some(CommandSource::Shared),
_ => None,
};
let admitted = admit(envelope, source_of, self.answer());
self.dispatch(admitted);
}
fn answer(&self) -> impl FnOnce(u64, AckOutcome) + Send + 'static {
let (answers, waker) = (self.answers.clone(), self.waker.clone());
move |ack, outcome| {
answers.lock().push(LinkReport::Ack { ack, outcome });
waker.wake();
}
}
fn dispatch(&mut self, admitted: Admitted) {
match admitted {
Admitted::Levels(on, pending) => {
self.levels = on.then(|| crate::remote::levels::feed().watch(&self.waker));
if let Some(pending) = pending {
pending.finish(AckOutcome::Done);
}
}
Admitted::Command(cmd, source, pending) => {
(self.local.on_command)(cmd, source, pending)
}
Admitted::Neither => {}
}
}
}
impl wire::Session for Serving<'_> {
fn outgoing(&mut self) -> Vec<String> {
let mut out = Vec::new();
if !self.greeted {
let cfg = Config::load().unwrap_or_default();
let id = &self.local.identity;
out.push(LinkReport::Hello(LinkHello {
id: id.device_id.clone(),
name: id.name.clone(),
platform: id.platform.clone(),
library: crate::remote::link::library_fingerprint(&cfg),
acks: true,
nonce: self.nonce.clone(),
}));
self.greeted = true;
}
out.append(&mut self.answers.lock());
let now = for_the_network((self.local.state)(), full_control());
if self
.sent
.as_ref()
.is_none_or(|(s, at)| now.differs(s, at.elapsed()))
{
out.push(LinkReport::State(now.clone()));
self.sent = Some((now, Instant::now()));
}
if let Some(f) = self.levels.as_mut().and_then(|w| w.take()) {
out.push(LinkReport::Levels { f });
}
let mut texts = std::mem::take(&mut self.pending);
for report in out {
let Ok(json) = serde_json::to_string(&report) else {
continue;
};
match &mut self.reports {
Some(session) => match session.sign(&json) {
Some((seq, sig)) => texts.extend(frame(&ProofFrame::SignedReport {
seq,
sig,
report: json,
})),
None => log::warn!("nearby: no key to sign a report with; dropped"),
},
None => texts.push(json),
}
}
texts
}
fn incoming(&mut self, text: &str) {
if let Ok(f) = serde_json::from_str::<ProofFrame>(text) {
return self.proof(f);
}
let envelope = match serde_json::from_str::<Envelope>(text) {
Ok(envelope) => envelope,
Err(e) => {
log::warn!("nearby: not a command ({e}): {text}");
return;
}
};
if self.proven.is_some() {
log::warn!(
"nearby: an unsigned command on a proven connection; refused: {:?}",
envelope.command
);
return;
}
let full = full_control();
let admitted = admit(envelope, |cmd| cmd.from_the_network(full), self.answer());
self.dispatch(admitted);
}
fn done(&self) -> bool {
self.stop.stopped()
|| self
.listed
.as_ref()
.is_some_and(|(_, ended)| ended.load(Ordering::Relaxed))
}
}
enum Admitted {
Levels(bool, Option<crate::remote::acks::Pending>),
Command(
LinkCommand,
crate::remote::link::CommandSource,
Option<crate::remote::acks::Pending>,
),
Neither,
}
fn admit(
envelope: Envelope,
source_of: impl FnOnce(&LinkCommand) -> Option<crate::remote::link::CommandSource>,
answer: impl FnOnce(u64, AckOutcome) + Send + 'static,
) -> Admitted {
let source = match &envelope.command {
LinkCommand::WatchLevels { .. } => None,
cmd => match source_of(cmd) {
Some(source) => Some(source),
None => {
log::warn!("nearby: refused {cmd:?}");
if let Some(ack) = envelope.ack {
answer(
ack,
AckOutcome::Refused {
reason: "not allowed from this network".into(),
},
);
}
return Admitted::Neither;
}
},
};
let Some((command, pending)) = crate::remote::acks::take(envelope, answer) else {
return Admitted::Neither;
};
match (command, source) {
(LinkCommand::WatchLevels { on }, _) => Admitted::Levels(on, pending),
(cmd, Some(source)) => Admitted::Command(cmd, source, pending),
(_, None) => Admitted::Neither,
}
}
fn full_control() -> bool {
crate::config::Config::cached().devices.nearby_control == crate::config::NearbyControl::Full
}
fn for_the_network(state: LinkState, full: bool) -> LinkState {
if full {
return state;
}
LinkState {
outputs: None,
..state
}
}
const RETRY_MIN: Duration = Duration::from_secs(2);
const RETRY_MAX: Duration = Duration::from_secs(60);
const SERVED_LONG_ENOUGH: Duration = Duration::from_secs(30);
fn spawn_dialer(r: &mut Running, key: String, addr: String) {
let Some(stop) = Stop::new() else { return };
let addr = Arc::new(Mutex::new(addr));
let redial = Arc::new(AtomicBool::new(false));
r.dialers.insert(
key.clone(),
Dialer {
stop: stop.clone(),
addr: addr.clone(),
redial: redial.clone(),
},
);
{
let mut found = FOUND.lock();
if !found.iter().any(|(k, _)| *k == key) {
found.push((
key.clone(),
Found {
name: addr.lock().clone(),
bonjour: None,
id: None,
platform: None,
server: None,
problem: None,
},
));
}
}
devices::touch();
let _ = std::thread::Builder::new()
.name("koan-nearby-dial".into())
.spawn(move || dial(key, addr, stop, redial));
}
fn dial(key: String, at: Arc<Mutex<String>>, stop: Arc<Stop>, redial: Arc<AtomicBool>) {
let mut wait = RETRY_MIN;
while !stop.stopped() {
let addr = at.lock().clone();
let mut served = false;
let started = Instant::now();
let held = FOUND
.lock()
.iter()
.find(|(k, _)| *k == key)
.and_then(|(_, f)| f.id.clone())
.is_some_and(|id| held_id(&id));
if held {
let mut fds = [libc::pollfd {
fd: stop.waker_fd(),
events: libc::POLLIN,
revents: 0,
}];
unsafe { libc::poll(fds.as_mut_ptr(), 1, -1) };
stop.drain();
redial.store(false, Ordering::Relaxed);
wait = RETRY_MIN;
continue;
}
match connect(&addr) {
Ok(mut socket) => {
let fd = socket.get_ref().as_raw_fd();
let Ok(waker) = Waker::new() else { return };
stop.also_wake(&waker);
let mut session = Controlling {
stop: &stop,
waker: waker.clone(),
key: &key,
addr: &addr,
id: None,
this_device: false,
duplicate: false,
handshake: Handshake::Plain,
proven: None,
reports: None,
pending: Vec::new(),
};
let result = wire::drive(&mut socket, fd, &waker, &mut session);
served = session
.id
.take()
.inspect(|id| {
if let Some(conns) = CONNS.lock().as_mut() {
conns.remove(id);
}
devices::nearby_gone(id);
redial_device(id, &key);
})
.is_some()
&& started.elapsed() >= SERVED_LONG_ENOUGH;
if session.this_device {
FOUND.lock().retain(|(k, _)| *k != key);
devices::touch();
return;
}
if let Err(e) = result {
log::info!("nearby: {addr}: {e}");
}
}
Err(e) => {
log::debug!("nearby: {addr}: {e}");
if locally_refused(&e) {
set_blocked(true);
}
note(&key, Some(explain(&e)));
}
}
let mut fds = [libc::pollfd {
fd: stop.waker_fd(),
events: libc::POLLIN,
revents: 0,
}];
unsafe { libc::poll(fds.as_mut_ptr(), 1, wait.as_millis() as i32) };
stop.drain();
wait = next_wait(wait, served, redial.swap(false, Ordering::Relaxed));
}
}
fn next_wait(wait: Duration, served: bool, redialed: bool) -> Duration {
if redialed {
RETRY_MIN
} else if served {
RETRY_MIN * 2
} else {
(wait * 2).min(RETRY_MAX)
}
}
fn connect(addr: &str) -> Result<tungstenite::WebSocket<TcpStream>, String> {
let host_port = if addr
.rsplit_once(':')
.is_some_and(|(_, p)| p.parse::<u16>().is_ok())
&& !addr.ends_with(']')
{
addr.to_string()
} else {
format!("{addr}:{DEVICES_PORT}")
};
let targets = host_port.to_socket_addrs().map_err(|e| e.to_string())?;
let mut last = "no address".to_string();
for target in targets {
match TcpStream::connect_timeout(&target, Duration::from_secs(3)) {
Ok(stream) => {
stream
.set_read_timeout(Some(Duration::from_secs(10)))
.map_err(|e| e.to_string())?;
let url = format!("ws://{host_port}/");
let (socket, _) =
tungstenite::client(url.as_str(), stream).map_err(|e| e.to_string())?;
socket
.get_ref()
.set_nonblocking(true)
.map_err(|e| e.to_string())?;
let _ = socket.get_ref().set_nodelay(true);
return Ok(socket);
}
Err(e) => last = e.to_string(),
}
}
Err(last)
}
struct Controlling<'a> {
stop: &'a Arc<Stop>,
waker: Arc<Waker>,
key: &'a str,
addr: &'a str,
id: Option<String>,
this_device: bool,
duplicate: bool,
handshake: Handshake,
proven: Option<proof::Peer>,
reports: Option<(Proven, proof::Session)>,
pending: Vec<String>,
}
enum Handshake {
Plain,
Awaiting {
listener: String,
me: String,
listen_nonce: String,
dial_nonce: String,
},
Signed(proof::Session),
}
impl Controlling<'_> {
fn prove(&mut self, listener: &str, listen_nonce: String) {
let Some(me) = devices::this_id() else { return };
let Some(dial_nonce) = proof::nonce() else {
return;
};
let Some(sig) = proof::sign_dial(listener, &me, &listen_nonce, &dial_nonce) else {
return;
};
self.pending.extend(frame(&ProofFrame::Auth {
id: me.clone(),
nonce: dial_nonce.clone(),
sig,
}));
self.handshake = Handshake::Awaiting {
listener: listener.to_owned(),
me,
listen_nonce,
dial_nonce,
};
}
fn answered(&mut self, verified: bool, sig: Option<String>) {
let Handshake::Awaiting {
listener,
me,
listen_nonce,
dial_nonce,
} = &self.handshake
else {
return;
};
let listener_is = sig.and_then(|sig| {
proof::verify_listen(me, listener, dial_nonce, listen_nonce, verified, &sig)
});
log::info!(
"nearby: {listener} {} us; it proved itself: {:?}",
if verified { "took" } else { "did not take" },
listener_is.as_ref().map(|p| &p.peer)
);
self.reports = listener_is
.clone()
.filter(|_| proof::signs_reports(listen_nonce, dial_nonce))
.map(|p| {
let session = proof::Session::reports(listener, me, listen_nonce, dial_nonce);
(p, session)
});
self.proven = listener_is.map(|p| p.peer);
if let Some(id) = &self.id
&& let Some(conn) = CONNS.lock().as_mut().and_then(|c| c.get_mut(id))
{
conn.proven = self.proven.clone();
}
devices::touch();
self.handshake = if verified {
Handshake::Signed(proof::Session::new(listener, me, listen_nonce, dial_nonce))
} else {
Handshake::Plain
};
}
}
impl wire::Session for Controlling<'_> {
fn outgoing(&mut self) -> Vec<String> {
let mut out = std::mem::take(&mut self.pending);
if matches!(self.handshake, Handshake::Awaiting { .. }) {
return out;
}
let Some(id) = &self.id else {
return out;
};
let commands = {
let mut conns = CONNS.lock();
let Some(conn) = conns.as_mut().and_then(|c| c.get_mut(id)) else {
return out;
};
std::mem::take(&mut conn.outbox)
};
for command in commands {
let Ok(json) = serde_json::to_string(&command) else {
continue;
};
match &mut self.handshake {
Handshake::Signed(session) => match session.sign(&json) {
Some((seq, sig)) => out.extend(frame(&ProofFrame::Signed {
seq,
sig,
command: json,
})),
None => log::warn!("nearby: no key to sign with; {command:?} dropped"),
},
_ => out.push(json),
}
}
out
}
fn incoming(&mut self, text: &str) {
if let Ok(ProofFrame::Proof { verified, sig }) = serde_json::from_str(text) {
return self.answered(verified, sig);
}
if let Some(report) = self.checked(text) {
self.report(&report);
}
}
fn done(&self) -> bool {
self.this_device
|| self.duplicate
|| self.stop.stopped()
|| self.id.as_deref().is_some_and(held_id)
}
}
impl Controlling<'_> {
fn checked<'t>(&mut self, text: &'t str) -> Option<std::borrow::Cow<'t, str>> {
if let Ok(ProofFrame::SignedReport { seq, sig, report }) = serde_json::from_str(text) {
if let Some((by, session)) = self.reports.as_mut()
&& !session.accept(by, seq, &sig, &report)
{
log::warn!("nearby: a report not signed for this connection; refused");
return None;
}
return Some(report.into());
}
if self.reports.is_some() {
log::warn!("nearby: an unsigned report on a connection that signs them; refused");
return None;
}
Some(text.into())
}
fn vouched(&self) -> Option<&str> {
let id = self.id.as_deref()?;
let vouched = proven_as_listed(devices::listed_owner(id), self.proven.as_ref());
if !vouched {
log::debug!("nearby: a report from {id}, which did not prove it is; ignored");
}
vouched.then_some(id)
}
fn report(&mut self, text: &str) {
match serde_json::from_str::<LinkReport>(text) {
Ok(LinkReport::Hello(mut hello)) => {
if devices::this_id().as_deref() == Some(hello.id.as_str()) {
self.this_device = true;
return;
}
let mut guard = CONNS.lock();
let conns = guard.get_or_insert_with(HashMap::new);
if conns.contains_key(&hello.id) {
self.duplicate = true;
return;
}
conns.insert(
hello.id.clone(),
Conn {
outbox: Vec::new(),
waker: self.waker.clone(),
proven: None,
addr: self.addr.to_string(),
since: chrono::Utc::now().timestamp(),
},
);
self.id = Some(hello.id.clone());
log::info!("nearby: found {} ({})", hello.name, hello.platform);
if let Some(nonce) = hello.nonce.take() {
self.prove(&hello.id, nonce);
}
drop(guard);
set_blocked(false);
{
let mut found = FOUND.lock();
if let Some((_, f)) = found.iter_mut().find(|(k, _)| k == self.key) {
f.id = Some(hello.id.clone());
f.platform = Some(hello.platform.clone());
f.problem = None;
}
}
devices::nearby_hello(hello, self.addr);
}
Ok(LinkReport::State(state)) => {
if let Some(id) = self.vouched() {
devices::nearby_state(id, state);
}
}
Ok(LinkReport::Levels { f }) => {
if let Some(id) = self.vouched() {
crate::remote::levels::remote().received(id, f);
}
}
Ok(LinkReport::Ack { ack, outcome }) => {
if let Some(id) = self.vouched() {
crate::remote::acks::resolve(ack, id, outcome);
}
}
Ok(_) => {}
Err(e) => log::debug!("nearby: not a report ({e})"),
}
}
}
fn proven_as_listed(listed: Option<Option<String>>, proven: Option<&proof::Peer>) -> bool {
match listed {
None => true,
Some(None) => proven == Some(&proof::Peer::Own),
Some(Some(owner)) => matches!(proven, Some(proof::Peer::Shared(o)) if *o == owner),
}
}
#[cfg(target_vendor = "apple")]
struct Announced;
#[cfg(target_vendor = "apple")]
impl Announced {
fn new(port: u16, identity: &crate::remote::link::LinkIdentity) -> Self {
*ADVERT.lock() = bonjour::advertise(port, identity, &announced_server());
Self
}
}
#[cfg(target_vendor = "apple")]
impl Drop for Announced {
fn drop(&mut self) {
*ADVERT.lock() = None;
}
}
#[cfg(target_vendor = "apple")]
fn announced(
name: String,
host: String,
port: u16,
id: Option<String>,
platform: Option<String>,
server: Option<String>,
) {
if !crate::quiet::awake() || id.is_some() && id == devices::this_id() {
return;
}
set_blocked(false);
let key = match &id {
Some(id) => format!("id:{id}"),
None => format!("bonjour:{name}"),
};
let addr = format!("{}:{port}", host.trim_end_matches('.'));
{
let mut found = FOUND.lock();
let problem = found
.iter()
.find(|(k, _)| *k == key)
.and_then(|(_, f)| f.problem.clone());
found.retain(|(k, _)| *k != key);
found.push((
key.clone(),
Found {
name: name.clone(),
bonjour: Some(name),
id,
platform,
server: server.filter(|s| !s.is_empty()),
problem,
},
));
}
devices::touch();
let mut running = RUNNING.lock();
let Some(r) = running.as_mut() else { return };
match r.dialers.get(&key) {
Some(d) => {
*d.addr.lock() = addr;
d.redial();
}
None => spawn_dialer(r, key, addr),
}
}
#[cfg(target_vendor = "apple")]
fn lost(name: &str) {
let key = {
let mut found = FOUND.lock();
let key = found
.iter()
.find(|(_, f)| f.bonjour.as_deref() == Some(name))
.map(|(k, _)| k.clone());
if let Some(k) = &key {
found.retain(|(kk, _)| kk != k);
}
key
};
let Some(key) = key else { return };
devices::touch();
if let Some(r) = RUNNING.lock().as_mut()
&& let Some(d) = r.dialers.remove(&key)
{
d.stop.stop();
}
}
#[cfg(target_vendor = "apple")]
pub(crate) mod bonjour {
use std::ffi::{CStr, CString, c_char, c_void};
use crate::remote::link::LinkIdentity;
type Ref = *mut c_void;
type BrowseReply =
extern "C" fn(Ref, u32, u32, i32, *const c_char, *const c_char, *const c_char, *mut c_void);
type ResolveReply = extern "C" fn(
Ref,
u32,
u32,
i32,
*const c_char,
*const c_char,
u16,
u16,
*const u8,
*mut c_void,
);
unsafe extern "C" {
fn DNSServiceRegister(
sd: *mut Ref,
flags: u32,
interface: u32,
name: *const c_char,
regtype: *const c_char,
domain: *const c_char,
host: *const c_char,
port: u16,
txt_len: u16,
txt: *const c_void,
callback: *const c_void,
context: *mut c_void,
) -> i32;
fn DNSServiceBrowse(
sd: *mut Ref,
flags: u32,
interface: u32,
regtype: *const c_char,
domain: *const c_char,
callback: BrowseReply,
context: *mut c_void,
) -> i32;
fn DNSServiceResolve(
sd: *mut Ref,
flags: u32,
interface: u32,
name: *const c_char,
regtype: *const c_char,
domain: *const c_char,
callback: ResolveReply,
context: *mut c_void,
) -> i32;
fn DNSServiceRefSockFD(sd: Ref) -> i32;
fn DNSServiceProcessResult(sd: Ref) -> i32;
fn DNSServiceRefDeallocate(sd: Ref);
}
const FLAG_ADD: u32 = 0x2;
pub struct Advert(Ref);
unsafe impl Send for Advert {}
impl Drop for Advert {
fn drop(&mut self) {
unsafe { DNSServiceRefDeallocate(self.0) };
}
}
pub fn advertise(port: u16, identity: &LinkIdentity, server: &str) -> Option<Advert> {
let name = CString::new(identity.name.as_str()).ok()?;
let regtype = CString::new(super::SERVICE).ok()?;
let mut pairs = vec![
("id", identity.device_id.as_str()),
("platform", identity.platform.as_str()),
];
if !server.is_empty() {
pairs.push(("server", server));
}
let txt = txt_record(&pairs);
let mut sd: Ref = std::ptr::null_mut();
let err = unsafe {
DNSServiceRegister(
&mut sd,
0,
0,
name.as_ptr(),
regtype.as_ptr(),
std::ptr::null(),
std::ptr::null(),
port.to_be(),
txt.len() as u16,
txt.as_ptr().cast(),
std::ptr::null(),
std::ptr::null_mut(),
)
};
if err != 0 {
log::warn!("nearby: cannot announce this device ({err})");
if err == POLICY_DENIED {
super::set_blocked(true);
}
return None;
}
Some(Advert(sd))
}
fn txt_record(pairs: &[(&str, &str)]) -> Vec<u8> {
let mut out = Vec::new();
for (k, v) in pairs {
let entry = format!("{k}={v}");
let bytes = &entry.as_bytes()[..entry.len().min(255)];
out.push(bytes.len() as u8);
out.extend_from_slice(bytes);
}
out
}
fn txt_value(txt: &[u8], key: &str) -> Option<String> {
let mut rest = txt;
while let Some((&len, tail)) = rest.split_first() {
let (entry, next) = tail.split_at((len as usize).min(tail.len()));
if let Some(v) = std::str::from_utf8(entry)
.ok()
.and_then(|e| e.strip_prefix(key))
.and_then(|e| e.strip_prefix('='))
{
return Some(v.to_string());
}
rest = next;
}
None
}
struct Seen {
add: bool,
interface: u32,
name: CString,
regtype: CString,
domain: CString,
}
#[derive(Default)]
struct Browsed {
seen: Vec<Seen>,
err: i32,
}
extern "C" fn on_browse(
_: Ref,
flags: u32,
interface: u32,
err: i32,
name: *const c_char,
regtype: *const c_char,
domain: *const c_char,
context: *mut c_void,
) {
unsafe {
let browsed = &mut *(context as *mut Browsed);
if err != 0 {
browsed.err = err;
return;
}
browsed.seen.push(Seen {
add: flags & FLAG_ADD != 0,
interface,
name: CStr::from_ptr(name).to_owned(),
regtype: CStr::from_ptr(regtype).to_owned(),
domain: CStr::from_ptr(domain).to_owned(),
});
}
}
const POLICY_DENIED: i32 = -65570;
static RESTART: std::sync::OnceLock<std::sync::Arc<crate::remote::wire::Waker>> =
std::sync::OnceLock::new();
static RESTART_ASKED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
static PAUSED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
pub fn restart() {
PAUSED.store(false, std::sync::atomic::Ordering::Relaxed);
RESTART_ASKED.store(true, std::sync::atomic::Ordering::Relaxed);
if let Some(w) = RESTART.get() {
w.wake();
}
}
pub fn pause() {
PAUSED.store(true, std::sync::atomic::Ordering::Relaxed);
RESTART_ASKED.store(true, std::sync::atomic::Ordering::Relaxed);
if let Some(w) = RESTART.get() {
w.wake();
}
}
pub fn browse_forever() {
let Ok(waker) = crate::remote::wire::Waker::new() else {
return;
};
let _ = RESTART.set(waker.clone());
loop {
while PAUSED.load(std::sync::atomic::Ordering::Relaxed) {
let mut fds = [libc::pollfd {
fd: waker.read_fd(),
events: libc::POLLIN,
revents: 0,
}];
unsafe { libc::poll(fds.as_mut_ptr(), 1, -1) };
waker.drain();
}
if let Err(e) = browse_once(&waker) {
log::warn!("nearby: browsing stopped: {e}");
}
let mut fds = [libc::pollfd {
fd: waker.read_fd(),
events: libc::POLLIN,
revents: 0,
}];
if !RESTART_ASKED.load(std::sync::atomic::Ordering::Relaxed) {
unsafe { libc::poll(fds.as_mut_ptr(), 1, 2000) };
}
waker.drain();
}
}
fn browse_once(waker: &crate::remote::wire::Waker) -> Result<(), String> {
use std::sync::atomic::Ordering;
RESTART_ASKED.store(false, Ordering::Relaxed);
let regtype = CString::new(super::SERVICE).map_err(|e| e.to_string())?;
let mut browsed = Browsed::default();
let mut interfaces: std::collections::HashMap<String, usize> = Default::default();
let mut unconfirmed: std::collections::HashSet<String> = super::FOUND
.lock()
.iter()
.filter_map(|(_, f)| f.bonjour.clone())
.collect();
let settle = std::time::Instant::now() + std::time::Duration::from_secs(3);
let mut sd: Ref = std::ptr::null_mut();
let err = unsafe {
DNSServiceBrowse(
&mut sd,
0,
0,
regtype.as_ptr(),
std::ptr::null(),
on_browse,
(&mut browsed as *mut Browsed).cast(),
)
};
if err != 0 {
super::set_blocked(err == POLICY_DENIED);
return Err(format!("cannot browse ({err})"));
}
let finish = |sd: Ref, result: Result<(), String>| {
unsafe { DNSServiceRefDeallocate(sd) };
result
};
let fd = unsafe { DNSServiceRefSockFD(sd) };
loop {
let wait = if unconfirmed.is_empty() {
-1
} else {
settle
.saturating_duration_since(std::time::Instant::now())
.as_millis() as i32
};
let mut fds = [
libc::pollfd {
fd,
events: libc::POLLIN,
revents: 0,
},
libc::pollfd {
fd: waker.read_fd(),
events: libc::POLLIN,
revents: 0,
},
];
let ready = unsafe { libc::poll(fds.as_mut_ptr(), 2, wait) };
if RESTART_ASKED.load(Ordering::Relaxed) {
return finish(sd, Ok(()));
}
waker.drain();
if !unconfirmed.is_empty() && std::time::Instant::now() >= settle {
for name in unconfirmed.drain() {
super::lost(&name);
}
}
if ready <= 0 || fds[0].revents == 0 {
continue;
}
if unsafe { DNSServiceProcessResult(sd) } != 0 {
return finish(sd, Err("the responder went away".into()));
}
let err = std::mem::take(&mut browsed.err);
if err != 0 {
super::set_blocked(err == POLICY_DENIED);
return finish(sd, Err(format!("browse error {err}")));
}
for s in browsed.seen.drain(..) {
let name = s.name.to_string_lossy().into_owned();
let count = interfaces.entry(name.clone()).or_insert(0usize);
if !s.add {
*count = count.saturating_sub(1);
if *count == 0 {
interfaces.remove(&name);
unconfirmed.remove(&name);
super::lost(&name);
}
continue;
}
unconfirmed.remove(&name);
*count += 1;
if *count > 1 {
continue;
}
std::thread::spawn(move || {
if let Some((host, port, id, platform, server)) = resolve(&s) {
super::announced(name, host, port, id, platform, server);
}
});
}
}
}
pub struct Instance {
pub name: String,
pub interface: u32,
pub host: String,
pub port: u16,
pub txt: Vec<u8>,
}
impl Instance {
pub fn txt(&self, key: &str) -> Option<String> {
txt_value(&self.txt, key)
}
}
extern "C" fn on_resolve(
_: Ref,
_: u32,
interface: u32,
err: i32,
_: *const c_char,
host: *const c_char,
port: u16,
txt_len: u16,
txt: *const u8,
context: *mut c_void,
) {
if err != 0 {
return;
}
unsafe {
let out = &mut *(context as *mut Option<(u32, String, u16, Vec<u8>)>);
*out = Some((
interface,
CStr::from_ptr(host).to_string_lossy().into_owned(),
u16::from_be(port),
std::slice::from_raw_parts(txt, txt_len as usize).to_vec(),
));
}
}
fn resolve_raw(
flags: u32,
interface: u32,
name: &CStr,
regtype: &CStr,
domain: &CStr,
within: std::time::Duration,
) -> Option<Instance> {
let mut out: Option<(u32, String, u16, Vec<u8>)> = None;
let mut sd: Ref = std::ptr::null_mut();
unsafe {
if DNSServiceResolve(
&mut sd,
flags,
interface,
name.as_ptr(),
regtype.as_ptr(),
domain.as_ptr(),
on_resolve,
(&mut out as *mut Option<(u32, String, u16, Vec<u8>)>).cast(),
) != 0
{
return None;
}
let mut fds = [libc::pollfd {
fd: DNSServiceRefSockFD(sd),
events: libc::POLLIN,
revents: 0,
}];
if libc::poll(fds.as_mut_ptr(), 1, within.as_millis() as i32) > 0 {
DNSServiceProcessResult(sd);
}
DNSServiceRefDeallocate(sd);
}
let (interface, host, port, txt) = out?;
Some(Instance {
name: name.to_string_lossy().into_owned(),
interface,
host,
port,
txt,
})
}
type Resolved = Option<(String, u16, Option<String>, Option<String>, Option<String>)>;
fn resolve(s: &Seen) -> Resolved {
let i = resolve_raw(
0,
s.interface,
&s.name,
&s.regtype,
&s.domain,
std::time::Duration::from_secs(5),
)?;
Some((
i.host.clone(),
i.port,
i.txt("id"),
i.txt("platform"),
i.txt("server"),
))
}
const FLAG_WAKE_ON_RESOLVE: u32 = 0x40000;
pub fn instances(regtype: &str, within: std::time::Duration) -> Vec<Instance> {
let Ok(regtype) = CString::new(regtype) else {
return Vec::new();
};
let mut browsed = Browsed::default();
let mut sd: Ref = std::ptr::null_mut();
let err = unsafe {
DNSServiceBrowse(
&mut sd,
0,
0,
regtype.as_ptr(),
std::ptr::null(),
on_browse,
(&mut browsed as *mut Browsed).cast(),
)
};
if err != 0 {
return Vec::new();
}
let until = std::time::Instant::now() + within;
let fd = unsafe { DNSServiceRefSockFD(sd) };
loop {
let left = until.saturating_duration_since(std::time::Instant::now());
if left.is_zero() {
break;
}
let mut fds = [libc::pollfd {
fd,
events: libc::POLLIN,
revents: 0,
}];
if unsafe { libc::poll(fds.as_mut_ptr(), 1, left.as_millis() as i32) } <= 0 {
continue;
}
if unsafe { DNSServiceProcessResult(sd) } != 0 {
break;
}
if browsed.err != 0 {
log::info!("bonjour: browsing {regtype:?} failed ({})", browsed.err);
break;
}
}
unsafe { DNSServiceRefDeallocate(sd) };
let mut names = std::collections::HashSet::new();
browsed
.seen
.into_iter()
.filter(|s| s.add && names.insert(s.name.clone()))
.filter_map(|s| {
resolve_raw(
0,
s.interface,
&s.name,
&s.regtype,
&s.domain,
std::time::Duration::from_secs(2),
)
})
.collect()
}
pub fn resolve_named(
name: &str,
regtype: &str,
interface: u32,
within: std::time::Duration,
) -> Option<Instance> {
let name = CString::new(name).ok()?;
let regtype = CString::new(regtype).ok()?;
resolve_raw(0, interface, &name, ®type, c"local.", within)
}
pub fn wake_on_resolve(mac: &str, ip: &str, regtype: &str, interface: u32) {
let (Ok(name), Ok(regtype)) = (CString::new(format!("{mac}@{ip}")), CString::new(regtype))
else {
return;
};
if interface == 0 {
return;
}
resolve_raw(
FLAG_WAKE_ON_RESOLVE,
interface,
&name,
®type,
c"local.",
std::time::Duration::from_millis(300),
);
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_browse_error_lands_in_its_own_browse() {
let mut standing = Browsed::default();
let mut lookup = Browsed::default();
let browse = |browsed: &mut Browsed, err: i32| {
on_browse(
std::ptr::null_mut(),
FLAG_ADD,
0,
err,
c"tv".as_ptr(),
c"_airplay._tcp.".as_ptr(),
c"local.".as_ptr(),
(browsed as *mut Browsed).cast(),
);
};
browse(&mut lookup, POLICY_DENIED);
browse(&mut standing, 0);
assert_eq!(lookup.err, POLICY_DENIED);
assert!(lookup.seen.is_empty());
assert_eq!(standing.err, 0);
assert_eq!(standing.seen.len(), 1);
}
#[test]
fn a_txt_record_reads_back() {
let txt = txt_record(&[("id", "abc-123"), ("platform", "ios")]);
assert_eq!(txt_value(&txt, "id").as_deref(), Some("abc-123"));
assert_eq!(txt_value(&txt, "platform").as_deref(), Some("ios"));
assert_eq!(txt_value(&txt, "name"), None);
}
#[test]
fn a_txt_record_carries_the_server() {
let txt = txt_record(&[
("id", "abc"),
("platform", "ios"),
("server", "https://music.example.com"),
]);
assert_eq!(
txt_value(&txt, "server").as_deref(),
Some("https://music.example.com")
);
}
}
}
#[cfg(test)]
mod dial_tests {
use super::*;
#[test]
fn a_duplicate_backs_off_like_a_failure() {
let mut wait = RETRY_MIN;
for _ in 0..10 {
wait = next_wait(wait, false, false);
}
assert_eq!(wait, RETRY_MAX);
}
#[test]
fn a_connection_that_served_starts_over() {
assert_eq!(next_wait(RETRY_MAX, true, false), RETRY_MIN * 2);
}
#[test]
fn a_redial_skips_the_wait() {
assert_eq!(next_wait(RETRY_MAX, false, true), RETRY_MIN);
}
}
#[cfg(test)]
mod server_tests {
use super::is_server;
#[test]
fn a_device_announces_only_an_address_fit_to_offer() {
use super::announceable;
assert_eq!(
announceable("https://music.example.com/").as_deref(),
Some("https://music.example.com")
);
for url in [
"",
"https://user:secret@music.example.com",
"https://music.example.com/?u=me&p=secret",
"http://127.0.0.1:4799",
"http://localhost:4799",
] {
assert_eq!(announceable(url), None, "{url}");
}
let long = format!("https://music.example.com/{}", "a".repeat(240));
assert_eq!(announceable(&long), None);
}
#[test]
fn only_a_bare_web_address_is_offered() {
assert!(is_server("https://music.example.com"));
assert!(is_server("http://192.168.1.20:4533/koan"));
for s in [
"",
"music.example.com",
"ftp://music.example.com",
"javascript:alert(1)",
"https://user:secret@music.example.com",
"https://music.example.com/?u=me&p=secret",
"https://music.example.com/#p=secret",
"http://127.0.0.1:4799",
"http://localhost:4799",
"http://music.localhost",
"http://[::1]:4799",
"http://0.0.0.0:4799",
] {
assert!(!is_server(s), "{s}");
}
}
}
#[cfg(test)]
mod tests {
#[test]
fn a_device_on_the_network_is_told_the_outputs_only_under_full_control() {
let state = crate::remote::link::LinkState {
playing: true,
outputs: Some(Default::default()),
..Default::default()
};
let full = super::for_the_network(state.clone(), true);
assert!(full.playing && full.outputs.is_some());
let playback = super::for_the_network(state, false);
assert!(playback.playing);
assert_eq!(playback.outputs, None);
}
use super::*;
use crate::remote::link::LinkDeviceKey;
use crate::remote::wire::Session as _;
struct Rig {
_dir: tempfile::TempDir,
local: Local,
stop: Arc<Stop>,
ran: Arc<Mutex<Vec<(LinkCommand, CommandSource)>>>,
}
fn rig(keys: &[(&str, Option<&str>)]) -> Rig {
let dir = tempfile::tempdir().unwrap();
crate::config::set_config_dir(dir.path());
Config::persist(|c| {
c.remote.enabled = true;
c.remote.url = "http://koan.test".into();
c.remote.username = "jo".into();
c.remote.api_key = "key".into();
c.remote.device_key = proof::new_device_key().unwrap();
})
.unwrap();
let public = proof::public_key().unwrap();
proof::keep(
keys.iter()
.map(|(id, owner)| LinkDeviceKey {
id: (*id).into(),
key: public.clone(),
owner: owner.map(Into::into),
})
.collect(),
proof::account_of(&Config::cached()),
);
let ran = Arc::new(Mutex::new(Vec::new()));
let seen = ran.clone();
Rig {
_dir: dir,
local: Local {
identity: crate::remote::link::LinkIdentity {
name: "Mac".into(),
platform: "macos".into(),
device_id: "mac".into(),
},
state: Arc::new(LinkState::default),
on_command: Arc::new(move |cmd, source, _| seen.lock().push((cmd, source))),
},
stop: Stop::new().unwrap(),
ran,
}
}
fn serving<'a>(r: &'a Rig) -> Serving<'a> {
Serving {
local: &r.local,
stop: &r.stop,
greeted: false,
sent: None,
waker: Waker::new().unwrap(),
levels: None,
answers: Default::default(),
nonce: proof::nonce(),
answered: false,
proven: None,
reports: None,
pending: Vec::new(),
listed: None,
}
}
fn json(f: &ProofFrame) -> String {
frame(f).unwrap()
}
fn auth(s: &mut Serving, dialer: &str) -> (String, proof::Session) {
auth_with(s, dialer, proof::nonce().unwrap())
}
fn auth_with(s: &mut Serving, dialer: &str, dial_nonce: String) -> (String, proof::Session) {
let listen_nonce = s.nonce.clone().unwrap();
let sig = proof::sign_dial("mac", dialer, &listen_nonce, &dial_nonce).unwrap();
let session = proof::Session::new("mac", dialer, &listen_nonce, &dial_nonce);
s.incoming(&json(&ProofFrame::Auth {
id: dialer.into(),
nonce: dial_nonce.clone(),
sig,
}));
(dial_nonce, session)
}
fn signed(session: &mut proof::Session, cmd: &LinkCommand) -> String {
let command = serde_json::to_string(cmd).unwrap();
let (seq, sig) = session.sign(&command).unwrap();
json(&ProofFrame::Signed { seq, sig, command })
}
fn answer(s: &mut Serving) -> (bool, Option<String>) {
let out = wire::Session::outgoing(s);
out.iter()
.find_map(|t| match serde_json::from_str::<ProofFrame>(t) {
Ok(ProofFrame::Proof { verified, sig }) => Some((verified, sig)),
_ => None,
})
.expect("an answer")
}
#[test]
fn an_own_device_that_proves_itself_runs_with_the_accounts_powers() {
let _guard = crate::config::tests::PERSIST_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let r = rig(&[("phone", None), ("mac", None)]);
let mut s = serving(&r);
let hello = wire::Session::outgoing(&mut s);
let listen_nonce = s.nonce.clone().unwrap();
assert!(
hello[0].contains(&listen_nonce),
"the Hello carries the nonce"
);
let (dial_nonce, mut session) = auth(&mut s, "phone");
let (verified, sig) = answer(&mut s);
assert!(verified);
let proven = proof::verify_listen(
"phone",
"mac",
&dial_nonce,
&listen_nonce,
true,
&sig.unwrap(),
);
assert_eq!(proven.map(|p| p.peer), Some(proof::Peer::Own));
let sync = LinkCommand::Sync { full: true };
let frame = signed(&mut session, &sync);
s.incoming(&frame);
s.incoming(&frame);
s.incoming(&serde_json::to_string(&LinkCommand::Pause).unwrap());
s.incoming(&signed(
&mut session,
&LinkCommand::DeviceKeys { keys: vec![] },
));
s.incoming(&signed(&mut session, &LinkCommand::Next));
let ran = r.ran.lock();
assert_eq!(
*ran,
vec![
(sync, CommandSource::Account),
(LinkCommand::Next, CommandSource::Account),
]
);
}
#[test]
fn a_listener_signs_its_reports_when_both_ends_do() {
let _guard = crate::config::tests::PERSIST_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let r = rig(&[("phone", None), ("mac", None)]);
let mut s = serving(&r);
wire::Session::outgoing(&mut s);
let listen_nonce = s.nonce.clone().unwrap();
let (dial_nonce, _) = auth(&mut s, "phone");
let out = wire::Session::outgoing(&mut s);
let Ok(ProofFrame::Proof { sig, .. }) = serde_json::from_str(&out[0]) else {
panic!("the proof goes first: {out:?}");
};
let mac = proof::verify_listen(
"phone",
"mac",
&dial_nonce,
&listen_nonce,
true,
&sig.unwrap(),
)
.unwrap();
let mut session = proof::Session::reports("mac", "phone", &listen_nonce, &dial_nonce);
let reports: Vec<String> = out[1..]
.iter()
.map(|t| match serde_json::from_str(t) {
Ok(ProofFrame::SignedReport { seq, sig, report }) => {
assert!(session.accept(&mac, seq, &sig, &report));
report
}
_ => panic!("unsigned: {t}"),
})
.collect();
assert!(
reports.iter().any(|r| r.contains(r#""type":"state""#)),
"the state is sent again, signed: {reports:?}"
);
}
#[test]
fn a_listener_reports_unsigned_to_an_older_dialler() {
let _guard = crate::config::tests::PERSIST_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let r = rig(&[("phone", None), ("mac", None)]);
let mut s = serving(&r);
wire::Session::outgoing(&mut s);
auth_with(&mut s, "phone", "b2xkZXI=".into());
assert!(answer(&mut s).0);
s.sent = None;
let out = wire::Session::outgoing(&mut s);
assert!(!out.is_empty());
assert!(
out.iter()
.all(|t| serde_json::from_str::<LinkReport>(t).is_ok()),
"{out:?}"
);
}
fn controlling<'a>(stop: &'a Arc<Stop>) -> Controlling<'a> {
Controlling {
stop,
waker: Waker::new().unwrap(),
key: "id:mac",
addr: "10.0.0.2:7979",
id: None,
this_device: false,
duplicate: false,
handshake: Handshake::Plain,
proven: None,
reports: None,
pending: Vec::new(),
}
}
fn answered(c: &mut Controlling, listen_nonce: &str, dial_nonce: &str) {
c.handshake = Handshake::Awaiting {
listener: "mac".into(),
me: "phone".into(),
listen_nonce: listen_nonce.into(),
dial_nonce: dial_nonce.into(),
};
let sig = proof::sign_listen("phone", "mac", dial_nonce, listen_nonce, true);
c.answered(true, sig);
}
#[test]
fn a_dialler_reads_only_reports_signed_for_its_connection() {
let _guard = crate::config::tests::PERSIST_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let _r = rig(&[("phone", None), ("mac", None)]);
let stop = Stop::new().unwrap();
let mut c = controlling(&stop);
let (listen_nonce, dial_nonce) = (proof::nonce().unwrap(), proof::nonce().unwrap());
answered(&mut c, &listen_nonce, &dial_nonce);
assert_eq!(c.proven, Some(Peer::Own));
let state = serde_json::to_string(&LinkReport::State(LinkState::default())).unwrap();
assert!(c.checked(&state).is_none(), "unsigned");
let mut session = proof::Session::reports("mac", "phone", &listen_nonce, &dial_nonce);
let (seq, sig) = session.sign(&state).unwrap();
let signed = json(&ProofFrame::SignedReport {
seq,
sig: sig.clone(),
report: state.clone(),
});
assert_eq!(c.checked(&signed).as_deref(), Some(state.as_str()));
assert!(c.checked(&signed).is_none(), "replayed");
let forged = serde_json::to_string(&LinkReport::Ack {
ack: 1,
outcome: AckOutcome::Done,
})
.unwrap();
let altered = json(&ProofFrame::SignedReport {
seq: seq + 1,
sig,
report: forged,
});
assert!(c.checked(&altered).is_none(), "altered");
let mut other = proof::Session::reports("mac", "phone", &dial_nonce, &listen_nonce);
let (seq, sig) = other.sign(&state).unwrap();
let elsewhere = json(&ProofFrame::SignedReport {
seq: seq + 5,
sig,
report: state.clone(),
});
assert!(c.checked(&elsewhere).is_none(), "another session");
}
#[test]
fn an_account_device_that_proves_nothing_is_not_mirrored() {
let _store = crate::remote::devices::tests::STORE_LOCK.lock();
let _guard = crate::config::tests::PERSIST_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let _r = rig(&[("phone", None), ("mac", None)]);
let id = "unproven-mac";
devices::set_account(vec![crate::remote::link::LinkDevice {
id: id.into(),
name: id.into(),
platform: "macos".into(),
linked: true,
state: None,
last_seen: None,
wakeable: None,
owner: None,
acks: false,
}]);
let stop = Stop::new().unwrap();
let mut c = controlling(&stop);
let hello = LinkHello {
id: id.into(),
name: id.into(),
platform: "macos".into(),
library: None,
acks: true,
nonce: None,
};
c.incoming(&serde_json::to_string(&LinkReport::Hello(hello)).unwrap());
assert_eq!(c.id.as_deref(), Some(id));
let forged = LinkState {
playing: true,
..Default::default()
};
c.incoming(&serde_json::to_string(&LinkReport::State(forged)).unwrap());
assert_eq!(devices::last_report(id), None);
CONNS.lock().as_mut().unwrap().remove(id);
devices::set_account(Vec::new());
}
#[test]
fn a_dialler_reads_an_older_listeners_reports_unsigned() {
let _guard = crate::config::tests::PERSIST_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let _r = rig(&[("phone", None), ("mac", None)]);
let stop = Stop::new().unwrap();
let mut c = controlling(&stop);
answered(&mut c, "b2xkZXI=", &proof::nonce().unwrap());
assert_eq!(c.proven, Some(Peer::Own));
let state = serde_json::to_string(&LinkReport::State(LinkState::default())).unwrap();
assert_eq!(c.checked(&state).as_deref(), Some(state.as_str()));
}
#[test]
fn a_disconnected_device_is_held_off_until_reached_for() {
let _store = crate::remote::devices::tests::STORE_LOCK.lock();
let addr: std::net::SocketAddr = "[::ffff:192.0.2.77]:50000".parse().unwrap();
let waker = Waker::new().unwrap();
let (listed, ended) = Listed::new(&addr, &waker);
listed.said("held-phone", None);
let key = listed.0;
let session = sessions().into_iter().find(|s| s.key == Some(key)).unwrap();
assert_eq!(session.addr, "192.0.2.77");
assert!(session.inbound);
end(key);
assert!(ended.load(Ordering::Relaxed));
assert!(held_addr(&addr.ip()));
assert!(held_id("held-phone"));
drop(listed);
assert!(sessions().iter().all(|s| s.key != Some(key)));
assert!(held().contains(&("192.0.2.77".into(), None)));
release("held-phone");
assert!(!held_id("held-phone"));
assert!(!held_addr(&addr.ip()));
let (listed, _) = Listed::new(&addr, &waker);
end(listed.0);
drop(listed);
release("held-phone");
assert!(held_addr(&addr.ip()));
let (listed, _) = Listed::new(&addr, &waker);
end(listed.0);
end(listed.0);
drop(listed);
assert_eq!(held().iter().filter(|(a, _)| a == "192.0.2.77").count(), 1);
release_addr("192.0.2.77");
assert!(!held_addr(&addr.ip()));
}
#[test]
fn a_second_proof_on_one_connection_is_ignored() {
let _guard = crate::config::tests::PERSIST_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let r = rig(&[("phone", Some("kim"))]);
let mut s = serving(&r);
let (_, _) = auth(&mut s, "stranger");
assert!(!answer(&mut s).0);
let (_, mut session) = auth(&mut s, "phone");
assert!(
wire::Session::outgoing(&mut s)
.iter()
.all(|t| !t.contains("nearbyProof")),
"no second answer"
);
s.incoming(&signed(&mut session, &LinkCommand::Pause));
assert!(r.ran.lock().is_empty(), "still unproven");
}
#[test]
fn a_shared_device_gets_the_playback_set_as_its_account() {
let _guard = crate::config::tests::PERSIST_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let r = rig(&[("phone", Some("kim"))]);
let mut s = serving(&r);
let (_, mut session) = auth(&mut s, "phone");
assert!(answer(&mut s).0);
s.incoming(&signed(&mut session, &LinkCommand::Sync { full: false }));
s.incoming(&signed(&mut session, &LinkCommand::Pause));
assert_eq!(
*r.ran.lock(),
vec![(LinkCommand::Pause, CommandSource::Shared)]
);
}
#[test]
fn a_peer_that_proves_nothing_is_trusted_as_the_network_is() {
let _guard = crate::config::tests::PERSIST_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let r = rig(&[("mac", None)]);
Config::persist(|c| c.devices.nearby_control = crate::config::NearbyControl::Playback)
.unwrap();
let mut s = serving(&r);
let (_, mut session) = auth(&mut s, "phone");
assert!(!answer(&mut s).0);
s.incoming(&signed(&mut session, &LinkCommand::Pause));
s.incoming(&serde_json::to_string(&LinkCommand::Sync { full: false }).unwrap());
s.incoming(&serde_json::to_string(&LinkCommand::Pause).unwrap());
assert_eq!(
*r.ran.lock(),
vec![(LinkCommand::Pause, CommandSource::Stranger)]
);
}
#[test]
fn a_peer_that_never_asks_is_trusted_as_before() {
let _guard = crate::config::tests::PERSIST_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let r = rig(&[("phone", None)]);
let mut s = serving(&r);
s.incoming(&serde_json::to_string(&LinkCommand::Pause).unwrap());
assert_eq!(r.ran.lock().len(), 1);
assert_ne!(r.ran.lock()[0].1, CommandSource::Account);
}
}
#[cfg(test)]
mod admit_tests {
use super::*;
use crate::remote::acks;
#[test]
fn a_stranger_cannot_spend_an_id_ahead_of_the_real_command() {
let id = acks::next_id();
let refused = std::sync::Arc::new(parking_lot::Mutex::new(None));
let got = refused.clone();
let forged = Envelope {
command: LinkCommand::Sync { full: true },
ack: Some(id),
};
assert!(matches!(
admit(
forged,
|cmd| cmd.from_the_network(false),
move |_, o| *got.lock() = Some(o)
),
Admitted::Neither
));
assert!(matches!(*refused.lock(), Some(AckOutcome::Refused { .. })));
let real = Envelope {
command: LinkCommand::Pause,
ack: Some(id),
};
assert!(matches!(
admit(real, |cmd| cmd.from_the_network(false), |_, _| {}),
Admitted::Command(LinkCommand::Pause, _, Some(_))
));
}
}
#[cfg(test)]
mod proof_tests {
use super::*;
use crate::remote::proof::Peer;
#[test]
fn a_listed_devices_name_needs_its_proof() {
assert!(proven_as_listed(None, None));
assert!(proven_as_listed(Some(None), Some(&Peer::Own)));
assert!(!proven_as_listed(Some(None), None));
assert!(!proven_as_listed(
Some(None),
Some(&Peer::Shared("b".into()))
));
let shared = || Some(Some("b".to_string()));
assert!(proven_as_listed(shared(), Some(&Peer::Shared("b".into()))));
assert!(!proven_as_listed(shared(), Some(&Peer::Shared("c".into()))));
assert!(!proven_as_listed(shared(), Some(&Peer::Own)));
assert!(!proven_as_listed(shared(), None));
}
#[test]
fn a_listed_device_is_sent_nothing_until_it_proves_it_is_that_device() {
let id = "impostor-of-the-phone";
let waker = Waker::new().unwrap();
CONNS.lock().get_or_insert_with(HashMap::new).insert(
id.into(),
Conn {
outbox: Vec::new(),
waker,
proven: None,
addr: String::new(),
since: 0,
},
);
let outbox = || CONNS.lock().as_ref().unwrap()[id].outbox.len();
assert!(!queue(id, Some(None), LinkCommand::Pause.into()));
assert_eq!(outbox(), 0);
assert!(queue(id, None, LinkCommand::Pause.into()));
assert_eq!(outbox(), 1);
CONNS.lock().as_mut().unwrap().get_mut(id).unwrap().proven = Some(Peer::Own);
assert!(queue(id, Some(None), LinkCommand::Pause.into()));
assert_eq!(outbox(), 2);
CONNS.lock().as_mut().unwrap().remove(id);
}
}