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

Merge branch 'blossom'

Browse Source
This commit is contained in:
Mike Dilger 2024-11-19 10:29:39 +13:00
commit a576c0f471
No known key found for this signature in database
GPG Key ID: 47581A78D4329BA4
24 changed files with 1023 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
Cargo.lock generated
View File

@ -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"

3
Cargo.toml
View File

@ -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"

13
contrib/chorus.toml
View File

@ -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 =

12
docs/CONFIG.md
View File

@ -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

5
src/bin/chorus.rs
View File

@ -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");

9
src/config.rs
View File

@ -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<String>,
}
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<Pubkey> = 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<String>,
}
impl Default for Config {
@ -229,7 +234,7 @@ impl Default for Config {
}
impl Config {
pub fn url(&self, inner: Uri, http: bool) -> Result<Uri, Error> {
pub fn uri_parts(&self, inner: Uri, http: bool) -> Result<http::uri::Parts, Error> {
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)
}
}

52
src/error.rs
View File

@ -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<std::io::Error> for Error {
}
}
impl From<http::header::ToStrError> for Error {
#[track_caller]
fn from(err: http::header::ToStrError) -> Self {
Error {
inner: ChorusError::NonAsciiHttpHeaderValue(err),
location: std::panic::Location::caller(),
}
}
}
impl From<pocket_db::Error> for Error {
#[track_caller]
fn from(err: pocket_db::Error) -> Self {
@ -514,6 +548,16 @@ impl From<serde_json::Error> for Error {
}
}
impl From<hex::FromHexError> for Error {
#[track_caller]
fn from(err: hex::FromHexError) -> Self {
Error {
inner: ChorusError::FromHex(err),
location: std::panic::Location::caller(),
}
}
}
impl From<std::string::FromUtf8Error> for Error {
#[track_caller]
fn from(err: std::string::FromUtf8Error) -> Self {
@ -539,3 +583,9 @@ impl From<Infallible> for Error {
panic!("INFALLIBLE")
}
}
impl From<Error> for std::io::Error {
fn from(e: Error) -> std::io::Error {
std::io::Error::other(e)
}
}

65
src/filestore/hash_output.rs Normal file
View File

@ -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<HashOutput, Error> {
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<P: AsRef<Path>>(&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"
);
}
}

162
src/filestore/mod.rs Normal file
View File

@ -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<P: AsRef<Path>>(base: P) -> Result<FileStore, Error> {
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<Bytes, Error>,
expected_hash: Option<HashOutput>,
) -> 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<BoxBody<Bytes, Error>, 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<Metadata, Error> {
// 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(())
}
}

3
src/globals.rs
View File

@ -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<Config>,
pub store: OnceLock<Store>,
pub filestore: OnceLock<FileStore>,
pub http1builder: http1::Builder,
pub rid: OnceLock<String>,
@ -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,

1
src/lib.rs
View File

@ -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;

146
src/web/blossom/auth.rs Normal file
View File

@ -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<AuthData, Error> {
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<AuthVerb>,
/// If an 'x' tag was included, this is the hash
pub hash: Option<[u8; 32]>,
}
pub fn verify_auth(request: &Request<Incoming>) -> Result<AuthData, Error> {
// 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<Incoming>) -> Result<AuthData, Error> {
// 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<AuthVerb> = 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<u64, Error> {
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)
}
}

341
src/web/blossom/mod.rs Normal file
View File

@ -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<Incoming>) -> Result<Response<BoxBody<Bytes, Error>>, 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<Incoming>) -> Result<Response<BoxBody<Bytes, Error>>, 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<Response<BoxBody<Bytes, Error>>, 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<Incoming>,
methods: &str,
) -> Result<Response<BoxBody<Bytes, Error>>, 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<Incoming>,
) -> Result<Response<BoxBody<Bytes, Error>>, 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<Incoming>,
) -> Result<Response<BoxBody<Bytes, Error>>, 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<Incoming>,
) -> Result<Response<BoxBody<Bytes, Error>>, 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<Incoming>,
) -> Result<Response<BoxBody<Bytes, Error>>, 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,
}

6
src/web/management/auth.rs
View File

@ -86,7 +86,11 @@ pub async fn check_auth(request: Request<Incoming>) -> Result<Value, Error> {
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)
};

21
src/web/mod.rs
View File

@ -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", "*")

3
test_blossom/.gitignore vendored Normal file
View File

@ -0,0 +1,3 @@
event.map
lmdb/
downloaded

7
test_blossom/README.md Normal file
View File

@ -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`

BIN
test_blossom/avatar-placeholder.webp Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 2.4 KiB

2
test_blossom/blossom/.gitignore vendored Normal file
View File

@ -0,0 +1,2 @@
*
!.gitignore

3
test_blossom/clean.rs Executable file
View File

@ -0,0 +1,3 @@
#!/bin/bash
rm -rf ./event.map ./lmdb/ ./blossom/* ./downloaded

37
test_blossom/config.toml Normal file
View File

@ -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"

26
test_blossom/create_auth.sh Executable file
View File

@ -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"

6
test_blossom/run.rs Executable file
View File

@ -0,0 +1,6 @@
#!/bin/bash
pushd ..
cargo build --release || exit 1
popd
../target/release/chorus ./config.toml

72
test_blossom/test.sh Executable file
View File

@ -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