Skip to content

Commit 2914c9a

Browse files
yuyuyurekayuyuyureka
committed
don't expose client source port
1 parent 7e9c0d4 commit 2914c9a

File tree

7 files changed

+111
-56
lines changed

7 files changed

+111
-56
lines changed

‎frontend/src/resultsView.js‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,7 @@ const processResults = (results) => {
9191
// stage 1, combine seen and accepted routes
9292
// start out with Accepted
9393
const seenAndAccepted = {};
94-
const seenAndAcceptedKey = route => `${route.session_id.from_client}:${route.session_id.peer_address}:${route.net}`;
94+
const seenAndAcceptedKey = route => `${route.session_id.from_client}:${route.session_id.listener}:${route.session_id.peer_address}:${route.net}`;
9595
for (let route of routeResults) {
9696
if (route.state === "Accepted") {
9797
seenAndAccepted[seenAndAcceptedKey(route)] = route;

‎src/api.rs‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -340,7 +340,14 @@ async fn tables<T: Store>(State(AppState { store, .. }): State<AppState<T>>) ->
340340
serde_json::to_string(&store.get_tables()).unwrap()
341341
}
342342
async fn routers<T: Store>(State(AppState { store, .. }): State<AppState<T>>) -> impl IntoResponse {
343-
serde_json::to_string(&store.get_routers()).unwrap()
343+
serde_json::to_string(
344+
&store
345+
.get_routers()
346+
.into_iter()
347+
.map(|(k, v)| (format!("{},{}", k.0, k.1), v))
348+
.collect::<HashMap<_, _>>(),
349+
)
350+
.unwrap()
344351
}
345352

346353
async fn routing_instances<T: Store>(
@@ -349,7 +356,12 @@ async fn routing_instances<T: Store>(
349356
let instances = store
350357
.get_routing_instances()
351358
.into_iter()
352-
.map(|(k, v)| (k, v.into_iter().map(|v| (v, v)).collect::<Vec<_>>()))
359+
.map(|(k, v)| {
360+
(
361+
format!("{},{}", k.0, k.1),
362+
v.into_iter().map(|v| (v, v)).collect::<Vec<_>>(),
363+
)
364+
})
353365
.collect::<HashMap<_, _>>();
354366

355367
serde_json::to_string(&instances).unwrap()

‎src/bgp_collector.rs‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ pub async fn run_peer(
2121
store: impl Store,
2222
stream: TcpStream,
2323
client_addr: SocketAddr,
24+
listener_name: String,
2425
) -> anyhow::Result<BgpNotificationMessage> {
2526
let mut caps = vec![
2627
BgpCapability::SafiIPv4u,
@@ -61,7 +62,8 @@ pub async fn run_peer(
6162
.unwrap_or(client_addr.ip().to_string());
6263
store
6364
.client_up(
64-
client_addr,
65+
client_addr.ip(),
66+
listener_name.clone(),
6567
cfg.route_state,
6668
Client {
6769
client_name,
@@ -80,7 +82,8 @@ pub async fn run_peer(
8082
.insert_bgp_update(
8183
TableSelector {
8284
session_id: SessionId {
83-
from_client: client_addr,
85+
from_client: client_addr.ip(),
86+
listener: listener_name.clone(),
8487
peer_address: client_addr.ip(),
8588
},
8689
route_state: cfg.route_state,
@@ -111,6 +114,7 @@ pub struct BgpCollectorConfig {
111114
}
112115

113116
pub async fn run(
117+
name: String,
114118
cfg: BgpCollectorConfig,
115119
store: impl Store,
116120
mut shutdown: tokio::sync::watch::Receiver<bool>,
@@ -125,10 +129,11 @@ pub async fn run(
125129

126130
if let Some(peer_cfg) = cfg.peers.get(&client_addr.ip()).or(cfg.default_peer_config.as_ref()).cloned() {
127131
let store = store.clone();
132+
let name = name.clone();
128133
let mut shutdown = shutdown.clone();
129134
running_tasks.push(tokio::spawn(async move {
130135
tokio::select! {
131-
res = run_peer(peer_cfg, store.clone(), io, client_addr) => {
136+
res = run_peer(peer_cfg, store.clone(), io, client_addr, name.clone()) => {
132137
match res {
133138
Err(e) => warn!("disconnected {} {}", client_addr, e),
134139
Ok(notification) => info!("disconnected {} {:?}", client_addr, notification),
@@ -137,7 +142,7 @@ pub async fn run(
137142
_ = shutdown.changed() => {
138143
}
139144
};
140-
store.client_down(client_addr).await;
145+
store.client_down(client_addr.ip(), name.clone()).await;
141146
}));
142147
} else {
143148
info!("unexpected connection from {}", client_addr);

‎src/bmp_collector.rs‎

Lines changed: 27 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ use zettabgp::bmp::BmpMessage;
1818

1919
fn table_selector_for_peer(
2020
client_addr: SocketAddr,
21+
listener_name: String,
2122
peer: &BmpMessagePeerHeader,
2223
) -> Option<TableSelector> {
2324
let route_state = match (peer.peertype, peer.flags.view_bits::<Msb0>()[1]) {
@@ -37,7 +38,8 @@ fn table_selector_for_peer(
3738
route_distinguisher,
3839
route_state,
3940
session_id: SessionId {
40-
from_client: client_addr,
41+
from_client: client_addr.ip(),
42+
listener: listener_name,
4143
peer_address: peer.peeraddress,
4244
},
4345
})
@@ -46,9 +48,10 @@ fn table_selector_for_peer(
4648
async fn process_route_monitoring(
4749
store: &impl Store,
4850
client_addr: SocketAddr,
51+
listener_name: String,
4952
rm: BmpMessageRouteMonitoring,
5053
) {
51-
let session = match table_selector_for_peer(client_addr, &rm.peer) {
54+
let session = match table_selector_for_peer(client_addr, listener_name, &rm.peer) {
5255
Some(session) => session,
5356
None => {
5457
trace!(
@@ -65,6 +68,7 @@ async fn process_route_monitoring(
6568

6669
pub fn run_peer(
6770
client_addr: SocketAddr,
71+
listener_name: String,
6872
peer: BmpMessagePeerHeader,
6973
store: &impl Store,
7074
) -> mpsc::Sender<Result<BmpMessageRouteMonitoring, BmpMessagePeerDown>> {
@@ -73,7 +77,7 @@ pub fn run_peer(
7377

7478
tokio::task::spawn(async move {
7579
trace!("{} {:?}", client_addr, peer);
76-
if let Some(session_id) = table_selector_for_peer(client_addr, &peer)
80+
if let Some(session_id) = table_selector_for_peer(client_addr, listener_name.clone(), &peer)
7781
.and_then(|store| store.session_id().cloned())
7882
{
7983
store.session_up(session_id, Session {}).await;
@@ -82,7 +86,7 @@ pub fn run_peer(
8286
loop {
8387
match rx.recv().await {
8488
Some(Ok(rm)) => {
85-
process_route_monitoring(&store, client_addr, rm).await;
89+
process_route_monitoring(&store, client_addr, listener_name.clone(), rm).await;
8690
}
8791
Some(Err(down_msg)) => {
8892
trace!("{} {:?}", client_addr, down_msg);
@@ -94,7 +98,7 @@ pub fn run_peer(
9498
}
9599
}
96100
}
97-
if let Some(session_id) = table_selector_for_peer(client_addr, &peer)
101+
if let Some(session_id) = table_selector_for_peer(client_addr, listener_name.clone(), &peer)
98102
.and_then(|store| store.session_id().cloned())
99103
{
100104
store.session_down(session_id, None).await;
@@ -107,6 +111,7 @@ pub async fn run_client(
107111
cfg: PeerConfig,
108112
io: TcpStream,
109113
client_addr: SocketAddr,
114+
listener_name: String,
110115
store: &impl Store,
111116
) -> anyhow::Result<BmpMessageTermination> {
112117
let read = LengthDelimitedCodec::builder()
@@ -151,15 +156,21 @@ pub async fn run_client(
151156
> = HashMap::new();
152157
channels.insert(
153158
first_peer_up.peer.peeraddress,
154-
run_peer(client_addr, first_peer_up.peer, store),
159+
run_peer(
160+
client_addr,
161+
listener_name.clone(),
162+
first_peer_up.peer,
163+
store,
164+
),
155165
);
156166
let client_name = cfg
157167
.name_override
158168
.or(init_msg.sys_name)
159169
.unwrap_or(client_addr.ip().to_string());
160170
store
161171
.client_up(
162-
client_addr,
172+
client_addr.ip(),
173+
listener_name.clone(),
163174
RouteState::Selected,
164175
Client {
165176
client_name,
@@ -178,12 +189,15 @@ pub async fn run_client(
178189
BmpMessage::RouteMonitoring(rm) => {
179190
let channel = channels.entry(rm.peer.peeraddress).or_insert_with(|| {
180191
warn!("the bmp device {} sent a message for a nonexisting peer, we'll initialize the table now: {:?}", &client_addr, &rm);
181-
run_peer(client_addr, rm.peer.clone(), store)
192+
run_peer(client_addr, listener_name.clone(), rm.peer.clone(), store)
182193
});
183194
channel.send(Ok(rm)).await.unwrap();
184195
}
185196
BmpMessage::PeerUpNotification(n) => {
186-
channels.insert(n.peer.peeraddress, run_peer(client_addr, n.peer, store));
197+
channels.insert(
198+
n.peer.peeraddress,
199+
run_peer(client_addr, listener_name.clone(), n.peer, store),
200+
);
187201
}
188202
BmpMessage::PeerDownNotification(n) => match channels.remove(&n.peer.peeraddress) {
189203
Some(channel) => channel.send(Err(n)).await.unwrap(),
@@ -209,6 +223,7 @@ pub struct BmpCollectorConfig {
209223
}
210224

211225
pub async fn run(
226+
name: String,
212227
cfg: BmpCollectorConfig,
213228
store: impl Store,
214229
mut shutdown: tokio::sync::watch::Receiver<bool>,
@@ -222,11 +237,12 @@ pub async fn run(
222237
info!("connected {:?}", client_addr);
223238

224239
let store = store.clone();
240+
let name = name.clone();
225241
let mut shutdown = shutdown.clone();
226242
if let Some(peer_cfg) = cfg.peers.get(&client_addr.ip()).or(cfg.default_peer_config.as_ref()).cloned() {
227243
running_tasks.push(tokio::spawn(async move {
228244
tokio::select! {
229-
res = run_client(peer_cfg, io, client_addr, &store) => {
245+
res = run_client(peer_cfg, io, client_addr, name.clone(), &store) => {
230246
match res {
231247
Err(e) => warn!("disconnected {} {}", client_addr, e),
232248
Ok(notification) => info!("disconnected {} {:?}", client_addr, notification),
@@ -235,7 +251,7 @@ pub async fn run(
235251
_ = shutdown.changed() => {
236252
}
237253
};
238-
store.client_down(client_addr).await;
254+
store.client_down(client_addr.ip(), name.clone()).await;
239255
}));
240256
} else {
241257
info!("unexpected connection from {}", client_addr);

‎src/main.rs‎

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -44,14 +44,20 @@ async fn main() -> anyhow::Result<()> {
4444

4545
futures.extend(
4646
cfg.collectors
47-
.into_values()
48-
.map(|collector| match collector {
49-
CollectorConfig::Bmp(cfg) => {
50-
tokio::task::spawn(bmp_collector::run(cfg, store.clone(), shutdown_rx.clone()))
51-
}
52-
CollectorConfig::Bgp(cfg) => {
53-
tokio::task::spawn(bgp_collector::run(cfg, store.clone(), shutdown_rx.clone()))
54-
}
47+
.into_iter()
48+
.map(|(name, collector)| match collector {
49+
CollectorConfig::Bmp(cfg) => tokio::task::spawn(bmp_collector::run(
50+
name,
51+
cfg,
52+
store.clone(),
53+
shutdown_rx.clone(),
54+
)),
55+
CollectorConfig::Bgp(cfg) => tokio::task::spawn(bgp_collector::run(
56+
name,
57+
cfg,
58+
store.clone(),
59+
shutdown_rx.clone(),
60+
)),
5561
}),
5662
);
5763

‎src/store.rs‎

Lines changed: 15 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ use ipnet::{IpNet, Ipv4Net, Ipv6Net};
33
use log::*;
44
use serde::{Deserialize, Serialize};
55
use std::collections::{HashMap, HashSet};
6-
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
6+
use std::net::{IpAddr, Ipv4Addr};
77
use std::pin::Pin;
88
use zettabgp::prelude::{BgpAddrV4, BgpAddrV6};
99

@@ -33,7 +33,8 @@ pub struct RouteAttrs {
3333
#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
3434
#[serde(deny_unknown_fields)]
3535
pub struct SessionId {
36-
pub from_client: SocketAddr,
36+
pub from_client: IpAddr,
37+
pub listener: String,
3738
pub peer_address: IpAddr,
3839
}
3940

@@ -61,8 +62,11 @@ pub struct TableSelector {
6162
}
6263

6364
impl TableSelector {
64-
pub fn client_addr(&self) -> &SocketAddr {
65-
&self.session_id.from_client
65+
pub fn client_id(&self) -> (IpAddr, String) {
66+
(
67+
self.session_id.from_client,
68+
self.session_id.listener.clone(),
69+
)
6670
}
6771
pub fn session_id(&self) -> Option<&SessionId> {
6872
match self.route_state {
@@ -76,7 +80,7 @@ impl TableSelector {
7680
pub enum TableQuery {
7781
Table(TableSelector),
7882
Session(SessionId),
79-
Client(SocketAddr),
83+
Client(IpAddr, String),
8084
Router(RouterId),
8185
}
8286

@@ -164,20 +168,22 @@ pub trait Store: Clone + Send + Sync + 'static {
164168

165169
fn get_tables(&self) -> Vec<TableSelector>;
166170

167-
fn get_routers(&self) -> HashMap<SocketAddr, Client>;
171+
fn get_routers(&self) -> HashMap<(IpAddr, String), Client>;
168172

169-
fn get_routing_instances(&self) -> HashMap<SocketAddr, HashSet<RouteDistinguisher>>;
173+
fn get_routing_instances(&self) -> HashMap<(IpAddr, String), HashSet<RouteDistinguisher>>;
170174

171175
fn client_up(
172176
&self,
173-
client_addr: SocketAddr,
177+
client_ip: IpAddr,
178+
listener: String,
174179
route_state: RouteState,
175180
client_data: Client,
176181
) -> impl std::future::Future<Output = ()> + std::marker::Send;
177182

178183
fn client_down(
179184
&self,
180-
client_addr: SocketAddr,
185+
client_ip: IpAddr,
186+
listener: String,
181187
) -> impl std::future::Future<Output = ()> + std::marker::Send;
182188

183189
fn session_up(

0 commit comments

Comments
 (0)