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
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
|
{-# LANGUAGE NondecreasingIndentation #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE RecordWildCards #-}
{-# LANGUAGE CPP #-}
import Control.Arrow;
import Control.Monad
import Control.Monad.Logger
import Control.Monad.Reader
import Data.Char
import Data.Default
import Data.List as L
import Data.Maybe
import Data.String
import qualified Data.ByteString as B (ByteString,writeFile,readFile)
; import Data.ByteString (ByteString)
import qualified Data.ByteString.Char8 as B8
import System.IO
import System.IO.Error
import Text.PrettyPrint.HughesPJClass
import Text.Printf
import Control.Monad.Reader.Class
import System.Posix.Process (getProcessID)
import Network.BitTorrent.Address
import Network.BitTorrent.DHT
import qualified Network.BitTorrent.DHT.Routing as R
import Network.BitTorrent.DHT.Session
import Network.SocketLike
import Network.StreamServer
import Control.Exception.Lifted as Lifted
#ifdef THREAD_DEBUG
import Control.Concurrent.Lifted.Instrument
import Data.Time.Clock
#else
import Control.Concurrent
#endif
mkNodeAddr :: SockAddr -> NodeAddr IPv4
mkNodeAddr addr = NodeAddr (fromJust $ fromSockAddr addr)
(fromMaybe 0 $ sockAddrPort addr) -- FIXME
btBindAddr :: String -> Bool -> IO (NodeAddr IPv4)
btBindAddr s b = mkNodeAddr <$> getBindAddress s b
printReport :: MonadIO m => [(String,String)] -> m ()
printReport kvs = liftIO $ do
putStrLn (showReport kvs)
hFlush stdout
showReport :: [(String,String)] -> String
showReport kvs = do
let colwidth = maximum $ map (length . fst) kvs
(k,v) <- kvs
concat [ printf " %-*s" (colwidth+1) k, v, "\n" ]
showEnry :: Show a => (NodeInfo a, t) -> [Char]
showEnry (n,_) = intercalate " "
[ show $ pPrint (nodeId n)
, show $ nodeAddr n
]
printTable :: DHT IPv4 ()
printTable = do
t <- showTable
liftIO $ do
putStrLn t
hFlush stdout
showTable :: DHT IPv4 String
showTable = do
nodes <- R.toList <$> getTable
return $ showReport
$ map (show *** showEnry)
$ concat $ zipWith map (map (,) [0::Int ..]) nodes
bootstrapNodes :: IO [NodeAddr IPv4]
bootstrapNodes = mapMaybe fromAddr
<$> mapM resolveHostName defaultBootstrapNodes
-- ExtendedCaps (Map.singleton
noDebugPrints :: LogSource -> LogLevel -> Bool
noDebugPrints _ = \case LevelDebug -> False
_ -> True
noLogging :: LogSource -> LogLevel -> Bool
noLogging _ _ = False
allNoise :: LogSource -> LogLevel -> Bool
allNoise _ _ = True
resume :: DHT IPv4 (Maybe B.ByteString)
resume = do
restore_attempt <- liftIO $ tryIOError $ B.readFile "dht-nodes.dat"
saved_nodes <-
either (const $ do liftIO $ putStrLn "Error reading dht-nodes.dat"
return Nothing)
(return . Just)
restore_attempt
return saved_nodes
godht :: forall b. (NodeAddr IPv4 -> NodeId -> DHT IPv4 b) -> IO b
godht f = do
a <- btBindAddr "8008" False
dht def { optTimeout = 5 } a allNoise $ do
me0 <- asks tentativeNodeId
printReport [("tentative node-id",show $ pPrint me0)
,("listen-address", show a)
]
f a me0
marshalForClient :: String -> String
marshalForClient s = show (length s) ++ ":" ++ s
hPutClient :: Handle -> String -> IO ()
hPutClient h s = hPutStr h (marshalForClient s)
clientSession :: Node IPv4 -> MVar () -> RestrictedSocket -> Int -> Handle -> IO ()
clientSession st signalQuit sock n h = do
line <- map toLower . dropWhile isSpace <$> hGetLine h
let cmd0 action = action >> clientSession st signalQuit sock n h
cmd action = cmd0 $ join $ runDHT st action
(c,args) = second (dropWhile isSpace) $ break isSpace line
case (c,args) of
("quit", _) -> hPutClient h "" >> hClose h
("stop", _) -> do hPutClient h "Terminating DHT Daemon."
hClose h
putMVar signalQuit ()
("ls", _) -> cmd $ do
tbl <- getTable
t <- showTable
me <- myNodeIdAccordingTo (read "8.8.8.8:6881")
ip <- routableAddress
return $ do
hPutClient h $ unlines
[ t
, showReport
[ ("node-id", show $ pPrint me)
, ("internet address", show ip)
, ("buckets", show $ R.shape tbl)]
]
("external-ip", _) -> cmd $ do
ip <- routableAddress
return $ do
hPutClient h $ maybe "" (takeWhile (/=':') . show) ip
("swarms", s) -> cmd $ do
let fltr = case s of
('-':'v':cs) | all isSpace (take 1 cs)
-> const True
_ -> (\(h,c,n) -> c/=0 )
ss <- getSwarms
let r = map (\(h,c,n) -> (unwords [show h,show c], maybe "" show n))
$ filter fltr ss
return $ do
hPutClient h $ showReport r
("peers", s) -> cmd $ do
let ih = fromString s
ps <- allPeers ih
return $ do
hPutClient h $ showReport $ map (((,) "") . show . pPrint) ps
("pid", _) -> cmd $ return $ do
pid <- getProcessID
hPutClient h (show pid)
#ifdef THREAD_DEBUG
("threads", _) -> cmd $ return $ do
ts <- threadsInformation
tm <- getCurrentTime
let r = map (\PerThread{..} -> (show lbl,show (diffUTCTime tm startTime))) ts
hPutClient h $ showReport r
#endif
_ -> cmd0 $ hPutClient h "error."
main :: IO ()
main = do
godht $ \a me0 -> do
printTable
bs <- liftIO bootstrapNodes
`onException`
(Lifted.ioError $ userError "unable to resolve bootstrap nodes")
saved_nodes <- resume
peers'trial <- liftIO $ tryIOError $ B.readFile "bt-peers.dat"
saved_peers <-
either (const $ do liftIO $ putStrLn "Error reading bt-peers.dat"
return Nothing)
(return . Just)
peers'trial
maybe (return ()) mergeSavedPeers saved_peers
when (isJust saved_nodes) $ do
b <- isBootstrapped
tbl <- getTable
bc <- optBucketCount <$> asks options
printTable
me <- case concat $ R.toList tbl of
(n,_):_ -> myNodeIdAccordingTo (nodeAddr n)
_ -> return me0
printReport [("node-id",show $ pPrint me)
,("listen-address", show a)
,("bootstrapped", show b)
,("buckets", show $ R.shape tbl)
,("optBucketCount", show bc)
,("dht-nodes.dat", "Running bootstrap...")
]
st <- ask
waitForSignal <- liftIO $ do
signalQuit <- newEmptyMVar
srv <- streamServer (withSession $ clientSession st signalQuit) (SockAddrUnix "dht.sock")
return $ liftIO $ do
() <- takeMVar signalQuit
quitListening srv
bootstrap saved_nodes bs
b <- isBootstrapped
tbl <- getTable
bc <- optBucketCount <$> asks options
printTable
ip <- routableAddress
me <- case concat $ R.toList tbl of
(n,_):_ -> myNodeIdAccordingTo (nodeAddr n)
_ -> return me0
printReport [("node-id",show $ pPrint me)
,("internet address", show ip)
,("listen-address", show a)
,("bootstrapped", show b)
,("buckets", show $ R.shape tbl)
,("optBucketCount", show bc)
]
waitForSignal -- Await unix socket to signal termination.
snapshot >>= liftIO . B.writeFile "dht-nodes.dat"
savePeerStore >>= liftIO . B.writeFile "bt-peers.dat"
|
