summaryrefslogtreecommitdiff
path: root/examples/dhtd.hs
blob: 68f9144697b0021a99bcecaac2fd539b5f452f0f (plain)
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
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
{-# LANGUAGE NondecreasingIndentation #-}
{-# LANGUAGE FlexibleContexts      #-}
{-# LANGUAGE FlexibleInstances     #-}
{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE LambdaCase            #-}
{-# LANGUAGE OverloadedStrings     #-}
{-# LANGUAGE ScopedTypeVariables   #-}
{-# LANGUAGE TupleSections         #-}
{-# LANGUAGE RecordWildCards       #-}
{-# LANGUAGE PartialTypeSignatures #-}
{-# LANGUAGE CPP #-}
{-# LANGUAGE RankNTypes #-}

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.Set as Set
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 Text.Read
import Control.Monad.Reader.Class
import System.Posix.Process (getProcessID)
import GHC.Stats
import System.Mem

import Data.Torrent (InfoHash)
import Network.Address
import Network.BitTorrent.DHT
import Network.BitTorrent.DHT.Search
import Network.BitTorrent.DHT.Query
import Network.DHT.Mainline (FindNode(..),NodeFound(..),GetPeers(..),GotPeers(..))
import Network.DatagramServer (QueryFailure(..))
import Network.DatagramServer.Mainline (ReflectedIP(..),KMessageOf)
import qualified Network.DatagramServer.Tox as Tox
import qualified Network.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 ()
import Data.Time.Clock
#else
import Control.Concurrent
#endif
import Control.Concurrent.STM
import System.Environment
import Data.BEncode (BValue)
import Network.DHT.Types
import Network.DatagramServer.Types
import Data.Bits
import Data.Serialize
import Network.KRPC.Method
import Data.Typeable
import GHC.Generics

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
    , Pretty (NodeId dht)
    ) => (NodeInfo dht a u, t) -> [Char]
showEnry (n,_) = intercalate " "
    [ show $ pPrint (nodeId n)
    , show $ nodeAddr n
    ]

printTable ::
    ( Pretty (NodeId dht)
    ) => DHT raw dht u IPv4 ()
printTable = do
    t <- showTable
    liftIO $ do
        putStrLn t
        hFlush stdout

showTable ::
    ( Pretty (NodeId dht)
    ) => DHT raw dht u 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
                        LevelOther _ -> False
                        _            -> True

noLogging :: LogSource -> LogLevel -> Bool
noLogging _ _ = False

allNoise :: LogSource -> LogLevel ->  Bool
allNoise _ _ = True

resume :: DHT raw dht u 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 ::
    ( Eq (QueryMethod dht)
    , Show (QueryMethod dht)
    , Functor dht
    , Ord (TransactionID dht)
    , Serialize (TransactionID dht)
    , Kademlia dht
    , WireFormat raw dht
    , DataHandlers raw dht
    , SerializableTo raw (Query dht (FindNode dht IPv4))
    , SerializableTo raw (Response dht (NodeFound dht IPv4))
    , SerializableTo raw (Query dht (Ping dht))
    , SerializableTo raw (Response dht (Ping dht))
    , KRPC dht (Query dht (FindNode dht IPv4)) (Response dht (NodeFound dht IPv4))
    , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
    , Ord (NodeId dht)
    , FiniteBits (NodeId dht)
    , Serialize (NodeId dht)
    , Show (NodeId dht)
    , Pretty (NodeId dht)
    , Pretty (NodeInfo dht IPv4 u)
    , Default u
    , Show u
    ) => String -> (NodeAddr IPv4 -> NodeId dht -> DHT raw dht u IPv4 b) -> IO b
godht p f = do
    a <- btBindAddr p False
    dht def { optTimeout = 5 } a noDebugPrints $ 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)

hPutClientChunk :: Handle -> String -> IO ()
hPutClientChunk h s = hPutStr h (' ' : marshalForClient s)

data GenericDHT ip a
    = GenericDHT
        (forall raw dht u.
            ( Eq (QueryMethod dht)
            , Show (QueryMethod dht)
            , Functor dht
            , Ord (TransactionID dht)
            , Serialize (TransactionID dht)
            , Kademlia dht
            , WireFormat raw dht
            , DataHandlers raw dht
            , SerializableTo raw (Query dht (FindNode dht ip))
            , SerializableTo raw (Response dht (NodeFound dht ip))
            , SerializableTo raw (Query dht (Ping dht))
            , SerializableTo raw (Response dht (Ping dht))
            , KRPC dht (Query dht (FindNode dht ip)) (Response dht (NodeFound dht ip))
            , KRPC dht (Query dht (Ping dht)) (Response dht (Ping dht))
            , Ord (NodeId dht)
            , FiniteBits (NodeId dht)
            , Serialize (NodeId dht)
            , Show (NodeId dht)
            , Pretty (NodeId dht)
            , Pretty (NodeInfo dht ip ())
            , Pretty (NodeInfo dht ip u)
            , Default u
            , Show u
            , Read (NodeId dht)
            ) => DHT raw dht u ip a)
    | BtDHT (DHT BValue KMessageOf () ip a)

dhtType :: DHT raw dht u ip (Proxy dht)
dhtType = return Proxy

nodeIdType :: NodeId dht -> DHT raw dht u ip ()
nodeIdType _ = return ()

nodeAddrType :: NodeAddr ip -> DHT raw dht u ip ()
nodeAddrType _ = return ()

ipType :: f dht ip -> DHT raw dht u ip ()
ipType _ = return ()

instance Kademlia Tox.Message where
instance Pretty (NodeId Tox.Message) where
instance Pretty (NodeInfo Tox.Message IPv4 ()) where
instance Pretty (NodeInfo Tox.Message IPv4 Bool) where -- TODO
instance Read (NodeId KMessageOf) where
    readsPrec d s = map (\(ih,s) -> (toNodeId (ih::InfoHash),s)) $ readsPrec d s
instance Read (NodeId Tox.Message) where -- TODO
instance Serialize (FindNode Tox.Message IPv4) where
    get = error "TODO get"
    put = error "TODO put"
instance Serialize (NodeFound Tox.Message IPv4) where
    get = error "TODO get"
    put = error "TODO put"
instance Serialize (Ping Tox.Message) where
    get = error "TODO get"
    put = error "TODO put"
instance Serialize (Query Tox.Message (FindNode Tox.Message IPv4)) where
    get = error "TODO get"
    put = error "TODO put"
instance Serialize (Query Tox.Message (Ping Tox.Message)) where -- TODO
    get = error "TODO get"
    put = error "TODO put"
instance Serialize (Response Tox.Message (NodeFound Tox.Message IPv4)) where
    get = error "TODO get"
    put = error "TODO put"
instance Serialize (Response Tox.Message (Ping Tox.Message)) where -- TODO
    get = error "TODO get"
    put = error "TODO put"
instance KRPC Tox.Message (Query    Tox.Message (FindNode Tox.Message IPv4))
              (Response Tox.Message (NodeFound Tox.Message IPv4)) where
    method = error "TODO method"
instance KRPC Tox.Message (Query    Tox.Message (Ping Tox.Message ))
              (Response Tox.Message (Ping Tox.Message )) where
    method = error "TODO method"
instance DataHandlers ByteString Tox.Message where


-- instance Generic (Response Tox.Message (NodeFound Tox.Message IPv4)) where -- TODO

instance Default Bool where def = False

clientSession :: Node BValue KMessageOf () IPv4 -> Node B.ByteString Tox.Message Bool IPv4 -> MVar () -> Bool ->  RestrictedSocket -> Int -> Handle -> IO ()
clientSession bt tox signalQuit isBt sock n h = do
    line <- map toLower . dropWhile isSpace <$> hGetLine h
    let dht :: Either (Node BValue KMessageOf () IPv4)
                      (Node B.ByteString Tox.Message Bool IPv4)
        dht | isBt      = Left  bt
            | otherwise = Right tox
        cmd0 :: IO () -> IO ()
        cmd0 action = action >> clientSession bt tox signalQuit isBt sock n h
        cmd :: GenericDHT IPv4 (IO ()) -> IO ()
        cmd  (GenericDHT action) = cmd0 $ join $ either (flip runDHT action) (flip runDHT action) dht
        cmd  (BtDHT action)      = cmd0 $ join $ runDHT bt action
        (c,args) = second (dropWhile isSpace) $ break isSpace line
        switchNetwork dest = hPutClient h ("Network: "++if dest then "mainline" else "tox") >> clientSession bt tox signalQuit dest sock n h
    case (c,args) of

        ("bt", _) -> switchNetwork True

        ("tox", _) -> switchNetwork False

        ("quit", _) -> hPutClient h "" >> hClose h

        ("stop", _) -> do hPutClient h "Terminating DHT Daemon."
                          hClose h
                          putMVar signalQuit ()

        ("pid", _) -> cmd0 $ do
                    pid <- getProcessID
                    hPutClient h (show pid)

        -- DHT specific
        ("ls", _) -> cmd $ GenericDHT $ 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)
                            , ("network", if isBt then "mainline" else "tox") ]
                        ]
        ("external-ip", _) -> cmd $ BtDHT $ do
                    ip <- routableAddress
                    return $ do
                    hPutClient h $ maybe "" (takeWhile (/=':') . show) ip

        ("swarms", s) -> cmd $ BtDHT $ 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

        -- bittorrent only
        ("peers", s) -> cmd $ BtDHT $ case readEither s of
                        Right ih -> do
                            ps <- allPeers ih
                            seq ih $ return $ do
                            hPutClient h $ showReport $ map (((,) "") . show . pPrint) ps
                        Left er -> return $ hPutClient h er

#ifdef THREAD_DEBUG
        ("threads", _) -> cmd0 $ do
                    ts <- threadsInformation
                    tm <- getCurrentTime
                    let r = map (\PerThread{..} -> (show lbl,show (diffUTCTime tm startTime))) ts
                    hPutClient h $ showReport r
        ("mem", s) -> cmd0 $ do
                    case s of
                        "gc" -> do hPutClient h "Performing garbage collection..."
                                   performMajorGC
                        "" -> do
                            is_enabled <- getGCStatsEnabled
                            if is_enabled
                             then do
                                GCStats{..} <- getGCStats
                                let r = [ ("bytesAllocated", show bytesAllocated)
                                        , ("numGcs", show numGcs)
                                        , ("maxBytesUsed", show maxBytesUsed)
                                        , ("numByteUsageSamples", show numByteUsageSamples)
                                        , ("cumulativeBytesUsed", show cumulativeBytesUsed)
                                        , ("bytesCopied", show bytesCopied)
                                        , ("currentBytesUsed", show currentBytesUsed)
                                        , ("currentBytesSlop", show currentBytesSlop)
                                        , ("maxBytesSlop", show maxBytesSlop)
                                        , ("peakMegabytesAllocated", show peakMegabytesAllocated)
                                        , ("mutatorCpuSeconds", show mutatorCpuSeconds)
                                        , ("mutatorWallSeconds", show mutatorWallSeconds)
                                        , ("gcCpuSeconds", show gcCpuSeconds)
                                        , ("gcWallSeconds", show gcWallSeconds)
                                        , ("cpuSeconds", show cpuSeconds)
                                        , ("wallSeconds", show wallSeconds)
                                        , ("parTotBytesCopied", show parTotBytesCopied)
                                        , ("parMaxBytesCopied", show parMaxBytesCopied)
                                        ]
                                hPutClient h $ showReport r
                            else hPutClient h "Run with +RTS -T to obtain live memory-usage information."
                        _ -> hPutClient h "error."

#endif
        -- DHT specific
        ("closest", s) -> cmd $ GenericDHT $ do
                        let (ns,hs) = second (dropWhile isSpace) $ break isSpace s
                            parse | null hs = do
                                       ih <- readEither ns
                                       return (8 :: Int, ih)
                                  | otherwise = do
                                       n <- readEither ns
                                       ih <- readEither hs
                                       return (n :: Int, ih)
                        case parse of
                            Right (n,ih) -> do
                                nodeIdType ih
                                tbl <- getTable
                                let nodes = R.kclosest n ih tbl
                                return $ do
                                hPutClient h $ unlines $ map (showEnry . (flip (,) (error "showEnry"))) nodes
                            Left er -> return $ hPutClient h er

        -- DHT specific
        ("ping", s) -> cmd $ GenericDHT $ do
                    case readEither s of
                        Right addr -> do result <- try $ pingQ addr
                                         let rs = either (pure . showQueryFail) reportPong result
                                         return $ do
                                         hPutClient h $ unlines rs
                        Left er -> return $ hPutClient h er

        -- DHT specific
        ("find-nodes", s) -> cmd $ GenericDHT $ do
                    let (hs,as) = second (dropWhile isSpace) $ break isSpace s
                        parse = do ih <- readEither hs
                                   a <- readEither as
                                   -- XXX: using 'InfoHash' only because 'NodeId' currently
                                   -- has no 'Read' instance.
                                   return (ih, a :: NodeAddr IPv4)
                    case parse of
                        Right (ih,a) -> do
                            nodeIdType ih
                            nodeAddrType a
                            proxy <- dhtType
                            let fn = findNodeMessage proxy ih
                            ipType fn
                            result <- try $ queryNode' a fn
                            either (const $ return ()) (\(nid,nf,_) -> nodeIdType nid >> ipType nf) result
                            let rs = either (pure . showQueryFail) reportNodes result
                            return $ do
                            hPutClient h $ unlines rs
                        Left er -> return $ hPutClient h er

        -- bittorrent only
        ("get-peers", s) -> cmd $ BtDHT $ do
                    let (hs,as) = second (dropWhile isSpace) $ break isSpace s
                        parse = do ih <- readEither hs
                                   a <- readEither as
                                   return (ih :: InfoHash, a :: NodeAddr IPv4)
                    case parse of
                        Right (ih,a) -> do
                            result <- try $ queryNode' (a ::NodeAddr IPv4) $ GetPeers ih
                            let rs = either (pure . ( (,) "error" ) . showQueryFail) reportPeers result
                            return $ do
                            hPutClient h $ showReport rs
                        Left er -> return $ hPutClient h er

        -- bittorrent only
        ("search-peers", s) -> cmd $ BtDHT $ do
                    case readEither s of
                        Right ih -> do
                            (tid, s) <- isearch ioGetPeers ih
                            flip fix Set.empty $ \again shown -> do
                                (chk,fin) <- liftIO . atomically $ do
                                    r <- (Set.\\ shown) <$> readTVar (searchResults s)
                                    if not $ Set.null r
                                        then (,) r <$> searchIsFinished s
                                        else searchIsFinished s >>= check >> return (Set.empty,True)
                                let ps = case Set.toList chk of
                                            [] -> ""
                                            _  -> unlines $ map (show . pPrint) $ Set.toList chk
                                if fin then return $ hPutClient h ps
                                       else do
                                            liftIO $ hPutClientChunk h ps
                                            again (shown `Set.union` chk)
                        Left er -> return $ hPutClient h er

        _    -> cmd0 $ hPutClient h "error."

defaultPort = error "TODO defaultPort"

showQueryFail :: QueryFailure -> String
showQueryFail e = show e

consip (ReflectedIP ip) xs = ("(external-ip " ++ show ip ++ ")") : xs

consip' (ReflectedIP ip) xs = ("to", show ip) : xs

reportPong (info,myip) = maybe id consip myip [show $ pPrint info]

reportNodes ::
    ( Kademlia dht
    , Pretty (NodeInfo dht ip ())
    , Pretty (NodeId dht)
    ) => (NodeId dht, NodeFound dht ip, Maybe ReflectedIP) -> [String]
reportNodes (nid,ns,myip) = maybe id consip myip $ show (pPrint nid) : map (show . pPrint) (foundNodes ns)

reportPeers :: (NodeId KMessageOf, GotPeers IPv4, Maybe ReflectedIP) -> [(String,String)]
reportPeers (nid,GotPeers r tok,myip)
    = maybe id consip' myip $ ("from", show (pPrint nid))
     : ("token", show tok)
     : case r of
        Right ps -> map ( ( (,) "peer" ) . show . pPrint ) ps
        Left ns  -> map ( ( (,) "node" ) . show . pPrint ) ns

main :: IO ()
main = do
    args <- getArgs
    p <- case take 2 (dropWhile (/="-p") args) of
                ["-p",port] | not ("-" `isPrefixOf` port) -> return port
                ("-p":_)      -> error "Port not specified! (-p PORT)"
                _             -> defaultPort
    godht p $ \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 (error "todo: tox state") signalQuit True) (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"