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

nip4e: drop processed-requests persistence

Browse Source
This commit is contained in:
reya
2026-11-12 14:39:28 +07:00
parent 8acbbaef5f
commit 8fa79aef33
2 changed files with 26 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/nip4e/src/lib.rs
+25 -134
View File
@@ -1,5 +1,5 @@
use std::cell::Cell; use std::cell::Cell;
use std::collections::{HashSet, VecDeque}; use std::collections::HashSet;
use std::path::PathBuf; use std::path::PathBuf;
use std::rc::Rc; use std::rc::Rc;
use std::sync::Arc; use std::sync::Arc;
@@ -19,14 +19,12 @@ use nostr_sdk::prelude::*;
use person::Directory; use person::Directory;
use settings::{AppSettings, NIP4E_ROTATION_MAX_DAYS, NIP4E_ROTATION_MIN_DAYS}; use settings::{AppSettings, NIP4E_ROTATION_MAX_DAYS, NIP4E_ROTATION_MIN_DAYS};
use smallvec::{SmallVec, smallvec}; use smallvec::{SmallVec, smallvec};
use state::{Announcement, NostrRegistry, UniversalSigner, credentials}; use state::{Announcement, NostrRegistry, UniversalSigner};
use storage::{SECONDS_PER_DAY, archive_key, get_keys, load_device_keys, set_keys}; use storage::{SECONDS_PER_DAY, archive_key, get_keys, load_device_keys, set_keys};
use ui::Avatar; use ui::Avatar;
mod storage; mod storage;
/// Keep the durable processed-request log bounded
const PROCESSED_REQUESTS_LIMIT: usize = 256;
/// Ignore key requests older than a day so stale events on relays cannot resurface /// Ignore key requests older than a day so stale events on relays cannot resurface
const REQUEST_MAX_AGE: u64 = SECONDS_PER_DAY; const REQUEST_MAX_AGE: u64 = SECONDS_PER_DAY;
@@ -79,8 +77,6 @@ pub struct DeviceRegistry {
app_keys: Option<Keys>, app_keys: Option<Keys>,
/// The key request currently awaiting user approval on this device /// The key request currently awaiting user approval on this device
pending_request: Option<PendingRequest>, 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 /// Whether an announcement has been made for this device
pub(crate) announcement_existed: Arc<AtomicBool>, pub(crate) announcement_existed: Arc<AtomicBool>,
/// Whether the local key is stale against the current announcement. /// Whether the local key is stale against the current announcement.
@@ -110,7 +106,6 @@ impl DeviceRegistry {
/// Create a new device registry instance /// Create a new device registry instance
fn new(cx: &mut Context<Self>) -> Self { fn new(cx: &mut Context<Self>) -> Self {
let entity = cx.entity().downgrade();
let nostr = NostrRegistry::global(cx); let nostr = NostrRegistry::global(cx);
let settings = AppSettings::global(cx); let settings = AppSettings::global(cx);
@@ -153,17 +148,10 @@ impl DeviceRegistry {
}), }),
); );
cx.defer(move |cx| {
let _ = entity.update(cx, |this, cx| {
this.handle_notifications(cx);
});
});
Self { Self {
signer, signer,
app_keys: None, app_keys: None,
pending_request: None, pending_request: None,
processed_requests: VecDeque::new(),
announcement_existed: Arc::new(AtomicBool::new(false)), announcement_existed: Arc::new(AtomicBool::new(false)),
resync: Arc::new(AtomicBool::new(false)), resync: Arc::new(AtomicBool::new(false)),
rotating: false, rotating: false,
@@ -184,8 +172,6 @@ impl DeviceRegistry {
let announcement_existed = self.announcement_existed.clone(); let announcement_existed = self.announcement_existed.clone();
let (tx, rx) = flume::bounded::<Event>(100); let (tx, rx) = flume::bounded::<Event>(100);
let read = credentials::read_processed_requests(cx);
self.notification_listener = Some(cx.background_spawn(async move { self.notification_listener = Some(cx.background_spawn(async move {
let mut notifications = client.notifications(); let mut notifications = client.notifications();
let mut processed_events = HashSet::new(); let mut processed_events = HashSet::new();
@@ -220,21 +206,6 @@ impl DeviceRegistry {
})); }));
self.tasks.push(cx.spawn(async move |this, cx| { 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 { while let Ok(event) = rx.recv_async().await {
match event.kind { match event.kind {
Kind::Custom(10044) => { Kind::Custom(10044) => {
@@ -261,36 +232,11 @@ impl DeviceRegistry {
})); }));
} }
/// 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. /// Withdraw the pending approval.
fn resolve_pending_request(&mut self, cx: &mut Context<Self>) { fn resolve_pending_request(&mut self, cx: &mut Context<Self>) {
let Some(pending) = self.pending_request.take() else { if self.pending_request.take().is_none() {
return; return;
}; }
self.mark_request_processed(pending.event.id, cx);
if let Some(window) = cx.windows().first().copied() { if let Some(window) = cx.windows().first().copied() {
window window
@@ -504,6 +450,7 @@ impl DeviceRegistry {
let elapsed = Timestamp::now() let elapsed = Timestamp::now()
.as_secs() .as_secs()
.saturating_sub(keys.created_at.as_secs()); .saturating_sub(keys.created_at.as_secs());
if elapsed < u64::from(interval_days) * SECONDS_PER_DAY { if elapsed < u64::from(interval_days) * SECONDS_PER_DAY {
return Ok(()); return Ok(());
} }
@@ -530,7 +477,6 @@ impl DeviceRegistry {
let secret = keys.secret_key().to_secret_hex(); let secret = keys.secret_key().to_secret_hex();
let n = keys.public_key(); let n = keys.public_key();
let created_at = Timestamp::now();
cx.background_spawn(async move { cx.background_spawn(async move {
// Construct an announcement event // Construct an announcement event
@@ -540,7 +486,7 @@ impl DeviceRegistry {
.await?; .await?;
// Persist the key locally before announcing it // Persist the key locally before announcing it
set_keys(&client, &signer, &secret, created_at).await?; set_keys(&client, &signer, &secret, Timestamp::now()).await?;
// Publish announcement // Publish announcement
client client
@@ -831,7 +777,6 @@ impl DeviceRegistry {
let signer = nostr.read(cx).signer(); let signer = nostr.read(cx).signer();
let event = event.clone(); let event = event.clone();
let request_id = event.id;
let task: Task<Result<(), Error>> = cx.background_spawn(async move { let task: Task<Result<(), Error>> = cx.background_spawn(async move {
// Get device keys // Get device keys
@@ -890,9 +835,6 @@ impl DeviceRegistry {
match task.await { match task.await {
Ok(_) => { Ok(_) => {
this.update(cx, |this, cx| { 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.pending_request = None; this.pending_request = None;
cx.notify(); cx.notify();
})?; })?;
@@ -933,51 +875,27 @@ impl DeviceRegistry {
return; return;
}; };
let read = credentials::read_processed_requests(cx); match self.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 => {
pending.event = event;
self.pending_request = Some(pending);
}
cx.spawn(async move |this, cx| { // A different client key is a remounted requester
let processed = match read.await { other => {
Ok(bytes) => bytes if other.is_some()
.map(|bytes| parse_processed_requests(&bytes).contains(&event.id)) && let Some(window) = cx.windows().first().copied()
.unwrap_or(false), {
Err(e) => { window
log::error!("Failed to read processed key-sync requests: {e}"); .update(cx, |_view, window, cx| {
false window.remove_notification::<DeviceNotification>(cx);
})
.ok();
} }
}; self.show_approval(event, client_key, cx);
}
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 /// Build and surface the approval notification for a key request
@@ -1103,19 +1021,6 @@ impl DeviceRegistry {
) )
.into_any_element() .into_any_element()
}) })
.on_close({
let entity = entity.clone();
let request_id = event.id;
move |_window, cx| {
// Closing without approving rejects the request, so it must not resurface
entity
.update(cx, |this, cx| {
this.mark_request_processed(request_id, cx);
})
.ok();
}
})
.action(move |_this, _window, _cx| { .action(move |_this, _window, _cx| {
let view = entity.clone(); let view = entity.clone();
let event = event.clone(); let event = event.clone();
@@ -1140,20 +1045,6 @@ impl DeviceRegistry {
} }
} }
/// 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 client public key from an event. /// Extract the client public key from an event.
fn client_pubkey_from_event(event: &Event) -> Option<PublicKey> { fn client_pubkey_from_event(event: &Event) -> Option<PublicKey> {
let key = event let key = event
crates/state/src/credentials.rs
+1 -55
View File
@@ -1,13 +1,10 @@
use anyhow::Result; use anyhow::Result;
#[cfg(target_arch = "wasm32")] #[cfg(target_arch = "wasm32")]
use common::web_store; use common::web_store;
use gpui::{App, AppContext as _, Task}; use gpui::{App, Task};
use crate::USER_KEYRING; use crate::USER_KEYRING;
/// The storage key for the log of handled NIP-4e key-sync request ids
const PROCESSED_REQUESTS_KEY: &str = "coop/nip4e/processed-requests";
/// Read the stored credential as `(username, secret)` /// Read the stored credential as `(username, secret)`
#[allow(clippy::type_complexity)] #[allow(clippy::type_complexity)]
pub(crate) fn read(cx: &App) -> Task<Result<Option<(String, Vec<u8>)>>> { pub(crate) fn read(cx: &App) -> Task<Result<Option<(String, Vec<u8>)>>> {
@@ -52,54 +49,3 @@ pub(crate) fn delete(cx: &App) -> Task<Result<()>> {
Task::ready(web_store::remove(USER_KEYRING)) Task::ready(web_store::remove(USER_KEYRING))
} }
} }
/// Read named bytes from the platform keyring,
/// falling back to browser storage on web where no keyring exists
fn read_named(cx: &App, key: &str) -> Task<Result<Option<Vec<u8>>>> {
#[cfg(not(target_arch = "wasm32"))]
{
let key = key.to_owned();
let read = cx.read_credentials(&key);
cx.background_spawn(async move { Ok(read.await?.map(|(_, secret)| secret)) })
}
#[cfg(target_arch = "wasm32")]
{
let _ = cx;
let stored = web_store::get(key).map(|content| content.map(|s| s.into_bytes()));
Task::ready(stored)
}
}
/// Write named bytes to the platform keyring,
/// falling back to browser storage on web where no keyring exists
fn write_named(cx: &App, key: &str, secret: &[u8]) -> Task<Result<()>> {
#[cfg(not(target_arch = "wasm32"))]
{
let key = key.to_owned();
let secret = secret.to_vec();
let write = cx.write_credentials(&key, "", &secret);
cx.background_spawn(write)
}
#[cfg(target_arch = "wasm32")]
{
let _ = cx;
let content = match std::str::from_utf8(secret) {
Ok(content) => content,
Err(e) => return Task::ready(Err(anyhow::anyhow!("secret is not UTF-8: {e}"))),
};
Task::ready(web_store::set(key, content))
}
}
/// Read the log of handled key-sync request ids, as concatenated hex ids
pub fn read_processed_requests(cx: &App) -> Task<Result<Option<Vec<u8>>>> {
read_named(cx, PROCESSED_REQUESTS_KEY)
}
/// Persist the log of handled key-sync request ids, as concatenated hex ids
pub fn write_processed_requests(cx: &App, ids: &[u8]) -> Task<Result<()>> {
write_named(cx, PROCESSED_REQUESTS_KEY, ids)
}