diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2017 Confer Health, Inc. + +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 Confer Health nor the names of other + 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 +OWNER 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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/apns-http2.cabal b/apns-http2.cabal new file mode 100644 --- /dev/null +++ b/apns-http2.cabal @@ -0,0 +1,91 @@ +-- This file has been generated from package.yaml by hpack version 0.18.1. +-- +-- see: https://github.com/sol/hpack + +name: apns-http2 +version: 0.1.0.0 +synopsis: Apple Push Notification service HTTP/2 integration. +category: Network +homepage: https://github.com/ConferHealth/apns-http2#readme +author: Confer Health, Inc. +maintainer: oss@confer.health +copyright: 2017 Confer Health, Inc. +license: BSD3 +license-file: LICENSE +build-type: Simple +cabal-version: >= 1.10 + +library + hs-source-dirs: + src + default-extensions: FlexibleContexts FlexibleInstances LambdaCase NamedFieldPuns NoMonomorphismRestriction OverloadedStrings QuasiQuotes RankNTypes RecordWildCards ScopedTypeVariables TemplateHaskell TupleSections ViewPatterns + ghc-options: -Wall -O2 + build-depends: + base >= 4.7 && < 5 + , aeson + , async + , bytestring + , conduit + , conduit-extra + , containers + , data-default + , http2 + , keys + , lens + , lifted-base + , mmorph + , mtl + , network + , stm + , stm-chans + , stm-conduit + , text + , time + , tls + , x509 + , x509-store + , x509-validation + exposed-modules: + Network.Apns + Network.Apns.Internal + Network.Apns.Types + other-modules: + Paths_apns_http2 + default-language: Haskell2010 + +executable apns-http2-example + main-is: Main.hs + hs-source-dirs: + example + default-extensions: FlexibleContexts FlexibleInstances LambdaCase NamedFieldPuns NoMonomorphismRestriction OverloadedStrings QuasiQuotes RankNTypes RecordWildCards ScopedTypeVariables TemplateHaskell TupleSections ViewPatterns + ghc-options: -Wall -O2 -threaded -rtsopts -with-rtsopts=-N + build-depends: + base >= 4.7 && < 5 + , aeson + , async + , bytestring + , conduit + , conduit-extra + , containers + , data-default + , http2 + , keys + , lens + , lifted-base + , mmorph + , mtl + , network + , stm + , stm-chans + , stm-conduit + , text + , time + , tls + , x509 + , x509-store + , x509-validation + , aeson-qq + , apns-http2 + , base16-bytestring + , x509-system + default-language: Haskell2010 diff --git a/example/Main.hs b/example/Main.hs new file mode 100644 --- /dev/null +++ b/example/Main.hs @@ -0,0 +1,77 @@ +module Main where + +import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar) +import Data.Aeson (encode) +import Data.Aeson.QQ (aesonQQ) +import qualified Data.ByteString as BS +import qualified Data.ByteString.Base16 as Base16 +import qualified Data.ByteString.Char8 as BSC8 +import qualified Data.ByteString.Lazy as BL +import Data.Monoid ((<>)) +import qualified Data.Text as T +import qualified Data.Text.IO as TIO +import Data.Time.Clock (addUTCTime, getCurrentTime) +import Network.Apns +import Network.TLS (credentialLoadX509Chain) +import System.Environment (getArgs) +import System.X509 (getSystemCertificateStore) +import Text.Read (readMaybe) + +main :: IO () +main = do + (host, port, certPath, keyPath, deviceToken, topic) <- getArgs >>= \ case + [h, readMaybe -> Just p, cp, kp, Base16.decode . BSC8.pack -> (dt, dt'), T.pack -> t] + | BS.null dt' -> pure (h, p, cp, kp, dt, t) + | otherwise -> fail "invalid device token" + _ -> fail "usage: apns-http2-example " + + (clientCert, clientKey) <- either fail pure =<< credentialLoadX509Chain certPath [] keyPath + + terminationMv <- newEmptyMVar + systemCAStore <- getSystemCertificateStore + let params = ApnsConnectionParams + { _apnsConnectionParams_hostName = host + , _apnsConnectionParams_portNumber = port + , _apnsConnectionParams_clientCertificateChain = clientCert + , _apnsConnectionParams_clientCertificateKey = clientKey + , _apnsConnectionParams_serverCertificateStore = Just systemCAStore + , _apnsConnectionParams_onDebugLog = TIO.putStrLn . ("Network.Apns debug: " <>) + , _apnsConnectionParams_onServerCertificate = Nothing + -- Uncomment following line and comment previous one to disable server cert verification + -- , _apnsConnectionParams_onServerCertificate = Just $ \ _ _ _ _ -> pure [] + , _apnsConnectionParams_onTermination = putMVar terminationMv + , _apnsConnectionParams_readQueueSize = 10 + , _apnsConnectionParams_writeQueueSize = 10 + , _apnsConnectionParams_pushQueueSize = 10 + } + + connectApns params >>= \ case + Left connectError -> + putStrLn $ "Failed to connect: " <> show connectError + Right connection -> do + now <- getCurrentTime + let _apnsPushData_deviceToken = deviceToken + _apnsPushData_expiration = addUTCTime (86400*2) now + _apnsPushData_priority = Nothing + _apnsPushData_topic = topic + _apnsPushData_collapseId = Nothing + _apnsPushData_payload = BL.toStrict $ encode [aesonQQ| + { + "aps": { + "alert": { + "title": "Hello, APNs!", + "body": "If you're reading this you're too close to a working integration" + }, + "badge": 1 + } + } + |] + + _apnsConnection_submitPush connection $ ApnsPush (ApnsPushData {..}) $ \ res -> do + putStrLn . ("push result: " <>) . show $ res + putStrLn "closing connection" + _apnsConnection_close connection + + terminationReason <- takeMVar terminationMv + putStrLn . ("Connection terminated: " <>) . show $ terminationReason + diff --git a/src/Network/Apns.hs b/src/Network/Apns.hs new file mode 100644 --- /dev/null +++ b/src/Network/Apns.hs @@ -0,0 +1,8 @@ +-- |APNs integration using the HTTP/2 protocol rather than the legacy binary interface. +module Network.Apns + ( module Network.Apns.Internal + , module Network.Apns.Types + ) where + +import Network.Apns.Internal (connectApns) +import Network.Apns.Types diff --git a/src/Network/Apns/Internal.hs b/src/Network/Apns/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Network/Apns/Internal.hs @@ -0,0 +1,810 @@ +module Network.Apns.Internal where + +import Control.Applicative ((<|>)) +import Control.Concurrent.Async (async, cancel, waitCatchSTM) +import Control.Concurrent.Lifted (fork) +import Control.Concurrent.STM (STM, atomically, retry) +import Control.Concurrent.STM.TBMQueue (TBMQueue, closeTBMQueue, isClosedTBMQueue, newTBMQueueIO, readTBMQueue, writeTBMQueue) +import Control.Concurrent.STM.TVar (TVar, newTVarIO, readTVar, writeTVar) +import Control.Exception (IOException, SomeException) +import Control.Exception.Lifted (bracketOnError, handle, throwIO, try) +import Control.Monad ((<=<), guard, join, unless, when) +import Control.Monad.Error.Class (throwError) +import Control.Monad.Except (ExceptT(ExceptT), runExceptT) +import Control.Monad.IO.Class (MonadIO, liftIO) +import Control.Monad.Morph (hoist) +import Control.Monad.Reader (ReaderT, runReaderT) +import Control.Monad.Reader.Class (MonadReader) +import Control.Monad.State.Strict (StateT, runStateT) +import Control.Monad.Trans (lift) +import Control.Lens + ( _Cons, _Just, _Snoc, (<&>), assign, at, both, each, filtered, folded, ifolded, modifying, over, preview, set, sumOf, to, toListOf, use, uses, withIndex, view ) +import Control.Lens.TH (makeLenses, makePrisms) +import Data.Aeson ((.:), (.:?), Value, eitherDecode', withObject) +import Data.Aeson.Types (Parser, parseEither) +import Data.Bifunctor (first) +import Data.ByteString (ByteString) +import qualified Data.ByteString as BS +import Data.ByteString.Builder (byteStringHex, toLazyByteString) +import qualified Data.ByteString.Char8 as BSC8 +import qualified Data.ByteString.Lazy as BL +import Data.Conduit (ConduitM, Producer, (.|), ($$), awaitForever, yield) +import qualified Data.Conduit.Binary as CB +import qualified Data.Conduit.List as CL +import Data.Conduit.TQueue (sinkTBMQueue, sourceTBMQueue) +import Data.Default (def) +import Data.Foldable (for_, traverse_) +import Data.Functor (($>), void) +import Data.IntMap.Strict (IntMap) +import qualified Data.IntMap.Strict as IntMap +import Data.Key (traverseWithKey_) +import Data.Maybe (catMaybes, fromMaybe, isJust) +import Data.Monoid ((<>)) +import Data.Text (Text, pack) +import Data.Text.Encoding (encodeUtf8) +import Data.Time.Clock (NominalDiffTime, UTCTime) +import Data.Time.Clock.POSIX (posixSecondsToUTCTime, utcTimeToPOSIXSeconds) +import Data.Word (Word32) +import Data.X509.Validation (validateDefault) +import Network.Apns.Types + ( ApnsConnection(ApnsConnection) + , ApnsConnectionError(..) + , ApnsConnectionParams(..), apnsConnectionParams_hostName, apnsConnectionParams_portNumber, apnsConnectionParams_onDebugLog + , apnsConnectionParams_pushQueueSize, apnsConnectionParams_readQueueSize, apnsConnectionParams_writeQueueSize, apnsConnectionParams_onTermination + , ApnsError(ApnsError) + , ApnsErrorReason, apnsErrorReasonFromText + , ApnsPush, _apnsPush_callback, apnsPush_data + , ApnsPushData(ApnsPushData), _apnsPushData_collapseId, _apnsPushData_deviceToken, _apnsPushData_expiration, apnsPushData_payload, _apnsPushData_priority + , _apnsPushData_topic + , ApnsPushResult(..) + , ApnsTerminationReason(..) + ) +import Network.BSD (getProtocolNumber) +import qualified Network.HPACK as HP +import qualified Network.HTTP2 as H2 +import Network.Socket (Socket) +import qualified Network.Socket as Socket +import qualified Network.TLS as TLS +import Network.TLS.Extra.Cipher (ciphersuite_strong) +import Text.Read (readMaybe) + + +-- |Event emitted by the reader thread. +data ReaderEvent + = ReadFrame H2.Frame + -- ^Reader read a well formed and not obviously wrong HTTP2 frame which is not a SETTINGS frame. + | ReadSettingsUpdate H2.Settings + -- ^Reader encountered a SETTINGS frame and updated its internal settings. The SETTINGS frame still needs to be adopted by the processor and acknowledged. + | ReadH2Error H2.ErrorCodeId ByteString ApnsTerminationReason + -- ^Reader failed to process incoming data due to some catastrophic protocol error. + deriving Show + +-- |Environmental context for the processing system, passed around as a 'ReaderT'. +data Context = Context + { _context_params :: ApnsConnectionParams + -- ^The 'ApnsConnectionParams' used to open the connection. + , _context_authority :: ByteString + -- ^The cached HTTP authority value sent with each request. + , _context_closingTv :: TVar (Maybe ApnsTerminationReason) + -- ^'TVar' that is @Just@ when the connection is being closed due to local or remote action but not all streams have been processed yet. + , _context_pushQueue :: TBMQueue ApnsPush + -- ^The 'TBMQueue' that '_apnsConnection_submitPush' enqueues using. If this queue is closed, then the connection is closed and can't deliver any more + -- push notifications. + , _context_read :: STM (Either ApnsTerminationReason ReaderEvent) + -- ^STM action which dequeues either a reader failure or reader event from the read queue. + , _context_write :: H2.EncodeInfo -> H2.FramePayload -> STM (Either ApnsTerminationReason ()) + -- ^STM action which enqueues an HTTP/2 frame to write. + , _context_encodingTable :: HP.DynamicTable + -- ^The HPACK dynamic table for encoding. + , _context_decodingTable :: HP.DynamicTable + -- ^The HPACK dynamic table for decoding. + } + +-- |The dynamic state of the processing system, passed around as a 'StateT'. +data State = State + { _state_nextStreamId :: H2.StreamId + -- ^The next stream ID to assign to a new push. + , _state_settings :: H2.Settings + -- ^The current settings as accepted and acknowledged by the processor. + , _state_connectionWindow :: H2.WindowSize + -- ^How many DATA payload bytes can be written to the connection for any stream before a WINDOW_UPDATE frame from the server is required. + , _state_writableStreams :: IntMap WritableStream + -- ^Streams that are currently writing out their payload data but have had their headers already sent along. Streams will be written out as quickly as + -- flow-control windows allow. + , _state_readableStreams :: IntMap (ApnsPush, ReadableStream) + -- ^Streams whose request bodies have been fully transmitted via DATA frames and now are waiting for server frames to complete the response. + } + +-- |Data associated with a single push stream which is in the process of being written to the server as DATA frames. Streams are written as fast as +-- flow-control windows allow, so this structure is mostly concerned with tracking that. +data WritableStream = WritableStream + { _writableStream_push :: ApnsPush + -- ^The 'ApnsPush' that this stream is associated with. + , _writableStream_windowSize :: H2.WindowSize + -- ^The remaining bytes of payload that can be sent for this stream, assuming the connection flow-control window allows for it. + -- Each DATA frame that goes out will decrement this by the size of its payload. + , _writableStream_remaining :: ByteString + -- ^The request body content still to be written. + -- Each DATA frame that goes out will drop the sent bytes from the front of this bytestring. + } + +-- |Data associated with a stream which the processor is waiting for server frames to move along. +data ReadableStream + = StreamWaiting + -- ^So far nothing has been received from the server for the stream. + | StreamReadingHeaderContinuation BL.ByteString + -- ^A HEADERS frame without the END_HEADERS or END_STREAM flags was received, so the processor is expecting one or more CONTINUATION frames. + | StreamReadingBody HP.HeaderList BL.ByteString + -- ^A HEADERS/CONTINUATION with END_HEADERS set but END_STREAM not set was received, so the processor is expecting zero or more DATA frames constituting + -- the response body. + | StreamReadingTrailerContinuation HP.HeaderList BL.ByteString BL.ByteString + -- ^A HEADERS frame after the initial HEADERS frame and zero or more DATA frames was received with the END_STREAM flag but not the END_HEADERS flag, so the + -- processor is expecting one or more CONTINUATION frames, the last of which will finish the stream. + +makeLenses ''Context +makeLenses ''State +makeLenses ''WritableStream +makePrisms ''ReadableStream + +-- |The monad stack used within the processing system. +type ProcessorM = ExceptT ApnsTerminationReason (StateT State (ReaderT Context IO)) + +-- |Emit some debug log message. +processorDebug :: (MonadReader Context m, MonadIO m) => Text -> m () +processorDebug t = do + f <- view (context_params . apnsConnectionParams_onDebugLog) + liftIO $ f t + +-- |Enqueue a HTTP/2 frame to be written. Not appropriate for HEADERS/CONTINUATION as there's no guarantee of atomicity for a series of writes. +processorWrite :: H2.EncodeInfo -> H2.FramePayload -> ProcessorM () +processorWrite ei p = do + f <- view context_write + hoist (liftIO . atomically) . ExceptT $ f ei p + +-- |Enqueue many frames to be Written as one contiguous block with no gaps, e.g. as for HEADERS/CONTINUATION sequences. +processorWriteMany :: [(H2.EncodeInfo, H2.FramePayload)] -> ProcessorM () +processorWriteMany frames = do + f <- view context_write + hoist (liftIO . atomically) $ traverse_ (ExceptT . uncurry f) $ frames + +-- |Invoke some callback function in IO where we don't care at all about exceptions. +invokeWithoutException :: MonadIO m => IO () -> m () +invokeWithoutException action = liftIO $ handle (\ (_ :: SomeException) -> pure ()) action + +-- |Establish a single connection to APNs, yielding @'Right' 'ApnsConnection'@ on success, @'Left' 'ApnsConnectionError'@ otherwise. +connectApns :: forall m. MonadIO m => ApnsConnectionParams -> m (Either ApnsConnectionError ApnsConnection) +connectApns params@(ApnsConnectionParams {..}) = + liftIO . runExceptT $ + bracketOnError connectSocket (liftIO . Socket.close) $ \ sock -> do + tlsContext <- liftIO $ TLS.contextNew sock tlsClientParams + liftIO $ TLS.handshake tlsContext + liftIO $ TLS.sendData tlsContext (BL.fromStrict H2.connectionPreface) + + ExceptT $ processor params tlsContext sock + + where + onCertificateRequest _ = do + pure . Just $ + ( _apnsConnectionParams_clientCertificateChain + , _apnsConnectionParams_clientCertificateKey + ) + + tlsClientParams = + (TLS.defaultParamsClient _apnsConnectionParams_hostName (BSC8.pack . show $ _apnsConnectionParams_portNumber)) + { TLS.clientSupported = def + { TLS.supportedVersions = [TLS.TLS12] + , TLS.supportedCiphers = ciphersuite_strong + } + , TLS.clientHooks = def + { TLS.onCertificateRequest = onCertificateRequest + , TLS.onServerCertificate = fromMaybe validateDefault _apnsConnectionParams_onServerCertificate + , TLS.onSuggestALPN = pure (Just ["h2"]) + } + , TLS.clientShared = def + { TLS.sharedCAStore = fromMaybe mempty _apnsConnectionParams_serverCertificateStore + } + } + + connectSocket :: ExceptT ApnsConnectionError IO Socket + connectSocket = do + tcp <- liftIO $ getProtocolNumber "tcp" + let hints = Socket.defaultHints + { Socket.addrFlags = [Socket.AI_ADDRCONFIG] + , Socket.addrProtocol = tcp + , Socket.addrSocketType = Socket.Stream + } + addressInfos <- liftIO $ Socket.getAddrInfo (Just hints) (Just _apnsConnectionParams_hostName) (Just . show $ _apnsConnectionParams_portNumber) + + when (null addressInfos) $ + throwError $ ApnsConnectionErrorInvalidHostName _apnsConnectionParams_hostName + + let go :: [IOException] -> [Socket.AddrInfo] -> ExceptT ApnsConnectionError IO Socket + go exs [] = throwError $ ApnsConnectionErrorConnectFailed _apnsConnectionParams_hostName exs + go exs (ai:ais) = do + let createSocket = liftIO $ Socket.socket (Socket.addrFamily ai) (Socket.addrSocketType ai) (Socket.addrProtocol ai) + res <- liftIO . try $ bracketOnError createSocket Socket.close $ \ sock -> + Socket.connect sock (Socket.addrAddress ai) $> sock + case res of + Left ex -> go (ex:exs) ais + Right sock -> pure sock + + go [] addressInfos + +-- |Main processing for an APNs connection which sets up the reading and writing threads for exchanging HTTP/2 frames over an established TLS connection, +-- finishes the HTTP/2 connection initiation, and then forks off the asynchronous processing. +processor :: ApnsConnectionParams -> TLS.Context -> Socket -> IO (Either ApnsConnectionError ApnsConnection) +processor params tlsContext sock = do + let debugLog :: MonadIO n => Text -> n () + debugLog = liftIO . view apnsConnectionParams_onDebugLog params + + readerQueue :: TBMQueue ReaderEvent <- liftIO . newTBMQueueIO $ view apnsConnectionParams_readQueueSize params + writerQueue :: TBMQueue (H2.EncodeInfo, H2.FramePayload) <- liftIO . newTBMQueueIO $ view apnsConnectionParams_writeQueueSize params + + debugLog "Starting reader and writer threads" + readerAsync <- liftIO . async $ reader (view apnsConnectionParams_onDebugLog params) tlsContext sock readerQueue + writerAsync <- liftIO . async $ writer (view apnsConnectionParams_onDebugLog params) tlsContext sock writerQueue + + let cleanUpThreads = do + cancel readerAsync + cancel writerAsync + + liftIO . atomically $ writeTBMQueue writerQueue (H2.encodeInfo id 0, H2.SettingsFrame [(H2.SettingsEnablePush, 0)]) + + debugLog "Waiting for first reader event" + connectResult <- try $ do + firstReaderEvent <- liftIO . atomically $ readTBMQueue readerQueue + + case firstReaderEvent of + Just (ReadSettingsUpdate settings0) -> do + debugLog "First read event is a valid SETTINGS frame" + pure $ Right settings0 + + Just other -> do + debugLog $ "First read event is not a valid SETTINGS frame: " <> (pack . show) other + pure $ Left ApnsConnectionErrorInvalidServerPreface + + Nothing -> do + debugLog "EOF when reading first frame" + pure $ Left ApnsConnectionErrorInvalidServerPreface + + pushQueue <- liftIO . newTBMQueueIO $ view apnsConnectionParams_pushQueueSize params + closingTv <- liftIO . newTVarIO $ Nothing + + case connectResult of + Left ex -> do + liftIO cleanUpThreads + pure . Left $ ApnsConnectionErrorException ex + + Right (Left err) -> do + liftIO cleanUpThreads + pure . Left $ err + + Right (Right settings0) -> do + void . fork $ do + encodingTable <- liftIO $ HP.newDynamicTableForEncoding HP.defaultDynamicTableSize + decodingTable <- liftIO $ HP.newDynamicTableForDecoding HP.defaultDynamicTableSize 131072 + let readReaderQueue = + readTBMQueue readerQueue >>= \ case + Just a -> pure $ Right a + Nothing -> Left . either ApnsTerminatedReadError (const $ ApnsTerminatedRemotely (Left "unknown cause")) <$> waitCatchSTM readerAsync + writeWriterQueue encodeInfo framePayload = do + isClosedTBMQueue writerQueue >>= \ case + False -> Right <$> writeTBMQueue writerQueue (encodeInfo, framePayload) + True -> Left . either ApnsTerminatedWriteError (const $ ApnsTerminatedRemotely (Left "unknown cause")) <$> waitCatchSTM writerAsync + + authority = BSC8.pack $ view apnsConnectionParams_hostName params <> ":" <> show (view apnsConnectionParams_portNumber params) + + context = Context params authority closingTv pushQueue readReaderQueue writeWriterQueue encodingTable decodingTable + state0 = State 1 settings0 H2.defaultInitialWindowSize mempty mempty + + dropStreams state = do + let droppedWritableStreams = view state_writableStreams state + droppedReadableStreams = view state_readableStreams state + + dropped :: (v -> ApnsPush) -> H2.StreamId -> v -> IO () + dropped f droppedSid v = do + debugLog $ "stream " <> (pack . show) droppedSid <> " dropped at end" + invokeWithoutException . _apnsPush_callback (f v) $ ApnsPushDropped + + traverseWithKey_ (dropped _writableStream_push) droppedWritableStreams + traverseWithKey_ (dropped fst ) droppedReadableStreams + + + -- enter main processing loop which maintains the connection and eventually exits with some termination reason. Alternates between a write phase + -- where it tries to write out all stream data it's allowed to and a post-write phase where it waits for incoming frames from the server or new work + -- to process. + (try . liftIO . flip runReaderT context . flip runStateT state0 . runExceptT $ processUpdatedSettings settings0 >> postWritePhase False) >>= \ case + Left ex -> do + invokeWithoutException . view apnsConnectionParams_onTermination params $ ApnsTerminatedProcessingException ex + + Right (Left reason, state') -> do + dropStreams state' + invokeWithoutException . view apnsConnectionParams_onTermination params $ reason + + Right (Right (), state') -> do + dropStreams state' + invokeWithoutException . view apnsConnectionParams_onTermination params $ ApnsTerminatedLocally + + liftIO . atomically $ closeTBMQueue pushQueue + liftIO $ cleanUpThreads + + let submitPush push = + join . liftIO . atomically $ + (||) <$> isClosedTBMQueue pushQueue <*> (isJust <$> readTVar closingTv) >>= \ case + False -> pure () <$ writeTBMQueue pushQueue push + True -> pure (invokeWithoutException $ _apnsPush_callback push ApnsPushConnectionClosed) + + close = liftIO . atomically $ do + writeTVar closingTv $ Just ApnsTerminatedLocally + writeTBMQueue writerQueue (H2.encodeInfo id 0, H2.GoAwayFrame 0 H2.NoError "bye!") + + pure . Right $ ApnsConnection submitPush close + +-- |Perform the write phase, writing as many DATA frames as the flow-control windows will allow us +writePhase :: ProcessorM () +writePhase = do + go <=< uses state_writableStreams $ toListOf (ifolded . filtered ((> 0) . view writableStream_windowSize) . withIndex) + postWritePhase False + + where + -- process a list of writable streams all of which have at least 1 byte available in the flow control window, updating the connection flow control window + -- as we go and stopping as soon as it's exhausted. + go :: [(H2.StreamId, WritableStream)] -> ProcessorM () + go [] = pure () + go ((sid, ws) : wses) = do + maxFrameSize <- uses state_settings H2.maxFrameSize + connWindow <- use state_connectionWindow + processorDebug $ "entering write for " <> (pack . show) sid <> " with connection window " <> (pack . show) connWindow <> " and max frame size " <> (pack . show) maxFrameSize + unless (connWindow <= 0) $ do + let toWriteSize = min maxFrameSize . min connWindow . view writableStream_windowSize $ ws + (toWriteBs, remainingBs) = BS.splitAt toWriteSize . view writableStream_remaining $ ws + + if BS.null remainingBs + then do + processorDebug $ "stream " <> (pack . show) sid <> " finished writing" + processorWrite (H2.encodeInfo H2.setEndStream sid) (H2.DataFrame toWriteBs) + assign (state_writableStreams . at sid) Nothing + assign (state_readableStreams . at sid) (Just (view writableStream_push ws, StreamWaiting)) + else do + processorDebug $ "stream " <> (pack . show) sid <> " writing " <> (pack . show) toWriteSize <> " bytes with " <> (pack . show) (BS.length remainingBs) <> " bytes to go" + processorWrite (H2.encodeInfo id sid) (H2.DataFrame toWriteBs) + modifying (state_writableStreams . at sid . _Just) + $ over writableStream_windowSize (subtract toWriteSize) + . set writableStream_remaining remainingBs + + modifying state_connectionWindow (subtract toWriteSize) + + go wses + +-- |Perform the post-write phase, where incoming messages, work, and other state changes are processed prior to returning to the write phase +-- +-- at this point one of the following conditions is true: +-- the connection flow control window has been exhausted, +-- all stream flow-control windows have been exhausted, +-- all streams have been written out and are waiting for a server response, +-- or there are no pending pushes +-- so now we wait for one of the following things to happen: +-- an incoming SETTINGS frame, which we need to adopt and might change the initial window size and thus all flow-control windows +-- an incoming WINDOW_UPDATE frame, which expands one of the flow-control windows +-- a PING frame which we respond to immediately +-- a PRIORITY frame which we ignore +-- a PUSH_PROMISE frame which we complain about mightily +-- a HEADERS/CONTINUATION/DATA/RST_STREAM frame which advances the read state of some stream +-- a GOAWAY frame which causes us to go into the closing state and signal any dropped pushes as dropped +-- +-- The Bool argument indicates whether this 'postWritePhase' follows another post write phase which either adjusted flow-control windows or added streams +-- and thus controls whether this should block waiting for more to do or can immediately enter a write phase as soon as other work has been handled. +postWritePhase :: Bool -> ProcessorM () +postWritePhase haveUpdatedWindowsOrStreams = do + activeStreams <- (+) + <$> uses state_writableStreams length + <*> uses state_readableStreams length + + canMakeMoreStreams <- uses state_settings (maybe True (activeStreams <) . H2.maxConcurrentStreams) + + processorDebug $ "post write phase with " <> (pack . show) activeStreams <> " active streams. can make more streams? " <> (pack . show) canMakeMoreStreams + + closingTv <- view context_closingTv + readFunc <- view context_read + pushQueue <- view context_pushQueue + + -- in an STM transaction figure out what to do based on the reader queue, push queue, closing state, and so on, + -- choosing an outside of STM action to perform. + -- + -- this is hella finicky, so a couple of notes: + -- each branch is <|>ed with each other, and will fall through to the next case if a guard fails or there's nothing in a queue + -- each branch is of type STM (Bool -> StateT State IO ApnsTerminationReason) + toDo :: ProcessorM () <- liftIO . atomically $ + readTVar closingTv >>= \ closingMay -> + -- if there's a frame received to process or no more frames can be received due to reader failure, deal with that + ( ( readFunc <&> \ case + Left reason -> throwError reason + Right event -> processReaderEvent haveUpdatedWindowsOrStreams event ) + -- if we're in the closing state and there are no more active streams, then we're done + <|> ( case closingMay of + Just reason -> do + guard $ activeStreams == 0 + pure $ throwError reason + Nothing -> retry ) + -- if we're not in the closing state and we've not hit the stream limit, then process a new push that might have arrived + <|> ( do guard $ not (isJust closingMay) && canMakeMoreStreams + maybe retry (pure . processNewPush haveUpdatedWindowsOrStreams) =<< readTBMQueue pushQueue ) + -- if there are active streams and we've updated the flow-control windows or streams since the last write phase, then don't block and instead + -- do a write phase + <|> ( do guard (activeStreams > 0 && haveUpdatedWindowsOrStreams) + pure writePhase ) + ) + + toDo + +-- |Process a reader event in the 'postWritePhase'. This boils down to propagating a failure if the reader failed, adopting and acknowleding new settings if +-- the reader received and processed a SETTINGS frame, or case switching on the received frame and doing various state things based on the frame. +-- +-- The Bool is the 'haveUpdatedWindowsOrStreams' flag propagating for this 'postWritePhase' +processReaderEvent :: Bool -> ReaderEvent -> ProcessorM () +processReaderEvent haveUpdatedWindowsOrStreams = \ case + ReadH2Error errorCodeId errorData abortReason -> do + processorDebug $ "reader protocol error " <> (pack . show) errorCodeId <> " " <> (pack . show) errorData <> " " <> (pack . show) abortReason + nextStreamId <- use state_nextStreamId + processorWrite (H2.encodeInfo id 0) (H2.GoAwayFrame (pred nextStreamId) errorCodeId errorData) + throwError abortReason + + ReadFrame frame -> + processReadFrame haveUpdatedWindowsOrStreams frame + + ReadSettingsUpdate settings' -> do + windowSizeChanged <- processUpdatedSettings settings' + postWritePhase $ haveUpdatedWindowsOrStreams || windowSizeChanged + +-- |Do the work required when receiving new settings from the server, such as acknowledging the update. This is a separate function since it's used both +-- by the preface and the receive frame processing. Yields True iff the window sizes were updated. +processUpdatedSettings :: H2.Settings -> ProcessorM Bool +processUpdatedSettings settings' = do + processorDebug $ "read new settings " <> (pack . show) settings' + settings <- use state_settings + assign state_settings settings' + + when (H2.headerTableSize settings /= H2.headerTableSize settings') $ do + processorDebug " - updating encoding table size" + encodingTable <- view context_encodingTable + liftIO $ HP.setLimitForEncoding (H2.headerTableSize settings') encodingTable + + let windowSizeChanged = H2.initialWindowSize settings /= H2.initialWindowSize settings' + when windowSizeChanged $ do + let delta = H2.initialWindowSize settings' - H2.initialWindowSize settings + processorDebug $ " - adjusting all stream windows by " <> (pack . show) delta + modifying (state_writableStreams . each . writableStream_windowSize) (+ delta) + + processorDebug $ " - acknowledging" + processorWrite (H2.encodeInfo H2.setAck 0) (H2.SettingsFrame []) + + pure windowSizeChanged + +-- |Process an incoming frame which is not a SETTINGS frame. +processReadFrame :: Bool -> H2.Frame -> ProcessorM () +processReadFrame haveUpdatedWindowsOrStreams (H2.Frame (H2.FrameHeader _ flags sid) payload) = + case payload of + H2.HeadersFrame _ fragment -> do + withReadableStream $ \ push -> \ case + StreamWaiting -> + case (H2.testEndHeader flags, H2.testEndStream flags) of + (True, True) -> decodeHeaders fragment >>= \ trailers -> processEndStream push trailers "" + (True, False) -> decodeHeaders fragment >>= \ headers -> pure . Just $ StreamReadingBody headers "" + (False, True) -> pure . Just $ StreamReadingTrailerContinuation [] (BL.fromStrict fragment) "" + (False, False) -> pure . Just $ StreamReadingHeaderContinuation (BL.fromStrict fragment) + StreamReadingHeaderContinuation _ -> + throwError (H2.ProtocolError, "unexpected HEADERS while expecting CONTINUATION") + StreamReadingBody headers bodyBytes -> + case (H2.testEndHeader flags, H2.testEndStream flags) of + (True, True) -> decodeHeaders fragment >>= \ trailers -> processEndStream push (headers ++ trailers) bodyBytes + (True, False) -> streamError push H2.ProtocolError "unexpected HEADERS without END_STREAM after DATA" + (False, True) -> pure . Just $ StreamReadingTrailerContinuation headers (BL.fromStrict fragment) "" + (False, False) -> streamError push H2.ProtocolError "unexpected HEADERS without END_STREAM after DATA" + StreamReadingTrailerContinuation _ _ _ -> + throwError (H2.ProtocolError, "unexpected HEADERS while expecting CONTINUATION") + postWritePhase haveUpdatedWindowsOrStreams + + H2.ContinuationFrame fragment -> do + withReadableStream $ \ push -> \ case + StreamWaiting -> + throwError (H2.ProtocolError, "unexpected CONTINUATION while expecting HEADERS or DATA") + StreamReadingHeaderContinuation accumulatedFragments -> do + let accumulatedFragments' = accumulatedFragments <> BL.fromStrict fragment + if H2.testEndHeader flags + then decodeHeaders (BL.toStrict accumulatedFragments') >>= \ headers -> pure . Just $ StreamReadingBody headers "" + else pure . Just $ StreamReadingHeaderContinuation accumulatedFragments' + StreamReadingBody _ _ -> + throwError (H2.ProtocolError, "unexpected CONTINUATION while expecting DATA or HEADERS") + StreamReadingTrailerContinuation headers accumulatedFragments bodyBytes -> do + let accumulatedFragments' = accumulatedFragments <> BL.fromStrict fragment + if H2.testEndHeader flags + then decodeHeaders (BL.toStrict accumulatedFragments') >>= \ trailers -> processEndStream push (headers ++ trailers) bodyBytes + else pure . Just $ StreamReadingTrailerContinuation headers accumulatedFragments' bodyBytes + postWritePhase haveUpdatedWindowsOrStreams + + H2.DataFrame bytes -> do + withReadableStream $ \ push -> \ case + StreamWaiting -> do + if H2.testEndStream flags + then processEndStream push [] (BL.fromStrict bytes) + else pure . Just $ StreamReadingBody [] (BL.fromStrict bytes) + StreamReadingHeaderContinuation _ -> do + throwError (H2.ProtocolError, "unexpected DATA while expecting CONTINUATION") + StreamReadingBody headers bodyBytes -> do + processorDebug $ "received more DATA for stream " <> (pack . show) sid + let bodyBytes' = bodyBytes <> BL.fromStrict bytes + if H2.testEndStream flags + then processEndStream push headers bodyBytes' + else pure . Just $ StreamReadingBody headers bodyBytes' + StreamReadingTrailerContinuation _ _ _ -> + streamError push H2.ProtocolError "unexpected DATA frame after trailing HEADERS sequence" + postWritePhase haveUpdatedWindowsOrStreams + + H2.RSTStreamFrame errorCodeId -> do + processorDebug $ "received RST_STREAM for stream " <> (pack . show) sid <> " with " <> (pack . show) errorCodeId + let terminate push = invokeWithoutException . _apnsPush_callback push $ ApnsPushStreamError errorCodeId "received RST_STREAM from server" + traverse_ (terminate . _writableStream_push) =<< use (state_writableStreams . at sid) + traverse_ (terminate . fst) =<< use (state_readableStreams . at sid) + assign (state_writableStreams . at sid) Nothing + assign (state_readableStreams . at sid) Nothing + postWritePhase haveUpdatedWindowsOrStreams + + H2.PushPromiseFrame _ _ -> do + processorDebug "unexpected and unwanted PUSH_PROMISE frame" + nextStreamId <- use state_nextStreamId + processorWrite (H2.encodeInfo id 0) (H2.GoAwayFrame (pred nextStreamId) H2.ProtocolError "unwanted PUSH_PROMISE") + throwError . ApnsTerminatedHttp2Error $ H2.ConnectionError H2.ProtocolError "unwanted PUSH_PROMISE" + + H2.PingFrame bytes -> do + unless (H2.testAck flags) $ do + processorDebug "responding to PING" + processorWrite (H2.encodeInfo H2.setAck 0) (H2.PingFrame bytes) + postWritePhase haveUpdatedWindowsOrStreams + + H2.GoAwayFrame lastProcessedStreamId errorCodeId errorData -> do + let decodeGoAway = withObject "GOAWAY frame" $ \ o -> o .: "reason" >>= maybe (fail "unknown reason") pure . apnsErrorReasonFromText + apnsErrorOrRaw = first (const errorData) $ eitherDecode' (BL.fromStrict errorData) >>= parseEither decodeGoAway + processorDebug $ "received GOAWAY " <> (pack . show) lastProcessedStreamId <> " " <> (pack . show) errorCodeId <> " " <> (pack . show) errorData + closingTv <- view context_closingTv + liftIO . atomically . writeTVar closingTv . Just $ ApnsTerminatedRemotely apnsErrorOrRaw + + (validWritableStreams, droppedWritableStreams) <- uses state_writableStreams (IntMap.split (succ lastProcessedStreamId)) + (validReadableStreams, droppedReadableStreams) <- uses state_readableStreams (IntMap.split (succ lastProcessedStreamId)) + + let dropped :: (v -> ApnsPush) -> H2.StreamId -> v -> ProcessorM () + dropped f droppedSid v = do + processorDebug $ "stream " <> (pack . show) droppedSid <> " dropped" + invokeWithoutException . _apnsPush_callback (f v) $ ApnsPushDropped + + traverseWithKey_ (dropped _writableStream_push) droppedWritableStreams + traverseWithKey_ (dropped fst ) droppedReadableStreams + assign state_writableStreams validWritableStreams + assign state_readableStreams validReadableStreams + + postWritePhase haveUpdatedWindowsOrStreams + + H2.WindowUpdateFrame delta -> do + processorDebug $ "stream " <> (pack . show) sid <> " WINDOW_UPDATE " <> (pack . show) delta + + if sid == 0 + then modifying state_connectionWindow (+ delta) + else modifying (state_writableStreams . at sid . _Just . writableStream_windowSize) (+ delta) + + postWritePhase True + + other -> do + processorDebug $ "ignoring unexpected frame " <> (pack . show) other + postWritePhase haveUpdatedWindowsOrStreams + + where + withReadableStream :: (ApnsPush -> ReadableStream -> ExceptT (H2.ErrorCodeId, ByteString) ProcessorM (Maybe ReadableStream)) -> ProcessorM () + withReadableStream k = use (state_readableStreams . at sid) >>= \ case + Just (push, state) -> + runExceptT (k push state) >>= \ case + Left (errorCodeId, errorData) -> do + processorDebug $ "while handling a frame for stream " <> (pack . show) sid <> " encountered unrecoverable connection error " <> (pack . show) errorCodeId <> " " <> (pack . show) errorData + nextStreamId <- use state_nextStreamId + processorWrite (H2.encodeInfo id 0) (H2.GoAwayFrame (pred nextStreamId) errorCodeId errorData) + throwError . ApnsTerminatedHttp2Error $ H2.ConnectionError errorCodeId errorData + Right newStateMay -> do + assign (state_readableStreams . at sid) ((push,) <$> newStateMay) + Nothing -> do + processorDebug $ "unexpected readable stream " <> (pack . show) sid <> ", sending RST" + processorWrite (H2.encodeInfo id sid) (H2.RSTStreamFrame H2.StreamClosed) + + streamError :: ApnsPush -> H2.ErrorCodeId -> ByteString -> ExceptT (H2.ErrorCodeId, ByteString) ProcessorM (Maybe ReadableStream) + streamError push errorCodeId errorData = do + processorDebug $ "stream " <> (pack . show) sid <> " encountered processing error " <> (pack . show) errorCodeId <> " " <> (pack . show) errorData + lift $ processorWrite (H2.encodeInfo id sid) (H2.RSTStreamFrame errorCodeId) + invokeWithoutException . _apnsPush_callback push $ ApnsPushStreamError errorCodeId errorData + pure Nothing + + responseParseError :: ApnsPush -> Text -> ExceptT (H2.ErrorCodeId, ByteString) ProcessorM (Maybe ReadableStream) + responseParseError push explanation = do + processorDebug $ "stream " <> (pack . show) sid <> " encountered response parsing error: " <> explanation + invokeWithoutException . _apnsPush_callback push $ ApnsPushResponseParseError explanation + pure Nothing + + decodeHeaders :: ByteString -> ExceptT (H2.ErrorCodeId, ByteString) ProcessorM HP.HeaderList + decodeHeaders headerBlock = do + decodeTable <- view context_decodingTable + (try . liftIO $ HP.decodeHeader decodeTable headerBlock) >>= \ case + Left (ex :: SomeException) -> throwError (H2.CompressionError, "failed to decompress headers: " <> (BSC8.pack . show) ex) + Right headers -> pure headers + + processEndStream :: ApnsPush -> HP.HeaderList -> BL.ByteString -> ExceptT (H2.ErrorCodeId, ByteString) ProcessorM (Maybe ReadableStream) + processEndStream push headers body = either id pure <=< runExceptT $ do + status <- maybe (throwError (streamError push H2.ProtocolError "missing or invalid :status header")) pure $ + readMaybe . BSC8.unpack =<< lookup ":status" headers + + apnsId <- maybe (throwError (responseParseError push "missing or invalid apns-id header")) pure $ + lookup "apns-id" headers + + if status >= 200 && status <= 299 + then invokeWithoutException . _apnsPush_callback push $ ApnsPushDelivered apnsId + else do + (reason, timestamp) <- either (throwError . responseParseError push . pack) pure $ + eitherDecode' body >>= parseEither parseResponse + invokeWithoutException . _apnsPush_callback push . ApnsPushErrored $ ApnsError apnsId status reason timestamp + + pure Nothing + + parseResponse :: Value -> Parser (Either Text ApnsErrorReason, Maybe UTCTime) + parseResponse = withObject "APNs response" $ \ obj -> + (,) + <$> (obj .: "reason" <&> \ t -> maybe (Left t) Right (apnsErrorReasonFromText t)) + <*> (fmap posixSecondsToUTCTime <$> obj .:? "timestamp") + + +-- |Process a new 'ApnsPush' by allocating a stream for it, sending its headers, and preparing to write the body data for it, then continuing on with the +-- 'postWritePhase' in progress. +processNewPush :: Bool -> ApnsPush -> ProcessorM () +processNewPush haveUpdatedWindowsOrStreams push = do + sid <- use state_nextStreamId + if sid < 0 || sid >= maxBound - 1 + then do + processorDebug "cannot process any more pushes on this connection due to stream exhaustion" + invokeWithoutException $ _apnsPush_callback push ApnsPushStreamsExhausted + postWritePhase haveUpdatedWindowsOrStreams + else do + initialWindowSize <- uses state_settings H2.initialWindowSize + maxFrameSize <- uses state_settings H2.maxFrameSize + encodingTable <- view context_encodingTable + authority <- view context_authority + let headers = headersForPush authority push + headerBlockFragments <- liftIO $ encodeHeaderBlockFragments maxFrameSize encodingTable headers + + modifying state_nextStreamId (+2) + let ws = WritableStream push initialWindowSize $ view (apnsPush_data . apnsPushData_payload) push + assign (state_writableStreams . at sid) (Just ws) + + processorDebug $ "writing headers for new stream " <> (pack . show) sid <> " with " <> (pack . show) headers <> " in " <> (pack . show) (length headerBlockFragments) <> " header block fragments" + + case preview _Cons headerBlockFragments of + Just (firstFragment, preview _Snoc -> Just (middleFragments, lastFragment)) -> do + processorWriteMany + $ (H2.encodeInfo id sid, H2.HeadersFrame Nothing firstFragment) + : map ((H2.encodeInfo id sid,) . H2.ContinuationFrame) middleFragments + ++ [(H2.encodeInfo H2.setEndHeader sid, H2.ContinuationFrame lastFragment)] + + Just (firstFragment, _) -> + processorWrite (H2.encodeInfo H2.setEndHeader sid) (H2.HeadersFrame Nothing firstFragment) + + Nothing -> + -- should never happen, should be at least pseudo headers + processorWrite (H2.encodeInfo H2.setEndHeader sid) (H2.HeadersFrame Nothing "") + + postWritePhase True + +-- |Given an 'ApnsPush', compose the list of headers that should be sent for it. +headersForPush :: ByteString -> ApnsPush -> HP.HeaderList +headersForPush authority (view apnsPush_data -> ApnsPushData {..}) + -- FIXME this doesn't guide the HPACK encoding process at all, but should according to the Apple guidelines, so probably the dynamic table usage is not great + = (":method", "POST") + : (":scheme", "https") + : (":authority", authority) + : (":path", "/3/device/" <> (BL.toStrict . toLazyByteString . byteStringHex) _apnsPushData_deviceToken) + : ("apns-expiration", (BSC8.pack . show . (truncate :: NominalDiffTime -> Word32) . utcTimeToPOSIXSeconds) _apnsPushData_expiration) + : ("apns-topic", encodeUtf8 _apnsPushData_topic) + : catMaybes + [ ("apns-priority",) . BSC8.pack . show <$> _apnsPushData_priority + , ("apns-collapse-id",) <$> _apnsPushData_collapseId + ] + +-- |Given the maximum frame size, a 'HP.HeaderList', and the 'HP.DynamicTable' for encoding, encode a list of headers into a list of header block fragments. +-- The first such fragment gets sent in a HEADERS frame while the subsequent ones if any get sent in a CONTINUATION frame. +encodeHeaderBlockFragments :: Int -> HP.DynamicTable -> HP.HeaderList -> IO [H2.HeaderBlockFragment] +encodeHeaderBlockFragments maxFrameSize dynTbl headers = do + let maxPackSize = length headers + sumOf (folded . both . to (succ . BS.length)) headers + headerBlock <- HP.encodeHeader HP.defaultEncodeStrategy maxPackSize dynTbl headers + let chunk bs + | (h, t) <- BS.splitAt maxFrameSize bs + , not (BS.null h) = h : chunk t + | otherwise = [] + pure $ chunk headerBlock + +-- |Reader thread. Before this exits, some debug log will be emitted, the socket will be shut down for reads, and the reader queue will be closed. +reader :: (Text -> IO ()) -> TLS.Context -> Socket -> TBMQueue ReaderEvent -> IO () +reader debugLog tlsContext sock readerQueue = do + res <- try + $ sourceTls + $$ decodeH2Frames H2.defaultSettings + .| CL.iterM (debugLog . ("received: " <>) . pack . show) + .| sinkTBMQueue readerQueue True + + atomically $ closeTBMQueue readerQueue + invokeWithoutException $ Socket.shutdown sock Socket.ShutdownReceive + + case res of + Left (ex :: SomeException) -> do + debugLog $ "reader ended in exception: " <> (pack . show) ex + throwIO ex + Right _ -> + debugLog "reader ended normally" + + where + sourceTls :: Producer IO ByteString + sourceTls = + liftIO (TLS.recvData tlsContext) >>= \ case + bs | BS.null bs -> pure () + | otherwise -> yield bs >> sourceTls + + decodeH2Frames :: H2.Settings -> ConduitM ByteString ReaderEvent IO () + decodeH2Frames settings = do + let continue (eventMay, settings') = traverse_ yield eventMay >> decodeH2Frames settings' + + either yield continue <=< runExceptT $ do + headerBytes <- lift $ BL.toStrict <$> CB.take H2.frameHeaderLength + when (BS.length headerBytes < H2.frameHeaderLength) $ + throwError (ReadH2Error H2.ProtocolError "Insufficient header bytes" ApnsTerminatedInsufficientHeaderBytes) + + (frameType, frameHeader) <- + either (\ h2Error -> throwError $ ReadH2Error (H2.errorCodeId h2Error) (BSC8.pack . show $ h2Error) (ApnsTerminatedHttp2Error h2Error)) pure $ + H2.checkFrameHeader settings (H2.decodeFrameHeader headerBytes) + + when (H2.payloadLength frameHeader > H2.maxFrameSize settings) $ + throwError $ ReadH2Error H2.ProtocolError ("Payload larger than maximum of " <> (BSC8.pack . show) (H2.maxFrameSize settings)) ApnsTerminatedOversizedPayload + payloadBytes <- lift $ BL.toStrict <$> CB.take (H2.payloadLength frameHeader) + when (BS.length payloadBytes < H2.payloadLength frameHeader) $ + throwError $ ReadH2Error H2.ProtocolError "Insufficient payload bytes" (ApnsTerminatedInsufficientPayloadBytes frameType) + + payload <- + either (\ h2Error -> throwError $ ReadH2Error (H2.errorCodeId h2Error) (BSC8.pack $ show h2Error) (ApnsTerminatedHttp2Error h2Error)) pure $ + H2.decodeFramePayload frameType frameHeader payloadBytes + + case payload of + H2.SettingsFrame _ | H2.testAck (H2.flags frameHeader) -> + pure (Nothing, settings) + + H2.SettingsFrame settingsList -> do + for_ (H2.checkSettingsList settingsList) $ \ h2Error -> + throwError $ ReadH2Error (H2.errorCodeId h2Error) (BSC8.pack $ show h2Error) (ApnsTerminatedHttp2Error h2Error) + + let settings' = H2.updateSettings settings settingsList + + pure (Just (ReadSettingsUpdate settings'), settings') + + other -> + pure (Just (ReadFrame $ H2.Frame frameHeader other), settings) + +-- |Writer thread. Before this exits, some debug log will be emitted, the socket will be shut down for writes, and the writer queue will be closed. +writer :: (Text -> IO ()) -> TLS.Context -> Socket -> TBMQueue (H2.EncodeInfo, H2.FramePayload) -> IO () +writer debugLog tlsContext sock writerQueue = do + res <- try + $ sourceTBMQueue writerQueue + $$ CL.iterM (debugLog . ("sending: " <>) . pack . show) + .| CL.concatMap (map BL.fromStrict . uncurry H2.encodeFrameChunks) + .| awaitForever (TLS.sendData tlsContext) + + atomically $ closeTBMQueue writerQueue + invokeWithoutException $ do + TLS.bye tlsContext + Socket.shutdown sock Socket.ShutdownSend + + case res of + Left (ex :: SomeException) -> do + debugLog $ "writer ended in exception: " <> (pack . show) ex + throwIO ex + Right _ -> + debugLog "writer ended normally" + diff --git a/src/Network/Apns/Types.hs b/src/Network/Apns/Types.hs new file mode 100644 --- /dev/null +++ b/src/Network/Apns/Types.hs @@ -0,0 +1,268 @@ +-- |Public types for using the APNs integration. +module Network.Apns.Types + ( ApnsPushData(..), apnsPushData_deviceToken, apnsPushData_expiration, apnsPushData_priority, apnsPushData_topic, apnsPushData_collapseId, apnsPushData_payload + , ApnsErrorReason(..), apnsErrorReasonToText, apnsErrorReasonFromText, apnsErrorReasonFromTextMap + , ApnsError(..), apnsError_apnsId, apnsError_status, apnsError_reason, apnsError_timestamp + , ApnsPushResult(..), _ApnsPushDelivered, _ApnsPushErrored, _ApnsPushStreamError, _ApnsPushResponseParseError, _ApnsPushDropped, _ApnsPushConnectionClosed + , _ApnsPushStreamsExhausted + , ApnsPush(..), apnsPush_data, apnsPush_callback + , ApnsTerminationReason(..), _ApnsTerminatedLocally, _ApnsTerminatedRemotely, _ApnsTerminatedHttp2Error, _ApnsTerminatedInsufficientHeaderBytes + , _ApnsTerminatedInsufficientPayloadBytes, _ApnsTerminatedReadError, _ApnsTerminatedWriteError, _ApnsTerminatedProcessingException + , ApnsConnectionParams(..), apnsConnectionParams_hostName, apnsConnectionParams_portNumber, apnsConnectionParams_clientCertificateChain + , apnsConnectionParams_clientCertificateKey, apnsConnectionParams_serverCertificateStore, apnsConnectionParams_onDebugLog + , apnsConnectionParams_onServerCertificate, apnsConnectionParams_onTermination, apnsConnectionParams_readQueueSize, apnsConnectionParams_writeQueueSize + , apnsConnectionParams_pushQueueSize + , ApnsConnectionError(..), _ApnsConnectionErrorInvalidHostName, _ApnsConnectionErrorConnectFailed, _ApnsConnectionErrorInvalidServerPreface + , _ApnsConnectionErrorException + , ApnsConnection(..), apnsConnection_submitPush, apnsConnection_close + ) where + +import Control.Arrow ((&&&)) +import Control.Exception (IOException, SomeException) +import Control.Lens.TH (makeLenses, makePrisms) +import Data.ByteString (ByteString) +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as Map +import Data.Text (Text) +import Data.Time.Clock (UTCTime) +import Data.X509 (CertificateChain, PrivKey) +import Data.X509.CertificateStore (CertificateStore) +import Data.X509.Validation (FailedReason, ServiceID) +import Network.HTTP2 (ErrorCodeId, FrameTypeId, HTTP2Error) +import Network.Socket (HostName, PortNumber) +import Network.TLS (ValidationCache) + + +-- |Data for an APNs push message, consisting of its target device token, information that gets sent as headers with the push request such as expiration and +-- topic, and the payload itself. +data ApnsPushData = ApnsPushData + { _apnsPushData_deviceToken :: ByteString + -- ^The target device token, as its raw binary self. This value will be base16 encoded when sending the push message. + , _apnsPushData_expiration :: UTCTime + -- ^When the push should expire at APNs if it hasn't been delivered yet. + , _apnsPushData_priority :: Maybe Int + -- ^What priority the push should have at APNs, with 10 being immediately and must have a alert, sound or badge, and 5 being possibly deferred due to battery + -- conservation and other factors. + , _apnsPushData_topic :: Text + -- ^The topic (usually app bundle ID) that this push is being sent for. + , _apnsPushData_collapseId :: Maybe ByteString + -- ^An optional identifier that the target device can use to consolidate a series of updates to the same notification. + , _apnsPushData_payload :: ByteString + -- ^The payload data, usually formatted as JSON. + } deriving (Show) + +makeLenses ''ApnsPushData + +-- |Various error codes reported by APNs in response to a push or on the connection as a whole. See APNs documentation for meanings. +data ApnsErrorReason + = ApnsErrorBadCollapseId + | ApnsErrorBadDeviceToken + | ApnsErrorBadExpirationDate + | ApnsErrorBadMessageId + | ApnsErrorBadPriority + | ApnsErrorBadTopic + | ApnsErrorDeviceTokenNotForTopic + | ApnsErrorDuplicateHeaders + | ApnsErrorIdleTimeout + | ApnsErrorMissingDeviceToken + | ApnsErrorMissingTopic + | ApnsErrorPayloadEmpty + | ApnsErrorTopicDisallowed + | ApnsErrorBadCertificate + | ApnsErrorBadCertificateEnvironment + | ApnsErrorExpiredProviderToken + | ApnsErrorForbidden + | ApnsErrorInvalidProviderToken + | ApnsErrorMissingProviderToken + | ApnsErrorBadPath + | ApnsErrorMethodNotAllowed + | ApnsErrorUnregistered + | ApnsErrorPayloadTooLarge + | ApnsErrorTooManyProviderTokenUpdates + | ApnsErrorTooManyRequests + | ApnsErrorInternalServerError + | ApnsErrorServiceUnavailable + | ApnsErrorShutdown + deriving (Bounded, Enum, Eq, Ord, Show) + +makePrisms ''ApnsErrorReason + +-- |Convert an 'ApnsErrorReason' to its text equivalent. +apnsErrorReasonToText :: ApnsErrorReason -> Text +apnsErrorReasonToText = \ case + ApnsErrorBadCollapseId -> "BadCollapseId" + ApnsErrorBadDeviceToken -> "BadDeviceToken" + ApnsErrorBadExpirationDate -> "BadExpirationDate" + ApnsErrorBadMessageId -> "BadMessageId" + ApnsErrorBadPriority -> "BadPriority" + ApnsErrorBadTopic -> "BadTopic" + ApnsErrorDeviceTokenNotForTopic -> "DeviceTokenNotForTopic" + ApnsErrorDuplicateHeaders -> "DuplicateHeaders" + ApnsErrorIdleTimeout -> "IdleTimeout" + ApnsErrorMissingDeviceToken -> "MissingDeviceToken" + ApnsErrorMissingTopic -> "MissingTopic" + ApnsErrorPayloadEmpty -> "PayloadEmpty" + ApnsErrorTopicDisallowed -> "TopicDisallowed" + ApnsErrorBadCertificate -> "BadCertificate" + ApnsErrorBadCertificateEnvironment -> "BadCertificateEnvironment" + ApnsErrorExpiredProviderToken -> "ExpiredProviderToken" + ApnsErrorForbidden -> "Forbidden" + ApnsErrorInvalidProviderToken -> "InvalidProviderToken" + ApnsErrorMissingProviderToken -> "MissingProviderToken" + ApnsErrorBadPath -> "BadPath" + ApnsErrorMethodNotAllowed -> "MethodNotAllowed" + ApnsErrorUnregistered -> "Unregistered" + ApnsErrorPayloadTooLarge -> "PayloadTooLarge" + ApnsErrorTooManyProviderTokenUpdates -> "TooManyProviderTokenUpdates" + ApnsErrorTooManyRequests -> "TooManyRequests" + ApnsErrorInternalServerError -> "InternalServerError" + ApnsErrorServiceUnavailable -> "ServiceUnavailable" + ApnsErrorShutdown -> "Shutdown" + +-- |Mapping of 'ApnsErrorReason' text forms to values. +{-# NOINLINE apnsErrorReasonFromTextMap #-} +apnsErrorReasonFromTextMap :: Map Text ApnsErrorReason +apnsErrorReasonFromTextMap = Map.fromList . map (apnsErrorReasonToText &&& id) $ enumFromTo minBound maxBound + +-- |Convert a text value to a @Just 'ApnsErrorReason'@ if the text value is a valid error reason, @Nothing@ if it's invalid. +apnsErrorReasonFromText :: Text -> Maybe ApnsErrorReason +apnsErrorReasonFromText = (`Map.lookup` apnsErrorReasonFromTextMap) + +-- |Structure carrying error information returned by APNs in response to a push. +data ApnsError = ApnsError + { _apnsError_apnsId :: ByteString + -- ^The unique identifier assigned by APNs for the push. + , _apnsError_status :: Int + -- ^The HTTP status code returned. Should never be in the 2xx range. + , _apnsError_reason :: Either Text ApnsErrorReason + -- ^The reason for failure given by APNs (@Right@) or the raw value of the @reason@ field if its value is unrecognized (@Left@). + , _apnsError_timestamp :: Maybe UTCTime + -- ^For status @410@, the last time APNs confirmed that the device token was invalid. No pushes should be sent for this device token. + } deriving Show + +makeLenses ''ApnsError + +-- |The result of a push delivery attempt. +data ApnsPushResult + = ApnsPushDelivered ByteString + -- ^The push was successfully delivered to APNs along with its assigned @apns-id@. It may or may not be delivered to the target device, at the discretion + -- of APNs. + | ApnsPushErrored ApnsError + -- ^APNs rejected the push notification. See the 'ApnsError' for more detail. + | ApnsPushStreamError ErrorCodeId ByteString + -- ^APNs reset the HTTP/2 stream before delivery could be completed. The push was most likely not accepted by APNs. The HTTP/2 'ErrorCodeId' and error data + -- are given. + | ApnsPushResponseParseError Text + -- ^APNs gave an error response but it could not be parsed as JSON. The JSON parsing error is given. + | ApnsPushDropped + -- ^APNs closed the HTTP/2 connection and indicated it did not process the push. It should be resubmitted on a new connection. + | ApnsPushConnectionClosed + -- ^The connection to APNs was already closed by the time the push was requested. It should be resubmitted on a new connection. + | ApnsPushStreamsExhausted + -- ^The HTTP/2 connection cannot allocate any more new streams. The connection should be closed and the push resubmitted on a new connection. + deriving Show + +makePrisms ''ApnsPushResult + +-- |Push notification data along with a callback that's invoked with the disposition of the push as soon as it's known. +data ApnsPush = ApnsPush + { _apnsPush_data :: ApnsPushData + -- ^What data to send for the push. + , _apnsPush_callback :: ApnsPushResult -> IO () + -- ^What callback to invoke when the push result is known. Exceptions thrown by the callback will be silently ignored. + } + +makeLenses ''ApnsPush + +-- |The reason an APNs connection was terminated. +data ApnsTerminationReason + = ApnsTerminatedLocally + -- ^The APNs connection was locally closed using '_apnsConnection_close'. + | ApnsTerminatedRemotely (Either ByteString ApnsErrorReason) + -- ^The APNs connection was remotely closed by APNs. The APNs error reason is given as @Right@ if available, otherwise some other error string + -- given as @Left@. + | ApnsTerminatedHttp2Error HTTP2Error + -- ^The APNs connection was terminated due to an HTTP/2 protocol error. + | ApnsTerminatedInsufficientHeaderBytes + -- ^The APNs connection was terminated because the TCP stream ended before being able to read all the header bytes for an HTTP/2 frame. + | ApnsTerminatedInsufficientPayloadBytes FrameTypeId + -- ^The APNs connection was terminated because the TCP stream ended before being able to read all the payload bytes for an HTTP/2 frame. + | ApnsTerminatedOversizedPayload + -- ^The APNs connection was terminated because the server send a frame with a purported payload size larger than the currently negotiated MAX_FRAME_SIZE. + | ApnsTerminatedReadError SomeException + -- ^The APNs connection was terminated by an I/O error while reading. + | ApnsTerminatedWriteError SomeException + -- ^The APNs connection was terminated by an I/O error while writing. + | ApnsTerminatedProcessingException SomeException + -- ^The processing thread died due to some unexpected exception. Pushes currently in transit might or might not be delivered. + deriving Show + +makePrisms ''ApnsTerminationReason + +-- |Parameters for establishing an APNs connection. +data ApnsConnectionParams = ApnsConnectionParams + { _apnsConnectionParams_hostName :: HostName + -- ^The host name of APNs. Usually @api.development.push.apple.com@ or @api.push.apple.com@. + , _apnsConnectionParams_portNumber :: PortNumber + -- ^The port number of APNs. Usually @443@. + , _apnsConnectionParams_clientCertificateChain :: CertificateChain + -- ^The client TLS certificate and chain to send to APNs to authenticate. Usually generated by the Apple developer portal. + -- + -- See 'Network.TLS.credentialLoadX509Chain' for one way to provide this value. + , _apnsConnectionParams_clientCertificateKey :: PrivKey + -- ^The client TLS private key to use when authenticating with APNs. Usually generated locally and used to generate a certificate signing request which is + -- then submitted to the Apple developer portal. + -- + -- See 'Network.TLS.credentialLoadX509Chain' for one way to provide this value. + , _apnsConnectionParams_serverCertificateStore :: Maybe CertificateStore + -- ^The certificate store to use when validating the server certificate, typically the system's trusted CA store. + -- + -- See @getSystemCertificateStore@ from the @x509-system@ package for a method to load the system's trusted CA store, or @readCertificateStore@ from + -- @x509-store@ for a way to load a custom trusted CA store. + , _apnsConnectionParams_onDebugLog :: Text -> IO () + -- ^A callback invoked by the APNs library to report debugging information. Set to @\ _ -> pure ()@ for no debug log. Take care to prevent any exceptions + -- from being thrown as they will likely leave the APNs connection in an undefined state. + -- + -- WARNING: this will be both verbose and contain potentially sensitive information as it includes all traffic exchanged with APNs, including payloads + -- and headers. The TLS certificate or key will not be included however. + , _apnsConnectionParams_onServerCertificate :: Maybe (CertificateStore -> ValidationCache -> ServiceID -> CertificateChain -> IO [FailedReason]) + -- ^A callback invoked by the TLS library to validate the server's presented certificate. @Nothing@ means use @validateDefault@, which uses + -- '_apnsConnectionParams_serverCertificateStore' to validate the certificate in the usual way. + , _apnsConnectionParams_onTermination :: ApnsTerminationReason -> IO () + -- ^A callback invokved by the APNs library when the APNs connection terminates with the reason why it was terminated. + , _apnsConnectionParams_readQueueSize :: Int + -- ^How many HTTP/2 frames to queue. Usually a small number is appropriate as the processing thread does no major blocking. + , _apnsConnectionParams_writeQueueSize :: Int + -- ^How many HTTP/2 frames to queue. Usually a small number is appropriate as the server should direct flow control. + , _apnsConnectionParams_pushQueueSize :: Int + -- ^How many 'ApnsPush' requests to queue. Usually a small number is appropriate. + } + +makeLenses ''ApnsConnectionParams + +-- |Indication of reason for an APNs connection attempt failing before the main asynchronous processing could begin. +data ApnsConnectionError + = ApnsConnectionErrorInvalidHostName HostName + -- ^The host name given as '_apnsConnectionParams_hostName' was not resolvable. + | ApnsConnectionErrorConnectFailed HostName [IOException] + -- ^The host name and port given as '_apnsConnectionParams_hostName' and '_apnsConnectionParams_portNumber' could not be connected to on any of the resolved + -- addresses. The exceptions generated when connections were attempted are given. + | ApnsConnectionErrorInvalidServerPreface + -- ^The server did not initiate with a valid HTTP/2 server preface, defined as a SETTINGS frame. Typically this indicates that the host was not APNs, APNs + -- is catastrophically down, or something along those lines. + | ApnsConnectionErrorException SomeException + -- ^An unexpected error occurred while waiting for the server to complete the HTTP/2 handshake. + deriving (Show) + +makePrisms ''ApnsConnectionError + +-- |An established APNs connection. In the future this should get metric and status reporting functions. +data ApnsConnection = ApnsConnection + { _apnsConnection_submitPush :: ApnsPush -> IO () + -- ^Submit a push message. This should never fail immediately and any result of the push attempt should be reported via the '_apnsPush_callback' + , _apnsConnection_close :: IO () + -- ^Close the APNs connection in an orderly fashion. The '_apnsConnectionParams_onTermination' callback will be triggered with 'ApnsTerminatedLocally' + -- if the connection closes correctly. + } + +makeLenses ''ApnsConnection