astelm/telemt
1
0
Fork 0
mirror of https://github.com/telemt/telemt.git synced 2026-04-24 22:14:10 +03:00
Code Issues Packages Projects Releases Wiki Activity

Src-IP in ME Routing + more strict bind_addresses

Browse Source
This commit is contained in:
Alexey
2026-03-14 02:20:51 +03:00
parent a15f74a6f9
commit 780fafa604
7 changed files with 206 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/transport/middle_proxy/handshake.rs
View File

@@ -59,6 +59,7 @@ impl KdfClientPortSource {
pub(crate) struct HandshakeOutput {
pub rd: ReadHalf<TcpStream>,
pub wr: WriteHalf<TcpStream>,
pub source_ip: IpAddr,
pub read_key: [u8; 32],
pub read_iv: [u8; 16],
pub write_key: [u8; 32],
@@ -689,6 +690,7 @@ impl MePool {
Ok(HandshakeOutput {
rd,
wr,
source_ip: local_addr_nat.ip(),
read_key: rk,
read_iv,
write_key: wk,

1
src/transport/middle_proxy/pool.rs
View File

@@ -34,6 +34,7 @@ pub(super) struct RefillEndpointKey {
pub struct MeWriter {
pub id: u64,
pub addr: SocketAddr,
pub source_ip: IpAddr,
pub writer_dc: i32,
pub generation: u64,
pub contour: Arc<AtomicU8>,

1
src/transport/middle_proxy/pool_writer.rs
View File

@@ -163,6 +163,7 @@ impl MePool {
let writer = MeWriter {
id: writer_id,
addr,
source_ip: hs.source_ip,
writer_dc,
generation,
contour: contour.clone(),

44
src/transport/middle_proxy/send.rs
View File

@@ -42,20 +42,30 @@ impl MePool {
tag_override: Option<&[u8]>,
) -> Result<()> {
let tag = tag_override.or(self.proxy_tag.as_deref());
let payload = build_proxy_req_payload(
conn_id,
client_addr,
our_addr,
data,
tag,
proto_flags,
);
let meta = ConnMeta {
let fallback_meta = ConnMeta {
target_dc,
client_addr,
our_addr,
proto_flags,
};
let build_routed_payload = |effective_our_addr: SocketAddr| {
(
build_proxy_req_payload(
conn_id,
client_addr,
effective_our_addr,
data,
tag,
proto_flags,
),
ConnMeta {
target_dc,
client_addr,
our_addr: effective_our_addr,
proto_flags,
},
)
};
let no_writer_mode =
MeRouteNoWriterMode::from_u8(self.me_route_no_writer_mode.load(Ordering::Relaxed));
let (routed_dc, unknown_target_dc) = self
@@ -70,8 +80,14 @@ impl MePool {
let mut hybrid_wait_current = hybrid_wait_step;
loop {
let current_meta = self
.registry
.get_meta(conn_id)
.await
.unwrap_or_else(|| fallback_meta.clone());
let (current_payload, _) = build_routed_payload(current_meta.our_addr);
if let Some(current) = self.registry.get_writer(conn_id).await {
match current.tx.try_send(WriterCommand::Data(payload.clone())) {
match current.tx.try_send(WriterCommand::Data(current_payload.clone())) {
Ok(()) => return Ok(()),
Err(TrySendError::Full(cmd)) => {
if current.tx.send(cmd).await.is_ok() {
@@ -354,11 +370,13 @@ impl MePool {
if !self.writer_accepts_new_binding(w) {
continue;
}
let effective_our_addr = SocketAddr::new(w.source_ip, our_addr.port());
let (payload, meta) = build_routed_payload(effective_our_addr);
match w.tx.try_send(WriterCommand::Data(payload.clone())) {
Ok(()) => {
self.stats.increment_me_writer_pick_success_try_total(pick_mode);
self.registry
.bind_writer(conn_id, w.id, w.tx.clone(), meta.clone())
.bind_writer(conn_id, w.id, w.tx.clone(), meta)
.await;
if w.generation < self.current_generation() {
self.stats.increment_pool_stale_pick_total();
@@ -397,12 +415,14 @@ impl MePool {
continue;
}
self.stats.increment_me_writer_pick_blocking_fallback_total();
let effective_our_addr = SocketAddr::new(w.source_ip, our_addr.port());
let (payload, meta) = build_routed_payload(effective_our_addr);
match w.tx.send(WriterCommand::Data(payload.clone())).await {
Ok(()) => {
self.stats
.increment_me_writer_pick_success_fallback_total(pick_mode);
self.registry
.bind_writer(conn_id, w.id, w.tx.clone(), meta.clone())
.bind_writer(conn_id, w.id, w.tx.clone(), meta)
.await;
if w.generation < self.current_generation() {
self.stats.increment_pool_stale_pick_total();

46
src/transport/upstream.rs
View File

@@ -390,7 +390,7 @@ impl UpstreamManager {
out
}
fn resolve_bind_address(
pub(crate) fn resolve_bind_address(
interface: &Option<String>,
bind_addresses: &Option<Vec<String>>,
target: SocketAddr,
@@ -399,7 +399,7 @@ impl UpstreamManager {
) -> Option<IpAddr> {
let want_ipv6 = target.is_ipv6();
if let Some(addrs) = bind_addresses {
if let Some(addrs) = bind_addresses.as_ref().filter(|v| !v.is_empty()) {
let mut candidates: Vec<IpAddr> = addrs
.iter()
.filter_map(|s| s.parse::<IpAddr>().ok())
@@ -431,7 +431,7 @@ impl UpstreamManager {
warn!(
interface = %iface,
target = %target,
"Configured interface has no addresses for target family; falling back to direct connect without bind"
"Configured interface has no addresses for target family"
);
candidates.clear();
}
@@ -454,10 +454,11 @@ impl UpstreamManager {
warn!(
interface = interface.as_deref().unwrap_or(""),
target = %target,
"No valid bind_addresses left for interface; falling back to direct connect without bind"
"No valid bind_addresses left for interface"
);
return None;
}
return None;
}
if let Some(iface) = interface {
@@ -795,6 +796,13 @@ impl UpstreamManager {
bind_rr.as_deref(),
true,
);
if bind_ip.is_none()
&& bind_addresses.as_ref().is_some_and(|v| !v.is_empty())
{
return Err(ProxyError::Config(format!(
"No valid bind_addresses for target family {target}"
)));
}
let socket = create_outgoing_socket_bound(target, bind_ip)?;
if let Some(ip) = bind_ip {
@@ -1642,4 +1650,32 @@ mod tests {
};
assert!(!UpstreamManager::is_hard_connect_error(&error));
}
#[test]
fn resolve_bind_address_prefers_explicit_bind_ip() {
let target = "203.0.113.10:443".parse::<SocketAddr>().unwrap();
let bind = UpstreamManager::resolve_bind_address(
&Some("198.51.100.20".to_string()),
&Some(vec!["198.51.100.10".to_string()]),
target,
None,
true,
);
assert_eq!(bind, Some("198.51.100.10".parse::<IpAddr>().unwrap()));
}
#[test]
fn resolve_bind_address_does_not_fallback_to_interface_when_bind_addresses_present() {
let target = "203.0.113.10:443".parse::<SocketAddr>().unwrap();
let bind = UpstreamManager::resolve_bind_address(
&Some("198.51.100.20".to_string()),
&Some(vec!["2001:db8::10".to_string()]),
target,
None,
true,
);
assert_eq!(bind, None);
}
}