github-personal/firezone
1
0
Fork 0
mirror of https://github.com/outbackdingo/firezone.git synced 2026-01-27 10:18:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(connlib): implement idempotent control protocol for client (#6942)

Browse Source 
Building on top of the gateway PR (#6941), this PR transitions the
clients to the new control protocol. Clients are **not**
backwards-compatible with old gateways. As a result, a certain customer
environment MUST have at least one gateway with the above PR running in
order for clients to be able to establish connections.

With this transition, Clients send explicit events to Gateways whenever
they assign IPs to a DNS resource name. The actual assignment only
happens once and the IPs then remain stable for the duration of the
client session.

When the Gateway receives such an event, it will perform a DNS
resolution of the requested domain name and set up the NAT between the
assigned proxy IPs and the IPs the domain actually resolves to. In order
to support self-healing of any problems that happen during this process,
the client will send an "Assigned IPs" event every time it receives a
DNS query for a particular domain. This in turn will trigger another DNS
resolution on the Gateway. Effectively, this means that DNS queries for
DNS resources propagate to the Gateway, triggering a DNS resolution
there. In case the domain resolves to the same set of IPs, no state is
changed to ensure existing connections are not interrupted.

With this new functionality in place, we can delete the old logic around
detecting "expired" IPs. This is considered a bugfix as this logic isn't
currently working as intended. It has been observed multiple times that
the Gateway can loop on this behaviour and resolving the same domain
over and over again. The only theoretical "incompatibility" here is that
pre-1.4.0 clients won't have access to this functionality of triggering
DNS refreshes on a Gateway 1.4.2+ Gateway. However, as soon as this PR
merges, we expect all admins to have already upgraded to a 1.4.0+
Gateway anyway which already mandates clients to be on 1.4.0+.

Resolves: #7391.
Resolves: #6828.
This commit is contained in:
Thomas Eizinger
2024-12-04 12:05:35 +00:00
committed by GitHub
parent fd8ca853a3
commit b802021cc4
25 changed files with 622 additions and 1376 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
rust/Cargo.lock generated
View File

@@ -2216,6 +2216,7 @@ dependencies = [
"secrecy",
"serde",
"serde_json",
"sha2",
"snownet",
"socket-factory",
"socket2",

246
rust/connlib/clients/shared/src/eventloop.rs
View File

@@ -1,13 +1,17 @@
use crate::{callbacks::Callbacks, PHOENIX_TOPIC};
use anyhow::Result;
use connlib_model::ResourceId;
use connlib_model::{PublicKey, ResourceId};
use firezone_logging::{anyhow_dyn_err, err_with_src, telemetry_event};
use firezone_tunnel::messages::{client::*, *};
use firezone_tunnel::messages::client::{
EgressMessages, FailReason, FlowCreated, FlowCreationFailed, GatewayIceCandidates,
GatewaysIceCandidates, IngressMessages, InitClient,
};
use firezone_tunnel::messages::RelaysPresence;
use firezone_tunnel::ClientTunnel;
use phoenix_channel::{ErrorReply, OutboundRequestId, PhoenixChannel, PublicKeyParam};
use std::time::Instant;
use std::{
collections::{BTreeMap, BTreeSet},
collections::BTreeSet,
io,
net::IpAddr,
task::{Context, Poll},
@@ -18,10 +22,8 @@ pub struct Eventloop<C: Callbacks> {
tunnel: ClientTunnel,
callbacks: C,
portal: PhoenixChannel<(), IngressMessages, ReplyMessages, PublicKeyParam>,
portal: PhoenixChannel<(), IngressMessages, (), PublicKeyParam>,
rx: tokio::sync::mpsc::UnboundedReceiver<Command>,
connection_intents: SentConnectionIntents,
}
/// Commands that can be sent to the [`Eventloop`].
@@ -37,7 +39,7 @@ impl<C: Callbacks> Eventloop<C> {
pub(crate) fn new(
tunnel: ClientTunnel,
callbacks: C,
mut portal: PhoenixChannel<(), IngressMessages, ReplyMessages, PublicKeyParam>,
mut portal: PhoenixChannel<(), IngressMessages, (), PublicKeyParam>,
rx: tokio::sync::mpsc::UnboundedReceiver<Command>,
) -> Self {
portal.connect(PublicKeyParam(tunnel.public_key().to_bytes()));
@@ -45,7 +47,6 @@ impl<C: Callbacks> Eventloop<C> {
Self {
tunnel,
portal,
connection_intents: SentConnectionIntents::default(),
rx,
callbacks,
}
@@ -143,29 +144,13 @@ where
firezone_tunnel::ClientEvent::ConnectionIntent {
connected_gateway_ids,
resource,
..
} => {
let id = self.portal.send(
PHOENIX_TOPIC,
EgressMessages::PrepareConnection {
resource_id: resource,
connected_gateway_ids,
},
);
self.connection_intents.register_new_intent(id, resource);
}
firezone_tunnel::ClientEvent::RequestAccess {
resource_id,
gateway_id,
maybe_domain,
} => {
self.portal.send(
PHOENIX_TOPIC,
EgressMessages::ReuseConnection(ReuseConnection {
resource_id,
gateway_id,
payload: maybe_domain,
}),
EgressMessages::CreateFlow {
resource_id: resource,
connected_gateway_ids,
},
);
}
firezone_tunnel::ClientEvent::ResourcesChanged { resources } => {
@@ -181,40 +166,15 @@ where
Vec::from_iter(config.ipv6_routes),
);
}
firezone_tunnel::ClientEvent::RequestConnection {
gateway_id,
offer,
preshared_key,
resource_id,
maybe_domain,
} => {
self.portal.send(
PHOENIX_TOPIC,
EgressMessages::RequestConnection(RequestConnection {
gateway_id,
resource_id,
client_preshared_key: preshared_key,
client_payload: ClientPayload {
ice_parameters: offer,
domain: maybe_domain,
},
}),
);
}
}
}
fn handle_portal_event(
&mut self,
event: phoenix_channel::Event<IngressMessages, ReplyMessages>,
) {
fn handle_portal_event(&mut self, event: phoenix_channel::Event<IngressMessages, ()>) {
match event {
phoenix_channel::Event::InboundMessage { msg, .. } => {
self.handle_portal_inbound_message(msg);
}
phoenix_channel::Event::SuccessResponse { res, req_id, .. } => {
self.handle_portal_success_reply(res, req_id);
}
phoenix_channel::Event::SuccessResponse { res: (), .. } => {}
phoenix_channel::Event::ErrorResponse { res, req_id, topic } => {
self.handle_portal_error_reply(res, topic, req_id);
}
@@ -283,52 +243,23 @@ where
)
}
}
}
}
fn handle_portal_success_reply(&mut self, res: ReplyMessages, req_id: OutboundRequestId) {
match res {
ReplyMessages::Connect(Connect {
gateway_payload:
GatewayResponse::ConnectionAccepted(ConnectionAccepted { ice_parameters, .. }),
gateway_public_key,
IngressMessages::FlowCreated(FlowCreated {
resource_id,
..
}) => {
if let Err(e) = self.tunnel.state_mut().accept_answer(
ice_parameters,
resource_id,
gateway_public_key.0.into(),
Instant::now(),
) {
tracing::warn!(error = anyhow_dyn_err(&e), "Failed to accept connection");
}
}
ReplyMessages::Connect(Connect {
gateway_payload: GatewayResponse::ResourceAccepted(ResourceAccepted { .. }),
..
}) => {
tracing::trace!("Connection response received, ignored as it's deprecated")
}
ReplyMessages::ConnectionDetails(ConnectionDetails {
gateway_id,
resource_id,
site_id,
..
gateway_public_key,
preshared_key,
client_ice_credentials,
gateway_ice_credentials,
}) => {
let should_accept = self
.connection_intents
.handle_connection_details_received(req_id, resource_id);
if !should_accept {
tracing::debug!(%resource_id, "Ignoring stale connection details");
return;
}
match self.tunnel.state_mut().on_routing_details(
match self.tunnel.state_mut().handle_flow_created(
resource_id,
gateway_id,
PublicKey::from(gateway_public_key.0),
site_id,
preshared_key,
client_ice_credentials,
gateway_ice_credentials,
Instant::now(),
) {
Ok(Ok(())) => {}
@@ -349,6 +280,16 @@ where
}
};
}
IngressMessages::FlowCreationFailed(FlowCreationFailed {
resource_id,
reason: FailReason::Offline,
..
}) => {
self.tunnel.state_mut().set_resource_offline(resource_id);
}
IngressMessages::FlowCreationFailed(FlowCreationFailed { reason, .. }) => {
tracing::warn!("Failed to create flow: {reason:?}")
}
}
}
@@ -359,130 +300,15 @@ where
req_id: OutboundRequestId,
) {
match res {
ErrorReply::Offline => {
let Some(offline_resource) = self.connection_intents.handle_error(req_id) else {
return;
};
tracing::debug!(resource_id = %offline_resource, "Resource is offline");
self.tunnel
.state_mut()
.set_resource_offline(offline_resource);
}
ErrorReply::Disabled => {
tracing::debug!(%req_id, "Functionality is disabled");
}
ErrorReply::UnmatchedTopic => {
self.portal.join(topic, ());
}
reason @ (ErrorReply::InvalidVersion | ErrorReply::NotFound | ErrorReply::Other) => {
reason @ (ErrorReply::InvalidVersion | ErrorReply::Other) => {
tracing::debug!(%req_id, %reason, "Request failed");
}
}
}
}
#[derive(Default)]
struct SentConnectionIntents {
inner: BTreeMap<OutboundRequestId, ResourceId>,
}
impl SentConnectionIntents {
fn register_new_intent(&mut self, id: OutboundRequestId, resource: ResourceId) {
self.inner.insert(id, resource);
}
/// To be called when we receive the connection details for a particular resource.
///
/// Returns whether we should accept them.
fn handle_connection_details_received(
&mut self,
reference: OutboundRequestId,
r: ResourceId,
) -> bool {
let has_more_recent_intent = self
.inner
.iter()
.any(|(req, resource)| req > &reference && resource == &r);
if has_more_recent_intent {
return false;
}
let has_intent = self
.inner
.get(&reference)
.is_some_and(|resource| resource == &r);
if !has_intent {
return false;
}
self.inner.retain(|_, v| v != &r);
true
}
fn handle_error(&mut self, req: OutboundRequestId) -> Option<ResourceId> {
self.inner.remove(&req)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn discards_old_connection_intent() {
let mut intents = SentConnectionIntents::default();
let resource = ResourceId::random();
intents.register_new_intent(OutboundRequestId::for_test(1), resource);
intents.register_new_intent(OutboundRequestId::for_test(2), resource);
let should_accept =
intents.handle_connection_details_received(OutboundRequestId::for_test(1), resource);
assert!(!should_accept);
}
#[test]
fn allows_unrelated_intents() {
let mut intents = SentConnectionIntents::default();
let resource1 = ResourceId::random();
let resource2 = ResourceId::random();
intents.register_new_intent(OutboundRequestId::for_test(1), resource1);
intents.register_new_intent(OutboundRequestId::for_test(2), resource2);
let should_accept_1 =
intents.handle_connection_details_received(OutboundRequestId::for_test(1), resource1);
let should_accept_2 =
intents.handle_connection_details_received(OutboundRequestId::for_test(2), resource2);
assert!(should_accept_1);
assert!(should_accept_2);
}
#[test]
fn handles_out_of_order_responses() {
let mut intents = SentConnectionIntents::default();
let resource = ResourceId::random();
intents.register_new_intent(OutboundRequestId::for_test(1), resource);
intents.register_new_intent(OutboundRequestId::for_test(2), resource);
let should_accept_2 =
intents.handle_connection_details_received(OutboundRequestId::for_test(2), resource);
let should_accept_1 =
intents.handle_connection_details_received(OutboundRequestId::for_test(1), resource);
assert!(should_accept_2);
assert!(!should_accept_1);
}
}

8
rust/connlib/clients/shared/src/lib.rs
View File

@@ -3,9 +3,7 @@ pub use crate::serde_routelist::{V4RouteList, V6RouteList};
pub use callbacks::{Callbacks, DisconnectError};
pub use connlib_model::StaticSecret;
pub use eventloop::Eventloop;
pub use firezone_tunnel::messages::client::{
ResourceDescription, {IngressMessages, ReplyMessages},
};
pub use firezone_tunnel::messages::client::{IngressMessages, ResourceDescription};
use connlib_model::ResourceId;
use eventloop::Command;
@@ -41,7 +39,7 @@ impl Session {
tcp_socket_factory: Arc<dyn SocketFactory<TcpSocket>>,
udp_socket_factory: Arc<dyn SocketFactory<UdpSocket>>,
callbacks: CB,
portal: PhoenixChannel<(), IngressMessages, ReplyMessages, PublicKeyParam>,
portal: PhoenixChannel<(), IngressMessages, (), PublicKeyParam>,
handle: tokio::runtime::Handle,
) -> Self {
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
@@ -118,7 +116,7 @@ async fn connect<CB>(
tcp_socket_factory: Arc<dyn SocketFactory<TcpSocket>>,
udp_socket_factory: Arc<dyn SocketFactory<UdpSocket>>,
callbacks: CB,
portal: PhoenixChannel<(), IngressMessages, ReplyMessages, PublicKeyParam>,
portal: PhoenixChannel<(), IngressMessages, (), PublicKeyParam>,
rx: UnboundedReceiver<Command>,
) -> Result<(), phoenix_channel::Error>
where

5
rust/connlib/snownet/src/node.rs
View File

@@ -1001,11 +1001,6 @@ where
Ok(params)
}
/// Whether we have sent an [`Offer`] for this connection and are currently expecting an [`Answer`].
pub fn is_expecting_answer(&self, id: TId) -> bool {
self.connections.initial.contains_key(&id)
}
/// Accept an [`Answer`] from the remote for a connection previously created via [`Node::new_connection`].
#[tracing::instrument(level = "info", skip_all, fields(%cid))]
#[deprecated]

1
rust/connlib/tunnel/Cargo.toml
View File

@@ -48,6 +48,7 @@ firezone-relay = { workspace = true, features = ["proptest"] }
ip-packet = { workspace = true, features = ["proptest"] }
proptest-state-machine = { workspace = true }
rand = { workspace = true }
sha2 = { workspace = true }
test-case = { workspace = true }
test-strategy = { workspace = true }
tracing-subscriber = { workspace = true, features = ["env-filter"] }

4
rust/connlib/tunnel/proptest-regressions/tests.txt
View File

File diff suppressed because one or more lines are too long

467
rust/connlib/tunnel/src/client.rs
View File

@@ -5,14 +5,15 @@ pub(crate) use resource::{CidrResource, Resource};
pub(crate) use resource::{DnsResource, InternetResource};
use crate::dns::StubResolver;
use crate::messages::ResolveRequest;
use crate::messages::{DnsServer, Interface as InterfaceConfig, IpDnsServer, Key, Offer};
use crate::messages::{DnsServer, Interface as InterfaceConfig, IpDnsServer};
use crate::messages::{IceCredentials, SecretKey};
use crate::peer_store::PeerStore;
use crate::{dns, TunConfig};
use crate::{dns, p2p_control, TunConfig};
use anyhow::Context;
use bimap::BiMap;
use connlib_model::PublicKey;
use connlib_model::{GatewayId, RelayId, ResourceId, ResourceStatus, ResourceView};
use connlib_model::{
DomainName, GatewayId, PublicKey, RelayId, ResourceId, ResourceStatus, ResourceView,
};
use connlib_model::{Site, SiteId};
use firezone_logging::{
anyhow_dyn_err, err_with_src, telemetry_event, unwrap_or_debug, unwrap_or_warn,
@@ -88,9 +89,14 @@ pub struct ClientState {
node: ClientNode<GatewayId, RelayId>,
/// All gateways we are connected to and the associated, connection-specific state.
peers: PeerStore<GatewayId, GatewayOnClient>,
/// Which Resources we are trying to connect to.
awaiting_connection_details: HashMap<ResourceId, AwaitingConnectionDetails>,
/// Tracks the flows to resources that we are currently trying to establish.
pending_flows: HashMap<ResourceId, PendingFlow>,
/// Tracks the domains for which we have set up a NAT per gateway.
///
/// The IPs for DNS resources get assigned on the client.
/// In order to route them to the actual resource, the gateway needs to set up a NAT table.
/// Until the NAT is set up, packets sent to these resources are effectively black-holed.
dns_resource_nat_by_gateway: BTreeMap<(GatewayId, DomainName), DnsResourceNatState>,
/// Tracks which gateway to use for a particular Resource.
resources_gateways: HashMap<ResourceId, GatewayId>,
/// The site a gateway belongs to.
@@ -144,10 +150,19 @@ pub struct ClientState {
buffered_dns_queries: VecDeque<dns::RecursiveQuery>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct AwaitingConnectionDetails {
enum DnsResourceNatState {
Pending { sent_at: Instant },
Confirmed,
}
impl DnsResourceNatState {
fn confirm(&mut self) {
*self = Self::Confirmed;
}
}
struct PendingFlow {
last_intent_sent_at: Instant,
domain: Option<ResolveRequest>,
}
impl ClientState {
@@ -157,7 +172,6 @@ impl ClientState {
now: Instant,
) -> Self {
Self {
awaiting_connection_details: Default::default(),
resources_gateways: Default::default(),
active_cidr_resources: IpNetworkTable::new(),
resources_by_id: Default::default(),
@@ -181,6 +195,8 @@ impl ClientState {
tcp_dns_client: dns_over_tcp::Client::new(now, seed),
tcp_dns_server: dns_over_tcp::Server::new(now),
tcp_dns_sockets_by_upstream_and_query_id: Default::default(),
pending_flows: Default::default(),
dns_resource_nat_by_gateway: BTreeMap::new(),
}
}
@@ -256,29 +272,113 @@ impl ClientState {
self.node.public_key()
}
fn request_access(
&mut self,
resource_ip: &IpAddr,
resource_id: ResourceId,
gateway_id: GatewayId,
) {
let Some((fqdn, ips)) = self.stub_resolver.get_fqdn(resource_ip) else {
return;
};
self.peers
.add_ips_with_resource(&gateway_id, ips.iter().copied(), &resource_id);
self.buffered_events.push_back(ClientEvent::RequestAccess {
resource_id,
gateway_id,
maybe_domain: Some(ResolveRequest {
name: fqdn.clone(),
proxy_ips: ips.clone(),
}),
})
/// Updates the NAT for all domains resolved by the stub resolver on the corresponding gateway.
///
/// In order to route traffic for DNS resources, the designated gateway needs to set up NAT from
/// the IPs assigned by the client's stub resolver and the actual IPs the domains resolve to.
///
/// The corresponding control message containing the domain and IPs is sent over UDP through the tunnel.
/// UDP is unreliable, even through the WG tunnel, meaning we need our own way of making reliable.
/// The algorithm for that is simple:
/// 1. We track the timestamp when we've last sent the setup message.
/// 2. The message is designed to be idempotent on the gateway.
/// 3. If we don't receive a response within 2s and this function is called again, we send another message.
///
/// The complexity of this function is O(N) with the number of resolved DNS resources.
fn update_dns_resource_nat(&mut self, now: Instant) {
use std::collections::btree_map::Entry;
for (domain, rid, proxy_ips, gid) in
self.stub_resolver
.resolved_resources()
.map(|(domain, resource, proxy_ips)| {
let gateway = self.resources_gateways.get(resource);
(domain, resource, proxy_ips, gateway)
})
{
let Some(gid) = gid else {
tracing::trace!(
%domain, %rid,
"No gateway connected for resource, skipping DNS resource NAT setup"
);
continue;
};
match self
.dns_resource_nat_by_gateway
.entry((*gid, domain.clone()))
{
Entry::Vacant(v) => {
self.peers
.add_ips_with_resource(gid, proxy_ips.iter().copied(), rid);
v.insert(DnsResourceNatState::Pending { sent_at: now });
}
Entry::Occupied(mut o) => match o.get_mut() {
DnsResourceNatState::Confirmed => continue,
DnsResourceNatState::Pending { sent_at } => {
let time_since_last_attempt = now.duration_since(*sent_at);
if time_since_last_attempt < Duration::from_secs(2) {
continue;
}
*sent_at = now;
}
},
}
let packet = match p2p_control::dns_resource_nat::assigned_ips(
*rid,
domain.clone(),
proxy_ips.clone(),
) {
Ok(packet) => packet,
Err(e) => {
tracing::warn!(
error = anyhow_dyn_err(&e),
"Failed to create IP packet for `AssignedIp`s event"
);
continue;
}
};
tracing::debug!(%gid, %domain, "Setting up DNS resource NAT");
let Some(transmit) = self
.node
.encapsulate(*gid, packet, now)
.inspect_err(|e| tracing::debug!(%gid, "Failed to encapsulate: {e}"))
.ok()
.flatten()
else {
continue;
};
self.buffered_transmits
.push_back(transmit.to_transmit().into_owned());
}
}
fn is_dns_resource(&self, resource: &ResourceId) -> bool {
matches!(self.resources_by_id.get(resource), Some(Resource::Dns(_)))
/// Clears the DNS resource NAT state for a given domain.
///
/// Once cleared, this will trigger the client to submit another `AssignedIp`s event to the Gateway.
/// On the Gateway, such an event causes a new DNS resolution.
///
/// We call this function every time a client issues a DNS query for a certain domain.
/// Coupling this behaviour together allows a client to refresh the DNS resolution of a DNS resource on the Gateway
/// through local DNS resolutions.
fn clear_dns_resource_nat_for_domain(&mut self, message: Message<&[u8]>) {
let Ok(question) = message.sole_question() else {
return;
};
let domain = question.into_qname();
tracing::debug!(%domain, "Clearing DNS resource NAT");
self.dns_resource_nat_by_gateway
.retain(|(_, candidate), _| candidate != &domain);
}
fn is_cidr_resource_connected(&self, resource: &ResourceId) -> bool {
@@ -334,6 +434,11 @@ impl ClientState {
return None;
}
if let Some(fz_p2p_control) = packet.as_fz_p2p_control() {
handle_p2p_control_packet(gid, fz_p2p_control, &mut self.dns_resource_nat_by_gateway);
return None;
}
let Some(peer) = self.peers.get_mut(&gid) else {
tracing::error!(%gid, "Couldn't find connection by ID");
@@ -421,22 +526,26 @@ impl ClientState {
return None;
};
// We read this here to prevent problems with the borrow checker
let is_dns_resource = self.is_dns_resource(&resource);
let Some(peer) = peer_by_resource_mut(&self.resources_gateways, &mut self.peers, resource)
else {
self.on_not_connected_resource(resource, &dst, now);
self.on_not_connected_resource(resource, now);
return None;
};
// Allowed IPs will track the IPs that we have sent to the gateway along with a list of ResourceIds
// for DNS resource we will send the IP one at a time.
if is_dns_resource && peer.allowed_ips.exact_match(dst).is_none() {
let gateway_id = peer.id();
self.request_access(&dst, resource, gateway_id);
return None;
}
// TODO: Don't send packets unless we have a positive response for the DNS resource NAT.
// TODO: Check DNS resource NAT state for the domain that the destination IP belongs to.
// Re-send if older than X.
// if let Some((domain, _)) = self.stub_resolver.resolve_resource_by_ip(&dst) {
// if self
// .dns_resource_nat_by_gateway
// .get(&(peer.id(), domain.clone()))
// .is_some_and(|s| s.is_pending())
// {
// self.update_dns_resource_nat(now);
// }
// }
let gid = peer.id();
@@ -491,59 +600,29 @@ impl ClientState {
self.drain_node_events();
}
#[tracing::instrument(level = "trace", skip_all, fields(%resource_id))]
#[expect(deprecated, reason = "Will be deleted together with deprecated API")]
pub fn accept_answer(
&mut self,
answer: impl Into<snownet::Answer>,
resource_id: ResourceId,
gateway: PublicKey,
now: Instant,
) -> anyhow::Result<()> {
let answer = answer.into();
debug_assert!(!self.awaiting_connection_details.contains_key(&resource_id));
let gateway_id = self
.gateway_by_resource(&resource_id)
.with_context(|| format!("No gateway associated with resource {resource_id}"))?;
self.node.accept_answer(gateway_id, gateway, answer, now);
Ok(())
}
/// Updates the "routing table".
///
/// In a nutshell, this tells us which gateway in which site to use for the given resource.
#[tracing::instrument(level = "debug", skip_all, fields(%resource_id, %gateway_id))]
#[expect(
deprecated,
reason = "Will be refactored when deprecated control protocol is shipped"
)]
pub fn on_routing_details(
#[tracing::instrument(level = "debug", skip_all, fields(%gateway_id))]
#[expect(clippy::too_many_arguments)]
pub fn handle_flow_created(
&mut self,
resource_id: ResourceId,
gateway_id: GatewayId,
gateway_key: PublicKey,
site_id: SiteId,
preshared_key: SecretKey,
client_ice: IceCredentials,
gateway_ice: IceCredentials,
now: Instant,
) -> anyhow::Result<Result<(), NoTurnServers>> {
tracing::trace!("Updating resource routing table");
let desc = self
let resource = self
.resources_by_id
.get(&resource_id)
.context("Unknown resource")?;
if self.node.is_expecting_answer(gateway_id) {
return Ok(Ok(()));
}
let awaiting_connection_details = self
.awaiting_connection_details
self.pending_flows
.remove(&resource_id)
.context("No connection details found for resource")?;
let ips = get_addresses_for_awaiting_resource(desc, &awaiting_connection_details);
.context("No pending flow for resource")?;
if let Some(old_gateway_id) = self.resources_gateways.insert(resource_id, gateway_id) {
if self.peers.get(&old_gateway_id).is_some() {
@@ -551,48 +630,39 @@ impl ClientState {
}
}
match self.node.upsert_connection(
gateway_id,
gateway_key,
Secret::new(preshared_key.expose_secret().0),
snownet::Credentials {
username: client_ice.username,
password: client_ice.password,
},
snownet::Credentials {
username: gateway_ice.username,
password: gateway_ice.password,
},
now,
) {
Ok(()) => {}
Err(e) => return Ok(Err(e)),
};
self.resources_gateways.insert(resource_id, gateway_id);
self.gateways_site.insert(gateway_id, site_id);
self.recently_connected_gateways.put(gateway_id, ());
if self.peers.get(&gateway_id).is_some() {
self.peers
.add_ips_with_resource(&gateway_id, ips.into_iter(), &resource_id);
self.buffered_events.push_back(ClientEvent::RequestAccess {
resource_id,
gateway_id,
maybe_domain: awaiting_connection_details.domain,
});
return Ok(Ok(()));
if self.peers.get(&gateway_id).is_none() {
self.peers.insert(GatewayOnClient::new(gateway_id), &[]);
};
let offer = match self.node.new_connection(
gateway_id,
awaiting_connection_details.last_intent_sent_at,
now,
) {
Ok(o) => o,
Err(e) => return Ok(Err(e)),
};
self.peers.insert(
GatewayOnClient::new(gateway_id, &ips, HashSet::from([resource_id])),
&[],
// This only works for CIDR & Internet Resource.
self.peers.add_ips_with_resource(
&gateway_id,
resource.addresses().into_iter(),
&resource_id,
);
self.peers
.add_ips_with_resource(&gateway_id, ips.into_iter(), &resource_id);
self.buffered_events
.push_back(ClientEvent::RequestConnection {
gateway_id,
offer: Offer {
username: offer.credentials.username,
password: offer.credentials.password,
},
preshared_key: Secret::new(Key(*offer.session_key.expose_secret())),
resource_id,
maybe_domain: awaiting_connection_details.domain,
});
self.update_dns_resource_nat(now);
Ok(Ok(()))
}
@@ -633,72 +703,53 @@ impl ClientState {
}
pub fn on_connection_failed(&mut self, resource: ResourceId) {
self.awaiting_connection_details.remove(&resource);
self.resources_gateways.remove(&resource);
self.pending_flows.remove(&resource);
let Some(disconnected_gateway) = self.resources_gateways.remove(&resource) else {
return;
};
self.cleanup_connected_gateway(&disconnected_gateway);
}
#[tracing::instrument(level = "debug", skip_all, fields(%resource))]
fn on_not_connected_resource(
&mut self,
resource: ResourceId,
destination: &IpAddr,
now: Instant,
) {
fn on_not_connected_resource(&mut self, resource: ResourceId, now: Instant) {
debug_assert!(self.resources_by_id.contains_key(&resource));
if self
.gateway_by_resource(&resource)
.is_some_and(|gateway_id| self.node.is_expecting_answer(gateway_id))
{
tracing::debug!("Already connecting to gateway");
match self.pending_flows.entry(resource) {
Entry::Vacant(v) => {
v.insert(PendingFlow {
last_intent_sent_at: now,
});
}
Entry::Occupied(mut o) => {
let pending_flow = o.get_mut();
return;
}
match self.awaiting_connection_details.entry(resource) {
Entry::Occupied(mut occupied) => {
let time_since_last_intent = now.duration_since(occupied.get().last_intent_sent_at);
let time_since_last_intent = now.duration_since(pending_flow.last_intent_sent_at);
if time_since_last_intent < Duration::from_secs(2) {
tracing::trace!(?time_since_last_intent, "Skipping connection intent");
return;
}
occupied.get_mut().last_intent_sent_at = now;
}
Entry::Vacant(vacant) => {
vacant.insert(AwaitingConnectionDetails {
last_intent_sent_at: now,
// Note: in case of an overlapping CIDR resource this should be None instead of Some if the resource_id
// is for a CIDR resource.
// But this should never happen as DNS resources are always preferred, so we don't encode the logic here.
// Tests will prevent this from ever happening.
domain: self.stub_resolver.get_fqdn(destination).map(|(fqdn, ips)| {
ResolveRequest {
name: fqdn.clone(),
proxy_ips: ips.clone(),
}
}),
});
pending_flow.last_intent_sent_at = now;
}
}
tracing::debug!("Sending connection intent");
// We tell the portal about all gateways we ever connected to, to encourage re-connecting us to the same ones during a session.
// The LRU cache visits them in MRU order, meaning a gateway that we recently connected to should still be preferred.
let connected_gateway_ids = self
.recently_connected_gateways
.iter()
.map(|(g, _)| *g)
.collect();
self.buffered_events
.push_back(ClientEvent::ConnectionIntent {
resource,
connected_gateway_ids,
});
connected_gateway_ids: self.connected_gateway_ids(),
})
}
// We tell the portal about all gateways we ever connected to, to encourage re-connecting us to the same ones during a session.
// The LRU cache visits them in MRU order, meaning a gateway that we recently connected to should still be preferred.
fn connected_gateway_ids(&self) -> BTreeSet<GatewayId> {
self.recently_connected_gateways
.iter()
.map(|(g, _)| *g)
.collect()
}
pub fn gateway_by_resource(&self, resource: &ResourceId) -> Option<GatewayId> {
@@ -768,11 +819,14 @@ impl ClientState {
self.dns_mapping.clone()
}
#[tracing::instrument(level = "debug", skip_all, fields(gateway = %gateway_id))]
pub fn cleanup_connected_gateway(&mut self, gateway_id: &GatewayId) {
self.update_site_status_by_gateway(gateway_id, ResourceStatus::Unknown);
self.peers.remove(gateway_id);
self.resources_gateways.retain(|_, g| g != gateway_id);
#[tracing::instrument(level = "debug", skip_all, fields(gateway = %disconnected_gateway))]
fn cleanup_connected_gateway(&mut self, disconnected_gateway: &GatewayId) {
self.update_site_status_by_gateway(disconnected_gateway, ResourceStatus::Unknown);
self.peers.remove(disconnected_gateway);
self.resources_gateways
.retain(|_, g| g != disconnected_gateway);
self.dns_resource_nat_by_gateway
.retain(|(gateway, _), _| gateway != disconnected_gateway);
}
fn routes(&self) -> impl Iterator<Item = IpNetwork> + '_ {
@@ -802,6 +856,7 @@ impl ClientState {
let maybe_dns_resource_id = self
.stub_resolver
.resolve_resource_by_ip(&destination)
.map(|(_, r)| *r)
.filter(|resource| self.is_resource_enabled(resource))
.inspect(
|resource| tracing::trace!(target: "tunnel_test_coverage", %destination, %resource, "Packet for DNS resource"),
@@ -905,7 +960,7 @@ impl ClientState {
// Check if have any pending TCP DNS queries.
if let Some(query) = self.tcp_dns_server.poll_queries() {
self.handle_tcp_dns_query(query);
self.handle_tcp_dns_query(query, now);
continue;
}
@@ -976,6 +1031,9 @@ impl ClientState {
match self.stub_resolver.handle(message) {
dns::ResolveStrategy::LocalResponse(response) => {
self.clear_dns_resource_nat_for_domain(response.for_slice_ref());
self.update_dns_resource_nat(now);
unwrap_or_debug!(
self.try_queue_udp_dns_response(upstream, source, &response),
"Failed to queue UDP DNS response: {}"
@@ -1007,7 +1065,7 @@ impl ClientState {
ControlFlow::Break(())
}
fn handle_tcp_dns_query(&mut self, query: dns_over_tcp::Query) {
fn handle_tcp_dns_query(&mut self, query: dns_over_tcp::Query, now: Instant) {
let message = query.message;
let Some(upstream) = self.dns_mapping.get_by_left(&query.local.ip()) else {
@@ -1018,6 +1076,9 @@ impl ClientState {
match self.stub_resolver.handle(message.for_slice_ref()) {
dns::ResolveStrategy::LocalResponse(response) => {
self.clear_dns_resource_nat_for_domain(response.for_slice_ref());
self.update_dns_resource_nat(now);
unwrap_or_debug!(
self.tcp_dns_server.send_message(query.socket, response),
"Failed to send TCP DNS response: {}"
@@ -1206,6 +1267,7 @@ impl ClientState {
self.node.reset();
self.recently_connected_gateways.clear(); // Ensure we don't have sticky gateways when we roam.
self.dns_resource_nat_by_gateway.clear();
self.drain_node_events();
// Resetting the client will trigger a failed `QueryResult` for each one that is in-progress.
@@ -1358,7 +1420,7 @@ impl ClientState {
tracing::info!(%name, address, %sites, "Deactivating resource");
self.awaiting_connection_details.remove(&id);
self.pending_flows.remove(&id);
let Some(peer) = peer_by_resource_mut(&self.resources_gateways, &mut self.peers, id) else {
return;
@@ -1464,6 +1526,42 @@ fn parse_udp_dns_message<'b>(datagram: &UdpSlice<'b>) -> anyhow::Result<Message<
Ok(message)
}
fn handle_p2p_control_packet(
gid: GatewayId,
fz_p2p_control: ip_packet::FzP2pControlSlice,
dns_resource_nat_by_gateway: &mut BTreeMap<(GatewayId, DomainName), DnsResourceNatState>,
) {
use p2p_control::dns_resource_nat;
match fz_p2p_control.event_type() {
p2p_control::DOMAIN_STATUS_EVENT => {
let Ok(res) = dns_resource_nat::decode_domain_status(fz_p2p_control)
.inspect_err(|e| tracing::debug!("{e:#}"))
else {
return;
};
if res.status != dns_resource_nat::NatStatus::Active {
tracing::debug!(%gid, domain = %res.domain, "DNS resource NAT is not active");
return;
}
let Some(nat_state) = dns_resource_nat_by_gateway.get_mut(&(gid, res.domain.clone()))
else {
tracing::debug!(%gid, domain = %res.domain, "No DNS resource NAT state, ignoring response");
return;
};
tracing::debug!(%gid, domain = %res.domain, "DNS resource NAT is active");
nat_state.confirm();
}
code => {
tracing::debug!(code = %code.into_u8(), "Unknown control protocol");
}
}
}
fn peer_by_resource_mut<'p>(
resources_gateways: &HashMap<ResourceId, GatewayId>,
peers: &'p mut PeerStore<GatewayId, GatewayOnClient>,
@@ -1475,28 +1573,6 @@ fn peer_by_resource_mut<'p>(
Some(peer)
}
fn get_addresses_for_awaiting_resource(
desc: &Resource,
awaiting_connection_details: &AwaitingConnectionDetails,
) -> Vec<IpNetwork> {
match desc {
Resource::Dns(_) => awaiting_connection_details
.domain
.as_ref()
.expect("for dns resources the awaiting connection should have an ip")
.proxy_ips
.iter()
.copied()
.map_into()
.collect_vec(),
Resource::Cidr(r) => vec![r.address],
Resource::Internet(_) => vec![
Ipv4Network::DEFAULT_ROUTE.into(),
Ipv6Network::DEFAULT_ROUTE.into(),
],
}
}
fn effective_dns_servers(
upstream_dns: Vec<DnsServer>,
default_resolvers: Vec<IpAddr>,
@@ -1549,6 +1625,7 @@ fn sentinel_dns_mapping(
})
.collect()
}
/// Compares the given [`IpAddr`] against a static set of ignored IPs that are definitely not resources.
fn is_definitely_not_a_resource(ip: IpAddr) -> bool {
/// Source: https://en.wikipedia.org/wiki/Multicast_address#Notable_IPv4_multicast_addresses

13
rust/connlib/tunnel/src/client/resource.rs
View File

@@ -6,7 +6,7 @@ use connlib_model::{
CidrResourceView, DnsResourceView, InternetResourceView, ResourceId, ResourceStatus,
ResourceView, Site,
};
use ip_network::IpNetwork;
use ip_network::{IpNetwork, Ipv4Network, Ipv6Network};
use itertools::Itertools as _;
use crate::messages::client::{
@@ -131,6 +131,17 @@ impl Resource {
}
}
pub fn addresses(&self) -> Vec<IpNetwork> {
match self {
Resource::Dns(_) => vec![],
Resource::Cidr(c) => vec![c.address],
Resource::Internet(_) => vec![
Ipv4Network::DEFAULT_ROUTE.into(),
Ipv6Network::DEFAULT_ROUTE.into(),
],
}
}
pub fn with_status(self, status: ResourceStatus) -> ResourceView {
match self {
Resource::Dns(r) => ResourceView::Dns(r.with_status(status)),

26
rust/connlib/tunnel/src/dns.rs
View File

@@ -40,7 +40,7 @@ static DOH_CANARY_DOMAIN: LazyLock<DomainName> = LazyLock::new(|| {
});
pub struct StubResolver {
fqdn_to_ips: HashMap<DomainName, Vec<IpAddr>>,
fqdn_to_ips: BTreeMap<(DomainName, ResourceId), Vec<IpAddr>>,
ips_to_fqdn: HashMap<IpAddr, (DomainName, ResourceId)>,
ip_provider: IpProvider,
/// All DNS resources we know about, indexed by the glob pattern they match against.
@@ -166,22 +166,16 @@ impl StubResolver {
///
/// Semantically, this is like a PTR query, i.e. we check whether we handed out this IP as part of answering a DNS query for one of our resources.
/// This is in the hot-path of packet routing and must be fast!
pub(crate) fn resolve_resource_by_ip(&self, ip: &IpAddr) -> Option<ResourceId> {
let (_, resource_id) = self.ips_to_fqdn.get(ip)?;
Some(*resource_id)
pub(crate) fn resolve_resource_by_ip(&self, ip: &IpAddr) -> Option<&(DomainName, ResourceId)> {
self.ips_to_fqdn.get(ip)
}
pub(crate) fn get_fqdn(&self, ip: &IpAddr) -> Option<(&DomainName, &Vec<IpAddr>)> {
let (fqdn, _) = self.ips_to_fqdn.get(ip)?;
let ips = self.fqdn_to_ips.get(fqdn);
debug_assert!(
ips.is_some(),
"fqdn_to_ips and ips_to_fqdn are inconsistent"
);
Some((fqdn, ips?))
pub(crate) fn resolved_resources(
&self,
) -> impl Iterator<Item = (&DomainName, &ResourceId, &Vec<IpAddr>)> + '_ {
self.fqdn_to_ips
.iter()
.map(|((domain, resource), ips)| (domain, resource, ips))
}
pub(crate) fn add_resource(&mut self, id: ResourceId, pattern: String) -> bool {
@@ -221,7 +215,7 @@ impl StubResolver {
fn get_or_assign_ips(&mut self, fqdn: DomainName, resource_id: ResourceId) -> Vec<IpAddr> {
let ips = self
.fqdn_to_ips
.entry(fqdn.clone())
.entry((fqdn.clone(), resource_id))
.or_insert_with(|| {
let mut ips = self.ip_provider.get_n_ipv4(4);
ips.extend_from_slice(&self.ip_provider.get_n_ipv6(4));

63
rust/connlib/tunnel/src/gateway.rs
View File

@@ -1,6 +1,5 @@
use crate::messages::{
gateway::ResourceDescription, Answer, IceCredentials, ResolveRequest, SecretKey,
};
use crate::messages::gateway::ResourceDescription;
use crate::messages::{Answer, IceCredentials, ResolveRequest, SecretKey};
use crate::utils::earliest;
use crate::{p2p_control, GatewayEvent};
use crate::{peer::ClientOnGateway, peer_store::PeerStore};
@@ -8,7 +7,7 @@ use anyhow::{Context, Result};
use boringtun::x25519::PublicKey;
use chrono::{DateTime, Utc};
use connlib_model::{ClientId, DomainName, RelayId, ResourceId};
use firezone_logging::{anyhow_dyn_err, telemetry_span};
use firezone_logging::anyhow_dyn_err;
use ip_network::{Ipv4Network, Ipv6Network};
use ip_packet::{FzP2pControlSlice, IpPacket};
use secrecy::{ExposeSecret as _, Secret};
@@ -246,7 +245,7 @@ impl GatewayState {
now,
)?;
let result = self.allow_access(client_id, ipv4, ipv6, expires_at, resource, None, now);
let result = self.allow_access(client_id, ipv4, ipv6, expires_at, resource, None);
debug_assert!(
result.is_ok(),
"`allow_access` should never fail without a `DnsResourceEntry`"
@@ -255,26 +254,6 @@ impl GatewayState {
Ok(())
}
pub fn refresh_translation(
&mut self,
client: ClientId,
resource_id: ResourceId,
name: DomainName,
resolved_ips: Vec<IpAddr>,
now: Instant,
) {
let _span = telemetry_span!("refresh_translation").entered();
let Some(peer) = self.peers.get_mut(&client) else {
return;
};
if let Err(e) = peer.refresh_translation(name.clone(), resource_id, resolved_ips, now) {
tracing::warn!(error = anyhow_dyn_err(&e), rid = %resource_id, %name, "Failed to refresh DNS resource IP translations");
};
}
#[expect(clippy::too_many_arguments)]
pub fn allow_access(
&mut self,
client: ClientId,
@@ -283,7 +262,6 @@ impl GatewayState {
expires_at: Option<DateTime<Utc>>,
resource: ResourceDescription,
dns_resource_nat: Option<DnsResourceNatEntry>,
now: Instant,
) -> anyhow::Result<()> {
let peer = self
.peers
@@ -296,9 +274,8 @@ impl GatewayState {
peer.setup_nat(
entry.domain,
resource.id(),
&entry.resolved_ips,
entry.proxy_ips,
now,
BTreeSet::from_iter(entry.resolved_ips),
BTreeSet::from_iter(entry.proxy_ips),
)?;
}
@@ -311,23 +288,25 @@ impl GatewayState {
pub fn handle_domain_resolved(
&mut self,
req: ResolveDnsRequest,
addresses: Vec<IpAddr>,
resolve_result: Result<Vec<IpAddr>>,
now: Instant,
) -> anyhow::Result<()> {
use p2p_control::dns_resource_nat;
let nat_status = self
.peers
.get_mut(&req.client)
.context("Unknown peer")?
.setup_nat(
req.domain.clone(),
req.resource,
&addresses,
req.proxy_ips,
now,
)
.map(|()| dns_resource_nat::NatStatus::Active)
let nat_status = resolve_result
.and_then(|addresses| {
self.peers
.get_mut(&req.client)
.context("Unknown peer")?
.setup_nat(
req.domain.clone(),
req.resource,
BTreeSet::from_iter(addresses),
BTreeSet::from_iter(req.proxy_ips),
)?;
Ok(dns_resource_nat::NatStatus::Active)
})
.unwrap_or_else(|e| {
tracing::warn!(
error = anyhow_dyn_err(&e),

25
rust/connlib/tunnel/src/lib.rs
View File

@@ -5,7 +5,6 @@
#![cfg_attr(test, allow(clippy::unwrap_used))]
use crate::messages::{Offer, ResolveRequest, SecretKey};
use bimap::BiMap;
use chrono::Utc;
use connlib_model::{ClientId, DomainName, GatewayId, PublicKey, ResourceId, ResourceView};
@@ -295,25 +294,6 @@ pub enum ClientEvent {
resource: ResourceId,
connected_gateway_ids: BTreeSet<GatewayId>,
},
RequestAccess {
/// The resource we want to access.
resource_id: ResourceId,
/// The gateway we want to access the resource through.
gateway_id: GatewayId,
/// In the case of a DNS resource, its domain and the IPs we assigned to it.
maybe_domain: Option<ResolveRequest>,
},
RequestConnection {
/// The gateway we want to establish a connection to.
gateway_id: GatewayId,
/// The connection "offer". Contains our ICE credentials.
offer: Offer,
preshared_key: SecretKey,
/// The resource we want to access.
resource_id: ResourceId,
/// In the case of a DNS resource, its domain and the IPs we assigned to it.
maybe_domain: Option<ResolveRequest>,
},
/// The list of resources has changed and UI clients may have to be updated.
ResourcesChanged {
resources: Vec<ResourceView>,
@@ -349,11 +329,6 @@ pub enum GatewayEvent {
conn_id: ClientId,
candidates: BTreeSet<String>,
},
RefreshDns {
name: DomainName,
conn_id: ClientId,
resource_id: ResourceId,
},
ResolveDns(ResolveDnsRequest),
}

54
rust/connlib/tunnel/src/messages.rs
View File

@@ -2,9 +2,9 @@
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use chrono::{serde::ts_seconds, DateTime, Utc};
use connlib_model::{GatewayId, RelayId, ResourceId};
use connlib_model::RelayId;
use ip_network::IpNetwork;
use secrecy::{ExposeSecret, Secret};
use secrecy::{ExposeSecret as _, Secret};
use serde::{Deserialize, Serialize};
use std::fmt;
@@ -46,56 +46,12 @@ impl PartialEq for Peer {
}
}
/// Represent a connection request from a client to a given resource.
///
/// While this is a client-only message it's hosted in common since the tunnel
/// makes use of this message type.
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct RequestConnection {
/// Gateway id for the connection
pub gateway_id: GatewayId,
/// Resource id the request is for.
pub resource_id: ResourceId,
/// The preshared key the client generated for the connection that it is trying to establish.
pub client_preshared_key: SecretKey,
pub client_payload: ClientPayload,
}
#[derive(Debug, Deserialize, Serialize, Clone, PartialEq, Eq)]
pub struct ResolveRequest {
pub name: DomainName,
pub proxy_ips: Vec<IpAddr>,
}
#[derive(Debug, Deserialize, Serialize, Clone, PartialEq)]
pub struct ClientPayload {
pub ice_parameters: Offer,
pub domain: Option<ResolveRequest>,
}
/// Represent a request to reuse an existing gateway connection from a client to a given resource.
///
/// While this is a client-only message it's hosted in common since the tunnel
/// make use of this message type.
#[derive(Debug, Deserialize, Serialize, Clone, PartialEq, Eq)]
pub struct ReuseConnection {
/// Resource id the request is for.
pub resource_id: ResourceId,
/// Id of the gateway we want to reuse
pub gateway_id: GatewayId,
/// Payload that the gateway will receive
pub payload: Option<ResolveRequest>,
}
// Custom implementation of partial eq to ignore client_rtc_sdp
impl PartialEq for RequestConnection {
fn eq(&self, other: &Self) -> bool {
self.resource_id == other.resource_id
}
}
impl Eq for RequestConnection {}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
pub struct Answer {
pub username: String,
@@ -145,15 +101,9 @@ pub struct ConnectionAccepted {
pub ice_parameters: Answer,
}
#[derive(Debug, Deserialize, Serialize, Clone, PartialEq)]
pub struct ResourceAccepted {
pub domain_response: DomainResponse,
}
#[derive(Debug, Deserialize, Serialize, Clone, PartialEq)]
pub enum GatewayResponse {
ConnectionAccepted(ConnectionAccepted),
ResourceAccepted(ResourceAccepted),
}
#[derive(Debug, Deserialize, Serialize, Clone, PartialEq, Eq, Hash)]

140
rust/connlib/tunnel/src/messages/client.rs
View File

@@ -1,12 +1,10 @@
//! Client related messages that are needed within connlib
use crate::messages::{
GatewayResponse, Interface, Key, Relay, RelaysPresence, RequestConnection, ReuseConnection,
};
use crate::messages::{IceCredentials, Interface, Key, Relay, RelaysPresence, SecretKey};
use connlib_model::{GatewayId, ResourceId, Site, SiteId};
use ip_network::IpNetwork;
use serde::{Deserialize, Serialize};
use std::{collections::BTreeSet, net::IpAddr};
use std::collections::BTreeSet;
/// Description of a resource that maps to a DNS record.
#[derive(Debug, Deserialize)]
@@ -85,21 +83,46 @@ pub struct ConfigUpdate {
pub interface: Interface,
}
#[derive(Debug, Deserialize)]
pub struct ConnectionDetails {
#[derive(Debug, Deserialize, Clone)]
pub struct FlowCreated {
pub resource_id: ResourceId,
pub gateway_id: GatewayId,
pub gateway_remote_ip: IpAddr,
pub gateway_public_key: Key,
#[serde(rename = "gateway_group_id")]
pub site_id: SiteId,
pub preshared_key: SecretKey,
pub client_ice_credentials: IceCredentials,
pub gateway_ice_credentials: IceCredentials,
}
#[derive(Debug, Deserialize)]
pub struct Connect {
pub gateway_payload: GatewayResponse,
#[derive(Debug, Deserialize, Clone)]
pub struct FlowCreationFailed {
pub resource_id: ResourceId,
pub gateway_public_key: Key,
pub persistent_keepalive: u64,
pub reason: FailReason,
#[serde(default)]
pub violated_properties: Vec<ViolatedProperty>,
}
#[derive(Debug, Deserialize, Clone)]
#[serde(rename_all = "snake_case")]
pub enum FailReason {
NotFound,
Offline,
Forbidden,
#[serde(other)]
Unknown,
}
#[derive(Debug, Deserialize, Clone)]
#[serde(rename_all = "snake_case")]
pub enum ViolatedProperty {
RemoteIpLocationRegion,
RemoteIp,
ProviderId,
CurrentUtcDatetime,
ClientVerified,
#[serde(other)]
Unknown,
}
// These messages are the messages that can be received
@@ -119,6 +142,9 @@ pub enum IngressMessages {
ConfigChanged(ConfigUpdate),
RelaysPresence(RelaysPresence),
FlowCreated(FlowCreated),
FlowCreationFailed(FlowCreationFailed),
}
#[derive(Debug, Serialize)]
@@ -137,25 +163,15 @@ pub struct GatewayIceCandidates {
pub candidates: Vec<String>,
}
/// The replies that can arrive from the channel by a client
#[derive(Debug, Deserialize)]
#[serde(untagged)]
pub enum ReplyMessages {
ConnectionDetails(ConnectionDetails),
Connect(Connect),
}
// These messages can be sent from a client to a control pane
#[derive(Debug, Serialize)]
#[serde(rename_all = "snake_case", tag = "event", content = "payload")]
// enum_variant_names: These are the names in the portal!
pub enum EgressMessages {
PrepareConnection {
CreateFlow {
resource_id: ResourceId,
connected_gateway_ids: BTreeSet<GatewayId>,
},
RequestConnection(RequestConnection),
ReuseConnection(ReuseConnection),
/// Candidates that can be used by the addressed gateways.
BroadcastIceCandidates(GatewaysIceCandidates),
/// Candidates that should no longer be used by the addressed gateways.
@@ -259,46 +275,12 @@ mod tests {
}
#[test]
fn can_deserialize_connect_reply() {
let json = r#"{
"resource_id": "ea6570d1-47c7-49d2-9dc3-efff1c0c9e0b",
"gateway_public_key": "dvy0IwyxAi+txSbAdT7WKgf7K4TekhKzrnYwt5WfbSM=",
"gateway_payload": {
"ConnectionAccepted":{
"domain_response":{
"address":[
"2607:f8b0:4008:804::200e",
"142.250.64.206"
],
"domain":"google.com"
},
"ice_parameters":{
"username":"tGeqOjtGuPzPpuOx",
"password":"pMAxxTgHHSdpqHRzHGNvuNsZinLrMxwe"
}
}
},
"persistent_keepalive": 25
}"#;
fn can_deserialize_flow_created() {
let json = r#"{"event":"flow_created","ref":null,"topic":"client","payload":{"gateway_group_id":"ef42a07f-87d0-40da-baa7-e881e619ea1c","gateway_id":"d263d490-a0bb-452a-8990-01d27a1f1144","resource_id":"733e8d14-c18d-4931-af30-3639fa09c0c0","preshared_key":"anX2T9RH9mimT5Xd5+HqNGV0bfCodWDHQch1DLiFNls=","client_ice_credentials":{"username":"resc","password":"rqi3ibvfikfaxj3wgp7muh"},"gateway_ice_credentials":{"username":"jbi4","password":"a6oeevhlutevykcifd5r2a"},"gateway_public_key":"uMBCkAxTewfSgypIyxdQ18uCi84HLtKmQJy0wvQrYWY="}}"#;
let message = serde_json::from_str::<ReplyMessages>(json).unwrap();
let message = serde_json::from_str::<IngressMessages>(json).unwrap();
assert!(matches!(message, ReplyMessages::Connect(_)))
}
#[test]
fn can_deserialize_connection_details_reply() {
let json = r#"
{
"resource_id": "f16ecfa0-a94f-4bfd-a2ef-1cc1f2ef3da3",
"gateway_id": "73037362-715d-4a83-a749-f18eadd970e6",
"gateway_remote_ip": "172.28.0.1",
"gateway_group_id": "bf56f32d-7b2c-4f5d-a784-788977d014a4"
}"#;
let message = serde_json::from_str::<ReplyMessages>(json).unwrap();
assert!(matches!(message, ReplyMessages::ConnectionDetails(_)));
assert!(matches!(message, IngressMessages::FlowCreated(_)));
}
#[test]
@@ -415,12 +397,42 @@ mod tests {
}
#[test]
fn serialize_prepare_connection_message() {
let message = EgressMessages::PrepareConnection {
fn can_deserialize_unknown_flow_creation_failed_err() {
let json = r#"{"event":"flow_creation_failed","ref":null,"topic":"client","payload":{"resource_id":"f16ecfa0-a94f-4bfd-a2ef-1cc1f2ef3da3","reason":"foobar"}}"#;
let message = serde_json::from_str::<IngressMessages>(json).unwrap();
assert!(matches!(
message,
IngressMessages::FlowCreationFailed(FlowCreationFailed {
reason: FailReason::Unknown,
..
})
));
}
#[test]
fn can_deserialize_known_flow_creation_failed_err() {
let json = r#"{"event":"flow_creation_failed","ref":null,"topic":"client","payload":{"resource_id":"f16ecfa0-a94f-4bfd-a2ef-1cc1f2ef3da3","reason":"offline"}}"#;
let message = serde_json::from_str::<IngressMessages>(json).unwrap();
assert!(matches!(
message,
IngressMessages::FlowCreationFailed(FlowCreationFailed {
reason: FailReason::Offline,
..
})
));
}
#[test]
fn serialize_create_flow_message() {
let message = EgressMessages::CreateFlow {
resource_id: "f16ecfa0-a94f-4bfd-a2ef-1cc1f2ef3da3".parse().unwrap(),
connected_gateway_ids: BTreeSet::new(),
};
let expected_json = r#"{"event":"prepare_connection","payload":{"resource_id":"f16ecfa0-a94f-4bfd-a2ef-1cc1f2ef3da3","connected_gateway_ids":[]}}"#;
let expected_json = r#"{"event":"create_flow","payload":{"resource_id":"f16ecfa0-a94f-4bfd-a2ef-1cc1f2ef3da3","connected_gateway_ids":[]}}"#;
let actual_json = serde_json::to_string(&message).unwrap();
assert_eq!(actual_json, expected_json);

6
rust/connlib/tunnel/src/messages/gateway.rs
View File

@@ -1,8 +1,8 @@
//! Gateway related messages that are needed within connlib
use crate::messages::{
GatewayResponse, IceCredentials, Interface, Key, Offer, Peer, Relay, RelaysPresence,
ResolveRequest, SecretKey,
GatewayResponse, IceCredentials, Interface, Key, Peer, Relay, RelaysPresence, ResolveRequest,
SecretKey,
};
use chrono::{serde::ts_seconds_option, DateTime, Utc};
use connlib_model::{ClientId, ResourceId};
@@ -13,6 +13,8 @@ use std::{
net::{Ipv4Addr, Ipv6Addr},
};
use super::Offer;
pub type Filters = Vec<Filter>;
/// Description of a resource that maps to a DNS record.

2
rust/connlib/tunnel/src/p2p_control.rs
View File

@@ -15,8 +15,6 @@ use ip_packet::FzP2pEventType;
pub const ASSIGNED_IPS_EVENT: FzP2pEventType = FzP2pEventType::new(0);
pub const DOMAIN_STATUS_EVENT: FzP2pEventType = FzP2pEventType::new(1);
/// The namespace for the DNS resource NAT protocol.
#[cfg_attr(not(test), expect(dead_code, reason = "Will be used soon."))]
pub mod dns_resource_nat {
use super::*;
use anyhow::{Context as _, Result};

502
rust/connlib/tunnel/src/peer.rs
View File

@@ -1,7 +1,7 @@
use std::collections::{hash_map, BTreeMap, HashMap, HashSet, VecDeque};
use std::collections::{hash_map, BTreeMap, BTreeSet, HashMap, HashSet, VecDeque};
use std::iter;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use std::time::{Duration, Instant};
use std::time::Instant;
use crate::client::{IPV4_RESOURCES, IPV6_RESOURCES};
use crate::messages::gateway::ResourceDescription;
@@ -11,7 +11,6 @@ use connlib_model::{ClientId, DomainName, GatewayId, ResourceId};
use ip_network::{IpNetwork, Ipv4Network, Ipv6Network};
use ip_network_table::IpNetworkTable;
use ip_packet::IpPacket;
use itertools::Itertools;
use rangemap::RangeInclusiveSet;
use crate::utils::network_contains_network;
@@ -117,17 +116,11 @@ impl GatewayOnClient {
}
impl GatewayOnClient {
pub(crate) fn new(
id: GatewayId,
ips: &[IpNetwork],
resource: HashSet<ResourceId>,
) -> GatewayOnClient {
let mut allowed_ips = IpNetworkTable::new();
for ip in ips {
allowed_ips.insert(*ip, resource.clone());
pub(crate) fn new(id: GatewayId) -> GatewayOnClient {
GatewayOnClient {
id,
allowed_ips: IpNetworkTable::new(),
}
GatewayOnClient { id, allowed_ips }
}
}
@@ -167,65 +160,14 @@ impl ClientOnGateway {
[IpAddr::from(self.ipv4), IpAddr::from(self.ipv6)]
}
pub(crate) fn refresh_translation(
&mut self,
name: DomainName,
resource_id: ResourceId,
resolved_ips: Vec<IpAddr>,
now: Instant,
) -> Result<()> {
let resource_on_gateway = self
.resources
.get_mut(&resource_id)
.context("Unknown resource")?;
let domains = match resource_on_gateway {
ResourceOnGateway::Dns { domains, .. } => domains,
ResourceOnGateway::Cidr { .. } => {
bail!("Cannot refresh translation for CIDR resource")
}
ResourceOnGateway::Internet { .. } => {
bail!("Cannot refresh translation for Internet resource")
}
};
let old_ips: HashSet<&IpAddr> =
HashSet::from_iter(self.permanent_translations.values().filter_map(|state| {
(state.name == name && state.resource_id == resource_id)
.then_some(&state.resolved_ip)
}));
let new_ips: HashSet<&IpAddr> = HashSet::from_iter(resolved_ips.iter());
if old_ips == new_ips {
return Ok(());
}
domains.insert(
name.clone(),
resolved_ips.iter().copied().map_into().collect_vec(),
);
let proxy_ips = self
.permanent_translations
.iter()
.filter_map(|(k, state)| {
(state.name == name && state.resource_id == resource_id).then_some(*k)
})
.collect_vec();
self.setup_nat(name, resource_id, &resolved_ips, proxy_ips, now)?;
Ok(())
}
/// Setup the NAT for a particular domain within a wildcard DNS resource.
#[tracing::instrument(level = "debug", skip_all, fields(cid = %self.id))]
pub(crate) fn setup_nat(
&mut self,
name: DomainName,
resource_id: ResourceId,
resolved_ips: &[IpAddr],
proxy_ips: Vec<IpAddr>,
now: Instant,
resolved_ips: BTreeSet<IpAddr>,
proxy_ips: BTreeSet<IpAddr>,
) -> Result<()> {
let resource = self
.resources
@@ -241,33 +183,31 @@ impl ClientOnGateway {
anyhow::ensure!(crate::dns::is_subdomain(&name, address));
let mapped_ipv4 = mapped_ipv4(resolved_ips);
let mapped_ipv6 = mapped_ipv6(resolved_ips);
let mapped_ipv4 = mapped_ipv4(&resolved_ips);
let mapped_ipv6 = mapped_ipv6(&resolved_ips);
let ipv4_maps = proxy_ips
.iter()
.filter(|ip| ip.is_ipv4())
.zip(mapped_ipv4.into_iter().cycle());
.zip(mapped_ipv4.iter().cycle().copied());
let ipv6_maps = proxy_ips
.iter()
.filter(|ip| ip.is_ipv6())
.zip(mapped_ipv6.into_iter().cycle());
.zip(mapped_ipv6.iter().cycle().copied());
let ip_maps = ipv4_maps.chain(ipv6_maps);
for (proxy_ip, real_ip) in ip_maps {
tracing::debug!(%name, %proxy_ip, %real_ip);
self.permanent_translations.insert(
*proxy_ip,
TranslationState::new(resource_id, name.clone(), real_ip, now),
);
self.permanent_translations
.insert(*proxy_ip, TranslationState::new(resource_id, real_ip));
}
tracing::debug!(domain = %name, ?resolved_ips, ?proxy_ips, "Set up DNS resource NAT");
domains.insert(name, resolved_ips.to_vec());
domains.insert(name, resolved_ips);
self.recalculate_filters();
Ok(())
@@ -302,39 +242,6 @@ impl ClientOnGateway {
}
pub(crate) fn handle_timeout(&mut self, now: Instant) {
let expired_translations = self
.permanent_translations
.iter()
.filter(|(_, state)| state.is_expired(now));
let mut for_refresh = HashSet::new();
for (proxy_ip, expired_state) in expired_translations {
let domain = &expired_state.name;
let resource_id = expired_state.resource_id;
let resolved_ip = expired_state.resolved_ip;
// Only refresh DNS for a domain if all of the resolved IPs stop responding in order to not kill existing connections.
if self
.permanent_translations
.values()
.filter(|state| state.resource_id == resource_id && state.name == domain)
.all(|state| state.no_incoming_in_120s(now))
{
tracing::debug!(%domain, conn_id = %self.id, %resource_id, %resolved_ip, %proxy_ip, "Refreshing DNS");
for_refresh.insert((expired_state.name.clone(), expired_state.resource_id));
}
}
for (name, resource_id) in for_refresh {
self.buffered_events.push_back(GatewayEvent::RefreshDns {
name,
conn_id: self.id,
resource_id,
});
}
self.nat_table.handle_timeout(now);
}
@@ -433,8 +340,6 @@ impl ClientOnGateway {
.context("Failed to translate packet")?;
packet.update_checksum();
state.on_outgoing_traffic(now);
Ok(packet)
}
@@ -471,12 +376,6 @@ impl ClientOnGateway {
};
let mut packet = packet.translate_source(self.ipv4, self.ipv6, proto, ip)?;
self.permanent_translations
.get_mut(&ip)
.context("No translation state for outgoing packet")?
.on_incoming_traffic(now);
packet.update_checksum();
Ok(packet)
@@ -554,7 +453,7 @@ enum ResourceOnGateway {
},
Dns {
address: String,
domains: HashMap<DomainName, Vec<IpAddr>>,
domains: HashMap<DomainName, BTreeSet<IpAddr>>,
filters: Filters,
expires_at: Option<DateTime<Utc>>,
},
@@ -659,100 +558,28 @@ impl ResourceOnGateway {
struct TranslationState {
/// Which (DNS) resource we belong to.
resource_id: ResourceId,
/// The concrete domain we have resolved (could be a sub-domain of a `*` or `?` resource).
name: DomainName,
/// The IP we have resolved for the domain.
resolved_ip: IpAddr,
/// When we've last received a packet from the resolved IP.
last_incoming: Option<Instant>,
/// When we've sent the first packet to the resolved IP.
first_outgoing: Option<Instant>,
/// When we've last sent a packet to the resolved IP.
last_outgoing: Option<Instant>,
/// When was this translation created
created_at: Instant,
/// When we first detected that we aren't getting any responses from this IP.
///
/// This is set upon outgoing traffic if we haven't received inbound traffic for a while.
/// We don't want to immediately trigger a refresh in that case because protocols like TCP and ICMP have responses.
/// Thus, a DNS refresh is triggered after a grace-period of 1s after the packet that detected the missing responses.
ack_grace_period_started_at: Option<Instant>,
}
impl TranslationState {
const USED_WINDOW: Duration = Duration::from_secs(10);
fn new(resource_id: ResourceId, name: DomainName, resolved_ip: IpAddr, now: Instant) -> Self {
fn new(resource_id: ResourceId, resolved_ip: IpAddr) -> Self {
Self {
resource_id,
name,
resolved_ip,
created_at: now,
last_incoming: None,
first_outgoing: None,
last_outgoing: None,
ack_grace_period_started_at: None,
}
}
fn is_expired(&self, now: Instant) -> bool {
// Note: we don't need to check that it's used here because the ack grace period already implies it
self.ack_grace_period_expired(now) && self.no_incoming_in_120s(now)
}
fn ack_grace_period_expired(&self, now: Instant) -> bool {
self.ack_grace_period_started_at
.is_some_and(|missing_responses_detected_at| {
now.duration_since(missing_responses_detected_at) >= Duration::from_secs(1)
})
}
fn no_incoming_in_120s(&self, now: Instant) -> bool {
const CONNTRACK_UDP_STREAM_TIMEOUT: Duration = Duration::from_secs(120);
if let Some(last_incoming) = self.last_incoming {
now.duration_since(last_incoming) >= CONNTRACK_UDP_STREAM_TIMEOUT
} else {
now.duration_since(self.created_at) >= CONNTRACK_UDP_STREAM_TIMEOUT
}
}
fn on_incoming_traffic(&mut self, now: Instant) {
self.last_incoming = Some(now);
self.ack_grace_period_started_at = None;
}
fn on_outgoing_traffic(&mut self, now: Instant) {
// We need this because it means that if a packet arrives at some point less than 120s but more than 110s
// we still start the grace period so that the connection expires at some point after 120s
let with_this_packet_the_connection_will_be_considered_used_when_it_expires =
self.no_incoming_in_120s(now + Self::USED_WINDOW);
if self.ack_grace_period_started_at.is_none()
&& with_this_packet_the_connection_will_be_considered_used_when_it_expires
{
self.ack_grace_period_started_at = Some(now);
}
self.last_outgoing = Some(now);
if self.first_outgoing.is_some() {
return;
}
self.first_outgoing = Some(now);
}
}
fn ipv4_addresses(ip: &[IpAddr]) -> Vec<IpAddr> {
ip.iter().filter(|ip| ip.is_ipv4()).copied().collect_vec()
fn ipv4_addresses(ip: &BTreeSet<IpAddr>) -> BTreeSet<IpAddr> {
ip.iter().filter(|ip| ip.is_ipv4()).copied().collect()
}
fn ipv6_addresses(ip: &[IpAddr]) -> Vec<IpAddr> {
ip.iter().filter(|ip| ip.is_ipv6()).copied().collect_vec()
fn ipv6_addresses(ip: &BTreeSet<IpAddr>) -> BTreeSet<IpAddr> {
ip.iter().filter(|ip| ip.is_ipv6()).copied().collect()
}
fn mapped_ipv4(ips: &[IpAddr]) -> Vec<IpAddr> {
fn mapped_ipv4(ips: &BTreeSet<IpAddr>) -> BTreeSet<IpAddr> {
if !ipv4_addresses(ips).is_empty() {
ipv4_addresses(ips)
} else {
@@ -760,7 +587,7 @@ fn mapped_ipv4(ips: &[IpAddr]) -> Vec<IpAddr> {
}
}
fn mapped_ipv6(ips: &[IpAddr]) -> Vec<IpAddr> {
fn mapped_ipv6(ips: &BTreeSet<IpAddr>) -> BTreeSet<IpAddr> {
if !ipv6_addresses(ips).is_empty() {
ipv6_addresses(ips)
} else {
@@ -786,7 +613,8 @@ fn insert_filters<'a>(
#[cfg(test)]
mod tests {
use std::{
net::{IpAddr, Ipv4Addr, Ipv6Addr},
collections::BTreeSet,
net::{Ipv4Addr, Ipv6Addr},
time::{Duration, Instant},
};
@@ -797,7 +625,7 @@ mod tests {
use connlib_model::{ClientId, ResourceId};
use ip_network::{IpNetwork, Ipv4Network};
use super::{ClientOnGateway, TranslationState};
use super::ClientOnGateway;
#[test]
fn gateway_filters_expire_individually() {
@@ -864,273 +692,6 @@ mod tests {
assert!(peer.ensure_allowed_dst(&udp_packet).is_err());
}
#[test]
fn initial_translation_state_is_not_expired() {
let now = Instant::now();
let state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
assert!(!state.is_expired(now));
}
#[test]
fn translation_state_is_not_used_but_expired_after_120s() {
let mut now = Instant::now();
let state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_secs(121);
assert!(!state.is_expired(now));
}
#[test]
fn translation_state_is_used_and_expired_after_120s_with_outgoing_packets() {
let mut now = Instant::now();
let mut state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_secs(120);
state.on_outgoing_traffic(now);
now += Duration::from_secs(1);
assert!(state.is_expired(now));
}
#[test]
fn translation_state_is_used_and_expired_after_121s_with_outgoing_packets() {
let mut now = Instant::now();
let mut state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_secs(121);
state.on_outgoing_traffic(now);
now += Duration::from_secs(1);
assert!(state.is_expired(now));
}
#[test]
fn translation_state_is_not_expired_with_incoming_packets() {
let mut now = Instant::now();
let mut state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_secs(120);
state.on_incoming_traffic(now);
now += Duration::from_secs(1);
assert!(!state.is_expired(now));
}
#[test]
fn translation_state_doesnt_expire_with_incoming_and_outgoing_packets() {
let mut now = Instant::now();
let mut state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_secs(120);
state.on_outgoing_traffic(now);
now += Duration::from_millis(200);
state.on_incoming_traffic(now);
now += Duration::from_secs(1);
assert!(!state.is_expired(now));
}
#[test]
fn translation_state_still_has_grace_period_after_incoming_traffic() {
let mut now = Instant::now();
let mut state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_secs(120);
state.on_outgoing_traffic(now);
now += Duration::from_millis(200);
state.on_incoming_traffic(now);
now += Duration::from_secs(120);
state.on_outgoing_traffic(now);
assert!(!state.is_expired(now));
}
#[test]
fn translation_state_still_expires_after_grace_period_after_incoming_traffic_resetted_it() {
let mut now = Instant::now();
let mut state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_secs(120);
state.on_outgoing_traffic(now);
now += Duration::from_millis(200);
state.on_incoming_traffic(now);
now += Duration::from_secs(120);
state.on_outgoing_traffic(now);
now += Duration::from_secs(1);
assert!(state.is_expired(now));
}
#[test]
fn translation_state_doesnt_expire_with_first_packet_after_silence() {
let mut now = Instant::now();
let mut state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_secs(120);
state.on_outgoing_traffic(now);
assert!(!state.is_expired(now));
}
#[test]
fn translation_state_expires_after_silence_even_with_multiple_packets() {
let mut now = Instant::now();
let mut state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_secs(120);
state.on_outgoing_traffic(now);
now += Duration::from_secs(5);
state.on_outgoing_traffic(now);
assert!(state.is_expired(now));
}
#[test]
fn translation_doesnt_expire_before_expected_period() {
let mut now = Instant::now();
let mut state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_secs(110);
state.on_outgoing_traffic(now);
now += Duration::from_secs(5);
assert!(!state.is_expired(now));
}
#[test]
fn translation_expire_after_expected_period() {
let mut now = Instant::now();
let mut state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_secs(110);
state.on_outgoing_traffic(now);
now += Duration::from_secs(5);
state.on_outgoing_traffic(now);
now += Duration::from_secs(5);
assert!(state.is_expired(now));
}
#[test]
fn incoming_traffic_prevents_expiration() {
let mut now = Instant::now();
let mut state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_secs(120);
state.on_outgoing_traffic(now);
now += Duration::from_millis(500);
state.on_incoming_traffic(now);
now += Duration::from_secs(5);
assert!(!state.is_expired(now));
}
#[test]
fn translation_state_doesnt_expire_with_packet_that_didnt_had_time_to_be_responded() {
let mut now = Instant::now();
let mut state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_millis(119990);
state.on_outgoing_traffic(now);
now += Duration::from_millis(20);
assert!(!state.is_expired(now));
}
#[test]
fn translation_state_expires_with_packet_that_had_time_to_be_responded() {
let mut now = Instant::now();
let mut state = TranslationState::new(
ResourceId::random(),
"example.com".parse().unwrap(),
IpAddr::V4(Ipv4Addr::LOCALHOST),
now,
);
now += Duration::from_millis(119990);
state.on_outgoing_traffic(now);
now += Duration::from_secs(1);
assert!(state.is_expired(now));
}
#[test]
fn dns_and_cidr_filters_dot_mix() {
let mut peer = ClientOnGateway::new(client_id(), source_v4_addr(), source_v6_addr());
@@ -1139,9 +700,8 @@ mod tests {
peer.setup_nat(
foo_name().parse().unwrap(),
resource_id(),
&[foo_real_ip().into()],
vec![foo_proxy_ip().into()],
Instant::now(),
BTreeSet::from([foo_real_ip().into()]),
BTreeSet::from([foo_proxy_ip().into()]),
)
.unwrap();
@@ -1200,9 +760,8 @@ mod tests {
peer.setup_nat(
foo_name().parse().unwrap(),
resource_id(),
&[foo_real_ip().into()],
vec![foo_proxy_ip().into()],
Instant::now(),
BTreeSet::from([foo_real_ip().into()]),
BTreeSet::from([foo_proxy_ip().into()]),
)
.unwrap();
@@ -1335,6 +894,7 @@ mod proptests {
use crate::messages::gateway::{PortRange, ResourceDescription, ResourceDescriptionCidr};
use crate::proptest::*;
use ip_packet::make::{icmp_request_packet, tcp_packet, udp_packet};
use itertools::Itertools as _;
use proptest::{
arbitrary::any,
collection, prop_oneof,

44
rust/connlib/tunnel/src/tests/reference.rs
View File

@@ -6,7 +6,7 @@ use super::{
};
use crate::{client, DomainName};
use crate::{dns::is_subdomain, proptest::relay_id};
use connlib_model::{GatewayId, RelayId, ResourceId, StaticSecret};
use connlib_model::{GatewayId, RelayId, StaticSecret};
use domain::base::Rtype;
use ip_network::{Ipv4Network, Ipv6Network};
use prop::sample::select;
@@ -361,10 +361,7 @@ impl ReferenceState {
}
}),
Transition::SendDnsQueries(queries) => {
let mut new_connections_via_gateways_udp_triggered =
BTreeMap::<_, BTreeSet<ResourceId>>::new();
let mut new_connections_via_gateways_tcp_triggered =
BTreeMap::<_, BTreeSet<ResourceId>>::new();
let mut new_connections = BTreeSet::new();
for query in queries {
// Some queries get answered locally.
@@ -400,29 +397,7 @@ impl ReferenceState {
{
tracing::debug!(%resource, %gateway, "Not connected yet, dropping packet");
let connected_resources = match query.transport {
DnsTransport::Udp => &mut new_connections_via_gateways_udp_triggered,
DnsTransport::Tcp => &mut new_connections_via_gateways_tcp_triggered,
}
.entry(gateway)
.or_default();
if state.client.inner().is_connected_gateway(gateway) {
connected_resources.insert(resource);
} else {
match query.transport {
DnsTransport::Udp => {
// As part of batch-processing DNS queries, only the first resource per gateway will be connected / authorized.
if connected_resources.is_empty() {
connected_resources.insert(resource);
}
}
DnsTransport::Tcp => {
// TCP has retries, so those will always connect.
connected_resources.insert(resource);
}
}
}
new_connections.insert((resource, gateway));
continue;
}
@@ -430,15 +405,10 @@ impl ReferenceState {
state.client.exec_mut(|client| client.on_dns_query(query));
}
for (gateway, resources) in new_connections_via_gateways_udp_triggered
.into_iter()
.chain(new_connections_via_gateways_tcp_triggered)
{
for resource in resources {
state.client.exec_mut(|client| {
client.connect_to_internet_or_cidr_resource(resource, gateway)
});
}
for (resource, gateway) in new_connections.into_iter() {
state.client.exec_mut(|client| {
client.connect_to_internet_or_cidr_resource(resource, gateway)
});
}
}
Transition::SendIcmpPacket {

24
rust/connlib/tunnel/src/tests/sim_client.rs
View File

@@ -433,7 +433,7 @@ pub struct RefClient {
/// The DNS resources the client is connected to.
#[debug(skip)]
pub(crate) connected_dns_resources: HashSet<(ResourceId, DomainName)>,
pub(crate) connected_dns_resources: HashSet<ResourceId>,
#[debug(skip)]
pub(crate) connected_gateways: BTreeSet<GatewayId>,
@@ -486,7 +486,7 @@ impl RefClient {
self.ipv6_routes.remove(resource);
self.connected_cidr_resources.remove(resource);
self.connected_dns_resources.retain(|(r, _)| r != resource);
self.connected_dns_resources.remove(resource);
if self.internet_resource.is_some_and(|r| &r == resource) {
self.connected_internet_resource = false;
@@ -675,7 +675,7 @@ impl RefClient {
return;
};
if self.is_connected_to_resource(resource, &dst) && self.is_tunnel_ip(src) {
if self.is_connected_to_resource(resource) && self.is_tunnel_ip(src) {
tracing::debug!("Connected to resource, expecting packet to be routed");
map(self)
.entry(gateway)
@@ -692,6 +692,7 @@ impl RefClient {
}
tracing::debug!("Not connected to resource, expecting to trigger connection intent");
self.connect_to_resource(resource, dst, gateway);
}
@@ -702,9 +703,9 @@ impl RefClient {
gateway: GatewayId,
) {
match destination {
Destination::DomainName { name, .. } => {
Destination::DomainName { .. } => {
if !self.disabled_resources.contains(&resource) {
self.connected_dns_resources.insert((resource, name));
self.connected_dns_resources.insert(resource);
self.connected_gateways.insert(gateway);
}
}
@@ -716,10 +717,6 @@ impl RefClient {
self.is_connected_to_cidr(resource) || self.is_connected_to_internet(resource)
}
pub(crate) fn is_connected_gateway(&self, gateway: GatewayId) -> bool {
self.connected_gateways.contains(&gateway)
}
pub(crate) fn connect_to_internet_or_cidr_resource(
&mut self,
resource: ResourceId,
@@ -769,17 +766,12 @@ impl RefClient {
.collect_vec()
}
fn is_connected_to_resource(&self, resource: ResourceId, destination: &Destination) -> bool {
fn is_connected_to_resource(&self, resource: ResourceId) -> bool {
if self.is_connected_to_internet_or_cidr(resource) {
return true;
}
let Destination::DomainName { name, .. } = destination else {
return false;
};
self.connected_dns_resources
.contains(&(resource, name.clone()))
self.connected_dns_resources.contains(&resource)
}
fn is_connected_to_internet(&self, id: ResourceId) -> bool {

199
rust/connlib/tunnel/src/tests/sut.rs
View File

@@ -9,18 +9,20 @@ use super::stub_portal::StubPortal;
use super::transition::{Destination, DnsQuery};
use super::unreachable_hosts::UnreachableHosts;
use crate::client::Resource;
use crate::dns::{self, is_subdomain};
use crate::gateway::DnsResourceNatEntry;
use crate::dns::is_subdomain;
use crate::messages::{IceCredentials, Key, SecretKey};
use crate::tests::assertions::*;
use crate::tests::flux_capacitor::FluxCapacitor;
use crate::tests::transition::Transition;
use crate::utils::earliest;
use crate::{messages::Interface, ClientEvent, GatewayEvent};
use connlib_model::{ClientId, GatewayId, RelayId};
use crate::{dns, messages::Interface, ClientEvent, GatewayEvent};
use connlib_model::{ClientId, GatewayId, PublicKey, RelayId};
use domain::base::iana::{Class, Rcode};
use domain::base::{Message, MessageBuilder, Record, RecordData, ToName as _, Ttl};
use firezone_logging::anyhow_dyn_err;
use secrecy::ExposeSecret as _;
use rand::distributions::DistString;
use rand::SeedableRng;
use sha2::Digest;
use snownet::Transmit;
use std::iter;
use std::{
@@ -411,13 +413,9 @@ impl TunnelTest {
);
continue 'outer;
}
if let Some(event) = self.client.exec_mut(|c| c.sut.poll_event()) {
self.on_client_event(
self.client.inner().id,
event,
&ref_state.portal,
&ref_state.global_dns_records,
);
self.on_client_event(self.client.inner().id, event, &ref_state.portal);
continue;
}
if let Some(query) = self.client.exec_mut(|c| c.sut.poll_dns_queries()) {
@@ -489,6 +487,7 @@ impl TunnelTest {
buffered_transmits.push_from(transmit, &self.client, now);
continue;
}
self.client.exec_mut(|sim| {
while let Some(packet) = sim.sut.poll_packets() {
sim.on_received_packet(packet)
@@ -675,13 +674,7 @@ impl TunnelTest {
}
}
fn on_client_event(
&mut self,
src: ClientId,
event: ClientEvent,
portal: &StubPortal,
global_dns_records: &DnsRecords,
) {
fn on_client_event(&mut self, src: ClientId, event: ClientEvent, portal: &StubPortal) {
let now = self.flux_capacitor.now();
match event {
@@ -710,46 +703,51 @@ impl TunnelTest {
})
}
ClientEvent::ConnectionIntent {
resource,
resource: resource_id,
connected_gateway_ids,
} => {
let (gateway, site) =
portal.handle_connection_intent(resource, connected_gateway_ids);
self.client
.exec_mut(|c| c.sut.on_routing_details(resource, gateway, site, now))
.unwrap()
.unwrap();
}
ClientEvent::RequestAccess {
resource_id,
gateway_id,
maybe_domain,
} => {
let (gateway_id, site_id) =
portal.handle_connection_intent(resource_id, connected_gateway_ids);
let gateway = self.gateways.get_mut(&gateway_id).expect("unknown gateway");
let maybe_entry = maybe_domain.map(|r| {
let resolved_ips = global_dns_records.domain_ips_iter(&r.name).collect();
DnsResourceNatEntry::new(r, resolved_ips)
});
let resource = portal.map_client_resource_to_gateway_resource(resource_id);
gateway.exec_mut(|g| {
g.sut
.allow_access(
self.client.inner().id,
let client_key = self.client.inner().sut.public_key();
let gateway_key = gateway.inner().sut.public_key();
let (preshared_key, client_ice, gateway_ice) =
make_preshared_key_and_ice(client_key, gateway_key);
gateway
.exec_mut(|g| {
g.sut.authorize_flow(
src,
client_key,
preshared_key.clone(),
client_ice.clone(),
gateway_ice.clone(),
self.client.inner().sut.tunnel_ip4().unwrap(),
self.client.inner().sut.tunnel_ip6().unwrap(),
None,
resource.clone(),
maybe_entry,
resource,
now,
)
.unwrap();
});
})
.unwrap();
if let Err(e) = self.client.exec_mut(|c| {
c.sut.handle_flow_created(
resource_id,
gateway_id,
gateway_key,
site_id,
preshared_key,
client_ice,
gateway_ice,
now,
)
}) {
tracing::error!("{e:#}")
};
}
ClientEvent::ResourcesChanged { .. } => {
tracing::warn!("Unimplemented");
}
@@ -780,75 +778,6 @@ impl TunnelTest {
c.ipv6_routes = config.ipv6_routes;
});
}
#[expect(deprecated, reason = "Will be deleted together with deprecated API")]
ClientEvent::RequestConnection {
gateway_id,
offer,
preshared_key,
resource_id,
maybe_domain,
} => {
let maybe_entry = maybe_domain.map(|r| {
let resolved_ips = global_dns_records.domain_ips_iter(&r.name).collect();
DnsResourceNatEntry::new(r, resolved_ips)
});
let resource = portal.map_client_resource_to_gateway_resource(resource_id);
let Some(gateway) = self.gateways.get_mut(&gateway_id) else {
tracing::error!("Unknown gateway");
return;
};
let client_id = self.client.inner().id;
let answer = gateway.exec_mut(|g| {
let answer = g
.sut
.accept(
client_id,
snownet::Offer {
session_key: preshared_key.expose_secret().0.into(),
credentials: snownet::Credentials {
username: offer.username,
password: offer.password,
},
},
self.client.inner().sut.public_key(),
now,
)
.unwrap();
g.sut
.allow_access(
self.client.inner().id,
self.client.inner().sut.tunnel_ip4().unwrap(),
self.client.inner().sut.tunnel_ip6().unwrap(),
None,
resource.clone(),
maybe_entry,
now,
)
.unwrap();
answer
});
self.client
.exec_mut(|c| {
c.sut.accept_answer(
snownet::Answer {
credentials: snownet::Credentials {
username: answer.username,
password: answer.password,
},
},
resource_id,
gateway.inner().sut.public_key(),
now,
)
})
.unwrap();
}
}
}
@@ -929,6 +858,35 @@ fn address_from_destination(destination: &Destination, state: &TunnelTest, src:
}
}
fn make_preshared_key_and_ice(
client_key: PublicKey,
gateway_key: PublicKey,
) -> (SecretKey, IceCredentials, IceCredentials) {
let secret_key = SecretKey::new(Key(hkdf("SECRET_KEY_DOMAIN_SEP", client_key, gateway_key)));
let client_ice = ice_creds("CLIENT_ICE_DOMAIN_SEP", client_key, gateway_key);
let gateway_ice = ice_creds("GATEWAY_ICE_DOMAIN_SEP", client_key, gateway_key);
(secret_key, client_ice, gateway_ice)
}
fn ice_creds(domain: &str, client_key: PublicKey, gateway_key: PublicKey) -> IceCredentials {
let mut rng = rand::rngs::StdRng::from_seed(hkdf(domain, client_key, gateway_key));
IceCredentials {
username: rand::distributions::Alphanumeric.sample_string(&mut rng, 4),
password: rand::distributions::Alphanumeric.sample_string(&mut rng, 12),
}
}
fn hkdf(domain: &str, client_key: PublicKey, gateway_key: PublicKey) -> [u8; 32] {
sha2::Sha256::default()
.chain_update(domain)
.chain_update(client_key.as_bytes())
.chain_update(gateway_key.as_bytes())
.finalize()
.into()
}
fn on_gateway_event(
src: GatewayId,
event: GatewayEvent,
@@ -948,13 +906,14 @@ fn on_gateway_event(
c.sut.remove_ice_candidate(src, candidate, now)
}
}),
event @ GatewayEvent::RefreshDns { .. } => {
tracing::warn!("Handling `{event:?}` is not yet implemented")
}
GatewayEvent::ResolveDns(r) => {
let resolved_ips = global_dns_records.domain_ips_iter(r.domain()).collect();
gateway.exec_mut(|g| g.sut.handle_domain_resolved(r, resolved_ips, now).unwrap())
gateway.exec_mut(|g| {
g.sut
.handle_domain_resolved(r, Ok(resolved_ips), now)
.unwrap()
})
}
}
}

126
rust/gateway/src/eventloop.rs
View File

@@ -1,7 +1,6 @@
use anyhow::Result;
use anyhow::{Context as _, Result};
use boringtun::x25519::PublicKey;
use connlib_model::DomainName;
use connlib_model::{ClientId, ResourceId};
#[cfg(not(target_os = "windows"))]
use dns_lookup::{AddrInfoHints, AddrInfoIter, LookupError};
use firezone_logging::{
@@ -14,7 +13,6 @@ use firezone_tunnel::messages::gateway::{
use firezone_tunnel::messages::{ConnectionAccepted, GatewayResponse, Interface, RelaysPresence};
use firezone_tunnel::{DnsResourceNatEntry, GatewayTunnel, ResolveDnsRequest};
use futures::channel::mpsc;
use futures_bounded::Timeout;
use phoenix_channel::{PhoenixChannel, PublicKeyParam};
use std::collections::BTreeSet;
use std::convert::Infallible;
@@ -37,9 +35,8 @@ static_assertions::const_assert!(
#[derive(Debug)]
enum ResolveTrigger {
RequestConnection(RequestConnection), // Deprecated
AllowAccess(AllowAccess), // Deprecated
Refresh(DomainName, ClientId, ResourceId), // TODO: Can we delete this perhaps?
RequestConnection(RequestConnection), // Deprecated
AllowAccess(AllowAccess), // Deprecated
SetupNat(ResolveDnsRequest),
}
@@ -48,7 +45,7 @@ pub struct Eventloop {
portal: PhoenixChannel<(), IngressMessages, (), PublicKeyParam>,
tun_device_channel: mpsc::Sender<Interface>,
resolve_tasks: futures_bounded::FuturesTupleSet<Vec<IpAddr>, ResolveTrigger>,
resolve_tasks: futures_bounded::FuturesTupleSet<Result<Vec<IpAddr>>, ResolveTrigger>,
}
impl Eventloop {
@@ -86,7 +83,14 @@ impl Eventloop {
Poll::Pending => {}
}
match self.resolve_tasks.poll_unpin(cx) {
match self.resolve_tasks.poll_unpin(cx).map(|(r, trigger)| {
(
r.unwrap_or_else(|e| {
Err(anyhow::Error::new(e).context("DNS resolution timed out"))
}),
trigger,
)
}) {
Poll::Ready((result, ResolveTrigger::RequestConnection(req))) => {
self.accept_connection(result, req);
continue;
@@ -95,23 +99,10 @@ impl Eventloop {
self.allow_access(result, req);
continue;
}
Poll::Ready((result, ResolveTrigger::Refresh(name, conn_id, resource_id))) => {
self.refresh_translation(result, conn_id, resource_id, name);
continue;
}
Poll::Ready((result, ResolveTrigger::SetupNat(request))) => {
let addresses = result
.inspect_err(|e| {
tracing::debug!(
error = std_dyn_err(e),
"DNS resolution timed out as part of setup NAT request"
)
})
.unwrap_or_default();
if let Err(e) = self.tunnel.state_mut().handle_domain_resolved(
request,
addresses,
result,
Instant::now(),
) {
tracing::warn!(
@@ -163,22 +154,6 @@ impl Eventloop {
}),
);
}
firezone_tunnel::GatewayEvent::RefreshDns {
name,
conn_id,
resource_id,
} => {
if self
.resolve_tasks
.try_push(
resolve(Some(name.clone())),
ResolveTrigger::Refresh(name, conn_id, resource_id),
)
.is_err()
{
tracing::warn!("Too many dns resolution requests, dropping existing one");
};
}
firezone_tunnel::GatewayEvent::ResolveDns(setup_nat) => {
if self
.resolve_tasks
@@ -348,14 +323,15 @@ impl Eventloop {
}
}
pub fn accept_connection(
&mut self,
result: Result<Vec<IpAddr>, Timeout>,
req: RequestConnection,
) {
let addresses = result
.inspect_err(|e| tracing::debug!(client = %req.client.id, reference = %req.reference, "DNS resolution timed out as part of connection request: {}", err_with_src(e)))
.unwrap_or_default();
pub fn accept_connection(&mut self, result: Result<Vec<IpAddr>>, req: RequestConnection) {
let addresses = match result {
Ok(addresses) => addresses,
Err(e) => {
tracing::debug!(client = %req.client.id, reference = %req.reference, "DNS resolution failed as part of connection request: {e:#}");
return; // Fail the connection so the client runs into a timeout.
}
};
let answer = match self.tunnel.state_mut().accept(
req.client.id,
@@ -388,7 +364,6 @@ impl Eventloop {
.payload
.domain
.map(|r| DnsResourceNatEntry::new(r, addresses)),
Instant::now(),
) {
let client = req.client.id;
@@ -408,10 +383,17 @@ impl Eventloop {
);
}
pub fn allow_access(&mut self, result: Result<Vec<IpAddr>, Timeout>, req: AllowAccess) {
let addresses = result
.inspect_err(|e| tracing::debug!(client = %req.client_id, reference = %req.reference, "DNS resolution timed out as part of allow access request: {}", err_with_src(e)))
.unwrap_or_default();
pub fn allow_access(&mut self, result: Result<Vec<IpAddr>>, req: AllowAccess) {
// "allow access" doesn't have a response so we can't tell the client that things failed.
// It is legacy code so don't bother ...
let addresses = match result {
Ok(addresses) => addresses,
Err(e) => {
tracing::debug!(client = %req.client_id, reference = %req.reference, "DNS resolution failed as part of allow access request: {e:#}");
vec![]
}
};
if let Err(e) = self.tunnel.state_mut().allow_access(
req.client_id,
@@ -420,56 +402,26 @@ impl Eventloop {
req.expires_at,
req.resource,
req.payload.map(|r| DnsResourceNatEntry::new(r, addresses)),
Instant::now(),
) {
tracing::warn!(error = anyhow_dyn_err(&e), client = %req.client_id, "Allow access request failed");
};
}
pub fn refresh_translation(
&mut self,
result: Result<Vec<IpAddr>, Timeout>,
conn_id: ClientId,
resource_id: ResourceId,
name: DomainName,
) {
let addresses = result
.inspect_err(|e| tracing::debug!(%conn_id, "DNS resolution timed out as part of allow access request: {}", err_with_src(e)))
.unwrap_or_default();
self.tunnel.state_mut().refresh_translation(
conn_id,
resource_id,
name,
addresses,
Instant::now(),
);
}
}
async fn resolve(domain: Option<DomainName>) -> Vec<IpAddr> {
async fn resolve(domain: Option<DomainName>) -> Result<Vec<IpAddr>> {
let Some(domain) = domain.clone() else {
return vec![];
return Ok(vec![]);
};
let dname = domain.to_string();
match tokio::task::spawn_blocking(move || resolve_addresses(&dname))
let addresses = tokio::task::spawn_blocking(move || resolve_addresses(&dname))
.instrument(telemetry_span!("resolve_dns_resource"))
.await
{
Ok(Ok(addresses)) => addresses,
Ok(Err(e)) => {
tracing::warn!(error = std_dyn_err(&e), %domain, "DNS resolution failed");
.context("DNS resolution task failed")?
.context("DNS resolution failed")?;
vec![]
}
Err(e) => {
tracing::warn!(error = std_dyn_err(&e), %domain, "DNS resolution task failed");
vec![]
}
}
Ok(addresses)
}
#[cfg(target_os = "windows")]

26
rust/phoenix-channel/src/lib.rs
View File

@@ -737,9 +737,7 @@ enum OkReply<T> {
pub enum ErrorReply {
#[serde(rename = "unmatched topic")]
UnmatchedTopic,
NotFound,
InvalidVersion,
Offline,
Disabled,
#[serde(other)]
Other,
@@ -749,9 +747,7 @@ impl fmt::Display for ErrorReply {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
ErrorReply::UnmatchedTopic => write!(f, "unmatched topic"),
ErrorReply::NotFound => write!(f, "not found"),
ErrorReply::InvalidVersion => write!(f, "invalid version"),
ErrorReply::Offline => write!(f, "offline"),
ErrorReply::Disabled => write!(f, "disabled"),
ErrorReply::Other => write!(f, "other"),
}
@@ -935,28 +931,6 @@ mod tests {
assert_eq!(actual_reply, expected_reply);
}
#[test]
fn not_found() {
let actual_reply = r#"
{
"event": "phx_reply",
"ref": null,
"topic": "client",
"payload": {
"status": "error",
"response": {
"reason": "not_found"
}
}
}
"#;
let actual_reply: Payload<(), ()> = serde_json::from_str(actual_reply).unwrap();
let expected_reply = Payload::<(), ()>::Reply(Reply::Error {
reason: ErrorReply::NotFound,
});
assert_eq!(actual_reply, expected_reply);
}
#[test]
fn unexpected_error_reply() {
let actual_reply = r#"

4
website/src/components/Changelog/Android.tsx
View File

@@ -20,6 +20,10 @@ export default function Android() {
Adds support for GSO (Generic Segmentation Offload), delivering
throughput improvements of up to 60%.
</ChangeItem>
<ChangeItem>
Makes use of the new control protocol, delivering faster and more
robust connection establishment.
</ChangeItem>
</Unreleased>
<Entry version="1.4.7" date={new Date("2024-11-08")}>
<ChangeItem pull="7263">

4
website/src/components/Changelog/Apple.tsx
View File

@@ -20,6 +20,10 @@ export default function Apple() {
Adds support for GSO (Generic Segmentation Offload), delivering
throughput improvements of up to 60%.
</ChangeItem>
<ChangeItem>
Makes use of the new control protocol, delivering faster and more
robust connection establishment.
</ChangeItem>
</Unreleased>
<Entry version="1.3.9" date={new Date("2024-11-08")}>
<ChangeItem pull="7288">

4
website/src/components/Changelog/GUI.tsx
View File

@@ -19,6 +19,10 @@ export default function GUI({ title }: { title: string }) {
Adds support for GSO (Generic Segmentation Offload), delivering
throughput improvements of up to 60%.
</ChangeItem>
<ChangeItem>
Makes use of the new control protocol, delivering faster and more
robust connection establishment.
</ChangeItem>
</Unreleased>
<Entry version="1.3.13" date={new Date("2024-11-15")}>
<ChangeItem pull="7334">

4
website/src/components/Changelog/Headless.tsx
View File

@@ -19,6 +19,10 @@ export default function Headless() {
Adds support for GSO (Generic Segmentation Offload), delivering
throughput improvements of up to 60%.
</ChangeItem>
<ChangeItem>
Makes use of the new control protocol, delivering faster and more
robust connection establishment.
</ChangeItem>
</Unreleased>
<Entry version="1.3.7" date={new Date("2024-11-15")}>
<ChangeItem pull="7334">