diff --git a/services/tunnelbroker/src/libcpp/Tunnelbroker.cpp b/services/tunnelbroker/src/libcpp/Tunnelbroker.cpp index 0d40dfb64..1f11fc3b3 100644 --- a/services/tunnelbroker/src/libcpp/Tunnelbroker.cpp +++ b/services/tunnelbroker/src/libcpp/Tunnelbroker.cpp @@ -1,261 +1,266 @@ #include "Tunnelbroker.h" #include "AmqpManager.h" #include "AwsTools.h" #include "ConfigManager.h" #include "DatabaseManager.h" #include "DeliveryBroker.h" #include "GlobalTools.h" #include "Tools.h" #include "rust/cxx.h" #include "tunnelbroker/src/cxx_bridge.rs.h" #include void initialize() { comm::network::tools::InitLogging("tunnelbroker"); comm::network::config::ConfigManager::getInstance().load(); Aws::InitAPI({}); // List of AWS DynamoDB tables to check if they are created and can be // accessed before any AWS API methods const std::list tablesList = { comm::network::config::ConfigManager::getInstance().getParameter( comm::network::config::ConfigManager::OPTION_DYNAMODB_SESSIONS_TABLE), comm::network::config::ConfigManager::getInstance().getParameter( comm::network::config::ConfigManager:: OPTION_DYNAMODB_SESSIONS_VERIFICATION_TABLE), comm::network::config::ConfigManager::getInstance().getParameter( comm::network::config::ConfigManager:: OPTION_DYNAMODB_SESSIONS_PUBLIC_KEY_TABLE), comm::network::config::ConfigManager::getInstance().getParameter( comm::network::config::ConfigManager:: OPTION_DYNAMODB_MESSAGES_TABLE)}; for (const std::string &table : tablesList) { if (!comm::network::database::DatabaseManager::getInstance() .isTableAvailable(table)) { throw std::runtime_error( "Error: AWS DynamoDB table '" + table + "' is not available"); } }; comm::network::AmqpManager::getInstance().init(); } rust::String getConfigParameter(rust::Str parameter) { return rust::String{ comm::network::config::ConfigManager::getInstance().getParameter( std::string{parameter})}; } bool isConfigParameterSet(rust::Str parameter) { return comm::network::config::ConfigManager::getInstance().isParameterSet( std::string{parameter}); } bool isSandbox() { return comm::network::tools::isSandbox(); } SessionSignatureResult sessionSignatureHandler(rust::Str deviceID) { const std::string requestedDeviceID(deviceID); if (!comm::network::tools::validateDeviceID(requestedDeviceID)) { return SessionSignatureResult{ .grpcStatus = { .statusCode = GRPCStatusCodes::InvalidArgument, .errorText = "Format validation failed for deviceID: " + requestedDeviceID}}; } const std::string toSign = comm::network::tools::generateRandomString( comm::network::SIGNATURE_REQUEST_LENGTH); std::shared_ptr SessionSignItem = std::make_shared( toSign, requestedDeviceID); comm::network::database::DatabaseManager::getInstance().putSessionSignItem( *SessionSignItem); return SessionSignatureResult{ .toSign = toSign, .grpcStatus = {.statusCode = GRPCStatusCodes::Ok}}; } rust::String getSavedNonceToSign(rust::Str deviceID) { const auto sessionSignatureItem = comm::network::database::DatabaseManager::getInstance() .findSessionSignItem(std::string{deviceID}); if (sessionSignatureItem == nullptr) { throw std::invalid_argument( "No requests found for 'deviceID': " + std::string{deviceID}); }; return rust::String{sessionSignatureItem->getSign()}; } NewSessionResult newSessionHandler( rust::Str deviceID, rust::Str publicKey, int32_t deviceType, rust::Str deviceAppVersion, rust::Str deviceOS, rust::Str notifyToken) { std::shared_ptr deviceSessionItem; std::shared_ptr sessionSignItem; std::shared_ptr publicKeyItem; const std::string stringDeviceID{deviceID}; if (!comm::network::tools::validateDeviceID(stringDeviceID)) { return NewSessionResult{ .grpcStatus = { .statusCode = GRPCStatusCodes::InvalidArgument, .errorText = "Format validation failed for deviceID"}}; } const std::string stringPublicKey{publicKey}; const std::string newSessionID = comm::network::tools::generateUUID(); try { publicKeyItem = comm::network::database::DatabaseManager::getInstance() .findPublicKeyItem(stringDeviceID); if (publicKeyItem == nullptr) { std::shared_ptr newPublicKeyItem = std::make_shared( stringDeviceID, stringPublicKey); comm::network::database::DatabaseManager::getInstance().putPublicKeyItem( *newPublicKeyItem); } else if (stringPublicKey != publicKeyItem->getPublicKey()) { return NewSessionResult{ .grpcStatus = { .statusCode = GRPCStatusCodes::PermissionDenied, .errorText = "The public key doesn't match for deviceID"}}; } comm::network::database::DatabaseManager::getInstance() .removeSessionSignItem(stringDeviceID); deviceSessionItem = std::make_shared( newSessionID, stringDeviceID, stringPublicKey, std::string{notifyToken}, deviceType, std::string{deviceAppVersion}, std::string{deviceOS}); comm::network::database::DatabaseManager::getInstance().putSessionItem( *deviceSessionItem); } catch (std::runtime_error &e) { LOG(ERROR) << "gRPC: " << "Error while processing 'NewSession' request: " << e.what(); return NewSessionResult{ .grpcStatus = { .statusCode = GRPCStatusCodes::Internal, .errorText = e.what()}}; } return NewSessionResult{ .sessionID = newSessionID, .grpcStatus = {.statusCode = GRPCStatusCodes::Ok}}; } SessionItem getSessionItem(rust::Str sessionID) { const std::string stringSessionID = std::string{sessionID}; if (!comm::network::tools::validateSessionID(stringSessionID)) { throw std::invalid_argument("Invalid format for 'sessionID'"); } std::shared_ptr sessionItem = comm::network::database::DatabaseManager::getInstance().findSessionItem( stringSessionID); if (sessionItem == nullptr) { throw std::invalid_argument( "No sessions found for 'sessionID': " + stringSessionID); } return SessionItem{ .deviceID = sessionItem->getDeviceID(), .publicKey = sessionItem->getPubKey(), .notifyToken = sessionItem->getNotifyToken(), .deviceType = static_cast(sessionItem->getDeviceType()), .appVersion = sessionItem->getAppVersion(), .deviceOS = sessionItem->getDeviceOs(), .isOnline = sessionItem->getIsOnline()}; } void updateSessionItemIsOnline(rust::Str sessionID, bool isOnline) { comm::network::database::DatabaseManager::getInstance() .updateSessionItemIsOnline(std::string{sessionID}, isOnline); } void updateSessionItemDeviceToken( rust::Str sessionID, rust::Str newNotifToken) { + if (comm::network::config::ConfigManager::getInstance().isParameterSet( + comm::network::config::ConfigManager:: + OPTION_SESSIONS_SKIP_AUTH_KEY)) { + return; + } comm::network::database::DatabaseManager::getInstance() .updateSessionItemDeviceToken( std::string{sessionID}, std::string{newNotifToken}); } rust::Vec getMessagesFromDatabase(rust::Str deviceID) { std::vector> messagesFromDatabase = comm::network::database::DatabaseManager::getInstance() .findMessageItemsByReceiver(std::string{deviceID}); rust::Vec result; for (auto &messageFromDatabase : messagesFromDatabase) { result.push_back(MessageItem{ .messageID = messageFromDatabase->getMessageID(), .fromDeviceID = messageFromDatabase->getFromDeviceID(), .payload = messageFromDatabase->getPayload(), .blobHashes = messageFromDatabase->getBlobHashes(), }); } return result; } void eraseMessagesFromAMQP(rust::Str deviceID) { comm::network::DeliveryBroker::getInstance().erase(std::string{deviceID}); } void ackMessageFromAMQP(uint64_t deliveryTag) { comm::network::AmqpManager::getInstance().ack(deliveryTag); } MessageItem waitMessageFromDeliveryBroker(rust::Str deviceID) { const auto message = comm::network::DeliveryBroker::getInstance().pop(std::string{deviceID}); return MessageItem{ .messageID = message.messageID, .fromDeviceID = message.fromDeviceID, .payload = message.payload, .deliveryTag = message.deliveryTag}; } void removeMessages( rust::Str deviceID, const rust::Vec &messagesIDs) { std::vector vectorOfmessagesIDs; std::string stringDeviceID = std::string{deviceID}; for (auto id : messagesIDs) { vectorOfmessagesIDs.push_back(std::string{id}); }; comm::network::database::DatabaseManager::getInstance() .removeMessageItemsByIDsForDeviceID(vectorOfmessagesIDs, stringDeviceID); // If messages queue for `deviceID` is empty we don't need to store // `folly::MPMCQueue` for it and need to free memory to fix possible // 'ghost' queues in DeliveryBroker. // We call `deleteQueueIfEmpty()` for this purpose here after removing // messages. comm::network::DeliveryBroker::DeliveryBroker::getInstance() .deleteQueueIfEmpty(stringDeviceID); } rust::Vec sendMessages(const rust::Vec &messages) { std::vector vectorOfMessages; rust::Vec messagesIDs; for (auto &message : messages) { std::string messageID = comm::network::tools::generateUUID(); vectorOfMessages.push_back(comm::network::database::MessageItem{ comm::network::database::MessageItem{ messageID, std::string{message.fromDeviceID}, std::string{message.toDeviceID}, std::string{message.payload}, std::string{message.blobHashes}, }}); messagesIDs.push_back(rust::String{messageID}); }; comm::network::database::DatabaseManager::getInstance() .putMessageItemsByBatch(vectorOfMessages); for (auto message : vectorOfMessages) { comm::network::AmqpManager::getInstance().send(&message); } return messagesIDs; } diff --git a/services/tunnelbroker/src/server/mod.rs b/services/tunnelbroker/src/server/mod.rs index e656f80e1..22707de29 100644 --- a/services/tunnelbroker/src/server/mod.rs +++ b/services/tunnelbroker/src/server/mod.rs @@ -1,390 +1,414 @@ -use crate::cxx_bridge::ffi::MessageItem; +use crate::cxx_bridge::ffi::{MessageItem, SessionItem}; use super::constants; use super::cxx_bridge::ffi::{ ackMessageFromAMQP, eraseMessagesFromAMQP, getMessagesFromDatabase, - getSavedNonceToSign, getSessionItem, newSessionHandler, removeMessages, - sendMessages, sessionSignatureHandler, updateSessionItemDeviceToken, - updateSessionItemIsOnline, waitMessageFromDeliveryBroker, GRPCStatusCodes, + getSavedNonceToSign, getSessionItem, isConfigParameterSet, newSessionHandler, + removeMessages, sendMessages, sessionSignatureHandler, + updateSessionItemDeviceToken, updateSessionItemIsOnline, + waitMessageFromDeliveryBroker, GRPCStatusCodes, }; use anyhow::Result; use futures::Stream; use std::pin::Pin; use tokio::sync::mpsc; use tokio_stream::{wrappers::ReceiverStream, StreamExt}; use tonic::{transport::Server, Request, Response, Status, Streaming}; use tracing::{debug, error}; use tunnelbroker::message_to_tunnelbroker::Data::{ MessagesToSend, NewNotifyToken, ProcessedMessages, }; use tunnelbroker::tunnelbroker_service_server::{ TunnelbrokerService, TunnelbrokerServiceServer, }; mod tools; mod tunnelbroker { tonic::include_proto!("tunnelbroker"); } #[derive(Debug, Default)] struct TunnelbrokerServiceHandlers {} #[tonic::async_trait] impl TunnelbrokerService for TunnelbrokerServiceHandlers { async fn session_signature( &self, request: Request, ) -> Result, Status> { let result = sessionSignatureHandler(&request.into_inner().device_id); if result.grpcStatus.statusCode != GRPCStatusCodes::Ok { return Err(tools::create_tonic_status( result.grpcStatus.statusCode, &result.grpcStatus.errorText, )); } Ok(Response::new(tunnelbroker::SessionSignatureResponse { to_sign: result.toSign, })) } async fn new_session( &self, request: Request, ) -> Result, Status> { let inner_request = request.into_inner(); let notify_token = inner_request.notify_token.unwrap_or(String::new()); if !tunnelbroker::new_session_request::DeviceTypes::is_valid( inner_request.device_type, ) { return Err(tools::create_tonic_status( GRPCStatusCodes::InvalidArgument, "Unsupported device type", )); }; let nonce_to_be_signed = match getSavedNonceToSign(&inner_request.device_id) { Ok(saved_nonce) => saved_nonce, Err(err) => { return Err(tools::create_tonic_status( GRPCStatusCodes::Internal, &err.what(), )) } }; match tools::verify_signed_string( &inner_request.public_key, &nonce_to_be_signed, &inner_request.signature, ) { Ok(verifying_result) => { if !verifying_result { return Err(tools::create_tonic_status( GRPCStatusCodes::PermissionDenied, "Signature for the verification message is not valid", )); } } Err(_) => { return Err(tools::create_tonic_status( GRPCStatusCodes::Internal, "Error while verifying the signature", )) } } let result = newSessionHandler( &inner_request.device_id, &inner_request.public_key, inner_request.device_type, &inner_request.device_app_version, &inner_request.device_os, ¬ify_token, ); if result.grpcStatus.statusCode != GRPCStatusCodes::Ok { return Err(tools::create_tonic_status( result.grpcStatus.statusCode, &result.grpcStatus.errorText, )); } Ok(Response::new(tunnelbroker::NewSessionResponse { session_id: result.sessionID, })) } type MessagesStreamStream = Pin< Box< dyn Stream> + Send, >, >; async fn messages_stream( &self, request: Request>, ) -> Result, Status> { - let session_id = match request.metadata().get("sessionID") { - Some(metadata_session_id) => metadata_session_id + let session_id: String; + let session_item: SessionItem; + if isConfigParameterSet("sessions.skip_authentication").expect( + "Error while checking the skip_authentication config file parameter", + ) { + session_id = String::new(); + let device_id = request + .metadata() + .get("deviceID") + .expect("Expected 'deviceID' value in metadata is not provided") .to_str() - .expect("metadata session id was not valid UTF8") - .to_string(), - None => { - return Err(Status::invalid_argument( - "No 'sessionID' in metadata was provided", - )) - } - }; - let session_item = match getSessionItem(&session_id) { - Ok(database_item) => database_item, - Err(err) => return Err(Status::unauthenticated(err.what())), - }; - + .expect("Metadata 'deviceID' value is not a valid UTF8") + .to_string(); + session_item = SessionItem { + deviceID: device_id, + publicKey: String::new(), + notifyToken: String::new(), + deviceType: 0, + appVersion: String::new(), + deviceOS: String::new(), + isOnline: true, + }; + } else { + session_id = match request.metadata().get("sessionID") { + Some(metadata_session_id) => metadata_session_id + .to_str() + .expect("metadata session id was not valid UTF8") + .to_string(), + None => { + return Err(Status::invalid_argument( + "No 'sessionID' in metadata was provided", + )); + } + }; + session_item = match getSessionItem(&session_id) { + Ok(database_item) => database_item, + Err(err) => return Err(Status::unauthenticated(err.what())), + }; + } let (tx, rx) = mpsc::channel(constants::GRPC_TX_QUEUE_SIZE); // Through this function, we will write to the output stream from different Tokio // tasks and update the device's online status if the write was unsuccessful async fn tx_writer( session_id: &str, channel: &tokio::sync::mpsc::Sender, payload: T, ) -> Result<(), String> { let result = channel.send(payload).await; match result { Ok(result) => Ok(result), Err(err) => { if let Err(err) = updateSessionItemIsOnline(&session_id, false) { return Err(err.what().to_string()); } return Err(err.to_string()); } } } if let Err(err) = updateSessionItemIsOnline(&session_id, true) { return Err(Status::internal(err.what())); } // Checking for an empty notif token and requesting the new one from the client if session_item.notifyToken.is_empty() && session_item.deviceType == tunnelbroker::new_session_request::DeviceTypes::Mobile as i32 { let result = tx_writer( &session_id, &tx, Ok(tunnelbroker::MessageToClient { data: Some( tunnelbroker::message_to_client::Data::NewNotifyTokenRequired(()), ), }), ); if let Err(err) = result.await { debug!( "Error while sending notification token request to the client: {}", err ); }; } // When a client connects to the bidirectional messages stream, first we check // if there are undelivered messages in the database let messages_from_database = match getMessagesFromDatabase(&session_item.deviceID) { Ok(messages) => messages, Err(err) => return Err(Status::internal(err.what())), }; if messages_from_database.len() > 0 { if let Err(err) = eraseMessagesFromAMQP(&session_item.deviceID) { return Err(Status::internal(err.what())); }; let mut messages_to_response = vec![]; for message in &messages_from_database { messages_to_response.push(tunnelbroker::MessageToClientStruct { message_id: message.messageID.clone(), from_device_id: message.fromDeviceID.clone(), payload: message.payload.clone(), blob_hashes: vec![message.blobHashes.clone()], }); } let result_from_writer = tx_writer( &session_id, &tx, Ok(tunnelbroker::MessageToClient { data: Some(tunnelbroker::message_to_client::Data::MessagesToDeliver( tunnelbroker::MessagesToDeliver { messages: messages_to_response, }, )), }), ); if let Err(err) = result_from_writer.await { debug!( "Error while sending undelivered messages from database to the client: {}", err ); return Err(Status::aborted(err)); }; } // Spawning asynchronous Tokio task to deliver new messages // to the client from delivery broker tokio::spawn({ let device_id = session_item.deviceID.clone(); let session_id = session_id.clone(); let tx = tx.clone(); async move { loop { let message_to_deliver = match waitMessageFromDeliveryBroker(&device_id) { Ok(message_item) => message_item, Err(err) => { error!( "Error on waiting messages from DeliveryBroker: {}", err.what() ); return; } }; let writer_result = tx_writer( &session_id, &tx, Ok(tunnelbroker::MessageToClient { data: Some( tunnelbroker::message_to_client::Data::MessagesToDeliver( tunnelbroker::MessagesToDeliver { messages: vec![tunnelbroker::MessageToClientStruct { message_id: message_to_deliver.messageID, from_device_id: message_to_deliver.fromDeviceID, payload: message_to_deliver.payload, blob_hashes: vec![message_to_deliver.blobHashes], }], }, ), ), }), ); if let Err(err) = writer_result.await { debug!("Error on writing to the stream: {}", err); return; }; if let Err(err) = ackMessageFromAMQP(message_to_deliver.deliveryTag) { debug!("Error on message acknowledgement in AMQP queue: {}", err); return; }; } } }); let mut input_stream = request.into_inner(); // Spawning asynchronous Tokio task for handling incoming messages from the client tokio::spawn(async move { while let Some(result) = input_stream.next().await { if let Err(err) = result { debug!("Error in input stream: {}", err); break; } if let Some(message_data) = result.unwrap().data { match message_data { NewNotifyToken(new_token) => { if let Err(err) = updateSessionItemDeviceToken(&session_id, &new_token) { error!( "Error in updating the device notification token in the database: {}", err.what() ); let writer_result = tx_writer( &session_id, &tx, Err( Status::internal( "Error in updating the device notification token in the database" ) ), ); if let Err(err) = writer_result.await { debug!( "Failed to write internal error to a channel: {}", err ); }; } } MessagesToSend(messages_to_send) => { let mut messages_vec = vec![]; for message in messages_to_send.messages { messages_vec.push(MessageItem { messageID: String::new(), fromDeviceID: session_item.deviceID.clone(), toDeviceID: message.to_device_id, payload: message.payload, blobHashes: String::new(), deliveryTag: 0, }); } let messages_ids = match sendMessages(&messages_vec) { Err(err) => { error!("Error on sending messages: {}", err.what()); return; } Ok(ids) => ids, }; if let Err(err) = tx_writer( &session_id, &tx, Ok(tunnelbroker::MessageToClient { data: Some( tunnelbroker::message_to_client::Data::ProcessedMessages( tunnelbroker::ProcessedMessages { message_id: messages_ids, }, ), ), }), ) .await { debug!( "Error on sending back processed messages IDs to the stream: {}", err); }; } ProcessedMessages(processed_messages) => { if let Err(err) = removeMessages( &session_item.deviceID, &processed_messages.message_id, ) { error!( "Error removing messages from the database: {}", err.what() ); }; } } } } if let Err(err) = updateSessionItemIsOnline(&session_id, false) { error!( "Error in updating the session online state in the database: {}", err.what() ); } }); let output_stream = ReceiverStream::new(rx); Ok(Response::new( Box::pin(output_stream) as Self::MessagesStreamStream )) } } pub async fn run_grpc_server() -> Result<()> { let addr = format!("[::1]:{}", constants::GRPC_SERVER_PORT).parse()?; Server::builder() .http2_keepalive_interval(Some(constants::GRPC_KEEP_ALIVE_PING_INTERVAL)) .http2_keepalive_timeout(Some(constants::GRPC_KEEP_ALIVE_PING_TIMEOUT)) .add_service(TunnelbrokerServiceServer::new( TunnelbrokerServiceHandlers::default(), )) .serve(addr) .await?; Ok(()) }