Skip to content

Fix duplicate peer usrpwd transport reconciliation #2578

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
BOURBONCASK wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Fix duplicate peer usrpwd transport reconciliation #2578

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d029917
transport: reconcile duplicate usrpwd peer connections
BOURBONCASK Apr 18, 2026
14482b1
zenoh: refresh interceptors after usrpwd principal upgrades
BOURBONCASK Apr 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions io/zenoh-transport/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,6 +119,7 @@ pub trait TransportPeerEventHandler: Send + Sync {
fn handle_message(&self, msg: NetworkMessageMut) -> ZResult<()>;
fn new_link(&self, src: Link);
fn del_link(&self, link: Link);
fn metadata_changed(&self) {}
fn closed(&self);
fn as_any(&self) -> &dyn Any;
}
Expand Down
45 changes: 43 additions & 2 deletions io/zenoh-transport/src/unicast/authentication.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,10 +13,51 @@
//
use zenoh_link::LinkAuthId;
use zenoh_protocol::core::ZenohIdProto;
#[cfg(feature = "auth_usrpwd")]
use zenoh_result::{zerror, ZResult};

#[cfg(feature = "auth_usrpwd")]
use super::establishment::ext::auth::UsrPwdId;

#[cfg(feature = "auth_usrpwd")]
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum TransportUsrPwdPrincipal {
Unknown,
Known(Vec<u8>),
}

#[cfg(feature = "auth_usrpwd")]
impl TransportUsrPwdPrincipal {
pub(crate) fn from_auth_id(auth_id: UsrPwdId) -> Self {
if let Some(username) = auth_id.0 {
Self::Known(username)
} else {
Self::Unknown
}
}
}

#[cfg(feature = "auth_usrpwd")]
pub(crate) fn plan_usrpwd_principal_update(
existing: &TransportUsrPwdPrincipal,
incoming: &TransportUsrPwdPrincipal,
) -> ZResult<bool> {
match (existing, incoming) {
(TransportUsrPwdPrincipal::Unknown, TransportUsrPwdPrincipal::Unknown)
| (TransportUsrPwdPrincipal::Known(_), TransportUsrPwdPrincipal::Unknown) => Ok(false),
(TransportUsrPwdPrincipal::Unknown, TransportUsrPwdPrincipal::Known(_)) => Ok(true),
(TransportUsrPwdPrincipal::Known(a), TransportUsrPwdPrincipal::Known(b)) if a == b => {
Ok(false)
}
_ => Err(zerror!(
"Invalid authenticated principal: {:?}. Expected: {:?}.",
incoming,
existing
)
.into()),
}
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TransportAuthId {
username: Option<String>,
Expand All @@ -34,8 +75,8 @@ impl TransportAuthId {
}

#[cfg(feature = "auth_usrpwd")]
pub(crate) fn set_username(&mut self, user_pwd_id: &UsrPwdId) {
self.username = if let Some(username) = &user_pwd_id.0 {
pub(crate) fn set_username(&mut self, principal: &TransportUsrPwdPrincipal) {
self.username = if let TransportUsrPwdPrincipal::Known(username) = principal {
// Convert username from Vec<u8> to String
match std::str::from_utf8(username) {
Ok(name) => Some(name.to_owned()),
Expand Down
9 changes: 7 additions & 2 deletions io/zenoh-transport/src/unicast/establishment/accept.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ use super::ext::auth::UsrPwdId;
use super::ext::shm::AuthSegment;
#[cfg(feature = "shared-memory")]
use crate::shm::TransportShmConfig;
#[cfg(feature = "auth_usrpwd")]
use crate::unicast::authentication::TransportUsrPwdPrincipal;
use crate::{
common::batch::BatchConfig,
unicast::{
Expand Down Expand Up @@ -877,12 +879,13 @@ pub(crate) async fn accept_link(link: LinkUnicast, manager: &TransportManager) -
false => None,
},
is_lowlatency: state.transport.ext_lowlatency.is_lowlatency(),
#[cfg(feature = "auth_usrpwd")]
auth_id: osyn_out.other_auth_id,
patch: state.transport.ext_patch.get(),
region_name: state.transport.ext_region_name.other_region_name(),
};

#[cfg(feature = "auth_usrpwd")]
let usrpwd_principal = TransportUsrPwdPrincipal::from_auth_id(osyn_out.other_auth_id);

let a_config = TransportLinkUnicastConfig {
direction,
batch: BatchConfig {
Expand Down Expand Up @@ -929,6 +932,8 @@ pub(crate) async fn accept_link(link: LinkUnicast, manager: &TransportManager) -
let _transport = manager
.init_transport_unicast(
config,
#[cfg(feature = "auth_usrpwd")]
usrpwd_principal,
a_link,
osyn_out.other_initial_sn,
osyn_out.other_lease,
Expand Down
9 changes: 6 additions & 3 deletions io/zenoh-transport/src/unicast/establishment/open.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ use super::ext::shm::AuthSegment;
#[cfg(feature = "shared-memory")]
use crate::shm::TransportShmConfig;
#[cfg(feature = "auth_usrpwd")]
use crate::unicast::establishment::ext::auth::UsrPwdId;
use crate::unicast::authentication::TransportUsrPwdPrincipal;
use crate::{
common::batch::BatchConfig,
unicast::{
Expand Down Expand Up @@ -758,12 +758,13 @@ pub(crate) async fn open_link(
false => None,
},
is_lowlatency: state.transport.ext_lowlatency.is_lowlatency(),
#[cfg(feature = "auth_usrpwd")]
auth_id: UsrPwdId(None),
patch: state.transport.ext_patch.get(),
region_name: state.transport.ext_region_name.other_region_name(),
};

#[cfg(feature = "auth_usrpwd")]
let usrpwd_principal = TransportUsrPwdPrincipal::from_auth_id(super::ext::auth::UsrPwdId(None));

let o_config = TransportLinkUnicastConfig {
direction,
batch: BatchConfig {
Expand Down Expand Up @@ -810,6 +811,8 @@ pub(crate) async fn open_link(
let transport = manager
.init_transport_unicast(
config,
#[cfg(feature = "auth_usrpwd")]
usrpwd_principal,
o_link,
oack_out.other_initial_sn,
oack_out.other_lease,
Expand Down
31 changes: 29 additions & 2 deletions io/zenoh-transport/src/unicast/lowlatency/transport.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@ use zenoh_result::{zerror, ZResult};

#[cfg(feature = "shared-memory")]
use crate::shm_context::UnicastTransportShmContext;
#[cfg(feature = "auth_usrpwd")]
use crate::unicast::authentication::{plan_usrpwd_principal_update, TransportUsrPwdPrincipal};
use crate::{
unicast::{
authentication::TransportAuthId,
Expand Down Expand Up @@ -64,6 +66,8 @@ pub(crate) struct TransportUnicastLowlatency {
pub(super) stats: zenoh_stats::TransportStats,
#[cfg(feature = "stats")]
pub(super) link_stats: Arc<OnceLock<zenoh_stats::LinkStats>>,
#[cfg(feature = "auth_usrpwd")]
pub(super) usrpwd_principal: Arc<SyncRwLock<TransportUsrPwdPrincipal>>,

// The handles for TX/RX tasks
pub(crate) token: CancellationToken,
Expand All @@ -77,6 +81,7 @@ impl TransportUnicastLowlatency {
pub fn make(
manager: TransportManager,
config: TransportConfigUnicast,
#[cfg(feature = "auth_usrpwd")] usrpwd_principal: TransportUsrPwdPrincipal,
#[cfg(feature = "shared-memory")] shm_context: Option<UnicastTransportShmContext>,
#[cfg(feature = "stats")] stats: zenoh_stats::TransportStats,
) -> Arc<dyn TransportUnicastTrait> {
Expand All @@ -90,6 +95,8 @@ impl TransportUnicastLowlatency {
stats,
#[cfg(feature = "stats")]
link_stats: Arc::new(OnceLock::new()),
#[cfg(feature = "auth_usrpwd")]
usrpwd_principal: Arc::new(SyncRwLock::new(usrpwd_principal)),
token: CancellationToken::new(),
tracker: TaskTracker::new(),
#[cfg(feature = "shared-memory")]
Expand Down Expand Up @@ -197,6 +204,11 @@ impl TransportUnicastTrait for TransportUnicastLowlatency {
self.config.zid
}

#[cfg(feature = "auth_usrpwd")]
fn set_usrpwd_principal(&self, principal: TransportUsrPwdPrincipal) {
*zwrite!(self.usrpwd_principal) = principal;
}

fn get_auth_ids(&self) -> TransportAuthId {
// Convert LinkUnicast auth id to AuthId
let mut transport_auth_id = TransportAuthId::new(self.get_zid());
Expand All @@ -208,7 +220,7 @@ impl TransportUnicastTrait for TransportUnicastLowlatency {
}
// Convert usrpwd auth id to AuthId
#[cfg(feature = "auth_usrpwd")]
transport_auth_id.set_username(&self.config.auth_id);
transport_auth_id.set_username(&zread!(self.usrpwd_principal));
transport_auth_id
}

Expand Down Expand Up @@ -260,6 +272,7 @@ impl TransportUnicastTrait for TransportUnicastLowlatency {
async fn add_link(
&self,
link: LinkUnicastWithOpenAck,
#[cfg(feature = "auth_usrpwd")] usrpwd_principal: &TransportUsrPwdPrincipal,
other_initial_sn: TransportSn,
other_lease: Duration,
) -> AddLinkResult {
Expand All @@ -278,6 +291,20 @@ impl TransportUnicastTrait for TransportUnicastLowlatency {
}
};

#[cfg(feature = "auth_usrpwd")]
let principal_update = {
let current = zread!(self.usrpwd_principal);
match plan_usrpwd_principal_update(&current, usrpwd_principal) {
Ok(update) => update,
Err(e) => {
let (l, asl) = link.fail();
return Err((e, l, asl, close::reason::INVALID));
}
}
};
#[cfg(not(feature = "auth_usrpwd"))]
let principal_update = false;

let mut guard = zasyncwrite!(self.link);
if guard.is_some() {
let (l, asl) = link.fail();
Expand Down Expand Up @@ -321,7 +348,7 @@ impl TransportUnicastTrait for TransportUnicastLowlatency {
self.internal_start_rx(other_lease);
});

Ok((start_tx, start_rx, ack, status_guard))
Ok((start_tx, start_rx, ack, status_guard, principal_update))
}

/*************************************/
Expand Down
60 changes: 55 additions & 5 deletions io/zenoh-transport/src/unicast/manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@ use zenoh_protocol::{
use zenoh_result::{bail, zerror, ZResult};

use super::{link::LinkUnicastWithOpenAck, transport_unicast_inner::InitTransportResult};
#[cfg(feature = "auth_usrpwd")]
use crate::unicast::authentication::TransportUsrPwdPrincipal;
#[cfg(feature = "transport_auth")]
use crate::unicast::establishment::ext::auth::Auth;
#[cfg(feature = "transport_multilink")]
Expand Down Expand Up @@ -439,6 +441,7 @@ impl TransportManager {
async fn init_existing_transport_unicast(
&self,
config: TransportConfigUnicast,
#[cfg(feature = "auth_usrpwd")] usrpwd_principal: TransportUsrPwdPrincipal,
link: LinkUnicastWithOpenAck,
other_initial_sn: TransportSn,
other_lease: Duration,
Expand All @@ -447,7 +450,7 @@ impl TransportManager {
let existing_config = transport.get_config();
// Verify that fundamental parameters are correct.
// Ignore the non fundamental parameters like initial SN.
if *existing_config != config {
if !existing_config.is_compatible_with(&config) {
let e = zerror!(
"Transport with peer {} already exist. Invalid config: {:?}. Expected: {:?}.",
config.zid,
Expand All @@ -465,8 +468,14 @@ impl TransportManager {
}

// Add the link to the transport
let (start_tx, start_rx, ack, add_link_guard) = transport
.add_link(link, other_initial_sn, other_lease)
let (start_tx, start_rx, ack, add_link_guard, principal_update) = transport
.add_link(
link,
#[cfg(feature = "auth_usrpwd")]
&usrpwd_principal,
other_initial_sn,
other_lease,
)
.await
.map_err(InitTransportError::Link)?;

Expand All @@ -476,6 +485,16 @@ impl TransportManager {
InitTransportError::Transport((e, transport.clone(), close::reason::GENERIC))
})?;

#[cfg(feature = "auth_usrpwd")]
if principal_update {
transport.set_usrpwd_principal(usrpwd_principal);
Self::notify_metadata_changed_unicast(&transport)
.await
.map_err(|e| {
InitTransportError::Transport((e, transport.clone(), close::reason::GENERIC))
})?;
}

start_tx();

// notify transport's callback interface that there is a new link
Expand Down Expand Up @@ -506,6 +525,19 @@ impl TransportManager {
Ok(())
}

async fn notify_metadata_changed_unicast(
transport: &Arc<dyn TransportUnicastTrait>,
) -> ZResult<()> {
if let Some(callback) = transport.get_callback() {
tokio::task::spawn_blocking(move || {
callback.metadata_changed();
})
.await?;
}

Ok(())
}

fn notify_new_transport_unicast(
&self,
transport: &Arc<dyn TransportUnicastTrait>,
Expand Down Expand Up @@ -538,6 +570,7 @@ impl TransportManager {
pub(super) async fn init_new_transport_unicast(
&self,
config: TransportConfigUnicast,
#[cfg(feature = "auth_usrpwd")] usrpwd_principal: TransportUsrPwdPrincipal,
link: LinkUnicastWithOpenAck,
other_initial_sn: TransportSn,
other_lease: Duration,
Expand Down Expand Up @@ -627,6 +660,8 @@ impl TransportManager {
TransportUnicastLowlatency::make(
self.clone(),
config.clone(),
#[cfg(feature = "auth_usrpwd")]
usrpwd_principal.clone(),
#[cfg(feature = "shared-memory")]
shm_context,
#[cfg(feature = "stats")]
Expand All @@ -638,6 +673,8 @@ impl TransportManager {
TransportUnicastUniversal::make(
self.clone(),
config.clone(),
#[cfg(feature = "auth_usrpwd")]
usrpwd_principal.clone(),
#[cfg(feature = "shared-memory")]
shm_context,
#[cfg(feature = "stats")]
Expand All @@ -648,8 +685,16 @@ impl TransportManager {
};

// Add the link to the transport
let (start_tx, start_rx, ack, add_link_guard) =
match t.add_link(link, other_initial_sn, other_lease).await {
let (start_tx, start_rx, ack, add_link_guard, _) = match t
.add_link(
link,
#[cfg(feature = "auth_usrpwd")]
&usrpwd_principal,
other_initial_sn,
other_lease,
)
.await
{
Ok(val) => val,
Err(e) => {
let _ = t.close(e.3).await;
Expand Down Expand Up @@ -731,6 +776,7 @@ impl TransportManager {
pub(super) async fn init_transport_unicast(
&self,
config: TransportConfigUnicast,
#[cfg(feature = "auth_usrpwd")] usrpwd_principal: TransportUsrPwdPrincipal,
link: LinkUnicastWithOpenAck,
other_initial_sn: TransportSn,
other_lease: Duration,
Expand All @@ -744,6 +790,8 @@ impl TransportManager {
drop(guard);
self.init_existing_transport_unicast(
config,
#[cfg(feature = "auth_usrpwd")]
usrpwd_principal,
link,
other_initial_sn,
other_lease,
Expand All @@ -754,6 +802,8 @@ impl TransportManager {
None => {
self.init_new_transport_unicast(
config,
#[cfg(feature = "auth_usrpwd")]
usrpwd_principal,
link,
other_initial_sn,
other_lease,
Expand Down
Loading
Loading