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
|
-- |
--
-- 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 Data.IP (IP)
import Network.Tox.Crypto.Transport
import Network.Tox.NodeId
-- import Network.Tox.DHT.Handlers (nodeSearch)
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 System.Timeout
-- | This type indicates the progress of a tox encrypted friend link
-- connection. Two scenarios are illustrated below. The parenthesis show the
-- current 'G.Status' 'ToxProgress' of the session.
--
--
-- Perfect handshake scenario:
--
-- Peer 1 Peer 2
-- (InProgress AcquiringCookie) (Dormant/InProgress AcquiringCookie)
-- Cookie request ->
-- <- Cookie response
-- (InProgress AwaitingHandshake) (Dormant/InProgress AcquiringCookie)
-- Handshake packet ->
-- * accepts connection
-- (InProgress AwaitingSessionPacket)
-- <- Handshake packet
-- *accepts connection
-- (InProgress AwaitingSessionPacket)
-- Encrypted packet -> <- Encrypted packet
-- *confirms connection *confirms connection
-- (Established) (Established)
--
-- Connection successful.
--
-- Encrypted packets -> <- Encrypted packets
--
--
--
--
-- More realistic handshake scenario:
-- Peer 1 Peer 2
-- (InProgress AcquiringCookie) (Dormant/InProgress AcquiringCookie)
-- Cookie request -> *packet lost*
-- Cookie request ->
-- <- Cookie response
-- (InProgress AwaitingHandshake) (Dormant/InProgress AcquiringCookie)
--
-- *Peer 2 randomly starts new connection to peer 1
-- (InProgress AcquiringCookie)
-- <- Cookie request
-- Cookie response ->
-- (InProgress AwaitingHandshake)
--
-- Handshake packet -> <- Handshake packet
-- *accepts connection * accepts connection
-- (InProgress AwaitingSessionPacket) (InProgress AwaitingSessionPacket)
--
-- Encrypted packet -> <- Encrypted packet
-- *confirms connection *confirms connection
-- (Established) (Established)
--
-- Connection successful.
--
-- Encrypted packets -> <- Encrypted packets
data ToxProgress
= AwaitingDHTKey -- ^ Waiting to receive their DHT key.
| AcquiringIPAddress -- ^ Searching DHT to obtain their node's IP & port.
| AcquiringCookie -- ^ Attempting to obtain a cookie.
| AwaitingHandshake -- ^ Waiting to receive a handshake.
| AwaitingSessionPacket -- ^ Connection is "accepted" but not yet "confirmed".
deriving (Eq,Ord,Enum,Show)
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 -> IO ()
acceptContact getPolicy AcceptContactMethods{..} = fix $ \loop -> do
join $ atomically $ do
orElse
(getPolicy >>= \case
RefusingToConnect -> return $ return () -- QUIT Dormant/Established
_ -> retry)
(do hs <- getHandshake
handshakeIsSuitable hs >>= \case
True -> do
-- Here we allocate a NetCrypto session for handling CryptoPacket.
transitionToState (InProgress AwaitingSessionPacket)
return loop
False -> return loop)
whileTryingAndNotEstablished :: STM Policy -> STM (Status t) -> ((Int -> IO ()) -> STM (IO ())) -> IO ()
whileTryingAndNotEstablished getPolicy getStatus body = fix $ \loop -> do
let retryWhileTrying k = getPolicy >>= \case
TryingToConnect -> retry
_ -> 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
{ getHandshakeParams :: STM params
, sendHandshake :: params -> IO ()
, retryInterval :: Int
}
-- | 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 -> IO ()
persueContact getPolicy getStatus PersueContactMethods{..}
= whileTryingAndNotEstablished getPolicy getStatus
$ \retryAfterTimeout -> do
-- AwaitingDHTKey
-- AcquiringIPAddress
params <- getHandshakeParams
return $ do -- AcquiringCookie
-- AwaitingHandshake
-- AwaitingSessionPacket
sendHandshake params
retryAfterTimeout retryInterval
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 -> IO ()
freshenContact getPolicy getStatus FreshenContactMethods{..}
= whileTryingAndNotEstablished getPolicy getStatus
-- retryAfterTimeout :: Int -> IO ()
$ \retryAfterTimeout ->
getDHTKey >>= \case
Nothing -> -- AwaitingDHTKey
retry
Just dk -> getSockAddr >>= return . \case
Nothing -> -- AcquiringIPAddress
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 -> -- AcquiringCookie
-- AwaitingHandshake
-- AwaitingSessionPacket
do _todo_search_toxid_send_dhtkey -- 123 _todo_search_toxid_send_dhtkey :: IO a0
retryAfterTimeout dhtkeyInterval
|
