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

harden proxy auth and event ordering

Browse Source
This commit is contained in:
reya
2026-11-12 17:59:28 +07:00
parent fe0ca04817
commit fd802a8870
16 changed files with 819 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/nip60/src/lib.rs
+62 -7
View File
@@ -55,6 +55,9 @@ pub struct HistoryEntry {
pub created_at: Timestamp,
}
/// How many times a failed nutzap redemption is retried before giving up
const MAX_REDEEM_ATTEMPTS: u32 = 3;
pub struct WalletRegistry {
wallet: Option<keys::WalletData>,
tokens: HashMap<EventId, TokenEvent>,
@@ -69,6 +72,10 @@ pub struct WalletRegistry {
pending_nutzaps: Vec<Event>,
/// Nutzaps with a redemption job in flight
redeeming: HashSet<EventId>,
/// Redemption attempts per nutzap, reset on success
redeem_attempts: HashMap<EventId, u32>,
/// A job errored after possibly mutating the store
publish_failed: bool,
store: Option<Arc<WalletRedbDatabase>>,
wallets: HashMap<MintUrl, Wallet>,
loading: bool,
@@ -119,6 +126,8 @@ impl WalletRegistry {
redeemed: HashSet::new(),
pending_nutzaps: Vec::new(),
redeeming: HashSet::new(),
redeem_attempts: HashMap::new(),
publish_failed: false,
store: None,
wallets: HashMap::new(),
loading: false,
@@ -378,6 +387,8 @@ impl WalletRegistry {
self.redeemed.clear();
self.pending_nutzaps.clear();
self.redeeming.clear();
self.redeem_attempts.clear();
self.publish_failed = false;
self.store = None;
self.wallets.clear();
self.loading = false;
@@ -397,6 +408,7 @@ impl WalletRegistry {
let signer = nostr.read(cx).signer().clone();
let (tx, rx) = flume::bounded::<Event>(100);
let consumer_signer = signer.clone();
self.notification_listener = Some(cx.background_spawn(async move {
const MAX_PROCESSED: usize = 1000;
@@ -417,7 +429,7 @@ impl WalletRegistry {
| Kind::CashuWalletSpendingHistory
);
let wallet_event = ours && is_cashu || is_token_deletion(&event);
let wallet_event = ours && (is_cashu || is_token_deletion(&event));
let nutzap = event.kind == Kind::Custom(9321)
&& me.is_some_and(|me| event.tags.public_keys().any(|key| key == me));
@@ -443,21 +455,22 @@ impl WalletRegistry {
}));
self.tasks.push(cx.spawn(async move |this, cx| {
let me = consumer_signer.get_public_key_async().await.ok();
while let Ok(event) = rx.recv_async().await {
this.update(cx, |this, cx| {
this.absorb(event, cx);
this.absorb(event, me, cx);
})
.ok();
}
}));
}
fn absorb(&mut self, event: Event, cx: &mut Context<Self>) {
fn absorb(&mut self, event: Event, me: Option<PublicKey>, cx: &mut Context<Self>) {
match event.kind {
Kind::CashuWallet => self.absorb_wallet(event, cx),
Kind::CashuWalletUnspentProof => self.absorb_token(event, cx),
Kind::CashuWalletSpendingHistory => self.absorb_history(event, cx),
Kind::EventDeletion => self.absorb_deletion(event, cx),
Kind::EventDeletion => self.absorb_deletion(event, me, cx),
Kind::Custom(9321) => self.absorb_nutzap(event, cx),
_ => {}
}
@@ -549,7 +562,15 @@ impl WalletRegistry {
}));
}
fn absorb_deletion(&mut self, event: Event, cx: &mut Context<Self>) {
fn absorb_deletion(&mut self, event: Event, me: Option<PublicKey>, cx: &mut Context<Self>) {
if me != Some(event.pubkey) {
log::warn!(
"ignoring kind 5 deletion {} from a foreign author",
event.id
);
return;
}
let mut removed_any = false;
for tag in event.tags.iter() {
if tag.kind() != "e" {
@@ -807,6 +828,9 @@ impl WalletRegistry {
self.job_running = true;
let job_handle = job.clone();
let publish_failed = self.publish_failed;
self.tasks.push(cx.spawn(async move |this, cx| {
let result = Tokio::spawn_result(cx, async move {
let ctx = ops::JobCtx {
@@ -815,6 +839,7 @@ impl WalletRegistry {
signer,
client,
tokens,
publish_failed,
};
job.run(ctx).await
@@ -823,7 +848,7 @@ impl WalletRegistry {
this.update(cx, |this, cx| {
this.job_running = false;
this.apply_outcome(result, cx);
this.apply_outcome(job_handle, result, cx);
for job in std::mem::take(&mut this.job_queued) {
this.schedule_job(job, cx);
@@ -833,10 +858,38 @@ impl WalletRegistry {
}));
}
fn apply_outcome(&mut self, result: Result<ops::Outcome, Error>, cx: &mut Context<Self>) {
fn apply_outcome(
&mut self,
job: ops::Job,
result: Result<ops::Outcome, Error>,
cx: &mut Context<Self>,
) {
let outcome = match result {
Ok(outcome) => outcome,
Err(error) => {
if !matches!(job, ops::Job::CreateQuote { .. }) {
self.publish_failed = true;
}
if !matches!(job, ops::Job::CreateQuote { .. } | ops::Job::Reconcile) {
self.schedule_job(ops::Job::Reconcile, cx);
}
if let ops::Job::RedeemNutzap { nutzap_id, .. } = &job {
self.redeeming.remove(nutzap_id);
let attempts = self.redeem_attempts.entry(*nutzap_id).or_default();
*attempts += 1;
if *attempts <= MAX_REDEEM_ATTEMPTS {
self.job_queued.push(job.clone());
} else {
log::warn!(
"giving up redeeming nutzap {nutzap_id} after {MAX_REDEEM_ATTEMPTS} attempts"
);
}
}
self.emit_error(format!("{error:#}"), cx);
return;
}
@@ -845,6 +898,7 @@ impl WalletRegistry {
match outcome {
ops::Outcome::Idle => {}
ops::Outcome::Reconciled { updates } => {
self.publish_failed = false;
let changed = !updates.is_empty();
for update in updates {
@@ -905,6 +959,7 @@ impl WalletRegistry {
amount,
} => {
self.redeeming.remove(&nutzap_id);
self.redeem_attempts.remove(&nutzap_id);
self.redeemed.insert(nutzap_id);
self.apply_token_update(update);
self.emit_updated(cx);
crates/nip60/src/ops.rs
+128 -38
View File
@@ -361,6 +361,8 @@ pub(crate) struct JobCtx<S> {
pub(crate) client: Client,
/// Relay token events at schedule time.
pub(crate) tokens: HashMap<EventId, TokenEvent>,
/// A previous job errored after possibly mutating the store
pub(crate) publish_failed: bool,
}
impl<S> JobCtx<S>
@@ -426,35 +428,14 @@ where
log::warn!("saga recovery failed for {mint_url}: {error:#}");
}
let minted = wallet.mint_unissued_quotes().await?;
if minted == Amount::ZERO {
continue;
}
let local_ys = proof_ys(wallet.get_unspent_proofs().await?.iter())?;
let states = vec![State::Pending, State::Reserved, State::PendingSpent];
let update = self
.sync_token_events(
wallet,
nostr_mint(mint_url)?,
&previous_ids(&tokens, mint_url),
Some(keys::History {
direction: TransactionDirection::In,
amount: u64::from(minted),
redeemed: Vec::new(),
sender: None,
}),
)
.await?;
let in_flight = wallet.get_proofs_by_states(states).await?;
let in_flight_ys = proof_ys(in_flight.iter())?;
for id in &update.destroyed {
tokens.remove(id);
}
tokens.insert(update.event_id, update.token.clone());
updates.push(update);
}
for (mint_url, wallet) in &self.wallets {
let mut relay_ys = HashSet::new();
let mut added = Vec::new();
let mut unseen = Vec::new();
for token in tokens
.values()
@@ -468,23 +449,104 @@ where
continue;
}
};
if relay_ys.insert(info.y) {
added.push(info);
if !relay_ys.insert(info.y) {
continue;
}
if local_ys.contains(&info.y) || in_flight_ys.contains(&info.y) {
continue;
}
unseen.push(info);
}
}
let unspent = wallet.get_unspent_proofs().await?;
let removed = unspent
.iter()
.filter(|proof| match proof.y() {
Ok(y) => !relay_ys.contains(&y),
Err(_) => false,
})
.map(|proof| proof.y())
.collect::<Result<Vec<_>, _>>()?;
if unseen.is_empty() {
continue;
}
wallet.localstore.update_proofs(added, removed).await?;
let unseen_proofs: Proofs = unseen.iter().map(|info| info.proof.clone()).collect();
let spent = match wallet.check_proofs_spent(unseen_proofs).await {
Ok(states) => states
.into_iter()
.filter(|state| state.state == State::Spent)
.map(|state| state.y)
.collect::<HashSet<_>>(),
Err(error) => {
log::warn!("mint state check failed, accepting unseen proofs: {error:#}");
HashSet::new()
}
};
let added: Vec<ProofInfo> = unseen
.into_iter()
.filter(|info| !spent.contains(&info.y))
.collect();
if !added.is_empty() {
wallet.localstore.update_proofs(added, Vec::new()).await?;
}
}
for (mint_url, wallet) in &self.wallets {
let minted = wallet.mint_unissued_quotes().await?;
if minted != Amount::ZERO {
let update = self
.sync_token_events(
wallet,
nostr_mint(mint_url)?,
&previous_ids(&tokens, mint_url),
Some(keys::History {
direction: TransactionDirection::In,
amount: u64::from(minted),
redeemed: Vec::new(),
sender: None,
}),
)
.await?;
for id in &update.destroyed {
tokens.remove(id);
}
tokens.insert(update.event_id, update.token.clone());
updates.push(update);
}
let relay_ys = relay_proof_ys(&tokens, mint_url);
let local_ys = proof_ys(wallet.get_unspent_proofs().await?.iter())?;
if local_ys != relay_ys {
if self.publish_failed {
let update = self
.sync_token_events(
wallet,
nostr_mint(mint_url)?,
&previous_ids(&tokens, mint_url),
None,
)
.await?;
for id in &update.destroyed {
tokens.remove(id);
}
tokens.insert(update.event_id, update.token.clone());
updates.push(update);
} else {
let removed = wallet
.get_unspent_proofs()
.await?
.into_iter()
.filter(|proof| match proof.y() {
Ok(y) => !relay_ys.contains(&y),
Err(_) => false,
})
.map(|proof| proof.y())
.collect::<Result<Vec<_>, _>>()?;
wallet.localstore.update_proofs(Vec::new(), removed).await?;
}
}
}
Ok(Outcome::Reconciled { updates })
@@ -499,6 +561,34 @@ fn previous_ids(tokens: &HashMap<EventId, TokenEvent>, mint: &MintUrl) -> Vec<Ev
.collect()
}
fn proof_ys<'a>(
proofs: impl Iterator<Item = &'a Proof>,
) -> Result<HashSet<cdk::nuts::PublicKey>, Error> {
proofs.map(|proof| proof.y().map_err(Error::from)).collect()
}
/// Proof ys referenced by the relay token events of one mint
fn relay_proof_ys(
tokens: &HashMap<EventId, TokenEvent>,
mint: &MintUrl,
) -> HashSet<cdk::nuts::PublicKey> {
let mut ys = HashSet::new();
for token in tokens
.values()
.filter(|token| token.mint.as_str() == mint.to_string())
{
for proof in &token.proofs {
match proof_info(proof, mint) {
Ok(info) => {
ys.insert(info.y);
}
Err(error) => log::warn!("skipping proof: {error:#}"),
}
}
}
ys
}
/// Resolve a recipient's 10019 wallet info and pick a mint we share with them.
async fn recipient_wallet_info<S>(
ctx: &JobCtx<S>,