upleb.uk

Public git repos — served from a NIP-34 GRASP relay at git.upleb.uk

summaryrefslogtreecommitdiff
path: root/src/sync
diff options
context:
space:
mode:
authorDanConwayDev <DanConwayDev@protonmail.com>2026-01-09 17:04:06 +0000
committerDanConwayDev <DanConwayDev@protonmail.com>2026-01-09 17:04:06 +0000
commit5ecd8d6a434f97da94daef2f59166086fbaf5a6b (patch)
tree54c7d3b953a6b1aedd1db6b9a719e18131659df5 /src/sync
parent895359aeb6746b98ff82944e4fca503f4a6e5439 (diff)
feat: implement state event authorization per GRASP-01 spec
Add comprehensive authorization checks to ensure state events are only accepted from maintainers of accepted repository announcements. This implements the core GRASP-01 requirement that pushes must match the latest state announcement "respecting the maintainer set." Changes: 1. StatePolicy authorization (src/nostr/policy/state.rs): - Check authorization BEFORE git data validation (fail-fast) - Reject if no announcement exists for repository - Reject if author not in maintainer set - Use existing helpers: fetch_repository_data() and pubkey_authorised_for_repo_owners() - Structured logging for all rejections 2. Purgatory invalidation (src/nostr/builder.rs): - New method: check_purgatory_state_events_for_identifier() - Called when announcements accepted (Accept and AcceptMaintainer) - Re-evaluates state events in purgatory for the identifier - Processes newly-authorized events (releases from purgatory) - Keeps unauthorized events for natural expiry (30 min) - Enables retroactive authorization when announcements arrive late 3. Purgatory sync authorization (src/git/sync.rs): - Check authorization BEFORE processing git data - Remove unauthorized events from purgatory (permanent rejection) - Prevents processing even if git data arrives first - Structured logging for monitoring 4. Rejected events tracking (src/sync/rejected_index.rs): - Add support for tracking rejected state events - New methods: add_state(), contains_state() - Separate metrics for state rejections - Enables sync to avoid re-fetching rejected states 5. Sync metrics (src/sync/metrics.rs, src/sync/mod.rs): - Add state-specific metrics (hot cache, cold index) - Track rejected states separately from announcements - Support monitoring of authorization rejections 6. Comprehensive tests (tests/state_authorization.rs): - test_reject_state_without_announcement - test_reject_state_from_unauthorized_author - test_accept_state_from_announcement_author - test_accept_state_from_maintainer Security Impact: - Before: State events could be published by anyone - After: Only maintainers can publish state events - Defense-in-depth: Authorization checked at 3 points: 1. On arrival (StatePolicy) 2. On announcement acceptance (purgatory re-evaluation) 3. On git data arrival (purgatory sync) All tests pass: - 248 unit tests - 51 NIP-34 announcement tests - 4 new state authorization tests - 9 rejected index tests Closes: State authorization requirement from GRASP-01 spec
Diffstat (limited to 'src/sync')
-rw-r--r--src/sync/metrics.rs103
-rw-r--r--src/sync/mod.rs40
-rw-r--r--src/sync/rejected_index.rs109
3 files changed, 249 insertions, 3 deletions
diff --git a/src/sync/metrics.rs b/src/sync/metrics.rs
index a175210..7d6d42d 100644
--- a/src/sync/metrics.rs
+++ b/src/sync/metrics.rs
@@ -56,6 +56,22 @@ pub struct SyncMetrics {
56 rejected_announcements_cold_index_expired_total: IntCounter, 56 rejected_announcements_cold_index_expired_total: IntCounter,
57 /// Total invalidations (maintainer announcements invalidated) 57 /// Total invalidations (maintainer announcements invalidated)
58 rejected_announcements_invalidated_total: IntCounter, 58 rejected_announcements_invalidated_total: IntCounter,
59
60 // === Rejected States Index Metrics ===
61 /// Current number of state events in hot cache
62 rejected_states_hot_cache_current: IntGauge,
63 /// Total hot cache hits (state events re-processed from cache)
64 rejected_states_hot_cache_hits_total: IntCounter,
65 /// Total hot cache misses (state events not in cache)
66 rejected_states_hot_cache_misses_total: IntCounter,
67 /// Total expired state events removed from hot cache
68 rejected_states_hot_cache_expired_total: IntCounter,
69 /// Current number of state event entries in cold index
70 rejected_states_cold_index_current: IntGauge,
71 /// Total state event cold index entries expired and removed
72 rejected_states_cold_index_expired_total: IntCounter,
73 /// Total state event invalidations
74 rejected_states_invalidated_total: IntCounter,
59} 75}
60 76
61impl SyncMetrics { 77impl SyncMetrics {
@@ -172,6 +188,49 @@ impl SyncMetrics {
172 ))?; 188 ))?;
173 registry.register(Box::new(rejected_announcements_invalidated_total.clone()))?; 189 registry.register(Box::new(rejected_announcements_invalidated_total.clone()))?;
174 190
191 // Rejected states metrics
192 let rejected_states_hot_cache_current = IntGauge::with_opts(Opts::new(
193 "ngit_sync_rejected_states_hot_cache_current",
194 "Current number of state events in hot cache (full events, 2 min expiry)",
195 ))?;
196 registry.register(Box::new(rejected_states_hot_cache_current.clone()))?;
197
198 let rejected_states_hot_cache_hits_total = IntCounter::with_opts(Opts::new(
199 "ngit_sync_rejected_states_hot_cache_hits_total",
200 "Total hot cache hits (state events re-processed from cache)",
201 ))?;
202 registry.register(Box::new(rejected_states_hot_cache_hits_total.clone()))?;
203
204 let rejected_states_hot_cache_misses_total = IntCounter::with_opts(Opts::new(
205 "ngit_sync_rejected_states_hot_cache_misses_total",
206 "Total hot cache misses (state events not in cache when invalidated)",
207 ))?;
208 registry.register(Box::new(rejected_states_hot_cache_misses_total.clone()))?;
209
210 let rejected_states_hot_cache_expired_total = IntCounter::with_opts(Opts::new(
211 "ngit_sync_rejected_states_hot_cache_expired_total",
212 "Total expired state events removed from hot cache",
213 ))?;
214 registry.register(Box::new(rejected_states_hot_cache_expired_total.clone()))?;
215
216 let rejected_states_cold_index_current = IntGauge::with_opts(Opts::new(
217 "ngit_sync_rejected_states_cold_index_current",
218 "Current number of state event entries in cold index (metadata only, 7 day expiry)",
219 ))?;
220 registry.register(Box::new(rejected_states_cold_index_current.clone()))?;
221
222 let rejected_states_cold_index_expired_total = IntCounter::with_opts(Opts::new(
223 "ngit_sync_rejected_states_cold_index_expired_total",
224 "Total state event cold index entries expired and removed",
225 ))?;
226 registry.register(Box::new(rejected_states_cold_index_expired_total.clone()))?;
227
228 let rejected_states_invalidated_total = IntCounter::with_opts(Opts::new(
229 "ngit_sync_rejected_states_invalidated_total",
230 "Total state event invalidations (when announcements accepted)",
231 ))?;
232 registry.register(Box::new(rejected_states_invalidated_total.clone()))?;
233
175 Ok(Self { 234 Ok(Self {
176 relay_connected, 235 relay_connected,
177 connection_attempts_total, 236 connection_attempts_total,
@@ -188,6 +247,13 @@ impl SyncMetrics {
188 rejected_announcements_cold_index_current, 247 rejected_announcements_cold_index_current,
189 rejected_announcements_cold_index_expired_total, 248 rejected_announcements_cold_index_expired_total,
190 rejected_announcements_invalidated_total, 249 rejected_announcements_invalidated_total,
250 rejected_states_hot_cache_current,
251 rejected_states_hot_cache_hits_total,
252 rejected_states_hot_cache_misses_total,
253 rejected_states_hot_cache_expired_total,
254 rejected_states_cold_index_current,
255 rejected_states_cold_index_expired_total,
256 rejected_states_invalidated_total,
191 }) 257 })
192 } 258 }
193 259
@@ -396,6 +462,43 @@ impl SyncMetrics {
396 pub fn record_invalidation(&self, count: usize) { 462 pub fn record_invalidation(&self, count: usize) {
397 self.rejected_announcements_invalidated_total.inc_by(count as u64); 463 self.rejected_announcements_invalidated_total.inc_by(count as u64);
398 } 464 }
465
466 // === Rejected States Recording Methods ===
467
468 /// Update state events hot cache current size gauge.
469 pub fn update_states_hot_cache_size(&self, size: usize) {
470 self.rejected_states_hot_cache_current.set(size as i64);
471 }
472
473 /// Record state event hot cache hit (event re-processed from cache).
474 pub fn record_states_hot_cache_hit(&self) {
475 self.rejected_states_hot_cache_hits_total.inc();
476 }
477
478 /// Record state event hot cache miss (event not in cache when invalidated).
479 pub fn record_states_hot_cache_miss(&self) {
480 self.rejected_states_hot_cache_misses_total.inc();
481 }
482
483 /// Record state event hot cache expired entries.
484 pub fn record_states_hot_cache_expired(&self, count: usize) {
485 self.rejected_states_hot_cache_expired_total.inc_by(count as u64);
486 }
487
488 /// Update state events cold index current size gauge.
489 pub fn update_states_cold_index_size(&self, size: usize) {
490 self.rejected_states_cold_index_current.set(size as i64);
491 }
492
493 /// Record state event cold index expired entries.
494 pub fn record_states_cold_index_expired(&self, count: usize) {
495 self.rejected_states_cold_index_expired_total.inc_by(count as u64);
496 }
497
498 /// Record state event invalidation.
499 pub fn record_states_invalidation(&self, count: usize) {
500 self.rejected_states_invalidated_total.inc_by(count as u64);
501 }
399} 502}
400 503
401#[cfg(test)] 504#[cfg(test)]
diff --git a/src/sync/mod.rs b/src/sync/mod.rs
index f296c0f..93b0e38 100644
--- a/src/sync/mod.rs
+++ b/src/sync/mod.rs
@@ -367,12 +367,14 @@ async fn run_daily_timer(
367/// Run the combined health and metrics checker 367/// Run the combined health and metrics checker
368/// 368///
369/// This function runs in a loop with a 2-second interval, performing three tasks: 369/// This function runs in a loop with a 2-second interval, performing three tasks:
370/// Background task for cleaning up expired entries from the rejected events index 370/// Background task for cleaning up expired entries from the rejected events indexes
371/// 371///
372/// This task runs two cleanup operations at different intervals: 372/// This task runs two cleanup operations at different intervals:
373/// 1. **Hot cache cleanup (60s)**: Remove events older than 2 minutes from hot cache 373/// 1. **Hot cache cleanup (60s)**: Remove events older than 2 minutes from hot cache
374/// 2. **Cold index cleanup (daily)**: Remove metadata older than 7 days from cold index 374/// 2. **Cold index cleanup (daily)**: Remove metadata older than 7 days from cold index
375/// 375///
376/// Cleans up both the announcements index and the states index.
377///
376/// The hot cache cleanup runs frequently to keep memory usage low (events expire quickly). 378/// The hot cache cleanup runs frequently to keep memory usage low (events expire quickly).
377/// The cold index cleanup runs daily since metadata is small and expires slowly. 379/// The cold index cleanup runs daily since metadata is small and expires slowly.
378async fn run_rejected_index_cleanup( 380async fn run_rejected_index_cleanup(
@@ -397,6 +399,8 @@ async fn run_rejected_index_cleanup(
397 tokio::select! { 399 tokio::select! {
398 _ = hot_cache_timer.tick() => { 400 _ = hot_cache_timer.tick() => {
399 let manager = sync_manager.lock().await; 401 let manager = sync_manager.lock().await;
402
403 // Clean up announcements index
400 let (hot_expired, _) = manager.rejected_events_index.cleanup_expired(); 404 let (hot_expired, _) = manager.rejected_events_index.cleanup_expired();
401 if hot_expired > 0 { 405 if hot_expired > 0 {
402 tracing::debug!( 406 tracing::debug!(
@@ -404,9 +408,20 @@ async fn run_rejected_index_cleanup(
404 hot_expired 408 hot_expired
405 ); 409 );
406 } 410 }
411
412 // Clean up states index
413 let (states_hot_expired, _) = manager.rejected_states_index.cleanup_states_expired();
414 if states_hot_expired > 0 {
415 tracing::debug!(
416 "Cleaned up {} expired entries from rejected states hot cache",
417 states_hot_expired
418 );
419 }
407 } 420 }
408 _ = cold_index_timer.tick() => { 421 _ = cold_index_timer.tick() => {
409 let manager = sync_manager.lock().await; 422 let manager = sync_manager.lock().await;
423
424 // Clean up announcements index
410 let (_, cold_expired) = manager.rejected_events_index.cleanup_expired(); 425 let (_, cold_expired) = manager.rejected_events_index.cleanup_expired();
411 if cold_expired > 0 { 426 if cold_expired > 0 {
412 tracing::info!( 427 tracing::info!(
@@ -414,6 +429,15 @@ async fn run_rejected_index_cleanup(
414 cold_expired 429 cold_expired
415 ); 430 );
416 } 431 }
432
433 // Clean up states index
434 let (_, states_cold_expired) = manager.rejected_states_index.cleanup_states_expired();
435 if states_cold_expired > 0 {
436 tracing::info!(
437 "Cleaned up {} expired entries from rejected states cold index",
438 states_cold_expired
439 );
440 }
417 } 441 }
418 _ = shutdown_rx.recv() => { 442 _ = shutdown_rx.recv() => {
419 tracing::info!("Rejected index cleanup received shutdown signal"); 443 tracing::info!("Rejected index cleanup received shutdown signal");
@@ -507,6 +531,8 @@ pub struct SyncManager {
507 pending_sync_index: PendingSyncIndex, 531 pending_sync_index: PendingSyncIndex,
508 /// Rejected announcement events (30617/30618) - two-tier storage for re-processing 532 /// Rejected announcement events (30617/30618) - two-tier storage for re-processing
509 rejected_events_index: Arc<RejectedEventsIndex>, 533 rejected_events_index: Arc<RejectedEventsIndex>,
534 /// Rejected state events (30618) - two-tier storage for re-processing
535 rejected_states_index: Arc<RejectedEventsIndex>,
510 /// Active relay connections - keyed by relay URL 536 /// Active relay connections - keyed by relay URL
511 connections: HashMap<String, RelayConnection>, 537 connections: HashMap<String, RelayConnection>,
512 /// Health tracker for relay connection state 538 /// Health tracker for relay connection state
@@ -571,6 +597,18 @@ impl SyncManager {
571 Duration::from_secs(config.rejected_cold_index_expiry_secs), 597 Duration::from_secs(config.rejected_cold_index_expiry_secs),
572 ) 598 )
573 }), 599 }),
600 rejected_states_index: Arc::new(if let Some(ref metrics) = sync_metrics {
601 RejectedEventsIndex::with_metrics(
602 Duration::from_secs(config.rejected_hot_cache_duration_secs),
603 Duration::from_secs(config.rejected_cold_index_expiry_secs),
604 metrics.clone(),
605 )
606 } else {
607 RejectedEventsIndex::new(
608 Duration::from_secs(config.rejected_hot_cache_duration_secs),
609 Duration::from_secs(config.rejected_cold_index_expiry_secs),
610 )
611 }),
574 connections: HashMap::new(), 612 connections: HashMap::new(),
575 health_tracker: Arc::new(RelayHealthTracker::new(config)), 613 health_tracker: Arc::new(RelayHealthTracker::new(config)),
576 next_batch_id: 0, 614 next_batch_id: 0,
diff --git a/src/sync/rejected_index.rs b/src/sync/rejected_index.rs
index 4733d80..f5ffef4 100644
--- a/src/sync/rejected_index.rs
+++ b/src/sync/rejected_index.rs
@@ -355,7 +355,7 @@ impl RejectedEventsIndex {
355 index 355 index
356 } 356 }
357 357
358 /// Update metrics with current sizes 358 /// Update metrics with current sizes (for announcements)
359 fn update_metrics(&self) { 359 fn update_metrics(&self) {
360 if let Some(ref metrics) = self.metrics { 360 if let Some(ref metrics) = self.metrics {
361 metrics.update_hot_cache_size(self.hot_cache.len()); 361 metrics.update_hot_cache_size(self.hot_cache.len());
@@ -363,6 +363,14 @@ impl RejectedEventsIndex {
363 } 363 }
364 } 364 }
365 365
366 /// Update metrics with current sizes (for states)
367 fn update_states_metrics(&self) {
368 if let Some(ref metrics) = self.metrics {
369 metrics.update_states_hot_cache_size(self.hot_cache.len());
370 metrics.update_states_cold_index_size(self.cold_index.len());
371 }
372 }
373
366 /// Add rejected announcement to both tiers 374 /// Add rejected announcement to both tiers
367 /// 375 ///
368 /// # Arguments 376 /// # Arguments
@@ -393,6 +401,36 @@ impl RejectedEventsIndex {
393 self.update_metrics(); 401 self.update_metrics();
394 } 402 }
395 403
404 /// Add rejected state event to both tiers
405 ///
406 /// # Arguments
407 ///
408 /// * `event` - Full event object (stored in hot cache)
409 /// * `pubkey` - Author's public key
410 /// * `identifier` - Repository identifier (d tag)
411 /// * `reason` - Why the state event was rejected
412 pub fn add_state(
413 &self,
414 event: Event,
415 pubkey: PublicKey,
416 identifier: String,
417 reason: RejectionReason,
418 ) {
419 // Add to hot cache (full event)
420 self.hot_cache.add(
421 event.clone(),
422 pubkey,
423 identifier.clone(),
424 reason,
425 );
426
427 // Add to cold index (metadata only)
428 self.cold_index.add(event.id, pubkey, identifier, reason);
429
430 // Update metrics (using states metrics)
431 self.update_states_metrics();
432 }
433
396 /// Check if event is already rejected (in either tier) 434 /// Check if event is already rejected (in either tier)
397 pub fn contains(&self, event_id: &EventId) -> bool { 435 pub fn contains(&self, event_id: &EventId) -> bool {
398 self.hot_cache.contains(event_id) || self.cold_index.contains(event_id) 436 self.hot_cache.contains(event_id) || self.cold_index.contains(event_id)
@@ -442,7 +480,51 @@ impl RejectedEventsIndex {
442 (removed, events) 480 (removed, events)
443 } 481 }
444 482
445 /// Clean up expired entries from both tiers 483 /// Invalidate state events and get events for immediate re-processing
484 ///
485 /// This is called when an announcement is accepted that authorizes state events.
486 /// It removes the cold index entries (so they can be re-fetched on next sync) and
487 /// returns any events still in the hot cache for immediate re-processing.
488 ///
489 /// # Returns
490 ///
491 /// Tuple of (number of cold index entries removed, events from hot cache)
492 pub fn invalidate_and_get_state_events(
493 &self,
494 maintainer_pubkey: &PublicKey,
495 identifier: &str,
496 ) -> (usize, Vec<Event>) {
497 // Remove from cold index (prevents re-fetch)
498 let removed = self
499 .cold_index
500 .invalidate_maintainer_announcements(maintainer_pubkey, identifier);
501
502 // Get from hot cache (for immediate re-processing)
503 let events = self
504 .hot_cache
505 .get_maintainer_events(maintainer_pubkey, identifier);
506
507 // Track metrics (using states metrics)
508 if let Some(ref metrics) = self.metrics {
509 if removed > 0 {
510 metrics.record_states_invalidation(removed);
511 }
512 if events.is_empty() {
513 metrics.record_states_hot_cache_miss();
514 } else {
515 for _ in &events {
516 metrics.record_states_hot_cache_hit();
517 }
518 }
519 }
520
521 // Update size metrics (using states metrics)
522 self.update_states_metrics();
523
524 (removed, events)
525 }
526
527 /// Clean up expired entries from both tiers (for announcements)
446 /// 528 ///
447 /// Returns tuple of (hot cache expired, cold index expired) 529 /// Returns tuple of (hot cache expired, cold index expired)
448 pub fn cleanup_expired(&self) -> (usize, usize) { 530 pub fn cleanup_expired(&self) -> (usize, usize) {
@@ -465,6 +547,29 @@ impl RejectedEventsIndex {
465 (hot_expired, cold_expired) 547 (hot_expired, cold_expired)
466 } 548 }
467 549
550 /// Clean up expired entries from both tiers (for states)
551 ///
552 /// Returns tuple of (hot cache expired, cold index expired)
553 pub fn cleanup_states_expired(&self) -> (usize, usize) {
554 let hot_expired = self.hot_cache.cleanup_expired();
555 let cold_expired = self.cold_index.cleanup_expired();
556
557 // Track metrics (using states metrics)
558 if let Some(ref metrics) = self.metrics {
559 if hot_expired > 0 {
560 metrics.record_states_hot_cache_expired(hot_expired);
561 }
562 if cold_expired > 0 {
563 metrics.record_states_cold_index_expired(cold_expired);
564 }
565 }
566
567 // Update size metrics (using states metrics)
568 self.update_states_metrics();
569
570 (hot_expired, cold_expired)
571 }
572
468 /// Get current number of entries in hot cache 573 /// Get current number of entries in hot cache
469 pub fn hot_cache_len(&self) -> usize { 574 pub fn hot_cache_len(&self) -> usize {
470 self.hot_cache.len() 575 self.hot_cache.len()