Assistant monad, stage 3
All toplevel named threads are converted to the Assistant monad. Some utility functions still need to be converted.
This commit is contained in:
Joey Hess
parent
1df7417403
commit
67ce7929a5
7 changed files with 174 additions and 182 deletions
|
|
@ -11,8 +11,6 @@ import Assistant.Common
|
|||
import Assistant.Pairing
|
||||
import Assistant.Pairing.Network
|
||||
import Assistant.Pairing.MakeRemote
|
||||
import Assistant.ThreadedMonad
|
||||
import Assistant.ScanRemotes
|
||||
import Assistant.DaemonStatus
|
||||
import Assistant.WebApp
|
||||
import Assistant.WebApp.Types
|
||||
|
|
@ -27,117 +25,116 @@ import Data.Char
|
|||
thisThread :: ThreadName
|
||||
thisThread = "PairListener"
|
||||
|
||||
pairListenerThread :: ThreadState -> DaemonStatusHandle -> ScanRemoteMap -> UrlRenderer -> NamedThread
|
||||
pairListenerThread st dstatus scanremotes urlrenderer = thread $ liftIO $ withSocketsDo $
|
||||
runEvery (Seconds 1) $ void $ tryIO $ do
|
||||
sock <- getsock
|
||||
go sock [] []
|
||||
where
|
||||
thread = NamedThread thisThread
|
||||
pairListenerThread :: UrlRenderer -> NamedThread
|
||||
pairListenerThread urlrenderer = NamedThread "PairListener" $ do
|
||||
listener <- asIO $ go [] []
|
||||
liftIO $ withSocketsDo $
|
||||
runEvery (Seconds 1) $ void $ tryIO $
|
||||
listener =<< getsock
|
||||
where
|
||||
{- Note this can crash if there's no network interface,
|
||||
- or only one like lo that doesn't support multicast. -}
|
||||
getsock = multicastReceiver (multicastAddress $ IPv4Addr undefined) pairingPort
|
||||
|
||||
{- Note this can crash if there's no network interface,
|
||||
- or only one like lo that doesn't support multicast. -}
|
||||
getsock = multicastReceiver (multicastAddress $ IPv4Addr undefined) pairingPort
|
||||
go reqs cache sock = liftIO (getmsg sock []) >>= \msg -> case readish msg of
|
||||
Nothing -> go reqs cache sock
|
||||
Just m -> do
|
||||
sane <- checkSane msg
|
||||
(pip, verified) <- verificationCheck m
|
||||
=<< (pairingInProgress <$> daemonStatus)
|
||||
let wrongstage = maybe False (\p -> pairMsgStage m <= inProgressPairStage p) pip
|
||||
case (wrongstage, sane, pairMsgStage m) of
|
||||
-- ignore our own messages, and
|
||||
-- out of order messages
|
||||
(True, _, _) -> go reqs cache sock
|
||||
(_, False, _) -> go reqs cache sock
|
||||
(_, _, PairReq) -> if m `elem` reqs
|
||||
then go reqs (invalidateCache m cache) sock
|
||||
else do
|
||||
pairReqReceived verified urlrenderer m
|
||||
go (m:take 10 reqs) (invalidateCache m cache) sock
|
||||
(_, _, PairAck) -> do
|
||||
cache' <- pairAckReceived verified pip m cache
|
||||
go reqs cache' sock
|
||||
(_, _, PairDone) -> do
|
||||
pairDoneReceived verified pip m
|
||||
go reqs cache sock
|
||||
|
||||
{- As well as verifying the message using the shared secret,
|
||||
- check its UUID against the UUID we have stored. If
|
||||
- they're the same, someone is sending bogus messages,
|
||||
- which could be an attempt to brute force the shared secret. -}
|
||||
verificationCheck _ Nothing = return (Nothing, False)
|
||||
verificationCheck m (Just pip)
|
||||
| not verified && sameuuid = do
|
||||
liftAnnex $ warning
|
||||
"detected possible pairing brute force attempt; disabled pairing"
|
||||
stopSending pip <<~ daemonStatusHandle
|
||||
return (Nothing, False)
|
||||
|otherwise = return (Just pip, verified && sameuuid)
|
||||
where
|
||||
verified = verifiedPairMsg m pip
|
||||
sameuuid = pairUUID (inProgressPairData pip) == pairUUID (pairMsgData m)
|
||||
|
||||
go sock reqs cache = getmsg sock [] >>= \msg -> case readish msg of
|
||||
Nothing -> go sock reqs cache
|
||||
Just m -> do
|
||||
sane <- checkSane msg
|
||||
(pip, verified) <- verificationCheck m
|
||||
=<< (pairingInProgress <$> getDaemonStatus dstatus)
|
||||
let wrongstage = maybe False (\p -> pairMsgStage m <= inProgressPairStage p) pip
|
||||
case (wrongstage, sane, pairMsgStage m) of
|
||||
-- ignore our own messages, and
|
||||
-- out of order messages
|
||||
(True, _, _) -> go sock reqs cache
|
||||
(_, False, _) -> go sock reqs cache
|
||||
(_, _, PairReq) -> if m `elem` reqs
|
||||
then go sock reqs (invalidateCache m cache)
|
||||
else do
|
||||
pairReqReceived verified dstatus urlrenderer m
|
||||
go sock (m:take 10 reqs) (invalidateCache m cache)
|
||||
(_, _, PairAck) ->
|
||||
pairAckReceived verified pip st dstatus scanremotes m cache
|
||||
>>= go sock reqs
|
||||
(_, _, PairDone) -> do
|
||||
pairDoneReceived verified pip st dstatus scanremotes m
|
||||
go sock reqs cache
|
||||
{- Various sanity checks on the content of the message. -}
|
||||
checkSane msg
|
||||
{- Control characters could be used in a
|
||||
- console poisoning attack. -}
|
||||
| any isControl msg || any (`elem` "\r\n") msg = do
|
||||
liftAnnex $ warning
|
||||
"illegal control characters in pairing message; ignoring"
|
||||
return False
|
||||
| otherwise = return True
|
||||
|
||||
{- As well as verifying the message using the shared secret,
|
||||
- check its UUID against the UUID we have stored. If
|
||||
- they're the same, someone is sending bogus messages,
|
||||
- which could be an attempt to brute force the shared
|
||||
- secret.
|
||||
-}
|
||||
verificationCheck m (Just pip) = do
|
||||
let verified = verifiedPairMsg m pip
|
||||
let sameuuid = pairUUID (inProgressPairData pip) == pairUUID (pairMsgData m)
|
||||
if not verified && sameuuid
|
||||
then do
|
||||
runThreadState st $
|
||||
warning "detected possible pairing brute force attempt; disabled pairing"
|
||||
stopSending dstatus pip
|
||||
return (Nothing, False)
|
||||
else return (Just pip, verified && sameuuid)
|
||||
verificationCheck _ Nothing = return (Nothing, False)
|
||||
|
||||
{- Various sanity checks on the content of the message. -}
|
||||
checkSane msg
|
||||
{- Control characters could be used in a
|
||||
- console poisoning attack. -}
|
||||
| any isControl msg || any (`elem` "\r\n") msg = do
|
||||
runThreadState st $
|
||||
warning "illegal control characters in pairing message; ignoring"
|
||||
return False
|
||||
| otherwise = return True
|
||||
{- PairReqs invalidate the cache of recently finished pairings.
|
||||
- This is so that, if a new pairing is started with the
|
||||
- same secret used before, a bogus PairDone is not sent. -}
|
||||
invalidateCache msg = filter (not . verifiedPairMsg msg)
|
||||
|
||||
{- PairReqs invalidate the cache of recently finished pairings.
|
||||
- This is so that, if a new pairing is started with the
|
||||
- same secret used before, a bogus PairDone is not sent. -}
|
||||
invalidateCache msg = filter (not . verifiedPairMsg msg)
|
||||
|
||||
getmsg sock c = do
|
||||
(msg, n, _) <- recvFrom sock chunksz
|
||||
if n < chunksz
|
||||
then return $ c ++ msg
|
||||
else getmsg sock $ c ++ msg
|
||||
where
|
||||
chunksz = 1024
|
||||
getmsg sock c = do
|
||||
(msg, n, _) <- recvFrom sock chunksz
|
||||
if n < chunksz
|
||||
then return $ c ++ msg
|
||||
else getmsg sock $ c ++ msg
|
||||
where
|
||||
chunksz = 1024
|
||||
|
||||
{- Show an alert when a PairReq is seen. -}
|
||||
pairReqReceived :: Bool -> DaemonStatusHandle -> UrlRenderer -> PairMsg -> IO ()
|
||||
pairReqReceived True _ _ _ = noop -- ignore our own PairReq
|
||||
pairReqReceived False dstatus urlrenderer msg = do
|
||||
url <- renderUrl urlrenderer (FinishPairR msg) []
|
||||
void $ addAlert dstatus $ pairRequestReceivedAlert repo
|
||||
pairReqReceived :: Bool -> UrlRenderer -> PairMsg -> Assistant ()
|
||||
pairReqReceived True _ _ = noop -- ignore our own PairReq
|
||||
pairReqReceived False urlrenderer msg = do
|
||||
url <- liftIO $ renderUrl urlrenderer (FinishPairR msg) []
|
||||
dstatus <- getAssistant daemonStatusHandle
|
||||
liftIO $ void $ addAlert dstatus $ pairRequestReceivedAlert repo
|
||||
AlertButton
|
||||
{ buttonUrl = url
|
||||
, buttonLabel = T.pack "Respond"
|
||||
, buttonAction = Just $ removeAlert dstatus
|
||||
}
|
||||
where
|
||||
repo = pairRepo msg
|
||||
where
|
||||
repo = pairRepo msg
|
||||
|
||||
{- When a verified PairAck is seen, a host is ready to pair with us, and has
|
||||
- already configured our ssh key. Stop sending PairReqs, finish the pairing,
|
||||
- and send a single PairDone.
|
||||
-}
|
||||
pairAckReceived :: Bool -> Maybe PairingInProgress -> ThreadState -> DaemonStatusHandle -> ScanRemoteMap -> PairMsg -> [PairingInProgress] -> IO [PairingInProgress]
|
||||
pairAckReceived True (Just pip) st dstatus scanremotes msg cache = do
|
||||
stopSending dstatus pip
|
||||
setupAuthorizedKeys msg
|
||||
finishedPairing st dstatus scanremotes msg (inProgressSshKeyPair pip)
|
||||
startSending dstatus pip PairDone $ multicastPairMsg
|
||||
- and send a single PairDone. -}
|
||||
pairAckReceived :: Bool -> Maybe PairingInProgress -> PairMsg -> [PairingInProgress] -> Assistant [PairingInProgress]
|
||||
pairAckReceived True (Just pip) msg cache = do
|
||||
stopSending pip <<~ daemonStatusHandle
|
||||
liftIO $ setupAuthorizedKeys msg
|
||||
finishedPairing msg (inProgressSshKeyPair pip)
|
||||
dstatus <- getAssistant daemonStatusHandle
|
||||
liftIO $ startSending dstatus pip PairDone $ multicastPairMsg
|
||||
(Just 1) (inProgressSecret pip) (inProgressPairData pip)
|
||||
return $ pip : take 10 cache
|
||||
{- A stale PairAck might also be seen, after we've finished pairing.
|
||||
- Perhaps our PairDone was not received. To handle this, we keep
|
||||
- a cache of recently finished pairings, and re-send PairDone in
|
||||
- response to stale PairAcks for them. -}
|
||||
pairAckReceived _ _ _ dstatus _ msg cache = do
|
||||
pairAckReceived _ _ msg cache = do
|
||||
let pips = filter (verifiedPairMsg msg) cache
|
||||
dstatus <- getAssistant daemonStatusHandle
|
||||
unless (null pips) $
|
||||
forM_ pips $ \pip ->
|
||||
liftIO $ forM_ pips $ \pip ->
|
||||
startSending dstatus pip PairDone $ multicastPairMsg
|
||||
(Just 1) (inProgressSecret pip) (inProgressPairData pip)
|
||||
return cache
|
||||
|
|
@ -151,9 +148,9 @@ pairAckReceived _ _ _ dstatus _ msg cache = do
|
|||
- entering the secret. Would be better to start a fresh pair request in this
|
||||
- situation.
|
||||
-}
|
||||
pairDoneReceived :: Bool -> Maybe PairingInProgress -> ThreadState -> DaemonStatusHandle -> ScanRemoteMap -> PairMsg -> IO ()
|
||||
pairDoneReceived False _ _ _ _ _ = noop -- not verified
|
||||
pairDoneReceived True Nothing _ _ _ _ = noop -- not in progress
|
||||
pairDoneReceived True (Just pip) st dstatus scanremotes msg = do
|
||||
stopSending dstatus pip
|
||||
finishedPairing st dstatus scanremotes msg (inProgressSshKeyPair pip)
|
||||
pairDoneReceived :: Bool -> Maybe PairingInProgress -> PairMsg -> Assistant ()
|
||||
pairDoneReceived False _ _ = noop -- not verified
|
||||
pairDoneReceived True Nothing _ = noop -- not in progress
|
||||
pairDoneReceived True (Just pip) msg = do
|
||||
stopSending pip <<~ daemonStatusHandle
|
||||
finishedPairing msg (inProgressSshKeyPair pip)
|
||||
|
|
|
|||
|
|
@ -8,7 +8,6 @@
|
|||
module Assistant.Threads.Transferrer where
|
||||
|
||||
import Assistant.Common
|
||||
import Assistant.ThreadedMonad
|
||||
import Assistant.DaemonStatus
|
||||
import Assistant.TransferQueue
|
||||
import Assistant.TransferSlots
|
||||
|
|
@ -23,75 +22,78 @@ import Locations.UserConfig
|
|||
|
||||
import System.Process (create_group)
|
||||
|
||||
thisThread :: ThreadName
|
||||
thisThread = "Transferrer"
|
||||
|
||||
{- For now only one transfer is run at a time. -}
|
||||
maxTransfers :: Int
|
||||
maxTransfers = 1
|
||||
|
||||
{- Dispatches transfers from the queue. -}
|
||||
transfererThread :: ThreadState -> DaemonStatusHandle -> TransferQueue -> TransferSlots -> CommitChan -> NamedThread
|
||||
transfererThread st dstatus transferqueue slots commitchan = thread $ liftIO $ go =<< readProgramFile
|
||||
where
|
||||
thread = NamedThread thisThread
|
||||
go program = forever $ inTransferSlot dstatus slots $
|
||||
maybe (return Nothing) (uncurry $ startTransfer st dstatus commitchan program)
|
||||
=<< getNextTransfer transferqueue dstatus notrunning
|
||||
{- Skip transfers that are already running. -}
|
||||
notrunning = isNothing . startedTime
|
||||
transfererThread :: NamedThread
|
||||
transfererThread = NamedThread "Transferr" $ do
|
||||
program <- liftIO readProgramFile
|
||||
transferqueue <- getAssistant transferQueue
|
||||
dstatus <- getAssistant daemonStatusHandle
|
||||
slots <- getAssistant transferSlots
|
||||
starter <- asIO2 $ startTransfer program
|
||||
liftIO $ forever $ inTransferSlot dstatus slots $
|
||||
maybe (return Nothing) (uncurry starter)
|
||||
=<< getNextTransfer transferqueue dstatus notrunning
|
||||
where
|
||||
{- Skip transfers that are already running. -}
|
||||
notrunning = isNothing . startedTime
|
||||
|
||||
{- By the time this is called, the daemonstatus's transfer map should
|
||||
- already have been updated to include the transfer. -}
|
||||
startTransfer :: ThreadState -> DaemonStatusHandle -> CommitChan -> FilePath -> Transfer -> TransferInfo -> TransferGenerator
|
||||
startTransfer st dstatus commitchan program t info = case (transferRemote info, associatedFile info) of
|
||||
(Just remote, Just file) -> ifM (runThreadState st $ shouldTransfer t info)
|
||||
startTransfer :: FilePath -> Transfer -> TransferInfo -> Assistant (Maybe (Transfer, TransferInfo, IO ()))
|
||||
startTransfer program t info = case (transferRemote info, associatedFile info) of
|
||||
(Just remote, Just file) -> ifM (liftAnnex $ shouldTransfer t info)
|
||||
( do
|
||||
brokendebug thisThread [ "Transferring:" , show t ]
|
||||
notifyTransfer dstatus
|
||||
return $ Just (t, info, transferprocess remote file)
|
||||
debug [ "Transferring:" , show t ]
|
||||
notifyTransfer <<~ daemonStatusHandle
|
||||
tp <- asIO2 transferprocess
|
||||
return $ Just (t, info, tp remote file)
|
||||
, do
|
||||
brokendebug thisThread [ "Skipping unnecessary transfer:" , show t ]
|
||||
void $ removeTransfer dstatus t
|
||||
debug [ "Skipping unnecessary transfer:" , show t ]
|
||||
void $ flip removeTransfer t <<~ daemonStatusHandle
|
||||
return Nothing
|
||||
)
|
||||
_ -> return Nothing
|
||||
where
|
||||
direction = transferDirection t
|
||||
isdownload = direction == Download
|
||||
where
|
||||
direction = transferDirection t
|
||||
isdownload = direction == Download
|
||||
|
||||
transferprocess remote file = void $ do
|
||||
(_, _, _, pid)
|
||||
<- createProcess (proc program $ toCommand params)
|
||||
{ create_group = True }
|
||||
{- Alerts are only shown for successful transfers.
|
||||
- Transfers can temporarily fail for many reasons,
|
||||
- so there's no point in bothering the user about
|
||||
- those. The assistant should recover.
|
||||
-
|
||||
- Also, after a successful transfer, the location
|
||||
- log has changed. Indicate that a commit has been
|
||||
- made, in order to queue a push of the git-annex
|
||||
- branch out to remotes that did not participate
|
||||
- in the transfer.
|
||||
-}
|
||||
whenM ((==) ExitSuccess <$> waitForProcess pid) $ do
|
||||
void $ addAlert dstatus $
|
||||
makeAlertFiller True $
|
||||
transferFileAlert direction True file
|
||||
recordCommit commitchan
|
||||
where
|
||||
params =
|
||||
[ Param "transferkey"
|
||||
, Param "--quiet"
|
||||
, Param $ key2file $ transferKey t
|
||||
, Param $ if isdownload
|
||||
then "--from"
|
||||
else "--to"
|
||||
, Param $ Remote.name remote
|
||||
, Param "--file"
|
||||
, File file
|
||||
]
|
||||
transferprocess remote file = void $ do
|
||||
(_, _, _, pid)
|
||||
<- liftIO $ createProcess (proc program $ toCommand params)
|
||||
{ create_group = True }
|
||||
{- Alerts are only shown for successful transfers.
|
||||
- Transfers can temporarily fail for many reasons,
|
||||
- so there's no point in bothering the user about
|
||||
- those. The assistant should recover.
|
||||
-
|
||||
- Also, after a successful transfer, the location
|
||||
- log has changed. Indicate that a commit has been
|
||||
- made, in order to queue a push of the git-annex
|
||||
- branch out to remotes that did not participate
|
||||
- in the transfer.
|
||||
-}
|
||||
whenM (liftIO $ (==) ExitSuccess <$> waitForProcess pid) $ do
|
||||
dstatus <- getAssistant daemonStatusHandle
|
||||
liftIO $ void $ addAlert dstatus $
|
||||
makeAlertFiller True $
|
||||
transferFileAlert direction True file
|
||||
recordCommit <<~ commitChan
|
||||
where
|
||||
params =
|
||||
[ Param "transferkey"
|
||||
, Param "--quiet"
|
||||
, Param $ key2file $ transferKey t
|
||||
, Param $ if isdownload
|
||||
then "--from"
|
||||
else "--to"
|
||||
, Param $ Remote.name remote
|
||||
, Param "--file"
|
||||
, File file
|
||||
]
|
||||
|
||||
{- Checks if the file to download is already present, or the remote
|
||||
- being uploaded to isn't known to have the file. -}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue