Support passing access tokens via Sec-WebSocket-Protocol header

Previously, the access token needed to be passed via the query string;
with this commit, the token can be passed *either* through the query
string or the Sec-WebSocket-Protocol header.

This was done to correspond to the changes made to the streaming.js
version in [Improve streaming server security](mastodon/mastodon#10818).
However, I am not sure that it *does* increase security; as explained
at <https://support.ably.io/support/solutions/articles/3000075120-is-it-secure-to-send-the-access-token-as-part-of-the-websocket-url-query-params->,
there is generally no security advantage to passing sensitive information
via websocket headers instead of the query string—the entire connection
is encrypted and is not stored in the browser history, so the typical
reasons to keep sensitive info out of the query string don't apply.

I would welcome any corrections on this/reasons this change improves
security.

Author: codesections

src/main.rs

@@ -40,7 +40,7 @@ use receiver::Receiver;
 use std::env;
 use std::net::SocketAddr;
 use stream::StreamManager;
-use user::{Method, Scope, User};
+use user::{Scope, User};
 use warp::path;
 use warp::Filter as WarpFilter;
 
@@ -96,7 +96,7 @@ fn main() {
 
     //let redis_updates_ws = StreamManager::new(Receiver::new());
     let websocket = path!("api" / "v1" / "streaming")
-        .and(Scope::Public.get_access_token(Method::WS))
+        .and(Scope::Public.get_access_token())
         .and_then(|token| User::from_access_token(token, Scope::Public))
         .and(warp::query())
         .and(query::Media::to_filter())
@@ -136,18 +136,16 @@ fn main() {
                     // Other endpoints don't exist:
                     _ => return Err(warp::reject::custom("Error: Nonexistent WebSocket query")),
                 };
+                let token = user.access_token.clone();
                 let stream = redis_updates_ws.configure_copy(&timeline, user);
 
-                Ok(ws.on_upgrade(move |socket| ws::send_replies(socket, stream)))
+                Ok((
+                    ws.on_upgrade(move |socket| ws::send_replies(socket, stream)),
+                    token,
+                ))
             },
         )
-        .map(|reply| {
-            warp::reply::with_header(
-                reply,
-                "sec-websocket-protocol",
-                "LhbVOxKckgqyMg3nDLaEu5vgqY6Yzc9Pk1w8_yKQwS8",
-            )
-        });
+        .map(|(reply, token)| warp::reply::with_header(reply, "sec-websocket-protocol", token));
 
     let address: SocketAddr = env::var("SERVER_ADDR")
         .unwrap_or("127.0.0.1:4000".to_owned())

src/stream.rs

@@ -51,7 +51,10 @@ impl Stream for StreamManager {
     type Error = Error;
 
     fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-        let mut receiver = self.receiver.lock().expect("No other thread panic");
+        let mut receiver = self
+            .receiver
+            .lock()
+            .expect("StreamManager: No other thread panic");
         receiver.update(self.id, &self.target_timeline.clone());
         match receiver.poll() {
             Ok(Async::Ready(Some(value))) => {
@@ -61,19 +64,19 @@ impl Stream for StreamManager {
                     .expect("Previously set current user");
 
                 let user_langs = user.langs.clone();
-                let copy = value.clone();
-                let event = copy["event"].as_str().expect("Redis string");
-                let copy = value.clone();
-                let payload = copy["payload"].to_string();
-                let copy = value.clone();
-                let toot_lang = copy["payload"]["language"]
-                    .as_str()
-                    .expect("redis str")
-                    .to_string();
+                let event = value["event"].as_str().expect("Redis string");
+                let payload = value["payload"].to_string();
 
                 match (&user.filter, user_langs) {
                     (Filter::Notification, _) if event != "notification" => Ok(Async::NotReady),
-                    (Filter::Language, Some(ref langs)) if !langs.contains(&toot_lang) => {
+                    (Filter::Language, Some(ref user_langs))
+                        if !user_langs.contains(
+                            &value["payload"]["language"]
+                                .as_str()
+                                .expect("Redis str")
+                                .to_string(),
+                        ) =>
+                    {
                         Ok(Async::NotReady)
                     }
                     _ => Ok(Async::Ready(Some(json!(

src/timeline.rs

@@ -1,6 +1,6 @@
 //! Filters for all the endpoints accessible for Server Sent Event updates
 use crate::query;
-use crate::user::{Method, Scope, User};
+use crate::user::{Scope, User};
 use warp::filters::BoxedFilter;
 use warp::{path, Filter};
 
@@ -14,7 +14,7 @@ type TimelineUser = ((String, User),);
 pub fn user() -> BoxedFilter<TimelineUser> {
     path!("api" / "v1" / "streaming" / "user")
         .and(path::end())
-        .and(Scope::Private.get_access_token(Method::HttpPush))
+        .and(Scope::Private.get_access_token())
         .and_then(|token| User::from_access_token(token, Scope::Private))
         .map(|user: User| (user.id.to_string(), user))
         .boxed()
@@ -30,7 +30,7 @@ pub fn user() -> BoxedFilter<TimelineUser> {
 pub fn user_notifications() -> BoxedFilter<TimelineUser> {
     path!("api" / "v1" / "streaming" / "user" / "notification")
         .and(path::end())
-        .and(Scope::Private.get_access_token(Method::HttpPush))
+        .and(Scope::Private.get_access_token())
         .and_then(|token| User::from_access_token(token, Scope::Private))
         .map(|user: User| (user.id.to_string(), user.with_notification_filter()))
         .boxed()
@@ -43,7 +43,7 @@ pub fn user_notifications() -> BoxedFilter<TimelineUser> {
 pub fn public() -> BoxedFilter<TimelineUser> {
     path!("api" / "v1" / "streaming" / "public")
         .and(path::end())
-        .and(Scope::Public.get_access_token(Method::HttpPush))
+        .and(Scope::Public.get_access_token())
         .and_then(|token| User::from_access_token(token, Scope::Public))
         .map(|user: User| ("public".to_owned(), user.with_language_filter()))
         .boxed()
@@ -56,7 +56,7 @@ pub fn public() -> BoxedFilter<TimelineUser> {
 pub fn public_media() -> BoxedFilter<TimelineUser> {
     path!("api" / "v1" / "streaming" / "public")
         .and(path::end())
-        .and(Scope::Public.get_access_token(Method::HttpPush))
+        .and(Scope::Public.get_access_token())
         .and_then(|token| User::from_access_token(token, Scope::Public))
         .and(warp::query())
         .map(|user: User, q: query::Media| match q.only_media.as_ref() {
@@ -73,7 +73,7 @@ pub fn public_media() -> BoxedFilter<TimelineUser> {
 pub fn public_local() -> BoxedFilter<TimelineUser> {
     path!("api" / "v1" / "streaming" / "public" / "local")
         .and(path::end())
-        .and(Scope::Public.get_access_token(Method::HttpPush))
+        .and(Scope::Public.get_access_token())
         .and_then(|token| User::from_access_token(token, Scope::Public))
         .map(|user: User| ("public:local".to_owned(), user.with_language_filter()))
         .boxed()
@@ -85,7 +85,7 @@ pub fn public_local() -> BoxedFilter<TimelineUser> {
 /// **public**.  Filter: `Language`
 pub fn public_local_media() -> BoxedFilter<TimelineUser> {
     path!("api" / "v1" / "streaming" / "public" / "local")
-        .and(Scope::Public.get_access_token(Method::HttpPush))
+        .and(Scope::Public.get_access_token())
         .and_then(|token| User::from_access_token(token, Scope::Public))
         .and(warp::query())
         .and(path::end())
@@ -103,7 +103,7 @@ pub fn public_local_media() -> BoxedFilter<TimelineUser> {
 pub fn direct() -> BoxedFilter<TimelineUser> {
     path!("api" / "v1" / "streaming" / "direct")
         .and(path::end())
-        .and(Scope::Private.get_access_token(Method::HttpPush))
+        .and(Scope::Private.get_access_token())
         .and_then(|token| User::from_access_token(token, Scope::Private))
         .map(|user: User| (format!("direct:{}", user.id), user.with_no_filter()))
         .boxed()
@@ -139,7 +139,7 @@ pub fn hashtag_local() -> BoxedFilter<TimelineUser> {
 /// **private**.  Filter: `None`
 pub fn list() -> BoxedFilter<TimelineUser> {
     path!("api" / "v1" / "streaming" / "list")
-        .and(Scope::Private.get_access_token(Method::HttpPush))
+        .and(Scope::Private.get_access_token())
         .and_then(|token| User::from_access_token(token, Scope::Private))
         .and(warp::query())
         .and_then(|user: User, q: query::List| (user.authorized_for_list(q.list), Ok(user)))

src/user.rs

@@ -50,7 +50,6 @@ LIMIT 1",
                 &[&token],
             )
             .expect("Hard-coded query will return Some([0 or more rows])");
-        dbg!(&result);
         if !result.is_empty() {
             let only_row = result.get(0);
             let id: i64 = only_row.get(1);
@@ -133,41 +132,27 @@ pub enum Scope {
     Public,
     Private,
 }
-pub enum Method {
-    WS,
-    HttpPush,
-}
 impl Scope {
-    pub fn get_access_token(self, method: Method) -> warp::filters::BoxedFilter<(String,)> {
-        let token_from_header_http_push =
-            warp::header::header::<String>("authorization").map(|auth: String| {
-                dbg!(auth.split(' ').nth(1).unwrap_or("invalid").to_string());
-                auth.split(' ').nth(1).unwrap_or("invalid").to_string()
-            });
+    pub fn get_access_token(self) -> warp::filters::BoxedFilter<(String,)> {
+        let token_from_header_http_push = warp::header::header::<String>("authorization")
+            .map(|auth: String| auth.split(' ').nth(1).unwrap_or("invalid").to_string());
         let token_from_header_ws =
-            warp::header::header::<String>("Sec-WebSocket-Protocol").map(|auth: String| {
-                dbg!(&auth);
-                auth
-            });
-        let token_from_query = warp::query().map(|q: query::Auth| {
-            dbg!(&q.access_token);
-            q.access_token
-        });
+            warp::header::header::<String>("Sec-WebSocket-Protocol").map(|auth: String| auth);
+        let token_from_query = warp::query().map(|q: query::Auth| q.access_token);
+
+        let private_scopes = any_of!(
+            token_from_header_http_push,
+            token_from_header_ws,
+            token_from_query
+        );
+
         let public = warp::any().map(|| "no access token".to_string());
 
-        match (self, method) {
+        match self {
             // if they're trying to access a private scope without an access token, reject the request
-            (Scope::Private, Method::HttpPush) => {
-                any_of!(token_from_query, token_from_header_http_push).boxed()
-            }
-            (Scope::Private, Method::WS) => any_of!(token_from_query, token_from_header_ws).boxed(),
+            Scope::Private => private_scopes.boxed(),
             // if they're trying to access a public scope without an access token, proceed
-            (Scope::Public, Method::HttpPush) => {
-                any_of!(token_from_query, token_from_header_http_push, public).boxed()
-            }
-            (Scope::Public, Method::WS) => {
-                any_of!(token_from_query, token_from_header_ws, public).boxed()
-            }
+            Scope::Public => any_of!(private_scopes, public).boxed(),
         }
     }
 }