net-mqtt-0.8.8.0: test/AuditSpec.hs
{-# LANGUAGE TypeApplications #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}
-- Reproductions for the Nov 2026 net-mqtt audit. Each test asserts the
-- correct behaviour, so a failure is a reproduced bug.
module AuditSpec where
import Control.Concurrent (threadDelay)
import Control.Concurrent.Async (async, wait, waitCatch, cancel)
import Control.Concurrent.STM
import qualified Control.Exception as E
import Control.Monad (forM, replicateM_, when, unless)
import Control.Monad.IO.Class (liftIO)
import qualified Data.Attoparsec.ByteString.Lazy as A
import qualified Data.ByteString as BS
import qualified Data.ByteString.Lazy as BL
import Data.Conduit (awaitForever, yield)
import Data.Either (isLeft)
import Data.Time.Clock (diffUTCTime, getCurrentTime)
import System.Timeout (timeout)
import Test.Tasty.HUnit
import Network.MQTT.Client
import Network.MQTT.Types as MT
data Broker = Broker {
written :: TVar [MQTTPkt] -- newest first, after the CONNECT
, inject :: MQTTPkt -> IO ()
, close :: IO () -- end the connection from the broker side
, connack :: IO () -- send the CONNACK (when held)
, conn :: (MQTTConduit -> IO ()) -> IO ()
}
-- | A fake broker. It records what the client writes (parsed), can inject
-- packets, can close the connection, and optionally closes as soon as
-- the client sends DISCONNECT (as real brokers do).
broker :: Bool -> Bool -> IO Broker
broker holdConnack closeOnDisco = do
w <- newTVarIO []
inj <- newTChanIO
closed <- newTVarIO False
ack <- newTVarIO (not holdConnack)
first <- newTVarIO True
let enc = BL.toStrict . toByteString Protocol50
next = (Just <$> readTChan inj) `orElse` (readTVar closed >>= check >> pure Nothing)
loop = liftIO (atomically next) >>= maybe (pure ()) (\p -> yield (enc p) >> loop)
src = do
liftIO . atomically $ readTVar ack >>= check
yield (enc $ ConnACKPkt (ConnACKFlags NewSession ConnAccepted []))
loop
sink = awaitForever $ \bs -> liftIO $ do
isFirst <- atomically $ readTVar first <* writeTVar first False
unless isFirst $ case A.eitherResult (A.parse (parsePacket Protocol50) (BL.fromStrict bs)) of
Right p -> do
atomically $ modifyTVar' w (p:)
case p of
DisconnectPkt _ | closeOnDisco -> atomically $ writeTVar closed True
_ -> pure ()
Left e -> error e
pure Broker{written=w, inject=atomically . writeTChan inj, close=atomically (writeTVar closed True),
connack=atomically (writeTVar ack True), conn=(\f -> f (src, sink))}
v5 :: MQTTConfig
v5 = mqttConfig{_protocol=Protocol50}
waitWritten :: Broker -> (MQTTPkt -> Bool) -> IO (Maybe MQTTPkt)
waitWritten b f = timeout 3000000 . atomically $ do
ws <- readTVar (written b)
case filter f ws of
(x:_) -> pure x
_ -> retry
-- QoS 0 publishes still reserve a packet ID. After 65535 of them the
-- counter wraps onto the ID of a QoS 1 publish that is still waiting for
-- its PUBACK; the QoS 0 publish replaces, then deletes, the waiter's ack
-- registration, so the real PUBACK is dropped and the publish hangs.
unit_audit_qos0WrapLosesQoS1Ack :: Assertion
unit_audit_qos0WrapLosesQoS1Ack = wrapCase 65535
-- Control: the same with fewer QoS 0 publishes passes.
unit_audit_qos0NoWrapControl :: Assertion
unit_audit_qos0NoWrapControl = wrapCase 1000
wrapCase :: Int -> Assertion
wrapCase n = do
b <- broker False False
c <- runMQTTConduit (conn b) v5
p <- async $ publishq c "a/b" "x" False QoS1 []
Just (PublishPkt PublishRequest{_pubPktID=pid}) <- waitWritten b (\case PublishPkt _ -> True; _ -> False)
replicateM_ n $ publish c "a/b" "y" False
inject b (PubACKPkt (PubACK pid 0 []))
r <- timeout 3000000 (wait p)
stopClient c
assertEqual ("QoS 1 publish (pid " <> show pid <> ") completes after its PUBACK") (Just ()) r
-- waitForClient after stopClient should return normally (the state is
-- Stopped), but it can see the transient Disconnected written by markDisco
-- before stopClient writes Stopped.
unit_audit_waitForClientStopRace :: Assertion
unit_audit_waitForClientStopRace = do
rs <- forM [1 .. 300 :: Int] $ \_ -> do
b <- broker False False
c <- runMQTTConduit (conn b) v5
w <- async (waitForClient c)
threadDelay 1000
stopClient c
either (Just . show) (const Nothing) <$> waitCatch w
let bad = [e | Just e <- rs]
assertEqual ("waitForClient threw after stopClient (" <> show (take 1 bad) <> ")") 0 (length bad)
-- Same race with normalDisconnect against a broker that closes on DISCONNECT.
unit_audit_waitForClientDisconnectRace :: Assertion
unit_audit_waitForClientDisconnectRace = do
rs <- forM [1 .. 300 :: Int] $ \_ -> do
b <- broker False True
c <- runMQTTConduit (conn b) v5
w <- async (waitForClient c)
threadDelay 1000
normalDisconnect c
either (Just . show) (const Nothing) <$> waitCatch w
let bad = [e | Just e <- rs]
assertEqual ("waitForClient threw after normalDisconnect (" <> show (take 1 bad) <> ")") 0 (length bad)
-- After sending DISCONNECT the client must close the connection itself
-- [MQTT-3.14.4-1]. net-mqtt waits for the server to close it, so against a
-- broker that waits for the client, disconnect takes 10s and waitForClient
-- reports a Timeout.
unit_audit_disconnectDoesNotClose :: Assertion
unit_audit_disconnectDoesNotClose = do
b <- broker False False
c <- runMQTTConduit (conn b) v5
w <- async (waitForClient c)
t0 <- getCurrentTime
normalDisconnect c
t1 <- getCurrentTime
r <- waitCatch w
assertBool ("normalDisconnect took " <> show (diffUTCTime t1 t0) <> ", waitForClient: " <> show r)
(diffUTCTime t1 t0 < 2 && not (isLeft r))
-- connectURI wraps the connect in a timeout, but the client thread is a
-- separate async that the timeout doesn't cancel. If the CONNACK shows up
-- after the timeout, the "failed" client connects and keeps pinging.
unit_audit_connectTimeoutLeaksClient :: Assertion
unit_audit_connectTimeoutLeaksClient = do
b <- broker True False
r <- timeout 300000 (runMQTTConduit (conn b) v5{_pingPeriod=100000})
assertBool "connect timed out" (null' r)
connack b
threadDelay 1000000
pings <- length . filter (== PingPkt) <$> readTVarIO (written b)
assertEqual "PINGREQs from a client whose connect timed out" 0 pings
where null' = maybe True (const False)
-- QoS 2 inbound messages are only delivered on PUBREL. The PUBREC already
-- told the broker the client owns the message [MQTT-4.3.3-9], so if the
-- connection drops in between, the broker resends only PUBREL on the
-- resumed session, the new client answers PUBCOMP 0x92, and the message is
-- never delivered.
unit_audit_qos2LostAcrossReconnect :: Assertion
unit_audit_qos2LostAcrossReconnect = do
got <- newTVarIO (0 :: Int)
let cfg = v5{Network.MQTT.Client._cleanSession=False, _msgCB=SimpleCallback (\_ _ _ _ -> atomically $ modifyTVar' got succ)}
b1 <- broker False False
c1 <- runMQTTConduit (conn b1) cfg
inject b1 (PublishPkt PublishRequest{_pubDup=False, _pubQoS=QoS2, _pubRetain=False, _pubTopic="a/b",
_pubPktID=7, _pubBody="x", _pubProps=[]})
Just _ <- waitWritten b1 (\case PubRECPkt _ -> True; _ -> False)
close b1
_ <- E.try @E.SomeException (waitForClient c1)
b2 <- broker False False
c2 <- runMQTTConduit (conn b2) cfg
inject b2 (PubRELPkt (PubREL 7 0 []))
Just comp <- waitWritten b2 (\case PubCOMPPkt _ -> True; _ -> False)
threadDelay 200000
n <- readTVarIO got
stopClient c2
assertEqual ("deliveries of the QoS 2 message (client sent " <> show comp <> ")") 1 n
-- Messages already received and queued for an ordered callback are thrown
-- away when the connection ends (waitForClient cancels the callback thread).
unit_audit_orderedQueueDroppedOnDisconnect :: Assertion
unit_audit_orderedQueueDroppedOnDisconnect = do
release <- newTVarIO False
got <- newTVarIO (0 :: Int)
b <- broker False False
c <- runMQTTConduit (conn b) v5{_msgCB=OrderedCallback (\_ _ _ _ -> do
atomically $ readTVar release >>= check
atomically $ modifyTVar' got succ)}
replicateM_ 3 $ inject b (PublishPkt PublishRequest{_pubDup=False, _pubQoS=QoS0, _pubRetain=False, _pubTopic="a/b",
_pubPktID=0, _pubBody="x", _pubProps=[]})
threadDelay 100000
close b
_ <- E.try @E.SomeException (waitForClient c)
atomically $ writeTVar release True
threadDelay 300000
n <- readTVarIO got
assertEqual "QoS 0 messages received before the close that reached the callback" 3 n