use std::{
net::{Ipv4Addr, SocketAddrV4},
num::NonZeroU16,
sync::Arc,
time::{Duration, Instant},
};
use current_mapping::CurrentMapping;
use futures_lite::StreamExt;
use n0_error::{e, stack_error};
use netwatch::interfaces::HomeRouter;
use tokio::sync::{mpsc, oneshot, watch};
use tokio_util::task::AbortOnDropHandle;
use tracing::{Instrument, debug, info_span, trace};
mod current_mapping;
mod mapping;
mod metrics;
mod nat_pmp;
mod pcp;
mod upnp;
mod util;
mod defaults {
use std::time::Duration;
pub(crate) const UPNP_SEARCH_TIMEOUT: Duration = Duration::from_secs(1);
pub(crate) const PCP_RECV_TIMEOUT: Duration = Duration::from_millis(500);
pub(crate) const NAT_PMP_RECV_TIMEOUT: Duration = Duration::from_millis(500);
}
pub use metrics::Metrics;
const AVAILABILITY_TRUST_DURATION: Duration = Duration::from_secs(60 * 10);
const SERVICE_CHANNEL_CAPACITY: usize = 32;
const UNAVAILABILITY_TRUST_DURATION: Duration = Duration::from_secs(5);
#[derive(Debug, Clone, PartialEq, Eq, derive_more::Display)]
#[display("portmap={{ UPnP: {upnp}, PMP: {nat_pmp}, PCP: {pcp} }}")]
pub struct ProbeOutput {
pub upnp: bool,
pub pcp: bool,
pub nat_pmp: bool,
}
impl ProbeOutput {
pub fn all_available(&self) -> bool {
self.upnp && self.pcp && self.nat_pmp
}
}
#[allow(missing_docs)]
#[stack_error(derive, add_meta)]
#[derive(Clone)]
#[non_exhaustive]
pub enum ProbeError {
#[error("Mapping channel is full")]
ChannelFull,
#[error("Mapping channel is closed")]
ChannelClosed,
#[error("No gateway found for probe")]
NoGateway,
#[error("gateway found is ipv6, ignoring")]
Ipv6Gateway,
#[error("Probe task stopped. is_panic: {is_panic}, is_cancelled: {is_cancelled}")]
Join { is_panic: bool, is_cancelled: bool },
}
#[derive(derive_more::Debug)]
enum Message {
ProcureMapping,
UpdateLocalPort { local_port: Option<NonZeroU16> },
Probe {
#[debug("_")]
result_tx: oneshot::Sender<Result<ProbeOutput, ProbeError>>,
},
}
#[derive(Debug, Clone, Copy)]
pub enum Protocol {
Udp,
Tcp,
}
#[derive(Debug, Clone)]
pub struct Config {
pub enable_upnp: bool,
pub enable_pcp: bool,
pub enable_nat_pmp: bool,
pub protocol: Protocol,
}
impl Default for Config {
fn default() -> Self {
Config {
enable_upnp: true,
enable_pcp: true,
enable_nat_pmp: true,
protocol: Protocol::Udp,
}
}
}
#[derive(Debug, Clone)]
pub struct Client {
port_mapping: watch::Receiver<Option<SocketAddrV4>>,
service_tx: mpsc::Sender<Message>,
metrics: Arc<Metrics>,
_service_handle: std::sync::Arc<AbortOnDropHandle<()>>,
}
impl Default for Client {
fn default() -> Self {
Self::new(Config::default())
}
}
impl Client {
pub fn new(config: Config) -> Self {
Self::with_metrics(config, Default::default())
}
pub fn with_metrics(config: Config, metrics: Arc<Metrics>) -> Self {
let (service_tx, service_rx) = mpsc::channel(SERVICE_CHANNEL_CAPACITY);
let (service, watcher) = Service::new(config, service_rx, metrics.clone());
let handle = AbortOnDropHandle::new(tokio::spawn(
async move { service.run().await }.instrument(info_span!("portmapper.service")),
));
Client {
port_mapping: watcher,
service_tx,
metrics,
_service_handle: std::sync::Arc::new(handle),
}
}
pub fn probe(&self) -> oneshot::Receiver<Result<ProbeOutput, ProbeError>> {
let (result_tx, result_rx) = oneshot::channel();
if let Err(e) = self.service_tx.try_send(Message::Probe { result_tx }) {
use mpsc::error::TrySendError::*;
let (result_tx, e) = match e {
Full(Message::Probe { result_tx }) => (result_tx, e!(ProbeError::ChannelFull)),
Closed(Message::Probe { result_tx }) => (result_tx, e!(ProbeError::ChannelClosed)),
Full(_) | Closed(_) => unreachable!("Sent value is a probe."),
};
if let Err(Err(e)) = result_tx.send(Err(e)) {
trace!("Failed to request probe: {e}")
}
}
result_rx
}
pub fn procure_mapping(&self) {
if let Err(e) = self.service_tx.try_send(Message::ProcureMapping) {
trace!("Failed to request mapping {e}")
}
}
pub fn update_local_port(&self, local_port: NonZeroU16) {
let local_port = Some(local_port);
if let Err(e) = self
.service_tx
.try_send(Message::UpdateLocalPort { local_port })
{
trace!("Failed to update local port {e}")
}
}
pub fn deactivate(&self) {
if let Err(e) = self
.service_tx
.try_send(Message::UpdateLocalPort { local_port: None })
{
trace!("Failed to deactivate port mapping {e}")
}
}
pub fn watch_external_address(&self) -> watch::Receiver<Option<SocketAddrV4>> {
self.port_mapping.clone()
}
pub fn metrics(&self) -> &Arc<Metrics> {
&self.metrics
}
}
#[derive(Debug)]
struct Probe {
last_probe: Instant,
last_upnp_gateway_addr: Option<(upnp::Gateway, Instant)>,
last_pcp: Option<Instant>,
last_nat_pmp: Option<Instant>,
}
impl Probe {
fn empty() -> Self {
Self {
last_probe: Instant::now(),
last_upnp_gateway_addr: None,
last_pcp: None,
last_nat_pmp: None,
}
}
async fn from_output(
config: Config,
output: ProbeOutput,
local_ip: Ipv4Addr,
gateway: Ipv4Addr,
metrics: Arc<Metrics>,
) -> Probe {
let ProbeOutput { upnp, pcp, nat_pmp } = output;
let Config {
enable_upnp,
enable_pcp,
enable_nat_pmp,
protocol: _,
} = config;
let mut upnp_probing_task = util::MaybeFuture {
inner: (enable_upnp && !upnp).then(|| {
let metrics = metrics.clone();
Box::pin(async move {
upnp::probe_available(&metrics)
.await
.map(|addr| (addr, Instant::now()))
})
}),
};
let mut pcp_probing_task = util::MaybeFuture {
inner: (enable_pcp && !pcp).then(|| {
let metrics = metrics.clone();
Box::pin(async move {
metrics.pcp_probes.inc();
pcp::probe_available(local_ip, gateway)
.await
.then(Instant::now)
})
}),
};
let mut nat_pmp_probing_task = util::MaybeFuture {
inner: (enable_nat_pmp && !nat_pmp).then(|| {
Box::pin(async {
nat_pmp::probe_available(local_ip, gateway)
.await
.then(Instant::now)
})
}),
};
if upnp_probing_task.inner.is_some() {
metrics.upnp_probes.inc();
}
let mut upnp_done = upnp_probing_task.inner.is_none();
let mut pcp_done = pcp_probing_task.inner.is_none();
let mut nat_pmp_done = nat_pmp_probing_task.inner.is_none();
let mut probe = Probe::empty();
while !upnp_done || !pcp_done || !nat_pmp_done {
tokio::select! {
last_upnp_gateway_addr = &mut upnp_probing_task, if !upnp_done => {
trace!("tick: upnp probe ready");
probe.last_upnp_gateway_addr = last_upnp_gateway_addr;
upnp_done = true;
},
last_nat_pmp = &mut nat_pmp_probing_task, if !nat_pmp_done => {
trace!("tick: nat_pmp probe ready");
probe.last_nat_pmp = last_nat_pmp;
nat_pmp_done = true;
},
last_pcp = &mut pcp_probing_task, if !pcp_done => {
trace!("tick: pcp probe ready");
probe.last_pcp = last_pcp;
pcp_done = true;
},
}
}
probe
}
fn output(&self) -> ProbeOutput {
let now = Instant::now();
let upnp = self
.last_upnp_gateway_addr
.as_ref()
.map(|(_gateway_addr, last_probed)| *last_probed + AVAILABILITY_TRUST_DURATION > now)
.unwrap_or_default();
let pcp = self
.last_pcp
.as_ref()
.map(|last_probed| *last_probed + AVAILABILITY_TRUST_DURATION > now)
.unwrap_or_default();
let nat_pmp = self
.last_nat_pmp
.as_ref()
.map(|last_probed| *last_probed + AVAILABILITY_TRUST_DURATION > now)
.unwrap_or_default();
ProbeOutput { upnp, pcp, nat_pmp }
}
fn update(&mut self, probe: Probe, metrics: &Arc<Metrics>) {
let Probe {
last_probe,
last_upnp_gateway_addr,
last_pcp,
last_nat_pmp,
} = probe;
if last_upnp_gateway_addr.is_some() {
metrics.upnp_available.inc();
let new_gateway = last_upnp_gateway_addr
.as_ref()
.map(|(addr, _last_seen)| addr);
let old_gateway = self
.last_upnp_gateway_addr
.as_ref()
.map(|(addr, _last_seen)| addr);
if new_gateway != old_gateway {
metrics.upnp_gateway_updated.inc();
debug!(
"upnp gateway changed {:?} -> {:?}",
old_gateway
.map(|gw| gw.to_string())
.unwrap_or("None".into()),
new_gateway
.map(|gw| gw.to_string())
.unwrap_or("None".into())
)
};
self.last_upnp_gateway_addr = last_upnp_gateway_addr;
}
if last_pcp.is_some() {
metrics.pcp_available.inc();
self.last_pcp = last_pcp;
}
if last_nat_pmp.is_some() {
self.last_nat_pmp = last_nat_pmp;
}
self.last_probe = last_probe;
}
}
type ProbeResult = Result<ProbeOutput, ProbeError>;
#[derive(Debug)]
pub struct Service {
config: Config,
local_port: Option<NonZeroU16>,
rx: mpsc::Receiver<Message>,
current_mapping: CurrentMapping,
full_probe: Probe,
mapping_task: Option<AbortOnDropHandle<Result<mapping::Mapping, mapping::Error>>>,
probing_task: Option<(AbortOnDropHandle<Probe>, Vec<oneshot::Sender<ProbeResult>>)>,
metrics: Arc<Metrics>,
}
impl Service {
fn new(
config: Config,
rx: mpsc::Receiver<Message>,
metrics: Arc<Metrics>,
) -> (Self, watch::Receiver<Option<SocketAddrV4>>) {
let (current_mapping, watcher) = CurrentMapping::new(metrics.clone());
let mut full_probe = Probe::empty();
if let Some(in_the_past) = full_probe
.last_probe
.checked_sub(AVAILABILITY_TRUST_DURATION)
{
full_probe.last_probe = in_the_past;
}
let service = Service {
config,
local_port: None,
rx,
current_mapping,
full_probe,
mapping_task: None,
probing_task: None,
metrics,
};
(service, watcher)
}
async fn invalidate_mapping(&mut self) {
if let Some(old_mapping) = self.current_mapping.update(None)
&& let Err(e) = old_mapping.release().await
{
debug!("failed to release mapping {e}");
}
}
async fn run(mut self) {
debug!("portmap starting");
loop {
tokio::select! {
msg = self.rx.recv() => {
trace!("tick: msg {msg:?}");
match msg {
Some(msg) => {
self.handle_msg(msg).await;
},
None => {
debug!("portmap service channel dropped. Likely shutting down.");
break;
}
}
}
mapping_result = util::MaybeFuture{ inner: self.mapping_task.as_mut() } => {
trace!("tick: mapping ready");
self.mapping_task = None;
self.on_mapping_result(mapping_result);
}
probe_result = util::MaybeFuture{ inner: self.probing_task.as_mut().map(|(fut, _rec)| fut) } => {
trace!("tick: probe ready");
let receivers = self.probing_task.take().expect("is some").1;
let probe_result = probe_result.map_err(|e| e!(ProbeError::Join { is_panic: e.is_panic(), is_cancelled: e.is_cancelled() }));
self.on_probe_result(probe_result, receivers);
}
Some(event) = self.current_mapping.next() => {
trace!("tick: mapping event {event:?}");
match event {
current_mapping::Event::Renew { external_ip, external_port } | current_mapping::Event::Expired { external_ip, external_port } => {
self.get_mapping(Some((external_ip, external_port)));
},
}
}
}
}
}
fn on_probe_result(
&mut self,
result: Result<Probe, ProbeError>,
receivers: Vec<oneshot::Sender<ProbeResult>>,
) {
let result = result.map(|probe| {
self.full_probe.update(probe, &self.metrics);
let output = self.full_probe.output();
trace!(?output, "probe output");
output
});
for tx in receivers {
let _ = tx.send(result.clone());
}
}
fn on_mapping_result(
&mut self,
result: Result<Result<mapping::Mapping, mapping::Error>, tokio::task::JoinError>,
) {
match result {
Ok(Ok(mapping)) => {
self.current_mapping.update(Some(mapping));
}
Ok(Err(e)) => {
debug!("failed to get a port mapping {e}");
self.metrics.mapping_failures.inc();
}
Err(e) => {
debug!("failed to get a port mapping {e}");
self.metrics.mapping_failures.inc();
}
}
}
async fn handle_msg(&mut self, msg: Message) {
match msg {
Message::ProcureMapping => self.update_local_port(self.local_port).await,
Message::UpdateLocalPort { local_port } => self.update_local_port(local_port).await,
Message::Probe { result_tx } => self.probe_request(result_tx),
}
}
async fn update_local_port(&mut self, local_port: Option<NonZeroU16>) {
if local_port != self.local_port {
self.metrics.local_port_updates.inc();
let old_port = std::mem::replace(&mut self.local_port, local_port);
let dropped_task = self.mapping_task.take();
let did_cancel = dropped_task
.map(|task| !task.is_finished())
.unwrap_or_default();
if did_cancel {
debug!(
"canceled mapping task due to local port update. Old: {:?} New: {:?}",
old_port, self.local_port
)
}
let external_addr = self.current_mapping.external();
if external_addr.is_some() {
self.invalidate_mapping().await;
}
self.get_mapping(external_addr);
} else if self.current_mapping.external().is_none() {
self.get_mapping(None)
}
}
fn get_mapping(&mut self, external_addr: Option<(Ipv4Addr, NonZeroU16)>) {
if let Some(local_port) = self.local_port {
self.metrics.mapping_attempts.inc();
let (local_ip, gateway) = match ip_and_gateway() {
Ok(ip_and_gw) => ip_and_gw,
Err(e) => return debug!("can't get mapping: {e}"),
};
let ProbeOutput { upnp, pcp, nat_pmp } = self.full_probe.output();
debug!("getting a port mapping for {local_ip}:{local_port} -> {external_addr:?}");
let recently_probed =
self.full_probe.last_probe + UNAVAILABILITY_TRUST_DURATION > Instant::now();
let protocol = self.config.protocol;
self.mapping_task = if pcp {
let task = mapping::Mapping::new_pcp(
protocol,
local_ip,
local_port,
gateway,
external_addr,
);
Some(AbortOnDropHandle::new(tokio::spawn(
task.instrument(info_span!("pcp")),
)))
} else if nat_pmp {
let task = mapping::Mapping::new_nat_pmp(
protocol,
local_ip,
local_port,
gateway,
external_addr,
);
Some(AbortOnDropHandle::new(tokio::spawn(
task.instrument(info_span!("pmp")),
)))
} else if upnp || self.config.enable_upnp {
let external_port = external_addr.map(|(_addr, port)| port);
let gateway = self
.full_probe
.last_upnp_gateway_addr
.as_ref()
.map(|(gateway, _last_seen)| gateway.clone());
let task = mapping::Mapping::new_upnp(
protocol,
local_ip,
local_port,
gateway,
external_port,
);
Some(AbortOnDropHandle::new(tokio::spawn(
task.instrument(info_span!("upnp")),
)))
} else if !recently_probed && self.config.enable_pcp {
let task = mapping::Mapping::new_pcp(
protocol,
local_ip,
local_port,
gateway,
external_addr,
);
Some(AbortOnDropHandle::new(tokio::spawn(
task.instrument(info_span!("pcp")),
)))
} else if !recently_probed && self.config.enable_nat_pmp {
let task = mapping::Mapping::new_nat_pmp(
protocol,
local_ip,
local_port,
gateway,
external_addr,
);
Some(AbortOnDropHandle::new(tokio::spawn(
task.instrument(info_span!("pmp")),
)))
} else {
return;
}
}
}
fn probe_request(&mut self, result_tx: oneshot::Sender<Result<ProbeOutput, ProbeError>>) {
match self.probing_task.as_mut() {
Some((_task_handle, receivers)) => receivers.push(result_tx),
None => {
let probe_output = self.full_probe.output();
if probe_output.all_available() {
let _ = result_tx.send(Ok(probe_output));
} else {
self.metrics.probes_started.inc();
let (local_ip, gateway) = match ip_and_gateway() {
Ok(ip_and_gw) => ip_and_gw,
Err(e) => {
debug!("could not start probe: {e}");
let _ = result_tx.send(Err(e));
return;
}
};
let config = self.config.clone();
let metrics = self.metrics.clone();
let handle = tokio::spawn(
async move {
Probe::from_output(config, probe_output, local_ip, gateway, metrics)
.await
}
.instrument(info_span!("portmapper.probe")),
);
let receivers = vec![result_tx];
self.probing_task = Some((AbortOnDropHandle::new(handle), receivers));
}
}
}
}
}
fn ip_and_gateway() -> Result<(Ipv4Addr, Ipv4Addr), ProbeError> {
let Some(HomeRouter { gateway, my_ip }) = HomeRouter::new() else {
return Err(e!(ProbeError::NoGateway));
};
let local_ip = match my_ip {
Some(std::net::IpAddr::V4(ip))
if !ip.is_unspecified() && !ip.is_loopback() && !ip.is_multicast() =>
{
ip
}
other => {
debug!("no address suitable for port mapping found ({other:?}), using localhost");
Ipv4Addr::LOCALHOST
}
};
let std::net::IpAddr::V4(gateway) = gateway else {
return Err(e!(ProbeError::Ipv6Gateway));
};
Ok((local_ip, gateway))
}