ts_capability_version, ts_control*, ts_runtime, ts_nodecapability: handle PeerChange, OnlineStatus, PeerSeen messages

Applies peer updates from MapResponse::{peers_changed_patch, peer_seen_change,
online_change} to the peer tracker's `Node` state. Introduces a number of
types to ease handling of patch/delta updates, especially concerning online
and last-seen timestamps. Updates order that delta updates are applied to match
the Golang codebase.

Signed-off-by: Dylan Bargatze <dylan@tailscale.com>

Author: dylan-tailscale

ts_capabilityversion/src/lib.rs

@@ -16,7 +16,7 @@ use core::fmt;
 ///
 /// Note: Prior to 2022-03-06, this value was known as the "`MapRequest` version", `mapVer`, or "map
 /// cap"; you'll still see that name used in comments throughout the Golang codebase.
-#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd)]
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
 pub struct CapabilityVersion(u16);
 

ts_control/Cargo.toml

@@ -28,7 +28,7 @@ ts_transport_derp.workspace = true
 
 # Unconditionally required dependencies.
 bytes.workspace = true
-chrono = { workspace = true, features = ["serde"] }
+chrono = { workspace = true, features = ["clock", "serde"] }
 gethostname.workspace = true
 ipnet = { workspace = true, features = ["serde"] }
 lazy_static.workspace = true

ts_control/src/lib.rs

@@ -28,7 +28,9 @@ pub use config::{Config, DEFAULT_CONTROL_SERVER};
 pub use control_dialer::{ControlDialer, TcpDialer, complete_connection};
 pub use derp::{Map as DerpMap, Region as DerpRegion, convert_derp_map};
 pub use dial_plan::{DialCandidate, DialMode, DialPlan};
-pub use node::{Id as NodeId, Node, StableId as StableNodeId, TailnetAddress};
+pub use node::{
+    Id as NodeId, Node, NodeStatus, NodeUpdate, StableId as StableNodeId, TailnetAddress,
+};
 
 #[cfg(feature = "async_tokio")]
 pub use crate::tokio::{AsyncControlClient, FilterUpdate, PeerUpdate, StateUpdate};

ts_control/src/node.rs

@@ -1,6 +1,11 @@
-use core::net::{IpAddr, SocketAddr};
+use alloc::collections::BTreeMap;
+use core::{
+    fmt,
+    net::{IpAddr, SocketAddr},
+};
 
 use chrono::{DateTime, Utc};
+use ts_capabilityversion::CapabilityVersion;
 use ts_keys::{DiscoPublicKey, MachinePublicKey, NodePublicKey};
 
 /// The unique id of a node.
@@ -10,20 +15,125 @@ pub type Id = i64;
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct StableId(pub String);
 
+/// Timestamp when an offline node was last connected to the control plane. Only applies to nodes
+/// that are offline (disconnected from the control plane).
+///
+/// Some timestamp values are only accurate to a resolution of ~10 minutes for privacy reasons, and
+/// do not take into account differences between the control plane clock and local device clock. See
+/// each variant for more information.
+#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
+pub enum NodeLastSeen {
+    /// The last time the node was connected to the control plane, as reported by the control plane
+    /// itself.
+    ///
+    /// The timestamp value may be rounded to the nearest 10-minute boundary before being reported
+    /// to this device by the control plane; in other words, many `Control` timestamps will be
+    /// aligned to 10-minute boundaries. This is a privacy-preserving measure implemented by the
+    /// control plane. Note that we have seen some non-rounded timestamps being reported from the
+    /// control plane, although it's unclear if this is a bug or by design.
+    ///
+    /// Note that the timestamp value was generated by the control plane. The control plane's
+    /// "clock" isn't synchronized with the local device's clock, so the timestamp value may be
+    /// much more than 5 minutes in the past/future when compared to the local device clock.
+    Control(DateTime<Utc>),
+    /// The last time the node was connected to the control plane, as estimated by this device.
+    ///
+    /// For some types of node updates, we're only notified that the node is offline, not when it
+    /// was last seen by the control plane. When we receive one of these offline updates, we
+    /// estimate a last-seen timestamp using the local device clock. As the control plane "clock"
+    /// and the local device clock aren't synchronized, this estimate may differ much more than 5
+    /// minutes in the past/future when compared to a fuzzed timestamp generated by the control
+    /// plane "clock".
+    Estimated(DateTime<Utc>),
+}
+
+impl fmt::Display for NodeLastSeen {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            NodeLastSeen::Control(dt) => write!(f, "last seen (by control): {}", dt),
+            NodeLastSeen::Estimated(dt) => write!(f, "last seen (estimated): {}", dt),
+        }
+    }
+}
+
+/// Whether a node is online (connected to the control plane) or offline (disconnected from the
+/// control plane).
+#[derive(Debug, Default, Copy, Clone, PartialEq, Eq, Hash)]
+pub enum NodeStatus {
+    /// The node may be online or offline; the control plane hasn't informed us yet.
+    #[default]
+    Unknown,
+    /// The node is online (connected to the control plane).
+    Online,
+    /// The node is offline (disconnected from the control plane).
+    Offline(NodeLastSeen),
+}
+
+impl fmt::Display for NodeStatus {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            NodeStatus::Unknown => write!(f, "unknown"),
+            NodeStatus::Online => write!(f, "online"),
+            NodeStatus::Offline(nls) => write!(f, "offline, {nls}"),
+        }
+    }
+}
+
+impl NodeStatus {
+    /// Construct a new `NodeStatus` from a combination of online and last-seen timestamp.
+    ///
+    /// The netmap messages we get from the control plane have multiple ways to indicate a peer is
+    /// online and/or the timestamp the peer was last connected to the control plane. This method
+    /// wraps the sometimes-painful logic of creating a `NodeStatus` from the various combinations
+    /// of netmap field values.
+    pub fn new(online: Option<bool>, last_seen: Option<DateTime<Utc>>) -> Self {
+        match (online, last_seen) {
+            (Some(true), None) => Self::Online,
+            (Some(false), None) => {
+                // We intentionally don't fuzz the estimated timestamp value because the Go client
+                // code doesn't either.
+                // See: https://github.com/tailscale/tailscale/blob/ee0a03b140021541495b25bdb6642b589431758b/control/controlclient/map.go#L753-L764
+                Self::Offline(NodeLastSeen::Estimated(Utc::now()))
+            }
+            (Some(false), Some(dt)) | (None, Some(dt)) => Self::Offline(NodeLastSeen::Control(dt)),
+            (None, None) => Self::Unknown,
+            (online, last_seen) => {
+                tracing::warn!(
+                    ?online,
+                    ?last_seen,
+                    "unexpected combination of online/last_seen states"
+                );
+                Self::Unknown
+            }
+        }
+    }
+}
+
 /// A node in a tailnet.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct Node {
     /// The node's id.
     pub id: Id,
     /// The node's stable id.
     pub stable_id: StableId,
-
-    /// This node's hostname.
+    /// The node's hostname.
     pub hostname: String,
 
+    /// The node's capability version.
+    pub capability_version: CapabilityVersion,
+
+    /// Whether the node is connected to the control plane or not. If [`NodeStatus::Unknown`], the
+    /// control plane hasn't told us yet.
+    ///
+    /// This field does _not_ indicate whether the node is reachable or visible from this device.
+    pub status: NodeStatus,
+
     /// The tailnet this node belongs to.
     pub tailnet: Option<String>,
 
+    /// The node capabilities assigned to this node.
+    pub node_capabilities: BTreeMap<String, Vec<String>>,
+
     /// The tags assigned to this node.
     pub tags: Vec<String>,
 
@@ -39,6 +149,9 @@ pub struct Node {
     pub machine_key: Option<MachinePublicKey>,
     /// The node's [`DiscoPublicKey`], if known.
     pub disco_key: Option<DiscoPublicKey>,
+    /// The signature of the node's public key with the Tailnet Lock signing key, if Tailnet Lock
+    /// is enabled and the signature is known.
+    pub tailnet_lock_key_signature: Option<Vec<u8>>,
 
     /// The routes this node accepts traffic for.
     pub accepted_routes: Vec<ipnet::IpNet>,
@@ -50,6 +163,67 @@ pub struct Node {
 }
 
 impl Node {
+    /// Apply the given partial update to this `Node`. Returns `true` if the update was successfully
+    /// applied; `false` otherwise.
+    #[tracing::instrument(skip_all, fields(id = self.stable_id.0, hostname = self.hostname))]
+    pub fn apply_update(&mut self, update: &NodeUpdate) -> bool {
+        if self.id != update.id {
+            tracing::warn!(
+                node_id = self.id,
+                update_id = update.id,
+                "ignoring node update with incorrect node ID"
+            );
+            return false;
+        }
+
+        if update.status != NodeStatus::Unknown {
+            tracing::debug!(old_status = %self.status, "peer {}", update.status);
+            self.status = update.status;
+        }
+
+        if let Some(cap) = update.cap {
+            tracing::debug!(old=?self.capability_version, new=?cap, "updating capability version");
+            self.capability_version = cap;
+        }
+
+        if let Some(cap_map) = update.cap_map.as_ref() {
+            tracing::debug!(old=?self.node_capabilities, new=?cap_map, "updating node capabilities");
+            self.node_capabilities = cap_map.clone();
+        }
+
+        if let Some(derp_region) = update.derp_region {
+            tracing::debug!(old=?self.derp_region, new=?derp_region, "updating derp home region");
+            self.derp_region = Some(derp_region);
+        }
+
+        if let Some(disco_key) = update.disco_key {
+            tracing::debug!(old=?self.disco_key, new=?disco_key, "updating disco key");
+            self.disco_key = Some(disco_key);
+        }
+
+        if let Some(node_key) = update.node_key {
+            tracing::debug!(old=?self.node_key, new=?node_key, "updating node key");
+            self.node_key = node_key;
+        }
+
+        if let Some(node_key_expiry) = update.node_key_expiry {
+            tracing::debug!(old=?self.node_key_expiry, new=?node_key_expiry, "updating node key expiry");
+            self.node_key_expiry = Some(node_key_expiry);
+        }
+
+        if let Some(tl_sig) = &update.tailnet_lock_key_signature {
+            tracing::debug!(old=?self.tailnet_lock_key_signature, new=?tl_sig, "updating tailnet lock key signature");
+            self.tailnet_lock_key_signature = Some(tl_sig.clone());
+        }
+
+        if let Some(underlay_addresses) = &update.underlay_addresses {
+            tracing::debug!(old=?self.underlay_addresses, new=?underlay_addresses, "updating underlay addresses");
+            self.underlay_addresses = underlay_addresses.clone();
+        }
+
+        true
+    }
+
     /// The fully-qualified domain name of the node.
     ///
     /// This is a string of the form `$HOST.$TAILNET_DOMAIN.`. For tailnets controlled by
@@ -124,10 +298,18 @@ impl From<&ts_control_serde::Node<'_>> for Node {
         Self {
             id: value.id,
             stable_id: StableId(value.stable_id.0.to_string()),
-
             hostname: hostname.to_owned(),
+
+            capability_version: value.cap,
+
+            status: NodeStatus::new(value.online, value.last_seen),
             tailnet,
 
+            node_capabilities: value
+                .cap_map
+                .iter()
+                .map(|(k, v)| (k.to_string(), v.into()))
+                .collect(),
             tags: value
                 .tags
                 .as_ref()
@@ -142,6 +324,7 @@ impl From<&ts_control_serde::Node<'_>> for Node {
             node_key_expiry: value.key_expiry,
             machine_key: value.machine,
             disco_key: value.disco_key,
+            tailnet_lock_key_signature: value.key_signature.as_ref().map(|s| Vec::<u8>::from(*s)),
 
             accepted_routes: value
                 .allowed_ips
@@ -159,3 +342,75 @@ impl From<&ts_control_serde::Node<'_>> for Node {
         }
     }
 }
+
+/// A partial update to a [`Node`] in a tailnet.
+///
+/// Devices should reject any `NodeUpdate` that it doesn't have a corresponding [`Node`] for. This
+/// type combines multiple different update/patch-style fields from netmap messages into a single
+/// type, such as the
+#[derive(Debug, Default, Clone, PartialEq, Eq, Hash)]
+pub struct NodeUpdate {
+    /// The node's id.
+    pub id: Id,
+
+    /// Whether this node is online or offline (with last-seen timestamp), according to the control
+    /// plane. If [`NodeStatus::Unknown`], has not changed.
+    pub status: NodeStatus,
+
+    /// The DERP region for this node. If `None`, has not changed.
+    pub derp_region: Option<ts_transport_derp::RegionId>,
+
+    /// The node's capability version. If `None`, has not changed.
+    pub cap: Option<CapabilityVersion>,
+    /// The node's capabilities (node caps, not peer caps). If `None`, has not changed.
+    pub cap_map: Option<BTreeMap<String, Vec<String>>>,
+
+    /// The node's [`NodePublicKey`]. If `None`, has not changed.
+    pub node_key: Option<NodePublicKey>,
+    /// The node key's expiration. If `None`, has not changed.
+    pub node_key_expiry: Option<DateTime<Utc>>,
+
+    /// The node's [`DiscoPublicKey`]. If `None`, has not changed.
+    pub disco_key: Option<DiscoPublicKey>,
+
+    /// The node's key signature for Tailnet Lock. If `None`, has not changed.
+    pub tailnet_lock_key_signature: Option<Vec<u8>>,
+
+    /// The underlay addresses this node is reachable on (`Endpoints` in Go). If `None`, has not
+    /// changed.
+    pub underlay_addresses: Option<Vec<SocketAddr>>,
+}
+
+impl From<&ts_control_serde::PeerChange<'_>> for NodeUpdate {
+    fn from(value: &ts_control_serde::PeerChange<'_>) -> Self {
+        let cap_map = value.cap_map.as_ref().map(|m| {
+            m.iter()
+                .map(|(name, values)| {
+                    (
+                        String::from(*name),
+                        values
+                            .0
+                            .iter()
+                            .map(|v| v.to_string())
+                            .collect::<Vec<String>>(),
+                    )
+                })
+                .collect()
+        });
+
+        Self {
+            id: value.node_id,
+            status: NodeStatus::new(value.online, value.last_seen),
+            derp_region: value
+                .derp_region
+                .map(|x| ts_transport_derp::RegionId(x.into())),
+            cap: value.cap,
+            cap_map,
+            node_key: value.key,
+            node_key_expiry: value.key_expiry,
+            disco_key: value.disco_key,
+            tailnet_lock_key_signature: value.key_signature.map(|x| x.into()),
+            underlay_addresses: value.endpoints.clone(),
+        }
+    }
+}