Locutus/locutus-server
2
1
Fork 0
Code Issues Pull Requests Projects Releases Activity

Subscriptions working

Browse Source
This commit is contained in:
Lorenz Stechauner
2021-06-05 14:17:23 +02:00
parent a96cbdc059
commit 473b553662
12 changed files with 306 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/database.rs
View File

@ -1,8 +1,6 @@
use crate::error::*; use crate::error::*;
use bb8_postgres::tokio_postgres::NoTls; use bb8_postgres::tokio_postgres::NoTls;
use bb8_postgres::PostgresConnectionManager; use bb8_postgres::PostgresConnectionManager;
use std::ops::Deref;
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration; use std::time::Duration;
pub enum Pool { pub enum Pool {

22
src/error.rs
View File

@ -1,4 +1,4 @@
use crate::usimp::{InputEnvelope, OutputEnvelope}; use crate::usimp::{InputEnvelope, OutputEnvelope, Event};
use serde_json::{Value, Map}; use serde_json::{Value, Map};
use bb8_postgres::tokio_postgres; use bb8_postgres::tokio_postgres;
@ -25,6 +25,10 @@ pub enum ErrorKind {
UsimpError, UsimpError,
WebSocketError, WebSocketError,
DatabaseError, DatabaseError,
InvalidSessionError,
AuthenticationError,
SubscriptionError,
InternalError,
} }
impl InputEnvelope { impl InputEnvelope {
@ -64,7 +68,10 @@ impl Error {
ErrorKind::NotImplemented => "NOT_IMPLEMENTED", ErrorKind::NotImplemented => "NOT_IMPLEMENTED",
ErrorKind::UsimpError => "USIMP_ERROR", ErrorKind::UsimpError => "USIMP_ERROR",
ErrorKind::WebSocketError => "WEBSOCKET_ERROR", ErrorKind::WebSocketError => "WEBSOCKET_ERROR",
ErrorKind::DatabaseError => "BACKEND_ERROR", ErrorKind::DatabaseError | ErrorKind::InternalError => "SERVER_ERROR",
ErrorKind::InvalidSessionError => "INVALID_SESSION_ERROR",
ErrorKind::AuthenticationError => "AUTHENTICATION_ERROR",
ErrorKind::SubscriptionError => "SUBSCRIPTION_ERROR",
} }
} }
} }
@ -149,3 +156,14 @@ impl From<bb8_postgres::bb8::RunError<tokio_postgres::Error>> for Error {
} }
} }
} }
impl From<tokio::sync::mpsc::error::SendError<Event>> for Error {
fn from(error: tokio::sync::mpsc::error::SendError<Event>) -> Self {
Error {
kind: ErrorKind::InternalError,
class: ErrorClass::ServerError,
msg: None,
desc: Some(error.to_string()),
}
}
}

4
src/http.rs
View File

@ -55,7 +55,7 @@ async fn endpoint_handler(req: &mut Request<Body>, endpoint: String) -> Result<O
data, data,
}; };
Ok(Some(usimp::endpoint(&input).await?)) Ok(Some(usimp::endpoint(&input, None).await?))
} }
pub async fn handler(mut req: Request<Body>) -> Result<Response<Body>, hyper::Error> { pub async fn handler(mut req: Request<Body>) -> Result<Response<Body>, hyper::Error> {
@ -135,7 +135,7 @@ pub async fn handler(mut req: Request<Body>) -> Result<Response<Body>, hyper::Er
match val { match val {
Ok(val) => Ok(val), Ok(val) => Ok(val),
Err(error) => { Err(_error) => {
todo!("help") todo!("help")
} }
} }

1
src/main.rs
View File

@ -63,6 +63,7 @@ async fn main() -> Result<(), Error> {
println!("Locutus server"); println!("Locutus server");
database::init().await?; database::init().await?;
usimp::subscription::init();
let server1 = Server::bind(&SocketAddr::from(([0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0], 8080))); let server1 = Server::bind(&SocketAddr::from(([0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0], 8080)));
let service = make_service_fn(|_: &AddrStream| async { let service = make_service_fn(|_: &AddrStream| async {

53
src/usimp/handler/authenticate.rs
View File

@ -1,9 +1,10 @@
use crate::usimp;
use crate::usimp::*; use crate::usimp::*;
use crate::database; use crate::database;
use serde_json::{Value, from_value, to_value}; use serde_json::{Value, from_value, to_value};
use serde::{Serialize, Deserialize}; use serde::{Serialize, Deserialize};
use std::ops::Deref; use rand::Rng;
#[derive(Serialize, Deserialize, Clone)] #[derive(Serialize, Deserialize, Clone)]
struct Input { struct Input {
@ -13,22 +14,56 @@ struct Input {
#[derive(Serialize, Deserialize, Clone)] #[derive(Serialize, Deserialize, Clone)]
struct Output { struct Output {
session: String, session_id: String,
token: String, token: String,
} }
pub async fn handle(input: &InputEnvelope, session: &Session) -> Result<Value, Error> { pub async fn handle(input: &InputEnvelope, session: Option<Session>) -> Result<Value, Error> {
Ok(to_value(authenticate(from_value(input.data.clone())?).await?)?) Ok(to_value(authenticate(from_value(input.data.clone())?, session).await?)?)
} }
async fn authenticate(input: Input) -> Result<Output, Error> { async fn authenticate(input: Input, _session: Option<Session>) -> Result<Output, Error> {
match database::client().await? { let backend = database::client().await?;
let token;
let session_id;
match backend {
database::Client::Postgres(client) => { database::Client::Postgres(client) => {
client.execute("SELECT * FROM asdf;", &[]).await?; let res = client.query(
"SELECT account_id, domain_id \
FROM accounts \
WHERE account_name = $1",
&[&input.name]
).await?;
if res.len() == 0 {
return Err(Error::new(ErrorKind::AuthenticationError, ErrorClass::ClientError, None));
}
let row = &res[0];
let account_id: String = row.get(0);
let domain_id: String = row.get(1);
// TODO password check
if !input.password.eq("MichaelScott") {
return Err(Error::new(ErrorKind::AuthenticationError, ErrorClass::ClientError, None));
}
session_id = usimp::get_id(&[domain_id.as_str(), account_id.as_str()]);
token = rand::thread_rng()
.sample_iter(&rand::distributions::Alphanumeric)
.take(256)
.map(char::from)
.collect();
client.execute(
"INSERT INTO sessions (account_id, session_nr, session_id, session_token) \
VALUES ($1, COALESCE((SELECT MAX(session_nr) + 1 \
FROM sessions \
WHERE account_id = $1), 1), $2, $3);",
&[&account_id, &session_id, &token],
).await?;
} }
} }
Ok(Output { Ok(Output {
session: "".to_string(), session_id,
token: "".to_string(), token,
}) })
} }

27
src/usimp/handler/mod.rs
View File

@ -1,18 +1,29 @@
mod ping; mod ping;
mod authenticate; mod authenticate;
mod subscribe;
mod new_event;
use crate::usimp::*; use crate::usimp::*;
use tokio::sync::mpsc;
pub async fn endpoint(input: &InputEnvelope, tx: Option<mpsc::Sender<OutputEnvelope>>) -> Result<OutputEnvelope, Error> {
if input.from_domain != None {
// TODO
return Err(Error::new(ErrorKind::NotImplemented, ErrorClass::ServerError, None));
}
let session;
if let Some(token) = &input.token {
session = Some(Session::from_token(token).await?);
} else {
session = None;
}
pub async fn endpoint(input: &InputEnvelope) -> Result<OutputEnvelope, Error> {
println!("Endpoint: {}", input.endpoint); println!("Endpoint: {}", input.endpoint);
let session= Session {
account: None,
id: "".to_string(),
nr: 0,
};
Ok(match input.endpoint.as_str() { Ok(match input.endpoint.as_str() {
"ping" => input.respond(ping::handle(&input, &session).await?), "ping" => input.respond(ping::handle(&input, session).await?),
"authenticate" => input.respond(authenticate::handle(&input, &session).await?), "authenticate" => input.respond(authenticate::handle(&input, session).await?),
"subscribe" => input.respond(subscribe::handle(&input, session, tx).await?),
"new_event" => input.respond(new_event::handle(&input, session).await?),
_ => input.new_error(ErrorKind::UsimpError, ErrorClass::ClientProtocolError, Some("Invalid endpoint".to_string())), _ => input.new_error(ErrorKind::UsimpError, ErrorClass::ClientProtocolError, Some("Invalid endpoint".to_string())),
}) })
} }

28
src/usimp/handler/new_event.rs Normal file
View File

@ -0,0 +1,28 @@
use crate::usimp::*;
use crate::usimp::subscription;
use serde_json::{Value, from_value, to_value};
use serde::{Serialize, Deserialize};
#[derive(Serialize, Deserialize, Clone)]
struct Input {
room_id: String,
events: Vec<Event>,
}
#[derive(Serialize, Deserialize, Clone)]
struct Output {
}
pub async fn handle(input: &InputEnvelope, session: Option<Session>) -> Result<Value, Error> {
Ok(to_value(new_event(from_value(input.data.clone())?, session).await?)?)
}
async fn new_event(input: Input, session: Option<Session>) -> Result<Output, Error> {
let _account = get_account(&session)?;
// TODO check permissions
for event in input.events {
subscription::push(input.room_id.as_str(), event).await?;
}
Ok(Output {})
}

6
src/usimp/handler/ping.rs
View File

@ -2,10 +2,10 @@ use crate::usimp::*;
use serde_json::Value; use serde_json::Value;
pub async fn handle(input: &InputEnvelope, session: &Session) -> Result<Value, Error> { pub async fn handle(input: &InputEnvelope, session: Option<Session>) -> Result<Value, Error> {
ping(&input.data).await ping(input.data.clone(), session).await
} }
async fn ping(input: &Value) -> Result<Value, Error> { async fn ping(input: Value, _session: Option<Session>) -> Result<Value, Error> {
Ok(input.clone()) Ok(input.clone())
} }

50
src/usimp/handler/subscribe.rs Normal file
View File

@ -0,0 +1,50 @@
use crate::usimp::*;
use crate::usimp::subscription;
use serde_json::{Value, from_value, to_value};
use serde::{Serialize, Deserialize};
use tokio::sync::mpsc;
#[derive(Serialize, Deserialize, Clone)]
struct Input {
}
#[derive(Serialize, Deserialize, Clone)]
struct Output {
event: Option<Event>,
}
pub async fn handle(input: &InputEnvelope, session: Option<Session>, tx: Option<mpsc::Sender<OutputEnvelope>>) -> Result<Value, Error> {
Ok(to_value(subscribe(from_value(input.data.clone())?, session, input.request_nr, tx).await?)?)
}
async fn subscribe(_input: Input, session: Option<Session>, req_nr: Option<u64>, tx: Option<mpsc::Sender<OutputEnvelope>>) -> Result<Output, Error> {
let account = get_account(&session)?;
let mut rx = subscription::subscribe_account(account).await;
match tx {
Some(tx) => {
tokio::spawn(async move {
while let Some(event) = rx.recv().await {
let _res = tx.send(OutputEnvelope {
error: None,
request_nr: req_nr,
data: to_value(event).unwrap(),
}).await;
}
});
Ok(Output {
event: None,
})
}
None => {
if let Some(event) = rx.recv().await {
Ok(Output {
event: Some(event),
})
} else {
Err(Error::new(ErrorKind::SubscriptionError, ErrorClass::ServerError, None))
}
}
}
}

67
src/usimp/mod.rs
View File

@ -1,10 +1,15 @@
mod handler; mod handler;
pub mod subscription;
pub use handler::endpoint; pub use handler::endpoint;
use serde_json::Value;
use crate::error::{Error, ErrorClass, ErrorKind}; use crate::error::{Error, ErrorClass, ErrorKind};
use crate::database;
use serde_json::Value;
use serde::{Serialize, Deserialize}; use serde::{Serialize, Deserialize};
use crypto::sha2::Sha256;
use crypto::digest::Digest;
use base64_url;
#[derive(Serialize, Deserialize)] #[derive(Serialize, Deserialize)]
pub struct InputEnvelope { pub struct InputEnvelope {
@ -22,14 +27,43 @@ pub struct OutputEnvelope {
pub data: Value, pub data: Value,
} }
#[derive(Clone, Serialize, Deserialize)]
pub struct Event {
data: Value,
}
pub struct Account {
id: String,
name: String,
domain: String,
}
pub struct Session { pub struct Session {
id: String, id: String,
nr: i32, nr: i32,
account: Option<Account>, account: Option<Account>,
} }
pub struct Account { pub fn get_id(input: &[&str]) -> String {
let mut hasher = Sha256::new();
hasher.input_str(chrono::Utc::now().timestamp_millis().to_string().as_str());
for part in input {
hasher.input_str(" ");
hasher.input_str(part);
}
let mut result = [0u8; 32];
hasher.result(&mut result);
base64_url::encode(&result)
}
pub fn get_account(session: &Option<Session>) -> Result<&Account, Error> {
match session {
Some(session) => match &session.account {
Some(account) => Ok(&account),
None => return Err(Error::new(ErrorKind::UsimpError, ErrorClass::ClientProtocolError, None))
},
None => return Err(Error::new(ErrorKind::UsimpError, ErrorClass::ClientProtocolError, None))
}
} }
impl InputEnvelope { impl InputEnvelope {
@ -43,7 +77,32 @@ impl InputEnvelope {
} }
impl Session { impl Session {
pub async fn from_token(token: &str) -> Self { pub async fn from_token(token: &str) -> Result<Self, Error> {
todo!("session") let backend = database::client().await?;
let session;
match backend {
database::Client::Postgres(client) => {
let res = client.query(
"SELECT session_id, session_nr, a.account_id, account_name, domain_id \
FROM accounts a JOIN sessions s ON a.account_id = s.account_id \
WHERE session_token = $1;",
&[&token]
).await?;
if res.len() == 0 {
return Err(Error::new(ErrorKind::InvalidSessionError, ErrorClass::ClientError, None));
}
let row = &res[0];
session = Session {
id: row.get(0),
nr: row.get(1),
account: Some(Account {
id: row.get(2),
name: row.get(3),
domain: row.get(4),
}),
};
}
}
Ok(session)
} }
} }

72
src/usimp/subscription.rs Normal file
View File

@ -0,0 +1,72 @@
use crate::usimp::*;
use crate::database;
use tokio::sync::{mpsc, Mutex};
use std::collections::HashMap;
use std::sync::Arc;
static mut ROOMS: Option<Arc<Mutex<HashMap<String, Vec<mpsc::Sender<Event>>>>>> = None;
static mut ACCOUNTS: Option<Arc<Mutex<HashMap<String, Vec<mpsc::Sender<Event>>>>>> = None;
pub fn init() {
unsafe {
ROOMS = Some(Arc::new(Mutex::new(HashMap::new())));
ACCOUNTS = Some(Arc::new(Mutex::new(HashMap::new())));
}
}
pub async fn subscribe_account(account: &Account) -> mpsc::Receiver<Event> {
let (tx, rx) = mpsc::channel::<Event>(64);
unsafe {
let mut acc = ACCOUNTS.as_ref().unwrap().lock().await;
match acc.get_mut(account.id.as_str()) {
Some(vec) => {
vec.push(tx);
},
None => {
acc.insert(account.id.clone(), vec!{tx});
},
}
}
rx
}
pub async fn push(room_id: &str, event: Event) -> Result<(), Error> {
let backend = database::client().await?;
let accounts = match backend {
database::Client::Postgres(client) => {
let res = client.query(
"SELECT account_id \
FROM members \
WHERE room_id = $1;",
&[&room_id]
).await?;
let mut acc: Vec<String> = Vec::new();
for row in res {
acc.push(row.get(0));
}
acc
}
};
unsafe {
let mut rooms = ROOMS.as_ref().unwrap().lock().await;
if let Some(rooms) = rooms.get_mut(room_id) {
for tx in rooms {
let _res = tx.send(event.clone()).await;
}
}
}
for account in accounts {
unsafe {
let mut accounts = ACCOUNTS.as_ref().unwrap().lock().await;
if let Some(acc) = accounts.get_mut(account.as_str()) {
for tx in acc {
let _res = tx.send(event.clone()).await;
}
}
}
}
Ok(())
}

8
src/websocket.rs
View File

@ -1,4 +1,4 @@
use hyper::{Request, Response, Body, StatusCode, header}; use hyper::{Request, Body, StatusCode, header};
use crate::usimp::*; use crate::usimp::*;
use crate::usimp; use crate::usimp;
use crate::error::*; use crate::error::*;
@ -40,11 +40,11 @@ async fn receiver(mut stream: SplitStream<WebSocketStream<Upgraded>>, tx: mpsc::
match res { match res {
Ok(msg) => { Ok(msg) => {
let input: InputEnvelope = serde_json::from_slice(&msg.into_data()[..]).unwrap(); let input: InputEnvelope = serde_json::from_slice(&msg.into_data()[..]).unwrap();
let output = match usimp::endpoint(&input).await { let output = match usimp::endpoint(&input, Some(tx.clone())).await {
Ok(output) => output, Ok(output) => output,
Err(error) => input.error(error), Err(error) => input.error(error),
}; };
tx.send(output).await; let _res = tx.send(output).await;
}, },
Err(error) => println!("{:?}", error), Err(error) => println!("{:?}", error),
} }
@ -76,7 +76,7 @@ pub async fn handler(req: Request<Body>, res: hyper::http::response::Builder) ->
Role::Server, Role::Server,
None, None,
).await; ).await;
let (tx, rx) = mpsc::channel::<OutputEnvelope>(16); let (tx, rx) = mpsc::channel::<OutputEnvelope>(64);
let (sink, stream) = ws_stream.split(); let (sink, stream) = ws_stream.split();
tokio::spawn(async move { tokio::spawn(async move {
sender(sink, rx).await sender(sink, rx).await