Merge branch 'ws'

# Conflicts:
#	Cargo.toml
#	src/api/core/ciphers.rs
#	src/main.rs
This commit is contained in:
Daniel García 2018-09-13 15:59:45 +02:00
commit a01fee0b9f
No known key found for this signature in database
GPG Key ID: FC8A7D14C3CD543A
12 changed files with 889 additions and 230 deletions

537
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@ -15,9 +15,18 @@ reqwest = "0.8.8"
# multipart/form-data support
multipart = "0.15.2"
# WebSockets library
ws = "0.7.8"
# MessagePack library
rmpv = "0.4.0"
# Concurrent hashmap implementation
chashmap = "2.2.0"
# A generic serialization/deserialization framework
serde = "1.0.74"
serde_derive = "1.0.74"
serde = "1.0.75"
serde_derive = "1.0.75"
serde_json = "1.0.26"
# A safe, extensible ORM and Query builder
@ -34,7 +43,7 @@ ring = { version = "= 0.11.0", features = ["rsa_signing"] }
uuid = { version = "0.6.5", features = ["v4"] }
# Date and time library for Rust
chrono = "0.4.5"
chrono = "0.4.6"
# TOTP library
oath = "0.10.2"
@ -58,14 +67,19 @@ lazy_static = "1.1.0"
num-traits = "0.2.5"
num-derive = "0.2.2"
# Email libraries
lettre = "0.8.2"
lettre_email = "0.8.2"
native-tls = "0.1.5"
fast_chemail = "0.9.5"
# Number encoding library
byteorder = "1.2.6"
[patch.crates-io]
# Make jwt use ring 0.11, to match rocket
jsonwebtoken = { path = "libs/jsonwebtoken" }
rmp = { git = 'https://github.com/dani-garcia/msgpack-rust' }
# Version 0.1.2 from crates.io lacks a commit that fixes a certificate error
u2f = { git = 'https://github.com/wisespace-io/u2f-rs', rev = '193de35093a44' }

View File

@ -76,6 +76,7 @@ RUN apt-get update && apt-get install -y\
RUN mkdir /data
VOLUME /data
EXPOSE 80
EXPOSE 3012
# Copies the files from the context (env file and web-vault)
# and the binary from the "build" stage to the current stage

View File

@ -68,6 +68,7 @@ RUN apk add \
RUN mkdir /data
VOLUME /data
EXPOSE 80
EXPOSE 3012
# Copies the files from the context (env file and web-vault)
# and the binary from the "build" stage to the current stage

View File

@ -25,6 +25,7 @@ _*Note, that this project is not associated with the [Bitwarden](https://bitward
- [Disable registration of new users](#disable-registration-of-new-users)
- [Disable invitations](#disable-invitations)
- [Enabling HTTPS](#enabling-https)
- [Enabling WebSocket notifications](#enabling-websocket-notifications)
- [Enabling U2F authentication](#enabling-u2f-authentication)
- [Changing persistent data location](#changing-persistent-data-location)
- [/data prefix:](#data-prefix)
@ -175,6 +176,37 @@ docker run -d --name bitwarden \
```
Note that you need to mount ssl files and you need to forward appropriate port.
### Enabling WebSocket notifications
*Important: This does not apply to the mobile clients, which use push notifications.*
To enable WebSockets notifications, an external reverse proxy is necessary, and it must be configured to do the following:
- Route the `/notifications/hub` endpoint to the WebSocket server, by default at port `3012`, making sure to pass the `Connection` and `Upgrade` headers.
- Route everything else, including `/notifications/hub/negotiate`, to the standard Rocket server, by default at port `80`.
- If using Docker, you may need to map both ports with the `-p` flag
An example configuration is included next for a [Caddy](https://caddyserver.com/) proxy server, and assumes the proxy is running in the same computer as `bitwarden_rs`:
```r
localhost:2015 {
# The negotiation endpoint is also proxied to Rocket
proxy /notifications/hub/negotiate 0.0.0.0:80 {
transparent
}
# Notifications redirected to the websockets server
proxy /notifications/hub 0.0.0.0:3012 {
websocket
}
# Proxy the Root directory to Rocket
proxy / 0.0.0.0:80 {
transparent
}
}
```
Note: The reason for this workaround is the lack of support for WebSockets from Rocket (though [it's a planned feature](https://github.com/SergioBenitez/Rocket/issues/90)), which forces us to launch a secondary server on a separate port.
### Enabling U2F authentication
To enable U2F authentication, you must be serving bitwarden_rs from an HTTPS domain with a valid certificate (Either using the included
HTTPS options or with a reverse proxy). We recommend using a free certificate from Let's Encrypt.

View File

@ -1,6 +1,7 @@
use std::path::Path;
use std::collections::HashSet;
use rocket::State;
use rocket::Data;
use rocket::http::ContentType;
@ -16,7 +17,7 @@ use db::models::*;
use crypto;
use api::{self, PasswordData, JsonResult, EmptyResult, JsonUpcase};
use api::{self, PasswordData, JsonResult, EmptyResult, JsonUpcase, WebSocketUsers, UpdateType};
use auth::Headers;
use CONFIG;
@ -117,22 +118,22 @@ pub struct CipherData {
}
#[post("/ciphers/admin", data = "<data>")]
fn post_ciphers_admin(data: JsonUpcase<CipherData>, headers: Headers, conn: DbConn) -> JsonResult {
fn post_ciphers_admin(data: JsonUpcase<CipherData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> JsonResult {
// TODO: Implement this correctly
post_ciphers(data, headers, conn)
post_ciphers(data, headers, conn, ws)
}
#[post("/ciphers", data = "<data>")]
fn post_ciphers(data: JsonUpcase<CipherData>, headers: Headers, conn: DbConn) -> JsonResult {
fn post_ciphers(data: JsonUpcase<CipherData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> JsonResult {
let data: CipherData = data.into_inner().data;
let mut cipher = Cipher::new(data.Type, data.Name.clone());
update_cipher_from_data(&mut cipher, data, &headers, false, &conn)?;
update_cipher_from_data(&mut cipher, data, &headers, false, &conn, &ws, UpdateType::SyncCipherCreate)?;
Ok(Json(cipher.to_json(&headers.host, &headers.user.uuid, &conn)))
}
pub fn update_cipher_from_data(cipher: &mut Cipher, data: CipherData, headers: &Headers, shared_to_collection: bool, conn: &DbConn) -> EmptyResult {
pub fn update_cipher_from_data(cipher: &mut Cipher, data: CipherData, headers: &Headers, shared_to_collection: bool, conn: &DbConn, ws: &State<WebSocketUsers>, ut: UpdateType) -> EmptyResult {
if let Some(org_id) = data.OrganizationId {
match UserOrganization::find_by_user_and_org(&headers.user.uuid, &org_id, &conn) {
None => err!("You don't have permission to add item to organization"),
@ -190,6 +191,7 @@ pub fn update_cipher_from_data(cipher: &mut Cipher, data: CipherData, headers: &
cipher.password_history = data.PasswordHistory.map(|f| f.to_string());
cipher.save(&conn);
ws.send_cipher_update(ut, &cipher, &cipher.update_users_revision(&conn));
if cipher.move_to_folder(data.FolderId, &headers.user.uuid, &conn).is_err() {
err!("Error saving the folder information")
@ -219,7 +221,7 @@ struct RelationsData {
#[post("/ciphers/import", data = "<data>")]
fn post_ciphers_import(data: JsonUpcase<ImportData>, headers: Headers, conn: DbConn) -> EmptyResult {
fn post_ciphers_import(data: JsonUpcase<ImportData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
let data: ImportData = data.into_inner().data;
// Read and create the folders
@ -243,7 +245,7 @@ fn post_ciphers_import(data: JsonUpcase<ImportData>, headers: Headers, conn: DbC
.map(|i| folders[*i].uuid.clone());
let mut cipher = Cipher::new(cipher_data.Type, cipher_data.Name.clone());
update_cipher_from_data(&mut cipher, cipher_data, &headers, false, &conn)?;
update_cipher_from_data(&mut cipher, cipher_data, &headers, false, &conn, &ws, UpdateType::SyncCipherCreate)?;
cipher.move_to_folder(folder_uuid, &headers.user.uuid.clone(), &conn).ok();
}
@ -257,22 +259,22 @@ fn post_ciphers_import(data: JsonUpcase<ImportData>, headers: Headers, conn: DbC
#[put("/ciphers/<uuid>/admin", data = "<data>")]
fn put_cipher_admin(uuid: String, data: JsonUpcase<CipherData>, headers: Headers, conn: DbConn) -> JsonResult {
put_cipher(uuid, data, headers, conn)
fn put_cipher_admin(uuid: String, data: JsonUpcase<CipherData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> JsonResult {
put_cipher(uuid, data, headers, conn, ws)
}
#[post("/ciphers/<uuid>/admin", data = "<data>")]
fn post_cipher_admin(uuid: String, data: JsonUpcase<CipherData>, headers: Headers, conn: DbConn) -> JsonResult {
post_cipher(uuid, data, headers, conn)
fn post_cipher_admin(uuid: String, data: JsonUpcase<CipherData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> JsonResult {
post_cipher(uuid, data, headers, conn, ws)
}
#[post("/ciphers/<uuid>", data = "<data>")]
fn post_cipher(uuid: String, data: JsonUpcase<CipherData>, headers: Headers, conn: DbConn) -> JsonResult {
put_cipher(uuid, data, headers, conn)
fn post_cipher(uuid: String, data: JsonUpcase<CipherData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> JsonResult {
put_cipher(uuid, data, headers, conn, ws)
}
#[put("/ciphers/<uuid>", data = "<data>")]
fn put_cipher(uuid: String, data: JsonUpcase<CipherData>, headers: Headers, conn: DbConn) -> JsonResult {
fn put_cipher(uuid: String, data: JsonUpcase<CipherData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> JsonResult {
let data: CipherData = data.into_inner().data;
let mut cipher = match Cipher::find_by_uuid(&uuid, &conn) {
@ -284,7 +286,7 @@ fn put_cipher(uuid: String, data: JsonUpcase<CipherData>, headers: Headers, conn
err!("Cipher is not write accessible")
}
update_cipher_from_data(&mut cipher, data, &headers, false, &conn)?;
update_cipher_from_data(&mut cipher, data, &headers, false, &conn, &ws, UpdateType::SyncCipherUpdate)?;
Ok(Json(cipher.to_json(&headers.host, &headers.user.uuid, &conn)))
}
@ -349,17 +351,17 @@ struct ShareCipherData {
}
#[post("/ciphers/<uuid>/share", data = "<data>")]
fn post_cipher_share(uuid: String, data: JsonUpcase<ShareCipherData>, headers: Headers, conn: DbConn) -> JsonResult {
fn post_cipher_share(uuid: String, data: JsonUpcase<ShareCipherData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> JsonResult {
let data: ShareCipherData = data.into_inner().data;
share_cipher_by_uuid(&uuid, data, &headers, &conn)
share_cipher_by_uuid(&uuid, data, &headers, &conn, &ws)
}
#[put("/ciphers/<uuid>/share", data = "<data>")]
fn put_cipher_share(uuid: String, data: JsonUpcase<ShareCipherData>, headers: Headers, conn: DbConn) -> JsonResult {
fn put_cipher_share(uuid: String, data: JsonUpcase<ShareCipherData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> JsonResult {
let data: ShareCipherData = data.into_inner().data;
share_cipher_by_uuid(&uuid, data, &headers, &conn)
share_cipher_by_uuid(&uuid, data, &headers, &conn, &ws)
}
#[derive(Deserialize)]
@ -370,7 +372,7 @@ struct ShareSelectedCipherData {
}
#[put("/ciphers/share", data = "<data>")]
fn put_cipher_share_seleted(data: JsonUpcase<ShareSelectedCipherData>, headers: Headers, conn: DbConn) -> EmptyResult {
fn put_cipher_share_seleted(data: JsonUpcase<ShareSelectedCipherData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
let mut data: ShareSelectedCipherData = data.into_inner().data;
let mut cipher_ids: Vec<String> = Vec::new();
@ -402,15 +404,16 @@ fn put_cipher_share_seleted(data: JsonUpcase<ShareSelectedCipherData>, headers:
};
match shared_cipher_data.Cipher.Id.take() {
Some(id) => share_cipher_by_uuid(&id, shared_cipher_data , &headers, &conn)?,
Some(id) => share_cipher_by_uuid(&id, shared_cipher_data , &headers, &conn, &ws)?,
None => err!("Request missing ids field")
};
}
Ok(())
}
fn share_cipher_by_uuid(uuid: &str, data: ShareCipherData, headers: &Headers, conn: &DbConn) -> JsonResult {
fn share_cipher_by_uuid(uuid: &str, data: ShareCipherData, headers: &Headers, conn: &DbConn, ws: &State<WebSocketUsers>) -> JsonResult {
let mut cipher = match Cipher::find_by_uuid(&uuid, &conn) {
Some(cipher) => {
if cipher.is_write_accessible_to_user(&headers.user.uuid, &conn) {
@ -443,7 +446,7 @@ fn share_cipher_by_uuid(uuid: &str, data: ShareCipherData, headers: &Headers, co
}
}
}
update_cipher_from_data(&mut cipher, data.Cipher, &headers, shared_to_collection, &conn)?;
update_cipher_from_data(&mut cipher, data.Cipher, &headers, shared_to_collection, &conn, &ws, UpdateType::SyncCipherUpdate)?;
Ok(Json(cipher.to_json(&headers.host, &headers.user.uuid, &conn)))
}
@ -509,53 +512,53 @@ fn post_attachment_admin(uuid: String, data: Data, content_type: &ContentType, h
}
#[post("/ciphers/<uuid>/attachment/<attachment_id>/share", format = "multipart/form-data", data = "<data>")]
fn post_attachment_share(uuid: String, attachment_id: String, data: Data, content_type: &ContentType, headers: Headers, conn: DbConn) -> JsonResult {
_delete_cipher_attachment_by_id(&uuid, &attachment_id, &headers, &conn)?;
fn post_attachment_share(uuid: String, attachment_id: String, data: Data, content_type: &ContentType, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> JsonResult {
_delete_cipher_attachment_by_id(&uuid, &attachment_id, &headers, &conn, &ws)?;
post_attachment(uuid, data, content_type, headers, conn)
}
#[post("/ciphers/<uuid>/attachment/<attachment_id>/delete-admin")]
fn delete_attachment_post_admin(uuid: String, attachment_id: String, headers: Headers, conn: DbConn) -> EmptyResult {
delete_attachment(uuid, attachment_id, headers, conn)
fn delete_attachment_post_admin(uuid: String, attachment_id: String, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
delete_attachment(uuid, attachment_id, headers, conn, ws)
}
#[post("/ciphers/<uuid>/attachment/<attachment_id>/delete")]
fn delete_attachment_post(uuid: String, attachment_id: String, headers: Headers, conn: DbConn) -> EmptyResult {
delete_attachment(uuid, attachment_id, headers, conn)
fn delete_attachment_post(uuid: String, attachment_id: String, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
delete_attachment(uuid, attachment_id, headers, conn, ws)
}
#[delete("/ciphers/<uuid>/attachment/<attachment_id>")]
fn delete_attachment(uuid: String, attachment_id: String, headers: Headers, conn: DbConn) -> EmptyResult {
_delete_cipher_attachment_by_id(&uuid, &attachment_id, &headers, &conn)
fn delete_attachment(uuid: String, attachment_id: String, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
_delete_cipher_attachment_by_id(&uuid, &attachment_id, &headers, &conn, &ws)
}
#[delete("/ciphers/<uuid>/attachment/<attachment_id>/admin")]
fn delete_attachment_admin(uuid: String, attachment_id: String, headers: Headers, conn: DbConn) -> EmptyResult {
_delete_cipher_attachment_by_id(&uuid, &attachment_id, &headers, &conn)
fn delete_attachment_admin(uuid: String, attachment_id: String, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
_delete_cipher_attachment_by_id(&uuid, &attachment_id, &headers, &conn, &ws)
}
#[post("/ciphers/<uuid>/delete")]
fn delete_cipher_post(uuid: String, headers: Headers, conn: DbConn) -> EmptyResult {
_delete_cipher_by_uuid(&uuid, &headers, &conn)
fn delete_cipher_post(uuid: String, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
_delete_cipher_by_uuid(&uuid, &headers, &conn, &ws)
}
#[post("/ciphers/<uuid>/delete-admin")]
fn delete_cipher_post_admin(uuid: String, headers: Headers, conn: DbConn) -> EmptyResult {
_delete_cipher_by_uuid(&uuid, &headers, &conn)
fn delete_cipher_post_admin(uuid: String, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
_delete_cipher_by_uuid(&uuid, &headers, &conn, &ws)
}
#[delete("/ciphers/<uuid>")]
fn delete_cipher(uuid: String, headers: Headers, conn: DbConn) -> EmptyResult {
_delete_cipher_by_uuid(&uuid, &headers, &conn)
fn delete_cipher(uuid: String, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
_delete_cipher_by_uuid(&uuid, &headers, &conn, &ws)
}
#[delete("/ciphers/<uuid>/admin")]
fn delete_cipher_admin(uuid: String, headers: Headers, conn: DbConn) -> EmptyResult {
_delete_cipher_by_uuid(&uuid, &headers, &conn)
fn delete_cipher_admin(uuid: String, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
_delete_cipher_by_uuid(&uuid, &headers, &conn, &ws)
}
#[delete("/ciphers", data = "<data>")]
fn delete_cipher_selected(data: JsonUpcase<Value>, headers: Headers, conn: DbConn) -> EmptyResult {
fn delete_cipher_selected(data: JsonUpcase<Value>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
let data: Value = data.into_inner().data;
let uuids = match data.get("Ids") {
@ -567,7 +570,7 @@ fn delete_cipher_selected(data: JsonUpcase<Value>, headers: Headers, conn: DbCon
};
for uuid in uuids {
if let error @ Err(_) = _delete_cipher_by_uuid(uuid, &headers, &conn) {
if let error @ Err(_) = _delete_cipher_by_uuid(uuid, &headers, &conn, &ws) {
return error;
};
}
@ -576,12 +579,12 @@ fn delete_cipher_selected(data: JsonUpcase<Value>, headers: Headers, conn: DbCon
}
#[post("/ciphers/delete", data = "<data>")]
fn delete_cipher_selected_post(data: JsonUpcase<Value>, headers: Headers, conn: DbConn) -> EmptyResult {
delete_cipher_selected(data, headers, conn)
fn delete_cipher_selected_post(data: JsonUpcase<Value>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
delete_cipher_selected(data, headers, conn, ws)
}
#[post("/ciphers/move", data = "<data>")]
fn move_cipher_selected(data: JsonUpcase<Value>, headers: Headers, conn: DbConn) -> EmptyResult {
fn move_cipher_selected(data: JsonUpcase<Value>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
let data = data.into_inner().data;
let folder_id = match data.get("FolderId") {
@ -627,18 +630,19 @@ fn move_cipher_selected(data: JsonUpcase<Value>, headers: Headers, conn: DbConn)
err!("Error saving the folder information")
}
cipher.save(&conn);
ws.send_cipher_update(UpdateType::SyncCipherUpdate, &cipher, &cipher.update_users_revision(&conn));
}
Ok(())
}
#[put("/ciphers/move", data = "<data>")]
fn move_cipher_selected_put(data: JsonUpcase<Value>, headers: Headers, conn: DbConn) -> EmptyResult {
move_cipher_selected(data, headers, conn)
fn move_cipher_selected_put(data: JsonUpcase<Value>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
move_cipher_selected(data, headers, conn, ws)
}
#[post("/ciphers/purge", data = "<data>")]
fn delete_all(data: JsonUpcase<PasswordData>, headers: Headers, conn: DbConn) -> EmptyResult {
fn delete_all(data: JsonUpcase<PasswordData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
let data: PasswordData = data.into_inner().data;
let password_hash = data.MasterPasswordHash;
@ -653,6 +657,9 @@ fn delete_all(data: JsonUpcase<PasswordData>, headers: Headers, conn: DbConn) ->
if cipher.delete(&conn).is_err() {
err!("Failed deleting cipher")
}
else {
ws.send_cipher_update(UpdateType::SyncCipherDelete, &cipher, &cipher.update_users_revision(&conn));
}
}
// Delete folders
@ -660,13 +667,16 @@ fn delete_all(data: JsonUpcase<PasswordData>, headers: Headers, conn: DbConn) ->
if f.delete(&conn).is_err() {
err!("Failed deleting folder")
}
else {
ws.send_folder_update(UpdateType::SyncFolderCreate, &f);
}
}
Ok(())
}
fn _delete_cipher_by_uuid(uuid: &str, headers: &Headers, conn: &DbConn) -> EmptyResult {
let cipher = match Cipher::find_by_uuid(uuid, conn) {
fn _delete_cipher_by_uuid(uuid: &str, headers: &Headers, conn: &DbConn, ws: &State<WebSocketUsers>) -> EmptyResult {
let cipher = match Cipher::find_by_uuid(&uuid, &conn) {
Some(cipher) => cipher,
None => err!("Cipher doesn't exist"),
};
@ -675,13 +685,16 @@ fn _delete_cipher_by_uuid(uuid: &str, headers: &Headers, conn: &DbConn) -> Empty
err!("Cipher can't be deleted by user")
}
match cipher.delete(conn) {
Ok(()) => Ok(()),
match cipher.delete(&conn) {
Ok(()) => {
ws.send_cipher_update(UpdateType::SyncCipherDelete, &cipher, &cipher.update_users_revision(&conn));
Ok(())
}
Err(_) => err!("Failed deleting cipher")
}
}
fn _delete_cipher_attachment_by_id(uuid: &str, attachment_id: &str, headers: &Headers, conn: &DbConn) -> EmptyResult {
fn _delete_cipher_attachment_by_id(uuid: &str, attachment_id: &str, headers: &Headers, conn: &DbConn, ws: &State<WebSocketUsers>) -> EmptyResult {
let attachment = match Attachment::find_by_id(&attachment_id, &conn) {
Some(attachment) => attachment,
None => err!("Attachment doesn't exist")
@ -702,7 +715,10 @@ fn _delete_cipher_attachment_by_id(uuid: &str, attachment_id: &str, headers: &He
// Delete attachment
match attachment.delete(&conn) {
Ok(()) => Ok(()),
Ok(()) => {
ws.send_cipher_update(UpdateType::SyncCipherDelete, &cipher, &cipher.update_users_revision(&conn));
Ok(())
}
Err(_) => err!("Deleting attachement failed")
}
}

View File

@ -1,9 +1,10 @@
use rocket::State;
use rocket_contrib::{Json, Value};
use db::DbConn;
use db::models::*;
use api::{JsonResult, EmptyResult, JsonUpcase};
use api::{JsonResult, EmptyResult, JsonUpcase, WebSocketUsers, UpdateType};
use auth::Headers;
#[get("/folders")]
@ -40,23 +41,24 @@ pub struct FolderData {
}
#[post("/folders", data = "<data>")]
fn post_folders(data: JsonUpcase<FolderData>, headers: Headers, conn: DbConn) -> JsonResult {
fn post_folders(data: JsonUpcase<FolderData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> JsonResult {
let data: FolderData = data.into_inner().data;
let mut folder = Folder::new(headers.user.uuid.clone(), data.Name);
folder.save(&conn);
ws.send_folder_update(UpdateType::SyncFolderCreate, &folder);
Ok(Json(folder.to_json()))
}
#[post("/folders/<uuid>", data = "<data>")]
fn post_folder(uuid: String, data: JsonUpcase<FolderData>, headers: Headers, conn: DbConn) -> JsonResult {
put_folder(uuid, data, headers, conn)
fn post_folder(uuid: String, data: JsonUpcase<FolderData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> JsonResult {
put_folder(uuid, data, headers, conn, ws)
}
#[put("/folders/<uuid>", data = "<data>")]
fn put_folder(uuid: String, data: JsonUpcase<FolderData>, headers: Headers, conn: DbConn) -> JsonResult {
fn put_folder(uuid: String, data: JsonUpcase<FolderData>, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> JsonResult {
let data: FolderData = data.into_inner().data;
let mut folder = match Folder::find_by_uuid(&uuid, &conn) {
@ -71,17 +73,18 @@ fn put_folder(uuid: String, data: JsonUpcase<FolderData>, headers: Headers, conn
folder.name = data.Name;
folder.save(&conn);
ws.send_folder_update(UpdateType::SyncFolderUpdate, &folder);
Ok(Json(folder.to_json()))
}
#[post("/folders/<uuid>/delete")]
fn delete_folder_post(uuid: String, headers: Headers, conn: DbConn) -> EmptyResult {
delete_folder(uuid, headers, conn)
fn delete_folder_post(uuid: String, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
delete_folder(uuid, headers, conn, ws)
}
#[delete("/folders/<uuid>")]
fn delete_folder(uuid: String, headers: Headers, conn: DbConn) -> EmptyResult {
fn delete_folder(uuid: String, headers: Headers, conn: DbConn, ws: State<WebSocketUsers>) -> EmptyResult {
let folder = match Folder::find_by_uuid(&uuid, &conn) {
Some(folder) => folder,
_ => err!("Invalid folder")
@ -93,7 +96,10 @@ fn delete_folder(uuid: String, headers: Headers, conn: DbConn) -> EmptyResult {
// Delete the actual folder entry
match folder.delete(&conn) {
Ok(()) => Ok(()),
Ok(()) => {
ws.send_folder_update(UpdateType::SyncFolderDelete, &folder);
Ok(())
}
Err(_) => err!("Failed deleting folder")
}
}

View File

@ -9,6 +9,7 @@ pub use self::icons::routes as icons_routes;
pub use self::identity::routes as identity_routes;
pub use self::web::routes as web_routes;
pub use self::notifications::routes as notifications_routes;
pub use self::notifications::{start_notification_server, WebSocketUsers, UpdateType};
use rocket::response::status::BadRequest;
use rocket_contrib::Json;

View File

@ -1,20 +1,24 @@
use rocket::Route;
use rocket_contrib::Json;
use db::DbConn;
use api::JsonResult;
use auth::Headers;
use db::DbConn;
pub fn routes() -> Vec<Route> {
routes![negotiate]
routes![negotiate, websockets_err]
}
#[get("/hub")]
fn websockets_err() -> JsonResult {
err!("'/notifications/hub' should be proxied towards the websocket server, otherwise notifications will not work. Go to the README for more info.")
}
#[post("/hub/negotiate")]
fn negotiate(_headers: Headers, _conn: DbConn) -> JsonResult {
use data_encoding::BASE64URL;
use crypto;
use data_encoding::BASE64URL;
// Store this in db?
let conn_id = BASE64URL.encode(&crypto::get_random(vec![0u8; 16]));
// TODO: Implement transports
@ -23,9 +27,338 @@ fn negotiate(_headers: Headers, _conn: DbConn) -> JsonResult {
Ok(Json(json!({
"connectionId": conn_id,
"availableTransports":[
// {"transport":"WebSockets", "transferFormats":["Text","Binary"]},
{"transport":"WebSockets", "transferFormats":["Text","Binary"]},
// {"transport":"ServerSentEvents", "transferFormats":["Text"]},
// {"transport":"LongPolling", "transferFormats":["Text","Binary"]}
]
})))
}
///
/// Websockets server
///
use std::sync::Arc;
use std::thread;
use ws::{self, util::Token, Factory, Handler, Handshake, Message, Sender, WebSocket};
use chashmap::CHashMap;
use chrono::NaiveDateTime;
use serde_json::from_str;
use db::models::{Cipher, Folder, User};
use rmpv::Value;
fn serialize(val: Value) -> Vec<u8> {
use rmpv::encode::write_value;
let mut buf = Vec::new();
write_value(&mut buf, &val).expect("Error encoding MsgPack");
// Add size bytes at the start
// Extracted from BinaryMessageFormat.js
let mut size = buf.len();
let mut len_buf: Vec<u8> = Vec::new();
loop {
let mut size_part = size & 0x7f;
size = size >> 7;
if size > 0 {
size_part = size_part | 0x80;
}
len_buf.push(size_part as u8);
if size <= 0 {
break;
}
}
len_buf.append(&mut buf);
len_buf
}
fn serialize_date(date: NaiveDateTime) -> Value {
let seconds: i64 = date.timestamp();
let nanos: i64 = date.timestamp_subsec_nanos() as i64;
let timestamp = nanos << 34 | seconds;
use byteorder::{BigEndian, WriteBytesExt};
let mut bs = [0u8; 8];
bs.as_mut()
.write_i64::<BigEndian>(timestamp)
.expect("Unable to write");
// -1 is Timestamp
// https://github.com/msgpack/msgpack/blob/master/spec.md#timestamp-extension-type
Value::Ext(-1, bs.to_vec())
}
fn convert_option<T: Into<Value>>(option: Option<T>) -> Value {
match option {
Some(a) => a.into(),
None => Value::Nil,
}
}
// Server WebSocket handler
pub struct WSHandler {
out: Sender,
user_uuid: Option<String>,
users: WebSocketUsers,
}
const RECORD_SEPARATOR: u8 = 0x1e;
const INITIAL_RESPONSE: [u8; 3] = [0x7b, 0x7d, RECORD_SEPARATOR]; // {, }, <RS>
#[derive(Deserialize)]
struct InitialMessage {
protocol: String,
version: i32,
}
const PING_MS: u64 = 15_000;
const PING: Token = Token(1);
impl Handler for WSHandler {
fn on_open(&mut self, hs: Handshake) -> ws::Result<()> {
// TODO: Improve this split
let path = hs.request.resource();
let mut query_split: Vec<_> = path.split("?").nth(1).unwrap().split("&").collect();
query_split.sort();
let access_token = &query_split[0][13..];
let _id = &query_split[1][3..];
// Validate the user
use auth;
let claims = match auth::decode_jwt(access_token) {
Ok(claims) => claims,
Err(_) => {
return Err(ws::Error::new(
ws::ErrorKind::Internal,
"Invalid access token provided",
))
}
};
// Assign the user to the handler
let user_uuid = claims.sub;
self.user_uuid = Some(user_uuid.clone());
// Add the current Sender to the user list
let handler_insert = self.out.clone();
let handler_update = self.out.clone();
self.users.map.upsert(
user_uuid,
|| vec![handler_insert],
|ref mut v| v.push(handler_update),
);
// Schedule a ping to keep the connection alive
self.out.timeout(PING_MS, PING)
}
fn on_message(&mut self, msg: Message) -> ws::Result<()> {
println!("Server got message '{}'. ", msg);
if let Message::Text(text) = msg.clone() {
let json = &text[..text.len() - 1]; // Remove last char
if let Ok(InitialMessage { protocol, version }) = from_str::<InitialMessage>(json) {
if &protocol == "messagepack" && version == 1 {
return self.out.send(&INITIAL_RESPONSE[..]); // Respond to initial message
}
}
}
// If it's not the initial message, just echo the message
self.out.send(msg)
}
fn on_timeout(&mut self, event: Token) -> ws::Result<()> {
if event == PING {
// send ping
self.out.send(create_ping())?;
// reschedule the timeout
self.out.timeout(PING_MS, PING)
} else {
Err(ws::Error::new(
ws::ErrorKind::Internal,
"Invalid timeout token provided",
))
}
}
}
struct WSFactory {
pub users: WebSocketUsers,
}
impl WSFactory {
pub fn init() -> Self {
WSFactory {
users: WebSocketUsers {
map: Arc::new(CHashMap::new()),
},
}
}
}
impl Factory for WSFactory {
type Handler = WSHandler;
fn connection_made(&mut self, out: Sender) -> Self::Handler {
println!("WS: Connection made");
WSHandler {
out,
user_uuid: None,
users: self.users.clone(),
}
}
fn connection_lost(&mut self, handler: Self::Handler) {
println!("WS: Connection lost");
// Remove handler
let user_uuid = &handler.user_uuid.unwrap();
if let Some(mut user_conn) = self.users.map.get_mut(user_uuid) {
user_conn.remove_item(&handler.out);
}
}
}
#[derive(Clone)]
pub struct WebSocketUsers {
pub map: Arc<CHashMap<String, Vec<Sender>>>,
}
impl WebSocketUsers {
fn send_update(&self, user_uuid: &String, data: Vec<u8>) -> ws::Result<()> {
if let Some(user) = self.map.get(user_uuid) {
for sender in user.iter() {
sender.send(data.clone())?;
}
}
Ok(())
}
// NOTE: The last modified date needs to be updated before calling these methods
pub fn send_user_update(&self, ut: UpdateType, user: &User) {
let data = create_update(
vec![
("UserId".into(), user.uuid.clone().into()),
("Date".into(), serialize_date(user.updated_at)),
].into(),
ut,
);
self.send_update(&user.uuid.clone(), data).ok();
}
pub fn send_folder_update(&self, ut: UpdateType, folder: &Folder) {
let data = create_update(
vec![
("Id".into(), folder.uuid.clone().into()),
("UserId".into(), folder.user_uuid.clone().into()),
("RevisionDate".into(), serialize_date(folder.updated_at)),
].into(),
ut,
);
self.send_update(&folder.user_uuid, data).ok();
}
pub fn send_cipher_update(&self, ut: UpdateType, cipher: &Cipher, user_uuids: &Vec<String>) {
let user_uuid = convert_option(cipher.user_uuid.clone());
let org_uuid = convert_option(cipher.organization_uuid.clone());
let data = create_update(
vec![
("Id".into(), cipher.uuid.clone().into()),
("UserId".into(), user_uuid),
("OrganizationId".into(), org_uuid),
("CollectionIds".into(), Value::Nil),
("RevisionDate".into(), serialize_date(cipher.updated_at)),
].into(),
ut,
);
for uuid in user_uuids {
self.send_update(&uuid, data.clone()).ok();
}
}
}
/* Message Structure
[
1, // MessageType.Invocation
{}, // Headers
null, // InvocationId
"ReceiveMessage", // Target
[ // Arguments
{
"ContextId": "app_id",
"Type": ut as i32,
"Payload": {}
}
]
]
*/
fn create_update(payload: Vec<(Value, Value)>, ut: UpdateType) -> Vec<u8> {
use rmpv::Value as V;
let value = V::Array(vec![
1.into(),
V::Array(vec![]),
V::Nil,
"ReceiveMessage".into(),
V::Array(vec![V::Map(vec![
("ContextId".into(), "app_id".into()),
("Type".into(), (ut as i32).into()),
("Payload".into(), payload.into()),
])]),
]);
serialize(value)
}
fn create_ping() -> Vec<u8> {
serialize(Value::Array(vec![6.into()]))
}
#[allow(dead_code)]
pub enum UpdateType {
SyncCipherUpdate = 0,
SyncCipherCreate = 1,
SyncLoginDelete = 2,
SyncFolderDelete = 3,
SyncCiphers = 4,
SyncVault = 5,
SyncOrgKeys = 6,
SyncFolderCreate = 7,
SyncFolderUpdate = 8,
SyncCipherDelete = 9,
SyncSettings = 10,
LogOut = 11,
}
pub fn start_notification_server() -> WebSocketUsers {
let factory = WSFactory::init();
let users = factory.users.clone();
thread::spawn(move || {
WebSocket::new(factory)
.unwrap()
.listen("0.0.0.0:3012")
.unwrap();
});
users
}

View File

@ -130,19 +130,25 @@ impl Cipher {
json_object
}
pub fn update_users_revision(&self, conn: &DbConn) {
pub fn update_users_revision(&self, conn: &DbConn) -> Vec<String> {
let mut user_uuids = Vec::new();
match self.user_uuid {
Some(ref user_uuid) => User::update_uuid_revision(&user_uuid, conn),
Some(ref user_uuid) => {
User::update_uuid_revision(&user_uuid, conn);
user_uuids.push(user_uuid.clone())
},
None => { // Belongs to Organization, need to update affected users
if let Some(ref org_uuid) = self.organization_uuid {
UserOrganization::find_by_cipher_and_org(&self.uuid, &org_uuid, conn)
.iter()
.for_each(|user_org| {
User::update_uuid_revision(&user_org.user_uuid, conn)
User::update_uuid_revision(&user_org.user_uuid, conn);
user_uuids.push(user_org.user_uuid.clone())
});
}
}
};
user_uuids
}
pub fn save(&mut self, conn: &DbConn) -> bool {
@ -157,7 +163,7 @@ impl Cipher {
}
}
pub fn delete(self, conn: &DbConn) -> QueryResult<()> {
pub fn delete(&self, conn: &DbConn) -> QueryResult<()> {
self.update_users_revision(conn);
FolderCipher::delete_all_by_cipher(&self.uuid, &conn)?;
@ -166,7 +172,7 @@ impl Cipher {
diesel::delete(
ciphers::table.filter(
ciphers::uuid.eq(self.uuid)
ciphers::uuid.eq(&self.uuid)
)
).execute(&**conn).and(Ok(()))
}

View File

@ -82,13 +82,13 @@ impl Folder {
}
}
pub fn delete(self, conn: &DbConn) -> QueryResult<()> {
pub fn delete(&self, conn: &DbConn) -> QueryResult<()> {
User::update_uuid_revision(&self.user_uuid, conn);
FolderCipher::delete_all_by_folder(&self.uuid, &conn)?;
diesel::delete(
folders::table.filter(
folders::uuid.eq(self.uuid)
folders::uuid.eq(&self.uuid)
)
).execute(&**conn).and(Ok(()))
}

View File

@ -1,10 +1,13 @@
#![feature(plugin, custom_derive)]
#![feature(plugin, custom_derive, vec_remove_item)]
#![plugin(rocket_codegen)]
#![allow(proc_macro_derive_resolution_fallback)] // TODO: Remove this when diesel update fixes warnings
extern crate rocket;
extern crate rocket_contrib;
extern crate reqwest;
extern crate multipart;
extern crate ws;
extern crate rmpv;
extern crate chashmap;
extern crate serde;
#[macro_use]
extern crate serde_derive;
@ -31,6 +34,7 @@ extern crate lettre;
extern crate lettre_email;
extern crate native_tls;
extern crate fast_chemail;
extern crate byteorder;
use std::{env, path::Path, process::{exit, Command}};
use rocket::Rocket;
@ -52,6 +56,7 @@ fn init_rocket() -> Rocket {
.mount("/icons", api::icons_routes())
.mount("/notifications", api::notifications_routes())
.manage(db::init_pool())
.manage(api::start_notification_server())
}
// Embed the migrations from the migrations folder into the application
@ -76,7 +81,6 @@ fn main() {
check_web_vault();
migrations::run_migrations();
init_rocket().launch();
}