Merge pull request parallaxsecond#207 from joechrisellis/master

Create `Connection` abstraction for client communication

src/authenticators/direct_authenticator/mod.rs

@@ -10,6 +10,7 @@
 
 use super::ApplicationName;
 use super::Authenticate;
+use crate::front::listener::ConnectionMetadata;
 use log::error;
 use parsec_interface::requests::request::RequestAuth;
 use parsec_interface::requests::{ResponseStatus, Result};
@@ -20,7 +21,11 @@ use std::str;
 pub struct DirectAuthenticator;
 
 impl Authenticate for DirectAuthenticator {
-    fn authenticate(&self, auth: &RequestAuth) -> Result<ApplicationName> {
+    fn authenticate(
+        &self,
+        auth: &RequestAuth,
+        _: Option<ConnectionMetadata>,
+    ) -> Result<ApplicationName> {
         if auth.buffer.expose_secret().is_empty() {
             error!("The direct authenticator does not expect empty authentication values.");
             Err(ResponseStatus::AuthenticationError)
@@ -49,9 +54,10 @@ mod test {
 
         let app_name = "app_name".to_string();
         let req_auth = RequestAuth::new(app_name.clone().into_bytes());
+        let conn_metadata = None;
 
         let auth_name = authenticator
-            .authenticate(&req_auth)
+            .authenticate(&req_auth, conn_metadata)
             .expect("Failed to authenticate");
 
         assert_eq!(auth_name.get_name(), app_name);
@@ -60,8 +66,9 @@ mod test {
     #[test]
     fn failed_authentication() {
         let authenticator = DirectAuthenticator {};
+        let conn_metadata = None;
         let status = authenticator
-            .authenticate(&RequestAuth::new(vec![0xff; 5]))
+            .authenticate(&RequestAuth::new(vec![0xff; 5]), conn_metadata)
             .expect_err("Authentication should have failed");
 
         assert_eq!(status, ResponseStatus::AuthenticationError);
@@ -70,8 +77,9 @@ mod test {
     #[test]
     fn empty_auth() {
         let authenticator = DirectAuthenticator {};
+        let conn_metadata = None;
         let status = authenticator
-            .authenticate(&RequestAuth::new(Vec::new()))
+            .authenticate(&RequestAuth::new(Vec::new()), conn_metadata)
             .expect_err("Empty auth should have failed");
 
         assert_eq!(status, ResponseStatus::AuthenticationError);

src/authenticators/mod.rs

@@ -13,6 +13,7 @@
 
 pub mod direct_authenticator;
 
+use crate::front::listener::ConnectionMetadata;
 use parsec_interface::requests::request::RequestAuth;
 use parsec_interface::requests::Result;
 
@@ -24,12 +25,19 @@ pub struct ApplicationName(String);
 ///
 /// Interface that must be implemented for each authentication type available for the service.
 pub trait Authenticate {
-    /// Authenticates a `RequestAuth` payload and returns the `ApplicationName` if successfull.
+    /// Authenticates a `RequestAuth` payload and returns the `ApplicationName` if successful. A
+    /// optional `ConnectionMetadata` object is passed in too, since it is sometimes possible to
+    /// perform authentication based on the connection's metadata (i.e. as is the case for UNIX
+    /// domain sockets with peer credentials).
     ///
     /// # Errors
     ///
     /// If the authentification fails, returns a `ResponseStatus::AuthenticationError`.
-    fn authenticate(&self, auth: &RequestAuth) -> Result<ApplicationName>;
+    fn authenticate(
+        &self,
+        auth: &RequestAuth,
+        meta: Option<ConnectionMetadata>,
+    ) -> Result<ApplicationName>;
 }
 
 impl ApplicationName {

src/bin/main.rs

@@ -125,10 +125,10 @@ fn main() -> Result<()> {
             info!("Parsec configuration reloaded.");
         }
 
-        if let Some(stream) = listener.accept() {
+        if let Some(connection) = listener.accept() {
             let front_end_handler = front_end_handler.clone();
             threadpool.execute(move || {
-                front_end_handler.handle_request(stream);
+                front_end_handler.handle_request(connection);
                 trace!("handle_request egress");
             });
         } else {

src/front/domain_socket.rs

@@ -5,8 +5,8 @@
 //! Expose Parsec functionality using Unix domain sockets as an IPC layer.
 //! The local socket is created at a predefined location.
 use super::listener;
+use listener::Connection;
 use listener::Listen;
-use listener::ReadWrite;
 use log::error;
 use std::fs;
 use std::fs::Permissions;
@@ -91,7 +91,7 @@ impl Listen for DomainSocketListener {
         self.timeout = duration;
     }
 
-    fn accept(&self) -> Option<Box<dyn ReadWrite + Send>> {
+    fn accept(&self) -> Option<Connection> {
         let stream_result = self.listener.accept();
         match stream_result {
             Ok((stream, _)) => {
@@ -105,7 +105,12 @@ impl Listen for DomainSocketListener {
                     format_error!("Failed to set stream as blocking", err);
                     None
                 } else {
-                    Some(Box::from(stream))
+                    Some(Connection {
+                        stream: Box::new(stream),
+                        // TODO: when possible, we want to replace this with the (uid, gid, pid)
+                        // triple for peer credentials. See listener.rs.
+                        metadata: None,
+                    })
                 }
             }
             Err(err) => {

src/front/front_end.rs

@@ -6,14 +6,14 @@
 //! pass them to the rest of the service and write the responses back.
 use crate::authenticators::Authenticate;
 use crate::back::dispatcher::Dispatcher;
+use crate::front::listener::Connection;
 use derivative::Derivative;
 use log::{info, trace};
 use parsec_interface::requests::AuthType;
 use parsec_interface::requests::ResponseStatus;
 use parsec_interface::requests::{Request, Response};
 use std::collections::HashMap;
 use std::io::{Error, ErrorKind, Result};
-use std::io::{Read, Write};
 
 /// Read and verify request from IPC stream
 ///
@@ -40,17 +40,17 @@ impl FrontEndHandler {
     ///
     /// If an error occurs during (un)marshalling, no operation will be performed and the
     /// method will return.
-    pub fn handle_request<T: Read + Write>(&self, mut stream: T) {
+    pub fn handle_request(&self, mut connection: Connection) {
         trace!("handle_request ingress");
         // Read bytes from stream
         // De-Serialise bytes into a request
-        let request = match Request::read_from_stream(&mut stream, self.body_len_limit) {
+        let request = match Request::read_from_stream(&mut connection.stream, self.body_len_limit) {
             Ok(request) => request,
             Err(status) => {
                 format_error!("Failed to read request", status);
 
                 let response = Response::from_status(status);
-                if let Err(status) = response.write_to_stream(&mut stream) {
+                if let Err(status) = response.write_to_stream(&mut connection.stream) {
                     format_error!("Failed to write response", status);
                 }
                 return;
@@ -63,7 +63,7 @@ impl FrontEndHandler {
         // Otherwise find an authenticator that is capable to authenticate the request
         } else if let Some(authenticator) = self.authenticators.get(&request.header.auth_type) {
             // Authenticate the request
-            match authenticator.authenticate(&request.auth) {
+            match authenticator.authenticate(&request.auth, connection.metadata) {
                 // Send the request to the dispatcher
                 // Get a response back
                 Ok(app_name) => (Some(app_name), None),
@@ -102,7 +102,7 @@ impl FrontEndHandler {
 
         // Serialise the response into bytes
         // Write bytes to stream
-        match response.write_to_stream(&mut stream) {
+        match response.write_to_stream(&mut connection.stream) {
             Ok(_) => {
                 if crate::utils::GlobalConfig::log_error_details() {
                     if let Some(app_name_string) = app_name {

src/front/listener.rs

@@ -5,6 +5,7 @@
 //! The [`Listen`](https://parallaxsecond.github.io/parsec-book/parsec_service/listeners.html)
 //! trait acts as an interface for the operations that must be supported by any implementation
 //! of the IPC mechanism used as a Parsec front.
+use derivative::Derivative;
 use serde::Deserialize;
 use std::time::Duration;
 
@@ -25,6 +26,23 @@ pub struct ListenerConfig {
     pub timeout: u64,
 }
 
+/// Specifies metadata associated with a connection, if any.
+#[derive(Copy, Clone, Debug)]
+pub enum ConnectionMetadata {
+    // TODO: nothing here right now. Metadata types will be added as needed.
+}
+
+/// Represents a connection to a single client. Contains a stream, used for communication with the
+/// client, and some metadata associated with the connection that might be useful elsewhere (i.e.
+/// authentication, etc).
+#[derive(Derivative)]
+#[derivative(Debug)]
+pub struct Connection {
+    #[derivative(Debug = "ignore")]
+    pub stream: Box<dyn ReadWrite + Send>,
+    pub metadata: Option<ConnectionMetadata>,
+}
+
 /// IPC front manager interface
 ///
 /// Interface defining the functionality that any IPC front manager has to expose to Parsec for normal
@@ -45,5 +63,5 @@ pub trait Listen {
     /// # Panics
     ///
     /// If the listener has not been initialised before, with the `init` method.
-    fn accept(&self) -> Option<Box<dyn ReadWrite + Send>>;
+    fn accept(&self) -> Option<Connection>;
 }