reya/coop
1
0
Fork 1
Code Issues Pull Requests Releases 5 Activity

track key-sync request state to prevent duplicates

Browse Source
This commit is contained in:
reya
2026-11-04 08:11:43 +07:00
parent 690dc37fc5
commit 198f13bd1d
4 changed files with 386 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/nip4e/src/lib.rs
+269 -99
View File
@@ -1,9 +1,10 @@
use std::cell::Cell;
use std::collections::HashSet;
use std::collections::{HashSet, VecDeque};
use std::path::PathBuf;
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use anyhow::{Context as AnyhowContext, Error, anyhow};
use gpui::{
@@ -59,8 +60,12 @@ impl DeviceEvent {
#[derive(Debug)]
pub struct DeviceRegistry {
signer: Entity<Option<UniversalSigner>>,
/// Whether there is a pending request for encryption key approval
pub(crate) pending_request: bool,
/// Ephemeral client key for the in-flight key request. never persisted
app_keys: Option<Keys>,
/// The key request currently awaiting user approval on this device
pending_request: Option<PendingRequest>,
/// Handled key-sync request ids, bounded and persisted
processed_requests: VecDeque<EventId>,
/// Whether an announcement has been made for this device
pub(crate) announcement_existed: Arc<AtomicBool>,
/// Person directory for requesters shown in notifications
@@ -108,6 +113,20 @@ impl DeviceRegistry {
if event.signer_changed() && settings.read(cx).is_nip4e_enabled(cx) {
// The old account's state must not leak into the new one
this.announcement_existed.store(false, Ordering::Relaxed);
// The in-flight request and client key belong to the old account
this.app_keys = None;
if this.pending_request.take().is_some() {
// The old approval must not target the new account's signer
if let Some(window) = cx.windows().first().copied() {
window
.update(cx, |_view, window, cx| {
window.remove_notification::<DeviceNotification>(cx);
})
.ok();
}
}
this.handle_notifications(cx);
this.get_announcement(cx);
}
@@ -115,14 +134,16 @@ impl DeviceRegistry {
);
cx.defer(move |cx| {
entity
.update(cx, |this, cx| this.handle_notifications(cx))
.ok();
let _ = entity.update(cx, |this, cx| {
this.handle_notifications(cx);
});
});
Self {
signer,
pending_request: false,
app_keys: None,
pending_request: None,
processed_requests: VecDeque::new(),
announcement_existed: Arc::new(AtomicBool::new(false)),
directory: Directory::new(cx),
notification_listener: None,
@@ -141,6 +162,8 @@ impl DeviceRegistry {
let announcement_existed = self.announcement_existed.clone();
let (tx, rx) = flume::bounded::<Event>(100);
let read = credentials::read_processed_requests(cx);
self.notification_listener = Some(cx.background_spawn(async move {
let mut notifications = client.notifications();
let mut processed_events = HashSet::new();
@@ -175,6 +198,21 @@ impl DeviceRegistry {
}));
self.tasks.push(cx.spawn(async move |this, cx| {
// Hydrate the processed-request log before routing any events
match read.await {
Ok(bytes) => {
let processed = bytes
.map(|bytes| parse_processed_requests(&bytes))
.unwrap_or_default();
this.update(cx, |this, _cx| {
this.processed_requests = processed;
})
.ok();
}
Err(e) => log::error!("Failed to read processed key-sync requests: {e}"),
}
while let Ok(event) = rx.recv_async().await {
match event.kind {
Kind::Custom(10044) => {
@@ -191,7 +229,7 @@ impl DeviceRegistry {
// New response event from the master device
Kind::Custom(4455) => {
this.update(cx, |this, cx| {
this.extract_encryption(event, cx);
this.handle_transfer(event, cx);
})?;
}
_ => {}
@@ -201,12 +239,62 @@ impl DeviceRegistry {
}));
}
/// Set whether there is a pending request for encryption key approval
fn set_pending_request(&mut self, pending: bool, cx: &mut Context<Self>) {
self.pending_request = pending;
/// Record a handled key-sync request id, persisting a bounded log
fn mark_request_processed(&mut self, event_id: EventId, cx: &mut Context<Self>) {
if self.processed_requests.contains(&event_id) {
return;
}
self.processed_requests.push_back(event_id);
while self.processed_requests.len() > PROCESSED_REQUESTS_LIMIT {
self.processed_requests.pop_front();
}
let bytes: Vec<u8> = self
.processed_requests
.iter()
.flat_map(|id| id.to_hex().into_bytes())
.collect();
let write = credentials::write_processed_requests(cx, &bytes);
cx.background_spawn(async move {
if let Err(e) = write.await {
log::error!("Failed to persist processed key-sync requests: {e}");
}
})
.detach();
}
/// Withdraw the pending approval.
fn resolve_pending_request(&mut self, cx: &mut Context<Self>) {
let Some(pending) = self.pending_request.take() else {
return;
};
self.mark_request_processed(pending.event.id, cx);
if let Some(window) = cx.windows().first().copied() {
window
.update(cx, |_view, window, cx| {
window.remove_notification::<DeviceNotification>(cx);
})
.ok();
}
cx.notify();
}
/// Route a key transfer.
fn handle_transfer(&mut self, event: Event, cx: &mut Context<Self>) {
let resolved = self
.pending_request
.as_ref()
.is_some_and(|pending| transfer_addresses(&event, pending.client_key));
if resolved {
self.resolve_pending_request(cx);
return;
}
self.extract_encryption(event, cx);
}
/// Get the signer
pub fn signer(&self, cx: &App) -> Option<UniversalSigner> {
self.signer.read(cx).clone()
@@ -418,11 +506,9 @@ impl DeviceRegistry {
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let signer = nostr.read(cx).signer();
let app_keys_task = get_or_init_app_keys(cx);
let app_keys = self.app_keys.get_or_insert_with(Keys::generate).clone();
let task: Task<Result<Option<Event>, Error>> = cx.background_spawn(async move {
let app_keys = app_keys_task.await?;
let app_pubkey = app_keys.public_key();
let public_key = signer.get_public_key_async().await?;
@@ -434,8 +520,7 @@ impl DeviceRegistry {
.limit(1);
if let Some(event) = client.database().query(filter).await?.into_iter().next() {
// Only reuse the stored approval if its key is still valid,
// otherwise it's stale and a fresh request is needed
// Only reuse the stored approval if its key is still valid
if decrypt_approval(&client, &app_keys, public_key, &event)
.await
.is_ok()
@@ -483,19 +568,13 @@ impl DeviceRegistry {
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let signer = nostr.read(cx).signer();
let app_keys_task = get_or_init_app_keys(cx);
let Some(app_keys) = self.app_keys.clone() else {
log::warn!("Cannot wait for approval without a client key");
return;
};
self.tasks.push(cx.spawn(async move |this, cx| {
let app_keys = match app_keys_task.await {
Ok(app_keys) => app_keys,
Err(e) => {
this.update(cx, |_this, cx| {
cx.emit(DeviceEvent::error(e.to_string()));
})?;
return Ok(());
}
};
this.update(cx, |_this, cx| {
cx.emit(DeviceEvent::Requesting(authorization_code(
&app_keys.public_key(),
@@ -505,7 +584,7 @@ impl DeviceRegistry {
let task: Task<Result<(), Error>> = cx.background_spawn(async move {
let public_key = signer.get_public_key_async().await?;
// Only approvals addressed to this device's app key matter
// Only approvals addressed to this device's client key matter
let filter = Filter::new()
.kind(Kind::Custom(4455))
.author(public_key)
@@ -529,15 +608,18 @@ impl DeviceRegistry {
/// Parse the approval event to get encryption key then set it
fn extract_encryption(&mut self, event: Event, cx: &mut Context<Self>) {
let app_keys_task = get_or_init_app_keys(cx);
let nostr = NostrRegistry::global(cx);
let client = nostr.read(cx).client();
let signer = nostr.read(cx).signer();
let Some(app_keys) = self.app_keys.clone() else {
log::warn!("Ignoring key transfer addressed to another device");
return;
};
let approval_id = event.id;
let task: Task<Result<Keys, Error>> = cx.background_spawn(async move {
let app_keys = app_keys_task.await?;
let identity = signer.get_public_key_async().await?;
let keys = decrypt_approval(&client, &app_keys, identity, &event).await?;
@@ -549,14 +631,14 @@ impl DeviceRegistry {
Ok(keys) => {
this.update(cx, |this, cx| {
this.set_signer(keys, cx);
// The client key was consumed; start fresh on the next request
this.app_keys = None;
// Clean up our request and the approval event from relays
this.cleanup_sync_events(approval_id, cx);
})?;
}
Err(e) => {
this.update(cx, |_this, cx| {
cx.emit(DeviceEvent::error(e.to_string()));
})?;
log::warn!("Ignoring key transfer: {e}");
}
}
Ok(())
@@ -612,6 +694,7 @@ impl DeviceRegistry {
// Get user's write relays
let event = event.clone();
let request_id = event.id;
let task: Task<Result<(), Error>> = cx.background_spawn(async move {
// Get device keys
@@ -619,24 +702,21 @@ impl DeviceRegistry {
let secret = keys.secret_key().to_secret_hex();
// Extract the target public key from the event tags
let target = event
.tags
.iter()
.find(|tag| tag.kind() == "P")
.and_then(|tag| tag.content())
.and_then(|content| PublicKey::parse(content).ok())
.context("Target is not a valid public key")?;
let target =
client_pubkey_from_request(&event).context("Target is not a valid public key")?;
// Encrypt the device keys with the user's signer
let payload = keys.nip44_encrypt_async(&target, &secret).await?;
// Encrypt to a fresh one-time keypair so the transfer event
// can never be decrypted with the long-lived encryption key.
let sender = Keys::generate();
let payload = sender.nip44_encrypt_async(&target, &secret).await?;
// Construct the response event
//
// P tag: the current device's public key
// p tag: the requester's public key
// P tag: the one-time sender key used for the encryption
// p tag: the requester's client key
let event = EventBuilder::new(Kind::Custom(4455), payload)
.tags(vec![
Tag::custom("P", vec![keys.public_key().to_hex()]),
Tag::custom("P", vec![sender.public_key().to_hex()]),
Tag::public_key(target),
])
.finalize_async(&signer)
@@ -652,8 +732,11 @@ impl DeviceRegistry {
match task.await {
Ok(_) => {
this.update(cx, |this, cx| {
// Record the request as handled so it cannot resurface
this.mark_request_processed(request_id, cx);
// Allow future requests to surface again
this.set_pending_request(false, cx);
this.pending_request = None;
cx.notify();
})?;
cx.update(|window, cx| {
@@ -678,21 +761,70 @@ impl DeviceRegistry {
/// Handle encryption request
fn ask_for_approval(&mut self, event: Event, cx: &mut Context<Self>) {
// Ignore if there is already a pending request
if self.pending_request {
let Some(client_key) = client_pubkey_from_request(&event) else {
log::warn!("Ignoring invalid encryption key request: no valid requester key");
return;
}
self.set_pending_request(true, cx);
};
let notification = match self.notification(event, cx) {
let read = credentials::read_processed_requests(cx);
cx.spawn(async move |this, cx| {
let processed = match read.await {
Ok(bytes) => bytes
.map(|bytes| parse_processed_requests(&bytes).contains(&event.id))
.unwrap_or(false),
Err(e) => {
log::error!("Failed to read processed key-sync requests: {e}");
false
}
};
this.update(cx, |this, cx| {
if processed || this.processed_requests.contains(&event.id) {
return;
}
match this.pending_request.take() {
// A retry signs a new request id but keeps the same client key
Some(mut pending) if pending.client_key == client_key => {
this.mark_request_processed(event.id, cx);
pending.event = event;
this.pending_request = Some(pending);
}
// A different client key is a remounted requester
other => {
if let Some(pending) = other {
this.mark_request_processed(pending.event.id, cx);
if let Some(window) = cx.windows().first().copied() {
window
.update(cx, |_view, window, cx| {
window.remove_notification::<DeviceNotification>(cx);
})
.ok();
}
}
this.show_approval(event, client_key, cx);
}
}
})
.ok();
})
.detach();
}
/// Build and surface the approval notification for a key request
fn show_approval(&mut self, event: Event, client_key: PublicKey, cx: &mut Context<Self>) {
let notification = match self.notification(event.clone(), cx) {
Ok(notification) => notification,
Err(e) => {
log::warn!("Ignoring invalid encryption key request: {e}");
self.set_pending_request(false, cx);
return;
}
};
self.pending_request = Some(PendingRequest { event, client_key });
cx.notify();
if let Some(window) = cx.windows().first().copied() {
if let Err(error) = window.update(cx, |_view, window, cx| {
window.push_notification(notification, cx);
@@ -715,13 +847,8 @@ impl DeviceRegistry {
compare the code with the one shown on the requesting \
device.";
let requester = event
.tags
.iter()
.find(|tag| tag.kind() == "P")
.and_then(|tag| tag.content())
.and_then(|content| PublicKey::parse(content).ok())
.context("Request has no valid requester key")?;
let requester =
client_pubkey_from_request(&event).context("Request has no valid requester key")?;
let code = authorization_code(&requester);
let directory = self.directory.clone();
@@ -831,6 +958,51 @@ impl DeviceRegistry {
struct DeviceNotification;
/// A key request awaiting user approval on this device
#[derive(Debug)]
struct PendingRequest {
event: Event,
client_key: PublicKey,
}
/// Keep the durable processed-request log bounded
const PROCESSED_REQUESTS_LIMIT: usize = 256;
/// Parse concatenated hex event ids
fn parse_processed_requests(bytes: &[u8]) -> VecDeque<EventId> {
let Ok(hex) = std::str::from_utf8(bytes) else {
return VecDeque::new();
};
hex.as_bytes()
.as_chunks::<64>()
.0
.iter()
.filter_map(|chunk| EventId::from_hex(std::str::from_utf8(chunk).ok()?).ok())
.collect()
}
/// Extract the requester client key from a key request's `P` tag
fn client_pubkey_from_request(event: &Event) -> Option<PublicKey> {
event
.tags
.iter()
.find(|tag| tag.kind() == "P")
.and_then(|tag| tag.content())
.and_then(|content| PublicKey::parse(content).ok())
}
/// Whether the transfer event is p-tagged to the given client key
fn transfer_addresses(event: &Event, client_key: PublicKey) -> bool {
event.tags.iter().any(|tag| {
tag.kind() == "p"
&& tag
.content()
.and_then(|content| PublicKey::parse(content).ok())
.is_some_and(|key| key == client_key)
})
}
/// The first eight hexadecimal characters of the client public key.
fn authorization_code(public_key: &PublicKey) -> SharedString {
let hex = public_key.to_hex();
@@ -838,35 +1010,6 @@ fn authorization_code(public_key: &PublicKey) -> SharedString {
SharedString::from(format!("{} {}", &code[..4], &code[4..]))
}
/// Get or create new app keys (async, returns a task)
fn get_or_init_app_keys(cx: &App) -> Task<Result<Keys, Error>> {
let read = credentials::read_app_keys(cx);
cx.spawn(async move |cx| {
if let Ok(Some(secret)) = read.await
&& let Ok(keys) = SecretKey::from_slice(&secret).map(Keys::new)
{
return Ok(keys);
}
// No stored keys found or invalid — generate new ones
let keys = Keys::generate();
let secret = keys.secret_key().to_secret_bytes();
cx.update(|cx| {
let write = credentials::write_app_keys(cx, &secret);
cx.background_spawn(async move {
if let Err(e) = write.await {
log::error!("Failed to persist app keys: {e}")
}
})
.detach();
});
Ok(keys)
})
}
/// Decrypt and validate an approval event against the current announcement.
async fn decrypt_approval(
client: &Client,
@@ -889,24 +1032,51 @@ async fn decrypt_approval(
let secret = SecretKey::from_hex(&decrypted)?;
let keys = Keys::new(secret);
// Reject approvals carrying a key that no longer matches the published announcement.
// Reject approvals carrying a key that no longer matches the published announcement
let filter = Filter::new()
.kind(Kind::Custom(10044))
.author(identity)
.limit(1);
if let Some(announcement) = client.database().query(filter).await?.into_iter().next() {
let announced = announcement
.tags
.iter()
.find(|tag| tag.kind() == "n")
.and_then(|tag| tag.content())
.and_then(|content| PublicKey::parse(content).ok())
.context("Announcement has no valid encryption key")?;
let announcement = client
.database()
.query(filter.clone())
.await?
.into_iter()
.next();
if keys.public_key() != announced {
return Err(anyhow!("Approval key doesn't match the announcement"));
let announcement = match announcement {
Some(announcement) => Some(announcement),
None => {
let mut stream = client
.stream_events(filter)
.timeout(Duration::from_secs(3))
.await?;
stream
.next()
.await
.map(|(_url, result)| result)
.transpose()?
}
};
let Some(announcement) = announcement else {
return Err(anyhow!(
"No encryption announcement available to validate the approval"
));
};
let announced = announcement
.tags
.iter()
.find(|tag| tag.kind() == "n")
.and_then(|tag| tag.content())
.and_then(|content| PublicKey::parse(content).ok())
.context("Announcement has no valid encryption key")?;
if keys.public_key() != announced {
return Err(anyhow!("Approval key doesn't match the announcement"));
}
Ok(keys)