Fixes mutual connections not being seen and hence requests failing.

Cargo.lock

@@ -1,5 +1,7 @@
 # This file is automatically @generated by Cargo.
 # It is not intended for manual editing.
+version = 3
+
 [[package]]
 name = "addr2line"
 version = "0.14.1"
@@ -2473,7 +2475,7 @@ dependencies = [
 
 [[package]]
 name = "revolt"
-version = "0.3.3-alpha.6"
+version = "0.3.3-alpha.8"
 dependencies = [
  "async-std",
  "async-tungstenite",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "revolt"
-version = "0.3.3-alpha.6"
+version = "0.3.3-alpha.8"
 authors = ["Paul Makles <paulmakles@gmail.com>"]
 edition = "2018"
 

Dockerfile

@@ -1,5 +1,6 @@
 # Build Stage
 FROM ekidd/rust-musl-builder:nightly-2021-01-01 AS builder
+USER 0:0
 WORKDIR /home/rust/src
 
 RUN USER=root cargo new --bin revolt

src/database/entities/message.rs

@@ -13,7 +13,9 @@ use mongodb::{
 use rocket_contrib::json::JsonValue;
 use serde::{Deserialize, Serialize};
 use ulid::Ulid;
-use web_push::{ContentEncoding, SubscriptionInfo, VapidSignatureBuilder, WebPushClient, WebPushMessageBuilder};
+use web_push::{
+    ContentEncoding, SubscriptionInfo, VapidSignatureBuilder, WebPushClient, WebPushMessageBuilder,
+};
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Message {
@@ -171,13 +173,13 @@ impl Message {
             for subscription in subscriptions {
                 let mut builder = WebPushMessageBuilder::new(&subscription).unwrap();
                 let sig_builder =
-                    VapidSignatureBuilder::from_pem(std::io::Cursor::new(&key), &subscription).unwrap();
+                    VapidSignatureBuilder::from_pem(std::io::Cursor::new(&key), &subscription)
+                        .unwrap();
                 let signature = sig_builder.build().unwrap();
                 builder.set_vapid_signature(signature);
                 builder.set_payload(ContentEncoding::AesGcm, enc.as_bytes());
                 let m = builder.build().unwrap();
-                let response = client.send(m).await.unwrap();
-                dbg!(response);
+                let response = client.send(m).await.ok();
             }
         }
 
@@ -198,6 +200,26 @@ impl Message {
     }
 
     pub async fn delete(&self) -> Result<()> {
+        if let Some(attachment) = &self.attachment {
+            get_collection("attachments")
+                .update_one(
+                    doc! {
+                        "_id": &attachment.id
+                    },
+                    doc! {
+                        "$set": {
+                            "deleted": true
+                        }
+                    },
+                    None
+                )
+                .await
+                .map_err(|_| Error::DatabaseError {
+                    operation: "update_one",
+                    with: "attachment",
+                })?;
+        }
+
         get_collection("messages")
             .delete_one(
                 doc! {

src/database/entities/user.rs

@@ -1,6 +1,11 @@
+use mongodb::bson::doc;
 use serde::{Deserialize, Serialize};
+use mongodb::options::{Collation, FindOneOptions};
 
-use crate::{database::permissions::user::UserPermissions, notifications::websocket::is_online};
+use crate::database::*;
+use crate::util::result::{Error, Result};
+use crate::notifications::websocket::is_online;
+use crate::database::permissions::user::UserPermissions;
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
 pub enum RelationshipStatus {
@@ -62,4 +67,32 @@ impl User {
 
         self
     }
+
+    /// Utility function for checking claimed usernames.
+    pub async fn is_username_taken(username: &str) -> Result<bool> {
+        if username == "revolt" && username == "admin" {
+            return Ok(true)
+        }
+    
+        if get_collection("users")
+            .find_one(
+                doc! {
+                    "username": username
+                },
+                FindOneOptions::builder()
+                    .collation(Collation::builder().locale("en").strength(2).build())
+                    .build(),
+            )
+            .await
+            .map_err(|_| Error::DatabaseError {
+                operation: "find_one",
+                with: "user",
+            })?
+            .is_some()
+        {
+            Ok(true)
+        } else {
+            Ok(false)
+        }
+    }
 }

src/database/permissions/user.rs

@@ -68,14 +68,12 @@ impl<'a> PermissionCalculator<'a> {
             || get_collection("channels")
                 .find_one(
                     doc! {
-                        "$or": [
-                            { "type": "Group" },
-                            { "type": "DirectMessage" },
-                        ],
-                        "$and": [
-                            { "recipients": &self.perspective.id },
-                            { "recipients": target }
-                        ]
+                        "channel_type": {
+                            "$in": ["Group", "DirectMessage"]
+                        },
+                        "recipients": {
+                            "$all": [ &self.perspective.id, target ]
+                        }
                     },
                     None,
                 )

src/notifications/events.rs

@@ -26,6 +26,12 @@ pub enum WebSocketError {
 #[serde(tag = "type")]
 pub enum ServerboundNotification {
     Authenticate(Session),
+    BeginTyping {
+        channel: String
+    },
+    EndTyping {
+        channel: String
+    }
 }
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -63,6 +69,14 @@ pub enum ClientboundNotification {
     ChannelDelete {
         id: String,
     },
+    ChannelStartTyping {
+        id: String,
+        user: String
+    },
+    ChannelStopTyping {
+        id: String,
+        user: String
+    },
 
     UserRelationship {
         id: String,

src/notifications/websocket.rs

@@ -71,8 +71,6 @@ async fn accept(stream: TcpStream) {
     let fwd = rx.map(Ok).forward(write);
     let incoming = read.try_for_each(async move |msg| {
         let mutex = mutex_generator();
-        //dbg!(&mutex.lock().unwrap());
-
         if let Message::Text(text) = msg {
             if let Ok(notification) = serde_json::from_str::<ServerboundNotification>(&text) {
                 match notification {
@@ -161,6 +159,54 @@ async fn accept(stream: TcpStream) {
                             ));
                         }
                     }
+                    // ! TEMP: verify user part of channel
+                    // ! Could just run permission check here.
+                    ServerboundNotification::BeginTyping { channel } => {
+                        if mutex.lock().unwrap().is_some() {
+                            let user = {
+                                let mutex = mutex.lock().unwrap();
+                                let session = mutex.as_ref().unwrap();
+                                session.user_id.clone()
+                            };
+
+                            ClientboundNotification::ChannelStartTyping {
+                                id: channel.clone(),
+                                user
+                            }
+                            .publish(channel)
+                            .await
+                            .ok();
+                        } else {
+                            send(ClientboundNotification::Error(
+                                WebSocketError::AlreadyAuthenticated,
+                            ));
+
+                            return Ok(());
+                        }
+                    }
+                    ServerboundNotification::EndTyping { channel } => {
+                        if mutex.lock().unwrap().is_some() {
+                            let user = {
+                                let mutex = mutex.lock().unwrap();
+                                let session = mutex.as_ref().unwrap();
+                                session.user_id.clone()
+                            };
+
+                            ClientboundNotification::ChannelStopTyping {
+                                id: channel.clone(),
+                                user
+                            }
+                            .publish(channel)
+                            .await
+                            .ok();
+                        } else {
+                            send(ClientboundNotification::Error(
+                                WebSocketError::AlreadyAuthenticated,
+                            ));
+
+                            return Ok(());
+                        }
+                    }
                 }
             }
         }

src/routes/channels/delete_channel.rs

@@ -12,7 +12,7 @@ pub async fn req(user: User, target: Ref) -> Result<()> {
         .for_channel()
         .await?;
     if !perm.get_view() {
-        Err(Error::LabelMe)?
+        Err(Error::MissingPermission)?
     }
 
     match &target {
@@ -102,7 +102,8 @@ pub async fn req(user: User, target: Ref) -> Result<()> {
             Message::create(
                 "00000000000000000000000000".to_string(),
                 id.clone(),
-                format!("<@{}> left the group.", user.id),
+                // ! FIXME: make a schema for this
+                format!("{{\"type\":\"user_left\",\"id\":\"{}\"}}", user.id),
             )
             .publish(&target)
             .await

src/routes/channels/fetch_channel.rs

@@ -12,7 +12,7 @@ pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
         .for_channel()
         .await?;
     if !perm.get_view() {
-        Err(Error::LabelMe)?
+        Err(Error::MissingPermission)?
     }
 
     Ok(json!(target))

src/routes/channels/group_add_member.rs

@@ -16,7 +16,7 @@ pub async fn req(user: User, target: Ref, member: Ref) -> Result<()> {
         .for_channel()
         .await?;
     if !perm.get_view() {
-        Err(Error::LabelMe)?
+        Err(Error::MissingPermission)?
     }
 
     if let Channel::Group { id, recipients, .. } = &channel {
@@ -59,7 +59,8 @@ pub async fn req(user: User, target: Ref, member: Ref) -> Result<()> {
         Message::create(
             "00000000000000000000000000".to_string(),
             id.clone(),
-            format!("<@{}> added <@{}> to the group.", user.id, member.id),
+            // ! FIXME: make a schema for this
+            format!("{{\"type\":\"user_added\",\"id\":\"{}\",\"by\":\"{}\"}}", member.id, user.id),
         )
         .publish(&channel)
         .await

src/routes/channels/group_remove_member.rs

@@ -20,7 +20,7 @@ pub async fn req(user: User, target: Ref, member: Ref) -> Result<()> {
     {
         if &user.id != owner {
             // figure out if we want to use perm system here
-            Err(Error::LabelMe)?
+            Err(Error::MissingPermission)?
         }
 
         if recipients.iter().find(|x| *x == &member.id).is_none() {
@@ -56,7 +56,8 @@ pub async fn req(user: User, target: Ref, member: Ref) -> Result<()> {
         Message::create(
             "00000000000000000000000000".to_string(),
             id.clone(),
-            format!("<@{}> removed <@{}> from the group.", user.id, member.id),
+            // ! FIXME: make a schema for this
+            format!("{{\"type\":\"user_remove\",\"id\":\"{}\",\"by\":\"{}\"}}", member.id, user.id),
         )
         .publish(&channel)
         .await

src/routes/channels/message_delete.rs

@@ -12,7 +12,7 @@ pub async fn req(user: User, target: Ref, msg: Ref) -> Result<()> {
         .for_channel()
         .await?;
     if !perm.get_view() {
-        Err(Error::LabelMe)?
+        Err(Error::MissingPermission)?
     }
 
     let message = msg.fetch_message(&channel).await?;

src/routes/channels/message_edit.rs

@@ -24,7 +24,7 @@ pub async fn req(user: User, target: Ref, msg: Ref, edit: Json<Data>) -> Result<
         .for_channel()
         .await?;
     if !perm.get_view() {
-        Err(Error::LabelMe)?
+        Err(Error::MissingPermission)?
     }
 
     let message = msg.fetch_message(&channel).await?;

src/routes/channels/message_fetch.rs

@@ -12,7 +12,7 @@ pub async fn req(user: User, target: Ref, msg: Ref) -> Result<JsonValue> {
         .for_channel()
         .await?;
     if !perm.get_view() {
-        Err(Error::LabelMe)?
+        Err(Error::MissingPermission)?
     }
 
     let message = msg.fetch_message(&channel).await?;

src/routes/channels/message_query.rs

@@ -34,7 +34,7 @@ pub async fn req(user: User, target: Ref, options: Form<Options>) -> Result<Json
         .for_channel()
         .await?;
     if !perm.get_view() {
-        Err(Error::LabelMe)?
+        Err(Error::MissingPermission)?
     }
 
     let mut query = doc! { "channel": target.id() };

src/routes/channels/message_query_stale.rs

@@ -14,7 +14,7 @@ pub struct Options {
 #[post("/<target>/messages/stale", data = "<data>")]
 pub async fn req(user: User, target: Ref, data: Json<Options>) -> Result<JsonValue> {
     if data.ids.len() > 150 {
-        return Err(Error::LabelMe);
+        return Err(Error::TooManyIds);
     }
 
     let target = target.fetch_channel().await?;
@@ -24,7 +24,7 @@ pub async fn req(user: User, target: Ref, data: Json<Options>) -> Result<JsonVal
         .for_channel()
         .await?;
     if !perm.get_view() {
-        Err(Error::LabelMe)?
+        Err(Error::MissingPermission)?
     }
 
     let mut cursor = get_collection("messages")

src/routes/channels/message_send.rs

@@ -12,7 +12,7 @@ use validator::Validate;
 
 #[derive(Validate, Serialize, Deserialize)]
 pub struct Data {
-    #[validate(length(min = 1, max = 2000))]
+    #[validate(length(min = 0, max = 2000))]
     content: String,
     // Maximum length of 36 allows both ULIDs and UUIDs.
     #[validate(length(min = 1, max = 36))]
@@ -26,6 +26,10 @@ pub async fn req(user: User, target: Ref, message: Json<Data>) -> Result<JsonVal
     message
         .validate()
         .map_err(|error| Error::FailedValidation { error })?;
+    
+    if message.content.len() == 0 && message.attachment.is_none() {
+        return Err(Error::EmptyMessage);
+    }
 
     let target = target.fetch_channel().await?;
 
@@ -34,7 +38,7 @@ pub async fn req(user: User, target: Ref, message: Json<Data>) -> Result<JsonVal
         .for_channel()
         .await?;
     if !perm.get_send_message() {
-        Err(Error::LabelMe)?
+        Err(Error::MissingPermission)?
     }
 
     if get_collection("messages")

src/routes/onboard/complete.rs

@@ -2,7 +2,6 @@ use crate::database::*;
 use crate::util::result::{Error, Result};
 
 use mongodb::bson::doc;
-use mongodb::options::{Collation, FindOneOptions};
 use rauth::auth::Session;
 use regex::Regex;
 use rocket_contrib::json::Json;
@@ -10,7 +9,7 @@ use serde::{Deserialize, Serialize};
 use validator::Validate;
 
 lazy_static! {
-    static ref RE_USERNAME: Regex = Regex::new(r"^[a-zA-Z0-9_]+$").unwrap();
+    static ref RE_USERNAME: Regex = Regex::new(r"^[a-zA-Z0-9_.]+$").unwrap();
 }
 
 #[derive(Validate, Serialize, Deserialize)]
@@ -28,31 +27,11 @@ pub async fn req(session: Session, user: Option<User>, data: Json<Data>) -> Resu
     data.validate()
         .map_err(|error| Error::FailedValidation { error })?;
 
-    if data.username == "revolt" {
-        Err(Error::UsernameTaken)?
+    if User::is_username_taken(&data.username).await? {
+        return Err(Error::UsernameTaken)
     }
 
-    let col = get_collection("users");
-    if col
-        .find_one(
-            doc! {
-                "username": &data.username
-            },
-            FindOneOptions::builder()
-                .collation(Collation::builder().locale("en").strength(2).build())
-                .build(),
-        )
-        .await
-        .map_err(|_| Error::DatabaseError {
-            operation: "find_one",
-            with: "user",
-        })?
-        .is_some()
-    {
-        Err(Error::UsernameTaken)?
-    }
-
-    col.insert_one(
+    get_collection("users").insert_one(
         doc! {
             "_id": session.user_id,
             "username": &data.username

src/routes/push/subscribe.rs

@@ -16,7 +16,7 @@ pub struct Subscription {
 #[post("/subscribe", data = "<data>")]
 pub async fn req(session: Session, data: Json<Subscription>) -> Result<()> {
     let data = data.into_inner();
-    let col = get_collection("accounts")
+    get_collection("accounts")
         .update_one(
             doc! {
                 "_id": session.user_id,
@@ -24,13 +24,14 @@ pub async fn req(session: Session, data: Json<Subscription>) -> Result<()> {
             },
             doc! {
                 "$set": {
-                    "sessions.$.subscription": to_document(&data).unwrap()
+                    "sessions.$.subscription": to_document(&data)
+                        .map_err(|_| Error::DatabaseError { operation: "to_document", with: "subscription" })?
                 }
             },
             None,
         )
         .await
-        .unwrap();
+        .map_err(|_| Error::DatabaseError { operation: "update_one", with: "account" })?;
 
     Ok(())
 }

src/routes/push/unsubscribe.rs

@@ -6,7 +6,7 @@ use rauth::auth::Session;
 
 #[post("/unsubscribe")]
 pub async fn req(session: Session) -> Result<()> {
-    let col = get_collection("accounts")
+    get_collection("accounts")
         .update_one(
             doc! {
                 "_id": session.user_id,
@@ -20,7 +20,10 @@ pub async fn req(session: Session) -> Result<()> {
             None,
         )
         .await
-        .unwrap();
+        .map_err(|_| Error::DatabaseError {
+            operation: "to_document",
+            with: "subscription",
+        })?;
 
     Ok(())
 }

src/routes/root.rs

@@ -9,7 +9,7 @@ use rocket_contrib::json::JsonValue;
 #[get("/")]
 pub async fn root() -> JsonValue {
     json!({
-        "revolt": "0.3.3-alpha.6",
+        "revolt": "0.3.3-alpha.8",
         "features": {
             "registration": !*DISABLE_REGISTRATION,
             "captcha": {

src/routes/users/fetch_user.rs

@@ -13,7 +13,7 @@ pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
         .await?;
 
     if !perm.get_access() {
-        Err(Error::LabelMe)?
+        Err(Error::MissingPermission)?
     }
 
     Ok(json!(target.from(&user).with(perm)))

src/routes/users/find_mutual.rs

@@ -8,15 +8,12 @@ use rocket_contrib::json::JsonValue;
 
 #[get("/<target>/mutual")]
 pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
-    let channels = get_collection("channels")
+    let users = get_collection("users")
         .find(
             doc! {
-                "$or": [
-                    { "type": "Group" },
-                ],
                 "$and": [
-                    { "recipients": &user.id },
-                    { "recipients": &target.id }
+                    { "relations._id": &user.id, "relations.status": "Friend" },
+                    { "relations._id": &target.id, "relations.status": "Friend" }
                 ]
             },
             FindOptions::builder().projection(doc! { "_id": 1 }).build(),
@@ -24,7 +21,7 @@ pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
         .await
         .map_err(|_| Error::DatabaseError {
             operation: "find",
-            with: "channels",
+            with: "users",
         })?
         .filter_map(async move |s| s.ok())
         .collect::<Vec<Document>>()
@@ -33,5 +30,5 @@ pub async fn req(user: User, target: Ref) -> Result<JsonValue> {
         .filter_map(|x| x.get_str("_id").ok().map(|x| x.to_string()))
         .collect::<Vec<String>>();
 
-    Ok(json!({ "channels": channels }))
+    Ok(json!({ "users": users }))
 }

src/util/result.rs

@@ -40,6 +40,8 @@ pub enum Error {
     UnknownAttachment,
     #[snafu(display("Cannot edit someone else's message."))]
     CannotEditMessage,
+    #[snafu(display("Cannot send empty message."))]
+    EmptyMessage,
     #[snafu(display("Cannot remove yourself from a group, use delete channel instead."))]
     CannotRemoveYourself,
     #[snafu(display("Group size is too large."))]
@@ -50,6 +52,8 @@ pub enum Error {
     NotInGroup,
 
     // ? General errors.
+    #[snafu(display("Trying to fetch too much data."))]
+    TooManyIds,
     #[snafu(display("Failed to validate fields."))]
     FailedValidation { error: ValidationErrors },
     #[snafu(display("Encountered a database error."))]
@@ -59,6 +63,8 @@ pub enum Error {
     },
     #[snafu(display("Internal server error."))]
     InternalError,
+    #[snafu(display("Missing permission."))]
+    MissingPermission,
     #[snafu(display("Operation cannot be performed on this object."))]
     InvalidOperation,
     #[snafu(display("Already created an object with this nonce."))]
@@ -88,6 +94,7 @@ impl<'r> Responder<'r, 'static> for Error {
             Error::UnknownChannel => Status::NotFound,
             Error::UnknownAttachment => Status::BadRequest,
             Error::CannotEditMessage => Status::Forbidden,
+            Error::EmptyMessage => Status::UnprocessableEntity,
             Error::CannotRemoveYourself => Status::BadRequest,
             Error::GroupTooLarge { .. } => Status::Forbidden,
             Error::AlreadyInGroup => Status::Conflict,
@@ -96,7 +103,9 @@ impl<'r> Responder<'r, 'static> for Error {
             Error::FailedValidation { .. } => Status::UnprocessableEntity,
             Error::DatabaseError { .. } => Status::InternalServerError,
             Error::InternalError => Status::InternalServerError,
+            Error::MissingPermission => Status::Forbidden,
             Error::InvalidOperation => Status::BadRequest,
+            Error::TooManyIds => Status::BadRequest,
             Error::DuplicateNonce => Status::Conflict,
             Error::NoEffect => Status::Ok,
         };

src/util/variables.rs

@@ -52,8 +52,6 @@ lazy_static! {
     // Application Logic Settings
     pub static ref MAX_GROUP_SIZE: usize =
         env::var("REVOLT_MAX_GROUP_SIZE").unwrap_or_else(|_| "50".to_string()).parse().unwrap();
-    pub static ref PUSH_LIMIT: usize =
-        env::var("REVOLT_PUSH_LIMIT").unwrap_or_else(|_| "50".to_string()).parse().unwrap();
 }
 
 pub fn preflight_checks() {