1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
|
-- |
--
-- This module defines three tasks intended to be run in separate threads:
--
-- * 'acceptContact'
--
-- * 'persueContact'
--
-- * 'freshenContact'
--
{-# LANGUAGE LambdaCase #-}
module Connection.Tox.Threads where
import Connection
-- import Connection.Tox
import Crypto.Tox
import Data.IP (IP)
import Network.Tox.Crypto.Transport
import Network.Tox.Crypto.Handlers
import Network.Tox.NodeId
import Network.Tox.ContactInfo
import Network.Tox.DHT.Handlers {- (nodeSearch) -} as DHT
import Network.Tox.DHT.Transport as DHT (dhtpk)
import Network.Socket
import Network.Kademlia.Search
import Network.Kademlia.Routing (BucketList)
import Control.Concurrent.STM
import Control.Monad
import Data.Function
import Data.Functor.Identity
import Data.Time.Clock.POSIX
import System.IO
import System.Timeout
type NodeSearch = Search NodeId (IP,PortNumber) () NodeInfo NodeInfo
data AcceptContactMethods = AcceptContactMethods
{ getHandshake :: STM (Handshake Identity)
, handshakeIsSuitable :: Handshake Identity -> STM Bool
, transitionToState :: Status ToxProgress -> STM ()
}
-- | Invokes an STM action on each incoming handshake.
--
-- Does not return until getPolicy yields RefusingToConnect.
acceptContact :: STM Policy -> AcceptContactMethods -> (Status ToxProgress -> STM ()) -> IO ()
acceptContact getPolicy AcceptContactMethods{..} writeState = fix $ \loop -> do
join $ atomically $ do
orElse
(getPolicy >>= \case
RefusingToConnect -> do writeState Dormant
return $ return () -- QUIT Dormant/Established
_ -> retry)
(do hs <- getHandshake
handshakeIsSuitable hs >>= \case
True -> do
-- Here we allocate a NetCrypto session for handling CryptoPacket.
writeState (InProgress AwaitingSessionPacket)
transitionToState (InProgress AwaitingSessionPacket)
return loop
False -> return loop)
whileTryingAndNotEstablished :: STM Policy
-> STM (Status t)
-> (Status ToxProgress -> STM ())
-> ((Int -> IO ()) -> STM (IO ()))
-> IO ()
whileTryingAndNotEstablished getPolicy getStatus writeStatus body = fix $ \loop -> do
let retryWhileTrying k = getPolicy >>= \case
TryingToConnect -> retry
_ -> do writeStatus Dormant
return k
ifEstablished t e = getStatus >>= \case
Established -> t
_ -> e
retryAfterTimeout interval = do
timeout interval $ atomically
$ orElse
(retryWhileTrying ())
(ifEstablished (return ()) retry)
loop
join $ atomically $ orElse
(retryWhileTrying (return ())) -- QUIT Dormant/Established
(ifEstablished retry
(body retryAfterTimeout))
data PersueContactMethods params = PersueContactMethods
{ allsessions :: NetCryptoSessions
, myseckey :: SecretKey
, theirpubkey :: PublicKey
, client :: DHT.Client
, shortRetryInterval :: Int -- successful cookie, try again soon.
, longRetryInterval :: Int -- no cookie, he's offline, give it some time.
, contact :: Contact
}
retryUntilJust :: TVar (Maybe a) -> STM a
retryUntilJust tvar = maybe retry return =<< readTVar tvar
-- | Continuously attempt to send handshake packets until a connection is
-- established.
--
-- As long as getPolicy is TryingToConnect and there is no established
-- connection, this function will continue.
persueContact :: STM Policy
-> STM (Status t)
-> PersueContactMethods a
-> (Status ToxProgress -> STM ())
-> IO ()
persueContact getPolicy getStatus PersueContactMethods{..} writeStatus = do
-- AwaitingDHTKey
atomically $ writeStatus (InProgress AwaitingDHTKey)
whileTryingAndNotEstablished getPolicy getStatus writeStatus
$ \retryAfterTimeout ->
orElse (do
theirDhtKey <- DHT.dhtpk <$> retryUntilJust (contactKeyPacket contact)
-- We don't have an IP address yet.
maybe (return ()) (const retry) =<< readTVar (contactLastSeenAddr contact)
return $ do -- AcquiringIPAddress
atomically $ writeStatus (InProgress AcquiringIPAddress)
retryAfterTimeout 0)
(do
theirDhtKey <- DHT.dhtpk <$> retryUntilJust (contactKeyPacket contact)
saddr <- retryUntilJust (contactLastSeenAddr contact)
ni <- either (const retry) return $ nodeInfo (key2id theirDhtKey) saddr
return $ do
-- AcquiringCookie
atomically $ writeStatus (InProgress AcquiringCookie)
let mykeyAsId = key2id (toPublic myseckey)
theirkeyAsId = key2id theirpubkey
crypto = transportCrypto allsessions
mbCookie <- -- TODO: Check for recent cached cookie.
DHT.cookieRequest crypto client (toPublic myseckey) ni
interval <- case mbCookie of
Nothing -> do
hPutStrLn stderr ("persueContact: (" ++ show mykeyAsId ++") <--> (" ++ show theirkeyAsId ++ ").")
hPutStrLn stderr ("persueContact: CookieRequest failed. TODO: dhtpkNodes thingy")
return longRetryInterval
Just cookie -> do
hPutStrLn stderr "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 allsessions saddr newsession timestamp hp
ioAction
-- send handshake
forM myhandshake $ \response_handshake -> do
sendHandshake allsessions saddr response_handshake
atomically $ writeStatus $ InProgress AwaitingHandshake
return shortRetryInterval
-- AwaitingHandshake
-- AwaitingSessionPacket
retryAfterTimeout interval)
data FreshenContactMethods = FreshenContactMethods
{ dhtkeyInterval :: Int
, sockAddrInterval :: Int
, nodeSch :: NodeSearch
, getDHTKey :: STM (Maybe NodeId)
, getSockAddr :: STM (Maybe SockAddr)
, getBuckets :: STM (BucketList NodeInfo)
}
-- send my dht key
-- search for their sockaddr
-- monitor new dht key
-- monitor new sockaddr
--
-- Keep going while TryingToConnect
-- pause while Established
-- Useful:
-- toxidSearch onionTimeout
-- newSearch
-- searchLoop
-- searchCancel
-- -> (r -> STM Bool) -- ^ Invoked on each result. Return False to quit searching.
-- | Continuously search the DHT to obtain ip addresses and to send your dht
-- key to contacts.
--
-- As long as getPolicy is TryingToConnect and there is no established
-- connection, this function will continue.
freshenContact :: STM Policy -> STM (Status t) -> FreshenContactMethods
-> (Status ToxProgress -> STM ())
-> IO ()
freshenContact getPolicy getStatus FreshenContactMethods{..} writeStatus
= whileTryingAndNotEstablished getPolicy getStatus writeStatus
-- retryAfterTimeout :: Int -> IO ()
$ \retryAfterTimeout ->
getDHTKey >>= \case
Nothing -> -- AwaitingDHTKey
retry
Just dk -> getSockAddr >>= \case
Nothing -> do -- AcquiringIPAddress
writeStatus (InProgress AcquiringIPAddress)
return $
do bkts <- atomically $ getBuckets
st <- search nodeSch bkts dk $
\r -> do -- TODO: store saddr, check for finish
return True
atomically $ searchIsFinished st >>= check
-- TODO: searchCancel on stop condition
retryAfterTimeout sockAddrInterval
Just a -> do
writeStatus (InProgress AcquiringCookie)
return $
-- AcquiringCookie
-- AwaitingHandshake
-- AwaitingSessionPacket
do _todo_search_toxid_send_dhtkey -- 123 _todo_search_toxid_send_dhtkey :: IO a0
retryAfterTimeout dhtkeyInterval
|
