diff options
Diffstat (limited to 'src/Erebos/Network.hs')
| -rw-r--r-- | src/Erebos/Network.hs | 815 |
1 files changed, 815 insertions, 0 deletions
diff --git a/src/Erebos/Network.hs b/src/Erebos/Network.hs new file mode 100644 index 0000000..b26ada5 --- /dev/null +++ b/src/Erebos/Network.hs @@ -0,0 +1,815 @@ +{-# LANGUAGE CPP #-} + +module Erebos.Network ( + Server, + startServer, + stopServer, + getNextPeerChange, + ServerOptions(..), serverIdentity, defaultServerOptions, + + Peer, peerServer, peerStorage, + PeerAddress(..), peerAddress, + PeerIdentity(..), peerIdentity, + WaitingRef, wrDigest, + Service(..), + serverPeer, +#ifdef ENABLE_ICE_SUPPORT + serverPeerIce, +#endif + sendToPeer, sendToPeerStored, sendToPeerWith, + runPeerService, + + discoveryPort, +) where + +import Control.Concurrent +import Control.Concurrent.STM +import Control.Exception +import Control.Monad +import Control.Monad.Except +import Control.Monad.Reader +import Control.Monad.State + +import qualified Data.ByteString.Char8 as BC +import Data.Function +import Data.IP qualified as IP +import Data.List +import Data.Map (Map) +import qualified Data.Map as M +import Data.Maybe +import Data.Typeable +import Data.Word + +import Foreign.Ptr +import Foreign.Storable + +import GHC.Conc.Sync (unsafeIOToSTM) + +import Network.Socket hiding (ControlMessage) +import qualified Network.Socket.ByteString as S + +import Erebos.Channel +#ifdef ENABLE_ICE_SUPPORT +import Erebos.ICE +#endif +import Erebos.Identity +import Erebos.Network.Protocol +import Erebos.PubKey +import Erebos.Service +import Erebos.State +import Erebos.Storage +import Erebos.Storage.Key +import Erebos.Storage.Merge + + +discoveryPort :: PortNumber +discoveryPort = 29665 + +announceIntervalSeconds :: Int +announceIntervalSeconds = 60 + + +data Server = Server + { serverStorage :: Storage + , serverOrigHead :: Head LocalState + , serverIdentity_ :: MVar UnifiedIdentity + , serverThreads :: MVar [ThreadId] + , serverSocket :: MVar Socket + , serverRawPath :: SymFlow (PeerAddress, BC.ByteString) + , serverControlFlow :: Flow (ControlMessage PeerAddress) (ControlRequest PeerAddress) + , serverDataResponse :: TQueue (Peer, Maybe PartialRef) + , serverIOActions :: TQueue (ExceptT String IO ()) + , serverServices :: [SomeService] + , serverServiceStates :: TMVar (M.Map ServiceID SomeServiceGlobalState) + , serverPeers :: MVar (Map PeerAddress Peer) + , serverChanPeer :: TChan Peer + , serverErrorLog :: TQueue String + } + +serverIdentity :: Server -> IO UnifiedIdentity +serverIdentity = readMVar . serverIdentity_ + +getNextPeerChange :: Server -> IO Peer +getNextPeerChange = atomically . readTChan . serverChanPeer + +data ServerOptions = ServerOptions + { serverPort :: PortNumber + , serverLocalDiscovery :: Bool + } + +defaultServerOptions :: ServerOptions +defaultServerOptions = ServerOptions + { serverPort = discoveryPort + , serverLocalDiscovery = True + } + + +data Peer = Peer + { peerAddress :: PeerAddress + , peerServer_ :: Server + , peerConnection :: TVar (Either [(Bool, TransportPacket Ref, [TransportHeaderItem])] (Connection PeerAddress)) + , peerIdentityVar :: TVar PeerIdentity + , peerStorage_ :: Storage + , peerInStorage :: PartialStorage + , peerServiceState :: TMVar (M.Map ServiceID SomeServiceState) + , peerWaitingRefs :: TMVar [WaitingRef] + } + +peerServer :: Peer -> Server +peerServer = peerServer_ + +peerStorage :: Peer -> Storage +peerStorage = peerStorage_ + +getPeerChannel :: Peer -> STM ChannelState +getPeerChannel Peer {..} = either (const $ return ChannelNone) connGetChannel =<< readTVar peerConnection + +setPeerChannel :: Peer -> ChannelState -> STM () +setPeerChannel Peer {..} ch = do + readTVar peerConnection >>= \case + Left _ -> retry + Right conn -> connSetChannel conn ch + +instance Eq Peer where + (==) = (==) `on` peerIdentityVar + +data PeerAddress = DatagramAddress SockAddr +#ifdef ENABLE_ICE_SUPPORT + | PeerIceSession IceSession +#endif + +instance Show PeerAddress where + show (DatagramAddress saddr) = unwords $ case IP.fromSockAddr saddr of + Just (IP.IPv6 ipv6, port) + | (0, 0, 0xffff, ipv4) <- IP.fromIPv6w ipv6 + -> [show (IP.toIPv4w ipv4), show port] + Just (addr, port) + -> [show addr, show port] + _ -> [show saddr] +#ifdef ENABLE_ICE_SUPPORT + show (PeerIceSession ice) = show ice +#endif + +instance Eq PeerAddress where + DatagramAddress addr == DatagramAddress addr' = addr == addr' +#ifdef ENABLE_ICE_SUPPORT + PeerIceSession ice == PeerIceSession ice' = ice == ice' + _ == _ = False +#endif + +instance Ord PeerAddress where + compare (DatagramAddress addr) (DatagramAddress addr') = compare addr addr' +#ifdef ENABLE_ICE_SUPPORT + compare (DatagramAddress _ ) _ = LT + compare _ (DatagramAddress _ ) = GT + compare (PeerIceSession ice ) (PeerIceSession ice') = compare ice ice' +#endif + + +data PeerIdentity = PeerIdentityUnknown (TVar [UnifiedIdentity -> ExceptT String IO ()]) + | PeerIdentityRef WaitingRef (TVar [UnifiedIdentity -> ExceptT String IO ()]) + | PeerIdentityFull UnifiedIdentity + +peerIdentity :: MonadIO m => Peer -> m PeerIdentity +peerIdentity = liftIO . atomically . readTVar . peerIdentityVar + + +lookupServiceType :: [TransportHeaderItem] -> Maybe ServiceID +lookupServiceType (ServiceType stype : _) = Just stype +lookupServiceType (_ : hs) = lookupServiceType hs +lookupServiceType [] = Nothing + + +newWaitingRef :: RefDigest -> (Ref -> WaitingRefCallback) -> PacketHandler WaitingRef +newWaitingRef dgst act = do + peer@Peer {..} <- gets phPeer + wref <- WaitingRef peerStorage_ (partialRefFromDigest peerInStorage dgst) act <$> liftSTM (newTVar (Left [])) + modifyTMVarP peerWaitingRefs (wref:) + liftSTM $ writeTQueue (serverDataResponse $ peerServer peer) (peer, Nothing) + return wref + + +forkServerThread :: Server -> IO () -> IO () +forkServerThread server act = modifyMVar_ (serverThreads server) $ \ts -> do + (:ts) <$> forkIO act + +startServer :: ServerOptions -> Head LocalState -> (String -> IO ()) -> [SomeService] -> IO Server +startServer opt serverOrigHead logd' serverServices = do + let serverStorage = headStorage serverOrigHead + serverIdentity_ <- newMVar $ headLocalIdentity serverOrigHead + serverThreads <- newMVar [] + serverSocket <- newEmptyMVar + (serverRawPath, protocolRawPath) <- newFlowIO + (serverControlFlow, protocolControlFlow) <- newFlowIO + serverDataResponse <- newTQueueIO + serverIOActions <- newTQueueIO + serverServiceStates <- newTMVarIO M.empty + serverPeers <- newMVar M.empty + serverChanPeer <- newTChanIO + serverErrorLog <- newTQueueIO + let server = Server {..} + + chanSvc <- newTQueueIO + + let logd = writeTQueue serverErrorLog + forkServerThread server $ forever $ do + logd' =<< atomically (readTQueue serverErrorLog) + + forkServerThread server $ dataResponseWorker server + forkServerThread server $ forever $ do + either (atomically . logd) return =<< runExceptT =<< + atomically (readTQueue serverIOActions) + + broadcastAddreses <- getBroadcastAddresses discoveryPort + + let open addr = do + sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr) + putMVar serverSocket sock + setSocketOption sock ReuseAddr 1 + setSocketOption sock Broadcast 1 + withFdSocket sock setCloseOnExecIfNeeded + bind sock (addrAddress addr) + return sock + + loop sock = do + when (serverLocalDiscovery opt) $ forkServerThread server $ forever $ do + atomically $ writeFlowBulk serverControlFlow $ map (SendAnnounce . DatagramAddress) broadcastAddreses + threadDelay $ announceIntervalSeconds * 1000 * 1000 + + let announceUpdate identity = do + st <- derivePartialStorage serverStorage + let selfRef = partialRef st $ storedRef $ idExtData identity + updateRefs = map refDigest $ selfRef : map (partialRef st . storedRef) (idUpdates identity) + ackedBy = concat [[ Acknowledged r, Rejected r, DataRequest r ] | r <- updateRefs ] + hitems = map AnnounceUpdate updateRefs + packet = TransportPacket (TransportHeader $ hitems) [] + + ps <- readMVar serverPeers + forM_ ps $ \peer -> atomically $ do + ((,) <$> readTVar (peerIdentityVar peer) <*> getPeerChannel peer) >>= \case + (PeerIdentityFull _, ChannelEstablished _) -> + sendToPeerS peer ackedBy packet + _ -> return () + + void $ watchHead serverOrigHead $ \h -> do + let idt = headLocalIdentity h + changedId <- modifyMVar serverIdentity_ $ \cur -> + return (idt, cur /= idt) + when changedId $ do + writeFlowIO serverControlFlow $ UpdateSelfIdentity idt + announceUpdate idt + + forM_ serverServices $ \(SomeService service _) -> do + forM_ (serviceStorageWatchers service) $ \(SomeStorageWatcher sel act) -> do + watchHeadWith serverOrigHead (sel . headStoredObject) $ \x -> do + withMVar serverPeers $ mapM_ $ \peer -> atomically $ do + readTVar (peerIdentityVar peer) >>= \case + PeerIdentityFull _ -> writeTQueue serverIOActions $ do + runPeerService peer $ act x + _ -> return () + + forkServerThread server $ forever $ do + (msg, saddr) <- S.recvFrom sock 4096 + writeFlowIO serverRawPath (DatagramAddress saddr, msg) + + forkServerThread server $ forever $ do + (paddr, msg) <- readFlowIO serverRawPath + case paddr of + DatagramAddress addr -> void $ S.sendTo sock msg addr +#ifdef ENABLE_ICE_SUPPORT + PeerIceSession ice -> iceSend ice msg +#endif + + forkServerThread server $ forever $ do + readFlowIO serverControlFlow >>= \case + NewConnection conn mbpid -> do + let paddr = connAddress conn + peer <- modifyMVar serverPeers $ \pvalue -> do + case M.lookup paddr pvalue of + Just peer -> return (pvalue, peer) + Nothing -> do + peer <- mkPeer server paddr + return (M.insert paddr peer pvalue, peer) + + forkServerThread server $ do + atomically $ do + readTVar (peerConnection peer) >>= \case + Left packets -> writeFlowBulk (connData conn) $ reverse packets + Right _ -> return () + writeTVar (peerConnection peer) (Right conn) + + case mbpid of + Just dgst -> do + identity <- readMVar serverIdentity_ + atomically $ runPacketHandler False peer $ do + wref <- newWaitingRef dgst $ handleIdentityAnnounce identity peer + readTVarP (peerIdentityVar peer) >>= \case + PeerIdentityUnknown idwait -> do + addHeader $ AnnounceSelf $ refDigest $ storedRef $ idData identity + writeTVarP (peerIdentityVar peer) $ PeerIdentityRef wref idwait + liftSTM $ writeTChan serverChanPeer peer + _ -> return () + Nothing -> return () + + forever $ do + (secure, TransportPacket header objs) <- readFlowIO $ connData conn + prefs <- forM objs $ storeObject $ peerInStorage peer + identity <- readMVar serverIdentity_ + let svcs = map someServiceID serverServices + handlePacket identity secure peer chanSvc svcs header prefs + + ReceivedAnnounce addr _ -> do + void $ serverPeer' server addr + + erebosNetworkProtocol (headLocalIdentity serverOrigHead) logd protocolRawPath protocolControlFlow + + forkServerThread server $ withSocketsDo $ do + let hints = defaultHints + { addrFlags = [AI_PASSIVE] + , addrFamily = AF_INET6 + , addrSocketType = Datagram + } + addr:_ <- getAddrInfo (Just hints) Nothing (Just $ show $ serverPort opt) + bracket (open addr) close loop + + forkServerThread server $ forever $ do + (peer, svc, ref) <- atomically $ readTQueue chanSvc + case find ((svc ==) . someServiceID) serverServices of + Just service@(SomeService (_ :: Proxy s) attr) -> runPeerServiceOn (Just (service, attr)) peer (serviceHandler $ wrappedLoad @s ref) + _ -> atomically $ logd $ "unhandled service '" ++ show (toUUID svc) ++ "'" + + return server + +stopServer :: Server -> IO () +stopServer Server {..} = do + mapM_ killThread =<< takeMVar serverThreads + +dataResponseWorker :: Server -> IO () +dataResponseWorker server = forever $ do + (peer, npref) <- atomically (readTQueue $ serverDataResponse server) + + wait <- atomically $ takeTMVar (peerWaitingRefs peer) + list <- forM wait $ \wr@WaitingRef { wrefStatus = tvar } -> + atomically (readTVar tvar) >>= \case + Left ds -> case maybe id (filter . (/=) . refDigest) npref $ ds of + [] -> copyRef (wrefStorage wr) (wrefPartial wr) >>= \case + Right ref -> do + atomically (writeTVar tvar $ Right ref) + forkServerThread server $ runExceptT (wrefAction wr ref) >>= \case + Left err -> atomically $ writeTQueue (serverErrorLog server) err + Right () -> return () + + return (Nothing, []) + Left dgst -> do + atomically (writeTVar tvar $ Left [dgst]) + return (Just wr, [dgst]) + ds' -> do + atomically (writeTVar tvar $ Left ds') + return (Just wr, []) + Right _ -> return (Nothing, []) + atomically $ putTMVar (peerWaitingRefs peer) $ catMaybes $ map fst list + + let reqs = concat $ map snd list + when (not $ null reqs) $ do + let packet = TransportPacket (TransportHeader $ map DataRequest reqs) [] + ackedBy = concat [[ Rejected r, DataResponse r ] | r <- reqs ] + atomically $ sendToPeerPlain peer ackedBy packet + + +newtype PacketHandler a = PacketHandler { unPacketHandler :: StateT PacketHandlerState (ExceptT String STM) a } + deriving (Functor, Applicative, Monad, MonadState PacketHandlerState, MonadError String) + +instance MonadFail PacketHandler where + fail = throwError + +runPacketHandler :: Bool -> Peer -> PacketHandler () -> STM () +runPacketHandler secure peer@Peer {..} act = do + let logd = writeTQueue $ serverErrorLog peerServer_ + runExceptT (flip execStateT (PacketHandlerState peer [] [] []) $ unPacketHandler act) >>= \case + Left err -> do + logd $ "Error in handling packet from " ++ show peerAddress ++ ": " ++ err + Right ph -> do + when (not $ null $ phHead ph) $ do + let packet = TransportPacket (TransportHeader $ phHead ph) (phBody ph) + sendToPeerS' secure peer (phAckedBy ph) packet + +liftSTM :: STM a -> PacketHandler a +liftSTM = PacketHandler . lift . lift + +readTVarP :: TVar a -> PacketHandler a +readTVarP = liftSTM . readTVar + +writeTVarP :: TVar a -> a -> PacketHandler () +writeTVarP v = liftSTM . writeTVar v + +modifyTMVarP :: TMVar a -> (a -> a) -> PacketHandler () +modifyTMVarP v f = liftSTM $ putTMVar v . f =<< takeTMVar v + +data PacketHandlerState = PacketHandlerState + { phPeer :: Peer + , phHead :: [TransportHeaderItem] + , phAckedBy :: [TransportHeaderItem] + , phBody :: [Ref] + } + +addHeader :: TransportHeaderItem -> PacketHandler () +addHeader h = modify $ \ph -> ph { phHead = h `appendDistinct` phHead ph } + +addAckedBy :: [TransportHeaderItem] -> PacketHandler () +addAckedBy hs = modify $ \ph -> ph { phAckedBy = foldr appendDistinct (phAckedBy ph) hs } + +addBody :: Ref -> PacketHandler () +addBody r = modify $ \ph -> ph { phBody = r `appendDistinct` phBody ph } + +appendDistinct :: Eq a => a -> [a] -> [a] +appendDistinct x (y:ys) | x == y = y : ys + | otherwise = y : appendDistinct x ys +appendDistinct x [] = [x] + +handlePacket :: UnifiedIdentity -> Bool + -> Peer -> TQueue (Peer, ServiceID, Ref) -> [ServiceID] + -> TransportHeader -> [PartialRef] -> IO () +handlePacket identity secure peer chanSvc svcs (TransportHeader headers) prefs = atomically $ do + let server = peerServer peer + ochannel <- getPeerChannel peer + let sidentity = idData identity + plaintextRefs = map (refDigest . storedRef) $ concatMap (collectStoredObjects . wrappedLoad) $ concat + [ [ storedRef sidentity ] + , map storedRef $ idUpdates identity + , case ochannel of + ChannelOurRequest _ req -> [ storedRef req ] + ChannelOurAccept _ acc _ -> [ storedRef acc ] + _ -> [] + ] + + runPacketHandler secure peer $ do + let logd = liftSTM . writeTQueue (serverErrorLog server) + forM_ headers $ \case + Acknowledged dgst -> do + liftSTM (getPeerChannel peer) >>= \case + ChannelOurAccept _ acc ch | refDigest (storedRef acc) == dgst -> do + liftSTM $ finalizedChannel peer ch identity + _ -> return () + + Rejected dgst -> do + logd $ "rejected by peer: " ++ show dgst + + DataRequest dgst + | secure || dgst `elem` plaintextRefs -> do + Right mref <- liftSTM $ unsafeIOToSTM $ + copyRef (peerStorage peer) $ + partialRefFromDigest (peerInStorage peer) dgst + addHeader $ DataResponse dgst + addAckedBy [ Acknowledged dgst, Rejected dgst ] + addBody $ mref + | otherwise -> do + logd $ "unauthorized data request for " ++ show dgst + addHeader $ Rejected dgst + + DataResponse dgst -> if + | Just pref <- find ((==dgst) . refDigest) prefs -> do + when (not secure) $ do + addHeader $ Acknowledged dgst + liftSTM $ writeTQueue (serverDataResponse server) (peer, Just pref) + | otherwise -> throwError $ "mismatched data response " ++ show dgst + + AnnounceSelf dgst + | dgst == refDigest (storedRef sidentity) -> return () + | otherwise -> do + wref <- newWaitingRef dgst $ handleIdentityAnnounce identity peer + readTVarP (peerIdentityVar peer) >>= \case + PeerIdentityUnknown idwait -> do + addHeader $ AnnounceSelf $ refDigest $ storedRef $ idData identity + writeTVarP (peerIdentityVar peer) $ PeerIdentityRef wref idwait + liftSTM $ writeTChan (serverChanPeer $ peerServer peer) peer + _ -> return () + + AnnounceUpdate dgst -> do + readTVarP (peerIdentityVar peer) >>= \case + PeerIdentityFull _ -> do + void $ newWaitingRef dgst $ handleIdentityUpdate peer + _ -> return () + + TrChannelRequest dgst -> do + let process cookie = do + addHeader $ Acknowledged dgst + wref <- newWaitingRef dgst $ handleChannelRequest peer identity + liftSTM $ setPeerChannel peer $ ChannelPeerRequest cookie wref + reject = addHeader $ Rejected dgst + + liftSTM (getPeerChannel peer) >>= \case + ChannelNone {} -> return () + ChannelCookieWait {} -> return () + ChannelCookieReceived cookie -> process $ Just cookie + ChannelCookieConfirmed cookie -> process $ Just cookie + ChannelOurRequest mbcookie our | dgst < refDigest (storedRef our) -> process mbcookie + | otherwise -> reject + ChannelPeerRequest mbcookie _ -> process mbcookie + ChannelOurAccept {} -> reject + ChannelEstablished {} -> process Nothing + + TrChannelAccept dgst -> do + let process = do + handleChannelAccept identity $ partialRefFromDigest (peerInStorage peer) dgst + reject = addHeader $ Rejected dgst + liftSTM (getPeerChannel peer) >>= \case + ChannelNone {} -> reject + ChannelCookieWait {} -> reject + ChannelCookieReceived {} -> reject + ChannelCookieConfirmed {} -> reject + ChannelOurRequest {} -> process + ChannelPeerRequest {} -> process + ChannelOurAccept _ our _ | dgst < refDigest (storedRef our) -> process + | otherwise -> addHeader $ Rejected dgst + ChannelEstablished {} -> process + + ServiceType _ -> return () + ServiceRef dgst + | not secure -> throwError $ "service packet without secure channel" + | Just svc <- lookupServiceType headers -> if + | svc `elem` svcs -> do + if dgst `elem` map refDigest prefs || True {- TODO: used by Message service to confirm receive -} + then do + void $ newWaitingRef dgst $ \ref -> + liftIO $ atomically $ writeTQueue chanSvc (peer, svc, ref) + else throwError $ "missing service object " ++ show dgst + | otherwise -> addHeader $ Rejected dgst + | otherwise -> throwError $ "service ref without type" + + _ -> return () + + +withPeerIdentity :: MonadIO m => Peer -> (UnifiedIdentity -> ExceptT String IO ()) -> m () +withPeerIdentity peer act = liftIO $ atomically $ readTVar (peerIdentityVar peer) >>= \case + PeerIdentityUnknown tvar -> modifyTVar' tvar (act:) + PeerIdentityRef _ tvar -> modifyTVar' tvar (act:) + PeerIdentityFull idt -> writeTQueue (serverIOActions $ peerServer peer) (act idt) + + +setupChannel :: UnifiedIdentity -> Peer -> UnifiedIdentity -> WaitingRefCallback +setupChannel identity peer upid = do + req <- flip runReaderT (peerStorage peer) $ createChannelRequest identity upid + let reqref = refDigest $ storedRef req + let hitems = + [ TrChannelRequest reqref + , AnnounceSelf $ refDigest $ storedRef $ idData identity + ] + liftIO $ atomically $ do + getPeerChannel peer >>= \case + ChannelCookieConfirmed cookie -> do + sendToPeerPlain peer [ Acknowledged reqref, Rejected reqref ] $ + TransportPacket (TransportHeader hitems) [storedRef req] + setPeerChannel peer $ ChannelOurRequest (Just cookie) req + _ -> return () + +handleChannelRequest :: Peer -> UnifiedIdentity -> Ref -> WaitingRefCallback +handleChannelRequest peer identity req = do + withPeerIdentity peer $ \upid -> do + (acc, ch) <- flip runReaderT (peerStorage peer) $ acceptChannelRequest identity upid (wrappedLoad req) + liftIO $ atomically $ do + getPeerChannel peer >>= \case + ChannelPeerRequest mbcookie wr | wrDigest wr == refDigest req -> do + setPeerChannel peer $ ChannelOurAccept mbcookie acc ch + let accref = refDigest $ storedRef acc + header = TrChannelAccept accref + ackedBy = [ Acknowledged accref, Rejected accref ] + sendToPeerPlain peer ackedBy $ TransportPacket (TransportHeader [header]) $ concat + [ [ storedRef $ acc ] + , [ storedRef $ signedData $ fromStored acc ] + , [ storedRef $ caKey $ fromStored $ signedData $ fromStored acc ] + , map storedRef $ signedSignature $ fromStored acc + ] + _ -> writeTQueue (serverErrorLog $ peerServer peer) $ "unexpected channel request" + +handleChannelAccept :: UnifiedIdentity -> PartialRef -> PacketHandler () +handleChannelAccept identity accref = do + peer <- gets phPeer + liftSTM $ writeTQueue (serverIOActions $ peerServer peer) $ do + withPeerIdentity peer $ \upid -> do + copyRef (peerStorage peer) accref >>= \case + Right acc -> do + ch <- acceptedChannel identity upid (wrappedLoad acc) + liftIO $ atomically $ do + sendToPeerS peer [] $ TransportPacket (TransportHeader [Acknowledged $ refDigest accref]) [] + finalizedChannel peer ch identity + + Left dgst -> throwError $ "missing accept data " ++ BC.unpack (showRefDigest dgst) + + +finalizedChannel :: Peer -> Channel -> UnifiedIdentity -> STM () +finalizedChannel peer@Peer {..} ch self = do + setPeerChannel peer $ ChannelEstablished ch + + -- Identity update + writeTQueue (serverIOActions peerServer_) $ liftIO $ atomically $ do + let selfRef = refDigest $ storedRef $ idExtData $ self + updateRefs = selfRef : map (refDigest . storedRef) (idUpdates self) + ackedBy = concat [[ Acknowledged r, Rejected r, DataRequest r ] | r <- updateRefs ] + sendToPeerS peer ackedBy $ flip TransportPacket [] $ TransportHeader $ map AnnounceUpdate updateRefs + + -- Notify services about new peer + readTVar peerIdentityVar >>= \case + PeerIdentityFull _ -> notifyServicesOfPeer peer + _ -> return () + + +handleIdentityAnnounce :: UnifiedIdentity -> Peer -> Ref -> WaitingRefCallback +handleIdentityAnnounce self peer ref = liftIO $ atomically $ do + let validateAndUpdate upds act = case validateIdentity $ wrappedLoad ref of + Just pid' -> do + let pid = fromMaybe pid' $ toUnifiedIdentity (updateIdentity upds pid') + writeTVar (peerIdentityVar peer) $ PeerIdentityFull pid + writeTChan (serverChanPeer $ peerServer peer) peer + act pid + writeTQueue (serverIOActions $ peerServer peer) $ do + setupChannel self peer pid + Nothing -> return () + + readTVar (peerIdentityVar peer) >>= \case + PeerIdentityRef wref wact + | wrDigest wref == refDigest ref + -> validateAndUpdate [] $ \pid -> do + mapM_ (writeTQueue (serverIOActions $ peerServer peer) . ($ pid)) . + reverse =<< readTVar wact + + PeerIdentityFull pid + | idData pid `precedes` wrappedLoad ref + -> validateAndUpdate (idUpdates pid) $ \_ -> do + notifyServicesOfPeer peer + + _ -> return () + +handleIdentityUpdate :: Peer -> Ref -> WaitingRefCallback +handleIdentityUpdate peer ref = liftIO $ atomically $ do + pidentity <- readTVar (peerIdentityVar peer) + if | PeerIdentityFull pid <- pidentity + , Just pid' <- toUnifiedIdentity $ updateIdentity [wrappedLoad ref] pid + -> do + writeTVar (peerIdentityVar peer) $ PeerIdentityFull pid' + writeTChan (serverChanPeer $ peerServer peer) peer + when (idData pid /= idData pid') $ notifyServicesOfPeer peer + + | otherwise -> return () + +notifyServicesOfPeer :: Peer -> STM () +notifyServicesOfPeer peer@Peer { peerServer_ = Server {..} } = do + writeTQueue serverIOActions $ do + forM_ serverServices $ \service@(SomeService _ attrs) -> + runPeerServiceOn (Just (service, attrs)) peer serviceNewPeer + + +mkPeer :: Server -> PeerAddress -> IO Peer +mkPeer peerServer_ peerAddress = do + peerConnection <- newTVarIO (Left []) + peerIdentityVar <- newTVarIO . PeerIdentityUnknown =<< newTVarIO [] + peerStorage_ <- deriveEphemeralStorage $ serverStorage peerServer_ + peerInStorage <- derivePartialStorage peerStorage_ + peerServiceState <- newTMVarIO M.empty + peerWaitingRefs <- newTMVarIO [] + return Peer {..} + +serverPeer :: Server -> SockAddr -> IO Peer +serverPeer server paddr = do + serverPeer' server (DatagramAddress paddr) + +#ifdef ENABLE_ICE_SUPPORT +serverPeerIce :: Server -> IceSession -> IO Peer +serverPeerIce server@Server {..} ice = do + let paddr = PeerIceSession ice + peer <- serverPeer' server paddr + iceSetChan ice $ mapFlow undefined (paddr,) serverRawPath + return peer +#endif + +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 server paddr + return (M.insert paddr peer pvalue, (peer, True)) + when hello $ atomically $ do + writeFlow (serverControlFlow server) (RequestConnection paddr) + return peer + + +sendToPeer :: (Service s, MonadIO m) => Peer -> s -> m () +sendToPeer peer packet = sendToPeerList peer [ServiceReply (Left packet) True] + +sendToPeerStored :: (Service s, MonadIO m) => Peer -> Stored s -> m () +sendToPeerStored peer spacket = sendToPeerList peer [ServiceReply (Right spacket) True] + +sendToPeerList :: (Service s, MonadIO m) => Peer -> [ServiceReply s] -> m () +sendToPeerList peer parts = do + let st = peerStorage peer + srefs <- liftIO $ fmap catMaybes $ forM parts $ \case + ServiceReply (Left x) use -> Just . (,use) <$> store st x + ServiceReply (Right sx) use -> return $ Just (storedRef sx, use) + ServiceFinally act -> act >> return Nothing + let dgsts = map (refDigest . fst) srefs + let content = map fst $ filter snd srefs + header = TransportHeader (ServiceType (serviceID $ head parts) : map ServiceRef dgsts) + packet = TransportPacket header content + ackedBy = concat [[ Acknowledged r, Rejected r, DataRequest r ] | r <- dgsts ] + liftIO $ atomically $ sendToPeerS peer ackedBy packet + +sendToPeerS' :: Bool -> Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM () +sendToPeerS' secure Peer {..} ackedBy packet = do + readTVar peerConnection >>= \case + Left xs -> writeTVar peerConnection $ Left $ (secure, packet, ackedBy) : xs + Right conn -> writeFlow (connData conn) (secure, packet, ackedBy) + +sendToPeerS :: Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM () +sendToPeerS = sendToPeerS' True + +sendToPeerPlain :: Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM () +sendToPeerPlain = sendToPeerS' False + +sendToPeerWith :: forall s m. (Service s, MonadIO m, MonadError String m) => Peer -> (ServiceState s -> ExceptT String IO (Maybe s, ServiceState s)) -> m () +sendToPeerWith peer fobj = do + let sproxy = Proxy @s + sid = serviceID sproxy + res <- liftIO $ do + svcs <- atomically $ takeTMVar (peerServiceState peer) + (svcs', res) <- runExceptT (fobj $ fromMaybe (emptyServiceState sproxy) $ fromServiceState sproxy =<< M.lookup sid svcs) >>= \case + Right (obj, s') -> return $ (M.insert sid (SomeServiceState sproxy s') svcs, Right obj) + Left err -> return $ (svcs, Left err) + atomically $ putTMVar (peerServiceState peer) svcs' + return res + + case res of + Right (Just obj) -> sendToPeer peer obj + Right Nothing -> return () + Left err -> throwError err + + +lookupService :: forall s. Service s => Proxy s -> [SomeService] -> Maybe (SomeService, ServiceAttributes s) +lookupService proxy (service@(SomeService (_ :: Proxy t) attr) : rest) + | Just (Refl :: s :~: t) <- eqT = Just (service, attr) + | otherwise = lookupService proxy rest +lookupService _ [] = Nothing + +runPeerService :: forall s m. (Service s, MonadIO m) => Peer -> ServiceHandler s () -> m () +runPeerService = runPeerServiceOn Nothing + +runPeerServiceOn :: forall s m. (Service s, MonadIO m) => Maybe (SomeService, ServiceAttributes s) -> Peer -> ServiceHandler s () -> m () +runPeerServiceOn mbservice peer handler = liftIO $ do + let server = peerServer peer + proxy = Proxy @s + svc = serviceID proxy + logd = writeTQueue (serverErrorLog server) + case mbservice `mplus` lookupService proxy (serverServices server) of + Just (service, attr) -> + atomically (readTVar (peerIdentityVar peer)) >>= \case + PeerIdentityFull peerId -> do + (global, svcs) <- atomically $ (,) + <$> takeTMVar (serverServiceStates server) + <*> takeTMVar (peerServiceState peer) + case (fromMaybe (someServiceEmptyState service) $ M.lookup svc svcs, + fromMaybe (someServiceEmptyGlobalState service) $ M.lookup svc global) of + ((SomeServiceState (_ :: Proxy ps) ps), + (SomeServiceGlobalState (_ :: Proxy gs) gs)) -> do + Just (Refl :: s :~: ps) <- return $ eqT + Just (Refl :: s :~: gs) <- return $ eqT + + let inp = ServiceInput + { svcAttributes = attr + , svcPeer = peer + , svcPeerIdentity = peerId + , svcServer = server + , svcPrintOp = atomically . logd + } + reloadHead (serverOrigHead server) >>= \case + Nothing -> atomically $ do + logd $ "current head deleted" + putTMVar (peerServiceState peer) svcs + putTMVar (serverServiceStates server) global + Just h -> do + (rsp, (s', gs')) <- runServiceHandler h inp ps gs handler + moveKeys (peerStorage peer) (serverStorage server) + when (not (null rsp)) $ do + sendToPeerList peer rsp + atomically $ do + putTMVar (peerServiceState peer) $ M.insert svc (SomeServiceState proxy s') svcs + putTMVar (serverServiceStates server) $ M.insert svc (SomeServiceGlobalState proxy gs') global + _ -> do + atomically $ logd $ "can't run service handler on peer with incomplete identity " ++ show (peerAddress peer) + + _ -> atomically $ do + logd $ "unhandled service '" ++ show (toUUID svc) ++ "'" + + +foreign import ccall unsafe "Network/ifaddrs.h broadcast_addresses" cBroadcastAddresses :: IO (Ptr Word32) +foreign import ccall unsafe "stdlib.h free" cFree :: Ptr Word32 -> IO () + +getBroadcastAddresses :: PortNumber -> IO [SockAddr] +getBroadcastAddresses port = do + ptr <- cBroadcastAddresses + let parse i = do + w <- peekElemOff ptr i + if w == 0 then return [] + else (SockAddrInet port w:) <$> parse (i + 1) + addrs <- parse 0 + cFree ptr + return addrs |