ConcurrentUtils 0.4.1.0 → 0.4.2.0

raw patch · 4 files changed

+140/−42 lines, 4 filesdep +RSAdep +crypto-randomdep +cryptohashsetup-changedPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: RSA, crypto-random, cryptohash, reexport-crypto-random, securemem, tagged

API changes (from Hackage documentation)

Files

• ConcurrentUtils.cabal +2/−2
• Control/CUtils/Deadlock.hs +13/−20
• Control/CUtils/NetChan.hs +122/−20
• Setup.hs +3/−0

@@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/
 
 name:                ConcurrentUtils
-version:             0.4.1.0
+version:             0.4.2.0
 synopsis:            Concurrent utilities
 -- description:         
 homepage:            http://alkalisoftware.net
@@ -18,4 +18,4 @@ library
   exposed-modules:     Control.CUtils.Processes, Control.CUtils.NetChan, Control.CUtils.FChan, Control.CUtils.Deadlock, Control.CUtils.DataParallel, Control.CUtils.Conc, Control.CUtils.Channel, Control.CUtils.AList
   other-modules:       Control.CUtils.Split   
-  build-depends:       base >=4 && <=5, process, network >=2.4, bytestring, binary, containers, array, parallel
+  build-depends:       base >=4 && <=5, process, network >=2.4, bytestring, binary, containers, array, parallel, cryptohash >=0.11.6, RSA >=2.1.0, crypto-random >=0.0.8, securemem >= 0.1.7, reexport-crypto-random, tagged >= 0.7.3

@@ -5,7 +5,7 @@ -- Automatic deadlock detection is inefficient, and computations cannot be rolled
 -- back or aborted in general.
 --
--- Instead, we prevent deadlocks before they happen.
+-- Instead, I prevent deadlocks before they happen.
 module Control.CUtils.Deadlock (Res(Lift, Acq, Rel, Fork, Plus, Id), run, lft) where
 
 import Control.Category
@@ -20,36 +20,29 @@ import Control.Concurrent
 import Prelude hiding (id, (.))
 
--- The typical sequence that produces a deadlock is as follows:
+-- | The typical sequence that produces a deadlock is as follows:
 --
 -- (1) Thread 1 acquires lock A
+--
 -- (2) Thread 2 acquires lock B
+--
 -- (3) Thread 1 tries to acquire B
+--
 -- (4) Thread 2 tries to acquire A
+--
 -- Deadlock.
 --
 -- Standard deadlock detection intervenes after (4) has occurred.
--- We should intervene in a lock acquisition that is followed
--- by an unsafe schedule (here at (2)). We suspend thread 2
+-- I intervene in a lock acquisition that is followed
+-- by an unsafe schedule (here at (2)). I suspend thread 2
 -- until a safe schedule is guaranteed -- in this case until
 -- thread 1 relinquishes lock A.
 --
--- We need to do some kind of static analysis on the threads
--- to do this. Haskell arrows make possible a kind of JIT
--- static analysis. We leverage the fact that considerable
--- computation has been done to reach a certain point --
--- we only have to analyse the immediate continuation of
--- a thread.
-
--- | The Res arrow.
+-- The Res arrow.
 --
---   Computations are built with these constructors (and the arrow
---   interface). The implementation guarantees progress provided:
---    * Pieces of the arrow that hold locks are finitely examinable,
---    * threads are programmed to eventually release a lock they hold,
---    * locks are the only source of deadlock,
---    * and all locks are used only with the Acq and Rel ctors (which
---    acquire and release a lock resp.).
+-- Computations are built with these constructors (and the arrow
+-- interface). Pieces of the arrow that hold locks have to be finitely examinable,
+-- Locks have to be used with the Acq and Rel constructors.
 data Res t u where
 	Lift :: Kleisli IO t v -> Res v u -> Res t u
 	Acq :: MVar () -> Res t u -> Res t u -- acquire a lock
@@ -86,7 +79,7 @@ {-# NOINLINE resource #-}
 resource = unsafePerformIO (newMVar M.empty)
 
--- A hazard is an ACQUIRE-HOLD cycle among threads.
+-- A hazard is a HOLD-ACQUIRE cycle among threads.
 -- We generate all sequences looking for a cycle.
 
 selects ls = [ (y, xs ++ ys) | xs <- inits ls | y:ys <- tails ls ]

@@ -1,21 +1,23 @@ {-# LANGUAGE CPP, ScopedTypeVariables #-}
 
 -- | A channel module with transparent network communication.
-module Control.CUtils.NetChan (NetSend, NetRecv, localHost, newNetChan, newNetSend, newNetRecv, send, recv, recvSend, sendRecv, recvRecv, activateSend, activateRecv) where
+module Control.CUtils.NetChan (NetSend, NetRecv, localHost, newNetChan, newNetSend, newNetRecv, send, receive, recv, recvSend, sendRecv, recvRecv, activateSend, activateRecv, Auth, authServer, authClient, example) where
 
 -- This module has a strategy for routing around dead nodes. See 'routeAround'.
 
 import System.IO
 import System.Process
-import Data.List (find, isPrefixOf, isInfixOf, (\\))
+import Data.List (find, isPrefixOf, (\\))
 import Network
 import Network.Socket (socketToHandle, SockAddr(..))
 import Network.BSD
 import Control.Concurrent
 import Control.Monad
-import Data.ByteString.Lazy (ByteString, hGet, hPut, length, fromChunks, append, empty)
+import Data.ByteString.Lazy hiding (map, isPrefixOf, dropWhile, drop, head, split)
 import qualified Data.ByteString as B
 import Data.Binary
+import Data.Binary.Get
+import Data.Binary.Put
 import qualified Data.Map as M
 import Data.Maybe
 import Data.Char
@@ -23,7 +25,14 @@ import Data.Bits
 import Control.Exception
 import System.IO.Unsafe
-import Prelude hiding (lookup, length, catch)
+import Crypto.Hash.SHA512
+import Codec.Crypto.RSA.Pure
+import Crypto.Random
+import Reexport.Crypto.Random
+import Data.SecureMem
+import Foreign.Storable
+import Data.Tagged
+import Prelude hiding (lookup, length, splitAt, catch)
 
 import Control.CUtils.Split
 
@@ -36,13 +45,13 @@ table :: MVar (M.Map Ident (ByteString -> IO ()))
 table = unsafePerformIO (newMVar (M.singleton empty (\_ -> return ())))
 
-data ChannelFibre t = ChannelFibre (MVar Bool) Handle
+data ChannelFibre = ChannelFibre (MVar Bool) Handle
 
-data NetSend t = NetSend HostName Ident (MVar [HostName]) (MVar [ChannelFibre t])
+data NetSend t = NetSend HostName Ident (MVar [HostName]) (MVar [ChannelFibre])
 
-data NetRecv t = NetRecv Ident (NetSend t) (NetSend HostName) (Chan t)
+data NetRecv t = NetRecv Ident (NetSend t) (NetSend HostName) (Chan ByteString)
 
-instance Eq (ChannelFibre t) where
+instance Eq ChannelFibre where
 	ChannelFibre _ hdl == ChannelFibre _ hdl2 = hdl == hdl2
 
 instance Eq (NetSend t) where
@@ -74,7 +83,7 @@ 	let ident = identifier host (fromIntegral (M.size mp))
 	liftM2 (,) (__newNetRecv True Nothing ident) (__newNetSend True host ident)
 
-modifyIdent b ident = append (fromChunks [B.pack $ map (fromIntegral . ord) $ if b then "main" else "back"]) ident
+modifyIdent b ident = append (pack $ map (fromIntegral . ord) $ if b then "main" else "back") ident
 
 __emptyNetSend :: Bool -> NetSend HostName -> HostName -> Ident -> IO (NetSend t)
 __emptyNetSend b backDown hostName ident = do
@@ -172,11 +181,10 @@ 	let listener bs = do
 		got <- readIORef gotUpstreams
 		if got then do
-				let x = decode bs
-				writeChan chan x
+				writeChan chan bs
 
 				-- Send the value to downstream receive ends.
-				send downstream x
+				__send downstream bs
 			else do
 				writeIORef gotUpstreams True
 				let x:xs = decode bs
@@ -205,20 +213,25 @@ 	mapM_ (__addConnection s) hosts
 	modifyMVar_ mvar (return . (\\[fib]))
 
--- | Sends something on a channel.
-send :: (Binary t) => NetSend t -> t -> IO ()
-send snd@(NetSend _ ident _ mvar) x = readMVar mvar >>= mapM_ (\fib@(ChannelFibre mvar hdl) -> do
-	b <- modifyMVar mvar (\b -> let s = encode x in
-		s `seq` catch (hPut hdl (encode (length s)) >> hPut hdl s) (\(_ :: SomeException) -> routeAround fib snd >> send snd x)
+__send snd@(NetSend _ ident _ mvar) s = readMVar mvar >>= mapM_ (\fib@(ChannelFibre mvar hdl) -> do
+	b <- modifyMVar mvar (\b ->
+		s `seq` catch (hPut hdl (encode (length s)) >> hPut hdl s) (\(_ :: SomeException) -> routeAround fib snd >> __send snd s)
 		>> return (True, b))
 	-- Buffering
 	unless b $ void $ forkIO $ do
 		threadDelay 100000
 		modifyMVar_ mvar (\_ -> return False)
-		catch (hFlush hdl) (\(_ :: SomeException) -> routeAround fib snd >> send snd x))
+		catch (hFlush hdl) (\(_ :: SomeException) -> routeAround fib snd >> __send snd s))
 
+-- | Sends something on a channel.
+send :: (Binary t) => NetSend t -> t -> IO ()
+send snd x = __send snd (encode x)
+
+receive (NetRecv _ _ _ chan) = readChan chan
+
 -- | Receives something from a channel.
-recv (NetRecv _ _ _ chan) = readChan chan
+recv :: (Binary t) => NetRecv t -> IO t
+recv r = liftM decode $ receive r
 
 --- Sending and receiving channels.
 
@@ -248,9 +261,98 @@ 	put (NetRecv ident _ _ _) = put ident
 	get = liftM (\x -> NetRecv x undefined undefined undefined) get
 
--- | 'get' produces channel ends with some data missing. Use these to make them usable.
 activateSend :: NetSend t -> IO (NetSend t)
 activateSend (NetSend hostName ident _ _) = __newNetSend True hostName ident
 
 activateRecv :: (Binary t) => NetRecv t -> IO (NetRecv t)
 activateRecv (NetRecv x _ _ _) = __newNetRecv True Nothing x
+
+repeatM m = m >> repeatM m
+
+data Auth t = Auth B.ByteString B.ByteString ByteString
+
+putLazy = mapM_ putByteString . toChunks
+
+instance Binary (Auth t) where
+	put (Auth b b2 b3) = putByteString b >> putByteString b2 >> putLazy b3
+	get = liftM3 Auth (getByteString 64) (getByteString 100) getRemainingLazyByteString
+
+instance CryptoRandomGen EntropyPool where
+	newGen _ = error "newGen: unsupported on SystemRNG"
+	genSeedLength = Tagged maxBound
+	genBytes l g = entropy `seq` Right (readSecureMem entropy, g) where
+		entropy = grabEntropy l g
+	reseed _ = Right
+	reseedInfo _ = Never
+	reseedPeriod _ = Never
+
+readSecureMem mem = unsafePerformIO $ withSecureMemPtrSz mem $ \n p -> liftM B.pack $ mapM (peekByteOff p) [0..n-1]
+
+-- | Remote exercise of authority. Commands are transmitted in the clear,
+--   but authenticated.
+--
+--   auth - The authority to be served (runs on a separate thread).
+--
+--   r - The receive end from the host.
+--
+--   s - The send end to the host.
+--
+--   publicKey - The public key of the intended recipient.
+authServer :: (Binary t) => (t -> IO ()) -> NetRecv (Auth t) -> NetSend ByteString -> PublicKey -> IO ()
+authServer auth r s publicKey = do
+	pool <- createEntropyPool
+
+	-- Engage in crypto to agree on a random certificate.
+	(cert, g) <- either throwIO return $ genBytes 100 pool
+	cert <- return $ fromChunks [cert]
+	let ei = encrypt g publicKey cert
+	(enc, _) <- either throwIO return ei
+	send s enc
+
+	-- Accept requests.
+	forkIO $ repeatM $ do
+		command <- receive r
+		let (tk, dr) = splitAt 64 command
+		let x = runGet (getByteString 100 >> get) dr
+		-- Check the hash before approving the command.
+		when (fromChunks [hashlazy $ append cert dr] == tk) $ auth x
+
+	return ()
+
+-- | privateKey - The private key for this host.
+--
+--   Returns a function that can be used to send messages.
+authClient :: (Binary t) => NetRecv ByteString -> NetSend (Auth t) -> PrivateKey -> IO (t -> IO ())
+authClient r s privateKey = do
+	-- Decrypt the certificate.
+	enc <- recv r
+	let ei = decrypt privateKey enc
+	cert <- either throwIO return ei
+
+	pool <- createEntropyPool
+	return $ \x -> do
+		salt <- grabEntropyIO 100 pool
+		salt <- return $ readSecureMem salt
+		let enc = encode x
+		send s $ Auth (hashlazy $ cert `append` fromChunks [salt] `append` enc) salt enc
+-- The format of an authenticated record is:
+-- 
+-- * An eight-byte record length, in bytes
+--
+-- * A 64-byte hash
+--
+-- * A 100-byte salt
+--
+-- * The remainder of the record contains the data
+
+example = do
+	pool <- createEntropyPool
+	let Right (pub, priv, _) = generateKeyPair pool 1024
+	(r :: NetRecv ByteString, s) <- newNetChan
+	(r2 :: NetRecv (Auth Int), s2) <- newNetChan
+	authServer print r2 s pub
+	threadDelay 100000
+	f <- authClient r s2 priv
+	f 1
+	f 2
+	f 5

@@ -1,2 +1,5 @@ import Distribution.Simple
+import System.Process
+import System.Directory
+
 main = defaultMain