Add Coturn integration (#20)

- add dockerized Coturn and Redis
- add TurnAuthRepo and TurnAuthService
- extend protocol with ICE servers

.env

@@ -1,2 +1,8 @@
 RUST_LOG=debug
 MEDEA_CONF=config.toml
+
+COMPOSE_PROJECT_NAME=medea
+COMPOSE_IMAGE_NAME=hub.instrumentisto.com/medea
+COMPOSE_IMAGE_VER=dev
+
+COMPOSE_EXTERNAL_IP=127.0.0.1

Cargo.toml

@@ -22,6 +22,8 @@ lto = "thin"
 [dependencies]
 actix = "0.7"
 actix-web = "0.7"
+bb8 = "0.3.0"
+bb8-redis = "0.3.0"
 config = "0.9"
 chrono = "0.4"
 dotenv = "0.13"
@@ -33,7 +35,10 @@ macro-attr = "0.2"
 medea-client-api-proto = { path = "proto/client-api", features = ["medea"] }
 medea-macro = { path = "crates/medea-macro" }
 newtype_derive = "0.1"
-serde = { version = "1.0", features = ['derive'] }
+rand = "0.6"
+rust-crypto = "0.2"
+redis = "0.10.0"
+serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 slog = "2.4"
 slog-envlogger = "2.1"
@@ -43,11 +48,11 @@ slog-json = "2.3"
 slog-scope = "4.1"
 smart-default = "0.5"
 toml = "0.4"
+tokio = "0.1"
 [dependencies.serde-humantime]
     git = "https://github.com/tailhook/serde-humantime"
     branch = "serde_wrapper"
 
 [dev-dependencies]
 serial_test = "0.2"
 serial_test_derive = "0.2"
-tokio = "0.1"

Makefile

@@ -38,7 +38,7 @@ fmt: cargo.fmt
 #	make up
 
 up:
-	$(MAKE) -j2 up.jason up.medea
+	$(MAKE) -j3 up.coturn up.jason up.medea
 
 
 test: test.unit
@@ -170,6 +170,15 @@ up.medea:
 	cargo run --bin medea
 
 
+# Run Coturn and Redis.
+#
+# Usage:
+#	make up.coturn
+
+up.coturn:
+	docker-compose up
+
+
 
 
 ##################

_dev/coturn/redis.conf

@@ -0,0 +1,5 @@
+maxmemory 16mb
+maxmemory-policy allkeys-lfu
+requirepass turn
+timeout 0
+tcp-keepalive 60

_dev/coturn/turnserver.conf

@@ -0,0 +1,10 @@
+lt-cred-mech
+fingerprint
+no-cli
+no-tls
+no-dtls
+min-port=49160
+max-port=49200
+realm=medea
+redis-userdb="ip=127.0.0.1 port=6379 dbname=0 password=turn"
+user=USER:PASS

config.toml

@@ -9,9 +9,6 @@
 # Default:
 #   bind_port = 8080
 
-
-
-
 [rpc]
 # Duration, after which remote RPC client will be considered idle if no
 # heartbeat messages received.
@@ -24,3 +21,45 @@
 #
 # Default:
 #   reconnect_timeout = "10s"
+
+[turn]
+# Turn server IP address.
+#
+# Default:
+#   ip = "0.0.0.0"
+
+# Turn server port.
+#
+# Default:
+#   port = 3478
+
+# Static user on Turn server.
+#
+# Default:
+#   user = "USER"
+
+# Static user password on Turn server.
+#
+# Default:
+#   pass = "PASS"
+
+[turn.db.redis]
+# IP address to bind Redis server to.
+#
+# Default:
+#   ip = "0.0.0.0"
+
+# Redis database port.
+#
+# Default:
+#   port = 6379
+
+# Redis database password.
+#
+# Default:
+#   pass = "turn"
+
+# Redis database number.
+#
+# Default:
+#   db_number = 0

docker-compose.yml

@@ -0,0 +1,21 @@
+version: '3.4'
+
+services:
+  coturn-db:
+    container_name: ${COMPOSE_PROJECT_NAME}-coturn-db
+    image: redis:alpine
+    command: ["redis-server", "/etc/redis.conf"]
+    ports:
+      - "6379:6379"    # coturn redis
+    volumes:
+      - ./_dev/coturn/redis.conf:/etc/redis.conf:ro
+  coturn:
+    container_name: ${COMPOSE_PROJECT_NAME}-coturn
+    image: instrumentisto/coturn:4.5
+    depends_on: ["coturn-db"]
+    command:
+      - --log-file=stdout
+    volumes:
+      - ./_dev/coturn/turnserver.conf:/etc/coturn/turnserver.conf:ro
+      - ./.cache/coturn/data:/var/lib/coturn
+    network_mode: host

proto/client-api/src/lib.rs

@@ -57,6 +57,7 @@ pub enum Event {
         peer_id: u64,
         sdp_offer: Option<String>,
         tracks: Vec<Track>,
+        ice_servers: Vec<IceServer>,
     },
     /// Media Server notifies Web Client about necessity to apply specified SDP
     /// Answer to Web Client's RTCPeerConnection.
@@ -92,6 +93,17 @@ pub struct Track {
     pub media_type: MediaType,
 }
 
+/// Representation of [`iceServers`] field of [`RTCConfiguration`] dictionary.
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[cfg_attr(test, derive(PartialEq))]
+pub struct IceServer {
+    pub urls: Vec<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub username: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub credential: Option<String>,
+}
+
 /// Direction of [`Track`].
 #[cfg_attr(feature = "medea", derive(Serialize))]
 #[cfg_attr(feature = "jason", derive(Deserialize))]

signaling_test.html

@@ -65,7 +65,8 @@
                 }
             });
 
-            pcs[msg.peer_id] = new RTCPeerConnection();
+            var configuration = { iceServers: msg.ice_servers };
+            pcs[msg.peer_id] = new RTCPeerConnection(configuration);
 
             var pc = pcs[msg.peer_id];
             pc.ontrack = function (e) {

src/api/client/rpc_connection.rs

@@ -1,6 +1,6 @@
 //! [`RpcConnection`] with related messages.
 
-use std::fmt;
+use core::fmt;
 
 use actix::Message;
 use futures::Future;
@@ -162,6 +162,7 @@ pub mod test {
                     peer_id,
                     sdp_offer,
                     tracks: _,
+                    ice_servers: _,
                 } => {
                     match sdp_offer {
                         Some(_) => self.room.do_send(CommandMessage::from(

src/api/client/server.rs

@@ -81,7 +81,7 @@ pub struct Context {
 
 /// Starts HTTP server for handling WebSocket connections of Client API.
 pub fn run(rooms: RoomsRepository, config: Conf) {
-    let server_addr = config.server.get_bind_addr();
+    let server_addr = config.server.bind_addr();
 
     server::new(move || {
         App::with_state(Context {
@@ -110,21 +110,36 @@ mod test {
 
     use crate::{
         api::control::Member,
-        conf::{Conf, Server},
+        conf::{Conf, Server, Turn},
         media::create_peers,
         signalling::Room,
+        turn::new_turn_auth_service_mock,
     };
 
     use super::*;
 
     /// Creates [`RoomsRepository`] for tests filled with a single [`Room`].
     fn room(conf: Rpc) -> RoomsRepository {
         let members = hashmap! {
-            1 => Member{id: 1, credentials: "caller_credentials".into()},
-            2 => Member{id: 2, credentials: "responder_credentials".into()},
+            1 => Member{
+                id: 1,
+                credentials: "caller_credentials".into(),
+                ice_user: None
+            },
+            2 => Member{
+                id: 2,
+                credentials: "responder_credentials".into(),
+                ice_user: None
+            },
         };
         let room = Arbiter::start(move |_| {
-            Room::new(1, members, create_peers(1, 2), conf.reconnect_timeout)
+            Room::new(
+                1,
+                members,
+                create_peers(1, 2),
+                conf.reconnect_timeout,
+                new_turn_auth_service_mock(),
+            )
         });
         let rooms = hashmap! {1 => room};
         RoomsRepository::new(rooms)
@@ -161,6 +176,7 @@ mod test {
                 idle_timeout: Duration::new(2, 0),
                 reconnect_timeout: Default::default(),
             },
+            turn: Turn::default(),
             server: Server::default(),
         };
 

src/api/control/member.rs

@@ -1,5 +1,7 @@
 //! Member definitions and implementations.
 
+use crate::media::IceUser;
+
 /// ID of [`Member`].
 pub type Id = u64;
 
@@ -11,4 +13,18 @@ pub struct Member {
 
     /// Credentials to authorize [`Member`] with.
     pub credentials: String,
+
+    /// Turn server credentials.
+    pub ice_user: Option<IceUser>,
+}
+
+impl Member {
+    /// Returns new instance of [`Memebr`] with given credentials.
+    pub fn new(id: Id, credentials: String) -> Self {
+        Self {
+            id,
+            credentials,
+            ice_user: None,
+        }
+    }
 }

src/conf/mod.rs

@@ -2,14 +2,19 @@
 
 pub mod rpc;
 pub mod server;
+pub mod turn;
 
 use std::env;
 
 use config::{Config, Environment, File};
 use failure::Error;
 use serde::{Deserialize, Serialize};
 
-pub use self::{rpc::Rpc, server::Server};
+pub use self::{
+    rpc::Rpc,
+    server::Server,
+    turn::{Redis, Turn},
+};
 
 /// CLI argument that is responsible for holding application configuration
 /// file path.
@@ -22,11 +27,12 @@ static APP_CONF_PATH_ENV_VAR_NAME: &str = "MEDEA_CONF";
 #[derive(Clone, Debug, Deserialize, Serialize, Default)]
 #[serde(default)]
 pub struct Conf {
-    /// RPC connection settings.
-    pub rpc: rpc::Rpc,
-
     /// HTTP server settings.
-    pub server: server::Server,
+    pub rpc: Rpc,
+    /// RPC connection settings.
+    pub server: Server,
+    /// TURN server settings.
+    pub turn: Turn,
 }
 
 impl Conf {
@@ -71,8 +77,9 @@ where
 
 #[cfg(test)]
 mod tests {
+    use std::{fs, net::Ipv4Addr, time::Duration};
+
     use serial_test_derive::serial;
-    use std::{fs, time::Duration};
 
     use super::*;
 
@@ -186,4 +193,46 @@ mod tests {
 
         assert_eq!(file_env_config.rpc.idle_timeout, Duration::from_secs(48));
     }
+
+    #[test]
+    #[serial]
+    fn redis_conf_test() {
+        let default_conf = Conf::default();
+
+        env::set_var("MEDEA_TURN.DB.REDIS.IP", "5.5.5.5");
+        env::set_var("MEDEA_TURN.DB.REDIS.PORT", "1234");
+
+        let env_conf = Conf::parse().unwrap();
+
+        assert_ne!(default_conf.turn.db.redis.ip, env_conf.turn.db.redis.ip);
+        assert_ne!(
+            default_conf.turn.db.redis.port,
+            env_conf.turn.db.redis.port
+        );
+
+        assert_eq!(env_conf.turn.db.redis.ip, Ipv4Addr::new(5, 5, 5, 5));
+        assert_eq!(env_conf.turn.db.redis.port, 1234);
+        assert_eq!(
+            env_conf.turn.db.redis.addr(),
+            "5.5.5.5:1234".parse().unwrap()
+        );
+    }
+
+    #[test]
+    #[serial]
+    fn turn_conf_test() {
+        let default_conf = Conf::default();
+
+        env::set_var("MEDEA_TURN.IP", "5.5.5.5");
+        env::set_var("MEDEA_TURN.PORT", "1234");
+
+        let env_conf = Conf::parse().unwrap();
+
+        assert_ne!(default_conf.turn.ip, env_conf.turn.ip);
+        assert_ne!(default_conf.turn.port, env_conf.turn.port);
+
+        assert_eq!(env_conf.turn.ip, Ipv4Addr::new(5, 5, 5, 5));
+        assert_eq!(env_conf.turn.port, 1234);
+        assert_eq!(env_conf.turn.addr(), "5.5.5.5:1234".parse().unwrap());
+    }
 }