server notified of client profile updates
This commit is contained in:
@@ -13,6 +13,7 @@ pub struct Profile {
|
|||||||
#[serde(tag = "type", rename_all = "camelCase")]
|
#[serde(tag = "type", rename_all = "camelCase")]
|
||||||
pub enum ClientMessage {
|
pub enum ClientMessage {
|
||||||
Register { profile: Profile, signature: String },
|
Register { profile: Profile, signature: String },
|
||||||
|
ProfileUpdated { profile: Profile, signature: String },
|
||||||
Signed { payload: String, signature: String },
|
Signed { payload: String, signature: String },
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -34,3 +35,11 @@ pub fn register_bytes(challenge: &str, profile: &Profile) -> Vec<u8> {
|
|||||||
pub fn message_bytes(payload: &str) -> Vec<u8> {
|
pub fn message_bytes(payload: &str) -> Vec<u8> {
|
||||||
format!("wyd-message-v{VERSION}\n{payload}").into_bytes()
|
format!("wyd-message-v{VERSION}\n{payload}").into_bytes()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn profile_bytes(profile: &Profile) -> Vec<u8> {
|
||||||
|
format!(
|
||||||
|
"wyd-profile-v{VERSION}\n{}\n{}",
|
||||||
|
profile.id, profile.display_name
|
||||||
|
)
|
||||||
|
.into_bytes()
|
||||||
|
}
|
||||||
|
|||||||
@@ -13,7 +13,9 @@ use ed25519_dalek::{Signature, Verifier, VerifyingKey};
|
|||||||
use futures_util::{SinkExt, StreamExt};
|
use futures_util::{SinkExt, StreamExt};
|
||||||
use tokio::sync::{Mutex, mpsc};
|
use tokio::sync::{Mutex, mpsc};
|
||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
use wyd_common::{ClientMessage, Profile, ServerMessage, message_bytes, register_bytes};
|
use wyd_common::{
|
||||||
|
ClientMessage, Profile, ServerMessage, message_bytes, profile_bytes, register_bytes,
|
||||||
|
};
|
||||||
|
|
||||||
type Clients = Arc<Mutex<HashMap<String, Client>>>;
|
type Clients = Arc<Mutex<HashMap<String, Client>>>;
|
||||||
|
|
||||||
@@ -91,14 +93,21 @@ async fn connected(mut socket: WebSocket, clients: Clients) {
|
|||||||
if writer.send(Message::Ping(Vec::new().into())).await.is_err() { break; }
|
if writer.send(Message::Ping(Vec::new().into())).await.is_err() { break; }
|
||||||
}
|
}
|
||||||
message = reader.next() => match message {
|
message = reader.next() => match message {
|
||||||
Some(Ok(Message::Text(text))) => {
|
Some(Ok(Message::Text(text))) => match serde_json::from_str(&text) {
|
||||||
let Ok(ClientMessage::Signed { payload, signature }) = serde_json::from_str(&text) else { break };
|
Ok(ClientMessage::Signed { payload, signature }) => {
|
||||||
let registered_key = clients.lock().await.get(&public_key)
|
let registered_key = clients.lock().await.get(&public_key)
|
||||||
.filter(|client| client.connection_id == connection_id)
|
.filter(|client| client.connection_id == connection_id)
|
||||||
.map(|client| client.key);
|
.map(|client| client.key);
|
||||||
let Some(registered_key) = registered_key else { break };
|
let Some(registered_key) = registered_key else { break };
|
||||||
if !verify(®istered_key, &message_bytes(&payload), &signature) { break; }
|
if !verify(®istered_key, &message_bytes(&payload), &signature) { break; }
|
||||||
// The message is authenticated. Domain routing comes next.
|
// The message is authenticated. Domain routing comes next.
|
||||||
|
}
|
||||||
|
Ok(ClientMessage::ProfileUpdated { profile, signature }) => {
|
||||||
|
if !update_profile(&clients, &public_key, connection_id, profile, &signature).await {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
_ => break,
|
||||||
}
|
}
|
||||||
Some(Ok(Message::Ping(data))) => {
|
Some(Ok(Message::Ping(data))) => {
|
||||||
if writer.send(Message::Pong(data)).await.is_err() { break; }
|
if writer.send(Message::Pong(data)).await.is_err() { break; }
|
||||||
@@ -112,6 +121,27 @@ async fn connected(mut socket: WebSocket, clients: Clients) {
|
|||||||
remove(&clients, &public_key, connection_id).await;
|
remove(&clients, &public_key, connection_id).await;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn update_profile(
|
||||||
|
clients: &Clients,
|
||||||
|
public_key: &str,
|
||||||
|
connection_id: Uuid,
|
||||||
|
profile: Profile,
|
||||||
|
signature: &str,
|
||||||
|
) -> bool {
|
||||||
|
let mut clients = clients.lock().await;
|
||||||
|
let Some(client) = clients
|
||||||
|
.get_mut(public_key)
|
||||||
|
.filter(|client| client.connection_id == connection_id)
|
||||||
|
else {
|
||||||
|
return false;
|
||||||
|
};
|
||||||
|
if profile.id != public_key || !verify(&client.key, &profile_bytes(&profile), signature) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
client.profile = profile;
|
||||||
|
true
|
||||||
|
}
|
||||||
|
|
||||||
async fn remove(clients: &Clients, public_key: &str, connection_id: Uuid) {
|
async fn remove(clients: &Clients, public_key: &str, connection_id: Uuid) {
|
||||||
let mut clients = clients.lock().await;
|
let mut clients = clients.lock().await;
|
||||||
if clients
|
if clients
|
||||||
@@ -182,4 +212,43 @@ mod tests {
|
|||||||
&signature,
|
&signature,
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn authenticated_profile_update_changes_the_registered_client() {
|
||||||
|
let signing_key = SigningKey::from_bytes(&[7; 32]);
|
||||||
|
let public_key = URL_SAFE_NO_PAD.encode(signing_key.verifying_key().to_bytes());
|
||||||
|
let connection_id = Uuid::new_v4();
|
||||||
|
let (sender, _receiver) = mpsc::channel(1);
|
||||||
|
let clients = Clients::default();
|
||||||
|
clients.lock().await.insert(
|
||||||
|
public_key.clone(),
|
||||||
|
Client {
|
||||||
|
connection_id,
|
||||||
|
key: signing_key.verifying_key(),
|
||||||
|
profile: Profile {
|
||||||
|
id: public_key.clone(),
|
||||||
|
display_name: "Old".to_owned(),
|
||||||
|
},
|
||||||
|
sender,
|
||||||
|
},
|
||||||
|
);
|
||||||
|
let profile = Profile {
|
||||||
|
id: public_key.clone(),
|
||||||
|
display_name: "New".to_owned(),
|
||||||
|
};
|
||||||
|
let signature =
|
||||||
|
URL_SAFE_NO_PAD.encode(signing_key.sign(&profile_bytes(&profile)).to_bytes());
|
||||||
|
|
||||||
|
assert!(update_profile(&clients, &public_key, connection_id, profile, &signature).await);
|
||||||
|
assert_eq!(
|
||||||
|
clients
|
||||||
|
.lock()
|
||||||
|
.await
|
||||||
|
.get(&public_key)
|
||||||
|
.unwrap()
|
||||||
|
.profile
|
||||||
|
.display_name,
|
||||||
|
"New"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,9 +7,11 @@ use serde::{Deserialize, Serialize};
|
|||||||
use specta::Type;
|
use specta::Type;
|
||||||
use tauri::{AppHandle, Manager, State};
|
use tauri::{AppHandle, Manager, State};
|
||||||
use tauri_specta::Event;
|
use tauri_specta::Event;
|
||||||
use tokio::sync::mpsc;
|
use tokio::sync::{mpsc, watch};
|
||||||
use tokio_tungstenite::tungstenite::Message;
|
use tokio_tungstenite::tungstenite::Message;
|
||||||
use wyd_common::{ClientMessage, Profile, ServerMessage, message_bytes, register_bytes};
|
use wyd_common::{
|
||||||
|
ClientMessage, Profile, ServerMessage, message_bytes, profile_bytes, register_bytes,
|
||||||
|
};
|
||||||
|
|
||||||
use crate::db::AppDatabase;
|
use crate::db::AppDatabase;
|
||||||
use crate::keypair::AppKeypair;
|
use crate::keypair::AppKeypair;
|
||||||
@@ -43,6 +45,7 @@ pub struct NetworkStatusChanged {
|
|||||||
pub struct Network {
|
pub struct Network {
|
||||||
senders: Mutex<HashMap<String, mpsc::Sender<String>>>,
|
senders: Mutex<HashMap<String, mpsc::Sender<String>>>,
|
||||||
statuses: Statuses,
|
statuses: Statuses,
|
||||||
|
profile: watch::Sender<crate::user::User>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Network {
|
impl Network {
|
||||||
@@ -56,12 +59,17 @@ impl Network {
|
|||||||
.try_send(payload)
|
.try_send(payload)
|
||||||
.map_err(|error| format!("remote is not ready: {error}"))
|
.map_err(|error| format!("remote is not ready: {error}"))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn update_profile(&self, profile: crate::user::User) {
|
||||||
|
self.profile.send_replace(profile);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn init(handle: &AppHandle) -> Result<(), Box<dyn std::error::Error>> {
|
pub async fn init(handle: &AppHandle) -> Result<(), Box<dyn std::error::Error>> {
|
||||||
let database = handle.state::<AppDatabase>();
|
let database = handle.state::<AppDatabase>();
|
||||||
let keypair = handle.state::<AppKeypair>().inner().clone();
|
let keypair = handle.state::<AppKeypair>().inner().clone();
|
||||||
let profile = crate::profile::get(&database, keypair.public_key()).await?;
|
let profile = crate::profile::get(&database, keypair.public_key()).await?;
|
||||||
|
let (profile_sender, profile_receiver) = watch::channel(profile);
|
||||||
let remotes = remotes::all(&database).await?;
|
let remotes = remotes::all(&database).await?;
|
||||||
let mut senders = HashMap::new();
|
let mut senders = HashMap::new();
|
||||||
let statuses = Statuses::default();
|
let statuses = Statuses::default();
|
||||||
@@ -74,7 +82,7 @@ pub async fn init(handle: &AppHandle) -> Result<(), Box<dyn std::error::Error>>
|
|||||||
handle.clone(),
|
handle.clone(),
|
||||||
statuses.clone(),
|
statuses.clone(),
|
||||||
remote,
|
remote,
|
||||||
profile.clone(),
|
profile_receiver.clone(),
|
||||||
keypair.clone(),
|
keypair.clone(),
|
||||||
receiver,
|
receiver,
|
||||||
));
|
));
|
||||||
@@ -83,6 +91,7 @@ pub async fn init(handle: &AppHandle) -> Result<(), Box<dyn std::error::Error>>
|
|||||||
handle.manage(Network {
|
handle.manage(Network {
|
||||||
senders: Mutex::new(senders),
|
senders: Mutex::new(senders),
|
||||||
statuses,
|
statuses,
|
||||||
|
profile: profile_sender,
|
||||||
});
|
});
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
@@ -91,7 +100,7 @@ async fn run(
|
|||||||
handle: AppHandle,
|
handle: AppHandle,
|
||||||
statuses: Statuses,
|
statuses: Statuses,
|
||||||
remote: Remote,
|
remote: Remote,
|
||||||
profile: crate::user::User,
|
mut profiles: watch::Receiver<crate::user::User>,
|
||||||
keypair: AppKeypair,
|
keypair: AppKeypair,
|
||||||
mut outgoing: mpsc::Receiver<String>,
|
mut outgoing: mpsc::Receiver<String>,
|
||||||
) {
|
) {
|
||||||
@@ -101,7 +110,7 @@ async fn run(
|
|||||||
&handle,
|
&handle,
|
||||||
&statuses,
|
&statuses,
|
||||||
&remote,
|
&remote,
|
||||||
&profile,
|
&mut profiles,
|
||||||
&keypair,
|
&keypair,
|
||||||
&mut outgoing,
|
&mut outgoing,
|
||||||
)
|
)
|
||||||
@@ -118,7 +127,7 @@ async fn connect(
|
|||||||
handle: &AppHandle,
|
handle: &AppHandle,
|
||||||
statuses: &Statuses,
|
statuses: &Statuses,
|
||||||
remote: &Remote,
|
remote: &Remote,
|
||||||
profile: &crate::user::User,
|
profiles: &mut watch::Receiver<crate::user::User>,
|
||||||
keypair: &AppKeypair,
|
keypair: &AppKeypair,
|
||||||
outgoing: &mut mpsc::Receiver<String>,
|
outgoing: &mut mpsc::Receiver<String>,
|
||||||
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
|
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
|
||||||
@@ -131,15 +140,16 @@ async fn connect(
|
|||||||
}
|
}
|
||||||
_ => return Err("server did not send a compatible challenge".into()),
|
_ => return Err("server did not send a compatible challenge".into()),
|
||||||
};
|
};
|
||||||
let profile = Profile {
|
let current = profiles.borrow_and_update().clone();
|
||||||
id: profile.id.clone(),
|
let registration_profile = Profile {
|
||||||
display_name: profile.display_name.clone(),
|
id: current.id,
|
||||||
|
display_name: current.display_name,
|
||||||
};
|
};
|
||||||
send(
|
send(
|
||||||
&mut writer,
|
&mut writer,
|
||||||
&ClientMessage::Register {
|
&ClientMessage::Register {
|
||||||
signature: keypair.sign(®ister_bytes(&challenge, &profile)),
|
signature: keypair.sign(®ister_bytes(&challenge, ®istration_profile)),
|
||||||
profile,
|
profile: registration_profile,
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
@@ -158,6 +168,18 @@ async fn connect(
|
|||||||
payload,
|
payload,
|
||||||
}).await?;
|
}).await?;
|
||||||
}
|
}
|
||||||
|
changed = profiles.changed() => {
|
||||||
|
changed.map_err(|_| "profile sender closed")?;
|
||||||
|
let current = profiles.borrow_and_update().clone();
|
||||||
|
let profile = Profile {
|
||||||
|
id: current.id,
|
||||||
|
display_name: current.display_name,
|
||||||
|
};
|
||||||
|
send(&mut writer, &ClientMessage::ProfileUpdated {
|
||||||
|
signature: keypair.sign(&profile_bytes(&profile)),
|
||||||
|
profile,
|
||||||
|
}).await?;
|
||||||
|
}
|
||||||
message = reader.next() => match message.ok_or("server closed the socket")?? {
|
message = reader.next() => match message.ok_or("server closed the socket")?? {
|
||||||
Message::Ping(data) => writer.send(Message::Pong(data)).await?,
|
Message::Ping(data) => writer.send(Message::Pong(data)).await?,
|
||||||
Message::Close(_) => return Ok(()),
|
Message::Close(_) => return Ok(()),
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
use crate::db::{self, AppDatabase};
|
use crate::db::{self, AppDatabase};
|
||||||
use crate::keypair::AppKeypair;
|
use crate::keypair::AppKeypair;
|
||||||
|
use crate::network::Network;
|
||||||
use crate::user::User;
|
use crate::user::User;
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use specta::Type;
|
use specta::Type;
|
||||||
@@ -71,11 +72,13 @@ pub async fn update_profile(
|
|||||||
handle: AppHandle,
|
handle: AppHandle,
|
||||||
database: State<'_, AppDatabase>,
|
database: State<'_, AppDatabase>,
|
||||||
keypair: State<'_, AppKeypair>,
|
keypair: State<'_, AppKeypair>,
|
||||||
|
network: State<'_, Network>,
|
||||||
display_name: String,
|
display_name: String,
|
||||||
) -> Result<User, String> {
|
) -> Result<User, String> {
|
||||||
let profile = update(&database, keypair.public_key(), display_name)
|
let profile = update(&database, keypair.public_key(), display_name)
|
||||||
.await
|
.await
|
||||||
.map_err(db::command_error)?;
|
.map_err(db::command_error)?;
|
||||||
|
network.update_profile(profile.clone());
|
||||||
emit_changed(&handle, profile.clone())?;
|
emit_changed(&handle, profile.clone())?;
|
||||||
Ok(profile)
|
Ok(profile)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user