| 1 | use std::borrow::Cow; |
| 2 | use std::collections::{HashMap, HashSet}; |
| 3 | use std::net::{IpAddr, SocketAddr}; |
| 4 | use std::sync::{Arc, Mutex, PoisonError}; |
| 5 | use std::time::{Duration, Instant}; |
| 6 | |
| 7 | use nostr::event::{Event, EventId, Kind}; |
| 8 | use nostr::filter::Filter; |
| 9 | use nostr::key::PublicKey; |
| 10 | use nostr::message::MachineReadablePrefix; |
| 11 | use nostr::nips::nip65::extract_relay_list; |
| 12 | use nostr::types::Timestamp; |
| 13 | use nostr_database::{DatabaseEventStatus, NostrDatabase, error::Error}; |
| 14 | use serde::Deserialize; |
| 15 | |
| 16 | use crate::config::{Config, Limits}; |
| 17 | |
| 18 | #[derive(Debug)] |
| 19 | struct Usage { |
| 20 | window: Instant, |
| 21 | events: u32, |
| 22 | queries: u32, |
| 23 | } |
| 24 | |
| 25 | #[derive(Debug, Clone)] |
| 26 | pub struct Policy { |
| 27 | database: Arc<dyn NostrDatabase>, |
| 28 | limits: Limits, |
| 29 | forge_keys: HashSet<PublicKey>, |
| 30 | peers: Arc<Mutex<HashMap<IpAddr, Usage>>>, |
| 31 | } |
| 32 | |
| 33 | #[derive(Deserialize)] |
| 34 | #[serde(deny_unknown_fields)] |
| 35 | struct Binding { |
| 36 | version: u64, |
| 37 | action: String, |
| 38 | fingerprint: String, |
| 39 | certificate: Option<String>, |
| 40 | signature: Option<String>, |
| 41 | } |
| 42 | |
| 43 | pub(crate) enum WritePolicyResult { |
| 44 | Accept, |
| 45 | Respond { |
| 46 | prefix: MachineReadablePrefix, |
| 47 | message: Cow<'static, str>, |
| 48 | status: bool, |
| 49 | }, |
| 50 | } |
| 51 | |
| 52 | impl WritePolicyResult { |
| 53 | const fn accepted(prefix: MachineReadablePrefix, message: &'static str) -> Self { |
| 54 | Self::Respond { |
| 55 | prefix, |
| 56 | message: Cow::Borrowed(message), |
| 57 | status: true, |
| 58 | } |
| 59 | } |
| 60 | |
| 61 | const fn reject(prefix: MachineReadablePrefix, message: &'static str) -> Self { |
| 62 | Self::Respond { |
| 63 | prefix, |
| 64 | message: Cow::Borrowed(message), |
| 65 | status: false, |
| 66 | } |
| 67 | } |
| 68 | } |
| 69 | |
| 70 | pub(crate) enum QueryPolicyResult { |
| 71 | Accept, |
| 72 | Reject { |
| 73 | prefix: MachineReadablePrefix, |
| 74 | message: Cow<'static, str>, |
| 75 | }, |
| 76 | } |
| 77 | |
| 78 | impl QueryPolicyResult { |
| 79 | const fn reject(prefix: MachineReadablePrefix, message: &'static str) -> Self { |
| 80 | Self::Reject { |
| 81 | prefix, |
| 82 | message: Cow::Borrowed(message), |
| 83 | } |
| 84 | } |
| 85 | } |
| 86 | |
| 87 | impl Policy { |
| 88 | pub fn new(database: Arc<dyn NostrDatabase>, config: &Config) -> anyhow::Result<Self> { |
| 89 | config.validate()?; |
| 90 | Ok(Self { |
| 91 | database, |
| 92 | limits: config.limits.clone(), |
| 93 | forge_keys: config.forge_keys()?, |
| 94 | peers: Arc::new(Mutex::new(HashMap::new())), |
| 95 | }) |
| 96 | } |
| 97 | |
| 98 | fn charge(&self, ip: IpAddr, publish: bool) -> bool { |
| 99 | let mut peers = self.peers.lock().unwrap_or_else(PoisonError::into_inner); |
| 100 | peers.retain(|_, usage| usage.window.elapsed() < Duration::from_secs(60)); |
| 101 | if peers.len() >= 4096 && !peers.contains_key(&ip) { |
| 102 | return false; |
| 103 | } |
| 104 | let usage = peers.entry(ip).or_insert_with(|| Usage { |
| 105 | window: Instant::now(), |
| 106 | events: 0, |
| 107 | queries: 0, |
| 108 | }); |
| 109 | let (count, limit) = if publish { |
| 110 | (&mut usage.events, self.limits.events_per_minute) |
| 111 | } else { |
| 112 | (&mut usage.queries, self.limits.queries_per_minute) |
| 113 | }; |
| 114 | if *count >= limit { |
| 115 | return false; |
| 116 | } |
| 117 | *count += 1; |
| 118 | drop(peers); |
| 119 | true |
| 120 | } |
| 121 | |
| 122 | pub(crate) async fn admit_event( |
| 123 | &self, |
| 124 | event: &Event, |
| 125 | address: &SocketAddr, |
| 126 | ) -> WritePolicyResult { |
| 127 | if !self.charge(address.ip(), true) { |
| 128 | return WritePolicyResult::reject( |
| 129 | MachineReadablePrefix::RateLimited, |
| 130 | "publish rate exceeded", |
| 131 | ); |
| 132 | } |
| 133 | self.admit(event).await.unwrap_or_else(|error| { |
| 134 | tracing::error!(%error, "event admission database failure"); |
| 135 | WritePolicyResult::reject(MachineReadablePrefix::Error, "database unavailable") |
| 136 | }) |
| 137 | } |
| 138 | |
| 139 | pub(crate) async fn admit_peer(&self, event: &Event) -> WritePolicyResult { |
| 140 | match self.admit(event).await { |
| 141 | Ok(result) => result, |
| 142 | Err(error) => { |
| 143 | tracing::error!(%error, "event admission database failure"); |
| 144 | WritePolicyResult::reject(MachineReadablePrefix::Error, "database unavailable") |
| 145 | } |
| 146 | } |
| 147 | } |
| 148 | |
| 149 | pub(crate) async fn admit_query( |
| 150 | &self, |
| 151 | filter: &Filter, |
| 152 | address: &SocketAddr, |
| 153 | ) -> QueryPolicyResult { |
| 154 | if !self.charge(address.ip(), false) { |
| 155 | return QueryPolicyResult::reject( |
| 156 | MachineReadablePrefix::RateLimited, |
| 157 | "query rate exceeded", |
| 158 | ); |
| 159 | } |
| 160 | if filter.as_json().len() > self.limits.max_filter_bytes { |
| 161 | return QueryPolicyResult::reject( |
| 162 | MachineReadablePrefix::RateLimited, |
| 163 | "filter byte limit exceeded", |
| 164 | ); |
| 165 | } |
| 166 | if filter.search.is_some() { |
| 167 | return QueryPolicyResult::reject( |
| 168 | MachineReadablePrefix::Unsupported, |
| 169 | "search is not supported", |
| 170 | ); |
| 171 | } |
| 172 | if let Some(limit) = filter.limit { |
| 173 | return if limit > self.limits.max_query_results { |
| 174 | QueryPolicyResult::reject( |
| 175 | MachineReadablePrefix::RateLimited, |
| 176 | "query result limit exceeded", |
| 177 | ) |
| 178 | } else { |
| 179 | QueryPolicyResult::Accept |
| 180 | }; |
| 181 | } |
| 182 | match self |
| 183 | .database |
| 184 | .count(filter.clone().limit(self.limits.max_query_results + 1)) |
| 185 | .await |
| 186 | { |
| 187 | Ok(count) if count <= self.limits.max_query_results => QueryPolicyResult::Accept, |
| 188 | Ok(_) => QueryPolicyResult::reject( |
| 189 | MachineReadablePrefix::RateLimited, |
| 190 | "history result limit exceeded", |
| 191 | ), |
| 192 | Err(error) => { |
| 193 | tracing::error!(%error, "query admission database failure"); |
| 194 | QueryPolicyResult::reject(MachineReadablePrefix::Error, "database unavailable") |
| 195 | } |
| 196 | } |
| 197 | } |
| 198 | |
| 199 | async fn admit(&self, event: &Event) -> Result<WritePolicyResult, Error> { |
| 200 | if let Some(result) = self.check_event(event) { |
| 201 | return Ok(result); |
| 202 | } |
| 203 | match self.database.check_id(&event.id).await? { |
| 204 | DatabaseEventStatus::Saved => { |
| 205 | return Ok(WritePolicyResult::accepted( |
| 206 | MachineReadablePrefix::Duplicate, |
| 207 | "already stored", |
| 208 | )); |
| 209 | } |
| 210 | DatabaseEventStatus::Deleted => return Ok(blocked("event is deleted")), |
| 211 | DatabaseEventStatus::NotExistent => {} |
| 212 | } |
| 213 | self.admit_kind(event).await |
| 214 | } |
| 215 | |
| 216 | fn check_event(&self, event: &Event) -> Option<WritePolicyResult> { |
| 217 | if event.as_json().len() > self.limits.max_event_bytes |
| 218 | || event.tags.len() > self.limits.max_tags |
| 219 | { |
| 220 | return Some(blocked("event exceeds size or tag limit")); |
| 221 | } |
| 222 | if event.created_at.as_secs() == 0 |
| 223 | || event.created_at.as_secs() > Timestamp::now().as_secs().saturating_add(900) |
| 224 | { |
| 225 | return Some(blocked("event timestamp is outside the supported range")); |
| 226 | } |
| 227 | if event.is_expired() || event.is_protected() { |
| 228 | return Some(blocked("expired or protected events are not accepted")); |
| 229 | } |
| 230 | None |
| 231 | } |
| 232 | |
| 233 | async fn admit_kind(&self, event: &Event) -> Result<WritePolicyResult, Error> { |
| 234 | match event.kind.as_u16() { |
| 235 | 10002 => { |
| 236 | if !self.forge_keys.contains(&event.pubkey) |
| 237 | && !self.registered_forge(event.pubkey).await? |
| 238 | { |
| 239 | return Ok(blocked("relay list signer has no registered repository")); |
| 240 | } |
| 241 | let relay_tags = event |
| 242 | .tags |
| 243 | .iter() |
| 244 | .filter(|tag| tag.as_slice().first().is_some_and(|name| name == "r")) |
| 245 | .count(); |
| 246 | if relay_tags > 16 || extract_relay_list(event).count() != relay_tags { |
| 247 | return Ok(blocked("relay list is invalid or too large")); |
| 248 | } |
| 249 | } |
| 250 | 30617 => { |
| 251 | let Some(identifier) = identifier(event) else { |
| 252 | return Ok(blocked("one valid repository identifier is required")); |
| 253 | }; |
| 254 | if !self.registered(event.pubkey, identifier).await? { |
| 255 | let count = self |
| 256 | .database |
| 257 | .query( |
| 258 | Filter::new() |
| 259 | .kind(Kind::from(30617)) |
| 260 | .limit(self.limits.max_repositories + 1), |
| 261 | ) |
| 262 | .await? |
| 263 | .len(); |
| 264 | if count >= self.limits.max_repositories { |
| 265 | return Ok(WritePolicyResult::reject( |
| 266 | MachineReadablePrefix::RateLimited, |
| 267 | "repository directory is full", |
| 268 | )); |
| 269 | } |
| 270 | } |
| 271 | } |
| 272 | 30618 => { |
| 273 | let Some(identifier) = identifier(event) else { |
| 274 | return Ok(blocked("one valid repository identifier is required")); |
| 275 | }; |
| 276 | if !self.registered(event.pubkey, identifier).await? { |
| 277 | return Ok(blocked("repository is not registered")); |
| 278 | } |
| 279 | } |
| 280 | 1617 | 1621 | 1630..=1632 => { |
| 281 | let mut addresses = 0_usize; |
| 282 | for tag in event |
| 283 | .tags |
| 284 | .iter() |
| 285 | .filter(|tag| tag.as_slice().first().is_some_and(|name| name == "a")) |
| 286 | { |
| 287 | let Some((kind, author, identifier)) = |
| 288 | tag.as_slice().get(1).and_then(|value| coordinate(value)) |
| 289 | else { |
| 290 | return Ok(blocked("invalid repository address")); |
| 291 | }; |
| 292 | if kind != 30617 || !self.registered(author, identifier).await? { |
| 293 | return Ok(blocked("repository is not registered")); |
| 294 | } |
| 295 | if matches!(event.kind.as_u16(), 1630..=1632) && author != event.pubkey { |
| 296 | return Ok(blocked("status signer does not own the repository")); |
| 297 | } |
| 298 | addresses += 1; |
| 299 | } |
| 300 | if addresses != 1 { |
| 301 | return Ok(blocked("exactly one repository address is required")); |
| 302 | } |
| 303 | } |
| 304 | 30382 => { |
| 305 | if !valid_binding(event) { |
| 306 | return Ok(blocked("invalid openpgp binding envelope")); |
| 307 | } |
| 308 | } |
| 309 | 5 => return self.deletion(event).await, |
| 310 | _ => { |
| 311 | return Ok(WritePolicyResult::reject( |
| 312 | MachineReadablePrefix::Unsupported, |
| 313 | "event kind is not supported", |
| 314 | )); |
| 315 | } |
| 316 | } |
| 317 | Ok(WritePolicyResult::Accept) |
| 318 | } |
| 319 | |
| 320 | async fn registered(&self, author: PublicKey, identifier: &str) -> Result<bool, Error> { |
| 321 | Ok(self |
| 322 | .database |
| 323 | .query( |
| 324 | Filter::new() |
| 325 | .kind(Kind::from(30617)) |
| 326 | .author(author) |
| 327 | .identifier(identifier) |
| 328 | .limit(1), |
| 329 | ) |
| 330 | .await? |
| 331 | .iter() |
| 332 | .any(|event| !event.is_expired())) |
| 333 | } |
| 334 | |
| 335 | async fn registered_forge(&self, author: PublicKey) -> Result<bool, Error> { |
| 336 | Ok(self |
| 337 | .database |
| 338 | .query( |
| 339 | Filter::new() |
| 340 | .kind(Kind::from(30617)) |
| 341 | .author(author) |
| 342 | .limit(1), |
| 343 | ) |
| 344 | .await? |
| 345 | .iter() |
| 346 | .any(|event| !event.is_expired())) |
| 347 | } |
| 348 | |
| 349 | async fn deletion(&self, event: &Event) -> Result<WritePolicyResult, Error> { |
| 350 | let mut targets = 0_usize; |
| 351 | for tag in event.tags.iter() { |
| 352 | let values = tag.as_slice(); |
| 353 | match values.first().map(String::as_str) { |
| 354 | Some("e") => { |
| 355 | let Some(id) = values |
| 356 | .get(1) |
| 357 | .filter(|id| id.len() == 64) |
| 358 | .and_then(|id| id.parse::<EventId>().ok()) |
| 359 | else { |
| 360 | return Ok(blocked("invalid deletion target")); |
| 361 | }; |
| 362 | if let Some(target) = self.database.event_by_id(&id).await? { |
| 363 | if target.pubkey != event.pubkey || target.kind == Kind::EventDeletion { |
| 364 | return Ok(blocked("deletion target is not removable by this author")); |
| 365 | } |
| 366 | } |
| 367 | targets += 1; |
| 368 | } |
| 369 | Some("a") => { |
| 370 | let Some((kind, author, _)) = values.get(1).and_then(|value| coordinate(value)) |
| 371 | else { |
| 372 | return Ok(blocked("invalid deletion address")); |
| 373 | }; |
| 374 | if author != event.pubkey || !matches!(kind, 30617 | 30618 | 30382) { |
| 375 | return Ok(blocked("deletion address is not owned by this author")); |
| 376 | } |
| 377 | targets += 1; |
| 378 | } |
| 379 | _ => {} |
| 380 | } |
| 381 | } |
| 382 | Ok(if targets == 0 { |
| 383 | blocked("deletion target is required") |
| 384 | } else { |
| 385 | WritePolicyResult::Accept |
| 386 | }) |
| 387 | } |
| 388 | } |
| 389 | |
| 390 | const fn blocked(message: &'static str) -> WritePolicyResult { |
| 391 | WritePolicyResult::reject(MachineReadablePrefix::Blocked, message) |
| 392 | } |
| 393 | |
| 394 | fn valid_identifier(identifier: &str) -> bool { |
| 395 | !identifier.is_empty() && identifier.len() <= 512 && !identifier.chars().any(char::is_control) |
| 396 | } |
| 397 | |
| 398 | fn identifier(event: &Event) -> Option<&str> { |
| 399 | let mut tags = event |
| 400 | .tags |
| 401 | .iter() |
| 402 | .filter(|tag| tag.as_slice().first().is_some_and(|name| name == "d")); |
| 403 | let tag = tags.next()?.as_slice(); |
| 404 | if tags.next().is_some() || tag.len() != 2 || !valid_identifier(&tag[1]) { |
| 405 | return None; |
| 406 | } |
| 407 | Some(&tag[1]) |
| 408 | } |
| 409 | |
| 410 | pub(crate) fn coordinate(value: &str) -> Option<(u16, PublicKey, &str)> { |
| 411 | let mut parts = value.splitn(3, ':'); |
| 412 | let kind = parts.next()?.parse().ok()?; |
| 413 | let author_hex = parts.next()?; |
| 414 | if author_hex.len() != 64 || !author_hex.bytes().all(|byte| byte.is_ascii_hexdigit()) { |
| 415 | return None; |
| 416 | } |
| 417 | let author = author_hex.parse().ok()?; |
| 418 | let identifier = parts.next()?; |
| 419 | valid_identifier(identifier).then_some((kind, author, identifier)) |
| 420 | } |
| 421 | |
| 422 | fn valid_binding(event: &Event) -> bool { |
| 423 | if event.content.len() > 192 * 1024 { |
| 424 | return false; |
| 425 | } |
| 426 | let Ok(body) = serde_json::from_str::<Binding>(&event.content) else { |
| 427 | return false; |
| 428 | }; |
| 429 | if body.version != 1 |
| 430 | || !matches!(body.fingerprint.len(), 40 | 64) |
| 431 | || !body |
| 432 | .fingerprint |
| 433 | .bytes() |
| 434 | .all(|byte| byte.is_ascii_hexdigit()) |
| 435 | || identifier(event) != Some(body.fingerprint.as_str()) |
| 436 | { |
| 437 | return false; |
| 438 | } |
| 439 | match body.action.as_str() { |
| 440 | "bind" => { |
| 441 | body.certificate |
| 442 | .is_some_and(|certificate| !certificate.trim().is_empty()) |
| 443 | && body |
| 444 | .signature |
| 445 | .is_some_and(|signature| !signature.trim().is_empty()) |
| 446 | } |
| 447 | "revoke" => body.certificate.is_none() && body.signature.is_none(), |
| 448 | _ => false, |
| 449 | } |
| 450 | } |