Hermes (empty) → 0.0.1

raw patch · 14 files changed

+2229/−0 lines, 14 filesdep +AESdep +RSAdep +SHA2setup-changed

Dependencies added: AES, RSA, SHA2, base, bytestring, cereal, containers, hslogger, monads-tf, network, old-time, random, stm, syb, time, transformers, unamb, yjtools

Files

• COPYING +10/−0
• Hermes.cabal +37/−0
• Network/Hermes.hs +327/−0
• Network/Hermes/Address.hs +45/−0
• Network/Hermes/Core.hs +656/−0
• Network/Hermes/Gossip.hs +362/−0
• Network/Hermes/MChan.hs +69/−0
• Network/Hermes/Misc.hs +88/−0
• Network/Hermes/Net.hsc +116/−0
• Network/Hermes/Protocol.hs +263/−0
• Network/Hermes/RPC.hs +115/−0
• Network/Hermes/Signature.hs +53/−0
• Network/Hermes/Types.hs +84/−0
• Setup.hs +4/−0

@@ -0,0 +1,10 @@+Copyright (c) 2010, University of Tromsø+All rights reserved.++Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.+    * Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.+    * Neither the name of the university of Tromsø nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,37 @@+Name: Hermes+Synopsis: Message-based middleware layer+Description: A middleware providing best-effort unicast, remote procedure calls,+             probabilistic (and slow!) broadcast and automatic membership+             management, meant for small to medium networks+Version: 0.0.1+License: BSD3+Author: Svein Ove Aas <svein.ove@aas.no>+Maintainer: Svein Ove Aas <svein.ove@aas.no>+Copyright: Copyright (c) 2010, University of Tromsø+License-file: COPYING+Stability: experimental+Category: Middleware, Network+Tested-With: GHC == 6.12-rc2+Cabal-Version: >= 1.6+Build-Type: Simple++Library+  Build-Depends: base == 4.*, bytestring, cereal >= 0.2, stm, hslogger >= 1.0.7,+                 old-time, containers, AES >= 0.2.4, SHA2 >= 0.1.1, RSA >= 1.0.2,+                 network, yjtools >= 0.9.7, random, monads-tf == 0.0.*,+                 syb == 0.1.*, unamb >= 0.2.2, time >= 1.1.4, transformers >= 0.1.4.0++  ghc-options:+                 +  Exposed-Modules:+        Network.Hermes,+        Network.Hermes.Protocol,+        Network.Hermes.Types,+        Network.Hermes.Core,+        Network.Hermes.Misc,+        Network.Hermes.Net,+        Network.Hermes.RPC,+        Network.Hermes.Signature,+        Network.Hermes.MChan,+        Network.Hermes.Gossip,+        Network.Hermes.Address

@@ -0,0 +1,327 @@+{-# LANGUAGE ViewPatterns, RecordWildCards, NamedFieldPuns, ScopedTypeVariables #-}+-- | Hermes is a middleware layer providing best-effort unicast,+-- remote procedure calls, probabilistic (and slow!) broadcast and+-- automatic membership management. It is meant for small-to-medium+-- networks; its broadcast gossip protocol, which is used for+-- membership management, will scale poorly to very large ones.+--+-- Hermes uses HsLogger for event logging, using the \"hermes\" namespace.+module Network.Hermes(+  HermesException(..),+  withHermes,+  -- * Authorities +  -- | Unless you turn security off entirely, one Hermes node will not+  -- talk with another unless it trusts the other node. There are two+  -- ways to achieve this: You can specify trusted keys explicitly, or+  -- you can create an signature authority that can create trusted+  -- keys.+  --+  -- This section deals with the latter.+  SignatureRequest, Signature, Authority,+  newAuthority, newSignatureRequest, signRequest, installSignature, addAuthority,+  -- * Context control +  -- | All communication requires a Hermes context. This section deals+  -- with creating, saving and loading them.+  Context, TrustLevel(..), HermesID,+  newContext, newSignedContext, snapshotContext, snapshotContext',+  restoreContext, uuid, setTimeout, setTrustLimit,+  -- * Listeners+  startListener, Address(..),+  -- * Explicit connections+  -- | While Hermes will normally maintain a membership list on its+  -- own, you still need the address of at least one node in order+  -- to download the list.+  connect,+  -- * Messaging+  send, send', recv, recv', acceptType, refuseType,+  -- * Remote Procedure Calls+  call, registerCallback, ProcName,+  -- * Gossip+  writeFactoid, readFactoid, readFactoids, addCallback, setPeriod, TTL,+  -- * Address book+  snapshotAddresses, restoreAddresses,+  -- * Debugging+  setDebug, Priority(..)+  ) where++import Network.Hermes.Signature(newAuthority, SignatureRequest,+                                Authority(..), signRequest)+import Network.Hermes.Core(CoreContext, TrustLevel(..), HermesID, HermesException(..),+                           withHermes)+import Network.Hermes.Types+import Network.Hermes.Protocol(Address(..),decode')+import Network.Hermes.RPC(RPCContext, ProcName)+import Network.Hermes.Gossip(GossipContext,TTL)+import Network.Hermes.Address+import Network.Hermes.Misc+import System.Log.Logger++import qualified Network.Hermes.Core as Core+import qualified Network.Hermes.Signature as Sig+import qualified Network.Hermes.RPC as RPC+import qualified Network.Hermes.Gossip as Gossip++import qualified Data.Map as M+import Control.Concurrent.STM+import Data.ByteString+import Data.Typeable+import Data.Serialize+ +data Context = Context {+  core :: CoreContext+ ,rpc :: RPCContext+ ,gossip :: GossipContext+  }++-- | Creates a signature request for serialization+newSignatureRequest :: Context -> SignatureRequest+newSignatureRequest = Sig.newSignatureRequest . core++installSignature :: Context -> Signature -> IO ()+installSignature Context{core} = Core.setKeySignature core++-- | Adds an authority to the list of trusted authorities+addAuthority :: Context -> Authority -> IO ()+addAuthority Context{core} = Core.addAuthority core . Sig.authorityKey++-- * Context control++-- | Creates a new Hermes context allowing messaging, RPC and gossip,+-- and using automatic address dissemination via the gossip protocol.+--+-- The trust level defaults to Indirect. If you set it to Direct,+-- address dissemination will fail.+--+-- The gossip interval defaults to 300 seconds, call setPeriod to+-- change it.+newContext :: IO Context+newContext = do+  core <- Core.newContext+  rpc <- RPC.newContext core+  gossip <- Gossip.newGossiper core 300+  startAddresser core gossip+  return Context{..}++-- | Creates a pre-signed context. You may snapshot this to restore on+-- another computer, or use on this one.+newSignedContext :: Authority -> IO Context+newSignedContext authority = do+  core <- Sig.newSignedContext authority+  rpc <- RPC.newContext core+  gossip <- Gossip.newGossiper core 300+  return Context{..}+++-- | Snapshots a context for storage  +--+-- Transient state (RPC calls, messages) are discarded, as are+-- connection, listener information and RPC bindings.+snapshotContext' :: Context -> STM ByteString+snapshotContext' Context{..} = do+  coreSnap <- Core.snapshotContext core+  gossipSnap <- Gossip.snapshotGossiper gossip+  return $ encode (coreSnap,gossipSnap)++snapshotContext :: Context -> IO ByteString+snapshotContext = atomically . snapshotContext'++-- | Restores a context from storage+--+-- You will have to reset RPC bindings and listeners.+restoreContext :: ByteString -> IO Context+restoreContext snap = do+  let (coreSnap,gossipSnap) = decode' snap+  core <- atomically $ Core.restoreContext' coreSnap+  rpc <- RPC.newContext core+  gossip <- Gossip.restoreGossiper core gossipSnap+  startAddresser core gossip+  return Context{..}++uuid :: Context -> HermesID+uuid = Core.myHermesID . core++-- | For operations that may block, other than recv, this sets a+-- maximum wait time. Hermes will never block longer than this.+setTimeout :: Context +             -> Double -- ^ Desired timeout, in seconds+             -> IO ()+setTimeout Context{core} = Core.setTimeout core++-- | Sets the number of retries, for retryable operations. Setting+-- this to N and the timeout to T makes Hermes time-out after T/N+-- seconds, then retry.+setRetries :: Context -> Int -> IO ()+setRetries = undefined -- TODO++-- | Set the desired trust limit, which will take effect on next+-- connection+--+-- When connecting peers (either way), a degree of trust is required,+-- or the connection will be rejected.+setTrustLimit :: Context -> TrustLevel -> IO ()+setTrustLimit Context{core} = Core.setTrustLimit core++-- | Set up a listener for incoming connections. These are not stored+-- when snapshotting contexts. This function will return once the port+-- has been bound.+startListener :: Context+                -> Address -- ^ The local address we should bind to+                -> Maybe Address -- ^ An address to provide peers;+                                -- handy for firewalls.+                -> IO ()+startListener Context{core} = Core.startListener core++-- | Connects to a given address without knowing in advance who will+-- be answering. The answerer's HermesID is returned, assuming the+-- connection is properly established.+--+-- Typically used for bootstrapping.+connect :: Context -> Address -> IO HermesID+connect Context{core} = Core.connect core+++-- | Sends a message. The type representation is included, so a+-- modicum of type safety is provided, and recv will only attempt to+-- decode and return a message of the matching (not necessarily+-- correct! Make sure your de/serializers match!) type. There is, of+-- course, a possibility of exceptions if application versions differ.+--+-- You may use send' to provide an arbitrary tag to match on, in which+-- case recv' will only return a message with an equal tag; if you+-- don't, recv will only return messages without tags.+--+-- This function normally blocks until the entire message has been+-- sent, an exception occurs or a timeout is reached. It will retry+-- once if the connection fails within the timeout.+--+-- Unless acceptType or recv has been called in advance, sent messages+-- are thrown away instead of queued. Once either has been, they are+-- indefinitely queued until refuseType is called.+send :: (Serialize msg, Typeable msg)+         => Context -> HermesID -> msg -> IO ()+send Context{core} uuid msg = Core.send core uuid msg++send' :: (Serialize msg, Typeable msg, Serialize tag, Typeable tag)+         => Context -> HermesID -> msg -> tag -> IO ()+send' Context{core} uuid msg tag = Core.send' core uuid msg tag+++-- | Receives a message. This function blocks until a message of the+-- appropriate type has been received, possibly forever. You may use+-- multiple simultaneous recv calls; each message will only be+-- delivered once.+recv :: (Serialize msg, Typeable msg)+        => Context -> IO (HermesID,msg)+recv Context{core} = Core.recv core++recv' :: (Serialize msg, Typeable msg, Serialize tag, Typeable tag)+        => Context -> tag -> IO (HermesID,msg)+recv' Context{core} tag = Core.recv' core tag+++-- | Registers (or replaces) a callback that is to be executed+-- whenever we receive a properly typed call to this name.+--+-- You may register calls with the same name, so long as they have+-- different types.+--+-- If the callback already exists, it is overwritten.+registerCallback :: forall a b. (Serialize a, Serialize b, Typeable a, Typeable b)+                   => Context+                   -> ProcName   -- ^ Callback's name+                   -> (a -> IO b) -- ^ The callback itself+                   -> IO ()+registerCallback Context{rpc} = RPC.registerCallback rpc++-- | Remote procedure call+--+-- In addition to the usual core exceptions, this function may fail in+-- the specific case the the named procedure doesn't exist or has the+-- wrong type, in which case it returns Nothing.+call :: forall a b. (Serialize a, Typeable a, Serialize b, Typeable b) =>+       Context -> HermesID -> ProcName -> a -> IO (Maybe b)+call Context{rpc} = RPC.call rpc++-- | This utility function decides the lowest priority that will be+-- shown. The default is WARNING.+setDebug :: Priority -> IO ()+setDebug level = updateGlobalLogger "hermes" (setLevel level)+++-- | If you wish to queue messages without immediately calling recv, use this.+--+-- acceptType is idempotent.+acceptType :: forall tag msg. (Typeable msg, Serialize tag, Typeable tag)+              => Context+              -> msg -- ^ The message type to accept. Only the type is used, so undefined is fine.+              -> tag+              -> IO ()+acceptType Context{core} msg tag = Core.acceptType core msg tag++-- | If you wish to *stop* queueing messages of a given type, use this.+--+-- Calling refuseType will cause all recv calls to this type/tag+-- combination to throw RecvCancelled.+--+-- refuseType is idempotent.+refuseType :: forall tag msg. (Typeable msg, Serialize tag, Typeable tag)+              => Context+              -> msg -- ^ The message type to accept. Only the type is used, so undefined is fine.+              -> tag+              -> IO ()+refuseType Context{core} msg tag = Core.refuseType core msg tag+++-- | Insert a factoid in the gossip network. This will immediately+-- trigger a limited gossip exchange, hopefully spreading it to a+-- large fraction of the network.+--+-- Factoids are keyed by their type, source, and the type and+-- serialized value of an arbitrary tag. They can be replaced by+-- re-inserting later, and optionally expire after a timeout.+-- +-- Don't rely on the timeout, though. It's for garbage collection, and+-- is not required to be exact.+writeFactoid :: forall factoid tag. (Typeable factoid, Serialize factoid, Typeable tag, Serialize tag)+                => Context -> factoid -> tag+                -> Maybe TTL -- ^ The timeout, in seconds+                -> IO ()+writeFactoid Context{gossip} factoid tag ttl = Gossip.writeFactoid gossip factoid tag ttl++-- | Read a factoid, assuming it exists.+readFactoid :: forall factoid tag. (Typeable factoid, Serialize factoid, Typeable tag, Serialize tag)+               => Context -> tag -> HermesID -> IO (Maybe factoid)+readFactoid Context{gossip} tag source = Gossip.readFactoid gossip tag source++-- | Read all factoids with an appropriate type and tag. Useful if you+-- don't know what source to expect.+readFactoids :: forall factoid tag. (Typeable factoid, Serialize factoid, Typeable tag, Serialize tag)+                => Context -> tag -> IO [(HermesID,factoid)]+readFactoids Context{gossip} tag = Gossip.readFactoids gossip tag++-- | Add a callback to be called every time a type-matching factoid is+-- inserted or updated. It will not be called for writeFactoid calls.+addCallback :: forall msg tag. (Serialize tag, Typeable tag, Serialize msg, Typeable msg)+               => Context -> (HermesID -> tag -> msg -> IO ()) -> IO ()+addCallback Context{gossip} function = Gossip.addCallback gossip function++-- | Set the period for the periodic gossiper. It will take effect+-- after the next periodic gossip.+setPeriod :: Context+             -> Double -- ^ The period, in seconds+             -> IO ()+setPeriod Context{gossip} period = Gossip.setPeriod gossip period+++-- | The address snapshot contains address information for every node+-- we know of, which can be restored into another node to bootstrap+-- it.+snapshotAddresses :: Context -> STM ByteString+snapshotAddresses = fmap encode . readTVar . peerAddress . core++-- | Restore an address snapshot to bootstrap your node.+--+-- Returns Nothing on success, otherwise a parse error.+restoreAddresses :: Context -> ByteString -> STM (Maybe String)+restoreAddresses Context{core} (decode -> info) = either (return . Just) mergeInfo info+  where mergeInfo addresses = modifyTVar (peerAddress core) (M.union addresses) >> return Nothing

@@ -0,0 +1,45 @@+{-# OPTIONS_HADDOCK hide #-}+{-# LANGUAGE DeriveDataTypeable, ViewPatterns, GeneralizedNewtypeDeriving, RecordWildCards #-}+-- | Gossiping about addresses+module Network.Hermes.Address(startAddresser) where++import Network.Hermes.Protocol+import Network.Hermes.Misc+import Network.Hermes.Types+import Network.Hermes.Gossip+import Network.Hermes.Core++import System.Log.Logger+import Data.Typeable+import Data.Serialize+import Control.Concurrent+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Data.ByteString(ByteString)+import Control.Applicative+import Control.Monad+import Data.Maybe+import Data.Map(Map)+import qualified Data.Map as M+import qualified Data.Set as S++-- | We gossip about addresses using () for a tag, and.. well, this+-- for a value.+newtype AGossip = AGossip { agAddress :: Address }+                deriving(Serialize,Show,Typeable)++-- | Start the address gossiper. As there are no controls, there is no+-- context returned.+startAddresser :: CoreContext -> GossipContext -> IO ()+startAddresser core gossip = (fmap (const ()) . forkIO) $ do+  mbox <- newEmptyMVar+  -- A listener to check for and insert new addresses+  listenTVar (listeners core) $ \ls -> +    unless (S.null ls) $ do+      let address = remoteAddress $ S.findMin ls+      noticeM "hermes.address" $ "Updating our address to " ++ show address+      writeFactoid gossip (AGossip address) () Nothing+  -- And one to handle incoming address updates+  addCallback gossip $ \source () AGossip{..} -> do+    noticeM "hermes.address" $ "Updating address for " ++ show source ++ " to " ++ show agAddress+    atomically $ modifyTVar (peerAddress core) (M.insert source agAddress)  

@@ -0,0 +1,656 @@+{-# OPTIONS_HADDOCK hide #-}+{-# LANGUAGE ViewPatterns, PackageImports, RecordWildCards,+    NamedFieldPuns, ScopedTypeVariables, BangPatterns, DeriveDataTypeable #-}+module Network.Hermes.Core(+  withHermes, CoreContext, TrustLevel(..), HermesID, HermesException(..)+  -- * Context management+  ,newContext, restoreContext, restoreContext', snapshotContext, addAuthority, setKeySignature+  ,myHermesID, setTimeout, timeout, setTrustLimit, snapshotContext'+  -- * Listeners+  ,startListener+   -- * Peer management+  ,connect, setHermesID+  -- * Communication+  ,send, send', recv, recv', NoTag(..), acceptType, refuseType+) where++import Prelude hiding(catch)++import Control.Arrow+import Control.Applicative+import Control.Monad+import Control.Monad.Tools+import "monads-tf" Control.Monad.State+import Control.Concurrent.STM+import Control.Exception(throwIO, throw, onException, block, unblock, catch, IOException(..))+import Data.Typeable++import System.Log.Logger++import Data.Maybe+import Data.Word+import qualified Data.Set as S+import Data.Map(Map)+import qualified Data.Map as M+import qualified Network+import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as BL+import System.IO(Handle, hFlush, hClose)+import qualified System.Timeout++import qualified Data.Serialize+import Data.Serialize(encode,decode,Serialize)+import Data.Serialize.Put+import Data.Serialize.Get+import Network.Hermes.Protocol+import Network.Hermes.Types+import Network.Hermes.Misc+import Network.Hermes.MChan+import qualified Network.Hermes.Net as N+import Codec.Crypto.AES.Random+import Codec.Crypto.RSA as RSA+import Codec.Crypto.AES.IO as AES+import Codec.Digest.SHA++-- * Misc++hashKey :: PublicKey -> HermesID+hashKey = byteStringToInteger . hash SHA256 . encode++-- | All use of hermes must be wrapped with this (on windows)+withHermes :: IO a -> IO a+withHermes = Network.withSocketsDo++-- | Like Data.ByteString.hGet, except it throws EOF instead of returning partial data+hGet :: Handle -> Int -> IO B.ByteString+hGet h i = do+  bs <- B.hGet h i+  unless (B.length bs == i) $ throwIO EOF+  return bs+++-- * Sending and receiving++-- Wire messages are always strict bytestrings. The API uses Data.Serialize+-- to encode arbitrary values as bytestrings, and tags them with their+-- type.+--+-- All integers are unsigned and little-endian.+--+-- Each clause (marked with a - ) is separately encrypted+--+-- Wire message format:+--+-- - Header:+-- * Length of the message as a 32-bit integer+-- * Length of the tag as a 32-bit integer+-- * Message type index as a 32-bit integer+-- * Tag type index as a 32-bit integer+--+-- - 16-byte truncated HMAC of the header+--+-- - 32-byte HMAC of the tag and message+-- - Variable length tag+-- - The message itself++cryptSend :: Connection+             -> Int          -- ^ The tag's type index+             -> B.ByteString -- ^ The tag+             -> Int          -- ^ The message's type index+             -> B.ByteString -- ^ The message+             -> IO ()+cryptSend Connection{..} tagIndex tag messageIndex message = do+  let header      = runPut $ do+        putWord32le $ fromIntegral $ B.length message+        putWord32le $ fromIntegral $ B.length tag+        putWord32le $ fromIntegral messageIndex+        putWord32le $ fromIntegral tagIndex+      headerHMAC  = B.take 16 $ hmac SHA256 aesKey header+      payloadHMAC = hmac SHA256 aesKey (BL.fromChunks [tag,message])+  debugM "hermes.core" $ "  message index: " ++ show messageIndex+  debugM "hermes.core" $ "  tag index    : " ++ show tagIndex+  debugM "hermes.core" $ "  header HMAC  : " ++ showBSasHex headerHMAC+  debugM "hermes.core" $ "  payload HMAC : " ++ showBSasHex payloadHMAC+  debugM "hermes.core" $ "  tag          : " ++ showBSasHex tag+  debugM "hermes.core" $ "  message      : " ++ showBSasHex message+  forM_ [header,headerHMAC,payloadHMAC,tag,message] (\bs -> AES.crypt aesctx bs >>= B.hPut handle)+  hFlush handle++cryptRecv :: Handle+             -> B.ByteString -- ^ The HMAC secret key+             -> AESCtx+             -> IO (Int,B.ByteString,Int,B.ByteString) -- ^ (Tag type index, tag, message index, message)+cryptRecv h key ctx = do+  header <- AES.crypt ctx =<< hGet h 16+  expectedHeaderHMAC <- AES.crypt ctx =<< hGet h 16+  let (messageLength,tagLength,messageIndex,tagIndex) =+        flip runGet' header $ liftM4 (,,,) getWord32le getWord32le getWord32le getWord32le+      headerHMAC = B.take 16 $ hmac SHA256 key header+  debugM "hermes.core" $ "Receiving message, type index " ++ show (messageIndex,tagIndex)+  debugM "hermes.core" $ "  header HMAC : " ++ showBSasHex headerHMAC+  unless (headerHMAC == expectedHeaderHMAC) $ throwIO MessageError+  expectedPayloadHMAC <- AES.crypt ctx =<< hGet h 32+  tag <- AES.crypt ctx =<< hGet h (fromIntegral tagLength)+  message <- AES.crypt ctx =<< hGet h (fromIntegral messageLength)+  let payloadHMAC = hmac SHA256 key (BL.fromChunks [tag,message])+  debugM "hermes.core" $ "  payload HMAC: " ++ showBSasHex payloadHMAC+  unless (payloadHMAC == expectedPayloadHMAC) $ throwIO MessageError+  debugM "hermes.core" $ "  tag         : " ++ showBSasHex tag+  debugM "hermes.core" $ "  message     : " ++ showBSasHex message+  return (fromIntegral tagIndex, tag, fromIntegral messageIndex, message)++-- | Optimizes sends by converting textual type representation to+-- indexed.+baseSend :: CoreContext+            -> HermesID+            -> Type         -- ^ The tag's type+            -> B.ByteString -- ^ The tag+            -> Type         -- ^ The message's type+            -> B.ByteString -- ^ The message+            -> IO ()+baseSend ctx uuid tagType tag msgType msg = trySend True+  where+    trySend firstTry = flip catch (handler firstTry) $ do+      -- OPTIMIZE: Use Data.Typeable's tag int+      -- OPTIMIZE: Don't encode tag/message if they're already (strict) bytestrings+      infoM "hermes.core" $ "Sending message, type " ++ show (tagType,msgType)+      if uuid == myHermesID ctx+        then do insertMessage ctx msgType tagType tag uuid msg+        else do withConnection ctx uuid $ \conn -> do+                  let fetchIndex typeString = do+                        (typeIndex,sendType) <- atomically $ do+                          maybeIndex <- M.lookup typeString <$> readTVar (typeMap conn)+                          case maybeIndex of+                            Just index -> return (index,False)+                            Nothing -> do+                              index <- succ <$> readTVar (typeMax conn)+                              modifyTVar (typeMap conn) (M.insert typeString index)+                              writeTVar (typeMax conn) index+                              return (index,True)+                        when sendType $ cryptSend conn 0 B.empty 0 (encode (typeIndex,typeString))+                        return typeIndex+                      fetchIndex :: Type -> IO Int+                  tagIndex <- fetchIndex tagType+                  messageIndex <- fetchIndex msgType+                  cryptSend conn tagIndex tag messageIndex msg+                  atomically $ modifyTVar (peerFailures ctx) (M.insert uuid 0)+    handler :: Bool -> IOException -> IO ()+    handler firstTry e = do+      noticeM "hermes.core" $ "IO error while sending: " ++ show e ++ if firstTry then ", retrying" else ""+      killConnection ctx uuid+      if firstTry then trySend False else return ()+      +-- | Kills the connection to a peer, increasing its failure count in+-- the process+killConnection :: CoreContext -> HermesID -> IO ()+killConnection ctx uuid = do+  h <- atomically $ do+    mvar <- M.lookup uuid <$> readTVar (peerConnections ctx)+    var <- case mvar of+      Nothing -> return Nothing+      Just v -> Just . handle <$> takeTMVar v+    modifyTVar (peerConnections ctx) (M.delete uuid)+    return $ var+  maybe (return ()) hClose h++-- | Sends a message. The type representation is included, so+-- a modicum of type safety is provided, and recv will only attempt to+-- decode and return a message of the matching (not necessarily+-- correct!) type. There is, of course, a possibility of parse errors+-- if application versions differ.+--+-- This function blocks until the entire message has been sent, or a+-- timeout is reached.  It will retry once if the connection fails.+--+-- This function is equivalent to using NoTag for send's tag.+send :: (Serialize msg, Typeable msg) => CoreContext -> HermesID -> msg -> IO ()+send cc uuid msg = send' cc uuid msg NoTag++data NoTag = NoTag+           deriving(Typeable)++instance Data.Serialize.Serialize NoTag where+         get = return NoTag +         put _ = return ()++-- | Alternately, you may provide an arbitrary tag to match on, in+-- which case recv' will only return a message with an equal tag.+send' :: (Serialize msg, Typeable msg, Serialize tag, Typeable tag)+         => CoreContext -> HermesID -> msg -> tag -> IO ()+send' ctx uuid msg tag = baseSend ctx uuid (showType tag) (encode tag) (showType msg) (encode msg)++-- | Receives a message. This function blocks until a message of the+-- appropriate type has been received.+--+-- Multiple calls to recv are allowed; messages are only returned once.+recv :: forall msg. (Serialize msg, Typeable msg) => CoreContext -> IO (HermesID,msg)+recv ctx = recv' ctx NoTag++-- | You may also specify an arbitrary tag to match on, in which case+-- we'll block until we see a message of the appropriate type, with a+-- tag with the correct type and value.+--+-- Messages with a type/tag that has not been requested via+-- recv' or acceptType are automatically dropped.+--+-- Once recv has been called once, any further messages of the same+-- type are indefinitely queued.+recv' :: forall msg tag. (Serialize msg, Typeable msg, Serialize tag, Typeable tag)+         => CoreContext+         -> tag+         -> IO (HermesID,msg)+recv' ctx tag = do+  let tagType = showType tag+      messageType = showType (undefined :: msg)+      key = (messageType,tagType,encode tag)+  infoM "hermes.core" $ "Requesting message of type " ++ show (messageType,showType tag,encode tag)+  acceptType ctx (undefined :: msg) tag+  msg <- atomically $ readMChan (messageBox ctx) key+  case msg of+    Nothing -> throwIO RecvCancelled+    Just msg' -> do+      infoM "hermes.core" $ "Message of type " ++ show (tagType,messageType) ++ " returned"+      return $ second decode' msg'++-- | If you wish to queue messages without immediately calling recv, use this.+--+-- acceptType is idempotent.+acceptType :: forall tag msg. (Typeable msg, Serialize tag, Typeable tag)+              => CoreContext+              -> msg -- ^ The message type to accept. Only the type is used, so undefined is fine.+              -> tag+              -> IO ()+acceptType CoreContext{messageBox} (showType -> messageType) tag = do+  let key = (messageType,showType tag,encode tag)+  debugM "hermes.core" $ "Accepting key: " ++ show key+  atomically $ ensureMChan messageBox key++-- | If you wish to *stop* queueing messages of a given type, use this.+--+-- Calling refuseType will cause all recv calls to this type/tag+-- combination to throw RecvCancelled.+--+-- refuseType is idempotent.+refuseType :: forall tag msg. (Typeable msg, Serialize tag, Typeable tag)+              => CoreContext+              -> msg -- ^ The message type to accept. Only the type is used, so undefined is fine.+              -> tag+              -> IO ()+refuseType CoreContext{messageBox} (showType -> messageType) tag = do+  let key = (messageType,showType tag,encode tag)+  debugM "hermes.core" $ "Refusing key: " ++ show key+  atomically $ deleteMChan messageBox key+++-- * Context management++restoreContext' :: CoreContextSnapshot -> STM CoreContext+restoreContext' CoreContextSnapshot{..} = do+  let myKey        = myKeySnap+      myPrivateKey = myPrivateKeySnap+      myHermesID       = myHermesIDSnap+  myKeySignature  <- newTVar myKeySignatureSnap+  authorities     <- newTVar authoritiesSnap+  listeners       <- newTVar S.empty+  listenerKillers <- newTVar M.empty+  peerAddress     <- newTVar peerAddressSnap+  peerKeys        <- newTVar peerKeysSnap+  peerConnections <- newTVar M.empty+  peerFailures    <- newTVar peerFailuresSnap+  trustLimit      <- newTVar trustLimitSnap+  messageBox      <- newMChan+  timeLimit       <- newTVar timeLimitSnap+  return $ CoreContext {..}++-- | Restores a context from snapshot+restoreContext :: B.ByteString -> IO CoreContext+restoreContext (decode' -> snapshot) = do+  infoM "hermes.core" "Restoring context from snapshot"+  atomically $ restoreContext' snapshot++++-- | Snapshots a context for storage  +--+-- Messages that have been received but not yet processed are+-- discarded, as are the listeners, and obviously connections. Other+-- details - keys, authorities, signatures, etc. - are all saved.+snapshotContext :: CoreContext -> STM CoreContextSnapshot+snapshotContext ctx = do+ let myKeySnap        = myKey ctx+     myPrivateKeySnap = myPrivateKey ctx+     myHermesIDSnap       = myHermesID ctx+ myKeySignatureSnap <- readTVar (myKeySignature ctx)+ authoritiesSnap    <- readTVar (authorities ctx)+ peerAddressSnap    <- readTVar (peerAddress ctx)+ peerKeysSnap       <- readTVar (peerKeys ctx)+ trustLimitSnap     <- readTVar (trustLimit ctx)+ timeLimitSnap      <- readTVar (timeLimit ctx)+ peerFailuresSnap   <- readTVar (peerFailures ctx)+ return $ CoreContextSnapshot {..}++snapshotContext' :: CoreContext -> IO B.ByteString+snapshotContext' ctx = encode <$> atomically (snapshotContext ctx)++-- | Set key/addressing information for some HermesID. You should probably+-- | not call this function directly.+setHermesID :: CoreContext -> HermesID -> Maybe Address -> Maybe PeerKey -> IO ()+setHermesID ctx uuid address key = atomically $ do+  when (isJust address) $ modifyTVar (peerAddress ctx) (M.insert uuid (fromJust address))+  when (isJust key) $ modifyTVar (peerKeys ctx) (M.insert uuid (fromJust key))++-- | Set the desired trust limit, which will take effect on next connection.+setTrustLimit :: CoreContext -> TrustLevel -> IO ()+setTrustLimit ctx = atomically . writeTVar (trustLimit ctx)++-- | Set the desired time-limit for all operations. Defaults to 30 seconds.+setTimeout :: CoreContext -> Double -> IO ()+setTimeout ctx = atomically . writeTVar (timeLimit ctx)++-- | As for System.Timeout.timeout, but reads the context for the+-- | timeout and throws Timeout instead of returning Nothing+timeout :: CoreContext -> IO a -> IO a+timeout ctx act = do+  limit <- atomically $ readTVar (timeLimit ctx)+  ret <- System.Timeout.timeout (round $ limit * 1000000) act+  maybe (throwIO Timeout) return ret++-- | Add an authority key that we're supposed to trust+addAuthority :: CoreContext -> PublicKey -> IO ()+addAuthority ctx authority = atomically $ modifyTVar (authorities ctx) (authority :)++-- | Set our key signature+setKeySignature :: CoreContext -> Signature -> IO ()+setKeySignature ctx sig = atomically $ writeTVar (myKeySignature ctx) (Just sig)++-- | Creates an empty context, with a new, randomly generated key.+-- The default trust level is Indirect, but no authorities are set.+newContext :: IO CoreContext+newContext = do+  infoM "hermes.core" "Creating new context"+  aesGen <- newAESGen+  let (myKey,myPrivateKey,_) = RSA.generateKeyPair aesGen rsaKeySize+      myHermesID                 = hashKey myKey+  myKeySignature  <- newTVarIO Nothing+  authorities     <- newTVarIO []+  listeners       <- newTVarIO S.empty+  listenerKillers <- newTVarIO M.empty+  peerAddress     <- newTVarIO M.empty+  peerKeys        <- newTVarIO M.empty+  peerConnections <- newTVarIO M.empty+  peerFailures    <- newTVarIO M.empty+  trustLimit      <- newTVarIO Indirect+  timeLimit       <- newTVarIO 30+  messageBox      <- newMChanIO+  return CoreContext {..}++-- | Connects to a given address without first knowing who will be+-- | answering. The answerer's HermesID is returned, assuming the+-- | connection is properly established.+--+-- Typically used for bootstrapping.+connect :: CoreContext -> Address -> IO HermesID+connect ctx address = block $ do+  infoM "hermes.core" $ "Connecting to " ++ show address+  (conn, uuid) <- unblock $ negotiate ctx address Nothing+  closeIt <- atomically $ do+    modifyTVar (peerAddress ctx) (M.insert uuid address)+    -- We store this connection iff we don't already have one to this HermesID+    ifM (M.member uuid <$> readTVar (peerConnections ctx))+      (return True)+      (do box <- newTMVar conn+          modifyTVar (peerConnections ctx) (M.insert uuid box)+          return False)+  when closeIt (hClose (handle conn))+  return uuid+    ++-- * Connection negotiation & listeners++++startListener :: CoreContext+                -> Address -- ^ The local address we bind to on this system+                -> Maybe Address -- ^ If present, the address other systems see+                -> IO ()+startListener ctx localAddress remoteAddressMaybe = do+  infoM "hermes.core" $ "Listener started on address " ++ show localAddress+  let remoteAddress = maybe localAddress id remoteAddressMaybe+      address = ListenerAddress {..}+  ok <- atomically $ do+    set <- readTVar (listeners ctx)+    if S.member address set+      then return False+      else writeTVar (listeners ctx) (S.insert address set) >> return True+  unless ok $ throwM ListenerAlreadyExists+  killer <- N.streamServer localAddress (handleConnection ctx)+  atomically $ do+    False <- M.member address <$> readTVar (listenerKillers ctx)+    modifyTVar (listenerKillers ctx) (M.insert address killer)++-- | Handle an incoming connection+handleConnection :: CoreContext -> Handle -> Address -> IO ()+handleConnection ctx h address = traplogging "hermes.core" CRITICAL "trap: handleConnection" $ do+  -- Handle exceptions+  flip catch (\e -> case e of+                 EOF -> infoM "hermes.core.handleConnection" $ "EOF on " ++ show address+             ) $ do+    infoM "hermes.core" $ "Incoming connection from " ++ show address+    -- Exchange protocol version+    exchangeVersions h+    -- Exchange HermesIDs. We go last.+    AHermesID theirHermesID <- rawRecv h+    rawSend h (AHermesID $ myHermesID ctx)+    -- Exchange keys, if required. We go last.+    answerKeyQuery ctx h+    -- Then we get to ask client+    -- for a key.  If we've gotten this far, then both sides have the+    -- other's key.  The connecting client then creates a session key,+    -- and seals it for the server's (i.e. our) eyes only.  Note that+    -- there's no provision in the protocol for informing the other side+    -- of errors. That's probably a TODO, but we have to handle+    -- disconnections at any point regardless; doing so would be purely+    -- for courtesy.+    theirKey <- ensureKey ctx h theirHermesID+    -- We send the client a challenge that must be included with the+    -- session key, to prevent replay attacks+    challenge <- prandBytes 16+    debugM "hermes.core" $ "Sending challenge: " ++ showBSasHex challenge+    do gen <- newAESGen+       rawSend h $ AChallenge $ fst $ rsaEncrypt gen (peerKey theirKey) challenge+    ASessionSetup setupBS <- rawRecv h+    let setupMsg@SessionSetup{..} = decode' $ rsaDecrypt (myPrivateKey ctx) setupBS+    infoM "hermes.core" $ "Receiving session setup: " ++ show setupMsg+    unless (challenge == setupChallenge) (throwM $ AuthError "Challenge mismatch")+    -- Well, they must be who they say they are.+    -- Opportunistically (re)insert the client's address in our address database+    when (isJust clientAddress) $+      atomically $ modifyTVar (peerAddress ctx) (M.insert theirHermesID (fromJust clientAddress))+    -- Create the session decryption context+    aesctx <- AES.newCtx AES.CTR setupKey setupIV AES.Decrypt+    -- And a map for type indexes+    indexMap <- newTVarIO (M.empty :: Map Int Type)+    -- Enter a loop, receiving and forwarding messages.+    infoM "hermes.core" $ "Connection setup for " ++ show address ++ " complete"+    forever $ do+      -- Receive a new message.+      (tagIndex,tag,messageIndex,message) <- cryptRecv h setupKey aesctx+      case messageIndex of+        0 -> do+          let decoded = decode' message+          infoM "hermes.core" $ "New type registered: " ++ show decoded+          atomically $ modifyTVar indexMap (uncurry M.insert decoded)+        _ -> do+          let getType typeIndex = atomically $ fromJust . M.lookup typeIndex <$> readTVar indexMap+          tagType <- getType tagIndex+          messageType <- getType messageIndex+          infoM "hermes.core" $ "Received message of type " ++ show (tagType,messageType)+          insertMessage ctx messageType tagType tag theirHermesID message+          +insertMessage ctx messageType tagType tag theirHermesID message = do+  accepted <- atomically $ writeMChan (messageBox ctx) (messageType,tagType,tag) (theirHermesID,message)+  unless accepted $ do+    warningM "hermes.core" $ "Message discarded: type, tag type, tag value: " +++      show (messageType,tagType,tag)+    -- Return the reject message, assuming this isn't a reject message - no loops!+    unless (messageType == showType RejectedMessage) $+      send' ctx theirHermesID RejectedMessage (tagType,tag)++-- | Negotiate a connection, client side. Returns the HermesID of the+-- | remote host, along with an connection/encryption context.+negotiate :: CoreContext+            -> Address -- ^ Address to connect to+            -> Maybe HermesID -- ^ Expected HermesID of the remote host, if any+            -> IO (Connection,HermesID) -- ^ (Connection,actual HermesID)+negotiate ctx address expectedHermesID = block $ do+  infoM "hermes.core" $ "Negotiating connection to " ++ show address ++ ", HermesID " ++ show expectedHermesID+  h <- N.connectStream address+  flip onException (hClose h) $ unblock $ do+    -- Exchange protocol version+    exchangeVersions h+    -- Exchange HermesIDs. We go first.+    rawSend h $ AHermesID $ myHermesID ctx+    AHermesID theirHermesID <- rawRecv h+    unless (maybe True (== theirHermesID) expectedHermesID) $ do+      -- Address is apparently wrong. Delete it from our database.+      -- But check to make sure it hasn't been updated in the meantime+      atomically $ whenM ((== Just address) . M.lookup (fromJust expectedHermesID)+                          <$> readTVar (peerAddress ctx))+        (modifyTVar (peerAddress ctx) (M.delete theirHermesID))+      throwM $ AddressUnknown theirHermesID+    -- Exchange keys, if required. Client (that is, us) goes first.+    theirKey <- ensureKey ctx h theirHermesID+    answerKeyQuery ctx h+    -- Both sides have the other's key - see handleConnection for details.+    -- Generate the session key and send it, with the challenge+    AChallenge challengeEncrypted <- rawRecv h+    let setupChallenge = rsaDecrypt (myPrivateKey ctx) challengeEncrypted+    setupKey <- prandBytes (aesKeySize `div` 8)+    setupIV <- prandBytes 16+    -- TODO: Do something smarter than picking one listener at random, and abstract+    clientAddress <- atomically $ listToMaybe . map remoteAddress . S.toList <$> readTVar (listeners ctx)+    let setupMsg = SessionSetup {..}+    debugM "hermes.core" $ "Sending session setup: " ++ show setupMsg+    do g <- newAESGen+       rawSend h $ ASessionSetup $ fst $ rsaEncrypt g (peerKey theirKey) (encode setupMsg)+    -- Finally, create and return the session context+    aesctx <- AES.newCtx AES.CTR setupKey setupIV AES.Encrypt+    infoM "hermes.core" $ "Negotiation complete"+    typeMap <- newTVarIO M.empty+    typeMax <- newTVarIO 0+    return (Connection { aesctx = aesctx, handle = h, aesKey = setupKey, typeMap, typeMax }, theirHermesID)++-- | Locks a peer connection, negotiating a connection to said peer if necessary+withConnection :: CoreContext -> HermesID -> (Connection -> IO a) -> IO a+withConnection ctx theirHermesID act = do+  address <- atomically $ maybe (throw $ AddressUnknown theirHermesID) id . M.lookup theirHermesID <$> readTVar (peerAddress ctx)+  withTMVar+    (peerConnections ctx)+    theirHermesID+    (fst <$> negotiate ctx address (Just theirHermesID))+    (get >>= liftIO . act)++-- | Looks up a TMVar in a TVar Map. If doesn't exist, it is created+-- | and some code is executed to fill it. Once it does exist, another+-- | function is called with it held. Only after this function is+-- | complete is the value put in the TMVar. If an exception occurs,+-- | the TMVar is deleted from the map again.+withTMVar :: Ord a =>+            TVar (Map a (TMVar b)) -- ^ The map+            -> a -- ^ A key for the map+            -> IO b -- ^ Function used to fill the TMVar+            -> StateT b IO r -- ^ Code to run once the TMVar is full+            -> IO r+withTMVar tvar key filler act = block $ do+  (needFill,var) <- atomically $ do+    exists <- M.member key <$> readTVar tvar+    if exists+      then do var <- fromJust . M.lookup key <$> readTVar tvar+              return (False,var)+      else do placeholder <- newEmptyTMVar+              modifyTVar tvar (M.insert key placeholder)+              return (True,placeholder)+  if needFill+    then flip onException (atomically $ modifyTVar tvar (M.delete key)) $ unblock $ do+    fill <- filler+    (ret, fill') <- runStateT act fill+    atomically $ putTMVar var fill'+    return ret+    else unblock $ runTMVar var act++-- | Checks if we have a trusted key for the given HermesID, and asks the+-- | given handle to provide one if we don't - then checks it.+ensureKey :: MonadIO m => CoreContext -> Handle -> HermesID -> m PeerKey+ensureKey ctx h uuid = do+  maybeTheirKey <- getKey ctx uuid -- See if we need a new key+  if isNothing maybeTheirKey+    then do +    theirKey <- requestKey ctx h uuid+    limit <- liftIO $ atomically $ readTVar (trustLimit ctx)+    when (trust theirKey < limit)+      (throwM $ AuthError $ "Key insufficiently trusted: " ++ show (trust theirKey))+    liftIO $ atomically $ modifyTVar (peerKeys ctx) (M.insert uuid theirKey)+    return theirKey+    else do+    rawSend h $ AKeyQuery KeyOK+    return (fromJust maybeTheirKey)+++-- | Fetches and verifies a remote key. Does not test trust level+-- | against the Context limit.+requestKey :: MonadIO m => CoreContext -> Handle -> HermesID -> m PeerKey+requestKey ctx h uuid = liftIO $ do+  rawSend h $ AKeyQuery RequestKey+  AKeyReply reply <- rawRecv h+  let key = keyReplyKey reply+      keyHermesID = hashKey key+      keyBS = encode key+  unless (keyHermesID == uuid) (throwM $ AuthError "key HermesID mismatch")+  case keyReplySig reply of+    Nothing -> return $ PeerKey { peerKey = key, trust = None, signature = Nothing }+    Just sig -> do+      authorityKeys <- atomically $ readTVar (authorities ctx)+      let ok = or $ map (\authority -> rsaVerify authority keyBS sig) authorityKeys+      unless ok (throwM $ AuthError "signature not verifiable")+      return $ PeerKey { peerKey = key, trust = Indirect, signature = Just sig }++-- | Returns Nothing if the key doesn't exist, or if it is insufficiently trusted.+getKey :: MonadIO m => CoreContext -> HermesID -> m (Maybe PeerKey)+getKey ctx uuid = liftIO $ atomically $ do+  maybeKey <- M.lookup uuid <$> readTVar (peerKeys ctx)+  limit <- readTVar (trustLimit ctx)+  case maybeKey of+    Just key -> return $ if trust key < limit then Nothing else Just key+    Nothing -> return Nothing++answerKeyQuery :: MonadIO m => CoreContext -> Handle -> m ()+answerKeyQuery ctx h = answerKeyQuery' =<< rawRecv h+  where+    answerKeyQuery' (AKeyQuery KeyOK) = return ()+    answerKeyQuery' (AKeyQuery RequestKey) = do+      let keyReplyKey = myKey ctx+      keyReplySig <- liftIO $ atomically $ readTVar (myKeySignature ctx)+      rawSend h $ AKeyReply $ KeyReply{..}+    answerKeyQuery' _ = throwM $ AuthError "answerKeyQuery: Unexpected reply"+++exchangeVersions :: MonadIO m => Handle -> m ()+exchangeVersions h = liftIO $ do+  B.hPut h magicString+  B.hPut h (encode protocolVersion)+  hFlush h+  theirString <- hGet h (B.length magicString)+  unless (magicString == theirString) (throwIO WrongProtocol)+  theirVersion <- decode' <$> hGet h 4+  unless (theirVersion == protocolVersion) (throwIO $ ProtocolVersionMismatch protocolVersion theirVersion)++rawSend :: (MonadIO m) => Handle -> AnyMessage -> m ()+rawSend h (encode -> msg) = liftIO $ do+  B.hPut h $ encode (fromIntegral $ B.length msg :: Word32)+  B.hPut h msg+  hFlush h++rawRecv :: (MonadIO m) => Handle -> m AnyMessage+rawRecv h = liftIO $ do+  size <- decode' <$> hGet h 4 :: IO Word32+  decode' <$> hGet h (fromIntegral size)

@@ -0,0 +1,362 @@+{-# OPTIONS_HADDOCK hide #-}+{-# LANGUAGE RecordWildCards, ScopedTypeVariables, PackageImports, GADTs,+             DeriveDataTypeable, NamedFieldPuns, ViewPatterns, TupleSections, DoRec #-}+module Network.Hermes.Gossip(+  GossipContext, TTL,+  writeFactoid,+  readFactoid,+  readFactoids,+  addCallback,+  newGossiper,+  setPeriod,+  snapshotGossiper,+  restoreGossiper+  ) where++import Control.Applicative+import Control.Monad+import Control.Monad.Trans.Maybe+import Control.Concurrent+import Control.Concurrent.STM+import Control.Concurrent.MVar+ +import Codec.Crypto.RSA hiding(sign,verify)+import qualified Codec.Crypto.RSA as RSA+import Data.ByteString(ByteString)+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as BSL+import Data.Time+import Data.Map(Map)+import qualified Data.Map as M+import Data.Typeable+import Data.Serialize+import Data.Maybe+import System.Log.Logger+import System.Random++import Network.Hermes.Core+import Network.Hermes.Misc+import Network.Hermes.Protocol+import Network.Hermes.Types(CoreContext(..),PeerKey(peerKey))++-- RSA adaption++sign :: PrivateKey -> ByteString -> ByteString+sign key bs = BS.concat $ BSL.toChunks $ RSA.sign key (BSL.fromChunks [bs])++verify :: PublicKey -> ByteString -> ByteString -> Bool+verify key bs sig = RSA.verify key (BSL.fromChunks [bs]) (BSL.fromChunks [sig])+++-- * State++data GossipContext = GossipContext {+  core :: CoreContext+  ,factoids :: TVar (Map FactoidKey (Map HermesID Factoid))+  ,factoidTTLs :: TVar (Map TTL FactoidKey)+  ,gossipInterval :: TVar Int -- ^ Microseconds, oddly enough+  ,messageLoop :: ThreadId+  ,callbacks :: TVar (Map (Type,Type) [Callback])+  }++data Callback where+  Callback :: (Serialize tag, Serialize msg) => (HermesID -> tag -> msg -> IO ()) -> Callback++-- | Seconds+type Age = Double+-- | Seconds+type TTL = Double++-- * External protocol++data FactoidKey = FactoidKey {+  factoidType :: Type,+  factoidTagType :: Type,+  factoidTag :: ByteString+  } deriving(Typeable,Show,Eq,Ord)++instance Serialize FactoidKey where+  put (FactoidKey a b c) = put a >> put b >> put c+  get = liftM3 FactoidKey get get get++data Factoid = Factoid {+  factoidDate :: UTCTime,+  factoidInsertTime :: UTCTime, -- ^ The time at which it was inserted locally+  factoidTTL :: Maybe TTL,+  factoidData :: ByteString,+  factoidSignature :: ByteString,+  factoidSource :: HermesID+  } deriving(Typeable,Show)++makeFactoid :: CoreContext -> Maybe TTL -> ByteString -> IO Factoid+makeFactoid CoreContext{myPrivateKey,myHermesID} ttl bs = do+  now <- getCurrentTime+  return $ Factoid now now ttl bs (sign myPrivateKey bs) myHermesID++verifyFactoid :: CoreContext -> Factoid -> STM Bool+verifyFactoid (peerKeys -> keys) Factoid{factoidData,factoidSignature,factoidSource} = do+  key <- fmap peerKey . M.lookup factoidSource <$> readTVar keys+  return $ maybe False (\k -> verify k factoidData factoidSignature) key++instance Serialize Factoid where+  put (Factoid a b c d e f) = put a >> put b >> put c >> put d >> put e >> put f+  get = Factoid <$> get <*> get <*> get <*> get <*> get <*> get++instance Serialize UTCTime where+  put (UTCTime a b) = put a >> put b+  get = liftM2 UTCTime get get++instance Serialize Day where+  put = put . toModifiedJulianDay+  get = ModifiedJulianDay <$> get++instance Serialize DiffTime where+  put = put . toRational+  get = fromRational <$> get++data FactoidMessage =+  PushFactoid FactoidKey Factoid+  | PushAbort -- ^ Tagged by FactoidKey+  | PullFactoids+  | PulledFactoids [(FactoidKey,Factoid)]+    deriving(Typeable,Show)++instance Serialize FactoidMessage where+  put (PushFactoid a b) = putWord8 0 >> put a >> put b+  put PushAbort = putWord8 1+  put PullFactoids = putWord8 2+  put (PulledFactoids a) = putWord8 3 >> put a+  get = do+    tag <- getWord8+    case tag of+      0 -> liftM2 PushFactoid get get+      1 -> return PushAbort+      2 -> return PullFactoids+      3 -> liftM PulledFactoids get+      _ -> fail "Serialize FactoidMessage: Broken message"++-- | Insert a factoid in the gossip network. This will immediately+-- trigger a limited gossip exchange, hopefully spreading it to a+-- large fraction of the network.+--+-- Factoids are keyed by their type, source, and the type and+-- serialized value of an arbitrary tag. They can be replaced by+-- re-inserting later, and optionally expire after a timeout.+-- +-- Don't rely on the timeout, though. It's for garbage collection, and+-- is not required to be exact.+writeFactoid :: forall factoid tag. (Typeable factoid, Serialize factoid, Typeable tag, Serialize tag)+                => GossipContext -> factoid -> tag+                -> Maybe TTL -- ^ The timeout, in seconds+                -> IO ()+writeFactoid ctx@GossipContext{..} fact tag ttl = do+  factoid <- makeFactoid core ttl (encode fact)+  let key = FactoidKey (showType fact) (showType tag) (encode tag)+  -- Insert the factoid in gossip state+  -- print =<< (atomically $ readTVar factoids)+  insertFactoidUnchecked ctx key factoid+  -- print =<< (atomically $ readTVar factoids)+  -- Perform limited gossip+  forkIO $ limitedGossip core factoid key+  return ()+++-- | Returns true if the fact was valid and insertable, otherwise false+insertFactoid :: GossipContext -> (FactoidKey,Factoid) -> IO Bool+insertFactoid GossipContext{core} (_,factoid) | myHermesID core == factoidSource factoid = return False+insertFactoid ctx@GossipContext{..} (key,factoid) = do+  -- Make sure the factoid hasn't been altered+  -- print =<< (atomically $ readTVar factoids)+  ok <- atomically $ do+    key <- M.lookup (factoidSource factoid) <$> readTVar (peerKeys core)+    -- FIXME: Send the key along as well, and.. something.+    return $ maybe True (\k -> verify (peerKey k) (factoidData factoid) (factoidSignature factoid)) key+  -- And insert.+  if ok+    then do+    inserted <- insertFactoidUnchecked ctx key factoid+    -- Call callbacks+    when inserted $ do+      let mapKey = (factoidTagType key, factoidType key)+      -- print ("Calling callbacks for " ++ show mapKey)+      callbacks <- atomically $ M.findWithDefault [] mapKey <$> readTVar callbacks+      forM_ callbacks $ \(Callback callback) -> do+        -- print "Calling callback"+        let tag = decode $ factoidTag key+            fact = decode $ factoidData factoid+        case (tag,fact) of+          (Right t, Right f) -> callback (factoidSource factoid) t f+          otherwise -> alertM "hermes.gossip.callback" "Unable to decode factoid"+    return inserted+    else do+    alertM "hermes.gossip.push" $ "Incorrect factoid signature for " ++ show factoid+    return False++-- | Returns true if the fact was insertable (not older than one we already have), otherwise false+insertFactoidUnchecked :: GossipContext -> FactoidKey -> Factoid -> IO Bool+insertFactoidUnchecked GossipContext{..} key factoid = atomically $ do+  oldFact <- join . fmap (M.lookup (factoidSource factoid)) . M.lookup key <$> readTVar factoids+  let update = case (factoidDate <$> oldFact, factoidDate factoid) of+        (Just oldDate, newDate) | oldDate >= newDate -> False+        otherwise -> True+  when update $ do+    modifyTVar factoids (adjustWithDefault M.empty (M.insert (factoidSource factoid) factoid) key)+    when (isJust $ factoidTTL factoid) $ modifyTVar factoidTTLs (M.insert (fromJust $ factoidTTL factoid) key)+  return update++-- | Gossips about a fact until someone tells us they already know it+-- (or we run out of peers).+limitedGossip :: CoreContext -> Factoid -> FactoidKey -> IO ()+limitedGossip core factoid key = do+  return () -- FIXME: THIS ENTIRE FUNCTION+  +fooasdhu core factoid key = do+  -- HACK: Let the system bootstrap before we start looking for peers+  threadDelay 100000+  -- Get a list of every peer we know the address of+  -- FIXME: SHUFFLE THE LIST, FILTER OUT MYSELF+  peers <- (fmap . fmap) fst $ atomically $ M.toList <$> readTVar (peerAddress core)+  -- Then contact them, one by one. Instead of doing it serially, +  mvar <- newEmptyMVar+  acceptType core (undefined :: FactoidMessage) key+  timerThread <- forkIO $ forever $ do putMVar mvar Nothing; threadDelay 500000+  recvThread <- forkIO $ forever $ recv' core key >>= putMVar mvar . Just . snd+  let gossiper list@(peer:peers) = do+        input <- takeMVar mvar+        case input of+          Just PushAbort -> return () -- We're done.+          Nothing -> do+            -- Timer triggered, contact new peer+            infoM "hermes.gossip.push" $ "Pushing to " ++ show peer+            forkIO $ send core peer (PushFactoid key factoid)+            gossiper peers+          Just e -> do+            -- What?+            warningM "hermes.gossip.push" $ "Unexpected reply: " ++ show e+            gossiper list+      gossiper [] = return ()+  gossiper peers+  -- Finally, kill the threads.+  killThread timerThread+  killThread recvThread+  -- And reset core.+  refuseType core (undefined :: FactoidMessage) key++-- | Factoid message loop+factoidLoop :: GossipContext -> IO ()+factoidLoop ctx@GossipContext{..} = do+  -- Setup+  mbox <- newEmptyMVar+  delayThread <- forkIO $ do+    threadDelay 100000 -- FIXME: Ugly hack to let clients call setPeriod before we delay.+    forever $ do+      threadDelay =<< atomically (readTVar gossipInterval)+      putMVar mbox Nothing+      infoM "hermes.gossip.pull" "Initiating gossip"+  recvThread <- forkIO $ forever $ recv core >>= putMVar mbox . Just+  -- Aand.. go!+  forever $ do+    input <- takeMVar mbox+    forkIO $ case input of+      Nothing -> do+        -- Time for our periodic gossiping. Pick some random peer.+        peers <- atomically $ readTVar (peerAddress core)+        unless (M.null peers) $ do+          peer <- fst . flip M.elemAt peers <$> randomRIO (0, M.size peers - 1)+          send core peer PullFactoids+      Just (source,PushFactoid key factoid) -> do+        infoM "hermes.gossip.push" "Caught gossip"+        inserted <- insertFactoid ctx (key,factoid)+        if inserted+          then limitedGossip core factoid key+          else send' core source PushAbort key+      Just (source,PullFactoids) -> do+        -- Someone wants to know the truth, the full truth, and nothing but the factoids. Let's oblige.+        facts <- concatMap (\(key,m) -> map ((key,) . snd) $ M.toList m) . M.toList <$>+                 atomically (readTVar factoids)+        infoM "hermes.gossip.pull" $ "Sending gossip: " ++ show (length facts)+        -- print facts+        send core source $ PulledFactoids facts+      Just (_,PulledFactoids factoids) -> do+        infoM "hermes.gossip.pull" $ "Received gossip: " ++ show (length factoids)+        -- print factoids+        mapM_ (insertFactoid ctx) factoids+      Just (source,e) -> do+        warningM "hermes.gossip.pull" $ "Unexpected reply: " ++ show (source,e)++-- | Read a factoid, assuming it exists.+readFactoid :: forall factoid tag. (Typeable factoid, Serialize factoid, Typeable tag, Serialize tag)+               => GossipContext -> tag -> HermesID -> IO (Maybe factoid)+readFactoid GossipContext{factoids} tag source = do+  let key = FactoidKey (showType (undefined :: factoid)) (showType tag) (encode tag)+  factoid <- atomically $ fmap (decode . factoidData) . M.lookup source . M.findWithDefault M.empty key <$> readTVar factoids+  case factoid of+    Nothing -> return Nothing+    Just (Left err) -> alertM "hermes.gossip.readFactoid" ("Factoid decode error: " ++ err) >> return Nothing+    Just (Right f) -> return f++-- | Read all factoids with an appropriate type and tag. Useful if you+-- don't know what source to expect.+readFactoids :: forall factoid tag. (Typeable factoid, Serialize factoid, Typeable tag, Serialize tag)+                => GossipContext -> tag -> IO [(HermesID,factoid)]+readFactoids GossipContext{factoids} tag = do+  let key = FactoidKey (showType (undefined :: factoid)) (showType tag) (encode tag)+  facts <- M.toList . M.findWithDefault M.empty key <$> atomically (readTVar factoids)+  fmap catMaybes $ forM facts $ \(source, decode . factoidData -> factoid) -> do+    case factoid of+      Left err -> alertM "hermes.gossip.readFactoids" ("Factoid decode error: " ++ err) >> return Nothing+      Right f -> return $ Just f+  ++-- | Creates a gossiper from scratch+newGossiper :: CoreContext +               -> Double -- ^ The gossip period, in seconds+               -> IO GossipContext+newGossiper core period = do+  rec+    factoids <- newTVarIO M.empty+    factoidTTLs <- newTVarIO M.empty+    gossipInterval <- newTVarIO $ round $ period * 1000000+    messageLoop <- forkIO $ factoidLoop ctx+    callbacks <- newTVarIO M.empty+    let ctx = GossipContext{..}+  return ctx++-- | Set the period for the periodic gossiper. It will take effect+-- after the next periodic gossip.+setPeriod :: GossipContext+             -> Double -- ^ The period, in seconds+             -> IO ()+setPeriod GossipContext{gossipInterval} period =+  atomically $ writeTVar gossipInterval $ round $ period * 1000000++-- | Snapshots a gossip context for storage. All state is saved,+-- except callbacks.+snapshotGossiper :: GossipContext -> STM ByteString+snapshotGossiper GossipContext{..} = do+  facts <- readTVar factoids+  ttls <- readTVar factoidTTLs+  interval <- readTVar gossipInterval+  return $ encode (facts,ttls,interval)++-- | Restore a gossip context and start gossiping+restoreGossiper :: CoreContext -> ByteString -> IO GossipContext+restoreGossiper core (decode -> Right (facts,ttls,interval)) = do+  rec+    factoids <- newTVarIO facts+    factoidTTLs <- newTVarIO ttls+    gossipInterval <- newTVarIO interval+    messageLoop <- forkIO $ factoidLoop ctx+    callbacks <- newTVarIO M.empty+    let ctx = GossipContext{..}+  return ctx++-- | Add a callback to be called every time a type-matching factoid is+-- inserted or updated. It will not be called for writeFactoid calls.+addCallback :: forall msg tag. (Serialize tag, Typeable tag, Serialize msg, Typeable msg)+               => GossipContext -> (HermesID -> tag -> msg -> IO ()) -> IO ()+addCallback GossipContext{callbacks} function = do+  let callback = Callback function+      key = (showType (undefined :: tag), showType (undefined :: msg))+  -- print ("INserting callback for " ++ show key)+  atomically $ modifyTVar callbacks (adjustWithDefault [] (callback:) key)

@@ -0,0 +1,69 @@+{-# LANGUAGE DeriveDataTypeable #-}+-- | An MChan is a combination of a Map and a TChan.+module Network.Hermes.MChan(+  MChan ,newMChan, newMChanIO, readMChan, writeMChan, writeMChan'+  ,existsMChan, ensureMChan, deleteMChan+  ) where++import Prelude hiding(lookup)++import Control.Concurrent.STM+import Control.Applicative((<$>))+import Data.Typeable+import Data.Map++import Network.Hermes.Misc(modifyTVar)++-- | MChan is an abstract type representing a keyed, unbounded FIFO channel+newtype MChan k v = MChan (TVar (Map k (TChan v)))+                      deriving(Typeable)++-- | Builds and returns a new instance of MChan+newMChan :: Ord k => STM (MChan k v)+newMChan = MChan <$> newTVar empty++-- | IO version of 'newMChan'. This is useful for creating top-level+-- 'MChan's using 'System.IO.Unsafe.unsafePerformIO', because using+-- 'atomically' inside 'System.IO.Unsafe.unsafePerformIO' isn't safe.+newMChanIO :: Ord k => IO (MChan k v)+newMChanIO = MChan <$> newTVarIO empty++-- | Read the next value from an MChan. If the specified key doesn't+-- exist (or is removed while waiting), it returns Nothing.+readMChan :: Ord k => MChan k v -> k -> STM (Maybe v)+readMChan (MChan var) k = do+  chan <- lookup k <$> readTVar var+  case chan of+    Nothing -> return Nothing+    Just chan' -> Just <$> readTChan chan'++-- | Write a value to an MChan. Returns false and discards the value+-- if the specified key doesn't exist.+writeMChan :: Ord k => MChan k v -> k -> v -> STM Bool+writeMChan (MChan var) k v = do+  chan <- lookup k <$> readTVar var+  case chan of+    Nothing -> return False+    Just chan' -> writeTChan chan' v >> return True++-- | Write a value to an MChan, creating the key if it doesn't exist.+writeMChan' :: Ord k => MChan k v -> k -> v -> STM ()+writeMChan' mchan@(MChan var) k v = do+  chan <- lookup k <$> readTVar var+  case chan of+    Nothing -> ensureMChan mchan k >> writeMChan' mchan k v+    Just chan' -> writeTChan chan' v++-- | Checks whether the key exists+existsMChan :: Ord k => MChan k v -> k -> STM Bool+existsMChan (MChan var) k = member k <$> readTVar var++-- | Creates the key if it doesn't already exist+ensureMChan :: Ord k => MChan k v -> k -> STM ()+ensureMChan (MChan var) k = do+  chan <- newTChan+  modifyTVar var (insertWith (const id) k chan)++-- | Delete a key from an MChan+deleteMChan :: Ord k => MChan k v -> k -> STM ()+deleteMChan (MChan var) k = modifyTVar var (delete k)

@@ -0,0 +1,88 @@+{-# LANGUAGE PackageImports, DeriveDataTypeable #-}+-- | Miscellaneous functions. Not really for public consumption.+module Network.Hermes.Misc where++import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as BL+import System.Random(RandomGen)+import Control.Arrow+import Data.Bits+import Control.Exception+import Data.Data+import Control.Applicative+import Control.Concurrent.STM+import "monads-tf" Control.Monad.State+import System.Log.Logger+import Control.Concurrent+import Data.Generics+import Data.Maybe+import Codec.Crypto.RSA+import qualified Data.Map as M+import Data.Map(Map)++ghead :: (Data x, Typeable y) => x -> y+ghead = fromJust . gfindtype++byteStringToInteger :: B.ByteString -> Integer+byteStringToInteger = fst . B.foldl (\(acc,scale) word -> (acc + shiftL (toInteger word) scale, scale+8)) (0,0)++runTMVar :: TMVar s -> StateT s IO a -> IO a+runTMVar var act = block $ do+  initial <- atomically $ takeTMVar var+  (retVal, final) <- unblock (runStateT act initial) `onException` (atomically $ putTMVar var initial)+  atomically $ putTMVar var final+  return retVal+  +throwM :: (MonadIO m, Exception e) => e -> m a+throwM = liftIO . throwIO++modifyTVar :: TVar a -> (a -> a) -> STM ()+modifyTVar var f = writeTVar var =<< f <$> readTVar var++-- | Our notion of types: A shown Typeable.+type Type = String++showType :: Typeable a => a -> Type+showType = show . typeOf++-- | Logs any unhandled exceptions+trapForkIO :: String -> IO () -> IO ThreadId+trapForkIO mod act = forkIO (traplogging mod EMERGENCY ("trap: " ++ mod) act)++-- | Encryption stuff++rsaEncrypt :: (RandomGen g) => g -> PublicKey -> B.ByteString -> (B.ByteString,g)+rsaEncrypt g key bs = first (B.concat . BL.toChunks) $ encrypt g key (BL.fromChunks [bs])++rsaDecrypt :: PrivateKey -> B.ByteString -> B.ByteString+rsaDecrypt key = B.concat . BL.toChunks . decrypt key . BL.fromChunks . (:[])++rsaVerify :: PublicKey -> B.ByteString -> B.ByteString -> Bool+rsaVerify key msg sig = verify key (BL.fromChunks [msg]) (BL.fromChunks [sig])++rsaSign :: PrivateKey -> B.ByteString -> B.ByteString+rsaSign key msg = B.concat $ BL.toChunks $ sign key (BL.fromChunks [msg])++-- | Swap values in a Map. Returns the old value, if any.+swap :: Ord k => k -> v -> M.Map k v -> (Maybe v, M.Map k v)+swap = M.insertLookupWithKey (\_ v _ -> v)++-- | Executes an action once for each value of the TVar. May skip+-- values if it changes quickly.+listenTVar :: Eq a => TVar a -> (a -> IO ()) -> IO ThreadId+listenTVar var act = forkIO $ do+  cur <- atomically $ readTVar var+  act cur+  listen cur+  where listen cur = do+          new <- atomically $ do+            new <- readTVar var+            check (new /= cur)+            return new+          act new+          listen new++adjustWithDefault :: Ord k => a -> (a -> a) -> k -> Map k a -> Map k a+adjustWithDefault def f k m = M.alter adj k m+  where adj Nothing  = Just (f def)+        adj (Just a) = Just (f a)

@@ -0,0 +1,116 @@+{-# LANGUAGE ViewPatterns, ForeignFunctionInterface, CPP #-}+module Network.Hermes.Net(connectStream,streamServer,Address(..),resolve,reverseLookup) where++import Control.Concurrent+import Control.Exception+import Control.Monad+import Data.Maybe++import System.IO+import Network.Socket++import Foreign+import Foreign.C++import Network.Hermes.Protocol+import Network.Hermes.Misc++connectStream :: Address -> IO Handle+connectStream address = do+  ip <- resolve address+  s <- case ip of+    SockAddrInet _ _ -> socket AF_INET Stream defaultProtocol+    SockAddrInet6 _ _ _ _ -> socket AF_INET6 Stream defaultProtocol+    SockAddrUnix _ -> socket AF_UNIX Stream defaultProtocol+  connect s ip+  socketToHandle s ReadWriteMode++-- | Returns the /best/ fit only, or a DNSFailure exception+resolve :: Address -> IO SockAddr+resolve (Unix path) = return $ SockAddrUnix path+resolve address = do+  fits <- getAddrInfo (Just defaultHints{addrFlags=[AI_ADDRCONFIG,AI_NUMERICSERV]})+          (Just $ ghead address)+          (Just $ show (ghead address :: Int))+  let fits' = flip filter (map addrAddress fits) $ \fit ->+        case (fit,address) of+          (_,IP _ _) -> True+          (SockAddrInet _ _, IPv4 _ _) -> True+          (SockAddrInet6 _ _ _ _, IPv6 _ _) -> True+          _ -> False+  when (null fits') $ throwIO $ DNSFailure address+  return $ head fits'++-- | Creates a TCP server that will hand off incoming connections to+-- new threads.+--+-- Killing the server does not kill these forked threads.+--+-- The handle passed to your action will be automatically closed when+-- that action returns.+streamServer :: Address -- ^ Address we should bind to. Use "" for hostname to use all interfaces.+               -> (Handle -> Address -> IO ()) -- ^ Function called to handle connections+               -> IO (IO ()) -- ^ Returns an action you may use to kill the server+streamServer address act = block $ do+  socks <- listenAt address+  threads <- forM socks $ \server -> forkIO $ flip finally (sClose server) $ unblock $ forever $ do+    (sock,sockAddr) <- accept server+    address <- reverseLookup sockAddr+    handle <- socketToHandle sock ReadWriteMode+    trapForkIO "hermes.net.streamServer" $ act handle address `finally` hClose handle+  return $ mapM_ killThread threads++reverseLookup :: SockAddr -> IO Address+reverseLookup (SockAddrUnix path) = return $ Unix path+reverseLookup sockAddr = do+  (Just host, Just (read -> port)) <- getNameInfo [NI_NUMERICSERV] True True sockAddr+  return $ case sockAddr of+    SockAddrInet _ _ -> IPv4 host port+    SockAddrInet6 _ _ _ _ -> IPv6 host port++-- On linux, binding an IPv6 port by default also binds the corresponding IPv4 port. Disable that.+#ifdef __linux+#include <sys/socket.h>+#include <netinet/in.h>++type SockLen = #type socklen_t+iPPROTO_IPV6 :: CInt+iPPROTO_IPV6 = #const IPPROTO_IPV6+iPV6_V6ONLY :: CInt+iPV6_V6ONLY  = #const IPV6_V6ONLY+foreign import ccall unsafe "setsockopt" _setsockopt :: CInt -> CInt -> CInt -> Ptr CInt -> SockLen -> IO ()++#endif++tryListenAt :: Family -> HostName -> Int -> IO (Maybe Socket)+tryListenAt family host port = do+  addr <- getAddrInfo (Just defaultHints{addrFamily=family,addrFlags=[AI_PASSIVE,AI_ADDRCONFIG]})+                     (if null host then Nothing else Just host)+                     (Just (show port))+  case addr of+    [] -> return Nothing+    ((addrAddress -> address):_) -> do+      s <- socket family Stream defaultProtocol+      setSocketOption s ReuseAddr 1+#ifdef __linux+      when (family == AF_INET6) $ do+        alloca $ \intptr -> do+          poke intptr 1+          _setsockopt (fdSocket s) iPPROTO_IPV6 iPV6_V6ONLY intptr (#size int)+#endif+      bindSocket s address+      listen s 3+      return $ Just s+  +listenAt :: Address -> IO [Socket]  +listenAt (IP host port) = do+  v4 <- tryListenAt AF_INET host port+  v6 <- tryListenAt AF_INET6 host port+  when (isNothing v4 && isNothing v6) (error "listenAt: No good exception, FIXME")+  return $ concat $ [maybeToList v4,maybeToList v6]+listenAt (IPv4 host port) = do+  Just s <- tryListenAt AF_INET host port -- FIXME+  return [s]+listenAt (IPv6 host port) = do+  Just s <- tryListenAt AF_INET6 host port -- FIXME+  return [s]

@@ -0,0 +1,263 @@+{-# OPTIONS_HADDOCK hide #-}+{-# LANGUAGE RecordWildCards, BangPatterns, ViewPatterns, DeriveDataTypeable #-}+module Network.Hermes.Protocol where++import Control.Applicative+import Control.Monad+import Control.Exception+import Data.Typeable+import Data.Data++import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as B8++import Network.Hermes.Misc++import Data.Serialize+import Data.Serialize.Put+import Data.Serialize.Get+import Codec.Digest.SHA+import Codec.Crypto.RSA+import Network.Socket(HostName)+import Data.Serialize++-- * Errors++-- | Most Hermes functions can throw one of these exceptions, which+-- | are mainly triggered when (re)negotiating connections.+data HermesException = HermesIDUnknown HermesID+                       -- ^ Hermes has no idea who you're talking about. How did you even get the HermesID?+                       -- HermesID information is never discarded, so this exception should be rather uncommon.+                     | AddressUnknown HermesID+                       -- ^ We don't know where this HermesID is; we never did, or old information proved to be false.+                     | DNSFailure Address+                       -- ^ Failed to resolve the address+                     | WrongProtocol+                       -- ^ The remote server is not speaking Hermes-speak.+                     | ProtocolVersionMismatch Word32 Word32+                       -- ^ A different protocol version is in use at the remote host. Check library version.+                     | AuthError String+                       -- ^ Something went wrong while authenticating. Have a reason.+                     | DeserializationError String+                       -- ^ Something went wrong while deserializing your data.+                     | ListenerAlreadyExists+                       -- ^ Attempted to create a listener on a port we're already listening to+                     | MessageError+                       -- ^ Message corrupted (connection broken)+                     | Timeout+                       -- ^ Some operation took longer than the user-configured timeout+                     | RecvCancelled+                       -- ^ Receive was explicitly cancelled by the user+                     deriving (Typeable,Show,Eq)++instance Exception HermesException++-- | Exceptions that are handled by simply closing the connection+data CloseException = EOF+                    deriving (Typeable,Show,Eq)++instance Exception CloseException++decode' :: Serialize a => B.ByteString -> a+decode' = either (throw . DeserializationError) id . decode++runGet' :: Get a -> B.ByteString -> a+runGet' g = either (throw . DeserializationError) id . runGet g+++-- * And some types++data Address = IP HostName Int -- ^ Host name and port, IPv4, IPv6, or both+             | IPv4 HostName Int -- ^ IPv4 only+             | IPv6 HostName Int -- ^ IPv6 only+             | Unix FilePath -- ^ Unix domain socket, not available on Windows+             deriving(Show,Read,Eq,Ord,Typeable,Data)++instance Serialize Address where+  put (IP a b)   = putWord8 0 >> put a >> put b+  put (IPv4 a b) = putWord8 1 >> put a >> put b+  put (IPv6 a b) = putWord8 2 >> put a >> put b+  put (Unix a)   = putWord8 3 >> put a+  get = do+    tag <- getWord8+    case tag of+      0 -> IP <$> get <*> get+      1 -> IPv4 <$> get <*> get+      2 -> IPv6 <$> get <*> get+      3 -> Unix <$> get+      _ -> error "Corrupted binary data for Address"+++-- * Cryptographic parameters++-- | AES session key size, in bits+aesKeySize :: Int+aesKeySize = 128++-- | Cipher to use for encrypting the session key+evpCipher :: String+evpCipher = "aes-128-cbc"++-- | Hash used all over the place+evpHash :: String+evpHash = "sha256"++-- | RSA key size, in bits; 512 <= size <= 1024+rsaKeySize :: Int+rsaKeySize = 1024++-- | DSA key size, for the signature authorities+dsaKeySize :: Int+dsaKeySize = 1024++-- * Line protocol++-- | Unchangeable bytes telling peers that this.. is... HERMES!+magicString :: B.ByteString+magicString = B8.pack "This.. is... HERMES!\n"++protocolVersion :: Word32 -- Do not change type+protocolVersion = 0++data KeyQuery = KeyOK | RequestKey+              deriving(Show)++-- | A hash computed from a public key+type HermesID = Integer++data KeyReply = KeyReply { keyReplyKey :: PublicKey+                          ,keyReplySig :: Maybe B.ByteString+                         } deriving(Show)++-- | If Indirect, require a signature from an authority.+--+-- If Direct, require an OK from the library client.+--+-- If None, no trust is required.+data TrustLevel = None | Indirect | Direct+                deriving(Eq,Ord,Show)++data SessionSetup = SessionSetup { +  setupKey+  ,setupIV+  ,setupChallenge :: B.ByteString+  ,clientAddress :: Maybe Address+  }+                deriving(Show)++data AnyMessage = AKeyQuery KeyQuery+                | AKeyReply KeyReply+                | AChallenge B.ByteString+                | ASessionSetup B.ByteString+                | AHermesID HermesID+                deriving(Show)++-- | If a message (m :: t) is discarded, then a RejectedMessage is+-- sent in reply, with (showType t,encode (original tag)) as the tag. The message body+-- is discarded.+data RejectedMessage = RejectedMessage+                     deriving(Typeable)++instance Serialize RejectedMessage where+  put _ = return ()+  get   = return RejectedMessage+++instance Serialize PublicKey where+  put PublicKey{..} = put public_size >> put public_n >> put public_e+  get = do+    public_size <- get+    public_n <- get+    public_e <- get+    return PublicKey{..}++instance Serialize PrivateKey where+  put PrivateKey{..} = put private_size >> put private_n >> put private_d+  get = do+    private_size <- get+    private_n <- get+    private_d <- get+    return PrivateKey{..}++-- GENERATED START++instance Serialize KeyQuery where+        put x+          = case x of+                KeyOK -> putWord8 0+                RequestKey -> putWord8 1+        get+          = do i <- getWord8+               case i of+                   0 -> return KeyOK+                   1 -> return RequestKey+                   _ -> error "Corrupted binary data for KeyQuery"++ +instance Serialize KeyReply where+        put (KeyReply x1 x2)+          = do put x1+               put x2+        get+          = do x1 <- get+               x2 <- get+               return (KeyReply x1 x2)++ +instance Serialize TrustLevel where+        put x+          = case x of+                None -> putWord8 0+                Indirect -> putWord8 1+                Direct -> putWord8 2+        get+          = do i <- getWord8+               case i of+                   0 -> return None+                   1 -> return Indirect+                   2 -> return Direct+                   _ -> error "Corrupted binary data for TrustLevel"++ +instance Serialize SessionSetup where+        put (SessionSetup x1 x2 x3 x4)+          = do put x1+               put x2+               put x3+               put x4+        get+          = do x1 <- get+               x2 <- get+               x3 <- get+               x4 <- get+               return (SessionSetup x1 x2 x3 x4)++ +instance Serialize AnyMessage where+        put x+          = case x of+                AKeyQuery x1 -> do putWord8 1+                                   put x1+                AKeyReply x1 -> do putWord8 2+                                   put x1+                AChallenge x1 -> do putWord8 3+                                    put x1+                ASessionSetup x1 -> do putWord8 4+                                       put x1+                AHermesID x1 -> do putWord8 5+                                   put x1+        get+          = do i <- getWord8+               case i of+                   1 -> do x1 <- get+                           return (AKeyQuery x1)+                   2 -> do x1 <- get+                           return (AKeyReply x1)+                   3 -> do x1 <- get+                           return (AChallenge x1)+                   4 -> do x1 <- get+                           return (ASessionSetup x1)+                   5 -> do x1 <- get+                           return (AHermesID x1)+                   _ -> error "Corrupted binary data for AnyMessage"+-- GENERATED STOP

@@ -0,0 +1,115 @@+{-# OPTIONS_HADDOCK hide #-}+{-# LANGUAGE RecordWildCards, DeriveDataTypeable,+    ScopedTypeVariables, ExistentialQuantification, NamedFieldPuns, +    ViewPatterns #-}+-- | This module allows you to register procedures which can then be+-- | called from another node.  Due to limitations in GHC, it is not+-- | possible to snapshot its state; you will always need to+-- | re-register them at program startup.+--+-- | RPC calls are distinguished using an arbitrary name, as well as+-- | the type of the parameter and return value.+module Network.Hermes.RPC where++import Control.Monad+import Control.Applicative+import Control.Concurrent+import Control.Concurrent.MVar+import Control.Concurrent.STM+import Control.Exception++import qualified Data.Map as M+import Data.Map(Map)+import Data.ByteString+import Data.Typeable+import Data.Serialize+import Data.Serialize.Put+import Data.Serialize.Get+import System.Random+import System.Log.Logger+import Data.Int+import Data.Unamb(race)++import Network.Hermes.Core+import Network.Hermes.Misc+import Network.Hermes.Protocol+import Network.Hermes.MChan++type ProcName = String++-- | ProcName :: Type+type ProcId = (ProcName,Type)++-- OPTIMIZE: Use a proper Word128 or something.+type Serial = Integer++data RPCContext = RPCContext {+  core :: CoreContext+  ,nextSerial :: IO Serial+  ,callbacks :: MVar (Map ProcId ThreadId)+  }++data RPCQuery p r = RPCQuery {+  parameter :: p+  ,serial :: Serial+  } deriving(Typeable)++instance Serialize p => Serialize (RPCQuery p r) where+  put (RPCQuery p s) = put p >> put s+  get = liftM2 RPCQuery get get++data RPCReply p r = RPCReply {+  reply :: r+  } deriving(Typeable)+  +instance Serialize r => Serialize (RPCReply p r) where+  put (RPCReply r) = put r+  get = RPCReply <$> get++-- | Do NOT create multiple contexts for one CoreContext. They'll conflict.+newContext :: CoreContext+             -> IO RPCContext+newContext core = do+  serialV <- newMVar =<< randomRIO (0,2^128)+  let nextSerial = modifyMVar serialV (return . join (,) . succ)+  callbacks <- newMVar M.empty+  return RPCContext{..}++-- | Registers (or replaces) a callback to be executed+-- when we receive a call to this name.+--+-- Individual callbacks are currently always executed in serial. Do we+-- want parallel?+registerCallback :: forall p r. (Serialize p, Serialize r, Typeable p, Typeable r)+                   => RPCContext+                   -> ProcName   -- ^ Callback's name+                   -> (p -> IO r) -- ^ The callback itself+                   -> IO ()+registerCallback RPCContext{callbacks,core} name proc = block $ do+  modifyMVar_ callbacks $ \cbmap -> do+    let key = (name,showType proc)+        oldTid = M.lookup key cbmap+    maybe (return ()) killThread oldTid+    tid <- forkIO handleCallback+    return $ M.insert key tid cbmap+  where+    handleCallback = unblock $ forever $ do+      (uuid, RPCQuery{..} :: RPCQuery p r) <- recv' core name+      reply <- proc parameter+      send' core uuid (RPCReply{..} :: RPCReply p r) serial++-- | Remote procedure call+-- Apart from core exceptions, it may fail in the specific case that the procedure doesn't exist,+-- in which case it returns Nothing.+call :: forall p r. (Serialize p, Typeable p, Serialize r, Typeable r) =>+       RPCContext -> HermesID -> ProcName -> p -> IO (Maybe r)+call RPCContext{..} uuid name parameter = do+  serial <- nextSerial+  send' core uuid (RPCQuery{parameter,serial} :: RPCQuery p r) name+  let getReply = do+        (_,msg :: RPCReply p r) <- recv' core serial+        return $ Just $ reply msg+      getFailure = do+        (_,RejectedMessage) <- recv' core (showType serial, encode serial)+        return Nothing+  race getReply getFailure

@@ -0,0 +1,53 @@+{-# OPTIONS_HADDOCK hide #-}+{-# LANGUAGE GeneralizedNewtypeDeriving, ViewPatterns #-}+module Network.Hermes.Signature(+  Authority(..), SignatureRequest(..)+  ,newAuthority+  ,newSignatureRequest+  ,signRequest+  ,newSignedContext+  ) where++import Control.Applicative+import Data.Serialize+import qualified Data.ByteString as B+import Codec.Crypto.RSA++import Network.Hermes.Types+import Network.Hermes.Protocol+import Network.Hermes.Misc+import Codec.Crypto.AES.Random+import Network.Hermes.Core(newContext,setKeySignature,addAuthority)++newtype SignatureRequest = SignatureRequest B.ByteString+                           deriving(Serialize)++data Authority = Authority { authorityPrivateKey :: PrivateKey, authorityKey :: PublicKey }+               deriving(Show)++instance Serialize Authority where+  put (Authority a b) = put a >> put b+  get = Authority <$> get <*> get++newAuthority :: IO Authority+newAuthority = do+  g <- newAESGen+  let (pub,priv,_) = generateKeyPair g rsaKeySize+  return $ Authority priv pub++-- | Creates a signature request, to fullfill on another computer.+newSignatureRequest :: CoreContext -> SignatureRequest+newSignatureRequest = SignatureRequest . encode . myKey++-- | Sign a request. Use setKeySignature to install it.+signRequest :: Authority -> SignatureRequest -> Signature+signRequest (authorityPrivateKey -> key) (SignatureRequest req) = rsaSign key req++-- | Creates a pre-signed Context, with this authority set as trusted+newSignedContext :: Authority -> IO CoreContext+newSignedContext authority = do+  ctx <- newContext+  let sig = signRequest authority $ newSignatureRequest ctx+  setKeySignature ctx sig+  addAuthority ctx (authorityKey authority)+  return ctx

@@ -0,0 +1,84 @@+{-# OPTIONS_HADDOCK hide #-}+{-# LANGUAGE RecordWildCards #-}+module Network.Hermes.Types where++import Control.Applicative++import Control.Concurrent.STM(TVar,TChan,TMVar)+import Data.Set(Set)+import Data.Map(Map)+import qualified Data.ByteString as B+import System.IO(Handle)+import Data.Serialize++import Network.Hermes.Protocol+import Network.Hermes.Misc+import Network.Hermes.MChan+import Codec.Crypto.RSA(PublicKey,PrivateKey)+import Codec.Crypto.AES.IO(AESCtx)++-- * Types++data CoreContext = CoreContext {+  myKey :: PublicKey+  ,myPrivateKey :: PrivateKey+  ,myHermesID :: HermesID+  ,myKeySignature :: TVar (Maybe Signature)+  ,authorities :: TVar [PublicKey] -- ^ List of keys we trust to sign other keys for indirect trust+  ,listeners :: TVar (Set ListenerAddress) -- ^ List of open ports for incoming traffic+   -- CHECKME: Would killing these actually do anything useful?+  ,listenerKillers :: TVar (Map ListenerAddress (IO ()))+  ,peerAddress :: TVar (Map HermesID Address)+  ,peerKeys :: TVar (Map HermesID PeerKey)+  ,peerFailures :: TVar (Map HermesID Int)+  ,peerConnections :: TVar (Map HermesID (TMVar Connection))+  ,trustLimit :: TVar TrustLevel -- ^ Peers with less than this level of trust won't be trusted. AdHoc or Indirect are good values.+  ,messageBox :: MChan (Type,Type,B.ByteString) (HermesID,B.ByteString)+   -- ^ This is a.. map of message type, tag type and tag value to the message origin and message+  ,timeLimit :: TVar Double -- ^ In seconds+  }++data CoreContextSnapshot = CoreContextSnapshot {+  myKeySnap :: PublicKey+  ,myPrivateKeySnap :: PrivateKey+  ,myHermesIDSnap :: HermesID+  ,myKeySignatureSnap :: Maybe Signature+  ,authoritiesSnap :: [PublicKey]+  ,peerAddressSnap :: Map HermesID Address+  ,peerKeysSnap :: Map HermesID PeerKey+  ,peerFailuresSnap :: Map HermesID Int+  ,trustLimitSnap :: TrustLevel+  ,timeLimitSnap :: Double+   } deriving(Show)++instance Serialize CoreContextSnapshot where+  put (CoreContextSnapshot a b c d e f g h i j) = put a >> put b >> put c >> put d >> put e >> put f >> put g >> put h >> put i >> put j+  get = CoreContextSnapshot <$> get <*> get <*> get <*> get <*> get <*> get <*> get <*> get <*> get <*> get++data ListenerAddress = ListenerAddress {+  localAddress :: Address -- ^ The address we bound to on this system+  ,remoteAddress :: Address -- ^ The one other peers see. May be different due, mostly, to port-forwarding+  } deriving(Eq,Ord)+++-- * Cryptographic types and peers++type Signature = B.ByteString++data PeerKey = PeerKey {+  peerKey :: PublicKey+  ,trust :: TrustLevel+  ,signature :: Maybe Signature+  } deriving(Show)++instance Serialize PeerKey where+  put (PeerKey a b c) = put a >> put b >> put c+  get = PeerKey <$> get <*> get <*> get++data Connection = Connection {+  aesctx :: AESCtx+  ,aesKey :: B.ByteString+  ,handle :: Handle+  ,typeMap :: TVar (Map Type Int)+  ,typeMax :: TVar Int+  }

@@ -0,0 +1,4 @@+{-# LANGUAGE NamedFieldPuns #-}+import Distribution.Simple++main = defaultMain