Pushing files to the remote (#158)

* added pushing files to the remote

* refactored code

* added comment about the std::Mutex usage

Co-authored-by: Anastasiia Romaniuk <romaniuk.anastasiia@apriorit.com>

devolutions-gateway/src/config.rs

@@ -32,6 +32,11 @@ const ARG_LOG_FILE: &str = "log-file";
 const ARG_SERVICE_MODE: &str = "service";
 const ARG_PLUGINS: &str = "plugins";
 const ARG_RECORDING_PATH: &str = "recording-path";
+const ARG_SOGAR_UTIL_PATH: &str = "sogar-path";
+const ARG_SOGAR_REGISTRY_URL: &str = "sogar-registry-url";
+const ARG_SOGAR_USERNAME: &str = "sogar-username";
+const ARG_SOGAR_PASSWORD: &str = "sogar-password";
+const ARG_SOGAR_IMAGE_NAME: &str = "sogar-image-name";
 
 const SERVICE_NAME: &str = "devolutions-gateway";
 const DISPLAY_NAME: &str = "Devolutions Gateway";
@@ -75,6 +80,15 @@ pub struct CertificateConfig {
     pub private_key_data: Option<String>,
 }
 
+#[derive(Debug, Default, Clone)]
+pub struct RecordingInfo {
+    pub sogar_path: Option<String>,
+    pub registry_url: Option<String>,
+    pub username: Option<String>,
+    pub password: Option<String>,
+    pub image_name: Option<String>,
+}
+
 #[derive(Debug, Clone)]
 pub struct Config {
     pub service_mode: bool,
@@ -97,6 +111,7 @@ pub struct Config {
     pub delegation_private_key: Option<PrivateKey>,
     pub plugins: Option<Vec<String>>,
     pub recording_path: Option<String>,
+    pub recording_info: RecordingInfo,
 }
 
 impl Default for Config {
@@ -129,6 +144,13 @@ impl Default for Config {
             delegation_private_key: None,
             plugins: None,
             recording_path: None,
+            recording_info: RecordingInfo {
+                sogar_path: None,
+                registry_url: None,
+                username: None,
+                password: None,
+                image_name: None,
+            },
         }
     }
 }
@@ -203,6 +225,16 @@ pub struct ConfigFile {
     pub plugins: Option<Vec<String>>,
     #[serde(rename = "RecordingPath")]
     pub recording_path: Option<String>,
+    #[serde(rename = "SogarPath")]
+    pub sogar_path: Option<String>,
+    #[serde(rename = "SogarRegistryUrl")]
+    pub registry_url: Option<String>,
+    #[serde(rename = "SogarUsername")]
+    pub username: Option<String>,
+    #[serde(rename = "SogarPassword")]
+    pub password: Option<String>,
+    #[serde(rename = "SogarImageName")]
+    pub image_name: Option<String>,
 
     // unstable options (subject to change)
     #[serde(rename = "ApiKey")]
@@ -521,6 +553,52 @@ impl Config {
                             Err(String::from("The value does not exist or is not a path"))
                         }
                     }),
+            )
+            .arg(
+                Arg::with_name(ARG_SOGAR_UTIL_PATH)
+                    .long(ARG_SOGAR_UTIL_PATH)
+                    .value_name("PATH")
+                    .help("A path where the sogar utility is located including the name and extension.")
+                    .takes_value(true)
+                    .empty_values(false)
+                    .requires_all(&[
+                        ARG_SOGAR_REGISTRY_URL,
+                        ARG_SOGAR_USERNAME,
+                        ARG_SOGAR_PASSWORD,
+                        ARG_SOGAR_IMAGE_NAME,
+                    ]),
+            )
+            .arg(
+                Arg::with_name(ARG_SOGAR_REGISTRY_URL)
+                    .long(ARG_SOGAR_REGISTRY_URL)
+                    .value_name("URL")
+                    .help("Registry url to where the session recordings will be pushed.")
+                    .env("SOGAR_REGISTRY_URL")
+                    .takes_value(true)
+                    .empty_values(false),
+            )
+            .arg(
+                Arg::with_name(ARG_SOGAR_USERNAME)
+                    .long(ARG_SOGAR_USERNAME)
+                    .help("Registry username.")
+                    .env("SOGAR_REGISTRY_USERNAME")
+                    .takes_value(true)
+                    .empty_values(false),
+            )
+            .arg(
+                Arg::with_name(ARG_SOGAR_PASSWORD)
+                    .long(ARG_SOGAR_PASSWORD)
+                    .help("Registry password.")
+                    .env("SOGAR_REGISTRY_PASSWORD")
+                    .takes_value(true)
+                    .empty_values(false),
+            )
+            .arg(
+                Arg::with_name(ARG_SOGAR_IMAGE_NAME)
+                    .long(ARG_SOGAR_IMAGE_NAME)
+                    .help("Image name of the registry where to push the file. For example videos/demo")
+                    .takes_value(true)
+                    .empty_values(false),
             );
 
         let matches = cli_app.get_matches();
@@ -658,6 +736,26 @@ impl Config {
             }
         }
 
+        if let Some(sogar_path) = matches.value_of(ARG_SOGAR_UTIL_PATH) {
+            config.recording_info.sogar_path = Some(sogar_path.to_owned());
+        }
+
+        if let Some(registry_url) = matches.value_of(ARG_SOGAR_REGISTRY_URL) {
+            config.recording_info.registry_url = Some(registry_url.to_owned());
+        }
+
+        if let Some(username) = matches.value_of(ARG_SOGAR_USERNAME) {
+            config.recording_info.username = Some(username.to_owned());
+        }
+
+        if let Some(password) = matches.value_of(ARG_SOGAR_PASSWORD) {
+            config.recording_info.password = Some(password.to_owned());
+        }
+
+        if let Some(image_name) = matches.value_of(ARG_SOGAR_IMAGE_NAME) {
+            config.recording_info.image_name = Some(image_name.to_owned());
+        }
+
         // listeners parsing
 
         let mut listeners = Vec::new();
@@ -825,6 +923,11 @@ impl Config {
 
         let plugins = config_file.plugins;
         let recording_path = config_file.recording_path;
+        let sogar_path = config_file.sogar_path;
+        let registry_url = config_file.registry_url;
+        let username = config_file.username;
+        let password = config_file.password;
+        let image_name = config_file.image_name;
 
         // unstable options (subject to change)
         let api_key = config_file.api_key;
@@ -849,6 +952,13 @@ impl Config {
             delegation_private_key,
             plugins,
             recording_path,
+            recording_info: RecordingInfo {
+                sogar_path,
+                registry_url,
+                username,
+                password,
+                image_name,
+            },
             ..Default::default()
         })
     }

devolutions-gateway/src/interceptor/pcap_recording.rs

@@ -2,6 +2,7 @@ use crate::interceptor::{PacketInterceptor, PeerInfo};
 use crate::plugin_manager::{PacketsParser, Recorder, PLUGIN_MANAGER};
 use slog_scope::{debug, error};
 use std::net::SocketAddr;
+use std::path::PathBuf;
 use std::sync::{Arc, Condvar, Mutex};
 use std::thread;
 use std::time::Duration;
@@ -12,6 +13,8 @@ enum RecordingState {
     Finish,
 }
 
+// The interceptor implements the trait PacketInterceptor which is not async, therefore
+// the std::Mutex is used here
 #[derive(Clone)]
 pub struct PcapRecordingInterceptor {
     server_info: Arc<Mutex<PeerInfo>>,
@@ -20,15 +23,17 @@ pub struct PcapRecordingInterceptor {
     recorder: Arc<Mutex<Option<Recorder>>>,
     condition_timeout: Arc<(Mutex<RecordingState>, Condvar)>,
     handle: Arc<Mutex<Option<thread::JoinHandle<()>>>>,
+    file_name: Arc<Mutex<String>>,
 }
 
 impl PcapRecordingInterceptor {
     pub fn new(server_addr: SocketAddr, client_addr: SocketAddr, association_id: String, candidate_id: String) -> Self {
         debug!("Recording Interceptor was created");
         let recording_plugin = PLUGIN_MANAGER.lock().unwrap().get_recording_plugin();
+        let file_name = format!("{}-to-{}", association_id, candidate_id);
+
         if let Some(recorder) = &recording_plugin {
-            let filename = format!("{}-to-{}", association_id, candidate_id);
-            recorder.set_filename(filename.as_str());
+            recorder.set_filename(file_name.as_str());
         }
 
         let recorder = Arc::new(Mutex::new(recording_plugin));
@@ -74,6 +79,7 @@ impl PcapRecordingInterceptor {
             recorder,
             condition_timeout,
             handle: Arc::new(Mutex::new(Some(handle))),
+            file_name: Arc::new(Mutex::new(file_name)),
         }
     }
 
@@ -83,6 +89,25 @@ impl PcapRecordingInterceptor {
             recorder.set_directory(directory);
         }
     }
+
+    pub fn get_recording_directory(&self) -> Option<PathBuf> {
+        let rec = self.recorder.lock().unwrap();
+        if let Some(recorder) = rec.as_ref() {
+            match recorder.get_filepath() {
+                Ok(path_buf) => {
+                    debug!("the path is {:?}", path_buf.to_str());
+                    return Some(path_buf);
+                }
+                Err(e) => error!("Failed to get video path: {}", e),
+            }
+        }
+        None
+    }
+
+    pub fn get_filename_pattern(&self) -> String {
+        let rec = self.file_name.lock().unwrap();
+        rec.clone()
+    }
 }
 
 impl PacketInterceptor for PcapRecordingInterceptor {

devolutions-gateway/src/jet_client.rs

@@ -20,12 +20,14 @@ use crate::interceptor::pcap_recording::PcapRecordingInterceptor;
 use crate::jet::association::Association;
 use crate::jet::candidate::{Candidate, CandidateState};
 use crate::jet::TransportType;
+use crate::plugin_manager::SogarData;
 use crate::transport::tcp::TcpTransport;
 use crate::transport::{JetTransport, Transport};
 use crate::utils::association::{remove_jet_association, ACCEPT_REQUEST_TIMEOUT};
 use crate::utils::{create_tls_connector, into_other_io_error as error_other};
 use crate::Proxy;
 
+use std::path::PathBuf;
 use tokio_rustls::{TlsAcceptor, TlsStream};
 
 pub type JetAssociationsMap = Arc<Mutex<HashMap<Uuid, Association>>>;
@@ -112,6 +114,8 @@ async fn handle_build_proxy(
 ) -> Result<(), io::Error> {
     let mut recording_interceptor: Option<PcapRecordingInterceptor> = None;
     let association_id = response.association_id;
+    let mut remote_data = None;
+    let mut recording_dir = None;
 
     let associations = jet_associations.lock().await;
     if let Some(association) = associations.get(&association_id) {
@@ -124,17 +128,38 @@ async fn handle_build_proxy(
                     response.candidate_id.clone().to_string(),
                 );
 
-                if let Some(path) = &config.recording_path {
-                    interceptor.set_recording_directory(path.as_str());
-                }
+                recording_dir = match &config.recording_path {
+                    Some(path) => {
+                        interceptor.set_recording_directory(path.as_str());
+                        Some(PathBuf::from(path))
+                    }
+                    None => interceptor.get_recording_directory(),
+                };
+
+                let file_pattern = interceptor.get_filename_pattern();
+
+                let recording_info = config.recording_info.clone();
+                remote_data = SogarData::new(
+                    recording_info.sogar_path.clone(),
+                    recording_info.registry_url.clone(),
+                    recording_info.username.clone(),
+                    recording_info.password.clone(),
+                    recording_info.image_name,
+                    Some(file_pattern),
+                );
 
                 recording_interceptor = Some(interceptor);
             }
         }
     }
 
     if let Some(interceptor) = recording_interceptor {
-        handle_build_tls_proxy(config, response, interceptor, tls_acceptor).await
+        let proxy_result = handle_build_tls_proxy(config, response, interceptor, tls_acceptor).await;
+        if let (Some(push_data), Some(dir)) = (remote_data, recording_dir) {
+            push_data.push(dir.as_path(), association_id.clone().to_string())
+        };
+
+        proxy_result
     } else {
         Proxy::new(config)
             .build(response.server_transport, response.client_transport)

devolutions-gateway/src/plugin_manager.rs

@@ -6,10 +6,12 @@ use std::sync::{Arc, Mutex};
 
 mod packets_parsing;
 mod plugin_info;
+mod push_files;
 mod recording;
 use crate::utils::into_other_io_error;
 pub use packets_parsing::PacketsParser;
 use plugin_info::{PluginCapabilities, PluginInformation};
+pub use push_files::SogarData;
 pub use recording::Recorder;
 
 #[derive(Clone)]