nik/chorus
1
0
Fork 0
mirror of https://github.com/mikedilger/chorus.git synced 2026-08-31 07:31:01 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Use parking lot RwLock, move away from async RwLock, honor config changes to allow_scraping

Browse Source
This commit is contained in:
Mike Dilger 2024-03-16 14:44:42 +13:00
parent 6a65c116ca
commit b80f14d56b
8 changed files with 46 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Cargo.lock generated
View File

@ -153,6 +153,7 @@ dependencies = [
"hyper-tungstenite",
"lazy_static",
"log",
"parking_lot",
"rustls-pemfile",
"serde",
"tempfile",

1
chorus-bin/Cargo.toml
View File

@ -16,6 +16,7 @@ hyper = { version = "0.14", features = [ "http1", "server", "runtime", "stream"
hyper-tungstenite = "0.11"
lazy_static = "1.4"
log = "0.4"
parking_lot = "0.12"
rustls-pemfile = "1.0"
serde = { version = "1.0", features = ["derive"] }
textnonce = "1"

2
chorus-bin/src/bin/dump.rs
View File

@ -43,7 +43,7 @@ fn main() -> Result<(), Error> {
let (_incount, _outcount, filter) = Filter::from_json(b"{}", &mut buffer)?;
let screen = |_: &Event<'_>| -> bool { true };
let mut events = store.find_events(filter, screen)?;
let mut events = store.find_events(filter, screen, &config)?;
for event in events.drain(..) {
let bytes = event.as_json()?;
let s = unsafe { std::str::from_utf8_unchecked(&bytes) };

2
chorus-bin/src/globals.rs
View File

@ -2,11 +2,11 @@ use chorus_lib::config::Config;
use chorus_lib::store::Store;
use hyper::server::conn::Http;
use lazy_static::lazy_static;
use parking_lot::RwLock;
use std::sync::atomic::AtomicUsize;
use std::sync::OnceLock;
use tokio::sync::broadcast::Sender as BroadcastSender;
use tokio::sync::watch::Sender as WatchSender;
use tokio::sync::RwLock;
pub struct Globals {
pub config: RwLock<Config>,

19
chorus-bin/src/main.rs
View File

@ -80,7 +80,7 @@ async fn main() -> Result<(), Error> {
log::info!(target: "Server", "Running on {}:{}", config.ip_address, config.port);
// Store config into GLOBALS
*GLOBALS.config.write().await = config;
*GLOBALS.config.write() = config;
let mut interrupt_signal = signal(SignalKind::interrupt())?;
let mut quit_signal = signal(SignalKind::quit())?;
@ -114,7 +114,7 @@ async fn main() -> Result<(), Error> {
let friendly_config: FriendlyConfig = toml::from_str(&contents)?;
let config: Config = friendly_config.into_config()?;
*GLOBALS.config.write().await = config;
*GLOBALS.config.write() = config;
},
// Accepts network connections and spawn a task to serve each one
@ -475,22 +475,13 @@ impl WebSocketService {
.unwrap()
.get_event_by_offset(new_event_offset)?
{
let config = &*GLOBALS.config.read().await;
let event_flags = {
let user_keys_ref = &config.user_keys;
nostr::event_flags(&event, &self.user, user_keys_ref)
};
let authorized_user = nostr::authorized_user(&self.user).await;
let event_flags = nostr::event_flags(&event, &self.user);
let authorized_user = nostr::authorized_user(&self.user);
'subs: for (subid, filters) in self.subscriptions.iter() {
for filter in filters.iter() {
if filter.as_filter()?.event_matches(&event)?
&& nostr::screen_outgoing_event(
&event,
&event_flags,
authorized_user,
config,
)
&& nostr::screen_outgoing_event(&event, &event_flags, authorized_user)
{
let message = NostrReply::Event(subid, event);
self.websocket

71
chorus-bin/src/nostr.rs
View File

@ -1,6 +1,5 @@
use crate::globals::GLOBALS;
use crate::WebSocketService;
use chorus_lib::config::Config;
use chorus_lib::error::{ChorusError, Error};
use chorus_lib::reply::{NostrReply, NostrReplyPrefix};
use chorus_lib::types::parse::json_escape::json_unescape;
@ -83,7 +82,7 @@ impl WebSocketService {
if let Err(e) = self.req_inner(&subid, filters).await {
let reply = match e.inner {
ChorusError::TooManySubscriptions => {
let max_subscriptions = GLOBALS.config.read().await.max_subscriptions;
let max_subscriptions = GLOBALS.config.read().max_subscriptions;
NostrReply::Closed(
&subid,
NostrReplyPrefix::Blocked,
@ -106,13 +105,13 @@ impl WebSocketService {
}
async fn req_inner(&mut self, subid: &String, filters: Vec<OwnedFilter>) -> Result<(), Error> {
let max_subscriptions = GLOBALS.config.read().await.max_subscriptions;
let max_subscriptions = GLOBALS.config.read().max_subscriptions;
if self.subscriptions.len() >= max_subscriptions {
return Err(ChorusError::TooManySubscriptions.into());
}
let user = self.user;
let authorized_user = authorized_user(&user).await;
let authorized_user = authorized_user(&user);
// NOTE on private events (DMs, GiftWraps)
// Most relays check if you are seeking them, and of which pubkey, and if you are
@ -127,19 +126,19 @@ impl WebSocketService {
{
let mut events: Vec<Event> = Vec::new();
let config = GLOBALS.config.read().await;
let user_keys_ref = &config.user_keys;
for filter in filters.iter() {
let screen = |event: &Event| {
let event_flags = event_flags(event, &user, user_keys_ref);
screen_outgoing_event(event, &event_flags, authorized_user, &config)
let event_flags = event_flags(event, &user);
screen_outgoing_event(event, &event_flags, authorized_user)
};
let filter_events = {
let config = &*GLOBALS.config.read();
GLOBALS
.store
.get()
.unwrap()
.find_events(filter.as_filter()?, screen, config)?
};
let filter_events = GLOBALS
.store
.get()
.unwrap()
.find_events(filter.as_filter()?, screen)?;
events.extend(filter_events);
}
@ -231,17 +230,14 @@ impl WebSocketService {
async fn event_inner(&mut self) -> Result<(), Error> {
let user = self.user;
let authorized_user = authorized_user(&user).await;
let authorized_user = authorized_user(&user);
// Delineate the event back out of the session buffer
let event = Event::delineate(&self.buffer)?;
let event_flags = {
let user_keys_ref = &*GLOBALS.config.read().await.user_keys;
event_flags(&event, &user, user_keys_ref)
};
let event_flags = event_flags(&event, &user);
if !event_flags.author_is_an_authorized_user || GLOBALS.config.read().await.verify_events {
if !event_flags.author_is_an_authorized_user || GLOBALS.config.read().verify_events {
// Verify the event is valid (id is hash, signature is valid)
if let Err(e) = event.verify() {
return Err(ChorusError::EventIsInvalid(format!("{}", e)).into());
@ -352,7 +348,7 @@ impl WebSocketService {
};
if let Some(h) = url.host() {
let theirhost = h.to_owned();
if theirhost == GLOBALS.config.read().await.hostname {
if theirhost == GLOBALS.config.read().hostname {
relay_ok = true;
}
}
@ -409,7 +405,7 @@ async fn screen_incoming_event(
}
// Accept if an open relay
if GLOBALS.config.read().await.open_relay {
if GLOBALS.config.read().open_relay {
return Ok(true);
}
@ -419,23 +415,17 @@ async fn screen_incoming_event(
}
// Accept relay lists from anybody
if event.kind() == Kind(10002) && GLOBALS.config.read().await.serve_relay_lists {
if event.kind() == Kind(10002) && GLOBALS.config.read().serve_relay_lists {
return Ok(true);
}
// Allow if event kind ephemeral
if event.kind().is_ephemeral() && GLOBALS.config.read().await.serve_ephemeral {
if event.kind().is_ephemeral() && GLOBALS.config.read().serve_ephemeral {
return Ok(true);
}
// If the author is one of our users, always accept it
if GLOBALS
.config
.read()
.await
.user_keys
.contains(&event.pubkey())
{
if GLOBALS.config.read().user_keys.contains(&event.pubkey()) {
return Ok(true);
}
@ -443,7 +433,7 @@ async fn screen_incoming_event(
for mut tag in event.tags()?.iter() {
if tag.next() == Some(b"p") {
if let Some(value) = tag.next() {
for ukhex in &GLOBALS.config.read().await.user_hex_keys {
for ukhex in &GLOBALS.config.read().user_hex_keys {
if value == ukhex.as_bytes() {
return Ok(true);
}
@ -459,7 +449,6 @@ pub fn screen_outgoing_event(
event: &Event<'_>,
event_flags: &EventFlags,
authorized_user: bool,
config: &Config,
) -> bool {
// Forbid if it is a private event (DM or GiftWrap) and theey are neither the recipient
// nor the author
@ -468,17 +457,17 @@ pub fn screen_outgoing_event(
}
// Allow if an open relay
if config.open_relay {
if GLOBALS.config.read().open_relay {
return true;
}
// Allow Relay Lists
if event.kind() == Kind(10002) && config.serve_relay_lists {
if event.kind() == Kind(10002) && GLOBALS.config.read().serve_relay_lists {
return true;
}
// Allow if event kind ephemeral
if event.kind().is_ephemeral() && config.serve_ephemeral {
if event.kind().is_ephemeral() && GLOBALS.config.read().serve_ephemeral {
return true;
}
@ -496,10 +485,10 @@ pub fn screen_outgoing_event(
false
}
pub async fn authorized_user(user: &Option<Pubkey>) -> bool {
pub fn authorized_user(user: &Option<Pubkey>) -> bool {
match user {
None => false,
Some(pk) => GLOBALS.config.read().await.user_keys.contains(pk),
Some(pk) => GLOBALS.config.read().user_keys.contains(pk),
}
}
@ -510,8 +499,8 @@ pub struct EventFlags {
pub tags_current_user: bool,
}
pub fn event_flags(event: &Event<'_>, user: &Option<Pubkey>, user_keys: &[Pubkey]) -> EventFlags {
let author_is_an_authorized_user = user_keys.contains(&event.pubkey());
pub fn event_flags(event: &Event<'_>, user: &Option<Pubkey>) -> EventFlags {
let author_is_an_authorized_user = GLOBALS.config.read().user_keys.contains(&event.pubkey());
let author_is_current_user = match user {
None => false,
@ -532,7 +521,7 @@ pub fn event_flags(event: &Event<'_>, user: &Option<Pubkey>, user_keys: &[Pubkey
}
}
if user_keys.contains(&tagged_pk) {
if GLOBALS.config.read().user_keys.contains(&tagged_pk) {
tags_an_authorized_user = true;
}
}

2
chorus-bin/src/web.rs
View File

@ -18,7 +18,7 @@ pub async fn serve_http(peer: SocketAddr, request: Request<Body>) -> Result<Resp
pub async fn serve_nip11(peer: SocketAddr) -> Result<Response<Body>, Error> {
log::debug!(target: "Client", "{}: sent NIP-11", peer);
let rid = {
let config = &*GLOBALS.config.read().await;
let config = &*GLOBALS.config.read();
GLOBALS.rid.get_or_init(|| build_rid(config))
};

16
chorus-lib/src/store/mod.rs
View File

@ -31,9 +31,6 @@ pub struct Store {
deleted_offsets: Database<U64<BigEndian>, Unit>,
deleted_events: Database<UnalignedSlice<u8>, Unit>,
ip_data: Database<UnalignedSlice<u8>, UnalignedSlice<u8>>,
allow_scraping: bool,
allow_scrape_if_limited_to: u32,
allow_scrape_if_max_seconds: u64,
}
impl Store {
@ -187,9 +184,6 @@ impl Store {
deleted_offsets,
deleted_events,
ip_data,
allow_scraping: config.allow_scraping,
allow_scrape_if_limited_to: config.allow_scrape_if_limited_to,
allow_scrape_if_max_seconds: config.allow_scrape_if_max_seconds,
};
// This is in migrations.rs
@ -303,7 +297,8 @@ impl Store {
}
/// Find all events that match the filter
pub fn find_events<F>(&self, filter: Filter, screen: F) -> Result<Vec<Event>, Error>
pub fn find_events<F>(&self, filter: Filter, screen: F, config: &Config)
-> Result<Vec<Event>, Error>
where
F: Fn(&Event) -> bool,
{
@ -626,9 +621,10 @@ impl Store {
} else {
// SCRAPE:
let maxtime = filter.until().0.min(Time::now().0);
let allow = self.allow_scraping
|| filter.limit() <= self.allow_scrape_if_limited_to
|| (maxtime - filter.since().0) < self.allow_scrape_if_max_seconds;
let allow = config.allow_scraping
|| filter.limit() <= config.allow_scrape_if_limited_to
|| (maxtime - filter.since().0) < config.allow_scrape_if_max_seconds;
if !allow {
return Err(ChorusError::Scraper.into());
}