2 files changed, 110 insertions, 182 deletions

diff --git a/src/Network/Tox.hs b/src/Network/Tox.hs
index 861d71d3..88228c50 100644
--- a/src/Network/Tox.hs
+++ b/src/Network/Tox.hs
@@ -32,31 +32,25 @@ import qualified Data.ByteString              as B
 import qualified Data.ByteString.Char8        as C8
 import Data.Data
 import Data.Functor.Contravariant
-import Data.IP
 import Data.Maybe
 import qualified Data.MinMaxPSQ               as MinMaxPSQ
 import qualified Data.Serialize               as S
 import Data.Time.Clock.POSIX                  (getPOSIXTime)
 import Data.Word
+import Network.Socket
+import System.Endian
+
+import Network.BitTorrent.DHT.Token           as Token
 import qualified Data.Wrapper.PSQ             as PSQ
 import System.Global6
-import Network.Address                        (WantIP (..))
+import Network.Address                        (WantIP (..),IP)
 import qualified Network.Kademlia.Routing          as R
 import Network.QueryResponse
-import Network.Socket
-import System.Endian
-import Network.BitTorrent.DHT.Token           as Token
-
-import Connection
 import Crypto.Tox
 import Data.Word64Map                        (fitsInInt)
 import qualified Data.Word64Map              (empty)
-import HandshakeCache
 import Network.Kademlia.Bootstrap            (forkPollForRefresh, bootstrap)
-import Network.Kademlia.Search
 import Network.Tox.Crypto.Transport          (Handshake(..),CryptoPacket)
-import Network.Tox.Handshake
-import Network.Tox.Crypto.Handlers
 import qualified Network.Tox.DHT.Handlers    as DHT
 import qualified Network.Tox.DHT.Transport   as DHT
 import Network.Tox.NodeId
@@ -66,12 +60,12 @@ import Network.Tox.Transport
 import OnionRouter
 import Network.Tox.ContactInfo
 import Text.XXD
-import qualified Data.HashMap.Strict as HashMap
-import qualified Data.Map.Strict as Map
 import DPut
 import Network.Tox.Avahi
-import Text.Printf
-import Data.List
+import Network.Tox.Session
+import Network.SessionTransports
+import Network.Kademlia.Search
+import HandshakeCache
 
 newCrypto :: IO TransportCrypto
 newCrypto = do
@@ -207,7 +201,6 @@ data Tox extra  = Tox
     , toxCrypto         :: Transport String SockAddr (CryptoPacket Encrypted)
     , toxHandshakes     :: Transport String SockAddr (Handshake Encrypted)
     , toxHandshakeCache :: HandshakeCache
-    , toxCryptoSessions :: NetCryptoSessions
     , toxCryptoKeys     :: TransportCrypto
     , toxRouting        :: DHT.Routing
     , toxTokens         :: TVar SessionTokens
@@ -217,97 +210,7 @@ data Tox extra  = Tox
     , toxAnnounceToLan  :: IO ()
     }
 
--- | initiate a netcrypto session, blocking
-netCrypto :: Tox extra -> SecretKey -> PublicKey{-UserKey -} -> IO [NetCryptoSession]
-netCrypto tox myseckey theirpubkey = netCryptoWithBackoff 1000000 tox myseckey theirpubkey
-
--- | helper for 'netCrypto', initiate a netcrypto session, retry after specified millisecs
-netCryptoWithBackoff :: Int -> Tox extra -> SecretKey -> PublicKey -> IO [NetCryptoSession]
-netCryptoWithBackoff millisecs tox myseckey theirpubkey = do
-    let mykeyAsId = key2id (toPublic myseckey)
-    -- TODO: check status of connection here:
-    mbContactsVar <- fmap contacts . HashMap.lookup mykeyAsId <$> atomically (readTVar (accounts (toxContactInfo tox)))
-    case mbContactsVar of
-      Nothing -> do
-        dput XNetCrypto ("netCrypto: (" ++ show mykeyAsId ++") accounts lookup failed.")
-        return []
-
-      Just contactsVar -> do
-        let theirkeyAsId = key2id theirpubkey
-        mbContact <- HashMap.lookup theirkeyAsId <$> atomically (readTVar contactsVar)
-        tup <- atomically $ do
-            mc <- HashMap.lookup theirkeyAsId <$> readTVar contactsVar
-            kp <- fmap join $ forM mc $ \c -> readTVar (contactKeyPacket     c)
-            sa <- fmap join $ forM mc $ \c -> readTVar (contactLastSeenAddr  c)
-            fr <- fmap join $ forM mc $ \c -> readTVar (contactFriendRequest c)
-            cp <- fmap join $ forM mc $ \c -> readTVar (contactPolicy        c)
-            return (kp,sa,fr,cp)
-        case tup of
-         (Nothing,Nothing,Nothing,Nothing) -> do
-            dput XNetCrypto ("netCrypto: (" ++ show mykeyAsId ++") friend not found (" ++ show theirkeyAsId ++ ").")
-            return []
-         (mbKeyPkt,Nothing,mbFR,mbPolicy) -> do
-            dput XNetCrypto ("netCrypto: (" ++ show mykeyAsId ++") no SockAddr for friend (" ++ show theirkeyAsId ++ "). TODO: search their node?")
-            return []
-         (Nothing,_,_,_) -> do
-            dput XNetCrypto ("netCrypto: (" ++ show mykeyAsId ++") no DHT-key for friend (" ++ show theirkeyAsId ++ "). TODO: what?")
-            return []
-         (Just (stamp_theirDhtKey,keyPkt),Just (stamp_saddr,saddr),mbFR,mbPolicy)
-                                  | theirDhtKey <- DHT.dhtpk keyPkt -> do
-            -- Do we already have an active session with this user?
-            sessionsMap <- atomically $ readTVar (netCryptoSessionsByKey (toxCryptoSessions tox) )
-            let sessionUsesIdentity key session = key == ncMyPublicKey session
-            case Map.lookup theirpubkey sessionsMap of
-            -- if sessions found, is it using this private key?
-              Just sessions | matchedSessions <- filter (sessionUsesIdentity (toPublic myseckey)) sessions
-                            , not (null matchedSessions)
-                            -> do
-                dput XNetCrypto ("netCrypto: Already have a session for " ++ show mykeyAsId ++ "<-->" ++ show theirkeyAsId)
-                return matchedSessions
-            -- if not, send handshake, this is separate session
-              _ -> do
-                -- if no session:
-                -- Convert to NodeInfo, so we can send cookieRequest
-                let crypto = toxCryptoKeys tox
-                    client = toxDHT tox
-                case nodeInfo (key2id theirDhtKey) (nodeAddr saddr) of
-                   Left e   -> dput XNetCrypto ("netCrypto: nodeInfo fail... " ++ e) >> return []
-                   Right ni -> do
-                    mbCookie <- DHT.cookieRequest crypto client (toPublic myseckey) ni
-                    case mbCookie of
-                      Nothing -> do
-                        dput XNetCrypto ("netCrypto: (" ++ show mykeyAsId ++") <--> (" ++ show theirkeyAsId ++ ").")
-                        dput XNetCrypto ("netCrypto: CookieRequest failed. TODO: dhtpkNodes thingy")
-                        return []
-                      Just cookie -> do
-                        dput XNetCrypto "Have cookie, creating handshake packet..."
-                        let hp = HParam { hpOtherCookie = cookie
-                                        , hpMySecretKey = myseckey
-                                        , hpCookieRemotePubkey = theirpubkey
-                                        , hpCookieRemoteDhtkey = theirDhtKey
-                                        , hpTheirBaseNonce = Nothing
-                                        , hpTheirSessionKeyPublic = Nothing
-                                        }
-                        newsession <- generateSecretKey
-                        timestamp <- getPOSIXTime
-                        (myhandshake,ioAction)
-                                <- atomically $ freshCryptoSession (toxCryptoSessions tox) (nodeAddr saddr) newsession timestamp hp
-                        ioAction
-                        -- send handshake
-                        forM myhandshake $ \response_handshake -> do
-                                sendHandshake (toxCryptoSessions tox) (nodeAddr saddr) response_handshake
-                        let secnum :: Double
-                            secnum = fromIntegral millisecs / 1000000
-                            delay = (millisecs * 5 `div` 4)
-                        if secnum < 20000000
-                          then do
-                            dput XNetCrypto $ "sent handshake, now delaying " ++ show (secnum * 1.25) ++ "  second(s).."
-                            -- threadDelay delay
-                            -- Commenting loop for simpler debugging
-                            return [] -- netCryptoWithBackoff delay tox myseckey theirpubkey -- hopefully it will find an active session this time.
-                          else do
-                            dput XNetCrypto "Unable to establish session..."
-                            return []
+
 
 -- | Create a DHTPublicKey packet to send to a remote contact.
 getContactInfo :: Tox extra -> IO DHT.DHTPublicKey
@@ -365,30 +268,24 @@ getOnionAlias crypto dhtself remoteNode = atomically $ do
 
 newTox :: TVar Onion.AnnouncedKeys   -- ^ Store of announced keys we are a rendezvous for.
        -> SockAddr                   -- ^ Bind-address to listen on.
-       -> Maybe NetCryptoSessions    -- ^ State of all one-on-one Tox links.
+       -> ( ContactInfo extra -> SockAddr -> Session -> IO () )
        -> Maybe SecretKey            -- ^ Optional DHT secret key to use.
        -> IO (Tox extra)
-newTox keydb addr mbSessionsState suppliedDHTKey = do
+newTox keydb addr onsess suppliedDHTKey = do
     (udp,sock) <- {- addVerbosity <$> -} udpTransport' addr
-    tox <- newToxOverTransport keydb addr mbSessionsState suppliedDHTKey udp
+    tox <- newToxOverTransport keydb addr onsess suppliedDHTKey udp
     return tox { toxAnnounceToLan = announceToLan sock (key2id $ transportPublic $ toxCryptoKeys tox) }
 
 -- | This version of 'newTox' is useful for automated tests using 'testPairTransport'.
 newToxOverTransport :: TVar Onion.AnnouncedKeys
         -> SockAddr
-        -> Maybe NetCryptoSessions
+        -> ( ContactInfo extra -> SockAddr -> Session -> IO () )
         -> Maybe SecretKey
         -> Onion.UDPTransport
         -> IO (Tox extra)
-newToxOverTransport keydb addr mbSessionsState suppliedDHTKey udp = do
+newToxOverTransport keydb addr onNewSession suppliedDHTKey udp = do
     roster <- newContactInfo
-    (crypto0,sessionsState0) <- case mbSessionsState of
-                                 Nothing -> do
-                                    crypto <- newCrypto
-                                    sessionsState <- newSessionsState crypto (const $ dput XUnexpected "Missing destroy hook!") defaultUnRecHook defaultCryptoDataHooks
-                                    return (crypto,sessionsState)
-                                 Just s -> return (transportCrypto s, s)
-
+    crypto0 <- newCrypto
     let -- patch in supplied DHT key
         crypto1 = fromMaybe crypto0 $do
             k <- suppliedDHTKey
@@ -409,6 +306,7 @@ newToxOverTransport keydb addr mbSessionsState suppliedDHTKey udp = do
     mkrouting <- DHT.newRouting addr crypto updateIP updateIP
     orouter <- newOnionRouter $ dput XRoutes
     (cryptonet,dhtcrypt,onioncrypt,dtacrypt,handshakes) <- toxTransport crypto orouter lookupClose udp
+    sessions <- initSessions (sendMessage cryptonet)
 
     let dhtnet0 = layerTransportM (DHT.decrypt crypto) (DHT.encrypt crypto) dhtcrypt
         tbl4 = DHT.routing4 $ mkrouting (error "missing client")
@@ -417,22 +315,12 @@ newToxOverTransport keydb addr mbSessionsState suppliedDHTKey udp = do
                     $ \client net -> onInbound (DHT.updateRouting client (mkrouting client) orouter) net
 
     hscache <- newHandshakeCache crypto (sendMessage handshakes)
-
-    let sessionsState = sessionsState0 { sendHandshake     = sendMessage handshakes
-                                       , sendSessionPacket = sendMessage cryptonet
-                                       , transportCrypto   = crypto
-                                       -- ToxContact -> STM Policy
-                                       , netCryptoPolicyByKey = policylookup
-                                       }
-        policylookup (ToxContact me them) = do
-            macnt <- HashMap.lookup me <$> readTVar (accounts roster)
-            case macnt of
-                Nothing   -> return RefusingToConnect
-                Just acnt -> do
-                    mc <- HashMap.lookup them <$> readTVar (contacts acnt)
-                    case mc of
-                        Nothing -> return RefusingToConnect
-                        Just c  -> fromMaybe RefusingToConnect <$> readTVar (contactPolicy c)
+    let sparams = SessionParams
+            { spCrypto           = crypto
+            , spSessions         = sessions
+            , spGetSentHandshake = getSentHandshake hscache
+            , spOnNewSession     = onNewSession roster addr
+            }
 
     orouter' <- forkRouteBuilder orouter
         $ \nid ni -> fmap (\(_,ns,_)->ns)
@@ -453,10 +341,9 @@ newToxOverTransport keydb addr mbSessionsState suppliedDHTKey udp = do
         { toxDHT            = dhtclient
         , toxOnion          = onionclient
         , toxToRoute        = onInbound (updateContactInfo roster) dtacrypt
-        , toxCrypto         = addHandler (dput XMisc) (sessionPacketH sessionsState) cryptonet
-        , toxHandshakes     = addHandler (dput XMisc) (handshakeH     sessionsState) handshakes
+        , toxCrypto         = addHandler (dput XMisc) (sessionHandler sessions) cryptonet
+        , toxHandshakes     = addHandler (dput XMisc) (handshakeH sparams) handshakes
         , toxHandshakeCache = hscache
-        , toxCryptoSessions = sessionsState
         , toxCryptoKeys     = crypto
         , toxRouting        = mkrouting dhtclient
         , toxTokens         = toks
@@ -526,8 +413,10 @@ announceToLan sock nid = do
                                       (Just "33445")
     let broadcast = addrAddress broadcast_info
         bs = S.runPut $ DHT.putMessage (DHT.DHTLanDiscovery nid)
+    dput XLan $ show broadcast ++ " <-- LanAnnounce " ++ show nid
     saferSendTo sock bs broadcast
 
+
 toxQSearch :: Tox extra -> Search NodeId (IP, PortNumber) Nonce32 NodeInfo Onion.Rendezvous
 toxQSearch tox = Onion.toxidSearch (onionTimeout tox) (toxCryptoKeys tox) (toxOnion tox)
 
diff --git a/src/Network/Tox/AggregateSession.hs b/src/Network/Tox/AggregateSession.hs
index edb897e0..1dd10eef 100644
--- a/src/Network/Tox/AggregateSession.hs
+++ b/src/Network/Tox/AggregateSession.hs
@@ -22,7 +22,6 @@ module Network.Tox.AggregateSession
 
 import Control.Concurrent.STM
 import Control.Concurrent.STM.TMChan
-import Control.Concurrent.Supply
 import Control.Monad
 import Data.Function
 import qualified Data.IntMap.Strict  as IntMap
@@ -47,9 +46,7 @@ import Network.Tox.Crypto.Transport (CryptoMessage (..), pattern KillPacket,
                                      pattern PacketRequest)
 import Network.Tox.DHT.Transport    (key2id)
 import Network.Tox.NodeId           (ToxProgress (..))
-import Network.Tox.Crypto.Handlers
-
-type Session = NetCryptoSession
+import Network.Tox.Session
 
 -- | For each component session, we track the current status.
 data SingleCon = SingleCon
@@ -113,47 +110,94 @@ data KeepAliveEvents = DoTimeout        -- ^ A session timed-out, close it.
                      | DoRequestMissing -- ^ Detect and request lost packets.
                      deriving Enum
 
--- | This function forks a thread to read all packets from the provided
--- 'Session' and forward them to 'contactChannel' for a containing
--- 'AggregateSession'
+-- | This call loops until the provided sesison is closed or times out.  It
+-- monitors the provided (non-empty) priority queue for scheduled tasks (see
+-- 'KeepAliveEvents') to perform for the connection.
+keepAlive :: Session -> TVar (PSQ POSIXTime) -> IO ()
+keepAlive s q = do
+    myThreadId >>= flip labelThread
+        (intercalate "." ["beacon"
+                         , take 8 $ show $ key2id $ sTheirUserKey s
+                         , show $ sSessionID s])
+
+    let outPrint e = dput XNetCrypto $ shows (sSessionID s,sTheirAddr s) $ " <-- " ++ e
+
+        doAlive = do
+            -- outPrint $ "Beacon"
+            sendMessage (sTransport s) () (OneByte PING)
+
+        doRequestMissing = do
+            (ns,nmin) <- sMissingInbound s
+            -- outPrint $ "PacketRequest " ++ show (nmin,ns)
+            sendMessage (sTransport s) () (RequestResend PacketRequest ns)
+
+        re tm again e io = do
+            io
+            atomically $ modifyTVar' q $ PSQ.insert (fromEnum e) tm
+            again
+
+        doEvent again now e = case e of
+            DoTimeout        -> do dput XNetCrypto $ "TIMEOUT: " ++ show (sSessionID s)
+                                   sClose s
+            DoAlive          -> re (now + 10) again e doAlive
+            DoRequestMissing -> re (now + 5)  again e doRequestMissing -- tox-core does this at 1 second intervals
+
+    fix $ \again -> do
+
+    now <- getPOSIXTime
+    join $ atomically $ do
+        Just ( k :-> tm ) <- PSQ.findMin <$> readTVar q
+        return $ if now < tm then threadDelay (toMicroseconds $ tm - now) >> again
+                             else doEvent again now (toEnum k)
+
+-- | This function forks two threads: the 'keepAlive' beacon-sending thread and
+-- a thread to read all packets from the provided 'Session' and forward them to
+-- 'contactChannel' for a containing 'AggregateSession'
 forkSession :: AggregateSession -> Session -> (Status ToxProgress -> STM ()) -> IO ThreadId
 forkSession c s setStatus = forkIO $ do
     myThreadId >>= flip labelThread
         (intercalate "." ["s"
-                         , take 8 $ show $ key2id $ ncTheirPublicKey s
+                         , take 8 $ show $ key2id $ sTheirUserKey s
                          , show $ sSessionID s])
-    tmchan <- atomically $ do
-        tmchan <- newTMChan
-        supply <- readTVar (listenerIDSupply $ ncAllSessions s)
-        let (listenerId,supply') = freshId supply
-        writeTVar (listenerIDSupply $ ncAllSessions s) supply'
-        modifyTVar' (ncListeners s) (IntMap.insert listenerId (0,tmchan))
-        return tmchan
+
+    q <- atomically $ newTVar $ fromList
+            [ fromEnum DoAlive :-> 0
+            , fromEnum DoRequestMissing :-> 0
+            ]
 
     let sendPacket :: CryptoMessage -> STM ()
         sendPacket msg = writeTMChan (contactChannel c) (sSessionID s, msg)
 
-        inPrint e = dput XNetCrypto $ shows (sSessionID s,ncSockAddr s) $ " --> " ++ e
+        inPrint e = dput XNetCrypto $ shows (sSessionID s,sTheirAddr s) $ " --> " ++ e
+
+        bump = do
+            -- inPrint $ "BUMP: " ++ show (sSessionID s)
+            now <- getPOSIXTime
+            atomically $ modifyTVar' q $ PSQ.insert (fromEnum DoTimeout) (now + 15)
 
         onPacket body loop Nothing          = return ()
         onPacket body loop (Just (Left e))  = inPrint e >> loop
         onPacket body loop (Just (Right x)) = body loop x
 
-        awaitPacket body = fix $ (.) (fmap Right <$> atomically (readTMChan tmchan) >>=)
-                               $ onPacket body
+        awaitPacket body = fix $ awaitMessage (sTransport s) . onPacket body
 
     atomically $ setStatus $ InProgress AwaitingSessionPacket
-    atomically $ setStatus Established
-    awaitPacket $ \loop x -> do
-        case msgID x of
-            KillPacket -> return ()
-            _          -> atomically (sendPacket x) >> loop
-
-    atomically $ setStatus Dormant
-
-
-sSessionID :: Session -> Int
-sSessionID s = fromIntegral $ ncSessionId s
+    awaitPacket $ \_ (online,()) -> do
+        when (msgID online /= ONLINE) $ do
+            inPrint $ "Unexpected initial packet: " ++ show (msgID online)
+        atomically $ do setStatus Established
+                        sendPacket online
+        bump
+        beacon <- forkIO $ keepAlive s q
+        awaitPacket $ \awaitNext (x,()) -> do
+            bump
+            case msgID x of
+                PING       -> return ()
+                KillPacket -> sClose s
+                _          -> atomically $ sendPacket x
+            awaitNext
+        atomically $ setStatus Dormant
+        killThread beacon
 
 -- | Add a new session (in 'AwaitingSessionPacket' state) to the
 -- 'AggregateSession'.  If the supplied session is not compatible because it is
@@ -166,8 +210,8 @@ sSessionID s = fromIntegral $ ncSessionId s
 addSession :: AggregateSession -> Session -> IO AddResult
 addSession c s = do
     (result,mcon,replaced) <- atomically $ do
-        let them = ncTheirPublicKey s
-            me   = ncMyPublicKey s
+        let them = sTheirUserKey s
+            me   = toPublic $ sOurKey s
         compat <- checkCompatible me them c
         let result = case compat of
                         Nothing      -> FirstSession
@@ -184,7 +228,7 @@ addSession c s = do
                                 writeTVar (contactSession c) imap'
                                 return (result,Just con,s0)
 
-    mapM_ (destroySession . singleSession) replaced
+    mapM_ (sClose . singleSession) replaced
     forM_ mcon $ \con ->
         forkSession c s $ \progress -> do
             writeTVar (singleStatus con) progress
@@ -203,6 +247,7 @@ addSession c s = do
                     return emap'
             writeTVar (contactEstablished c) emap'
     return result
+
 -- | Information returned from 'delSession'.
 data DelResult = NoSession      -- ^ Contact is completely disconnected.
                | DeletedSession -- ^ Connection removed but session remains active.
@@ -230,11 +275,10 @@ delSession c sid = do
         writeTVar (contactSession c) imap'
         writeTVar (contactEstablished c) emap'
         return ( IntMap.lookup sid imap, IntMap.null imap')
-    mapM_ (destroySession . singleSession) con
+    mapM_ (sClose . singleSession) con
     return $ if r then NoSession
                   else DeletedSession
 
-
 -- | Send a packet to one or all of the component sessions in the aggregate.
 dispatchMessage :: AggregateSession -> Maybe Int -- ^ 'Nothing' to broadcast, otherwise SessionID.
                                     -> CryptoMessage -> IO ()
@@ -242,11 +286,7 @@ dispatchMessage c msid msg = join $ atomically $ do
     imap <- readTVar (contactSession c)
     let go = case msid of Nothing  -> forM_ imap
                           Just sid -> forM_ (IntMap.lookup sid imap)
-    return $ go $ \con -> do
-        eResult <- sendLossless (transportCrypto (ncAllSessions (singleSession con))) (singleSession con) msg
-        case eResult of
-            Left msg -> dput XJabber msg
-            Right pkt -> dput XJabber ("sendLossLess SUCCESS: " ++ show pkt)
+    return $ go $ \con -> sendMessage (sTransport $ singleSession con) () msg
 
 -- | Retry until:
 --
@@ -287,7 +327,6 @@ aggregateStatus c = do
           | not (IntMap.null imap) -> InProgress AwaitingSessionPacket
           | otherwise              -> Dormant
 
-
 -- | Query whether the supplied ToxID keys are compatible with this aggregate.
 --
 --   [ Nothing ]    Any keys would be compatible because there is not yet any
@@ -304,8 +343,8 @@ checkCompatible me them c = do
     imap <- readTVar (contactSession c)
     return $ case IntMap.elems imap of
                     _ | isclosed -> Just False -- All keys are incompatible (closed).
-                    con:_        -> Just $ ncTheirPublicKey (singleSession con)   == them
-                                           && (ncMyPublicKey $ singleSession con) == me
+                    con:_        -> Just $ sTheirUserKey (singleSession con)         == them
+                                           && toPublic (sOurKey $ singleSession con) == me
                     []           -> Nothing
 
 -- | Returns the local and remote keys that are compatible with this aggregate.
@@ -317,6 +356,6 @@ compatibleKeys c = do
     imap <- readTVar (contactSession c)
     return $ case IntMap.elems imap of
                     _ | isclosed -> Nothing -- none.
-                    con:_        -> Just ( ncMyPublicKey $ singleSession con
-                                         , ncTheirPublicKey (singleSession con))
+                    con:_        -> Just ( toPublic (sOurKey $ singleSession con)
+                                         , sTheirUserKey (singleSession con))
                     []           -> Nothing -- any.