summaryrefslogtreecommitdiff
path: root/src/Erebos
diff options
context:
space:
mode:
Diffstat (limited to 'src/Erebos')
-rw-r--r--src/Erebos/Discovery.hs135
-rw-r--r--src/Erebos/ICE.chs41
-rw-r--r--src/Erebos/ICE/pjproject.c111
-rw-r--r--src/Erebos/ICE/pjproject.h7
-rw-r--r--src/Erebos/Service.hs8
5 files changed, 230 insertions, 72 deletions
diff --git a/src/Erebos/Discovery.hs b/src/Erebos/Discovery.hs
index 6422a59..1ba11c5 100644
--- a/src/Erebos/Discovery.hs
+++ b/src/Erebos/Discovery.hs
@@ -2,6 +2,7 @@
module Erebos.Discovery (
DiscoveryService(..),
+ DiscoveryAttributes(..),
DiscoveryConnection(..)
) where
@@ -16,6 +17,7 @@ import Data.Map.Strict qualified as M
import Data.Maybe
import Data.Text (Text)
import Data.Text qualified as T
+import Data.Word
import Network.Socket
@@ -28,12 +30,28 @@ import Erebos.Service
import Erebos.Storage
-data DiscoveryService = DiscoverySelf [ Text ] (Maybe Int)
- | DiscoveryAcknowledged Text
- | DiscoverySearch Ref
- | DiscoveryResult Ref [ Text ]
- | DiscoveryConnectionRequest DiscoveryConnection
- | DiscoveryConnectionResponse DiscoveryConnection
+data DiscoveryService
+ = DiscoverySelf [ Text ] (Maybe Int)
+ | DiscoveryAcknowledged [ Text ] (Maybe Text) (Maybe Word16) (Maybe Text) (Maybe Word16)
+ | DiscoverySearch Ref
+ | DiscoveryResult Ref [ Text ]
+ | DiscoveryConnectionRequest DiscoveryConnection
+ | DiscoveryConnectionResponse DiscoveryConnection
+
+data DiscoveryAttributes = DiscoveryAttributes
+ { discoveryStunPort :: Maybe Word16
+ , discoveryStunServer :: Maybe Text
+ , discoveryTurnPort :: Maybe Word16
+ , discoveryTurnServer :: Maybe Text
+ }
+
+defaultDiscoveryAttributes :: DiscoveryAttributes
+defaultDiscoveryAttributes = DiscoveryAttributes
+ { discoveryStunPort = Nothing
+ , discoveryStunServer = Nothing
+ , discoveryTurnPort = Nothing
+ , discoveryTurnServer = Nothing
+ }
data DiscoveryConnection = DiscoveryConnection
{ dconnSource :: Ref
@@ -58,8 +76,13 @@ instance Storable DiscoveryService where
DiscoverySelf addrs priority -> do
mapM_ (storeText "self") addrs
mapM_ (storeInt "priority") priority
- DiscoveryAcknowledged addr -> do
- storeText "ack" addr
+ DiscoveryAcknowledged addrs stunServer stunPort turnServer turnPort -> do
+ if null addrs then storeEmpty "ack"
+ else mapM_ (storeText "ack") addrs
+ storeMbText "stun-server" stunServer
+ storeMbInt "stun-port" stunPort
+ storeMbText "turn-server" turnServer
+ storeMbInt "turn-port" turnPort
DiscoverySearch ref -> storeRawRef "search" ref
DiscoveryResult ref addr -> do
storeRawRef "result" ref
@@ -82,8 +105,16 @@ instance Storable DiscoveryService where
guard (not $ null addrs)
DiscoverySelf addrs
<$> loadMbInt "priority"
- , DiscoveryAcknowledged
- <$> loadText "ack"
+ , do
+ addrs <- loadTexts "ack"
+ mbEmpty <- loadMbEmpty "ack"
+ guard (not (null addrs) || isJust mbEmpty)
+ DiscoveryAcknowledged
+ <$> pure addrs
+ <*> loadMbText "stun-server"
+ <*> loadMbInt "stun-port"
+ <*> loadMbText "turn-server"
+ <*> loadMbInt "turn-port"
, DiscoverySearch <$> loadRawRef "search"
, DiscoveryResult
<$> loadRawRef "result"
@@ -114,6 +145,14 @@ data DiscoveryPeer = DiscoveryPeer
instance Service DiscoveryService where
serviceID _ = mkServiceID "dd59c89c-69cc-4703-b75b-4ddcd4b3c23c"
+ type ServiceAttributes DiscoveryService = DiscoveryAttributes
+ defaultServiceAttributes _ = defaultDiscoveryAttributes
+
+#ifdef ENABLE_ICE_SUPPORT
+ type ServiceState DiscoveryService = Maybe IceConfig
+ emptyServiceState _ = Nothing
+#endif
+
type ServiceGlobalState DiscoveryService = Map RefDigest DiscoveryPeer
emptyServiceGlobalState _ = M.empty
@@ -123,7 +162,7 @@ instance Service DiscoveryService where
peer <- asks svcPeer
let insertHelper new old | dpPriority new > dpPriority old = new
| otherwise = old
- mbaddr <- fmap (listToMaybe . catMaybes) $ forM addrs $ \addr -> case words (T.unpack addr) of
+ matchedAddrs <- fmap catMaybes $ forM addrs $ \addr -> case words (T.unpack addr) of
[ipaddr, port] | DatagramAddress paddr <- peerAddress peer -> do
saddr <- liftIO $ head <$> getAddrInfo (Just $ defaultHints { addrSocketType = Datagram }) (Just ipaddr) (Just port)
return $ if paddr == addrAddress saddr
@@ -139,9 +178,40 @@ instance Service DiscoveryService where
, dpIceSession = Nothing
#endif
}
- replyPacket $ DiscoveryAcknowledged $ fromMaybe (T.pack "ICE") mbaddr
-
- DiscoveryAcknowledged _ -> do
+ let matchedAddrs' = matchedAddrs
+#ifdef ENABLE_ICE_SUPPORT
+ ++ filter (== T.pack "ICE") addrs
+#endif
+ attrs <- asks svcAttributes
+ replyPacket $ DiscoveryAcknowledged matchedAddrs'
+ (discoveryStunServer attrs)
+ (discoveryStunPort attrs)
+ (discoveryTurnServer attrs)
+ (discoveryTurnPort attrs)
+
+ DiscoveryAcknowledged addrs stunServer stunPort turnServer turnPort -> do
+#ifdef ENABLE_ICE_SUPPORT
+ when (T.pack "ICE" `elem` addrs) $ do
+ paddr <- asks (peerAddress . svcPeer) >>= return . \case
+ (DatagramAddress saddr) -> case IP.fromSockAddr saddr of
+ Just (IP.IPv6 ipv6, _)
+ | (0, 0, 0xffff, ipv4) <- IP.fromIPv6w ipv6
+ -> Just $ T.pack $ show (IP.toIPv4w ipv4)
+ Just (addr, _)
+ -> Just $ T.pack $ show addr
+ _ -> Nothing
+ _ -> Nothing
+
+ let toIceServer Nothing Nothing = Nothing
+ toIceServer Nothing (Just port) = ( , port) <$> paddr
+ toIceServer (Just server) Nothing = Just ( server, 0 )
+ toIceServer (Just server) (Just port) = Just ( server, port )
+
+ cfg <- liftIO $ iceCreateConfig
+ (toIceServer stunServer stunPort)
+ (toIceServer turnServer turnPort)
+ svcSet cfg
+#endif
return ()
DiscoverySearch ref -> do
@@ -155,13 +225,16 @@ instance Service DiscoveryService where
-- TODO: check if we really requested that
server <- asks svcServer
self <- svcSelf
+ mbIceConfig <- svcGet
discoveryPeer <- asks svcPeer
let runAsService = runPeerService @DiscoveryService discoveryPeer
liftIO $ void $ forkIO $ forM_ addrs $ \addr -> if
- | addr == T.pack "ICE" -> do
+ | addr == T.pack "ICE"
#ifdef ENABLE_ICE_SUPPORT
- ice <- iceCreate PjIceSessRoleControlling $ \ice -> do
+ , Just config <- mbIceConfig
+ -> do
+ ice <- iceCreateSession config PjIceSessRoleControlling $ \ice -> do
rinfo <- iceRemoteInfo ice
res <- runExceptT $ sendToPeer discoveryPeer $
DiscoveryConnectionRequest (emptyConnection (storedRef $ idData self) ref) { dconnIceSession = Just rinfo }
@@ -177,6 +250,7 @@ instance Service DiscoveryService where
, dpIceSession = Just ice
}
#else
+ -> do
return ()
#endif
@@ -207,15 +281,19 @@ instance Service DiscoveryService where
-- request for us, create ICE sesssion
server <- asks svcServer
peer <- asks svcPeer
- liftIO $ void $ iceCreate PjIceSessRoleControlled $ \ice -> do
- rinfo <- iceRemoteInfo ice
- res <- runExceptT $ sendToPeer peer $ DiscoveryConnectionResponse rconn { dconnIceSession = Just rinfo }
- case res of
- Right _ -> do
- case dconnIceSession conn of
- Just prinfo -> iceConnect ice prinfo $ void $ serverPeerIce server ice
- Nothing -> putStrLn $ "Discovery: connection request without ICE remote info"
- Left err -> putStrLn $ "Discovery: failed to send connection response: " ++ err
+ svcGet >>= \case
+ Just config -> do
+ liftIO $ void $ iceCreateSession config PjIceSessRoleControlled $ \ice -> do
+ rinfo <- iceRemoteInfo ice
+ res <- runExceptT $ sendToPeer peer $ DiscoveryConnectionResponse rconn { dconnIceSession = Just rinfo }
+ case res of
+ Right _ -> do
+ case dconnIceSession conn of
+ Just prinfo -> iceConnect ice prinfo $ void $ serverPeerIce server ice
+ Nothing -> putStrLn $ "Discovery: connection request without ICE remote info"
+ Left err -> putStrLn $ "Discovery: failed to send connection response: " ++ err
+ Nothing -> do
+ svcPrint $ "Discovery: ICE request from peer without ICE configuration"
else do
-- request to some of our peers, relay
@@ -270,6 +348,11 @@ instance Service DiscoveryService where
let addrToText saddr = do
( addr, port ) <- IP.fromSockAddr saddr
Just $ T.pack $ show addr <> " " <> show port
- addrs <- catMaybes . map addrToText <$> liftIO (getServerAddresses server)
+ addrs <- concat <$> sequence
+ [ catMaybes . map addrToText <$> liftIO (getServerAddresses server)
+#ifdef ENABLE_ICE_SUPPORT
+ , return [ T.pack "ICE" ]
+#endif
+ ]
sendToPeer peer $ DiscoverySelf addrs Nothing
diff --git a/src/Erebos/ICE.chs b/src/Erebos/ICE.chs
index 096ee0d..2a6c174 100644
--- a/src/Erebos/ICE.chs
+++ b/src/Erebos/ICE.chs
@@ -4,9 +4,11 @@
module Erebos.ICE (
IceSession,
IceSessionRole(..),
+ IceConfig,
IceRemoteInfo,
- iceCreate,
+ iceCreateConfig,
+ iceCreateSession,
iceDestroy,
iceRemoteInfo,
iceShow,
@@ -23,17 +25,19 @@ import Control.Monad.Except
import Control.Monad.Identity
import Data.ByteString (ByteString, packCStringLen, useAsCString)
-import qualified Data.ByteString.Lazy.Char8 as BLC
+import Data.ByteString.Lazy.Char8 qualified as BLC
import Data.ByteString.Unsafe
import Data.Function
import Data.Text (Text)
-import qualified Data.Text as T
-import qualified Data.Text.Encoding as T
-import qualified Data.Text.Read as T
+import Data.Text qualified as T
+import Data.Text.Encoding qualified as T
+import Data.Text.Read qualified as T
import Data.Void
+import Data.Word
import Foreign.C.String
import Foreign.C.Types
+import Foreign.ForeignPtr
import Foreign.Marshal.Alloc
import Foreign.Marshal.Array
import Foreign.Ptr
@@ -46,6 +50,7 @@ import Erebos.Storage
data IceSession = IceSession
{ isStrans :: PjIceStrans
+ , _isConfig :: IceConfig
, isChan :: MVar (Either [ByteString] (Flow Void ByteString))
}
@@ -116,14 +121,34 @@ instance StorableText IceCandidate where
{#enum pj_ice_sess_role as IceSessionRole {underscoreToCase} deriving (Show, Eq) #}
+data PjIceStransCfg
+newtype IceConfig = IceConfig (ForeignPtr PjIceStransCfg)
+
+foreign import ccall unsafe "pjproject.h &ice_cfg_free"
+ ice_cfg_free :: FunPtr (Ptr PjIceStransCfg -> IO ())
+foreign import ccall unsafe "pjproject.h ice_cfg_create"
+ ice_cfg_create :: CString -> Word16 -> CString -> Word16 -> IO (Ptr PjIceStransCfg)
+
+iceCreateConfig :: Maybe ( Text, Word16 ) -> Maybe ( Text, Word16 ) -> IO (Maybe IceConfig)
+iceCreateConfig stun turn =
+ maybe ($ nullPtr) (withText . fst) stun $ \cstun ->
+ maybe ($ nullPtr) (withText . fst) turn $ \cturn -> do
+ cfg <- ice_cfg_create cstun (maybe 0 snd stun) cturn (maybe 0 snd turn)
+ if cfg == nullPtr
+ then return Nothing
+ else Just . IceConfig <$> newForeignPtr ice_cfg_free cfg
+
{#pointer *pj_ice_strans as ^ #}
-iceCreate :: IceSessionRole -> (IceSession -> IO ()) -> IO IceSession
-iceCreate role cb = do
+iceCreateSession :: IceConfig -> IceSessionRole -> (IceSession -> IO ()) -> IO IceSession
+iceCreateSession icfg@(IceConfig fcfg) role cb = do
rec sptr <- newStablePtr sess
cbptr <- newStablePtr $ cb sess
sess <- IceSession
- <$> {#call ice_create #} (fromIntegral $ fromEnum role) (castStablePtrToPtr sptr) (castStablePtrToPtr cbptr)
+ <$> (withForeignPtr fcfg $ \cfg ->
+ {#call ice_create #} (castPtr cfg) (fromIntegral $ fromEnum role) (castStablePtrToPtr sptr) (castStablePtrToPtr cbptr)
+ )
+ <*> pure icfg
<*> (newMVar $ Left [])
return $ sess
diff --git a/src/Erebos/ICE/pjproject.c b/src/Erebos/ICE/pjproject.c
index d3037bf..2374340 100644
--- a/src/Erebos/ICE/pjproject.c
+++ b/src/Erebos/ICE/pjproject.c
@@ -12,7 +12,6 @@ static struct
{
pj_caching_pool cp;
pj_pool_t * pool;
- pj_ice_strans_cfg cfg;
pj_sockaddr def_addr;
} ice;
@@ -31,9 +30,9 @@ static void ice_perror(const char * msg, pj_status_t status)
fprintf(stderr, "ICE: %s: %s\n", msg, err);
}
-static int ice_worker_thread(void * unused)
+static int ice_worker_thread(void * vcfg)
{
- PJ_UNUSED_ARG(unused);
+ pj_ice_strans_cfg * cfg = (pj_ice_strans_cfg *) vcfg;
while (true) {
pj_time_val max_timeout = { 0, 0 };
@@ -41,7 +40,7 @@ static int ice_worker_thread(void * unused)
max_timeout.msec = 500;
- pj_timer_heap_poll(ice.cfg.stun_cfg.timer_heap, &timeout);
+ pj_timer_heap_poll(cfg->stun_cfg.timer_heap, &timeout);
pj_assert(timeout.sec >= 0 && timeout.msec >= 0);
if (timeout.msec >= 1000)
@@ -50,7 +49,7 @@ static int ice_worker_thread(void * unused)
if (PJ_TIME_VAL_GT(timeout, max_timeout))
timeout = max_timeout;
- int c = pj_ioqueue_poll(ice.cfg.stun_cfg.ioqueue, &timeout);
+ int c = pj_ioqueue_poll(cfg->stun_cfg.ioqueue, &timeout);
if (c < 0)
pj_thread_sleep(PJ_TIME_VAL_MSEC(timeout));
}
@@ -105,7 +104,7 @@ static void ice_init(void)
if (done) {
pthread_mutex_unlock(&mutex);
- goto exit;
+ return;
}
pj_log_set_level(1);
@@ -125,48 +124,88 @@ static void ice_init(void)
pj_caching_pool_init(&ice.cp, NULL, 0);
- pj_ice_strans_cfg_default(&ice.cfg);
- ice.cfg.stun_cfg.pf = &ice.cp.factory;
-
ice.pool = pj_pool_create(&ice.cp.factory, "ice", 512, 512, NULL);
- if (pj_timer_heap_create(ice.pool, 100,
- &ice.cfg.stun_cfg.timer_heap) != PJ_SUCCESS) {
- fprintf(stderr, "pj_timer_heap_create failed\n");
- goto exit;
+exit:
+ done = true;
+ pthread_mutex_unlock(&mutex);
+}
+
+pj_ice_strans_cfg * ice_cfg_create( const char * stun_server, uint16_t stun_port,
+ const char * turn_server, uint16_t turn_port )
+{
+ ice_init();
+
+ pj_ice_strans_cfg * cfg = malloc( sizeof(pj_ice_strans_cfg) );
+ pj_ice_strans_cfg_default( cfg );
+
+ cfg->stun_cfg.pf = &ice.cp.factory;
+ if( pj_timer_heap_create( ice.pool, 100,
+ &cfg->stun_cfg.timer_heap ) != PJ_SUCCESS ){
+ fprintf( stderr, "pj_timer_heap_create failed\n" );
+ goto fail;
}
- if (pj_ioqueue_create(ice.pool, 16, &ice.cfg.stun_cfg.ioqueue) != PJ_SUCCESS) {
- fprintf(stderr, "pj_ioqueue_create failed\n");
- goto exit;
+ if( pj_ioqueue_create( ice.pool, 16, &cfg->stun_cfg.ioqueue ) != PJ_SUCCESS ){
+ fprintf( stderr, "pj_ioqueue_create failed\n" );
+ goto fail;
}
pj_thread_t * thread;
- if (pj_thread_create(ice.pool, "ice", &ice_worker_thread,
- NULL, 0, 0, &thread) != PJ_SUCCESS) {
- fprintf(stderr, "pj_thread_create failed\n");
- goto exit;
+ if( pj_thread_create( ice.pool, NULL, &ice_worker_thread,
+ cfg, 0, 0, &thread ) != PJ_SUCCESS ){
+ fprintf( stderr, "pj_thread_create failed\n" );
+ goto fail;
}
- ice.cfg.af = pj_AF_INET();
- ice.cfg.opt.aggressive = PJ_TRUE;
+ cfg->af = pj_AF_INET();
+ cfg->opt.aggressive = PJ_TRUE;
- ice.cfg.stun.server.ptr = "discovery1.erebosprotocol.net";
- ice.cfg.stun.server.slen = strlen(ice.cfg.stun.server.ptr);
- ice.cfg.stun.port = 29670;
+ if( stun_server ){
+ cfg->stun.server.ptr = malloc( strlen( stun_server ));
+ pj_strcpy2( &cfg->stun.server, stun_server );
+ if( stun_port )
+ cfg->stun.port = stun_port;
+ }
- ice.cfg.turn.server = ice.cfg.stun.server;
- ice.cfg.turn.port = ice.cfg.stun.port;
- ice.cfg.turn.auth_cred.type = PJ_STUN_AUTH_CRED_STATIC;
- ice.cfg.turn.auth_cred.data.static_cred.data_type = PJ_STUN_PASSWD_PLAIN;
- ice.cfg.turn.conn_type = PJ_TURN_TP_UDP;
+ if( turn_server ){
+ cfg->turn.server.ptr = malloc( strlen( turn_server ));
+ pj_strcpy2( &cfg->turn.server, turn_server );
+ if( turn_port )
+ cfg->turn.port = turn_port;
+ cfg->turn.auth_cred.type = PJ_STUN_AUTH_CRED_STATIC;
+ cfg->turn.auth_cred.data.static_cred.data_type = PJ_STUN_PASSWD_PLAIN;
+ cfg->turn.conn_type = PJ_TURN_TP_UDP;
+ }
-exit:
- done = true;
- pthread_mutex_unlock(&mutex);
+ return cfg;
+fail:
+ ice_cfg_free( cfg );
+ return NULL;
+}
+
+void ice_cfg_free( pj_ice_strans_cfg * cfg )
+{
+ if( ! cfg )
+ return;
+
+ if( cfg->turn.server.ptr )
+ free( cfg->turn.server.ptr );
+
+ if( cfg->stun.server.ptr )
+ free( cfg->stun.server.ptr );
+
+ if( cfg->stun_cfg.ioqueue )
+ pj_ioqueue_destroy( cfg->stun_cfg.ioqueue );
+
+ if( cfg->stun_cfg.timer_heap )
+ pj_timer_heap_destroy( cfg->stun_cfg.timer_heap );
+
+ free( cfg );
}
-pj_ice_strans * ice_create(pj_ice_sess_role role, HsStablePtr sptr, HsStablePtr cb)
+pj_ice_strans * ice_create( const pj_ice_strans_cfg * cfg, pj_ice_sess_role role,
+ HsStablePtr sptr, HsStablePtr cb )
{
ice_init();
@@ -182,8 +221,8 @@ pj_ice_strans * ice_create(pj_ice_sess_role role, HsStablePtr sptr, HsStablePtr
.on_ice_complete = cb_on_ice_complete,
};
- pj_status_t status = pj_ice_strans_create(NULL, &ice.cfg, 1,
- udata, &icecb, &res);
+ pj_status_t status = pj_ice_strans_create( NULL, cfg, 1,
+ udata, &icecb, &res );
if (status != PJ_SUCCESS)
ice_perror("error creating ice", status);
diff --git a/src/Erebos/ICE/pjproject.h b/src/Erebos/ICE/pjproject.h
index e230e75..e4fcbdb 100644
--- a/src/Erebos/ICE/pjproject.h
+++ b/src/Erebos/ICE/pjproject.h
@@ -3,7 +3,12 @@
#include <pjnath.h>
#include <HsFFI.h>
-pj_ice_strans * ice_create(pj_ice_sess_role role, HsStablePtr sptr, HsStablePtr cb);
+pj_ice_strans_cfg * ice_cfg_create( const char * stun_server, uint16_t stun_port,
+ const char * turn_server, uint16_t turn_port );
+void ice_cfg_free( pj_ice_strans_cfg * cfg );
+
+pj_ice_strans * ice_create( const pj_ice_strans_cfg *, pj_ice_sess_role role,
+ HsStablePtr sptr, HsStablePtr cb );
void ice_destroy(pj_ice_strans * strans);
ssize_t ice_encode_session(pj_ice_strans *, char * ufrag, char * pass,
diff --git a/src/Erebos/Service.hs b/src/Erebos/Service.hs
index f8428d1..d1943e1 100644
--- a/src/Erebos/Service.hs
+++ b/src/Erebos/Service.hs
@@ -37,7 +37,13 @@ import {-# SOURCE #-} Erebos.Network
import Erebos.State
import Erebos.Storage
-class (Typeable s, Storable s, Typeable (ServiceState s), Typeable (ServiceGlobalState s)) => Service s where
+class (
+ Typeable s, Storable s,
+ Typeable (ServiceAttributes s),
+ Typeable (ServiceState s),
+ Typeable (ServiceGlobalState s)
+ ) => Service s where
+
serviceID :: proxy s -> ServiceID
serviceHandler :: Stored s -> ServiceHandler s ()
generated by cgit v1.2.3 (git 2.25.1) at 2025-04-01 17:59:29 +0000