diff --git a/Cargo.lock b/Cargo.lock index a5f6d14..281716f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -179,6 +179,12 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9425c3bf7089c983facbae04de54513cce73b41c7f9ff8c845b54e7bc64ebbfb" +[[package]] +name = "bitcoin-io" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b47c4ab7a93edb0c7198c5535ed9b52b63095f4e9b45279c6736cec4b856baf" + [[package]] name = "bitcoin_hashes" version = "0.13.0" @@ -195,6 +201,7 @@ version = "0.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bb18c03d0db0247e147a21a6faafd5a7eb851c743db062de72018b6b7e8e4d16" dependencies = [ + "bitcoin-io", "hex-conservative 0.2.1", ] @@ -259,6 +266,7 @@ name = "chorus" version = "1.5.3" dependencies = [ "base64 0.22.1", + "bitcoin_hashes 0.14.0", "dashmap", "env_logger", "futures", @@ -283,6 +291,8 @@ dependencies = [ "textnonce", "tokio", "tokio-rustls", + "tokio-stream", + "tokio-util", "toml", "url", ] @@ -1827,6 +1837,17 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-stream" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4f4e6ce100d0eb49a2734f8c0812bcd324cf357d21810932c5df6b96ef2b86f1" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "tokio-tungstenite" version = "0.24.0" @@ -1839,6 +1860,19 @@ dependencies = [ "tungstenite", ] +[[package]] +name = "tokio-util" +version = "0.7.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61e7c3654c13bcd040d4a03abee2c75b1d14a37b423cf5a813ceae1cc903ec6a" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "pin-project-lite", + "tokio", +] + [[package]] name = "toml" version = "0.8.19" diff --git a/Cargo.toml b/Cargo.toml index 67d22d3..39281ff 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,7 @@ edition = "2021" [dependencies] base64 = "0.22" +bitcoin_hashes = { version = "0.14", features = [ "bitcoin-io" ] } dashmap = "6" env_logger = "0.11" futures = "0.3" @@ -32,6 +33,8 @@ speedy = "0.8" textnonce = "1" tokio = { version = "1", features = [ "full" ] } tokio-rustls = "0.26" +tokio-util = { version = "0.7", features = [ "io" ] } +tokio-stream = "0.1" toml = "0.8" url = "2.5" diff --git a/contrib/chorus.toml b/contrib/chorus.toml index 2108951..e360bd6 100644 --- a/contrib/chorus.toml +++ b/contrib/chorus.toml @@ -299,3 +299,16 @@ throttling_bytes_per_second = 1024576 # Default is 16777216 bytes. # throttling_burst = 16777216 + + +# Blossom server directory +# +# Set to a filesystem directory where you want chorus to store files. +# +# Blossom allows clients to upload files making them available for the public to download. +# Our implementation makes all files publicly readable, but only chorus users can upload +# or delete. See https://github.com/hzrd149/blossom +# +# Default is not set +# +# blossom_directory = diff --git a/docs/CONFIG.md b/docs/CONFIG.md index 7d58c91..3de5492 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -257,3 +257,15 @@ second. If that bucket doesn't have enough, the burst won't be allowed and the c will be closed. Default is 16777216 bytes. + +### blossom_directory + +Blossom server directory + +Set to a filesystem directory where you want chorus to store files. + +Blossom allows clients to upload files making them available for the public to download. +Our implementation makes all files publicly readable, but only chorus users can upload +or delete. See https://github.com/hzrd149/blossom + +Default is None diff --git a/src/bin/chorus.rs b/src/bin/chorus.rs index 8363a8e..ff93571 100644 --- a/src/bin/chorus.rs +++ b/src/bin/chorus.rs @@ -31,6 +31,11 @@ async fn main() -> Result<(), Error> { let store = chorus::setup_store(&config)?; let _ = GLOBALS.store.set(store); + if let Some(ref blossom_directory) = config.blossom_directory { + let filestore = chorus::filestore::FileStore::new(blossom_directory).await?; + let _ = GLOBALS.filestore.set(filestore); + } + // TLS setup let maybe_tls_acceptor = if config.use_tls { log::info!(target: "Server", "Using TLS"); diff --git a/src/config.rs b/src/config.rs index d6d3450..dd83e77 100644 --- a/src/config.rs +++ b/src/config.rs @@ -40,6 +40,7 @@ pub struct FriendlyConfig { pub max_connections_per_ip: usize, pub throttling_bytes_per_second: usize, pub throttling_burst: usize, + pub blossom_directory: Option, } impl Default for FriendlyConfig { @@ -77,6 +78,7 @@ impl Default for FriendlyConfig { max_connections_per_ip: 5, throttling_bytes_per_second: 1024 * 1024, throttling_burst: 1024 * 1024 * 16, + blossom_directory: None, } } } @@ -116,6 +118,7 @@ impl FriendlyConfig { max_connections_per_ip, throttling_bytes_per_second, throttling_burst, + blossom_directory, } = self; let mut public_key: Option = None; @@ -177,6 +180,7 @@ impl FriendlyConfig { max_connections_per_ip, throttling_bytes_per_second, throttling_burst, + blossom_directory, }) } } @@ -217,6 +221,7 @@ pub struct Config { pub max_connections_per_ip: usize, pub throttling_bytes_per_second: usize, pub throttling_burst: usize, + pub blossom_directory: Option, } impl Default for Config { @@ -229,7 +234,7 @@ impl Default for Config { } impl Config { - pub fn url(&self, inner: Uri, http: bool) -> Result { + pub fn uri_parts(&self, inner: Uri, http: bool) -> Result { let mut uri_parts = inner.into_parts(); let scheme = match (self.use_tls, http) { (false, false) => Scheme::from_str("ws").unwrap(), @@ -240,6 +245,6 @@ impl Config { uri_parts.scheme = Some(scheme); let authority = Authority::from_str(&format!("{}:{}", self.hostname, self.port))?; uri_parts.authority = Some(authority); - Ok(Uri::from_parts(uri_parts)?) + Ok(uri_parts) } } diff --git a/src/error.rs b/src/error.rs index 95d71da..59be098 100644 --- a/src/error.rs +++ b/src/error.rs @@ -23,7 +23,7 @@ impl std::fmt::Display for Error { /// Errors that can occur in the chorus crate #[derive(Debug)] pub enum ChorusError { - // Auth failure + // Nostr AUTH failure AuthFailure(String), // Auth required @@ -50,6 +50,9 @@ pub enum ChorusError { // Blocked IP BlockedIp, + // Blossom Authorization failure + BlossomAuthFailure(String), + // Channel Recv ChannelRecv(tokio::sync::broadcast::error::RecvError), @@ -68,6 +71,9 @@ pub enum ChorusError { // Event is Invalid EventIsInvalid(String), + // From hex + FromHex(hex::FromHexError), + // From UTF8 FromUtf8(std::string::FromUtf8Error), @@ -98,6 +104,9 @@ pub enum ChorusError { // Missing Table MissingTable(&'static str), + // Non-ASCII HTTP header value + NonAsciiHttpHeaderValue(http::header::ToStrError), + // No private key NoPrivateKey, @@ -137,6 +146,9 @@ pub enum ChorusError { // Serde JSON SerdeJson(serde_json::Error), + // Signal - Not Blossom Request + SignalNotBlossom, + // Speedy Speedy(speedy::Error), @@ -176,12 +188,14 @@ impl std::fmt::Display for ChorusError { ChorusError::BannedUser => write!(f, "User is banned"), ChorusError::Base64Decode(e) => write!(f, "{e}"), ChorusError::BlockedIp => write!(f, "IP is temporarily blocked"), + ChorusError::BlossomAuthFailure(s) => write!(f, "Authorization failure: {s}"), ChorusError::ChannelRecv(e) => write!(f, "{e}"), ChorusError::ChannelSend(e) => write!(f, "{e}"), ChorusError::Config(e) => write!(f, "{e}"), ChorusError::Crypto(e) => write!(f, "{e}"), ChorusError::ErrorClose => write!(f, "Closing due to error(s)"), ChorusError::EventIsInvalid(s) => write!(f, "Event is invalid: {s}"), + ChorusError::FromHex(e) => write!(f, "{e}"), ChorusError::FromUtf8(e) => write!(f, "{e}"), ChorusError::General(s) => write!(f, "{s}"), ChorusError::Http(e) => write!(f, "{e}"), @@ -192,6 +206,9 @@ impl std::fmt::Display for ChorusError { ChorusError::Io(e) => write!(f, "{e}"), ChorusError::ManagementAuthFailure(s) => write!(f, "Authorization failure: {s}"), ChorusError::MissingTable(t) => write!(f, "Missing table: {t}"), + ChorusError::NonAsciiHttpHeaderValue(e) => { + write!(f, "Non ASCII HTTP header value: {e}") + } ChorusError::NoPrivateKey => write!(f, "Private Key Not Found"), ChorusError::NotImplemented => write!(f, "Not implemented"), ChorusError::NoSuchSubscription => write!(f, "No such subscription"), @@ -205,6 +222,7 @@ impl std::fmt::Display for ChorusError { ChorusError::Rustls(e) => write!(f, "{e}"), ChorusError::Scraper => write!(f, "Filter is underspecified. Scrapers are not allowed"), ChorusError::SerdeJson(e) => write!(f, "{e}"), + ChorusError::SignalNotBlossom => write!(f, "internal-signal-not-blossom"), ChorusError::Speedy(e) => write!(f, "{e}"), ChorusError::TimedOut => write!(f, "Timed out"), ChorusError::TooManySubscriptions => write!(f, "Too many subscriptions"), @@ -225,12 +243,14 @@ impl StdError for ChorusError { ChorusError::ChannelSend(e) => Some(e), ChorusError::Config(e) => Some(e), ChorusError::Crypto(e) => Some(e), + ChorusError::FromHex(e) => Some(e), ChorusError::FromUtf8(e) => Some(e), ChorusError::Http(e) => Some(e), ChorusError::Hyper(e) => Some(e), ChorusError::InvalidUri(e) => Some(e), ChorusError::InvalidUriParts(e) => Some(e), ChorusError::Io(e) => Some(e), + ChorusError::NonAsciiHttpHeaderValue(e) => Some(e), ChorusError::PocketDb(e) => Some(e), ChorusError::PocketDbHeed(e) => Some(e), ChorusError::PocketType(e) => Some(e), @@ -266,12 +286,14 @@ impl ChorusError { ChorusError::BannedUser => 0.2, ChorusError::Base64Decode(_) => 0.0, ChorusError::BlockedIp => 0.0, + ChorusError::BlossomAuthFailure(_) => 0.0, ChorusError::ChannelRecv(_) => 0.0, ChorusError::ChannelSend(_) => 0.0, ChorusError::Config(_) => 0.0, ChorusError::Crypto(_) => 0.1, ChorusError::ErrorClose => 1.0, ChorusError::EventIsInvalid(_) => 0.2, + ChorusError::FromHex(_) => 0.2, ChorusError::FromUtf8(_) => 0.2, ChorusError::General(_) => 0.0, ChorusError::Http(_) => 0.0, @@ -282,6 +304,7 @@ impl ChorusError { ChorusError::Io(_) => 0.0, ChorusError::ManagementAuthFailure(_) => 0.0, ChorusError::MissingTable(_) => 0.0, + ChorusError::NonAsciiHttpHeaderValue(_) => 0.2, ChorusError::NoPrivateKey => 0.0, ChorusError::NotImplemented => 0.0, ChorusError::NoSuchSubscription => 0.05, @@ -295,6 +318,7 @@ impl ChorusError { ChorusError::Rustls(_) => 0.0, ChorusError::Scraper => 0.4, ChorusError::SerdeJson(_) => 0.0, + ChorusError::SignalNotBlossom => 0.0, ChorusError::Speedy(_) => 0.0, ChorusError::TimedOut => 0.1, ChorusError::TooManySubscriptions => 0.1, @@ -414,6 +438,16 @@ impl From for Error { } } +impl From for Error { + #[track_caller] + fn from(err: http::header::ToStrError) -> Self { + Error { + inner: ChorusError::NonAsciiHttpHeaderValue(err), + location: std::panic::Location::caller(), + } + } +} + impl From for Error { #[track_caller] fn from(err: pocket_db::Error) -> Self { @@ -514,6 +548,16 @@ impl From for Error { } } +impl From for Error { + #[track_caller] + fn from(err: hex::FromHexError) -> Self { + Error { + inner: ChorusError::FromHex(err), + location: std::panic::Location::caller(), + } + } +} + impl From for Error { #[track_caller] fn from(err: std::string::FromUtf8Error) -> Self { @@ -539,3 +583,9 @@ impl From for Error { panic!("INFALLIBLE") } } + +impl From for std::io::Error { + fn from(e: Error) -> std::io::Error { + std::io::Error::other(e) + } +} diff --git a/src/filestore/hash_output.rs b/src/filestore/hash_output.rs new file mode 100644 index 0000000..b34f030 --- /dev/null +++ b/src/filestore/hash_output.rs @@ -0,0 +1,65 @@ +use crate::{ChorusError, Error}; +use std::fmt; +use std::path::{Path, PathBuf}; + +/// A simple type for a SHA-256 hash output of 32 bytes +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct HashOutput([u8; 32]); + +impl HashOutput { + pub fn from_engine(engine: bitcoin_hashes::sha256::HashEngine) -> HashOutput { + use bitcoin_hashes::{sha256, Hash}; + let hashvalue = sha256::Hash::from_engine(engine); + HashOutput(hashvalue.as_byte_array()[0..32].try_into().unwrap()) + } + + pub fn from_hex(input: &str) -> Result { + let bytes = hex::decode(input)?; + if bytes.len() == 32 { + Ok(HashOutput(bytes.try_into().unwrap())) + } else { + Err( + ChorusError::General("HashOutput::from_hex() got wrong length string".to_string()) + .into(), + ) + } + } + + pub fn from_bytes(bytes: [u8; 32]) -> HashOutput { + HashOutput(bytes) + } + + pub fn to_pathbuf>(&self, base: P) -> PathBuf { + let s = hex::encode(self.0); + let mut output: PathBuf = PathBuf::new(); + output.push(base); + output.push(&s[0..=1]); + output.push(&s[2..=3]); + output.push(&s[4..]); + output + } +} + +impl fmt::Display for HashOutput { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", hex::encode(self.0)) + } +} + +#[cfg(test)] +mod test { + use super::*; + + #[test] + fn test_hash_output_to_pathbuf() { + let hash = HashOutput::from_hex( + "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", + ) + .unwrap(); + + assert_eq!( + &format!("{}", hash.to_pathbuf("/tmp").display()), + "/tmp/e3/b0/c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + ); + } +} diff --git a/src/filestore/mod.rs b/src/filestore/mod.rs new file mode 100644 index 0000000..7d85a48 --- /dev/null +++ b/src/filestore/mod.rs @@ -0,0 +1,162 @@ +use crate::error::{ChorusError, Error}; +use futures::TryStreamExt; +use http_body_util::combinators::BoxBody; +use http_body_util::{BodyDataStream, BodyExt, StreamBody}; +use hyper::body::{Bytes, Frame}; +use std::fs::Metadata; +use std::path::{Path, PathBuf}; +use tokio::fs; +use tokio::fs::File; +use tokio_util::io::{InspectReader, ReaderStream, StreamReader}; + +mod hash_output; +pub use hash_output::HashOutput; + +pub struct FileStore { + pub base: PathBuf, + pub temp: PathBuf, +} + +impl FileStore { + pub async fn new>(base: P) -> Result { + let base = base.as_ref().to_owned(); + + let temp = { + let mut temp = base.clone(); + temp.push("temp"); + temp + }; + + if !fs::try_exists(&temp).await? { + fs::create_dir_all(&temp).await?; + } + + Ok(FileStore { base, temp }) + } + + fn tmpfile(&self) -> PathBuf { + let mut tf = self.temp.clone(); + let nonce = textnonce::TextNonce::sized_urlsafe(32).unwrap(); + tf.push(&nonce.0); + tf + } + + /// Store a file in storage, streamed from a hyper BoxBody + /// + /// Returns it's HashOutput by which it can be later retrieved or deleted. + pub async fn store( + &self, + data: BoxBody, + expected_hash: Option, + ) -> Result<(u64, HashOutput), Error> { + use bitcoin_hashes::sha256; + use std::io::Write; // for hash_engine.write_all() + + // We will download into a temporary file (as we don't know the hash yet) + let temppathbuf = self.tmpfile(); + let mut tempfile = File::options() + .create(true) + .truncate(true) + .write(true) + .open(&temppathbuf) + .await?; + + // Convert the BoxBody into a Data Stream + let body_stream = BodyDataStream::new(data); + + // Convert the Data Stream into something that is AsyncRead (over bytes) + let stream_reader = StreamReader::new(body_stream); + + // Wrap this in something that lets us inspect the content so we can + // hash it as it comes in, as well as count the size + let mut size: u64 = 0; + let mut hash_engine = sha256::HashEngine::default(); + let mut inspect_reader = InspectReader::new(stream_reader, |bytes: &[u8]| { + size += bytes.len() as u64; + hash_engine.write_all(bytes).unwrap(); // I don't think hashing will fail + }); + + // Copy the data into the tempfile (hashing and counting as we go) + let count = tokio::io::copy(&mut inspect_reader, &mut tempfile).await?; + + // Verify our code was correct + if count != size { + return Err(ChorusError::General("INTERNAL COUNT MISMATCH".to_string()).into()); + } + + // Finish the hash + let hash = HashOutput::from_engine(hash_engine); + + // Verify the expected hash matches + if let Some(expected) = expected_hash { + if hash != expected { + // Remove the temporary file + fs::remove_file(&temppathbuf).await?; + + // And complain + return Err(ChorusError::BlossomAuthFailure( + "File hash does not match authorized hash".to_string(), + ) + .into()); + } + } + + // Compute the proper path + let pathbuf = hash.to_pathbuf(&self.base); + + // If it already exists, trust the existing copy + if fs::try_exists(&pathbuf).await? { + // Just clean up + fs::remove_file(&temppathbuf).await?; + + return Ok((size, hash)); + } + + // Make the parent directory + fs::create_dir_all(pathbuf.parent().unwrap()).await?; + + // Move the file + fs::rename(&temppathbuf, &pathbuf).await?; + + Ok((size, hash)) + } + + /// Retrieve a file from storage by its HashOutput, streamed to a hyper BoxBoxy + pub async fn retrieve(&self, hash: HashOutput) -> Result, Error> { + // Compute the path + let pathbuf = hash.to_pathbuf(&self.base); + + // Open the file + let file = File::open(&pathbuf).await?; + + // Convert the AsyncRead file into a Stream + let reader_stream = ReaderStream::new(file); + + // Convert the Stream into a Body + let stream_body = StreamBody::new(reader_stream.map_ok(Frame::data)); + + // Box the body, mapping the error + let boxed_body = BodyExt::map_err(stream_body, |e| e.into()).boxed(); + + Ok(boxed_body) + } + + /// Check if a file exists and provide it's metadata (including .len()) + pub async fn metadata(&self, hash: HashOutput) -> Result { + // Compute the path + let pathbuf = hash.to_pathbuf(&self.base); + + Ok(tokio::fs::metadata(&pathbuf).await?) + } + + /// Delete a file from storage by its HashOutput + pub async fn delete(&self, hash: HashOutput) -> Result<(), Error> { + // Compute the path + let pathbuf = hash.to_pathbuf(&self.base); + + // Delete the file + tokio::fs::remove_file(&pathbuf).await?; + + Ok(()) + } +} diff --git a/src/globals.rs b/src/globals.rs index b9de435..1940e36 100644 --- a/src/globals.rs +++ b/src/globals.rs @@ -1,4 +1,5 @@ use crate::config::Config; +use crate::filestore::FileStore; use crate::ip::HashedIp; use dashmap::DashMap; use hyper::server::conn::http1; @@ -18,6 +19,7 @@ pub struct Globals { pub bytes_outbound: AtomicU64, pub config: RwLock, pub store: OnceLock, + pub filestore: OnceLock, pub http1builder: http1::Builder, pub rid: OnceLock, @@ -48,6 +50,7 @@ lazy_static! { bytes_outbound: AtomicU64::new(0), config: RwLock::new(Default::default()), store: OnceLock::new(), + filestore: OnceLock::new(), http1builder, rid: OnceLock::new(), new_events, diff --git a/src/lib.rs b/src/lib.rs index 859750b..e3bb337 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,6 +1,7 @@ pub mod config; pub mod counting_stream; pub mod error; +pub mod filestore; pub mod globals; pub mod ip; pub mod nostr; diff --git a/src/web/blossom/auth.rs b/src/web/blossom/auth.rs new file mode 100644 index 0000000..e481716 --- /dev/null +++ b/src/web/blossom/auth.rs @@ -0,0 +1,146 @@ +use crate::error::{ChorusError, Error}; +use crate::globals::GLOBALS; +use base64::prelude::*; +use http::header::AUTHORIZATION; +use hyper::body::Incoming; +use hyper::Request; +use pocket_types::Event; + +fn s_err(s: &str) -> Result { + Err(ChorusError::BlossomAuthFailure(s.to_owned()).into()) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum AuthVerb { + Upload, + List, + Delete, + Mirror, +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct AuthData { + /// If a verb was included, this is it + pub verb: Option, + + /// If an 'x' tag was included, this is the hash + pub hash: Option<[u8; 32]>, +} + +pub fn verify_auth(request: &Request) -> Result { + // Force every other error into a BlossomAuthFailure error + match verify_auth_inner(request) { + Ok(ad) => Ok(ad), + Err(e) => match e.inner { + ChorusError::BlossomAuthFailure(_) => Err(e), + _ => Err(ChorusError::BlossomAuthFailure(format!("{e}")).into()), + }, + } +} + +fn verify_auth_inner(request: &Request) -> Result { + // Must have AUTHORIZATION header + let authz = match request.headers().get(AUTHORIZATION) { + Some(h) => h, + None => return s_err("Authorization Required"), + }; + + // Authorization header must be type "nostr" + if !authz.to_str()?.to_ascii_lowercase().starts_with("nostr ") { + return s_err("You must use the Nostr authorization scheme"); + } + + let base64 = match authz.to_str()?.get(6..) { + Some(x) => x, + None => return s_err("Missing auth base64 encoded event"), + }; + + // Authorization header must be base64 + let event_bytes = BASE64_STANDARD.decode(base64)?; + + // Authorization header base64 must decode to a nostr Event + let mut buffer = vec![0; base64.len()]; + let (_size, event) = Event::from_json(&event_bytes, &mut buffer)?; + + // Nostr event must be valid + if let Err(e) = event.verify() { + return s_err(&format!("Authorization event is invalid: {}", e)); + } + + // Nostr event must be signed by a chorus user + if !GLOBALS.config.read().user_keys.contains(&event.pubkey()) { + return s_err("You are not an authorized user"); + } + + // Event kind must be 24242 + if event.kind().as_u16() != 24242 { + return s_err("Authorization event not kind 24242"); + } + + // Event created_at must be in the past (we give 30 seconds leeway) + use pocket_types::Time; + let now = Time::now(); + if event.created_at() > now + 30 { + return s_err("Authorization event too far in the future"); + } + + let tags = event.tags()?; + + // Expiration tag must be in the future + if let Some(v) = tags.get_value(b"expiration") { + let u = parse_u64(v)?; + let expiration = Time::from_u64(u); + if expiration < now { + return s_err("Authorization event has expired"); + } + } else { + return s_err("Authorization event missing expiration tag"); + } + + // We let the caller check the verb and hash since those are specific + // to the endpoint (and the 'x' must be checked later on) + + let verb: Option = if let Some(t) = tags.get_value(b"t") { + if t == b"upload" { + Some(AuthVerb::Upload) + } else if t == b"list" { + Some(AuthVerb::List) + } else if t == b"delete" { + Some(AuthVerb::Delete) + } else { + None + } + } else { + None + }; + + let hash: Option<[u8; 32]> = if let Some(v) = tags.get_value(b"x") { + let vec = hex::decode(v)?; + if vec.len() == 32 { + Some(vec.try_into().unwrap()) + } else { + return s_err("Authorization event x tag is of the wrong length"); + } + } else { + None + }; + + Ok(AuthData { verb, hash }) +} + +// FIXME, expose these from pocket-types +fn parse_u64(input: &[u8]) -> Result { + let mut pos = 0; + let mut value: u64 = 0; + let mut any: bool = false; + while pos < input.len() && b"0123456789".contains(&input[pos]) { + any = true; + value = (value * 10) + (input[pos] - 48) as u64; + pos += 1; + } + if !any { + Err(ChorusError::General("Auth event expiration is not a number".to_string()).into()) + } else { + Ok(value) + } +} diff --git a/src/web/blossom/mod.rs b/src/web/blossom/mod.rs new file mode 100644 index 0000000..7a376a9 --- /dev/null +++ b/src/web/blossom/mod.rs @@ -0,0 +1,341 @@ +use crate::error::{ChorusError, Error}; +use crate::filestore::HashOutput; +use crate::globals::GLOBALS; +use http::header::{ + ACCESS_CONTROL_ALLOW_HEADERS, ACCESS_CONTROL_ALLOW_METHODS, ACCESS_CONTROL_ALLOW_ORIGIN, + ACCESS_CONTROL_REQUEST_HEADERS, ACCESS_CONTROL_REQUEST_METHOD, ALLOW, CONTENT_LENGTH, + CONTENT_TYPE, ETAG, IF_MATCH, IF_MODIFIED_SINCE, IF_NONE_MATCH, ORIGIN, WWW_AUTHENTICATE, +}; +use http::{Method, StatusCode}; +//ACCEPT, AUTHORIZATION, DATE, ETAG, ORIGIN +use http_body_util::combinators::BoxBody; +use http_body_util::{BodyExt, Empty, Full}; +use hyper::body::{Bytes, Incoming}; +use hyper::{Request, Response}; +use serde::{Deserialize, Serialize}; + +mod auth; +use auth::{verify_auth, AuthVerb}; + +pub async fn handle(request: Request) -> Result>, Error> { + match route(request).await { + Ok(response) => Ok(response), + Err(e) => match e.inner { + ChorusError::SignalNotBlossom => Err(e), + _ => error_response(e), + }, + } +} + +pub async fn route(request: Request) -> Result>, Error> { + let p = request.uri().path(); + #[allow(clippy::int_plus_one)] + if p.starts_with("/") + && p.len() >= 1 + 64 + && p.chars().skip(1).take(64).all(|c| c.is_ascii_hexdigit()) + { + handle_hash(request).await + } else if p == "/upload" { + handle_upload(request).await + } else if p.starts_with("/list/") + && p.len() >= 6 + 64 + && p.chars().skip(6).take(64).all(|c| c.is_ascii_hexdigit()) + { + handle_list(request).await + } else if p == "/mirror" { + handle_mirror(request).await + } else { + Err(ChorusError::SignalNotBlossom.into()) + } +} + +fn error_response(e: Error) -> Result>, Error> { + use std::io::ErrorKind; + + let mut response = Response::builder().header(ACCESS_CONTROL_ALLOW_ORIGIN, "*"); + + let (status, reason) = match e.inner { + ChorusError::BlossomAuthFailure(m) => { + response = response.header(WWW_AUTHENTICATE, "Nostr"); + (StatusCode::UNAUTHORIZED, m) + } + ChorusError::FromHex(_) => (StatusCode::BAD_REQUEST, format!("{e}")), + ChorusError::Io(ref ioerror) => match ioerror.kind() { + ErrorKind::NotFound => (StatusCode::NOT_FOUND, "Not Found".to_owned()), + _ => (StatusCode::INTERNAL_SERVER_ERROR, format!("{e}")), + }, + _ => (StatusCode::INTERNAL_SERVER_ERROR, format!("{e}")), + }; + + Ok(response + .header("X-Reason", reason) + .status(status) + .body(Empty::new().map_err(|e| e.into()).boxed())?) +} + +fn options_response( + request: Request, + methods: &str, +) -> Result>, Error> { + if request + .headers() + .contains_key(ACCESS_CONTROL_REQUEST_HEADERS) + || request + .headers() + .contains_key(ACCESS_CONTROL_REQUEST_METHOD) + || request.headers().contains_key(ORIGIN) + { + // CORS OPTIONS response + Ok(Response::builder() + .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") + .header(ACCESS_CONTROL_ALLOW_HEADERS, "Authorization, *") + .header(ACCESS_CONTROL_ALLOW_METHODS, methods) + .header(CONTENT_LENGTH, "0") + .status(StatusCode::OK) + .body(Empty::new().map_err(|e| e.into()).boxed())?) + } else { + // Normal OPTIONS response + Ok(Response::builder() + .header(ALLOW, methods) + .status(StatusCode::NO_CONTENT) + .body(Empty::new().map_err(|e| e.into()).boxed())?) + } +} + +pub async fn handle_hash( + request: Request, +) -> Result>, Error> { + if matches!(request.method(), &Method::OPTIONS) { + return options_response(request, "OPTIONS, HEAD, GET, DELETE"); + } + + // HEAD, GET, DELETE + let p = request.uri().path(); + let hashstr: String = p.chars().skip(1).take(64).collect(); + let hash = match HashOutput::from_hex(&hashstr) { + Ok(h) => h, + Err(e) => return error_response(e), + }; + + let metadata = GLOBALS.filestore.get().unwrap().metadata(hash).await?; + + match *request.method() { + Method::HEAD | Method::GET => { + // Honor If-Match (fail if they didn't specify an etag matching the hash) + if let Some(etags) = request.headers().get(IF_MATCH) { + let mut onematch: bool = false; + for part in etags.to_str()?.split(',') { + if &part[1..part.len() - 1] == &format!("{}", hash) { + onematch = true; + } + } + if !onematch { + return Ok(Response::builder() + .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") + .header(ETAG, format!("\"{}\"", hash)) + .status(StatusCode::PRECONDITION_FAILED) + .body(Empty::new().map_err(|e| e.into()).boxed())?); + } + } + + // Honor If-None-Match (send NOT_MODIFIED if they specified an etag matching the hash) + // Honor If-Modified-Since (always send NOT_MODIFIED) + let mut send_not_modified: bool = false; + if let Some(etags) = request.headers().get(IF_NONE_MATCH) { + for part in etags.to_str()?.split(',') { + if &part[1..part.len() - 1] == &format!("{}", hash) { + send_not_modified = true; + } + } + } + if request.headers().get(IF_MODIFIED_SINCE).is_some() { + send_not_modified = true; + } + if send_not_modified { + return Ok(Response::builder() + .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") + .header(ETAG, format!("\"{}\"", hash)) + .status(StatusCode::NOT_MODIFIED) + .body(Empty::new().map_err(|e| e.into()).boxed())?); + } + + // Normal reasponse (HEAD or GET) + let response = Response::builder() + .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") + .header(CONTENT_LENGTH, format!("{}", metadata.len())) + .header(ETAG, format!("\"{}\"", hash)) + .status(StatusCode::OK); + + if matches!(*request.method(), Method::GET) { + let body = GLOBALS.filestore.get().unwrap().retrieve(hash).await?; + Ok(response.body(body)?) + } else { + Ok(response.body(Empty::new().map_err(|e| e.into()).boxed())?) + } + } + Method::DELETE => { + let auth_data = verify_auth(&request)?; + if auth_data.verb != Some(AuthVerb::Delete) { + return Err(ChorusError::BlossomAuthFailure( + "Delete was not authorized".to_string(), + ) + .into()); + } + + GLOBALS.filestore.get().unwrap().delete(hash).await?; + Ok(Response::builder() + .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") + .header(CONTENT_LENGTH, "0") + .status(StatusCode::OK) + .body(Empty::new().map_err(|e| e.into()).boxed())?) + } + _ => Ok(Response::builder() + .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") + .header(CONTENT_LENGTH, "0") + .status(StatusCode::METHOD_NOT_ALLOWED) + .body(Empty::new().map_err(|e| e.into()).boxed())?), + } +} + +pub async fn handle_upload( + request: Request, +) -> Result>, Error> { + if matches!(request.method(), &Method::OPTIONS) { + return options_response(request, "OPTIONS, HEAD, PUT"); + } + + let auth_data = verify_auth(&request)?; + if auth_data.verb != Some(AuthVerb::Upload) { + return Err( + ChorusError::BlossomAuthFailure("Upload was not authorized".to_string()).into(), + ); + } + + match *request.method() { + Method::HEAD => Ok(Response::builder() + .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") + .header(CONTENT_LENGTH, "0") + .status(StatusCode::NOT_IMPLEMENTED) + .body(Empty::new().map_err(|e| e.into()).boxed())?), + Method::PUT => { + let expected_hash = auth_data.hash.map(HashOutput::from_bytes); + if expected_hash.is_none() { + return Err(ChorusError::BlossomAuthFailure( + "Put requires an expected hash value x tag in the authorization event" + .to_string(), + ) + .into()); + } + + let uri = request.uri().to_owned(); + + let (size, hash) = GLOBALS + .filestore + .get() + .unwrap() + .store( + request.into_body().map_err(|e| e.into()).boxed(), + expected_hash, + ) + .await?; + + let uri = { + let mut parts = GLOBALS.config.read().uri_parts(uri, true)?; + parts.path_and_query = Some(http::uri::PathAndQuery::from_maybe_shared(format!( + "/{}", + hash + ))?); + http::Uri::from_parts(parts)? + }; + + let blob_descriptor = BlobDescriptor { + url: format!("{}", uri), + sha256: format!("{}", hash), + size, + uploaded: pocket_types::Time::now().as_u64(), + }; + + let descriptor_json_string = serde_json::to_string(&blob_descriptor)?; + let body_bytes = descriptor_json_string.into_bytes(); + let len = body_bytes.len(); + let body = Full::new(Bytes::from(body_bytes)) + .map_err(|e| e.into()) + .boxed(); + + Ok(Response::builder() + .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") + .header(CONTENT_LENGTH, format!("{}", len)) + .header(CONTENT_TYPE, "application/json") + .status(StatusCode::OK) + .body(body)?) + } + _ => Ok(Response::builder() + .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") + .header(CONTENT_LENGTH, "0") + .status(StatusCode::METHOD_NOT_ALLOWED) + .body(Empty::new().map_err(|e| e.into()).boxed())?), + } +} + +pub async fn handle_list( + request: Request, +) -> Result>, Error> { + if matches!(request.method(), &Method::OPTIONS) { + return options_response(request, "OPTIONS, GET"); + } + + let auth_data = verify_auth(&request)?; + if auth_data.verb != Some(AuthVerb::List) { + return Err(ChorusError::BlossomAuthFailure("List was not authorized".to_string()).into()); + } + + match *request.method() { + Method::GET => Ok(Response::builder() + .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") + .header(CONTENT_LENGTH, "0") + .status(StatusCode::NOT_IMPLEMENTED) + .body(Empty::new().map_err(|e| e.into()).boxed())?), + _ => Ok(Response::builder() + .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") + .header(CONTENT_LENGTH, "0") + .status(StatusCode::METHOD_NOT_ALLOWED) + .body(Empty::new().map_err(|e| e.into()).boxed())?), + } +} + +pub async fn handle_mirror( + request: Request, +) -> Result>, Error> { + if matches!(request.method(), &Method::OPTIONS) { + return options_response(request, "OPTIONS, PUT"); + } + + let auth_data = verify_auth(&request)?; + if auth_data.verb != Some(AuthVerb::Mirror) { + return Err( + ChorusError::BlossomAuthFailure("Mirror was not authorized".to_string()).into(), + ); + } + + match *request.method() { + Method::PUT => Ok(Response::builder() + .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") + .header(CONTENT_LENGTH, "0") + .status(StatusCode::NOT_IMPLEMENTED) + .body(Empty::new().map_err(|e| e.into()).boxed())?), + _ => Ok(Response::builder() + .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*") + .header(CONTENT_LENGTH, "0") + .status(StatusCode::METHOD_NOT_ALLOWED) + .body(Empty::new().map_err(|e| e.into()).boxed())?), + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct BlobDescriptor { + pub url: String, + pub sha256: String, + pub size: u64, + // type: String + pub uploaded: u64, +} diff --git a/src/web/management/auth.rs b/src/web/management/auth.rs index d33932b..bc0816b 100644 --- a/src/web/management/auth.rs +++ b/src/web/management/auth.rs @@ -86,7 +86,11 @@ pub async fn check_auth(request: Request) -> Result { if let Some(u) = tags.get_value(b"u") { let auth_url = String::from_utf8(u.to_owned())?; let actual_url = { - let uri = GLOBALS.config.read().url(request.uri().to_owned(), true)?; + let uri_parts = GLOBALS + .config + .read() + .uri_parts(request.uri().to_owned(), true)?; + let uri = http::Uri::from_parts(uri_parts)?; format!("{}", uri) }; diff --git a/src/web/mod.rs b/src/web/mod.rs index 1989c96..8dd3490 100644 --- a/src/web/mod.rs +++ b/src/web/mod.rs @@ -1,7 +1,9 @@ +mod blossom; mod management; mod nip11; -use crate::error::Error; +use crate::error::{ChorusError, Error}; +use crate::globals::GLOBALS; use crate::ip::HashedPeer; use http::Method; use http_body_util::combinators::BoxBody; @@ -41,7 +43,22 @@ pub async fn serve_http( } } - log::debug!(target: "Client", "{}: HTTP request for {}", peer, request.uri()); + let uri = request.uri().to_owned(); + + // Try blossom if enabled + if GLOBALS.config.read().blossom_directory.is_some() { + match blossom::handle(request).await { + Ok(response) => return Ok(response), + Err(e) => { + if !matches!(e.inner, ChorusError::SignalNotBlossom) { + return Err(e); + } + } + } + } + + log::debug!(target: "Client", "{}: HTTP request for {}", peer, uri); + let response = Response::builder() .header("Access-Control-Allow-Origin", "*") .header("Access-Control-Allow-Headers", "*") diff --git a/test_blossom/.gitignore b/test_blossom/.gitignore new file mode 100644 index 0000000..20080a8 --- /dev/null +++ b/test_blossom/.gitignore @@ -0,0 +1,3 @@ +event.map +lmdb/ +downloaded diff --git a/test_blossom/README.md b/test_blossom/README.md new file mode 100644 index 0000000..dbdd73d --- /dev/null +++ b/test_blossom/README.md @@ -0,0 +1,7 @@ +# Testing Blossom functionality of Chorus + +In one shell run `run.sh` to run a local chorus. + +Then in another shell run `test.sh` to run tests against that running instance. + +When done, break out of the server and run `clean.sh` diff --git a/test_blossom/avatar-placeholder.webp b/test_blossom/avatar-placeholder.webp new file mode 100644 index 0000000..e8f5973 Binary files /dev/null and b/test_blossom/avatar-placeholder.webp differ diff --git a/test_blossom/blossom/.gitignore b/test_blossom/blossom/.gitignore new file mode 100644 index 0000000..d6b7ef3 --- /dev/null +++ b/test_blossom/blossom/.gitignore @@ -0,0 +1,2 @@ +* +!.gitignore diff --git a/test_blossom/clean.rs b/test_blossom/clean.rs new file mode 100755 index 0000000..80a123d --- /dev/null +++ b/test_blossom/clean.rs @@ -0,0 +1,3 @@ +#!/bin/bash + +rm -rf ./event.map ./lmdb/ ./blossom/* ./downloaded diff --git a/test_blossom/config.toml b/test_blossom/config.toml new file mode 100644 index 0000000..c6b2f8f --- /dev/null +++ b/test_blossom/config.toml @@ -0,0 +1,37 @@ +# See contrib/chorus.toml for a documented config file + +data_directory = "." +ip_address = "127.0.0.1" +port = 8089 +hostname = "localhost" +chorus_is_behind_a_proxy = false +use_tls = false +certchain_pem_path = "tls/fullchain.pem" +key_pem_path = "tls/privkey.pem" +name = "Chorus Sample" +description = "A sample run of the Chorus relay" +# icon_url = +open_relay = false +user_hex_keys = [ + "12bb541d03bfc3cab0f4a8e4db28947f60faae6fca4e315eb27f809c6eff9a0b" +] +moderator_hex_keys = [ + "12bb541d03bfc3cab0f4a8e4db28947f60faae6fca4e315eb27f809c6eff9a0b" +] +verify_events = true +allow_scraping = false +allow_scrape_if_limited_to = 100 +allow_scrape_if_max_seconds = 7200 +max_subscriptions = 128 +serve_ephemeral = true +serve_relay_lists = true +server_log_level = "Info" +library_log_level = "Info" +client_log_level = "Warn" +enable_ip_blocking = true +minimum_ban_seconds = 1 +timeout_seconds = 60 +max_connections_per_ip = 5 +throttling_bytes_per_second = 131072 +throttling_burst = 4194304 +blossom_directory = "./blossom" \ No newline at end of file diff --git a/test_blossom/create_auth.sh b/test_blossom/create_auth.sh new file mode 100755 index 0000000..a536e1b --- /dev/null +++ b/test_blossom/create_auth.sh @@ -0,0 +1,26 @@ +#!/bin/bash + +PUBKEY=12bb541d03bfc3cab0f4a8e4db28947f60faae6fca4e315eb27f809c6eff9a0b +PRIVKEY=b4a98d96270b6cd30c80e4fd594461d2b22d8dbcfbcd1f7b11bf0ef2b028a56b +AUTH_EXPIRATION=1900000000 + +VERB=$1 +HASH=$2 + +if [ x$VERB = x ] ; then + echo "USAGE: create_auth.sh VERB HASH" + exit 1 +fi + +if [ x$HASH = x ] ; then + echo "USAGE: create_auth.sh VERB HASH" + exit 1 +fi + +PRE_EVENT='{"pubkey": "'$PUBKEY'", "kind": 24242, "created_at": 0, "tags": [["expiration","'$AUTH_EXPIRATION'"], ["t","'$VERB'"], ["x","'$HASH'"]], "content":""}' + +EVENT=$(echo "$PRE_EVENT" | nak event --sec $PRIVKEY) + +EVENT_BASE64=$(echo $EVENT | base64 -w 0) + +echo "Authorization: Nostr $EVENT_BASE64" diff --git a/test_blossom/run.rs b/test_blossom/run.rs new file mode 100755 index 0000000..ecfbf4b --- /dev/null +++ b/test_blossom/run.rs @@ -0,0 +1,6 @@ +#!/bin/bash + +pushd .. +cargo build --release || exit 1 +popd +../target/release/chorus ./config.toml diff --git a/test_blossom/test.sh b/test_blossom/test.sh new file mode 100755 index 0000000..7ef675b --- /dev/null +++ b/test_blossom/test.sh @@ -0,0 +1,72 @@ +#!/bin/bash + +if ! command -v curl 2>&1 >/dev/null +then + echo "curl command is required." + exit 1 +fi + +if ! command -v jq 2>&1 >/dev/null +then + echo "jq command is required." + exit 1 +fi + +if ! command -v nak 2>&1 >/dev/null +then + echo "nak command is required. https://github.com/fiatjaf/nak" + exit 1 +fi + +# UPLOAD TEST ------------ + +FILE="./avatar-placeholder.webp" +HASH=$(sha256sum $FILE | awk '{print $1}') + +# Generate nostr auth +AUTH=$(./create_auth.sh upload $HASH) + +# Upload +DESCRIPTOR=$(curl -s --data-binary @"$FILE" -X PUT --header "$AUTH" http://127.0.0.1:8089/upload) +if [ $? -ne 0 ] ; then + echo "FAILED: Curl (uploading) exited with a non-zero status" + exit 1 +fi +echo "PASS: FILE UPLOADED" + +# Extract the sha256 and compare it +DHASH=$(echo "$DESCRIPTOR" | jq -r .sha256) +if [ $? -ne 0 ] ; then + echo "FAILED: jq failed extracting sha256 from descriptor" + exit 1 +fi +if [ "$HASH" != "$DHASH" ] ; then + echo "returned descriptor 'sha256' does not match the hash" +fi +echo "PASS: DESCRIPTOR HASH MATCHES" + +# Extract the URL for download +URL=$(echo "$DESCRIPTOR" | jq -r .url) +if [ $? -ne 0 ] ; then + echo "FAILED: jq failed extracting url from descriptor" + exit 1 +fi + +# DOWNLOAD TEST ----------- + +curl -s "$URL" > downloaded +if [ $? -ne 0 ] ; then + echo "FAILED: Curl (downloading) exited with a non-zero status" + exit 1 +fi +echo "PASS: FILE DOWNLOADED" + +# Compare the files +if cmp -s "$FILE" downloaded; then + echo "PASS: THE DOWNLOADED FILE MATCHES THE UPLOADED FILE" +else + echo "FAIL: THE DOWNLOADED FILE DOES NOT MATCH THE UPLOADED FILE" +fi + +echo "end." +exit 0