#13 authenticate websockets

lib/src/authentication.rs

@@ -1,3 +1,5 @@
+//! Check signatures in authentication headers, find the correct agent. Authorization is done in Hierarchies
+
 use crate::{commit::check_timestamp, errors::AtomicResult, Storelike};
 
 /// Set of values extracted from the request.

lib/src/plugins/versioning.rs

@@ -48,6 +48,7 @@ fn handle_version_request(
 fn handle_all_versions_request(
     url: url::Url,
     store: &impl Storelike,
+    // TODO: implement auth
     for_agent: Option<String>,
 ) -> AtomicResult<Resource> {
     let params = url.query_pairs();

server/src/actor_messages.rs

@@ -14,6 +14,7 @@ pub struct WsMessage(pub String);
 pub struct Subscribe {
     pub addr: Addr<crate::handlers::web_sockets::WebSocketConnection>,
     pub subject: String,
+    pub agent: String,
 }
 
 /// A message containing a Resource, which should be sent to subscribers

server/src/appstate.rs

@@ -7,7 +7,8 @@ use atomic_lib::{
     Storelike,
 };
 
-/// Context for the server (not an individual request).
+/// Data object available to handlers and actors.
+/// Contains the store, configuration and addresses for Actix Actors.
 // This struct is cloned accross all threads, so make sure the fields are thread safe.
 // A good option here is to use Actors for things that can change (e.g. commit_monitor)
 #[derive(Clone)]

server/src/commit_monitor.rs

@@ -12,11 +12,12 @@ use actix::{
     prelude::{Actor, Context, Handler},
     Addr,
 };
-use atomic_lib::Db;
+use atomic_lib::{Db, Storelike};
 use chrono::Local;
 use std::collections::{HashMap, HashSet};
 
 /// The Commit Monitor is an Actor that manages subscriptions for subjects and sends Commits to listeners.
+/// It's also responsible for checking whether the rights are present
 pub struct CommitMonitor {
     /// Maintains a list of all the resources that are being subscribed to, and maps these to websocket connections.
     subscriptions: HashMap<String, HashSet<Addr<WebSocketConnection>>>,
@@ -34,15 +35,31 @@ impl Actor for CommitMonitor {
 impl Handler<Subscribe> for CommitMonitor {
     type Result = ();
 
+    // A message comes in when a client subscribes to a subject.
     fn handle(&mut self, msg: Subscribe, _: &mut Context<Self>) {
-        let mut set = if let Some(set) = self.subscriptions.get(&msg.subject) {
-            set.clone()
-        } else {
-            HashSet::new()
-        };
-        set.insert(msg.addr);
-        log::info!("handle subscribe {} ", msg.subject);
-        self.subscriptions.insert(msg.subject, set);
+        // check if the agent has the rights to subscribe to this resource
+        if let Ok(resource) = self.store.get_resource(&msg.subject) {
+            if let Ok(can) =
+                atomic_lib::hierarchy::check_read(&self.store, &resource, msg.agent.clone())
+            {
+                if can {
+                    let mut set = if let Some(set) = self.subscriptions.get(&msg.subject) {
+                        set.clone()
+                    } else {
+                        HashSet::new()
+                    };
+                    set.insert(msg.addr);
+                    log::info!("handle subscribe {} ", msg.subject);
+                    self.subscriptions.insert(msg.subject.clone(), set);
+                }
+                log::info!(
+                    "Not allowed {}  to subscribe to {} ",
+                    &msg.agent,
+                    &msg.subject
+                );
+            }
+            // TODO: Handle errors
+        }
     }
 }
 

server/src/handlers/tpf.rs

@@ -20,8 +20,12 @@ pub async fn tpf(
     req: actix_web::HttpRequest,
     query: web::Query<TpfQuery>,
 ) -> AtomicServerResult<HttpResponse> {
-    let mut context = data.lock().unwrap();
-    let store = &mut context.store;
+    let appstate = data.lock().unwrap();
+    let store = &appstate.store;
+
+    if !appstate.config.opts.public_mode {
+        return Err("/tpf endpoint is only available on public mode".into());
+    }
     // This is how locally items are stored (which don't know their full subject URL) in Atomic Data
     let mut builder = HttpResponse::Ok();
     let content_type = get_accept(req.headers());

server/src/handlers/web_sockets.rs

@@ -1,26 +1,38 @@
 use actix::{Actor, ActorContext, Addr, AsyncContext, Handler, StreamHandler};
-use actix_web::{web, Error, HttpRequest, HttpResponse};
+use actix_web::{web, HttpRequest, HttpResponse};
 use actix_web_actors::ws::{self};
 use std::{
     sync::Mutex,
     time::{Duration, Instant},
 };
 
-use crate::{actor_messages::CommitMessage, appstate::AppState, commit_monitor::CommitMonitor};
+use crate::{
+    actor_messages::CommitMessage, appstate::AppState, commit_monitor::CommitMonitor,
+    errors::AtomicServerResult, helpers::get_auth_headers,
+};
 
 /// Get an HTTP request, upgrade it to a Websocket connection
 pub async fn web_socket_handler(
     req: HttpRequest,
     stream: web::Payload,
     data: web::Data<Mutex<AppState>>,
-) -> Result<HttpResponse, Error> {
+) -> AtomicServerResult<HttpResponse> {
     log::info!("Starting websocket");
     let context = data.lock().unwrap();
-    ws::start(
-        WebSocketConnection::new(context.commit_monitor.clone()),
+
+    // Authentication check. If the user has no headers, continue with the Public Agent.
+    let auth_header_values = get_auth_headers(req.headers(), "ws".into())?;
+    let for_agent = atomic_lib::authentication::get_agent_from_headers_and_check(
+        auth_header_values,
+        &context.store,
+    )?;
+
+    let result = ws::start(
+        WebSocketConnection::new(context.commit_monitor.clone(), for_agent),
         &req,
         stream,
-    )
+    )?;
+    Ok(result)
 }
 
 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
@@ -37,6 +49,9 @@ pub struct WebSocketConnection {
     subscribed: std::collections::HashSet<String>,
     /// The CommitMonitor Actor that receives and sends messages for Commits
     commit_monitor_addr: Addr<CommitMonitor>,
+    /// The Agent who is connected.
+    /// If it's not specified, it's the Public Agent.
+    agent: String,
 }
 
 impl Actor for WebSocketConnection {
@@ -68,6 +83,7 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WebSocketConnecti
                                 .do_send(crate::actor_messages::Subscribe {
                                     addr: ctx.address(),
                                     subject: subject.to_string(),
+                                    agent: self.agent.clone(),
                                 });
                             self.subscribed.insert(subject.into());
                         } else {
@@ -82,6 +98,12 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WebSocketConnecti
                             ctx.text("ERROR: UNSUBSCRIBE without subject")
                         }
                     }
+                    s if s.starts_with("GET ") => {
+                        let mut parts = s.split("GET ");
+                        if let Some(_subject) = parts.nth(1) {
+                            ctx.text("GET not yet supported, see https://github.com/joepio/atomic-data-rust/issues/180")
+                        }
+                    }
                     other => {
                         log::warn!("Unmatched message: {}", other);
                         ctx.text(format!("Server receieved unknown message: {}", other));
@@ -99,12 +121,13 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WebSocketConnecti
 }
 
 impl WebSocketConnection {
-    fn new(commit_monitor_addr: Addr<CommitMonitor>) -> Self {
+    fn new(commit_monitor_addr: Addr<CommitMonitor>, agent: String) -> Self {
         Self {
             hb: Instant::now(),
             // Maybe this should be stored only in the CommitMonitor, and not here.
             subscribed: std::collections::HashSet::new(),
             commit_monitor_addr,
+            agent,
         }
     }