liz/veilid
1
0
Fork 0
Code Pull Requests Packages Releases Activity

continue to refactor

Browse Source
This commit is contained in:
John Smith
2021-12-21 22:20:55 -05:00
parent c0a42ac90c
commit 5826551763
22 changed files with 381 additions and 551 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
veilid-core/src/intf/native/network/mod.rs
View File

@@ -39,13 +39,11 @@ pub const PEEK_DETECT_LEN: usize = 64;
struct NetworkInner {
routing_table: RoutingTable,
network_manager: NetworkManager,
network_started: bool,
network_needs_restart: bool,
udp_listen: bool,
protocol_config: Option<ProtocolConfig>,
udp_static_public_dialinfo: bool,
tcp_listen: bool,
tcp_static_public_dialinfo: bool,
ws_listen: bool,
wss_listen: bool,
network_class: Option<NetworkClass>,
join_handles: Vec<JoinHandle<()>>,
listener_states: BTreeMap<SocketAddr, Arc<RwLock<ListenerState>>>,
@@ -81,13 +79,11 @@ impl Network {
NetworkInner {
routing_table: network_manager.routing_table(),
network_manager,
network_started: false,
network_needs_restart: false,
udp_listen: false,
protocol_config: None,
udp_static_public_dialinfo: false,
tcp_listen: false,
tcp_static_public_dialinfo: false,
ws_listen: false,
wss_listen: false,
network_class: None,
join_handles: Vec::new(),
listener_states: BTreeMap::new(),
@@ -681,14 +677,11 @@ impl Network {
match descriptor.protocol_type() {
ProtocolType::UDP => {
// send over the best udp socket we have bound since UDP is not connection oriented
let peer_socket_addr = descriptor
.remote
.to_socket_addr()
.map_err(map_to_string)
.map_err(logthru_net!(error))?;
if let Some(ph) =
self.find_best_udp_protocol_handler(&peer_socket_addr, &descriptor.local)
{
let peer_socket_addr = descriptor.remote.to_socket_addr();
if let Some(ph) = self.find_best_udp_protocol_handler(
&peer_socket_addr,
&descriptor.local.map(|sa| sa.to_socket_addr()),
) {
ph.clone()
.send_message(data, peer_socket_addr)
.await
@@ -724,14 +717,13 @@ impl Network {
) -> Result<(), String> {
match &dial_info {
DialInfo::UDP(_) => {
let peer_socket_addr = dial_info.to_socket_addr().map_err(logthru_net!())?;
let peer_socket_addr = dial_info.to_socket_addr();
RawUdpProtocolHandler::send_unbound_message(data, peer_socket_addr)
.await
.map_err(logthru_net!())
}
DialInfo::TCP(_) => {
let peer_socket_addr = dial_info.to_socket_addr().map_err(logthru_net!())?;
let peer_socket_addr = dial_info.to_socket_addr();
RawTcpProtocolHandler::send_unbound_message(data, peer_socket_addr)
.await
.map_err(logthru_net!())
@@ -752,7 +744,7 @@ impl Network {
let conn = match &dial_info {
DialInfo::UDP(_) => {
let peer_socket_addr = dial_info.to_socket_addr().map_err(logthru_net!())?;
let peer_socket_addr = dial_info.to_socket_addr();
if let Some(ph) = self.find_best_udp_protocol_handler(&peer_socket_addr, &None) {
return ph
.send_message(data, peer_socket_addr)
@@ -764,7 +756,7 @@ impl Network {
}
}
DialInfo::TCP(_) => {
let peer_socket_addr = dial_info.to_socket_addr().map_err(logthru_net!())?;
let peer_socket_addr = dial_info.to_socket_addr();
let local_addr =
self.get_preferred_local_address(self.inner.lock().tcp_port, &peer_socket_addr);
RawTcpProtocolHandler::connect(network_manager, local_addr, peer_socket_addr)
@@ -772,22 +764,20 @@ impl Network {
.map_err(logthru_net!())?
}
DialInfo::WS(_) => {
let remote_ip_addr = dial_info.resolve()?;
let peer_socket_addr = dial_info.to_socket_addr();
let local_addr =
self.get_preferred_local_address(self.inner.lock().ws_port, &peer_socket_addr);
WebsocketProtocolHandler::connect(network_manager, local_addr, dial_info)
.await
.map_err(logthru_net!(error))?
}
DialInfo::WSS(_) => {
let peer_socket_addr = dial_info.to_socket_addr();
let local_addr =
self.get_preferred_local_address(self.inner.lock().ws_port, &peer_socket_addr);
WebsocketProtocolHandler::connect(network_manager, dial_info)
.await
.map_err(logthru_net!(error))?,
self.get_preferred_local_address(self.inner.lock().wss_port, &peer_socket_addr);
WebsocketProtocolHandler::connect(network_manager, local_addr, dial_info)
.await
.map_err(logthru_net!(error))?
}
};
@@ -865,7 +855,7 @@ impl Network {
} else {
// Register local dial info as public if it is publicly routable
for x in &dial_infos {
if x.is_public().unwrap_or(false) {
if x.is_global() {
routing_table.register_global_dial_info(
x.clone(),
Some(NetworkClass::Server),
@@ -875,7 +865,6 @@ impl Network {
}
}
}
self.inner.lock().udp_listen = true;
self.inner.lock().udp_static_public_dialinfo = static_public;
Ok(())
}
@@ -904,8 +893,8 @@ impl Network {
for (a, p) in addresses {
// Pick out WS port for outbound connections (they will all be the same)
self.inner.lock().ws_port = p;
let di = DialInfo::ws(a.address_string(), p, path.clone());
xxx continue here
let di = DialInfo::try_ws(a.address_string(), p, path.clone());
dial_infos.push(di.clone());
routing_table.register_local_dial_info(di, DialInfoOrigin::Static);
}
@@ -930,7 +919,6 @@ impl Network {
);
}
self.inner.lock().ws_listen = true;
Ok(())
}
@@ -990,7 +978,6 @@ impl Network {
return Err("WSS URL must be specified due to TLS requirements".to_owned());
}
self.inner.lock().wss_listen = true;
Ok(())
}
@@ -1044,7 +1031,7 @@ impl Network {
} else {
// Register local dial info as public if it is publicly routable
for x in &dial_infos {
if x.is_public().unwrap_or(false) {
if x.is_global() {
routing_table.register_global_dial_info(
x.clone(),
Some(NetworkClass::Server),
@@ -1055,12 +1042,15 @@ impl Network {
}
}
self.inner.lock().tcp_listen = true;
self.inner.lock().tcp_static_public_dialinfo = static_public;
Ok(())
}
pub fn get_protocol_config(&self) -> Option<ProtocolConfig> {
self.inner.lock().protocol_config.clone()
}
pub async fn startup(&self) -> Result<(), String> {
info!("starting network");
let network_manager = self.inner.lock().network_manager.clone();
@@ -1068,36 +1058,38 @@ impl Network {
// initialize interfaces
self.inner.lock().interfaces.refresh()?;
// get network config
let (enabled_udp, connect_tcp, listen_tcp, connect_ws, listen_ws, connect_wss, listen_wss) = {
// get protocol config
let protocol_config = {
let c = self.config.get();
(
c.network.protocol.udp.enabled && c.capabilities.protocol_udp,
c.network.protocol.tcp.connect && c.capabilities.protocol_connect_tcp,
c.network.protocol.tcp.listen && c.capabilities.protocol_accept_tcp,
c.network.protocol.ws.connect && c.capabilities.protocol_connect_ws,
c.network.protocol.ws.listen && c.capabilities.protocol_accept_ws,
c.network.protocol.wss.connect && c.capabilities.protocol_connect_wss,
c.network.protocol.wss.listen && c.capabilities.protocol_accept_wss,
)
ProtocolConfig {
udp_enabled: c.network.protocol.udp.enabled && c.capabilities.protocol_udp,
tcp_connect: c.network.protocol.tcp.connect && c.capabilities.protocol_connect_tcp,
tcp_listen: c.network.protocol.tcp.listen && c.capabilities.protocol_accept_tcp,
ws_connect: c.network.protocol.ws.connect && c.capabilities.protocol_connect_ws,
ws_listen: c.network.protocol.ws.listen && c.capabilities.protocol_accept_ws,
wss_connect: c.network.protocol.wss.connect && c.capabilities.protocol_connect_wss,
wss_listen: c.network.protocol.wss.listen && c.capabilities.protocol_accept_wss,
}
};
self.inner.lock().protocol_config = Some(protocol_config);
// start listeners
if enabled_udp {
if protocol_config.udp_enabled {
self.start_udp_listeners().await?;
self.create_udp_outbound_sockets().await?;
}
if listen_ws {
if protocol_config.ws_listen {
self.start_ws_listeners().await?;
}
if listen_wss {
if protocol_config.wss_listen {
self.start_wss_listeners().await?;
}
if listen_tcp {
if protocol_config.tcp_listen {
self.start_tcp_listeners().await?;
}
info!("network started");
self.inner.lock().network_started = true;
Ok(())
}
@@ -1123,13 +1115,17 @@ impl Network {
}
//////////////////////////////////////////
pub fn get_network_class(&self) -> NetworkClass {
pub fn get_network_class(&self) -> Option<NetworkClass> {
let inner = self.inner.lock();
let routing_table = inner.routing_table.clone();
if !inner.network_started {
return None;
}
// If we've fixed the network class, return it rather than calculating it
if let Some(network_class) = inner.network_class {
return network_class;
if inner.network_class.is_some() {
return inner.network_class;
}
// Go through our public dialinfo and see what our best network class is
@@ -1141,7 +1137,7 @@ impl Network {
}
}
}
network_class
Some(network_class)
}
//////////////////////////////////////////
@@ -1149,18 +1145,19 @@ impl Network {
pub async fn tick(&self) -> Result<(), String> {
let (
routing_table,
udp_listen,
protocol_config,
udp_static_public_dialinfo,
tcp_listen,
tcp_static_public_dialinfo,
network_class,
) = {
let inner = self.inner.lock();
(
inner.network_manager.routing_table(),
inner.udp_listen,
inner
.protocol_config
.clone()
.unwrap_or_else(|| ProtocolConfig::default()),
inner.udp_static_public_dialinfo,
inner.tcp_listen,
inner.tcp_static_public_dialinfo,
inner.network_class.unwrap_or(NetworkClass::Invalid),
)
@@ -1171,7 +1168,7 @@ impl Network {
// If we can have public dialinfo, or we haven't figured out our network class yet,
// and we're active for UDP, we should attempt to get our public dialinfo sorted out
// and assess our network class if we haven't already
if udp_listen
if protocol_config.udp_enabled
&& !udp_static_public_dialinfo
&& (network_class.inbound_capable() || network_class == NetworkClass::Invalid)
{
@@ -1189,7 +1186,7 @@ impl Network {
}
// Same but for TCPv4
if tcp_listen
if protocol_config.tcp_enabled
&& !tcp_static_public_dialinfo
&& (network_class.inbound_capable() || network_class == NetworkClass::Invalid)
{

19
veilid-core/src/intf/native/network/protocol/tcp.rs
View File

@@ -153,8 +153,7 @@ impl RawTcpProtocolHandler {
let conn = NetworkConnection::RawTcp(RawTcpNetworkConnection::new(stream));
let peer_addr = PeerAddress::new(
Address::from_socket_addr(socket_addr).to_canonical(),
socket_addr.port(),
SocketAddress::from_socket_addr(socket_addr),
ProtocolType::TCP,
);
let (network_manager, local_address) = {
@@ -162,7 +161,13 @@ impl RawTcpProtocolHandler {
(inner.network_manager.clone(), inner.local_address)
};
network_manager
.on_new_connection(ConnectionDescriptor::new(peer_addr, local_address), conn)
.on_new_connection(
ConnectionDescriptor::new(
peer_addr,
SocketAddress::from_socket_addr(local_address),
),
conn,
)
.await?;
Ok(true)
}
@@ -191,8 +196,7 @@ impl RawTcpProtocolHandler {
.map_err(logthru_net!())?;
let ps = AsyncPeekStream::new(ts);
let peer_addr = PeerAddress::new(
Address::from_socket_addr(remote_socket_addr).to_canonical(),
remote_socket_addr.port(),
SocketAddress::from_socket_addr(remote_socket_addr),
ProtocolType::TCP,
);
@@ -200,7 +204,10 @@ impl RawTcpProtocolHandler {
let conn = NetworkConnection::RawTcp(RawTcpNetworkConnection::new(ps));
network_manager
.on_new_connection(
ConnectionDescriptor::new(peer_addr, local_address),
ConnectionDescriptor::new(
peer_addr,
SocketAddress::from_socket_addr(local_address),
),
conn.clone(),
)
.await?;

8
veilid-core/src/intf/native/network/protocol/udp.rs
View File

@@ -48,15 +48,17 @@ impl RawUdpProtocolHandler {
};
let peer_addr = PeerAddress::new(
Address::from_socket_addr(remote_addr),
remote_addr.port(),
SocketAddress::from_socket_addr(remote_addr),
ProtocolType::UDP,
);
let local_socket_addr = socket.local_addr().map_err(|e| format!("{}", e))?;
network_manager
.on_recv_envelope(
data,
&ConnectionDescriptor::new(peer_addr, local_socket_addr),
&ConnectionDescriptor::new(
peer_addr,
SocketAddress::from_socket_addr(local_socket_addr),
),
)
.await
}

55
veilid-core/src/intf/native/network/protocol/ws.rs
View File

@@ -221,11 +221,8 @@ impl WebsocketProtocolHandler {
ProtocolType::WS
};
let peer_addr = PeerAddress::new(
Address::from_socket_addr(socket_addr),
socket_addr.port(),
protocol_type,
);
let peer_addr =
PeerAddress::new(SocketAddress::from_socket_addr(socket_addr), protocol_type);
let conn = NetworkConnection::WsAccepted(WebsocketNetworkConnection::new(
self.inner.tls,
@@ -235,7 +232,10 @@ impl WebsocketProtocolHandler {
.network_manager
.clone()
.on_new_connection(
ConnectionDescriptor::new(peer_addr, self.inner.local_address),
ConnectionDescriptor::new(
peer_addr,
SocketAddress::from_socket_addr(self.inner.local_address),
),
conn,
)
.await?;
@@ -248,27 +248,20 @@ impl WebsocketProtocolHandler {
dial_info: &DialInfo,
) -> Result<NetworkConnection, String> {
// Split dial info up
let (tls, request, domain, port, protocol_type) = match &dial_info {
DialInfo::WS(di) => (
false,
format!("ws://{}:{}/{}", di.host, di.port, di.path),
di.host.clone(),
di.port,
ProtocolType::WS,
),
DialInfo::WSS(di) => (
true,
format!("wss://{}:{}/{}", di.host, di.port, di.path),
di.host.clone(),
di.port,
ProtocolType::WSS,
),
let (tls, protocol_type, scheme) = match &dial_info {
DialInfo::WS(_) => (false, ProtocolType::WS, "ws"),
DialInfo::WSS(_) => (true, ProtocolType::WSS, "wss"),
_ => panic!("invalid dialinfo for WS/WSS protocol"),
};
let request = dial_info.request().unwrap();
let split_url = SplitUrl::from_str(&request)?;
if split_url.scheme != scheme {
return Err("invalid websocket url scheme".to_string());
}
let domain = split_url.host.clone();
// Resolve remote address
let remote_ip_addr = dial_info.resolve()?;
let remote_socket_addr = SocketAddr::new(remote_ip_addr, port);
let remote_socket_addr = dial_info.to_socket_addr();
// Make a shared socket
let socket = new_shared_tcp_socket(local_address)?;
@@ -304,15 +297,17 @@ impl WebsocketProtocolHandler {
// Make the connection descriptor peer address
let peer_addr = PeerAddress::new(
Address::from_socket_addr(remote_socket_addr),
port,
SocketAddress::from_socket_addr(remote_socket_addr),
ProtocolType::WSS,
);
// Register the WSS connection
network_manager
.on_new_connection(
ConnectionDescriptor::new(peer_addr, actual_local_addr),
ConnectionDescriptor::new(
peer_addr,
SocketAddress::from_socket_addr(actual_local_addr),
),
conn.clone(),
)
.await?;
@@ -326,15 +321,17 @@ impl WebsocketProtocolHandler {
// Make the connection descriptor peer address
let peer_addr = PeerAddress::new(
Address::from_socket_addr(remote_socket_addr),
port,
SocketAddress::from_socket_addr(remote_socket_addr),
ProtocolType::WS,
);
// Register the WS connection
network_manager
.on_new_connection(
ConnectionDescriptor::new(peer_addr, actual_local_addr),
ConnectionDescriptor::new(
peer_addr,
SocketAddress::from_socket_addr(actual_local_addr),
),
conn.clone(),
)
.await?;

35
veilid-core/src/intf/wasm/network/mod.rs
View File

@@ -12,12 +12,15 @@ pub use protocol::*;
struct NetworkInner {
network_manager: NetworkManager,
stop_network: Eventual,
network_started: bool,
network_needs_restart: bool,
protocol_config: Option<ProtocolConfig>,
//join_handle: TryJoin?
}
#[derive(Clone)]
pub struct Network {
config: VeilidConfig,
inner: Arc<Mutex<NetworkInner>>,
}
@@ -26,13 +29,15 @@ impl Network {
NetworkInner {
network_manager,
stop_network: Eventual::new(),
network_started: false,
network_needs_restart: false,
//join_handle: None,
protocol_config: None, //join_handle: None,
}
}
pub fn new(network_manager: NetworkManager) -> Self {
Self {
config: network_manager.config(),
inner: Arc::new(Mutex::new(Self::new_inner(network_manager))),
}
}
@@ -150,10 +155,21 @@ impl Network {
/////////////////////////////////////////////////////////////////
pub async fn startup(&self) -> Result<(), String> {
//let network_manager = self.inner.lock().network_manager.clone();
//let config_shared = network_manager.core().config();
//let config = config_shared.get();
// get protocol config
self.inner.lock().protocol_config = Some({
let c = self.config.get();
ProtocolConfig {
udp_enabled: false, //c.network.protocol.udp.enabled && c.capabilities.protocol_udp,
tcp_connect: false, //c.network.protocol.tcp.connect && c.capabilities.protocol_connect_tcp,
tcp_listen: false, //c.network.protocol.tcp.listen && c.capabilities.protocol_accept_tcp,
ws_connect: c.network.protocol.ws.connect && c.capabilities.protocol_connect_ws,
ws_listen: c.network.protocol.ws.listen && c.capabilities.protocol_accept_ws,
wss_connect: c.network.protocol.wss.connect && c.capabilities.protocol_connect_wss,
wss_listen: c.network.protocol.wss.listen && c.capabilities.protocol_accept_wss,
}
});
self.inner.lock().network_started = true;
Ok(())
}
@@ -179,9 +195,16 @@ impl Network {
}
//////////////////////////////////////////
pub fn get_network_class(&self) -> NetworkClass {
pub fn get_network_class(&self) -> Option<NetworkClass> {
// xxx eventually detect tor browser?
return NetworkClass::WebApp;
return if self.inner.lock().network_started {
Some(NetworkClass::WebApp)
} else {
None
};
}
pub fn get_protocol_config(&self) -> Option<ProtocolConfig> {
self.inner.lock().protocol_config.clone()
}
//////////////////////////////////////////