64d9f456ce
Avoid large logs of 127.0.0.1:5959 attack payloads See merge request veilid/veilid!158
438 lines
15 KiB
Rust
438 lines
15 KiB
Rust
use crate::settings::*;
|
|
use crate::tools::*;
|
|
use crate::veilid_logs::VeilidLogs;
|
|
use cfg_if::*;
|
|
use futures_util::{future::try_join_all, stream::FuturesUnordered, StreamExt};
|
|
use parking_lot::Mutex;
|
|
use std::collections::HashMap;
|
|
use std::net::SocketAddr;
|
|
use std::sync::Arc;
|
|
use stop_token::future::FutureExt as _;
|
|
use stop_token::*;
|
|
use tracing::*;
|
|
use veilid_core::json_api::JsonRequestProcessor;
|
|
use veilid_core::tools::*;
|
|
use veilid_core::*;
|
|
use wg::AsyncWaitGroup;
|
|
|
|
const MAX_NON_JSON_LOGGING: usize = 50;
|
|
|
|
cfg_if! {
|
|
if #[cfg(feature="rt-async-std")] {
|
|
use async_std::io::prelude::BufReadExt;
|
|
use async_std::io::WriteExt;
|
|
} else if #[cfg(feature="rt-tokio")] {
|
|
use tokio::io::AsyncBufReadExt;
|
|
use tokio::io::AsyncWriteExt;
|
|
} else {
|
|
compile_error!("needs executor implementation")
|
|
}
|
|
}
|
|
|
|
// --- Client API Server-Side ---------------------------------
|
|
|
|
type ClientApiAllFuturesJoinHandle = MustJoinHandle<std::io::Result<Vec<()>>>;
|
|
|
|
struct RequestLine {
|
|
// Request to process
|
|
line: String,
|
|
// Where to send the response
|
|
responses_tx: flume::Sender<String>,
|
|
}
|
|
|
|
struct ClientApiInner {
|
|
veilid_api: veilid_core::VeilidAPI,
|
|
veilid_logs: VeilidLogs,
|
|
settings: Settings,
|
|
stop: Option<StopSource>,
|
|
join_handle: Option<ClientApiAllFuturesJoinHandle>,
|
|
update_channels: HashMap<(SocketAddr, SocketAddr), flume::Sender<String>>,
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
pub struct ClientApi {
|
|
inner: Arc<Mutex<ClientApiInner>>,
|
|
}
|
|
|
|
impl ClientApi {
|
|
#[instrument(level = "trace", skip_all)]
|
|
pub fn new(
|
|
veilid_api: veilid_core::VeilidAPI,
|
|
veilid_logs: VeilidLogs,
|
|
settings: Settings,
|
|
) -> Self {
|
|
Self {
|
|
inner: Arc::new(Mutex::new(ClientApiInner {
|
|
veilid_api,
|
|
veilid_logs,
|
|
settings,
|
|
stop: Some(StopSource::new()),
|
|
join_handle: None,
|
|
update_channels: HashMap::new(),
|
|
})),
|
|
}
|
|
}
|
|
|
|
#[instrument(level = "trace", skip_all)]
|
|
fn shutdown(&self) {
|
|
trace!("ClientApi::shutdown");
|
|
|
|
crate::server::shutdown();
|
|
}
|
|
|
|
fn change_log_level(
|
|
&self,
|
|
layer: String,
|
|
log_level: VeilidConfigLogLevel,
|
|
) -> VeilidAPIResult<()> {
|
|
trace!("ClientApi::change_log_level");
|
|
|
|
let veilid_logs = self.inner.lock().veilid_logs.clone();
|
|
veilid_logs.change_log_level(layer, log_level)
|
|
}
|
|
|
|
#[instrument(level = "trace", skip(self))]
|
|
pub async fn stop(&self) {
|
|
trace!("ClientApi::stop requested");
|
|
let jh = {
|
|
let mut inner = self.inner.lock();
|
|
if inner.join_handle.is_none() {
|
|
trace!("ClientApi stop ignored");
|
|
return;
|
|
}
|
|
drop(inner.stop.take());
|
|
inner.join_handle.take().unwrap()
|
|
};
|
|
trace!("ClientApi::stop: waiting for stop");
|
|
if let Err(err) = jh.await {
|
|
eprintln!("{}", err);
|
|
}
|
|
trace!("ClientApi::stop: stopped");
|
|
}
|
|
|
|
async fn handle_incoming(self, bind_addr: SocketAddr) -> std::io::Result<()> {
|
|
let listener = TcpListener::bind(bind_addr).await?;
|
|
debug!("Client API listening on: {:?}", bind_addr);
|
|
|
|
// Process the incoming accept stream
|
|
cfg_if! {
|
|
if #[cfg(feature="rt-async-std")] {
|
|
let mut incoming_stream = listener.incoming();
|
|
} else {
|
|
let mut incoming_stream = tokio_stream::wrappers::TcpListenerStream::new(listener);
|
|
}
|
|
}
|
|
|
|
// Make wait group for all incoming connections
|
|
let awg = AsyncWaitGroup::new();
|
|
|
|
let stop_token = self.inner.lock().stop.as_ref().unwrap().token();
|
|
while let Ok(Some(stream_result)) =
|
|
incoming_stream.next().timeout_at(stop_token.clone()).await
|
|
{
|
|
// Get the stream to process
|
|
let stream = stream_result?;
|
|
stream.set_nodelay(true)?;
|
|
|
|
// Increment wait group
|
|
awg.add(1);
|
|
let t_awg = awg.clone();
|
|
|
|
// Process the connection
|
|
spawn(self.clone().handle_connection(stream, t_awg)).detach();
|
|
}
|
|
|
|
// Wait for all connections to terminate
|
|
awg.wait().await;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
// Process control messages for the server
|
|
async fn process_control(self, args: Vec<String>) -> VeilidAPIResult<String> {
|
|
if args.len() == 0 {
|
|
apibail_generic!("no control request specified");
|
|
}
|
|
if args[0] == "Shutdown" {
|
|
if args.len() != 1 {
|
|
apibail_generic!("wrong number of arguments");
|
|
}
|
|
self.shutdown();
|
|
Ok("".to_owned())
|
|
} else if args[0] == "ChangeLogLevel" {
|
|
if args.len() != 3 {
|
|
apibail_generic!("wrong number of arguments");
|
|
}
|
|
let log_level = VeilidConfigLogLevel::from_str(&args[2])?;
|
|
self.change_log_level(args[1].clone(), log_level)?;
|
|
Ok("".to_owned())
|
|
} else if args[0] == "GetServerSettings" {
|
|
if args.len() != 1 {
|
|
apibail_generic!("wrong number of arguments");
|
|
}
|
|
let settings = self.inner.lock().settings.clone();
|
|
let settings = &*settings.read();
|
|
let settings_json_string = serialize_json(settings);
|
|
let mut settings_json =
|
|
json::parse(&settings_json_string).map_err(VeilidAPIError::internal)?;
|
|
settings_json["core"]["network"].remove("node_id_secret");
|
|
settings_json["core"]["protected_store"].remove("device_encryption_key_password");
|
|
settings_json["core"]["protected_store"].remove("new_device_encryption_key_password");
|
|
let safe_settings_json = settings_json.to_string();
|
|
Ok(safe_settings_json)
|
|
} else if args[0] == "EmitSchema" {
|
|
if args.len() != 2 {
|
|
apibail_generic!("wrong number of arguments");
|
|
}
|
|
|
|
let mut schemas = HashMap::<String, String>::new();
|
|
veilid_core::json_api::emit_schemas(&mut schemas);
|
|
|
|
let Some(schema) = schemas.get(&args[1]) else {
|
|
apibail_invalid_argument!("invalid schema", "schema", args[1].clone());
|
|
};
|
|
|
|
Ok(schema.clone())
|
|
} else {
|
|
apibail_generic!("unknown control message");
|
|
}
|
|
}
|
|
|
|
async fn process_request_line(
|
|
self,
|
|
jrp: JsonRequestProcessor,
|
|
request_line: RequestLine,
|
|
) -> VeilidAPIResult<Option<RequestLine>> {
|
|
let line = request_line.line.trim_start();
|
|
|
|
// Avoid logging failed deserialization of large adversarial payloads from
|
|
// http://127.0.0.1:5959 by using an initial colon to force a parse error.
|
|
let sanitized_line = if line.len() > MAX_NON_JSON_LOGGING && !line.starts_with("{") {
|
|
":skipped long input that's not a JSON object".to_string()
|
|
} else {
|
|
line.to_string()
|
|
};
|
|
|
|
let responses_tx = request_line.responses_tx;
|
|
|
|
// Unmarshal NDJSON - newline => json
|
|
// (trim all whitespace around input lines just to make things more permissive for API users)
|
|
let request: json_api::Request = deserialize_json(&sanitized_line)?;
|
|
|
|
// See if this is a control message or a veilid-core message
|
|
let response = if let json_api::RequestOp::Control { args } = request.op {
|
|
// Process control messages
|
|
json_api::Response {
|
|
id: request.id,
|
|
op: json_api::ResponseOp::Control {
|
|
result: json_api::to_json_api_result(self.process_control(args).await),
|
|
},
|
|
}
|
|
} else {
|
|
// Process with ndjson api
|
|
jrp.clone().process_request(request).await
|
|
};
|
|
|
|
// Marshal json + newline => NDJSON
|
|
let response_string = serialize_json(json_api::RecvMessage::Response(response)) + "\n";
|
|
if let Err(e) = responses_tx.send_async(response_string).await {
|
|
eprintln!("response not sent: {}", e)
|
|
}
|
|
VeilidAPIResult::Ok(None)
|
|
}
|
|
|
|
async fn next_request_line(
|
|
requests_rx: flume::Receiver<Option<RequestLine>>,
|
|
) -> VeilidAPIResult<Option<RequestLine>> {
|
|
Ok(requests_rx.recv_async().await.ok().flatten())
|
|
}
|
|
|
|
async fn receive_requests<R: AsyncBufReadExt + Unpin>(
|
|
self,
|
|
mut reader: R,
|
|
requests_tx: flume::Sender<Option<RequestLine>>,
|
|
responses_tx: flume::Sender<String>,
|
|
) -> VeilidAPIResult<Option<RequestLine>> {
|
|
let mut linebuf = String::new();
|
|
while let Ok(size) = reader.read_line(&mut linebuf).await {
|
|
// Eof?
|
|
if size == 0 {
|
|
break;
|
|
}
|
|
|
|
// Put the processing in the async queue
|
|
let line = linebuf.trim().to_owned();
|
|
linebuf.clear();
|
|
|
|
// Ignore newlines
|
|
if line.len() == 0 {
|
|
continue;
|
|
}
|
|
|
|
// Enqueue the line for processing in parallel
|
|
let request_line = RequestLine {
|
|
line,
|
|
responses_tx: responses_tx.clone(),
|
|
};
|
|
if let Err(e) = requests_tx.send_async(Some(request_line)).await {
|
|
eprintln!("failed to enqueue request: {}", e);
|
|
break;
|
|
}
|
|
}
|
|
|
|
VeilidAPIResult::Ok(None)
|
|
}
|
|
|
|
async fn send_responses<W: AsyncWriteExt + Unpin>(
|
|
self,
|
|
responses_rx: flume::Receiver<String>,
|
|
mut writer: W,
|
|
) -> VeilidAPIResult<Option<RequestLine>> {
|
|
while let Ok(resp) = responses_rx.recv_async().await {
|
|
if let Err(_) = writer.write_all(resp.as_bytes()).await {
|
|
break;
|
|
}
|
|
}
|
|
VeilidAPIResult::Ok(None)
|
|
}
|
|
|
|
pub async fn handle_connection(self, stream: TcpStream, awg: AsyncWaitGroup) {
|
|
// Get address of peer
|
|
let peer_addr = match stream.peer_addr() {
|
|
Ok(v) => v,
|
|
Err(e) => {
|
|
eprintln!("can't get peer address: {}", e);
|
|
return;
|
|
}
|
|
};
|
|
// Get local address
|
|
let local_addr = match stream.local_addr() {
|
|
Ok(v) => v,
|
|
Err(e) => {
|
|
eprintln!("can't get local address: {}", e);
|
|
return;
|
|
}
|
|
};
|
|
// Get connection tuple
|
|
let conn_tuple = (local_addr, peer_addr);
|
|
|
|
debug!(
|
|
"Accepted Client API Connection: {:?} -> {:?}",
|
|
peer_addr, local_addr
|
|
);
|
|
|
|
// Make stop token to quit when stop() is requested externally
|
|
let stop_token = self.inner.lock().stop.as_ref().unwrap().token();
|
|
|
|
// Split into reader and writer halves
|
|
// with line buffering on the reader
|
|
cfg_if! {
|
|
if #[cfg(feature="rt-async-std")] {
|
|
use futures_util::AsyncReadExt;
|
|
let (reader, mut writer) = stream.split();
|
|
let reader = BufReader::new(reader);
|
|
} else {
|
|
let (reader, writer) = stream.into_split();
|
|
let reader = BufReader::new(reader);
|
|
}
|
|
}
|
|
|
|
// Make request processor for this connection
|
|
let api = self.inner.lock().veilid_api.clone();
|
|
let jrp = json_api::JsonRequestProcessor::new(api);
|
|
|
|
// Futures to process unordered
|
|
let mut unord = FuturesUnordered::new();
|
|
|
|
// Requests and responses are done serially to the socket
|
|
// but the requests are processed in parallel by the FuturesUnordered
|
|
let (requests_tx, requests_rx) = flume::unbounded();
|
|
let (responses_tx, responses_rx) = flume::unbounded();
|
|
|
|
// Start sending updates
|
|
self.inner
|
|
.lock()
|
|
.update_channels
|
|
.insert(conn_tuple, responses_tx.clone());
|
|
|
|
// Request receive processor future
|
|
// Receives from socket and enqueues RequestLines
|
|
// Completes when the connection is closed or there is a failure
|
|
unord.push(system_boxed(self.clone().receive_requests(
|
|
reader,
|
|
requests_tx,
|
|
responses_tx,
|
|
)));
|
|
|
|
// Response send processor
|
|
// Sends finished response strings out the socket
|
|
// Completes when the responses channel is closed
|
|
unord.push(system_boxed(
|
|
self.clone().send_responses(responses_rx, writer),
|
|
));
|
|
|
|
// Add future to process first request
|
|
unord.push(system_boxed(Self::next_request_line(requests_rx.clone())));
|
|
|
|
// Send and receive until we're done or a stop is requested
|
|
while let Ok(Some(r)) = unord.next().timeout_at(stop_token.clone()).await {
|
|
// See if we got some work to do
|
|
let request_line = match r {
|
|
Ok(Some(request_line)) => {
|
|
// Add future to process next request
|
|
unord.push(system_boxed(Self::next_request_line(requests_rx.clone())));
|
|
|
|
// Socket receive future returned something to process
|
|
request_line
|
|
}
|
|
Ok(None) => {
|
|
// Non-request future finished
|
|
continue;
|
|
}
|
|
Err(e) => {
|
|
// Connection processing failure, abort
|
|
eprintln!("Connection processing failure: {}", e);
|
|
break;
|
|
}
|
|
};
|
|
|
|
// Enqueue unordered future to process request line in parallel
|
|
unord.push(system_boxed(
|
|
self.clone().process_request_line(jrp.clone(), request_line),
|
|
));
|
|
}
|
|
|
|
// Stop sending updates
|
|
self.inner.lock().update_channels.remove(&conn_tuple);
|
|
|
|
debug!(
|
|
"Closed Client API Connection: {:?} -> {:?}",
|
|
peer_addr, local_addr
|
|
);
|
|
|
|
awg.done();
|
|
}
|
|
|
|
pub fn handle_update(&self, veilid_update: veilid_core::VeilidUpdate) {
|
|
// serialize update to NDJSON
|
|
let veilid_update = serialize_json(json_api::RecvMessage::Update(veilid_update)) + "\n";
|
|
|
|
// Pass other updates to clients
|
|
let inner = self.inner.lock();
|
|
for ch in inner.update_channels.values() {
|
|
if let Err(_) = ch.send(veilid_update.clone()) {
|
|
// eprintln!("failed to send update: {}", e);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[instrument(level = "trace", skip(self))]
|
|
pub fn run(&self, bind_addrs: Vec<SocketAddr>) {
|
|
let bind_futures = bind_addrs
|
|
.iter()
|
|
.map(|addr| self.clone().handle_incoming(*addr));
|
|
let bind_futures_join = try_join_all(bind_futures);
|
|
self.inner.lock().join_handle = Some(spawn(bind_futures_join));
|
|
}
|
|
}
|