diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/ICE.chs | 200 | ||||
| -rw-r--r-- | src/ICE/pjproject.c | 363 | ||||
| -rw-r--r-- | src/ICE/pjproject.h | 14 | ||||
| -rw-r--r-- | src/Main.hs | 61 | ||||
| -rw-r--r-- | src/Network.hs | 106 | ||||
| -rw-r--r-- | src/Storage.hs | 8 |
6 files changed, 710 insertions, 42 deletions
diff --git a/src/ICE.chs b/src/ICE.chs new file mode 100644 index 0000000..06ad7aa --- /dev/null +++ b/src/ICE.chs @@ -0,0 +1,200 @@ +{-# LANGUAGE ForeignFunctionInterface #-} +{-# LANGUAGE RecursiveDo #-} + +module ICE ( + IceSession, + IceSessionRole(..), + IceRemoteInfo, + + iceCreate, + iceDestroy, + iceRemoteInfo, + iceShow, + iceConnect, + iceSend, + + iceSetChan, +) where + +import Control.Arrow +import Control.Concurrent.Chan +import Control.Concurrent.MVar +import Control.Monad +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.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 Foreign.C.String +import Foreign.C.Types +import Foreign.Marshal.Alloc +import Foreign.Marshal.Array +import Foreign.Ptr +import Foreign.StablePtr + +import Storage + +#include "pjproject.h" + +data IceSession = IceSession + { isStrans :: PjIceStrans + , isChan :: MVar (Either [ByteString] (MappedChan ByteString)) + } + +data MappedChan a = forall b. MappedChan (a -> b) (Chan b) + +instance Eq IceSession where + (==) = (==) `on` isStrans + +instance Ord IceSession where + compare = compare `on` isStrans + +instance Show IceSession where + show _ = "<ICE>" + + +data IceRemoteInfo = IceRemoteInfo + { iriUsernameFrament :: Text + , iriPassword :: Text + , iriDefaultCandidate :: Text + , iriCandidates :: [Text] + } + +data IceCandidate = IceCandidate + { icandFoundation :: Text + , icandPriority :: Int + , icandAddr :: Text + , icandPort :: Int + , icandType :: Text + } + +instance Storable IceRemoteInfo where + store' x = storeRec $ do + storeText "ice-ufrag" $ iriUsernameFrament x + storeText "ice-pass" $ iriPassword x + storeText "ice-default" $ iriDefaultCandidate x + mapM_ (storeText "ice-candidate") $ iriCandidates x + + load' = loadRec $ IceRemoteInfo + <$> loadText "ice-ufrag" + <*> loadText "ice-pass" + <*> loadText "ice-default" + <*> loadTexts "ice-candidate" + +instance StorableText IceCandidate where + toText x = T.concat $ + [ icandFoundation x + , T.singleton ' ' + , T.pack $ show $ icandPriority x + , T.singleton ' ' + , icandAddr x + , T.singleton ' ' + , T.pack $ show $ icandPort x + , T.singleton ' ' + , icandType x + ] + + fromText t = case T.words t of + [found, tprio, addr, tport, ctype] + | Right (prio, _) <- T.decimal tprio + , Right (port, _) <- T.decimal tport + -> return $ IceCandidate + { icandFoundation = found + , icandPriority = prio + , icandAddr = addr + , icandPort = port + , icandType = ctype + } + _ -> throwError "failed to parse candidate" + + +{#enum pj_ice_sess_role as IceSessionRole {underscoreToCase} deriving (Show, Eq) #} + +{#pointer *pj_ice_strans as ^ #} + +iceCreate :: IceSessionRole -> (IceSession -> IO ()) -> IO IceSession +iceCreate role cb = do + rec sptr <- newStablePtr sess + cbptr <- newStablePtr $ cb sess + sess <- IceSession + <$> {#call ice_create #} (fromIntegral $ fromEnum role) (castStablePtrToPtr sptr) (castStablePtrToPtr cbptr) + <*> (newMVar $ Left []) + return $ sess + +{#fun ice_destroy as ^ { isStrans `IceSession' } -> `()' #} + +iceRemoteInfo :: IceSession -> IO IceRemoteInfo +iceRemoteInfo sess = + allocaBytes (32*128) $ \bytes -> + allocaArray 29 $ \carr -> do + let (ufrag : pass : def : cptrs) = take 32 $ iterate (`plusPtr` 128) bytes + pokeArray carr cptrs + + ncand <- {#call ice_encode_session #} (isStrans sess) ufrag pass def carr 128 29 + if ncand < 0 then fail "failed to generate ICE remote info" + else IceRemoteInfo + <$> (T.pack <$> peekCString ufrag) + <*> (T.pack <$> peekCString pass) + <*> (T.pack <$> peekCString def) + <*> (mapM (return . T.pack <=< peekCString) $ take (fromIntegral ncand) cptrs) + +iceShow :: IceSession -> IO String +iceShow sess = do + st <- memoryStorage + return . drop 1 . dropWhile (/='\n') . BLC.unpack . runIdentity =<< + ioLoadBytes =<< store st =<< iceRemoteInfo sess + +iceConnect :: IceSession -> IceRemoteInfo -> (IO ()) -> IO () +iceConnect sess remote cb = do + cbptr <- newStablePtr $ cb + ice_connect sess cbptr + (iriUsernameFrament remote) + (iriPassword remote) + (iriDefaultCandidate remote) + (iriCandidates remote) + +{#fun ice_connect { isStrans `IceSession', castStablePtrToPtr `StablePtr (IO ())', + withText* `Text', withText* `Text', withText* `Text', withTextArray* `[Text]'& } -> `()' #} + +withText :: Text -> (Ptr CChar -> IO a) -> IO a +withText t f = useAsCString (T.encodeUtf8 t) f + +withTextArray :: Num n => [Text] -> ((Ptr (Ptr CChar), n) -> IO ()) -> IO () +withTextArray tsAll f = helper tsAll [] + where helper (t:ts) bs = withText t $ \b -> helper ts (b:bs) + helper [] bs = allocaArray (length bs) $ \ptr -> do + pokeArray ptr $ reverse bs + f (ptr, fromIntegral $ length bs) + +withByteStringLen :: Num n => ByteString -> ((Ptr CChar, n) -> IO a) -> IO a +withByteStringLen t f = unsafeUseAsCStringLen t (f . (id *** fromIntegral)) + +{#fun ice_send as ^ { isStrans `IceSession', withByteStringLen* `ByteString'& } -> `()' #} + +foreign export ccall ice_call_cb :: StablePtr (IO ()) -> IO () +ice_call_cb :: StablePtr (IO ()) -> IO () +ice_call_cb = join . deRefStablePtr + +iceSetChan :: IceSession -> (ByteString -> a) -> Chan a -> IO () +iceSetChan sess f chan = do + modifyMVar_ (isChan sess) $ \orig -> do + case orig of + Left buf -> writeList2Chan chan $ map f $ reverse buf + Right _ -> return () + return $ Right $ MappedChan f chan + +foreign export ccall ice_rx_data :: StablePtr IceSession -> Ptr CChar -> Int -> IO () +ice_rx_data :: StablePtr IceSession -> Ptr CChar -> Int -> IO () +ice_rx_data sptr buf len = do + sess <- deRefStablePtr sptr + bs <- packCStringLen (buf, len) + modifyMVar_ (isChan sess) $ \case + mc@(Right (MappedChan f chan)) -> writeChan chan (f bs) >> return mc + Left bss -> return $ Left (bs:bss) diff --git a/src/ICE/pjproject.c b/src/ICE/pjproject.c new file mode 100644 index 0000000..0ae69e9 --- /dev/null +++ b/src/ICE/pjproject.c @@ -0,0 +1,363 @@ +#include "pjproject.h" +#include "ICE_stub.h" + +#include <stdio.h> +#include <stdlib.h> +#include <stdbool.h> +#include <pthread.h> +#include <pjlib.h> +#include <pjlib-util.h> + +static struct +{ + pj_caching_pool cp; + pj_pool_t * pool; + pj_ice_strans_cfg cfg; + pj_sockaddr def_addr; +} ice; + +struct user_data +{ + pj_ice_sess_role role; + HsStablePtr sptr; + HsStablePtr cb_init; + HsStablePtr cb_connect; +}; + +static void ice_perror(const char * msg, pj_status_t status) +{ + char err[PJ_ERR_MSG_SIZE]; + pj_strerror(status, err, sizeof(err)); + fprintf(stderr, "ICE: %s: %s\n", msg, err); +} + +static int ice_worker_thread(void * unused) +{ + PJ_UNUSED_ARG(unused); + + while (true) { + pj_time_val max_timeout = { 0, 0 }; + pj_time_val timeout = { 0, 0 }; + + max_timeout.msec = 500; + + pj_timer_heap_poll(ice.cfg.stun_cfg.timer_heap, &timeout); + + pj_assert(timeout.sec >= 0 && timeout.msec >= 0); + if (timeout.msec >= 1000) + timeout.msec = 999; + + if (PJ_TIME_VAL_GT(timeout, max_timeout)) + timeout = max_timeout; + + int c = pj_ioqueue_poll(ice.cfg.stun_cfg.ioqueue, &timeout); + if (c < 0) + pj_thread_sleep(PJ_TIME_VAL_MSEC(timeout)); + } + + return 0; +} + +static void cb_on_rx_data(pj_ice_strans * strans, unsigned comp_id, + void * pkt, pj_size_t size, + const pj_sockaddr_t * src_addr, unsigned src_addr_len) +{ + struct user_data * udata = pj_ice_strans_get_user_data(strans); + ice_rx_data(udata->sptr, pkt, size); +} + +static void cb_on_ice_complete(pj_ice_strans * strans, + pj_ice_strans_op op, pj_status_t status) +{ + if (status != PJ_SUCCESS) { + ice_perror("cb_on_ice_complete", status); + ice_destroy(strans); + return; + } + + struct user_data * udata = pj_ice_strans_get_user_data(strans); + if (op == PJ_ICE_STRANS_OP_INIT) { + pj_status_t istatus = pj_ice_strans_init_ice(strans, udata->role, NULL, NULL); + if (istatus != PJ_SUCCESS) + ice_perror("error creating session", istatus); + + if (udata->cb_init) { + ice_call_cb(udata->cb_init); + hs_free_stable_ptr(udata->cb_init); + udata->cb_init = NULL; + } + } + + if (op == PJ_ICE_STRANS_OP_NEGOTIATION) { + if (udata->cb_connect) { + ice_call_cb(udata->cb_connect); + hs_free_stable_ptr(udata->cb_connect); + udata->cb_connect = NULL; + } + } +} + +static void ice_init(void) +{ + static bool done = false; + static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; + pthread_mutex_lock(&mutex); + + if (done) { + pthread_mutex_unlock(&mutex); + goto exit; + } + + pj_log_set_level(1); + + if (pj_init() != PJ_SUCCESS) { + fprintf(stderr, "pj_init failed\n"); + goto exit; + } + if (pjlib_util_init() != PJ_SUCCESS) { + fprintf(stderr, "pjlib_util_init failed\n"); + goto exit; + } + if (pjnath_init() != PJ_SUCCESS) { + fprintf(stderr, "pjnath_init failed\n"); + goto exit; + } + + 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; + } + + if (pj_ioqueue_create(ice.pool, 16, &ice.cfg.stun_cfg.ioqueue) != PJ_SUCCESS) { + fprintf(stderr, "pj_ioqueue_create failed\n"); + goto exit; + } + + 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; + } + + ice.cfg.af = pj_AF_INET(); + ice.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; + + 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; + +exit: + done = true; + pthread_mutex_unlock(&mutex); +} + +pj_ice_strans * ice_create(pj_ice_sess_role role, HsStablePtr sptr, HsStablePtr cb) +{ + ice_init(); + + pj_ice_strans * res; + + struct user_data * udata = malloc(sizeof(struct user_data)); + udata->role = role; + udata->sptr = sptr; + udata->cb_init = cb; + + pj_ice_strans_cb icecb = { + .on_rx_data = cb_on_rx_data, + .on_ice_complete = cb_on_ice_complete, + }; + + pj_status_t status = pj_ice_strans_create(NULL, &ice.cfg, 1, + udata, &icecb, &res); + + if (status != PJ_SUCCESS) + ice_perror("error creating ice", status); + + return res; +} + +void ice_destroy(pj_ice_strans * strans) +{ + struct user_data * udata = pj_ice_strans_get_user_data(strans); + if (udata->sptr) + hs_free_stable_ptr(udata->sptr); + if (udata->cb_init) + hs_free_stable_ptr(udata->cb_init); + if (udata->cb_connect) + hs_free_stable_ptr(udata->cb_connect); + free(udata); + + pj_ice_strans_stop_ice(strans); + pj_ice_strans_destroy(strans); +} + +ssize_t ice_encode_session(pj_ice_strans * strans, char * ufrag, char * pass, + char * def, char * candidates[], size_t maxlen, size_t maxcand) +{ + int n; + pj_str_t local_ufrag, local_pwd; + pj_status_t status; + + pj_ice_strans_get_ufrag_pwd(strans, &local_ufrag, &local_pwd, NULL, NULL); + + n = snprintf(ufrag, maxlen, "%.*s", (int) local_ufrag.slen, local_ufrag.ptr); + if (n < 0 || n == maxlen) + return -PJ_ETOOSMALL; + + n = snprintf(pass, maxlen, "%.*s", (int) local_pwd.slen, local_pwd.ptr); + if (n < 0 || n == maxlen) + return -PJ_ETOOSMALL; + + pj_ice_sess_cand cand[PJ_ICE_ST_MAX_CAND]; + char ipaddr[PJ_INET6_ADDRSTRLEN]; + + status = pj_ice_strans_get_def_cand(strans, 1, &cand[0]); + if (status != PJ_SUCCESS) + return -status; + + n = snprintf(def, maxlen, "%s %d", + pj_sockaddr_print(&cand[0].addr, ipaddr, sizeof(ipaddr), 0), + (int) pj_sockaddr_get_port(&cand[0].addr)); + if (n < 0 || n == maxlen) + return -PJ_ETOOSMALL; + + unsigned cand_cnt = PJ_ARRAY_SIZE(cand); + status = pj_ice_strans_enum_cands(strans, 1, &cand_cnt, cand); + if (status != PJ_SUCCESS) + return -status; + + for (unsigned i = 0; i < cand_cnt && i < maxcand; i++) { + char ipaddr[PJ_INET6_ADDRSTRLEN]; + n = snprintf(candidates[i], maxlen, + "%.*s %u %s %u %s", + (int) cand[i].foundation.slen, cand[i].foundation.ptr, + cand[i].prio, + pj_sockaddr_print(&cand[i].addr, ipaddr, sizeof(ipaddr), 0), + (unsigned) pj_sockaddr_get_port(&cand[i].addr), + pj_ice_get_cand_type_name(cand[i].type)); + + if (n < 0 || n == maxlen) + return -PJ_ETOOSMALL; + } + + return cand_cnt; +} + +void ice_connect(pj_ice_strans * strans, HsStablePtr cb, + const char * ufrag, const char * pass, + const char * defcand, const char * tcandidates[], size_t ncand) +{ + unsigned def_port = 0; + char def_addr[80]; + pj_bool_t done = PJ_FALSE; + char line[256]; + pj_ice_sess_cand candidates[PJ_ICE_ST_MAX_CAND]; + + struct user_data * udata = pj_ice_strans_get_user_data(strans); + udata->cb_connect = cb; + + def_addr[0] = '\0'; + + if (ncand == 0) { + fprintf(stderr, "ICE: no candidates\n"); + return; + } + + int cnt = sscanf(defcand, "%s %u", def_addr, &def_port); + if (cnt != 2) { + fprintf(stderr, "ICE: error parsing default candidate\n"); + return; + } + + int okcand = 0; + for (int i = 0; i < ncand; i++) { + char foundation[32], ipaddr[80], type[32]; + int prio, port; + + int cnt = sscanf(tcandidates[i], "%s %d %s %d %s", + foundation, &prio, + ipaddr, &port, + type); + if (cnt != 5) + continue; + + pj_ice_sess_cand * cand = &candidates[okcand]; + pj_bzero(cand, sizeof(*cand)); + + if (strcmp(type, "host") == 0) + cand->type = PJ_ICE_CAND_TYPE_HOST; + else if (strcmp(type, "srflx") == 0) + cand->type = PJ_ICE_CAND_TYPE_SRFLX; + else if (strcmp(type, "relay") == 0) + cand->type = PJ_ICE_CAND_TYPE_RELAYED; + else + continue; + + cand->comp_id = 1; + pj_strdup2(ice.pool, &cand->foundation, foundation); + cand->prio = prio; + + int af = strchr(ipaddr, ':') ? pj_AF_INET6() : pj_AF_INET(); + pj_str_t tmpaddr = pj_str(ipaddr); + pj_sockaddr_init(af, &cand->addr, NULL, 0); + pj_status_t status = pj_sockaddr_set_str_addr(af, &cand->addr, &tmpaddr); + if (status != PJ_SUCCESS) { + fprintf(stderr, "ICE: invalid IP address \"%s\"\n", ipaddr); + continue; + } + + pj_sockaddr_set_port(&cand->addr, (pj_uint16_t)port); + okcand++; + } + + pj_str_t tmp_addr; + pj_status_t status; + + int af = strchr(def_addr, ':') ? pj_AF_INET6() : pj_AF_INET(); + + pj_sockaddr_init(af, &ice.def_addr, NULL, 0); + tmp_addr = pj_str(def_addr); + status = pj_sockaddr_set_str_addr(af, &ice.def_addr, &tmp_addr); + if (status != PJ_SUCCESS) { + fprintf(stderr, "ICE: invalid default IP address \"%s\"\n", def_addr); + return; + } + pj_sockaddr_set_port(&ice.def_addr, (pj_uint16_t) def_port); + + pj_str_t rufrag, rpwd; + status = pj_ice_strans_start_ice(strans, + pj_cstr(&rufrag, ufrag), pj_cstr(&rpwd, pass), + okcand, candidates); + if (status != PJ_SUCCESS) { + ice_perror("error starting ICE", status); + return; + } +} + +void ice_send(pj_ice_strans * strans, const char * data, size_t len) +{ + if (!pj_ice_strans_sess_is_complete(strans)) { + fprintf(stderr, "ICE: negotiation has not been started or is in progress\n"); + return; + } + + pj_status_t status = pj_ice_strans_sendto(strans, 1, data, len, + &ice.def_addr, pj_sockaddr_get_len(&ice.def_addr)); + if (status != PJ_SUCCESS && status != PJ_EPENDING) + ice_perror("error sending data", status); +} diff --git a/src/ICE/pjproject.h b/src/ICE/pjproject.h new file mode 100644 index 0000000..e230e75 --- /dev/null +++ b/src/ICE/pjproject.h @@ -0,0 +1,14 @@ +#pragma once + +#include <pjnath.h> +#include <HsFFI.h> + +pj_ice_strans * ice_create(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, + char * def, char * candidates[], size_t maxlen, size_t maxcand); +void ice_connect(pj_ice_strans * strans, HsStablePtr cb, + const char * ufrag, const char * pass, + const char * defcand, const char * candidates[], size_t ncand); +void ice_send(pj_ice_strans *, const char * data, size_t len); diff --git a/src/Main.hs b/src/Main.hs index 9404517..0e8970f 100644 --- a/src/Main.hs +++ b/src/Main.hs @@ -25,6 +25,7 @@ import System.Environment import Attach import Contact +import ICE import Identity import Message import Network @@ -95,7 +96,8 @@ interactiveLoop st bhost = runInputT defaultSettings $ do haveTerminalUI >>= \case True -> return () False -> error "Requires terminal" extPrint <- getExternalPrint - let extPrintLn str = extPrint $ str ++ "\n"; + let extPrintLn str = extPrint $ case reverse str of ('\n':_) -> str + _ -> str ++ "\n"; server <- liftIO $ do startServer erebosHead extPrintLn bhost [ SomeService @AttachService Proxy @@ -158,7 +160,10 @@ interactiveLoop st bhost = runInputT defaultSettings $ do let loop (Just cstate) = runMaybeT (process cstate) >>= loop loop Nothing = return () - loop $ Just $ CommandState { csPeer = Nothing } + loop $ Just $ CommandState + { csPeer = Nothing + , csIceSessions = [] + } data CommandInput = CommandInput @@ -171,6 +176,7 @@ data CommandInput = CommandInput data CommandState = CommandState { csPeer :: Maybe Peer + , csIceSessions :: [IceSession] } newtype CommandM a = CommandM (ReaderT CommandInput (StateT CommandState (ExceptT String IO)) a) @@ -195,6 +201,11 @@ commands = , ("contacts", cmdContacts) , ("contact-add", cmdContactAdd) , ("contact-accept", cmdContactAccept) + , ("ice-create", cmdIceCreate) + , ("ice-destroy", cmdIceDestroy) + , ("ice-show", cmdIceShow) + , ("ice-connect", cmdIceConnect) + , ("ice-send", cmdIceSend) ] cmdUnknown :: String -> Command @@ -212,8 +223,7 @@ showPeer peer = PeerIdentityUnknown -> "<noid>" PeerIdentityRef wref -> "<" ++ BC.unpack (showRefDigest $ wrDigest wref) ++ ">" PeerIdentityFull pid -> T.unpack $ displayIdentity pid - DatagramAddress addr = peerAddress peer - in name ++ " [" ++ show addr ++ "]" + in name ++ " [" ++ show (peerAddress peer) ++ "]" cmdSetPeer :: Int -> Command cmdSetPeer n | n < 1 = liftIO $ putStrLn "Invalid peer index" @@ -276,3 +286,46 @@ cmdContactAccept = join $ contactAccept <$> asks ciPrint <*> asks ciHead <*> (maybe (throwError "no peer selected") return =<< gets csPeer) + +cmdIceCreate :: Command +cmdIceCreate = do + role <- asks ciLine >>= return . \case + 'm':_ -> PjIceSessRoleControlling + 's':_ -> PjIceSessRoleControlled + _ -> PjIceSessRoleUnknown + eprint <- asks ciPrint + sess <- liftIO $ iceCreate role $ eprint <=< iceShow + modify $ \s -> s { csIceSessions = sess : csIceSessions s } + +cmdIceDestroy :: Command +cmdIceDestroy = do + s:ss <- gets csIceSessions + modify $ \st -> st { csIceSessions = ss } + liftIO $ iceDestroy s + +cmdIceShow :: Command +cmdIceShow = do + sess <- gets csIceSessions + eprint <- asks ciPrint + liftIO $ forM_ (zip [1::Int ..] sess) $ \(i, s) -> do + eprint $ "[" ++ show i ++ "]" + eprint =<< iceShow s + +cmdIceConnect :: Command +cmdIceConnect = do + s:_ <- gets csIceSessions + server <- asks ciServer + let loadInfo = BC.getLine >>= \case line | BC.null line -> return [] + | otherwise -> (line:) <$> loadInfo + Right remote <- liftIO $ do + st <- memoryStorage + pst <- derivePartialStorage st + rbytes <- (BL.fromStrict . BC.unlines) <$> loadInfo + copyRef st =<< storeRawBytes pst (BL.fromChunks [ BC.pack "rec ", BC.pack (show (BL.length rbytes)), BC.singleton '\n' ] `BL.append` rbytes) + liftIO $ iceConnect s (load remote) $ void $ serverPeerIce server s + +cmdIceSend :: Command +cmdIceSend = void $ do + s:_ <- gets csIceSessions + server <- asks ciServer + liftIO $ serverPeerIce server s diff --git a/src/Network.hs b/src/Network.hs index 5685627..cbc68b6 100644 --- a/src/Network.hs +++ b/src/Network.hs @@ -9,7 +9,7 @@ module Network ( PeerChannel(..), WaitingRef, wrDigest, Service(..), - serverPeer, + serverPeer, serverPeerIce, sendToPeer, sendToPeerStored, sendToPeerWith, discoveryPort, @@ -31,9 +31,10 @@ import Data.Maybe import Data.Typeable import Network.Socket -import Network.Socket.ByteString (recvFrom, sendTo) +import qualified Network.Socket.ByteString as S import Channel +import ICE import Identity import PubKey import Service @@ -53,7 +54,8 @@ data Server = Server { serverStorage :: Storage , serverIdentity :: MVar UnifiedIdentity , serverSocket :: MVar Socket - , serverPeers :: MVar (Map SockAddr Peer) + , serverChanPacket :: Chan (PeerAddress, BC.ByteString) + , serverPeers :: MVar (Map PeerAddress Peer) , serverChanPeer' :: Chan Peer } @@ -66,7 +68,6 @@ data Peer = Peer , peerIdentity :: PeerIdentity , peerIdentityUpdate :: [WaitingRef] , peerChannel :: PeerChannel - , peerSocket :: Socket , peerStorage :: Storage , peerInStorage :: PartialStorage , peerServiceState :: MVar (M.Map ServiceID SomeServiceState) @@ -75,8 +76,24 @@ data Peer = Peer , peerWaitingRefs :: [WaitingRef] } -data PeerAddress = DatagramAddress SockAddr - deriving (Show) +data PeerAddress = DatagramAddress Socket SockAddr + | PeerIceSession IceSession + +instance Show PeerAddress where + show (DatagramAddress _ addr) = show addr + show (PeerIceSession ice) = show ice + +instance Eq PeerAddress where + DatagramAddress _ addr == DatagramAddress _ addr' = addr == addr' + PeerIceSession ice == PeerIceSession ice' = ice == ice' + _ == _ = False + +instance Ord PeerAddress where + compare (DatagramAddress _ addr) (DatagramAddress _ addr') = compare addr addr' + compare (DatagramAddress _ _ ) _ = LT + compare _ (DatagramAddress _ _ ) = GT + compare (PeerIceSession ice ) (PeerIceSession ice') = compare ice ice' + data PeerIdentity = PeerIdentityUnknown | PeerIdentityRef WaitingRef @@ -184,6 +201,7 @@ receivedWaitingRef nref wr@(WaitingRef _ _ mvar) = do startServer :: Head LocalState -> (String -> IO ()) -> String -> [SomeService] -> IO Server startServer origHead logd bhost services = do let storage = refStorage $ headRef origHead + chanPacket <- newChan chanPeer <- newChan chanSvc <- newChan svcStates <- newMVar M.empty @@ -206,7 +224,7 @@ startServer origHead logd bhost services = do readMVar midentity >>= \identity -> do st <- derivePartialStorage storage baddr:_ <- getAddrInfo (Just $ defaultHints { addrSocketType = Datagram }) (Just bhost) (Just discoveryPort) - void $ sendTo sock (BL.toStrict $ serializeObject $ transportToObject $ TransportHeader [ AnnounceSelf $ partialRef st $ storedRef $ idData identity ]) (addrAddress baddr) + void $ S.sendTo sock (BL.toStrict $ serializeObject $ transportToObject $ TransportHeader [ AnnounceSelf $ partialRef st $ storedRef $ idData identity ]) (addrAddress baddr) threadDelay $ announceIntervalSeconds * 1000 * 1000 let announceUpdate identity = do @@ -220,9 +238,8 @@ startServer origHead logd bhost services = do peer | PeerIdentityFull _ <- peerIdentity peer , ChannelEstablished ch <- peerChannel peer - , DatagramAddress paddr <- peerAddress peer -> runExceptT (channelEncrypt ch plaintext) >>= \case - Right ctext -> void $ sendTo (peerSocket peer) ctext paddr + Right ctext -> void $ sendTo peer ctext Left err -> logd $ "Failed to encrypt data: " ++ err | otherwise -> return () @@ -246,8 +263,12 @@ startServer origHead logd bhost services = do when changedShared $ do mapM_ (shareState idt shared) =<< readMVar peers + void $ forkIO $ forever $ do + (msg, saddr) <- S.recvFrom sock 4096 + writeChan chanPacket (DatagramAddress sock saddr, msg) + forever $ do - (msg, paddr) <- recvFrom sock 4096 + (paddr, msg) <- readChan chanPacket modifyMVar_ peers $ \pvalue -> do let mbpeer = M.lookup paddr pvalue (peer, content, secure) <- if @@ -263,7 +284,7 @@ startServer origHead logd bhost services = do -> return (peer, msg, False) | otherwise - -> (, msg, False) <$> mkPeer storage sock paddr + -> (, msg, False) <$> mkPeer storage paddr case runExcept $ deserializeObjects (peerInStorage peer) $ BL.fromStrict content of Right (obj:objs) @@ -323,13 +344,14 @@ startServer origHead logd bhost services = do logd $ "unhandled service '" ++ show (toUUID svc) ++ "'" return (svcs, global) - | DatagramAddress paddr <- peerAddress peer -> do - logd $ "service packet from peer with incomplete identity " ++ show paddr + | otherwise -> do + logd $ "service packet from peer with incomplete identity " ++ show (peerAddress peer) return Server { serverStorage = storage , serverIdentity = midentity , serverSocket = ssocket + , serverChanPacket = chanPacket , serverPeers = peers , serverChanPeer' = chanPeer } @@ -362,7 +384,6 @@ handlePacket :: (String -> IO ()) -> Head LocalState -> UnifiedIdentity -> Bool -> TransportHeader -> IO (Maybe Peer) handlePacket logd origHead identity secure opeer chanSvc svcs (TransportHeader headers) = do let sidentity = idData identity - DatagramAddress paddr = peerAddress opeer plaintextRefs = map (refDigest . storedRef) $ concatMap (collectStoredObjects . wrappedLoad) $ concat [ [ storedRef sidentity ] , map storedRef $ idUpdates identity @@ -477,7 +498,7 @@ handlePacket logd origHead identity secure opeer chanSvc svcs (TransportHeader h case res of Left err -> do - logd $ "Error in handling packet from " ++ show paddr ++ ": " ++ err + logd $ "Error in handling packet from " ++ show (peerAddress opeer) ++ ": " ++ err return Nothing Right ph -> do when (not $ null $ phHead ph) $ do @@ -488,9 +509,9 @@ handlePacket logd origHead identity secure opeer chanSvc svcs (TransportHeader h case peerChannel $ phPeer ph of ChannelEstablished ch -> do x <- runExceptT (channelEncrypt ch plain) - case x of Right ctext -> void $ sendTo (peerSocket $ phPeer ph) ctext paddr + case x of Right ctext -> void $ sendTo (phPeer ph) ctext Left err -> logd $ "Failed to encrypt data: " ++ err - _ -> void $ sendTo (peerSocket $ phPeer ph) plain paddr + _ -> void $ sendTo (phPeer ph) plain return $ if phPeerChanged ph then Just $ phPeer ph else Nothing @@ -599,13 +620,12 @@ finalizedChannel oh self = do -- Outstanding service packets gets phPeer >>= \case - Peer { peerChannel = ChannelEstablished ch - , peerAddress = DatagramAddress paddr - , peerServiceOutQueue = oqueue - , peerSocket = sock - } -> do - ps <- liftIO $ modifyMVar oqueue $ return . ([],) - forM_ ps $ sendPacket sock paddr ch + peer@Peer + { peerChannel = ChannelEstablished ch + , peerServiceOutQueue = oqueue + } -> do + ps <- liftIO $ modifyMVar oqueue $ return . ([],) + forM_ ps $ sendPacket peer ch _ -> return () @@ -645,18 +665,17 @@ handleServices chan = gets (peerServiceInQueue . phPeer) >>= \case updatePeer $ \p -> p { peerServiceInQueue = queue' } -mkPeer :: Storage -> Socket -> SockAddr -> IO Peer -mkPeer st sock paddr = do +mkPeer :: Storage -> PeerAddress -> IO Peer +mkPeer st paddr = do pst <- deriveEphemeralStorage st ist <- derivePartialStorage pst svcs <- newMVar M.empty oqueue <- newMVar [] return $ Peer - { peerAddress = DatagramAddress paddr + { peerAddress = paddr , peerIdentity = PeerIdentityUnknown , peerIdentityUpdate = [] , peerChannel = ChannelWait - , peerSocket = sock , peerStorage = pst , peerInStorage = ist , peerServiceState = svcs @@ -668,21 +687,35 @@ mkPeer st sock paddr = do serverPeer :: Server -> SockAddr -> IO Peer serverPeer server paddr = do sock <- readMVar $ serverSocket server + serverPeer' server (DatagramAddress sock paddr) + +serverPeerIce :: Server -> IceSession -> IO Peer +serverPeerIce server ice = do + let paddr = PeerIceSession ice + peer <- serverPeer' server paddr + iceSetChan ice (paddr,) $ serverChanPacket server + return peer + +serverPeer' :: Server -> PeerAddress -> IO Peer +serverPeer' server paddr = do (peer, hello) <- modifyMVar (serverPeers server) $ \pvalue -> do case M.lookup paddr pvalue of Just peer -> return (pvalue, (peer, False)) Nothing -> do - peer <- mkPeer (serverStorage server) sock paddr + peer <- mkPeer (serverStorage server) paddr return (M.insert paddr peer pvalue, (peer, True)) when hello $ do identity <- readMVar (serverIdentity server) - void $ sendTo sock - (BL.toStrict $ serializeObject $ transportToObject $ TransportHeader + void $ sendTo peer $ + BL.toStrict $ serializeObject $ transportToObject $ TransportHeader [ AnnounceSelf $ partialRef (peerInStorage peer) $ storedRef $ idData identity ] - ) paddr return peer +sendTo :: Peer -> BC.ByteString -> IO () +sendTo Peer { peerAddress = DatagramAddress sock addr } msg = void $ S.sendTo sock msg addr +sendTo Peer { peerAddress = PeerIceSession ice } msg = iceSend ice msg + sendToPeer :: (Service s, MonadIO m, MonadError String m) => UnifiedIdentity -> Peer -> s -> m () sendToPeer self peer packet = sendToPeerList self peer [ServiceReply (Left packet) True] @@ -701,17 +734,16 @@ sendToPeerList _ peer parts = do packet = TransportPacket header content case peerChannel peer of ChannelEstablished ch -> do - let DatagramAddress paddr = peerAddress peer - sendPacket (peerSocket peer) paddr ch packet + sendPacket peer ch packet _ -> liftIO $ modifyMVar_ (peerServiceOutQueue peer) $ return . (packet:) -sendPacket :: (MonadIO m, MonadError String m) => Socket -> SockAddr -> Channel -> TransportPacket -> m () -sendPacket sock addr ch (TransportPacket header content) = do +sendPacket :: (MonadIO m, MonadError String m) => Peer -> Channel -> TransportPacket -> m () +sendPacket peer ch (TransportPacket header content) = do let plain = BL.toStrict $ BL.concat $ (serializeObject $ transportToObject header) : map lazyLoadBytes content ctext <- channelEncrypt ch plain - void $ liftIO $ sendTo sock ctext addr + void $ liftIO $ sendTo peer ctext sendToPeerWith :: forall s m. (Service s, MonadIO m, MonadError String m) => UnifiedIdentity -> Peer -> (ServiceState s -> ExceptT String IO (Maybe s, ServiceState s)) -> m () sendToPeerWith identity peer fobj = do diff --git a/src/Storage.hs b/src/Storage.hs index 92a1e1f..f73c420 100644 --- a/src/Storage.hs +++ b/src/Storage.hs @@ -35,7 +35,7 @@ module Storage ( loadBlob, loadRec, loadZero, loadInt, loadNum, loadText, loadBinary, loadDate, loadUUID, loadJson, loadRef, loadRawRef, loadMbInt, loadMbNum, loadMbText, loadMbBinary, loadMbDate, loadMbUUID, loadMbJson, loadMbRef, loadMbRawRef, - loadBinaries, loadRefs, loadRawRefs, + loadTexts, loadBinaries, loadRefs, loadRawRefs, loadZRef, Stored, @@ -720,6 +720,12 @@ loadMbText name = asks (lookup (BC.pack name) . snd) >>= \case Just (RecText x) -> Just <$> fromText x Just _ -> throwError $ "Expecting type text of record item '"++name++"'" +loadTexts :: StorableText a => String -> LoadRec [a] +loadTexts name = do + items <- map snd . filter ((BC.pack name ==) . fst) <$> asks snd + forM items $ \case RecText x -> fromText x + _ -> throwError $ "Expecting type text of record item '"++name++"'" + loadBinary :: BA.ByteArray a => String -> LoadRec a loadBinary name = maybe (throwError $ "Missing record item '"++name++"'") return =<< loadMbBinary name |