feat: show client IP in logs

This commit is contained in:
Greg Heartsfield 2022-11-02 18:33:44 -05:00
parent caffbbbede
commit 8ecce3f566
5 changed files with 35 additions and 20 deletions

View File

@ -14,6 +14,8 @@ const MAX_SUBSCRIPTION_ID_LEN: usize = 256;
/// State for a client connection /// State for a client connection
pub struct ClientConn { pub struct ClientConn {
/// Client IP (either from socket, or configured proxy header
client_ip: String,
/// Unique client identifier generated at connection time /// Unique client identifier generated at connection time
client_id: Uuid, client_id: Uuid,
/// The current set of active client subscriptions /// The current set of active client subscriptions
@ -24,16 +26,17 @@ pub struct ClientConn {
impl Default for ClientConn { impl Default for ClientConn {
fn default() -> Self { fn default() -> Self {
Self::new() Self::new("unknown".to_owned())
} }
} }
impl ClientConn { impl ClientConn {
/// Create a new, empty connection state. /// Create a new, empty connection state.
#[must_use] #[must_use]
pub fn new() -> Self { pub fn new(client_ip: String) -> Self {
let client_id = Uuid::new_v4(); let client_id = Uuid::new_v4();
ClientConn { ClientConn {
client_ip,
client_id, client_id,
subscriptions: HashMap::new(), subscriptions: HashMap::new(),
max_subs: 32, max_subs: 32,
@ -47,6 +50,11 @@ impl ClientConn {
self.client_id.to_string().chars().take(8).collect() self.client_id.to_string().chars().take(8).collect()
} }
#[must_use]
pub fn ip(&self) -> &str {
&self.client_ip
}
/// Find all matching subscriptions. /// Find all matching subscriptions.
#[must_use] #[must_use]
pub fn get_matching_subscriptions(&self, e: &Event) -> Vec<&str> { pub fn get_matching_subscriptions(&self, e: &Event) -> Vec<&str> {
@ -102,7 +110,7 @@ impl ClientConn {
// TODO: return notice if subscription did not exist. // TODO: return notice if subscription did not exist.
self.subscriptions.remove(&c.id); self.subscriptions.remove(&c.id);
debug!( debug!(
"removed subscription, currently have {} active subs (cid={})", "removed subscription, currently have {} active subs (cid={:?})",
self.subscriptions.len(), self.subscriptions.len(),
self.client_id self.client_id
); );

View File

@ -316,7 +316,7 @@ pub fn write_event(conn: &mut PooledConnection, e: &Event) -> Result<usize> {
if is_lower_hex(tagval) && (tagval.len() % 2 == 0) { if is_lower_hex(tagval) && (tagval.len() % 2 == 0) {
tx.execute( tx.execute(
"INSERT OR IGNORE INTO tag (event_id, name, value_hex) VALUES (?1, ?2, ?3)", "INSERT OR IGNORE INTO tag (event_id, name, value_hex) VALUES (?1, ?2, ?3)",
params![ev_id, &tagname, hex::decode(&tagval).ok()], params![ev_id, &tagname, hex::decode(tagval).ok()],
)?; )?;
} else { } else {
tx.execute( tx.execute(
@ -510,7 +510,7 @@ fn query_from_filter(f: &ReqFilter) -> (String, Vec<Box<dyn ToSql>>) {
let mut blob_vals: Vec<Box<dyn ToSql>> = vec![]; let mut blob_vals: Vec<Box<dyn ToSql>> = vec![];
for v in val { for v in val {
if (v.len() % 2 == 0) && is_lower_hex(v) { if (v.len() % 2 == 0) && is_lower_hex(v) {
if let Ok(h) = hex::decode(&v) { if let Ok(h) = hex::decode(v) {
blob_vals.push(Box::new(h)); blob_vals.push(Box::new(h));
} }
} else { } else {

View File

@ -34,11 +34,11 @@ fn main() {
// enable tracing with tokio-console // enable tracing with tokio-console
ConsoleLayer::builder().with_default_env().init(); ConsoleLayer::builder().with_default_env().init();
} }
// update with database location // update with database location
if let Some(db) = db_dir { if let Some(db) = db_dir {
settings.database.data_directory = db; settings.database.data_directory = db;
} }
let (_, ctrl_rx): (MpscSender<()>, MpscReceiver<()>) = syncmpsc::channel(); let (_, ctrl_rx): (MpscSender<()>, MpscReceiver<()>) = syncmpsc::channel();
// run this in a new thread // run this in a new thread
let handle = thread::spawn(|| { let handle = thread::spawn(|| {

View File

@ -332,7 +332,7 @@ fn mig_5_to_6(conn: &mut PooledConnection) -> Result<usize> {
if (tagval.len() % 2 == 0) && is_lower_hex(tagval) { if (tagval.len() % 2 == 0) && is_lower_hex(tagval) {
tx.execute( tx.execute(
"INSERT INTO tag (event_id, name, value_hex) VALUES (?1, ?2, ?3);", "INSERT INTO tag (event_id, name, value_hex) VALUES (?1, ?2, ?3);",
params![event_id, tagname, hex::decode(&tagval).ok()], params![event_id, tagname, hex::decode(tagval).ok()],
)?; )?;
} else { } else {
// otherwise, insert as text // otherwise, insert as text

View File

@ -84,11 +84,14 @@ async fn handle_web_request(
Some(config), Some(config),
) )
.await; .await;
// spawn server with info... but include IP here.
let remote_ip = remote_addr.ip().to_string();
tokio::spawn(nostr_server( tokio::spawn(nostr_server(
pool, settings, ws_stream, broadcast, event_tx, shutdown, pool, remote_ip, settings, ws_stream, broadcast, event_tx,
shutdown,
)); ));
} }
// todo: trace, don't print...
Err(e) => println!( Err(e) => println!(
"error when trying to upgrade connection \ "error when trying to upgrade connection \
from address {} to websocket connection. \ from address {} to websocket connection. \
@ -167,7 +170,6 @@ async fn ctrl_c_or_signal(mut shutdown_signal: Receiver<()>) {
info!("Shutting down webserver due to SIGTERM"); info!("Shutting down webserver due to SIGTERM");
break; break;
}, },
} }
} }
} }
@ -386,6 +388,7 @@ fn make_notice_message(msg: &str) -> Message {
/// for all client communication. /// for all client communication.
async fn nostr_server( async fn nostr_server(
pool: db::SqlitePool, pool: db::SqlitePool,
remote_ip: String,
settings: Settings, settings: Settings,
mut ws_stream: WebSocketStream<Upgraded>, mut ws_stream: WebSocketStream<Upgraded>,
broadcast: Sender<Event>, broadcast: Sender<Event>,
@ -395,7 +398,8 @@ async fn nostr_server(
// get a broadcast channel for clients to communicate on // get a broadcast channel for clients to communicate on
let mut bcast_rx = broadcast.subscribe(); let mut bcast_rx = broadcast.subscribe();
// Track internal client state // Track internal client state
let mut conn = conn::ClientConn::new(); let mut conn = conn::ClientConn::new(remote_ip);
// Use the remote IP as the client identifier
let cid = conn.get_client_prefix(); let cid = conn.get_client_prefix();
// Create a channel for receiving query results from the database. // Create a channel for receiving query results from the database.
// we will send out the tx handle to any query we generate. // we will send out the tx handle to any query we generate.
@ -424,11 +428,11 @@ async fn nostr_server(
// and how many it received from queries. // and how many it received from queries.
let mut client_published_event_count: usize = 0; let mut client_published_event_count: usize = 0;
let mut client_received_event_count: usize = 0; let mut client_received_event_count: usize = 0;
info!("new connection for client: {:?}", cid); info!("new connection for client: {:?}, ip: {:?}", cid, conn.ip());
loop { loop {
tokio::select! { tokio::select! {
_ = shutdown.recv() => { _ = shutdown.recv() => {
info!("Shutting client connection down due to shutdown: {:?}", cid); info!("Shutting client connection down due to shutdown: {:?}, ip: {:?}", cid, conn.ip());
// server shutting down, exit loop // server shutting down, exit loop
break; break;
}, },
@ -507,17 +511,17 @@ async fn nostr_server(
Err(WsError::AlreadyClosed | WsError::ConnectionClosed | Err(WsError::AlreadyClosed | WsError::ConnectionClosed |
WsError::Protocol(tungstenite::error::ProtocolError::ResetWithoutClosingHandshake))) WsError::Protocol(tungstenite::error::ProtocolError::ResetWithoutClosingHandshake)))
=> { => {
debug!("websocket close from client: {:?}",cid); debug!("websocket close from client: {:?}, ip: {:?}",cid, conn.ip());
break; break;
}, },
Some(Err(WsError::Io(e))) => { Some(Err(WsError::Io(e))) => {
// IO errors are considered fatal // IO errors are considered fatal
warn!("IO error (client: {:?}): {:?}", cid, e); warn!("IO error (client: {:?}, ip: {:?}): {:?}", cid, conn.ip(), e);
break; break;
} }
x => { x => {
// default condition on error is to close the client connection // default condition on error is to close the client connection
info!("unknown error (client: {:?}): {:?} (closing conn)", cid, x); info!("unknown error (client: {:?}, ip: {:?}): {:?} (closing conn)", cid, conn.ip(), x);
break; break;
} }
}; };
@ -546,7 +550,7 @@ async fn nostr_server(
} }
}, },
Err(_) => { Err(_) => {
info!("client {:?} sent an invalid event", cid); info!("client: {:?} sent an invalid event", cid);
ws_stream.send(make_notice_message("event was invalid")).await.ok(); ws_stream.send(make_notice_message("event was invalid")).await.ok();
} }
} }
@ -592,7 +596,7 @@ async fn nostr_server(
} }
}, },
Err(Error::ConnError) => { Err(Error::ConnError) => {
debug!("got connection close/error, disconnecting client: {:?}",cid); debug!("got connection close/error, disconnecting client: {:?}, ip: {:?}",cid, conn.ip());
break; break;
} }
Err(Error::EventMaxLengthError(s)) => { Err(Error::EventMaxLengthError(s)) => {
@ -615,7 +619,10 @@ async fn nostr_server(
stop_tx.send(()).ok(); stop_tx.send(()).ok();
} }
info!( info!(
"stopping connection for client: {:?} (client sent {} event(s), received {})", "stopping connection for client: {:?}, ip: {:?} (client sent {} event(s), received {})",
cid, client_published_event_count, client_received_event_count cid,
conn.ip(),
client_published_event_count,
client_received_event_count
); );
} }