icepeak (empty) → 0.7.1.0
raw patch · 30 files changed
+2697/−0 lines, 30 filesdep +QuickCheckdep +aesondep +async
Dependencies added: QuickCheck, aeson, async, base, bytestring, containers, directory, hashable, hspec, hspec-core, hspec-expectations, hspec-wai, http-types, icepeak, jwt, monad-logger, mtl, network, optparse-applicative, prometheus-client, prometheus-metrics-ghc, quickcheck-instances, random, raven-haskell, scotty, securemem, sqlite-simple, stm, text, time, unix, unordered-containers, uuid, wai, wai-extra, wai-middleware-prometheus, wai-websockets, warp, websockets
Files
- app/Icepeak/Main.hs +158/−0
- app/IcepeakTokenGen/Main.hs +90/−0
- icepeak.cabal +224/−0
- src/AccessControl.hs +97/−0
- src/Config.hs +154/−0
- src/Core.hs +174/−0
- src/HTTPMethodInvalid.hs +29/−0
- src/HttpServer.hs +93/−0
- src/JwtAuth.hs +104/−0
- src/JwtMiddleware.hs +117/−0
- src/Logger.hs +82/−0
- src/Metrics.hs +76/−0
- src/MetricsServer.hs +24/−0
- src/Persistence.hs +341/−0
- src/SentryLogging.hs +20/−0
- src/Server.hs +26/−0
- src/Store.hs +95/−0
- src/Subscription.hs +121/−0
- src/WebsocketServer.hs +144/−0
- tests/AccessControlSpec.hs +43/−0
- tests/ApiSpec.hs +6/−0
- tests/CoreSpec.hs +8/−0
- tests/JwtSpec.hs +76/−0
- tests/OrphanInstances.hs +33/−0
- tests/PersistenceSpec.hs +31/−0
- tests/RequestSpec.hs +56/−0
- tests/SocketSpec.hs +6/−0
- tests/Spec.hs +23/−0
- tests/StoreSpec.hs +132/−0
- tests/SubscriptionTreeSpec.hs +114/−0
+ app/Icepeak/Main.hs view
@@ -0,0 +1,158 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+module Main (main) where++import Control.Exception (fromException, catch, handle, AsyncException, SomeException)+import Control.Monad (forM, void, when)+import Data.Foldable (forM_)+import Data.Semigroup ((<>))+import Options.Applicative (execParser)+import System.Environment (getEnvironment)+import System.IO (BufferMode (..), hSetBuffering, stdout)++import qualified Control.Concurrent.Async as Async+import qualified Data.Text as Text+import qualified Prometheus+import qualified Prometheus.Metric.GHC+import qualified System.Posix.Signals as Signals++import Config (Config (..), configInfo)+import Core (Core (..))+import Persistence (getDataFile, setupStorageBackend)+import Logger (Logger, LogLevel(..), postLog)++import qualified Core+import qualified HttpServer+import qualified Server+import qualified WebsocketServer+import qualified Logger+import qualified Metrics+import qualified MetricsServer++-- Install SIGTERM and SIGINT handlers to do a graceful exit.+installHandlers :: Core -> IO ()+installHandlers core =+ let+ logHandle = do+ postLog (coreLogger core) LogInfo "\nTermination sequence initiated ..."+ Core.postQuit core+ handler = Signals.CatchOnce logHandle+ blockSignals = Nothing+ installHandler signal = Signals.installHandler signal handler blockSignals+ in do+ void $ installHandler Signals.sigTERM+ void $ installHandler Signals.sigINT+++main :: IO ()+main = do+ -- make sure output is flushed regularly+ hSetBuffering stdout LineBuffering++ env <- getEnvironment+ config <- execParser (configInfo env)++ -- make sure the storage file exists and that it has the right format - otherwise, fail early+ let dataFile = getDataFile (configStorageBackend config) (configDataFile config)+ setupStorageBackend (configStorageBackend config) dataFile++ -- start logging as early as possible+ logger <- Logger.newLogger config+ loggerThread <- Async.async $ Logger.processLogRecords logger+ handle (\e -> postLog logger LogError . Text.pack . show $ (e :: SomeException)) $ do+ -- setup metrics if enabled+ icepeakMetrics <- forM (configMetricsEndpoint config) $ const $ do+ void $ Prometheus.register Prometheus.Metric.GHC.ghcMetrics+ Metrics.createAndRegisterIcepeakMetrics++ eitherCore <- Core.newCore config logger icepeakMetrics+ either (postLog logger LogError . Text.pack) runCore eitherCore++ -- only stop logging when everything else has stopped+ Logger.postStop logger+ Async.wait loggerThread+++runCore :: Core -> IO ()+runCore core = do+ let config = coreConfig core+ let logger = coreLogger core+ httpServer <- HttpServer.new core+ let wsServer = WebsocketServer.acceptConnection core++ -- start threads+ commandLoopThread <- Async.async $ catchRunCoreResult CommandLoopException $ Core.runCommandLoop core+ webSocketThread <- Async.async $ catchRunCoreResult WebSocketsException $ WebsocketServer.processUpdates core+ httpThread <- Async.async $ catchRunCoreResult HttpException $ Server.runServer logger wsServer httpServer (configPort config)+ syncThread <- Async.async $ Core.runSyncTimer core+ metricsThread <- Async.async+ $ forM_ (configMetricsEndpoint config) (MetricsServer.runMetricsServer logger)++ installHandlers core+ logAuthSettings config logger+ logQueueSettings config logger+ logSyncSettings config logger++ postLog logger LogInfo "System online. ** robot sounds **"++ -- Everything should stop when any of these stops+ (_, runCoreResult) <- Async.waitAny [commandLoopThread, webSocketThread, httpThread]+ logRunCoreResult logger runCoreResult++ -- kill all threads when one of the main threads ended+ Core.postQuit core -- should stop commandLoopThread+ Async.cancel webSocketThread+ Async.cancel httpThread+ Async.cancel metricsThread+ Async.cancel syncThread+ void $ Async.wait commandLoopThread++-- | Data type to hold results for the async that finishes first+data RunCoreResult+ = CommandLoopException SomeException+ | WebSocketsException SomeException+ | HttpException SomeException+ | ThreadOk++-- | If a threads fails, catch the error and tag it so we know how to log it+catchRunCoreResult :: (SomeException -> RunCoreResult) -> IO () -> IO RunCoreResult+catchRunCoreResult tag action = catch (action >> pure ThreadOk) $ \exc -> case fromException exc of+ Just (_ :: AsyncException) -> pure ThreadOk+ _ -> pure (tag exc) -- we only worry about non-async exceptions++logRunCoreResult :: Logger -> RunCoreResult -> IO ()+logRunCoreResult logger rcr = do+ case rcr of+ CommandLoopException exc -> handleLog "core" exc+ WebSocketsException exc -> handleLog "web sockets server" exc+ HttpException exc -> handleLog "http server" exc+ ThreadOk -> pure ()+ where+ handleLog name exc+ | Just (_ :: AsyncException) <- fromException exc = pure ()+ | otherwise = do+ Logger.postLog logger LogError $ name <> " stopped with an exception: " <> Text.pack (show exc)++logAuthSettings :: Config -> Logger -> IO ()+logAuthSettings cfg logger+ | configEnableJwtAuth cfg = case configJwtSecret cfg of+ Just _ -> postLog logger LogInfo "JWT authorization enabled and secret provided, tokens will be verified."+ Nothing -> postLog logger LogInfo "JWT authorization enabled but no secret provided, tokens will NOT be verified."+ | otherwise = case configJwtSecret cfg of+ Just _ -> postLog logger LogInfo "WARNING a JWT secret has been provided, but JWT authorization is disabled."+ Nothing -> postLog logger LogInfo "JWT authorization disabled."++logQueueSettings :: Config -> Logger -> IO ()+logQueueSettings cfg logger =+ postLog logger LogInfo ("Queue capacity is set to " <> Text.pack (show (configQueueCapacity cfg)) <> ".")++logSyncSettings :: Config -> Logger -> IO ()+logSyncSettings cfg logger = case configSyncIntervalMicroSeconds cfg of+ Nothing -> do+ postLog logger LogInfo "Sync: Persisting after every modification"+ when (configEnableJournaling cfg) $ do+ postLog logger LogInfo "Journaling has no effect when periodic syncing is disabled"+ Just musecs -> do+ postLog logger LogInfo ("Sync: every " <> Text.pack (show musecs) <> " microseconds.")+ when (configEnableJournaling cfg) $ do+ postLog logger LogInfo "Journaling enabled"
+ app/IcepeakTokenGen/Main.hs view
@@ -0,0 +1,90 @@+{-# LANGUAGE OverloadedStrings #-}+module Main where++import Data.Semigroup ((<>))+import qualified Data.Text as Text+import qualified Data.Text.IO as Text+import Options.Applicative+import System.Environment (getEnvironment)+import qualified Web.JWT as JWT+import qualified Data.Time.Clock.POSIX as Clock++import AccessControl+import JwtAuth++data Config = Config+ { configJwtSecret :: Maybe JWT.Signer+ , configExpiresSeconds :: Maybe Integer+ , configWhitelist :: [AuthPath]+ }++type EnvironmentConfig = [(String, String)]++main :: IO ()+main = do+ env <- getEnvironment+ config <- execParser (configInfo env)+ now <- Clock.getPOSIXTime++ let joseHeader = JWT.JOSEHeader+ { JWT.typ = Just "JWT"+ , JWT.cty = Nothing+ , JWT.alg = Just JWT.HS256+ , JWT.kid = Nothing+ }++ let access = IcepeakClaim (configWhitelist config)+ claims = addIcepeakClaim access $ JWT.JWTClaimsSet+ { JWT.iss = Nothing+ , JWT.sub = Nothing+ , JWT.aud = Nothing+ , JWT.exp = fmap (\secs -> realToFrac secs + now) (configExpiresSeconds config) >>= JWT.numericDate+ , JWT.nbf = Nothing+ , JWT.iat = Nothing+ , JWT.jti = Nothing+ , JWT.unregisteredClaims = mempty+ }+ token = case configJwtSecret config of+ Nothing -> JWT.encodeUnsigned claims joseHeader+ Just key -> JWT.encodeSigned key joseHeader claims+ Text.putStrLn token++configParser :: EnvironmentConfig -> Parser Config+configParser environment = Config+ <$> optional (+ secretOption (long "jwt-secret"+ <> metavar "JWT_SECRET"+ <> environ "JWT_SECRET"+ <> short 's'+ <> help "Secret used for signing the JWT, defaults to the value of the JWT_SECRET environment variable if present. If no secret is passed, JWT tokens are not signed."))+ <*> optional(+ option auto (long "expires"+ <> short 'e'+ <> metavar "EXPIRES_SECONDS"+ <> help "Generate a token that expires in EXPIRES_SECONDS seconds from now."))+ <*> many (option authPathReader+ (long "path"+ <> short 'p'+ <> metavar "PATH:MODES"+ <> help "Adds the PATH to the whitelist, allowing the access modes MODES. MODES can be 'r' (read), 'w' (write) or 'rw' (read/write). This option may be used more than once."+ ))+ where+ environ var = foldMap value (lookup var environment)+ secretOption m = JWT.hmacSecret . Text.pack <$> strOption m++configInfo :: EnvironmentConfig -> ParserInfo Config+configInfo environment = info parser description+ where+ parser = helper <*> configParser environment+ description = fullDesc <>+ header "Icepeak Token Generator - Generates and signs JSON Web Tokens for authentication and authorization in Icepeak."++authPathReader :: ReadM AuthPath+authPathReader = eitherReader (go . Text.pack) where+ go input = let (pathtxt, modetxt) = Text.breakOn ":" input+ modeEither | modetxt == ":r" = Right [ModeRead]+ | modetxt == ":w" = Right [ModeWrite]+ | modetxt == ":rw" = Right [ModeRead, ModeWrite]+ | otherwise = Left $ "Invalid mode: " ++ Text.unpack modetxt+ pathComponents = filter (not . Text.null) $ Text.splitOn "/" pathtxt+ in AuthPath pathComponents <$> modeEither
+ icepeak.cabal view
@@ -0,0 +1,224 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.33.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: f5db6d897187842d6e4aa9c198fd2741665d7be7121d721ab1a1a9f5ac178418++name: icepeak+version: 0.7.1.0+synopsis: A fast JSON document store with push notification support.+description: Icepeak is a fast JSON document store with push notification support.+category: web+homepage: https://github.com/channable/icepeak+bug-reports: https://github.com/channable/icepeak/issues+author: Channable+maintainer: rkrzr+copyright: (c) 2020, Channable+license: BSD3+build-type: Simple++library+ exposed-modules:+ AccessControl+ Config+ Core+ HTTPMethodInvalid+ HttpServer+ JwtAuth+ JwtMiddleware+ Logger+ Metrics+ MetricsServer+ Persistence+ SentryLogging+ Server+ Store+ Subscription+ WebsocketServer+ other-modules:+ Paths_icepeak+ hs-source-dirs:+ src+ ghc-options: -Wall -Wno-orphans -Wno-unused-top-binds -O2 -fno-ignore-asserts -funbox-strict-fields+ build-depends:+ aeson >=1.4.6 && <1.5+ , async >=2.2.2 && <2.3+ , base >=4.12.0 && <4.13+ , bytestring >=0.10.8 && <0.11+ , containers >=0.6.0 && <0.7+ , directory >=1.3.3 && <1.4+ , hashable >=1.2.0 && <1.4+ , http-types >=0.12.3 && <0.13+ , jwt >=0.10.0 && <0.11+ , monad-logger >=0.3.31 && <0.4+ , mtl >=2.2.2 && <2.3+ , network >=2.8.0 && <3.2+ , optparse-applicative >=0.14.0 && <0.16+ , prometheus-client >=1.0.0 && <1.1+ , prometheus-metrics-ghc >=1.0.0 && <1.1+ , random >=1.1 && <1.2+ , raven-haskell >=0.1.2 && <0.2+ , scotty >=0.11.5 && <0.12+ , securemem >=0.1.10 && <0.2+ , sqlite-simple >=0.4.16 && <0.5+ , stm >=2.5.0 && <2.6+ , text >=1.2.3 && <1.3+ , time >=1.8.0 && <1.9+ , unix >=2.7.2 && <2.8+ , unordered-containers >=0.2.10 && <0.3+ , uuid >=1.3.13 && <1.4+ , wai >=3.2.2 && <3.3+ , wai-extra >=3.0.29 && <3.1+ , wai-middleware-prometheus >=1.0.0 && <1.1+ , wai-websockets >=3.0.1 && <3.1+ , warp >=3.3.0 && <3.4+ , websockets >=0.12.7 && <0.13+ default-language: Haskell2010++executable icepeak+ main-is: Main.hs+ other-modules:+ Paths_icepeak+ hs-source-dirs:+ app/Icepeak+ ghc-options: -Wall -O2 -threaded -rtsopts "-with-rtsopts=-N -I0"+ build-depends:+ aeson >=1.4.6 && <1.5+ , async >=2.2.2 && <2.3+ , base >=4.12.0 && <4.13+ , bytestring >=0.10.8 && <0.11+ , containers >=0.6.0 && <0.7+ , directory >=1.3.3 && <1.4+ , hashable >=1.2.0 && <1.4+ , http-types >=0.12.3 && <0.13+ , icepeak+ , jwt >=0.10.0 && <0.11+ , monad-logger >=0.3.31 && <0.4+ , mtl >=2.2.2 && <2.3+ , network >=2.8.0 && <3.2+ , optparse-applicative >=0.14.0 && <0.16+ , prometheus-client >=1.0.0 && <1.1+ , prometheus-metrics-ghc >=1.0.0 && <1.1+ , random >=1.1 && <1.2+ , raven-haskell >=0.1.2 && <0.2+ , scotty >=0.11.5 && <0.12+ , securemem >=0.1.10 && <0.2+ , sqlite-simple >=0.4.16 && <0.5+ , stm >=2.5.0 && <2.6+ , text >=1.2.3 && <1.3+ , time >=1.8.0 && <1.9+ , unix >=2.7.2 && <2.8+ , unordered-containers >=0.2.10 && <0.3+ , uuid >=1.3.13 && <1.4+ , wai >=3.2.2 && <3.3+ , wai-extra >=3.0.29 && <3.1+ , wai-middleware-prometheus >=1.0.0 && <1.1+ , wai-websockets >=3.0.1 && <3.1+ , warp >=3.3.0 && <3.4+ , websockets >=0.12.7 && <0.13+ default-language: Haskell2010++executable icepeak-token-gen+ main-is: Main.hs+ other-modules:+ Paths_icepeak+ hs-source-dirs:+ app/IcepeakTokenGen+ ghc-options: -Wall+ build-depends:+ aeson >=1.4.6 && <1.5+ , async >=2.2.2 && <2.3+ , base >=4.12.0 && <4.13+ , bytestring >=0.10.8 && <0.11+ , containers >=0.6.0 && <0.7+ , directory >=1.3.3 && <1.4+ , hashable >=1.2.0 && <1.4+ , http-types >=0.12.3 && <0.13+ , icepeak+ , jwt >=0.10.0 && <0.11+ , monad-logger >=0.3.31 && <0.4+ , mtl >=2.2.2 && <2.3+ , network >=2.8.0 && <3.2+ , optparse-applicative >=0.14.0 && <0.16+ , prometheus-client >=1.0.0 && <1.1+ , prometheus-metrics-ghc >=1.0.0 && <1.1+ , random >=1.1 && <1.2+ , raven-haskell >=0.1.2 && <0.2+ , scotty >=0.11.5 && <0.12+ , securemem >=0.1.10 && <0.2+ , sqlite-simple >=0.4.16 && <0.5+ , stm >=2.5.0 && <2.6+ , text >=1.2.3 && <1.3+ , time >=1.8.0 && <1.9+ , unix >=2.7.2 && <2.8+ , unordered-containers >=0.2.10 && <0.3+ , uuid >=1.3.13 && <1.4+ , wai >=3.2.2 && <3.3+ , wai-extra >=3.0.29 && <3.1+ , wai-middleware-prometheus >=1.0.0 && <1.1+ , wai-websockets >=3.0.1 && <3.1+ , warp >=3.3.0 && <3.4+ , websockets >=0.12.7 && <0.13+ default-language: Haskell2010++test-suite spec+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ other-modules:+ AccessControlSpec+ ApiSpec+ CoreSpec+ JwtSpec+ OrphanInstances+ PersistenceSpec+ RequestSpec+ SocketSpec+ StoreSpec+ SubscriptionTreeSpec+ Paths_icepeak+ hs-source-dirs:+ tests+ ghc-options: -Wall -Wno-orphans+ build-depends:+ QuickCheck+ , aeson >=1.4.6 && <1.5+ , async >=2.2.2 && <2.3+ , base >=4.12.0 && <4.13+ , bytestring >=0.10.8 && <0.11+ , containers >=0.6.0 && <0.7+ , directory >=1.3.3 && <1.4+ , hashable >=1.2.0 && <1.4+ , hspec+ , hspec-core+ , hspec-expectations+ , hspec-wai+ , http-types >=0.12.3 && <0.13+ , icepeak+ , jwt >=0.10.0 && <0.11+ , monad-logger >=0.3.31 && <0.4+ , mtl >=2.2.2 && <2.3+ , network >=2.8.0 && <3.2+ , optparse-applicative >=0.14.0 && <0.16+ , prometheus-client >=1.0.0 && <1.1+ , prometheus-metrics-ghc >=1.0.0 && <1.1+ , quickcheck-instances+ , random >=1.1 && <1.2+ , raven-haskell >=0.1.2 && <0.2+ , scotty >=0.11.5 && <0.12+ , securemem >=0.1.10 && <0.2+ , sqlite-simple >=0.4.16 && <0.5+ , stm >=2.5.0 && <2.6+ , text >=1.2.3 && <1.3+ , time >=1.8.0 && <1.9+ , unix >=2.7.2 && <2.8+ , unordered-containers >=0.2.10 && <0.3+ , uuid >=1.3.13 && <1.4+ , wai >=3.2.2 && <3.3+ , wai-extra >=3.0.29 && <3.1+ , wai-middleware-prometheus >=1.0.0 && <1.1+ , wai-websockets >=3.0.1 && <3.1+ , warp >=3.3.0 && <3.4+ , websockets >=0.12.7 && <0.13+ default-language: Haskell2010
+ src/AccessControl.hs view
@@ -0,0 +1,97 @@+{-# LANGUAGE OverloadedStrings #-}+-- | This module defines the kinds of permissions used in icepeak and provides+-- functions checking for sufficient permissions for certain operations.+module AccessControl+ ( AccessMode (..)+ , AuthPath (..)+ , IcepeakClaim (..)+ , Path+ , allowEverything+ , accessModeToText+ , textToAccessMode+ , isAuthorizedByClaim+ ) where++import Data.Aeson ((.:), (.=))+import qualified Data.Aeson as Aeson+import qualified Data.List as List+import Data.Text (Text)++import Store (Path)++-- * Claim datatypes++-- | Defines the structure of a JWT claim for Icepeak.+data IcepeakClaim = IcepeakClaim+ { icepeakClaimWhitelist :: [AuthPath]+ -- ^ The whitelist containing all authorizations.+ } deriving (Read, Show, Eq, Ord)++data AuthPath = AuthPath+ { authPathPrefix :: Path+ -- ^ The prefix of all the paths to which this authorization applies.+ , authPathModes :: [AccessMode]+ -- ^ The modes that are authorized on this path prefix.+ } deriving (Read, Show, Eq, Ord)++-- | Different modes for accessing the JSON store+data AccessMode = ModeRead | ModeWrite+ deriving (Read, Show, Eq, Ord, Enum, Bounded)+++-- | A claim that allows all operations.+allowEverything :: IcepeakClaim+allowEverything = IcepeakClaim [AuthPath [] [minBound..maxBound]]++-- * Authorization++-- | Check whether accessing the given path with the given mode is authorized by+-- the supplied claim.+isAuthorizedByClaim :: IcepeakClaim -> Path -> AccessMode -> Bool+isAuthorizedByClaim claim path mode = any allows (icepeakClaimWhitelist claim) where+ allows (AuthPath prefix modes) = List.isPrefixOf prefix path && mode `elem` modes+++-- * JSON encoding and decoding++accessModeToText :: AccessMode -> Text+accessModeToText mode = case mode of+ ModeRead -> "read"+ ModeWrite -> "write"++textToAccessMode :: Text -> Maybe AccessMode+textToAccessMode mode+ | mode == "read" = Just ModeRead+ | mode == "write" = Just ModeWrite+ | otherwise = Nothing++instance Aeson.ToJSON AccessMode where+ toJSON = Aeson.String . accessModeToText++instance Aeson.FromJSON AccessMode where+ parseJSON = Aeson.withText "mode string" $ \txt -> case textToAccessMode txt of+ Nothing -> fail "Invalid mode value."+ Just m -> pure m++instance Aeson.ToJSON AuthPath where+ toJSON (AuthPath prefix modes) = Aeson.object+ [ "prefix" .= prefix+ , "modes" .= modes ]++instance Aeson.FromJSON AuthPath where+ parseJSON = Aeson.withObject "path and modes" $ \v -> AuthPath+ <$> v .: "prefix"+ <*> v .: "modes"++instance Aeson.ToJSON IcepeakClaim where+ toJSON claim = Aeson.object+ [ "version" .= (1 :: Int)+ , "whitelist" .= icepeakClaimWhitelist claim+ ]++instance Aeson.FromJSON IcepeakClaim where+ parseJSON = Aeson.withObject "icepeak claim" $ \v -> do+ version <- v .: "version"+ if version == (1 :: Int)+ then IcepeakClaim <$> v .: "whitelist"+ else fail "unsupported version"
+ src/Config.hs view
@@ -0,0 +1,154 @@+module Config (+ Config (..),+ MetricsConfig (..),+ StorageBackend (..),+ periodicSyncingEnabled,+ configInfo,+) where++import Control.Applicative (optional)+import Data.Semigroup ((<>))+import Options.Applicative+import qualified Network.Wai.Handler.Warp as Warp+import qualified Text.Read as Read+import qualified Data.Char as Char+import Data.Maybe (isJust)+import Data.String (fromString)+import qualified Data.List as List+import qualified Data.Text as Text+import qualified Web.JWT as JWT+++data StorageBackend = File | Sqlite++-- command-line arguments+data Config = Config+ { configDataFile :: Maybe FilePath+ , configPort :: Int+ -- | Enables the use of JWT for authorization in JWT.+ , configEnableJwtAuth :: Bool+ -- | The secret used for verifying the JWT signatures. If no secret is+ -- specified even though JWT authorization is enabled, tokens will still be+ -- used, but not be verified.+ , configJwtSecret :: Maybe JWT.Signer+ , configMetricsEndpoint :: Maybe MetricsConfig+ , configQueueCapacity :: Word+ , configSyncIntervalMicroSeconds :: Maybe Int+ -- | Enable journaling, only in conjunction with periodic syncing+ , configEnableJournaling :: Bool+ -- | Indicates that the sentry logging is disabled, can be used to overwrite+ -- ```configSentryDSN``` or the environment variable+ , configDisableSentryLogging :: Bool+ -- | The SENTRY_DSN key that Sentry uses to communicate, if not set, use Nothing.+ -- Just indicates that a key is given.+ , configSentryDSN :: Maybe String+ , configStorageBackend :: StorageBackend+ }++data MetricsConfig = MetricsConfig+ { metricsConfigHost :: Warp.HostPreference+ , metricsConfigPort :: Warp.Port+ }++periodicSyncingEnabled :: Config -> Bool+periodicSyncingEnabled = isJust . configSyncIntervalMicroSeconds++-- Parsing of command-line arguments++type EnvironmentConfig = [(String, String)]++configParser :: EnvironmentConfig -> Parser Config+configParser environment = Config+ -- Note: If no --data-file is given we default either to icepeak.json or icepeak.db+ <$> optional (strOption (long "data-file" <>+ metavar "DATA_FILE" <>+ help "File where data is persisted to. Default: icepeak.json"))+ <*> option auto (long "port" <>+ metavar "PORT" <>+ maybe (value 3000) value (readFromEnvironment "ICEPEAK_PORT") <>+ help "Port to listen on, defaults to the value of the ICEPEAK_PORT environment variable if present, or 3000 if not")+ <*> switch (long "enable-jwt-auth" <>+ help "Enable authorization using JSON Web Tokens.")+ <*> optional (secretOption (+ long "jwt-secret" <>+ metavar "JWT_SECRET" <>+ environ "JWT_SECRET" <>+ help "Secret used for JWT verification, defaults to the value of the JWT_SECRET environment variable if present. If no secret is passed, JWT tokens are not checked for validity."))+ <*> optional (option metricsConfigReader+ (long "metrics" <>+ metavar "HOST:PORT" <>+ help "If provided, Icepeak collects various metrics and provides them to Prometheus on the given endpoint."+ ))+ <*> option auto+ (long "queue-capacity" <>+ metavar "INTEGER" <>+ value 256 <>+ help ("Smaller values decrease the risk of data loss during a crash, while " <>+ "higher values result in more requests being accepted in rapid succession."))+ <*> optional (option timeDurationReader+ (long "sync-interval" <>+ metavar "DURATION" <>+ help ("If supplied, data is only persisted to disc every DURATION time units." <>+ "The units 'm' (minutes), 's' (seconds) and 'ms' (milliseconds) can be used. " <>+ "When omitting this argument, data is persisted after every modification")))+ <*> switch (long "journaling" <>+ help "Enable journaling. This only has an effect when periodic syncing is enabled.")+ <*> switch (long "disable-sentry-logging" <>+ help "Disable error logging via Sentry")+ <*> optional (strOption (+ long "sentry-dsn" <>+ metavar "SENTRY_DSN" <>+ environ "SENTRY_DSN" <>+ help "Sentry DSN used for Sentry logging, defaults to the value of the SENTRY_DSN environment variable if present. If no secret is passed, Sentry logging will be disabled."))+ <*> storageBackend++ where+ environ var = foldMap value (lookup var environment)++ readFromEnvironment :: Read a => String -> Maybe a+ readFromEnvironment var = lookup var environment >>= Read.readMaybe++ secretOption m = JWT.hmacSecret . Text.pack <$> strOption m++configInfo :: EnvironmentConfig -> ParserInfo Config+configInfo environment = info parser description+ where+ parser = helper <*> configParser environment+ description = fullDesc <>+ header "Icepeak - Fast Json document store with push notification support."++-- * Parsers++storageBackend :: Parser StorageBackend+storageBackend = fileBackend <|> sqliteBackend++fileBackend :: Parser StorageBackend+-- The first 'File' here is the default value. We want --file to be used by default, when nothing+-- is specified on the command-line. This ensures backwards-compatibility.+fileBackend = flag File File (long "file" <> help "Use a file as the storage backend." )++sqliteBackend :: Parser StorageBackend+sqliteBackend = flag' Sqlite (long "sqlite" <> help "Use a sqlite file as the storage backend." )++-- * Reader functions++metricsConfigReader :: ReadM MetricsConfig+metricsConfigReader = eitherReader $ \input ->+ case List.break (== ':') input of+ (hostStr, ':':portStr) -> MetricsConfig (fromString hostStr) <$> Read.readEither portStr+ (_, _) -> Left "no port specified"++-- | Read an option as a time duration in microseconds.+timeDurationReader :: ReadM Int+timeDurationReader = eitherReader $ \input ->+ case List.break Char.isLetter input of+ ("", _) -> Left "no amount specified"+ (amount, unit) -> case lookup unit units of+ Nothing -> Left "invalid unit"+ Just factor -> fmap (* factor) $ Read.readEither amount+ where+ -- defines the available units and how they convert to microseconds+ units = [ ("s", 1000000)+ , ("ms", 1000)+ , ("m", 60000000)+ ]
+ src/Core.hs view
@@ -0,0 +1,174 @@+{-# LANGUAGE OverloadedStrings #-}+module Core+(+ Core (..), -- TODO: Expose only put for clients.+ EnqueueResult (..),+ Command (..),+ ServerState,+ Updated (..),+ enqueueCommand,+ tryEnqueueCommand,+ getCurrentValue,+ withCoreMetrics,+ lookup,+ newCore,+ postQuit,+ runCommandLoop,+ runSyncTimer+)+where++import Control.Concurrent (threadDelay)+import Control.Concurrent.MVar (MVar, newMVar, putMVar)+import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TBQueue (TBQueue, newTBQueueIO, readTBQueue, writeTBQueue, isFullTBQueue)+import Control.Concurrent.STM.TVar (TVar, newTVarIO)+import Control.Monad (forever, unless)+import Control.Monad.IO.Class+import Data.Aeson (Value (..))+import Data.Foldable (forM_)+import Data.Traversable (for)+import Data.UUID (UUID)+import Prelude hiding (log, writeFile)++import qualified Network.WebSockets as WS++import Config (Config (..), periodicSyncingEnabled)+import Logger (Logger)+import Store (Path, Modification (..))+import Subscription (SubscriptionTree, empty)+import Persistence (PersistentValue, PersistenceConfig (..))++import qualified Store+import qualified Persistence+import qualified Metrics++-- | Defines the kinds of commands that are handled by the event loop of the Core.+data Command+ = Sync+ -- ^ The @Sync@ command causes the core to write the JSON value to disk.+ | Modify Modification (Maybe (MVar ()))+ -- ^ The @Modify@ command applies a modification (writing or deleting) to the JSON value.+ -- The optional MVar is used to signal that the command has been processed by the core.+ | Stop+ -- ^ The @Stop@ command causes the event loop of the Core to exit.+ deriving (Eq)++-- The main value has been updated at the given path. The payload contains the+-- entire new value. (So not only the inner value at the updated path.)+data Updated = Updated Path Value deriving (Eq, Show)++data EnqueueResult = Enqueued | Dropped+ deriving (Show, Eq, Ord, Enum, Bounded)++data Core = Core+ { coreCurrentValue :: PersistentValue+ -- the "dirty" flag is set to True whenever the core value has been modified+ -- and is reset to False when it is persisted.+ , coreValueIsDirty :: TVar Bool+ , coreQueue :: TBQueue Command+ , coreUpdates :: TBQueue (Maybe Updated)+ , coreClients :: MVar ServerState+ , coreLogger :: Logger+ , coreConfig :: Config+ , coreMetrics :: Maybe Metrics.IcepeakMetrics+ }++type ServerState = SubscriptionTree UUID WS.Connection++newServerState :: ServerState+newServerState = empty++-- | Try to initialize the core. This loads the database and sets up the internal data structures.+newCore :: Config -> Logger -> Maybe Metrics.IcepeakMetrics -> IO (Either String Core)+newCore config logger metrics = do+ let queueCapacity = fromIntegral . configQueueCapacity $ config+ -- load the persistent data from disk+ let filePath = Persistence.getDataFile (configStorageBackend config) (configDataFile config)+ journalFile+ | configEnableJournaling config+ && periodicSyncingEnabled config = Just $ filePath ++ ".journal"+ | otherwise = Nothing+ eitherValue <- Persistence.loadFromBackend (configStorageBackend config) PersistenceConfig+ { pcDataFile = filePath+ , pcJournalFile = journalFile+ , pcLogger = logger+ , pcMetrics = metrics+ }+ for eitherValue $ \value -> do+ -- create synchronization channels+ tdirty <- newTVarIO False+ tqueue <- newTBQueueIO queueCapacity+ tupdates <- newTBQueueIO queueCapacity+ tclients <- newMVar newServerState+ pure (Core value tdirty tqueue tupdates tclients logger config metrics)++-- Tell the put handler loop and the update handler to quit.+postQuit :: Core -> IO ()+postQuit core = do+ atomically $ do+ writeTBQueue (coreQueue core) Stop+ writeTBQueue (coreUpdates core) Nothing++-- | Try to enqueue a command. It succeeds if the queue is not full, otherwise,+-- nothing is changed. This should be used for non-critical commands that can+-- also be retried later.+tryEnqueueCommand :: Command -> Core -> IO EnqueueResult+tryEnqueueCommand cmd core = atomically $ do+ isFull <- isFullTBQueue (coreQueue core)+ unless isFull $ writeTBQueue (coreQueue core) cmd+ pure $ if isFull then Dropped else Enqueued++-- | Enqueue a command. Blocks if the queue is full. This is used by the sync+-- timer to make sure the sync commands are actually enqueued. In general,+-- whenever it is critical that a command is executed eventually (when reaching+-- the front of the queue), this function should be used.+enqueueCommand :: Command -> Core -> IO ()+enqueueCommand cmd core = atomically $ writeTBQueue (coreQueue core) cmd++getCurrentValue :: Core -> Path -> IO (Maybe Value)+getCurrentValue core path =+ fmap (Store.lookup path) $ atomically $ Persistence.getValue $ coreCurrentValue core++withCoreMetrics :: MonadIO m => Core -> (Metrics.IcepeakMetrics -> IO ()) -> m ()+withCoreMetrics core act = liftIO $ forM_ (coreMetrics core) act++-- | Drain the command queue and execute them. Changes are published to all+-- subscribers. This function returns when executing the 'Stop' command from the+-- queue.+runCommandLoop :: Core -> IO ()+runCommandLoop core = go+ where+ config = coreConfig core+ currentValue = coreCurrentValue core+ storageBackend = configStorageBackend config+ go = do+ command <- atomically $ readTBQueue (coreQueue core)+ case command of+ Modify op maybeNotifyVar -> do+ Persistence.apply op currentValue+ postUpdate (Store.modificationPath op) core+ -- when periodic syncing is disabled, data is persisted after every modification+ unless (periodicSyncingEnabled $ coreConfig core) $+ Persistence.syncToBackend storageBackend currentValue+ mapM_ (`putMVar` ()) maybeNotifyVar+ go+ Sync -> do+ Persistence.syncToBackend storageBackend currentValue+ go+ Stop -> Persistence.syncToBackend storageBackend currentValue++-- | Post an update to the core's update queue (read by the websocket subscribers)+postUpdate :: Path -> Core -> IO ()+postUpdate path core = atomically $ do+ value <- Persistence.getValue (coreCurrentValue core)+ writeTBQueue (coreUpdates core) (Just $ Updated path value)++-- | Periodically send a 'Sync' command to the 'Core' if enabled in the core+-- configuration.+runSyncTimer :: Core -> IO ()+runSyncTimer core = mapM_ go (configSyncIntervalMicroSeconds $ coreConfig core)+ where+ go interval = forever $ do+ enqueueCommand Sync core+ threadDelay interval
+ src/HTTPMethodInvalid.hs view
@@ -0,0 +1,29 @@+module HTTPMethodInvalid (canonicalizeHTTPMethods, limitHTTPMethods) where++import qualified Network.Wai as Wai+import qualified Network.HTTP.Types as HTTP+import qualified Network.HTTP.Types.Method as Method+import qualified Data.ByteString.Char8 as ByteString+import qualified Data.ByteString.Lazy.Char8 as LBS++-- | Checks that the HTTP method is one of the StdMethods.+-- StdMethod: HTTP standard method (as defined by RFC 2616, and PATCH which is defined by RFC 5789).+-- Otherwise sets the HTTP method to INVALID.+-- post: HTTP method is canonicalized+canonicalizeHTTPMethods :: Wai.Middleware+canonicalizeHTTPMethods app request respond = do+ let method = Wai.requestMethod request+ parsedMethod = either (\_ -> invalid) (Method.renderStdMethod) (Method.parseMethod method)+ request' = request { Wai.requestMethod = parsedMethod }+ app request' respond++-- | Early exit for INVALID HTTP methods.+-- pre: HTTP method is canonicalized.+limitHTTPMethods :: Wai.Middleware+limitHTTPMethods app request respond =+ if Wai.requestMethod request == invalid+ then respond (Wai.responseLBS HTTP.badRequest400 [(HTTP.hContentType, (ByteString.pack "application/json"))] (LBS.pack "{}"))+ else app request respond++invalid :: Method.Method+invalid = ByteString.pack "INVALID"
+ src/HttpServer.hs view
@@ -0,0 +1,93 @@+{-# LANGUAGE OverloadedStrings #-}++module HttpServer (new) where++import Control.Concurrent.MVar (newEmptyMVar, takeMVar)+import Control.Monad+import Control.Monad.IO.Class+import Data.Foldable (for_)+import Data.Traversable (for)+import Data.Text (pack)+import Network.HTTP.Types+import Network.Wai (Application)+import Web.Scotty (delete, get, json, jsonData, put, regex, middleware, request, scottyApp, status, ActionM)++import qualified Data.Text.Lazy as LText+import qualified Network.Wai as Wai+import qualified Web.Scotty.Trans as Scotty++import HTTPMethodInvalid (canonicalizeHTTPMethods,limitHTTPMethods)+import JwtMiddleware (jwtMiddleware)+import Core (Core (..), EnqueueResult (..))+import Config (Config (..))+import Logger (postLog, LogLevel(LogError))+import qualified Store+import qualified Core+import qualified Metrics++new :: Core -> IO Application+new core =+ scottyApp $ do+ -- First we check whether the request HTTP method is a recognised HTTP method.+ -- Any arbitrary ByteString is accepted as a request method and we store those + -- in the exposed metrics, this is a DoS vector.+ middleware canonicalizeHTTPMethods+ -- Second middleware is the metrics middleware in order to intercept+ -- all requests and their corresponding responses+ forM_ (coreMetrics core) $ middleware . metricsMiddleware+ -- Exit on unknown HTTP verb after the request has been stored in the metrics.+ middleware limitHTTPMethods+ -- Use the Sentry logger if available+ -- Scottys error handler will only catch errors that are thrown from within+ -- a ```liftAndCatchIO``` function.+ Scotty.defaultHandler (\e -> do+ liftIO $ postLog (coreLogger core) LogError . pack . show $ e+ status status503+ Scotty.text "Internal server error"+ )++ when (configEnableJwtAuth $ coreConfig core) $+ middleware $ jwtMiddleware $ configJwtSecret $ coreConfig core++ get (regex "^") $ do+ path <- Wai.pathInfo <$> request+ maybeValue <- Scotty.liftAndCatchIO $ Core.getCurrentValue core path+ maybe (status status404) json maybeValue++ put (regex "^") $ do+ path <- Wai.pathInfo <$> request+ value <- jsonData+ result <- postModification core (Store.Put path value)+ buildResponse result++ delete (regex "^") $ do+ path <- Wai.pathInfo <$> request+ result <- postModification core (Store.Delete path)+ buildResponse result+++-- | Enqueue modification and wait for it to be processed, if desired by the client.+postModification :: (Scotty.ScottyError e, MonadIO m) => Core -> Store.Modification -> Scotty.ActionT e m EnqueueResult+postModification core op = do+ -- the parameter is parsed as type (), therefore only presence or absence is important+ durable <- maybeParam "durable"+ waitVar <- Scotty.liftAndCatchIO $ for durable $ \() -> newEmptyMVar+ result <- Scotty.liftAndCatchIO $ Core.tryEnqueueCommand (Core.Modify op waitVar) core+ when (result == Enqueued) $+ Scotty.liftAndCatchIO $ for_ waitVar $ takeMVar+ pure result++buildResponse :: EnqueueResult -> ActionM ()+buildResponse Enqueued = status accepted202+buildResponse Dropped = status serviceUnavailable503++metricsMiddleware :: Metrics.IcepeakMetrics -> Wai.Middleware+metricsMiddleware metrics app req sendResponse = app req sendWithMetrics+ where+ sendWithMetrics resp = do+ Metrics.notifyRequest (Wai.requestMethod req) (Wai.responseStatus resp) metrics+ sendResponse resp++maybeParam :: (Scotty.Parsable a, Scotty.ScottyError e, Monad m) => LText.Text -> Scotty.ActionT e m (Maybe a)+maybeParam name = fmap (parseMaybe <=< lookup name) Scotty.params where+ parseMaybe = either (const Nothing) Just . Scotty.parseParam
+ src/JwtAuth.hs view
@@ -0,0 +1,104 @@+{-# LANGUAGE OverloadedStrings #-}+-- | This module contains all the web framework independent code for parsing and verifying+-- JSON Web Tokens.+module JwtAuth where++import Control.Monad ((<=<))+import qualified Data.Aeson as Aeson+import qualified Data.Aeson.Types as Aeson+import Data.Bifunctor (first)+import qualified Data.ByteString as SBS+import qualified Data.Map.Strict as Map+import qualified Data.Text.Encoding as Text+import Data.Time.Clock.POSIX (POSIXTime)+import Web.JWT (JWT, UnverifiedJWT, VerifiedJWT)+import qualified Web.JWT as JWT++import AccessControl++-- * Token verification++data VerificationError+ = TokenUsedTooEarly+ | TokenExpired+ | TokenInvalid+ | TokenNotFound+ | TokenSignatureInvalid+ deriving (Show, Eq)++-- | Check that a token is valid at the given time for the given secret.+verifyToken :: POSIXTime -> JWT.Signer -> SBS.ByteString -> Either VerificationError (JWT VerifiedJWT)+verifyToken now secret = verifyNotBefore now+ <=< verifyExpiry now+ <=< verifySignature secret+ <=< decodeToken++-- | Verify that the token is not used before it was issued.+verifyNotBefore :: POSIXTime -> JWT VerifiedJWT -> Either VerificationError (JWT VerifiedJWT)+verifyNotBefore now token =+ case JWT.nbf . JWT.claims $ token of+ Nothing -> Right token+ Just notBefore ->+ if now <= JWT.secondsSinceEpoch notBefore+ then Left TokenUsedTooEarly+ else Right token++-- | Verify that the token is not used after is has expired.+verifyExpiry :: POSIXTime -> JWT VerifiedJWT -> Either VerificationError (JWT VerifiedJWT)+verifyExpiry now token =+ case JWT.exp . JWT.claims $ token of+ Nothing -> Right token+ Just expiry ->+ if now > JWT.secondsSinceEpoch expiry+ then Left TokenExpired+ else Right token++-- | Verify that the token contains a valid signature.+verifySignature :: JWT.Signer -> JWT UnverifiedJWT -> Either VerificationError (JWT VerifiedJWT)+verifySignature secret token =+ case JWT.verify secret token of+ Nothing -> Left TokenSignatureInvalid+ Just token' -> Right token'++decodeToken :: SBS.ByteString -> Either VerificationError (JWT UnverifiedJWT)+decodeToken bytes =+ case JWT.decode (Text.decodeUtf8 bytes) of+ Nothing -> Left TokenInvalid+ Just token -> Right token++-- * Claim parsing++data TokenError+ = VerificationError VerificationError -- ^ JWT could not be verified.+ | ClaimError String -- ^ The claims do not fit the schema.+ deriving (Show, Eq)++-- | Verify the token and extract the icepeak claim from it.+extractClaim :: POSIXTime -> JWT.Signer -> SBS.ByteString -> Either TokenError IcepeakClaim+extractClaim now secret tokenBytes = do+ jwt <- first VerificationError $ verifyToken now secret tokenBytes+ claim <- first ClaimError $ getIcepeakClaim jwt+ pure claim++-- | Extract the icepeak claim from the token without verifying it.+extractClaimUnverified :: SBS.ByteString -> Either TokenError IcepeakClaim+extractClaimUnverified tokenBytes = do+ jwt <- first VerificationError $ decodeToken tokenBytes+ claim <- first ClaimError $ getIcepeakClaim jwt+ pure claim++getIcepeakClaim :: JWT r -> Either String IcepeakClaim+getIcepeakClaim token = do+ let (JWT.ClaimsMap claimsMap) = JWT.unregisteredClaims $ JWT.claims token+ maybeClaim = Map.lookup "icepeak" claimsMap+ claimJson <- maybe (Left "Icepeak claim missing.") Right maybeClaim+ Aeson.parseEither Aeson.parseJSON claimJson++-- * Token generation++-- | Add the icepeak claim to a set of JWT claims.+addIcepeakClaim :: IcepeakClaim -> JWT.JWTClaimsSet -> JWT.JWTClaimsSet+addIcepeakClaim claim claims = claims+ { JWT.unregisteredClaims = newClaimsMap <> JWT.unregisteredClaims claims }+ where+ newClaimsMap = JWT.ClaimsMap $ Map.fromList [("icepeak", Aeson.toJSON claim)]
+ src/JwtMiddleware.hs view
@@ -0,0 +1,117 @@+{-# LANGUAGE OverloadedStrings #-}+-- | This module provides functionality for verifying the JSON Web Tokens in a wai setting.+module JwtMiddleware where++import Control.Applicative+import Control.Monad+import Data.Aeson ((.=))+import qualified Data.Aeson as Aeson+import qualified Data.ByteString as SBS+import qualified Data.ByteString.Lazy as LBS+import qualified Data.List as List+import qualified Data.Text as Text+import Data.Time.Clock.POSIX (POSIXTime)+import qualified Data.Time.Clock.POSIX as Clock+import qualified Network.HTTP.Types as Http+import qualified Network.Wai as Wai+import qualified Web.JWT as JWT++import AccessControl+import JwtAuth+import Store (Path)++-- | Defines the kinds of errors that cause authorization to fail.+data AuthError+ = TokenError TokenError+ -- ^ Authorization was denied due to an invalid token.+ | OperationNotAllowed+ -- ^ Authorization was denied because the operation is not allowed by the token.++-- | Result of checking authorization+data AuthResult+ = AuthRejected AuthError+ -- ^ Authorization was denied because of the specified reason+ | AuthAccepted+ -- ^ Authorization was successful++-- * Requests++-- | Check whether accessing the given path with the given mode is authorized by+-- the token supplied in the request headers or query string (which may not be+-- present, then failing the check).+isRequestAuthorized :: Http.RequestHeaders -> Http.Query -> POSIXTime -> Maybe JWT.Signer -> Path -> AccessMode -> AuthResult+isRequestAuthorized headers query now maybeSecret path mode =+ case getRequestClaim headers query now maybeSecret of+ Left err -> AuthRejected (TokenError err)+ Right claim | isAuthorizedByClaim claim path mode+ -> AuthAccepted+ | otherwise+ -> AuthRejected OperationNotAllowed++-- | Extract the JWT claim from the request.+getRequestClaim :: Http.RequestHeaders -> Http.Query -> POSIXTime -> Maybe JWT.Signer -> Either TokenError IcepeakClaim+getRequestClaim headers query now maybeSecret =+ let getTokenBytes = maybe (Left $ VerificationError TokenNotFound) Right (findTokenBytes headers query)+ in case maybeSecret of+ Nothing ->+ -- authorization is enabled, but no secret provided, accept all tokens+ getTokenBytes >>= extractClaimUnverified+ Just secret -> getTokenBytes >>= extractClaim now secret++-- | Lookup a token, first in the @Authorization@ header of the request, then+-- falling back to the @access_token@ query parameter.+findTokenBytes :: Http.RequestHeaders -> Http.Query -> Maybe SBS.ByteString+findTokenBytes headers query = headerToken headers <|> queryToken query++-- | Look up a token from the @Authorization@ header.+-- Header should be in the format @Bearer <token>@.+headerToken :: Http.RequestHeaders -> Maybe SBS.ByteString+headerToken =+ SBS.stripPrefix "Bearer " <=< List.lookup Http.hAuthorization++-- | Look up a token from the @access_token@ query parameter+queryToken :: Http.Query -> Maybe SBS.ByteString+queryToken = join . lookup "access_token"++-- * Responses++instance Aeson.ToJSON AuthError where+ toJSON aerr = case aerr of+ TokenError terr -> case terr of+ ClaimError ce -> Aeson.object [ "error" .= ce ]+ VerificationError ve | ve `elem` [TokenInvalid, TokenNotFound]+ -> Aeson.object [ "error" .= Text.pack "invalid token format" ]+ _ -> Aeson.object [ "data" .= Aeson.Null ]+ OperationNotAllowed -> Aeson.object [ "error" .= Text.pack "not allowed" ]++-- | Generate a 401 Unauthorized response for a given authorization error.+errorResponseBody :: AuthError -> LBS.ByteString+errorResponseBody = Aeson.encode++-- * Middleware++jwtMiddleware :: Maybe JWT.Signer -> Wai.Application -> Wai.Application+jwtMiddleware secret app req respond = do+ now <- Clock.getPOSIXTime+ case getRequestClaim headers query now secret of+ Left err -> rejectUnauthorized (TokenError err)+ Right claim | isAuthorized claim -> app req respond+ | otherwise -> rejectUnauthorized OperationNotAllowed+ where+ -- read request+ path = Wai.pathInfo req+ query = Wai.queryString req+ headers = Wai.requestHeaders req++ -- translate HTTP request methods to modes+ maybeMode | Wai.requestMethod req == Http.methodGet = Just ModeRead+ | Wai.requestMethod req == Http.methodPut = Just ModeWrite+ | Wai.requestMethod req == Http.methodDelete = Just ModeWrite+ | otherwise = Nothing++ isAuthorized claim = maybe False (isAuthorizedByClaim claim path) maybeMode++ rejectUnauthorized err = respond $ Wai.responseLBS+ Http.unauthorized401+ [(Http.hContentType, "application/json")]+ (Aeson.encode err)
+ src/Logger.hs view
@@ -0,0 +1,82 @@+module Logger+(+ Logger,+ LogRecord,+ LogQueue,+ LogLevel(..),+ newLogger,+ postLog,+ postLogBlocking,+ postStop,+ processLogRecords,+ loggerSentryService+)+where++import SentryLogging (getCrashLogger, logCrashMessage)+import Config (Config, configSentryDSN, configDisableSentryLogging, configQueueCapacity)++import Control.Monad (unless, when, forM_)+import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TBQueue (TBQueue, newTBQueue, readTBQueue, writeTBQueue, isFullTBQueue)+import Data.Text (Text, unpack)+import Prelude hiding (log)++import qualified System.Log.Raven.Types as Sentry++import qualified Data.Text.IO as T++type LogRecord = Text++data LogLevel = LogInfo | LogError+ deriving (Eq, Ord, Show, Read)++type LogQueue = TBQueue LogCommand++data Logger = Logger { loggerQueue :: LogQueue, loggerSentryService :: Maybe Sentry.SentryService }++data LogCommand = LogRecord LogLevel LogRecord | LogStop+ deriving (Eq, Ord, Show, Read)++newLogger :: Config -> IO Logger+newLogger config = Logger+ <$> createQueue+ <*> createSentryService+ where+ createQueue = atomically (newTBQueue (fromIntegral $ configQueueCapacity config))+ createSentryService+ | configDisableSentryLogging config = pure Nothing+ | otherwise = traverse getCrashLogger (configSentryDSN config)++-- | Post a non-essential log message to the queue. The message is discarded+-- when the queue is full.+postLog :: Logger -> LogLevel -> LogRecord -> IO ()+postLog logger level record = atomically $ do+ isFull <- isFullTBQueue (loggerQueue logger)+ unless isFull $ writeTBQueue (loggerQueue logger) (LogRecord level record)++-- | Post an essential log message to the queue. This function blocks when the+-- queue is full.+postLogBlocking :: Logger -> LogLevel -> LogRecord -> IO ()+postLogBlocking logger level record = atomically $+ writeTBQueue (loggerQueue logger) (LogRecord level record)++postStop :: Logger -> IO ()+postStop logger = atomically $ writeTBQueue (loggerQueue logger) LogStop++processLogRecords :: Logger -> IO ()+processLogRecords logger = go+ where+ go = do+ cmd <- atomically $ readTBQueue (loggerQueue logger)+ case cmd of+ LogRecord logLevel logRecord -> do+ T.putStrLn logRecord+ when (logLevel == LogError) (+ forM_+ (loggerSentryService logger)+ (\service -> logCrashMessage "Icepeak" service (unpack logRecord))+ )+ go+ -- stop the loop when asked so+ LogStop -> pure ()
+ src/Metrics.hs view
@@ -0,0 +1,76 @@+{-# LANGUAGE OverloadedStrings #-}+module Metrics where++import Data.Text (Text, pack)+import Data.Text.Encoding (decodeUtf8With)+import Data.Text.Encoding.Error (lenientDecode)+import Prometheus (Counter, Gauge, Info (..), MonadMonitor, Vector, addCounter, counter, decGauge,+ gauge, incCounter, incGauge, register, setGauge, vector, withLabel)++import qualified Network.HTTP.Types as Http+++type HttpMethodLabel = Text+type HttpStatusCode = Text++-- We want to store for each (HTTP method, HTTP status code) pair how many times it has been called+type HttpRequestCounter = Vector (HttpMethodLabel, HttpStatusCode) Counter++countHttpRequest :: MonadMonitor m => Http.Method -> Http.Status -> HttpRequestCounter -> m ()+countHttpRequest method status httpRequestCounter = withLabel httpRequestCounter label incCounter+ where+ label = (textMethod, textStatusCode)+ textMethod = decodeUtf8With lenientDecode method+ textStatusCode = pack $ show $ Http.statusCode status+++data IcepeakMetrics = IcepeakMetrics+ { icepeakMetricsRequestCounter :: HttpRequestCounter+ , icepeakMetricsDataSize :: Gauge+ , icepeakMetricsJournalSize :: Gauge+ , icepeakMetricsDataWritten :: Counter+ , icepeakMetricsJournalWritten :: Counter+ , icepeakMetricsSubscriberCount :: Gauge+ }++createAndRegisterIcepeakMetrics :: IO IcepeakMetrics+createAndRegisterIcepeakMetrics = IcepeakMetrics+ <$> register (vector ("method", "status") requestCounter)+ <*> register (gauge (Info "icepeak_data_size" "Size of data file in bytes."))+ <*> register (gauge (Info "icepeak_journal_size_bytes"+ "Size of journal file in bytes."))+ <*> register (counter (Info "icepeak_data_written" "Total number of bytes written so far."))+ <*> register (counter (Info "icepeak_journal_written_bytes_total"+ "Total number of bytes written to the journal so far."))+ <*> register (gauge+ (Info "icepeak_subscriber_count" "Number of websocket subscriber connections."))+ where+ requestCounter = counter (Info "icepeak_http_requests"+ "Total number of HTTP requests since starting Icepeak.")++notifyRequest :: Http.Method -> Http.Status -> IcepeakMetrics -> IO ()+notifyRequest method status = countHttpRequest method status . icepeakMetricsRequestCounter++setDataSize :: (MonadMonitor m, Real a) => a -> IcepeakMetrics -> m ()+setDataSize val metrics = setGauge (icepeakMetricsDataSize metrics) (realToFrac val)++setJournalSize :: (MonadMonitor m, Real a) => a -> IcepeakMetrics -> m ()+setJournalSize val metrics = setGauge (icepeakMetricsJournalSize metrics) (realToFrac val)++-- | Increment the total data written to disk by the given number of bytes.+-- Returns True, when it actually increased the counter and otherwise False.+incrementDataWritten :: (MonadMonitor m, Real a) => a -> IcepeakMetrics -> m Bool+incrementDataWritten num_bytes metrics = addCounter (icepeakMetricsDataWritten metrics)+ (realToFrac num_bytes)++-- | Increment the data written to the journal by the given number of bytes.+-- Returns True, when it actually increased the counter and otherwise False.+incrementJournalWritten :: (MonadMonitor m, Real a) => a -> IcepeakMetrics -> m Bool+incrementJournalWritten num_bytes metrics = addCounter (icepeakMetricsJournalWritten metrics)+ (realToFrac num_bytes)++incrementSubscribers :: MonadMonitor m => IcepeakMetrics -> m ()+incrementSubscribers = incGauge . icepeakMetricsSubscriberCount++decrementSubscribers :: MonadMonitor m => IcepeakMetrics -> m ()+decrementSubscribers = decGauge . icepeakMetricsSubscriberCount
+ src/MetricsServer.hs view
@@ -0,0 +1,24 @@+{-# LANGUAGE OverloadedStrings #-}+module MetricsServer where++import Data.Function ((&))+import Data.Monoid ((<>))+import qualified Data.Text as Text+import qualified Network.Wai.Handler.Warp as Warp+import qualified Network.Wai.Middleware.Prometheus as PrometheusWai++import Config (MetricsConfig (..))+import Logger (Logger, LogLevel(..), postLog)++metricsServerConfig :: MetricsConfig -> Warp.Settings+metricsServerConfig config = Warp.defaultSettings+ & Warp.setHost (metricsConfigHost config)+ & Warp.setPort (metricsConfigPort config)++runMetricsServer :: Logger -> MetricsConfig -> IO ()+runMetricsServer logger metricsConfig = do+ Logger.postLog logger LogInfo $ "Metrics provided on "+ <> (Text.pack $ show $ metricsConfigHost metricsConfig)+ <> ":"+ <> (Text.pack $ show $ metricsConfigPort metricsConfig)+ Warp.runSettings (metricsServerConfig metricsConfig) PrometheusWai.metricsApp
+ src/Persistence.hs view
@@ -0,0 +1,341 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-| This module abstracts over the details of persisting the value. Journaling is+ also handled here, if enabled. -}+module Persistence+ ( PersistentValue+ , PersistenceConfig (..)+ , getDataFile+ , getValue+ , apply+ , loadFromBackend+ , setupStorageBackend+ , syncToBackend+ ) where++import Control.Concurrent.STM+import Control.Exception+import Control.Monad.Except+import qualified Data.Aeson as Aeson+import qualified Data.Aeson.Types as Aeson+import qualified Data.ByteString as SBS+import qualified Data.ByteString.Char8 as SBS8+import qualified Data.ByteString.Lazy as LBS+import qualified Data.ByteString.Lazy.Char8 as LBS8+import Data.Foldable+import Data.Text (Text)+import qualified Data.Text as Text+import Data.Traversable+import Database.SQLite.Simple (FromRow (..), NamedParam (..), Only (..), execute,+ execute_, executeNamed, field, open, query_)+import System.Directory (getFileSize, renameFile)+import System.Exit (die)+import System.IO+import System.IO.Error (tryIOError, isDoesNotExistError, isPermissionError)+import Logger (Logger, LogLevel(..))+import qualified Logger+import qualified Metrics+import qualified Store++import Config (StorageBackend (..))++data PersistentValue = PersistentValue+ { pvConfig :: PersistenceConfig+ , pvValue :: TVar Store.Value+ -- ^ contains the state of the whole database+ , pvIsDirty :: TVar Bool+ -- ^ flag indicating whether the current value of 'pvValue' has not yet been persisted to disk+ , pvJournal :: Maybe Handle+ }++data PersistenceConfig = PersistenceConfig+ { pcDataFile :: FilePath+ , pcJournalFile :: Maybe FilePath+ , pcLogger :: Logger+ , pcMetrics :: Maybe Metrics.IcepeakMetrics+ }++-- | Get the actual value+getValue :: PersistentValue -> STM Store.Value+getValue = readTVar . pvValue++-- | Apply a modification, and write it to the journal if enabled.+apply :: Store.Modification -> PersistentValue -> IO ()+apply op val = do+ -- append to journal if enabled+ for_ (pvJournal val) $ \journalHandle -> do+ let entry = Aeson.encode op+ LBS8.hPutStrLn journalHandle entry+ for_ (pcMetrics . pvConfig $ val) $ \metrics -> do+ journalPos <- hTell journalHandle+ _ <- Metrics.incrementJournalWritten (LBS8.length entry) metrics+ Metrics.setJournalSize journalPos metrics+ -- update value+ atomically $ do+ modifyTVar (pvValue val) (Store.applyModification op)+ writeTVar (pvIsDirty val) True+++-- If no --data-file was supplied we default to either icepeak.json, for the file backend,+-- or icepeak.db, for the Sqlite backend+getDataFile :: StorageBackend -> Maybe FilePath -> FilePath+getDataFile _ (Just filePath) = filePath+getDataFile File _ = "icepeak.json"+getDataFile Sqlite _ = "icepeak.db"++-- * IO++-- | Ensure that we can access the chosen storage file, and validate that it has the right file format+setupStorageBackend :: StorageBackend -> FilePath -> IO ()+setupStorageBackend File filePath = do+ eitherEncodedValue <- tryIOError $ SBS.readFile filePath+ case eitherEncodedValue of+ Left e | isDoesNotExistError e -> do+ -- If there is no icepeak.json file yet, we create an empty one instead.+ let message = "WARNING: Could not read data from " ++ filePath +++ " because the file does not exist yet. Created an empty database instead."++ -- if this fails, we want the whole program to crash since something is wrong+ SBS.writeFile filePath "{}"+ putStrLn message++ Left e | isPermissionError e ->+ die $ "File " ++ filePath ++ " cannot be read due to a permission error. Please check the file permissions."+ -- other errors should also lead to program termination+ Left e -> die (show e)++ -- in case the data-file is empty we write the empty object "{}" to it and return it+ Right "" -> do+ let message = "WARNING: The provided --data-file " ++ filePath +++ " is empty. Will write a default database of {} to this file."+ putStrLn message+ SBS.writeFile filePath "{}"+ Right encodedValue -> case Aeson.eitherDecodeStrict encodedValue of+ Left msg -> die $ "Failed to decode the initial data in " ++ filePath ++ ": " ++ show msg+ Right (_value :: Aeson.Value) -> pure ()+setupStorageBackend Sqlite filePath = do+ -- read the data from SQLite+ conn <- liftIO $ open filePath+ liftIO $ execute_ conn "CREATE TABLE IF NOT EXISTS icepeak (value BLOB)"++ jsonRows <- liftIO $ (query_ conn "SELECT * from icepeak" :: IO [JsonRow])+ case jsonRows of+ -- ensure that there is one row in the table, so that we can UPDATE it later+ [] -> liftIO $ execute conn "INSERT INTO icepeak (value) VALUES (?)" (Only $ Aeson.encode Aeson.emptyObject)+ _ -> pure()++loadFromBackend :: StorageBackend -> PersistenceConfig -> IO (Either String PersistentValue)+loadFromBackend backend config = runExceptT $ do+ let metrics = pcMetrics config+ dataFilePath = pcDataFile config++ -- We immediately set the dataSize metric, so that Prometheus can start scraping it+ liftIO $ forM_ metrics $ \metric -> do+ size <- getFileSize dataFilePath+ Metrics.setDataSize size metric++ -- Read the data from disk and parse it as an Aeson.Value+ value <- case backend of+ File -> readData dataFilePath+ Sqlite -> readSqliteData dataFilePath++ valueVar <- lift $ newTVarIO value+ dirtyVar <- lift $ newTVarIO False+ journal <- for (pcJournalFile config) openJournal++ let val = PersistentValue+ { pvConfig = config+ , pvValue = valueVar+ , pvIsDirty = dirtyVar+ , pvJournal = journal+ }+ recoverJournal val+ return val++syncToBackend :: StorageBackend -> PersistentValue -> IO ()+syncToBackend File pv = syncFile pv+syncToBackend Sqlite pv = syncSqliteFile pv++-- * SQLite loading and syncing++-- | There is currently just one row and it contains only one column of type SBS.ByteString.+-- This single field holds the whole JSON value for now.+data JsonRow = JsonRow {jsonByteString :: SBS.ByteString} deriving (Show)++instance FromRow JsonRow where+ fromRow = JsonRow <$> field++-- | Read and decode the Sqlite data file from disk+readSqliteData :: FilePath -> ExceptT String IO Store.Value+readSqliteData filePath = ExceptT $ do+ -- read the data from SQLite+ conn <- liftIO $ open filePath+ jsonRows <- liftIO $ (query_ conn "SELECT * from icepeak" :: IO [JsonRow])++ case jsonRows of+ -- the 'setupStorageBackend' function verifies that we can read the database and that at least one row exists+ [] -> pure $ Right Aeson.emptyObject+ _ -> case Aeson.eitherDecodeStrict (jsonByteString $ head $ jsonRows) of+ Left msg -> pure $ Left $ "Failed to decode the initial data: " ++ show msg+ Right value -> pure $ Right (value :: Store.Value)++-- | Write the data to the SQLite file if it has changed.+syncSqliteFile :: PersistentValue -> IO ()+syncSqliteFile val = do+ (dirty, value) <- atomically $ (,) <$> readTVar (pvIsDirty val)+ <*> readTVar (pvValue val)+ <* writeTVar (pvIsDirty val) False+ -- simple optimization: only write when something changed+ when dirty $ do+ let filePath = pcDataFile $ pvConfig val++ conn <- open filePath+ -- we can always UPDATE here, since we know that there will be at least one row, since+ -- we issue an INSERT when we load in an empty database+ liftIO $ executeNamed conn "UPDATE icepeak SET value = :value" [":value" := Aeson.encode value]++ -- the journal is idempotent, so there is no harm if icepeak crashes between+ -- the previous and the next action+ truncateJournal val++ -- handle metrics last+ updateMetrics val++-- * File syncing++-- | Write the data to disk if it has changed.+syncFile :: PersistentValue -> IO ()+syncFile val = do+ (dirty, value) <- atomically $ (,) <$> readTVar (pvIsDirty val)+ <*> readTVar (pvValue val)+ <* writeTVar (pvIsDirty val) False+ -- simple optimization: only write when something changed+ when dirty $ do+ let fileName = pcDataFile $ pvConfig val+ tempFileName = fileName ++ ".new"+ -- we first write to a temporary file here and then do a rename on it+ -- because rename is atomic on Posix and a crash during writing the+ -- temporary file will thus not corrupt the datastore+ LBS.writeFile tempFileName (Aeson.encode value)+ renameFile tempFileName fileName++ -- the journal is idempotent, so there is no harm if icepeak crashes between+ -- the previous and the next action+ truncateJournal val++ -- handle metrics last+ updateMetrics val++-- * Private helper functions++-- Note that some of these functions are still exported in order to be usable in the test suite++-- | Seek to the beginning of the journal file and set the file size to zero.+-- This should be called after all journal entries have been replayed, and the data has been+-- synced to disk.+truncateJournal :: PersistentValue -> IO ()+truncateJournal val =+ for_ (pvJournal val) $ \journalHandle -> do+ -- we must seek back to the beginning of the file *before* calling hSetFileSize, since that+ -- function does not change the file cursor, which means that the first write that follows+ -- would fill up the file with \NUL bytes up to the original cursor position.+ hSeek journalHandle AbsoluteSeek 0+ hSetFileSize journalHandle 0++-- | We keep track of three metrics related to persistence:+-- 1. The current size of the data file+-- 2. The current size of the journal file+-- 3. The total amount of data written to disk since the process was started+-- (not counting journal writes)+updateMetrics :: PersistentValue -> IO ()+updateMetrics val = do+ let filePath = pcDataFile . pvConfig $ val+ metrics = pcMetrics . pvConfig $ val+ forM_ metrics $ \metric -> do+ size <- getFileSize filePath+ Metrics.setDataSize size metric+ Metrics.setJournalSize (0 :: Int) metric+ Metrics.incrementDataWritten size metric++-- | Open or create the journal file+openJournal :: FilePath -> ExceptT String IO Handle+openJournal journalFile = ExceptT $ do+ eitherHandle <- try $ do+ h <- openBinaryFile journalFile ReadWriteMode+ hSetBuffering h LineBuffering+ pure h+ case eitherHandle :: Either SomeException Handle of+ Left exc -> pure $ Left $ "Failed to open journal file: " ++ show exc+ Right fileHandle -> pure $ Right fileHandle++-- | Read the modifications from the journal file, apply them and sync again.+-- This should be done when loading the database from disk.+recoverJournal :: PersistentValue -> ExceptT String IO ()+recoverJournal pval = for_ (pvJournal pval) $ \journalHandle -> ExceptT $ fmap formatErr $ try $ do+ initialValue <- atomically $ readTVar (pvValue pval)+ (finalValue, successful, total) <- runRecovery journalHandle initialValue++ when (successful > 0) $ do+ atomically $ do+ writeTVar (pvValue pval) finalValue+ writeTVar (pvIsDirty pval) True+ -- syncing takes care of cleaning the journal+ syncFile pval++ when (total > 0) $ do+ logMessage pval "Journal replayed"+ logMessage pval $ " failed: " <> Text.pack (show $ total - successful)+ logMessage pval $ " successful: " <> Text.pack (show $ successful)++ where+ formatErr :: Either SomeException a -> Either String a+ formatErr (Left exc) = Left $ "Failed to read journal: " ++ show exc+ formatErr (Right x) = Right x++ runRecovery journalHandle value = do+ -- read modifications from the beginning+ hSeek journalHandle AbsoluteSeek 0+ foldJournalM journalHandle replayLine (value, 0 :: Integer, 0 :: Integer)++ replayLine line (!value, !successful, !total) = do+ when (total == 0) $ do+ logMessage pval "Journal not empty, recovering"+ case Aeson.eitherDecodeStrict line of+ Left err -> do+ let lineNumber = total + 1+ logMessage pval $ failedRecoveryMsg err lineNumber+ pure (value, successful, total + 1)+ Right op -> pure (Store.applyModification op value, successful + 1, total + 1)++ failedRecoveryMsg err line = "Failed to recover journal entry "+ <> Text.pack (show line) <> ": " <> Text.pack err++-- | Read and decode the data file from disk+readData :: FilePath -> ExceptT String IO Store.Value+readData filePath = ExceptT $ do+ eitherEncodedValue <- tryIOError $ SBS.readFile filePath+ case eitherEncodedValue of+ -- we do not expect any errors here, since we validated the file earlier already+ Left e -> pure $ Left (show e)+ Right encodedValue -> case Aeson.eitherDecodeStrict encodedValue of+ Left msg -> pure $ Left $ "Failed to decode the initial data: " ++ show msg+ Right value -> pure $ Right value++-- | Log a message in the context of a PersistentValue.+logMessage :: PersistentValue -> Text -> IO ()+logMessage pval msg = Logger.postLogBlocking (pcLogger $ pvConfig pval) LogInfo msg++-- | Left fold over all journal entries.+foldJournalM :: Handle -> (SBS8.ByteString -> a -> IO a) -> a -> IO a+foldJournalM h f = go+ where+ go !x = do+ eof <- hIsEOF h+ if eof+ then pure x+ else do+ line <- SBS8.hGetLine h+ x' <- f line x+ go x'
+ src/SentryLogging.hs view
@@ -0,0 +1,20 @@+-- | Module for logging (crash) reports to Sentry+module SentryLogging(+ getCrashLogger, logCrashMessage+) where++import qualified System.Log.Raven as Sentry+import qualified System.Log.Raven.Transport.HttpConduit as Sentry+import qualified System.Log.Raven.Types as Sentry++-- | Returns a Maybe SentryService which can be used to send error information+-- to Sentry. Return value is Nothing is the environment variable SENTRY_DSN is+-- not set.+getCrashLogger :: String -> IO Sentry.SentryService+getCrashLogger dsn = Sentry.initRaven dsn id Sentry.sendRecord Sentry.stderrFallback++-- | Send a crash message to Sentry, used in cases when no exception is available,+-- Which is the case for Scotty errors. Function does nothing when Sentry service+-- is Nothing.+logCrashMessage :: String -> Sentry.SentryService -> String -> IO ()+logCrashMessage name service message = Sentry.register service name Sentry.Fatal message id
+ src/Server.hs view
@@ -0,0 +1,26 @@+{-# LANGUAGE OverloadedStrings #-}+module Server+(+ runServer,+)+where++import Data.Semigroup ((<>))+import Data.Text (pack)++import Network.Wai (Application)+import Network.Wai.Handler.WebSockets (websocketsOr)+import Network.WebSockets (ServerApp)++import qualified Network.Wai.Handler.Warp as Warp+import qualified Network.WebSockets as WebSockets++import Logger (Logger, LogLevel(..), postLog)++runServer :: Logger -> ServerApp -> Application -> Int -> IO ()+runServer logger wsApp httpApp port =+ let+ wsConnectionOpts = WebSockets.defaultConnectionOptions+ in do+ postLog logger LogInfo $ pack $ "Listening on port " <> show port <> "."+ Warp.run port $ websocketsOr wsConnectionOpts wsApp httpApp
+ src/Store.hs view
@@ -0,0 +1,95 @@+{-# LANGUAGE OverloadedStrings #-}+module Store+(+ Modification (..),+ Path,+ Value,+ modificationPath,+ applyModification,+ delete,+ insert,+ lookup,+ lookupOrNull,+)+where++import Data.Aeson (Value (..), (.=), (.:))+import Data.Maybe (fromMaybe)+import Data.Text (Text)+import Prelude hiding (lookup)++import qualified Data.Aeson as Aeson+import qualified Data.Aeson.Types as Aeson+import qualified Data.HashMap.Strict as HashMap++type Path = [Text]++-- A modification operation.+data Modification+ = Put Path Value+ | Delete Path+ deriving (Eq, Show)++instance Aeson.ToJSON Modification where+ toJSON (Put path value) = Aeson.object+ [ "op" .= ("put" :: Text)+ , "path" .= path+ , "value" .= value+ ]+ toJSON (Delete path) = Aeson.object+ [ "op" .= ("delete" :: Text)+ , "path" .= path+ ]++instance Aeson.FromJSON Modification where+ parseJSON = Aeson.withObject "Modification" $ \v -> do+ op <- v .: "op"+ case op of+ "put" -> Put <$> v .: "path" <*> v .: "value"+ "delete" -> Delete <$> v .: "path"+ other -> Aeson.typeMismatch "Op" other++-- | Return the path that is touched by a modification.+modificationPath :: Modification -> Path+modificationPath op = case op of+ Put path _ -> path+ Delete path -> path++lookup :: Path -> Value -> Maybe Value+lookup path value =+ case path of+ [] -> Just value+ key : pathTail -> case value of+ Object dict -> HashMap.lookup key dict >>= lookup pathTail+ _notObject -> Nothing++-- Look up a value, returning null if the path does not exist.+lookupOrNull :: Path -> Value -> Value+lookupOrNull path = fromMaybe Null . lookup path++-- | Execute a modification.+applyModification :: Modification -> Value -> Value+applyModification (Delete path) value = Store.delete path value+applyModification (Put path newValue) value = Store.insert path newValue value++-- Overwrite a value at the given path, and create the path leading up to it if+-- it did not exist.+insert :: Path -> Value -> Value -> Value+insert path newValue value =+ case path of+ [] -> newValue+ key : pathTail -> Object $ case value of+ Object dict -> HashMap.alter (Just . (insert pathTail newValue) . fromMaybe Null) key dict+ _notObject -> HashMap.singleton key $ insert pathTail newValue Null++-- Delete key at the given path. If the path is empty, return null.+delete :: Path -> Value -> Value+delete path value =+ case path of+ [] -> Null+ key : [] -> case value of+ Object dict -> Object $ HashMap.delete key dict+ notObject -> notObject+ key : pathTail -> case value of+ Object dict -> Object $ HashMap.adjust (delete pathTail) key dict+ notObject -> notObject
+ src/Subscription.hs view
@@ -0,0 +1,121 @@+{-# LANGUAGE DeriveFunctor #-}++module Subscription+(+ SubscriptionTree (..),+ broadcast,+ broadcast',+ empty,+ subscribe,+ unsubscribe,+ showTree,+)+where++import Control.Monad (void)+import Control.Monad.Writer (Writer, tell, execWriter)+import Data.Aeson (Value)+import Data.Foldable (for_, traverse_)+import Data.HashMap.Strict (HashMap)+import Data.Hashable (Hashable)+import Data.Maybe (fromMaybe)+import Data.Monoid ((<>))+import Data.Text (Text)++import qualified Control.Concurrent.Async as Async+import qualified Data.HashMap.Strict as HashMap+import qualified Data.Text as Text++import qualified Store++-- Keeps subscriptions in a tree data structure, so we can efficiently determine+-- which clients need to be notified for a given update.+data SubscriptionTree id conn =+ SubscriptionTree (HashMap id conn) (HashMap Text (SubscriptionTree id conn))+ deriving (Eq, Functor, Show)++empty :: SubscriptionTree id conn+empty = SubscriptionTree HashMap.empty HashMap.empty++isEmpty :: SubscriptionTree id conn -> Bool+isEmpty (SubscriptionTree here inner) = HashMap.null here && HashMap.null inner++subscribe+ :: (Eq id, Hashable id)+ => [Text]+ -> id+ -> conn+ -> SubscriptionTree id conn+ -> SubscriptionTree id conn+subscribe path subid subval (SubscriptionTree here inner) =+ case path of+ [] -> SubscriptionTree (HashMap.insert subid subval here) inner+ key : pathTail ->+ let+ subscribeInner = subscribe pathTail subid subval+ newInner = HashMap.alter (Just . subscribeInner . fromMaybe empty) key inner+ in+ SubscriptionTree here newInner++unsubscribe+ :: (Eq id, Hashable id)+ => [Text]+ -> id+ -> SubscriptionTree id conn+ -> SubscriptionTree id conn+unsubscribe path subid (SubscriptionTree here inner) =+ case path of+ [] -> SubscriptionTree (HashMap.delete subid here) inner+ key : pathTail ->+ let+ -- Remove the tail from the inner tree (if it exists). If that left the+ -- inner tree empty, remove the key altogether to keep the tree clean.+ justNotEmpty tree = if isEmpty tree then Nothing else Just tree+ unsubscribeInner = justNotEmpty . unsubscribe pathTail subid+ newInner = HashMap.update unsubscribeInner key inner+ in+ SubscriptionTree here newInner++-- Invoke f for all subscribers to the path. The subscribers get passed the+-- subvalue at the path that they are subscribed to.+broadcast :: (conn -> Value -> IO ()) -> [Text] -> Value -> SubscriptionTree id conn -> IO ()+broadcast f path value tree =+ -- We broadcast concurrently since all updates are independent of each other+ Async.mapConcurrently_ (uncurry f) notifications+ where notifications = broadcast' path value tree++-- Like broadcast, but return a list of notifications rather than invoking an+-- effect on each of them.+broadcast' :: [Text] -> Value -> SubscriptionTree id conn -> [(conn, Value)]+broadcast' = \path value tree -> execWriter $ loop path value tree+ where+ loop :: [Text] -> Value -> SubscriptionTree id conn -> Writer [(conn, Value)] ()+ loop path value (SubscriptionTree here inner) = do+ case path of+ [] -> do+ -- When the path is empty, all subscribers that are "here" or at a deeper+ -- level should receive a notification.+ traverse_ (\v -> tell [(v, value)]) here+ let broadcastInner key = loop [] (Store.lookupOrNull [key] value)+ void $ HashMap.traverseWithKey broadcastInner inner++ key : pathTail -> do+ traverse_ (\v -> tell [(v, value)]) here+ for_ (HashMap.lookup key inner) $ \subs ->+ loop pathTail (Store.lookupOrNull [key] value) subs++-- Show subscriptions, for debugging purposes.+showTree :: Show id => SubscriptionTree id conn -> String+showTree tree =+ let+ withPrefix prefix (SubscriptionTree here inner) =+ let+ strHere :: String+ strHere = concatMap (\cid -> " * " <> (show cid) <> "\n") (HashMap.keys here)+ showInner iPrefix t = iPrefix <> "\n" <> withPrefix iPrefix t+ strInner :: String+ strInner = concat $ HashMap.mapWithKey (\key -> showInner (prefix <> "/" <> Text.unpack key)) inner+ in+ strHere <> strInner+ in+ "/\n" <> (withPrefix "" tree)
+ src/WebsocketServer.hs view
@@ -0,0 +1,144 @@+{-# LANGUAGE OverloadedStrings #-}++module WebsocketServer (+ ServerState,+ acceptConnection,+ processUpdates+) where++import Control.Concurrent (modifyMVar_, readMVar)+import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TBQueue (readTBQueue)+import Control.Exception (SomeAsyncException, SomeException, finally, fromException, catch, throwIO)+import Control.Monad (forever)+import Data.Aeson (Value)+import Data.Text (Text)+import Data.UUID+import System.Random (randomIO)++import qualified Data.Aeson as Aeson+import qualified Data.ByteString.Lazy as LBS+import qualified Data.Time.Clock.POSIX as Clock+import qualified Network.WebSockets as WS+import qualified Network.HTTP.Types.Header as HttpHeader+import qualified Network.HTTP.Types.URI as Uri++import Config (Config (..))+import Core (Core (..), ServerState, Updated (..), getCurrentValue, withCoreMetrics)+import Store (Path)+import AccessControl (AccessMode(..))+import JwtMiddleware (AuthResult (..), isRequestAuthorized, errorResponseBody)++import qualified Metrics+import qualified Subscription++newUUID :: IO UUID+newUUID = randomIO++-- send the updated data to all subscribers to the path+broadcast :: [Text] -> Value -> ServerState -> IO ()+broadcast =+ let+ send :: WS.Connection -> Value -> IO ()+ send conn value =+ WS.sendTextData conn (Aeson.encode value)+ `catch`+ sendFailed++ sendFailed :: SomeException -> IO ()+ sendFailed exc+ -- Rethrow async exceptions, they are meant for inter-thread communication+ -- (e.g. ThreadKilled) and we don't expect them at this point.+ | Just asyncExc <- fromException exc = throwIO (asyncExc :: SomeAsyncException)+ -- We want to catch all other errors in order to prevent them from+ -- bubbling up and disrupting the broadcasts to other clients.+ | otherwise = pure ()+ in+ Subscription.broadcast send++-- Called for each new client that connects.+acceptConnection :: Core -> WS.PendingConnection -> IO ()+acceptConnection core pending = do+ -- printRequest pending+ -- TODO: Validate the path and headers of the pending request+ authResult <- authorizePendingConnection core pending+ case authResult of+ AuthRejected err ->+ WS.rejectRequestWith pending $ WS.RejectRequest+ { WS.rejectCode = 401+ , WS.rejectMessage = "Unauthorized"+ , WS.rejectHeaders = [(HttpHeader.hContentType, "application/json")]+ , WS.rejectBody = LBS.toStrict $ errorResponseBody err+ }+ AuthAccepted -> do+ let path = fst $ Uri.decodePath $ WS.requestPath $ WS.pendingRequest pending+ connection <- WS.acceptRequest pending+ -- Fork a pinging thread, for each client, to keep idle connections open and to detect+ -- closed connections. Sends a ping message every 30 seconds.+ -- Note: The thread dies silently if the connection crashes or is closed.+ WS.withPingThread connection 30 (pure ()) $ handleClient connection path core++-- * Authorization++authorizePendingConnection :: Core -> WS.PendingConnection -> IO AuthResult+authorizePendingConnection core conn+ | configEnableJwtAuth (coreConfig core) = do+ now <- Clock.getPOSIXTime+ let req = WS.pendingRequest conn+ (path, query) = Uri.decodePath $ WS.requestPath req+ headers = WS.requestHeaders req+ return $ isRequestAuthorized headers query now (configJwtSecret (coreConfig core)) path ModeRead+ | otherwise = pure AuthAccepted++-- * Client handling++handleClient :: WS.Connection -> Path -> Core -> IO ()+handleClient conn path core = do+ uuid <- newUUID+ let+ state = coreClients core+ onConnect = do+ modifyMVar_ state (pure . Subscription.subscribe path uuid conn)+ withCoreMetrics core Metrics.incrementSubscribers+ onDisconnect = do+ modifyMVar_ state (pure . Subscription.unsubscribe path uuid)+ withCoreMetrics core Metrics.decrementSubscribers+ sendInitialValue = do+ currentValue <- getCurrentValue core path+ WS.sendTextData conn (Aeson.encode currentValue)++ -- simply ignore connection errors, otherwise, warp handles the exception+ -- and sends a 500 response in the middle of a websocket connection, and+ -- that violates the websocket protocol.+ -- Note that subscribers are still properly removed by the finally below+ handleConnectionError :: WS.ConnectionException -> IO ()+ handleConnectionError _ = pure ()+ -- Put the client in the subscription tree and keep the connection open.+ -- Remove it when the connection is closed.+ finally (onConnect >> sendInitialValue >> keepTalking conn) onDisconnect+ `catch` handleConnectionError++-- We don't send any messages here; sending is done by the update+-- loop; it finds the client in the set of subscriptions. But we do+-- need to keep the thread running, otherwise the connection will be+-- closed. So we go into an infinite loop here.+keepTalking :: WS.Connection -> IO ()+keepTalking conn = forever $ do+ -- Note: WS.receiveDataMessage will handle control messages automatically and e.g.+ -- do the closing handshake of the websocket protocol correctly+ WS.receiveDataMessage conn++-- loop that is called for every update and that broadcasts the values to all+-- subscribers of the updated path+processUpdates :: Core -> IO ()+processUpdates core = go+ where+ go = do+ maybeUpdate <- atomically $ readTBQueue (coreUpdates core)+ case maybeUpdate of+ Just (Updated path value) -> do+ clients <- readMVar (coreClients core)+ broadcast path value clients+ go+ -- Stop the loop when we receive a Nothing.+ Nothing -> pure ()
+ tests/AccessControlSpec.hs view
@@ -0,0 +1,43 @@+{-# LANGUAGE OverloadedStrings #-}+module AccessControlSpec (spec) where++import Test.Hspec (Spec, describe, it, shouldBe)++import qualified Data.Aeson as Aeson+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck.Instances ()++import AccessControl+import OrphanInstances ()++spec :: Spec+spec = do+ describe "AccessControl" $ do+ let testClaim = IcepeakClaim+ [ AuthPath ["foo"] [ModeRead]+ , AuthPath ["foo", "bar"] [ModeWrite]+ , AuthPath ["a", "b"] [ModeRead, ModeWrite]+ ]++ prop "empty whitelist disallows everything" $ \path mode ->+ let emptyClaim = IcepeakClaim []+ in not $ isAuthorizedByClaim emptyClaim path mode++ prop "empty path in whitelist allows everything" $ \path mode ->+ let allClaim = IcepeakClaim [AuthPath [] [ModeRead, ModeWrite]]+ in isAuthorizedByClaim allClaim path mode++ prop "allowEverything claim allows everything" $ isAuthorizedByClaim allowEverything++ it "should allow access to sub-paths" $ do+ isAuthorizedByClaim testClaim ["foo", "1"] ModeRead `shouldBe` True+ isAuthorizedByClaim testClaim ["foo", "bar"] ModeRead `shouldBe` True+ isAuthorizedByClaim testClaim ["foo", "bar", "1"] ModeWrite `shouldBe` True++ it "should disallow access to non-whitelisted paths" $ do+ isAuthorizedByClaim testClaim ["foo", "1"] ModeWrite `shouldBe` False+ isAuthorizedByClaim testClaim ["a"] ModeRead `shouldBe` False+ isAuthorizedByClaim testClaim ["a", "c"] ModeWrite `shouldBe` False++ prop "serializing and deserializing claim is identity" $ \claim ->+ Aeson.fromJSON (Aeson.toJSON claim) == Aeson.Success (claim :: IcepeakClaim)
+ tests/ApiSpec.hs view
@@ -0,0 +1,6 @@+module ApiSpec (spec) where++import Test.Hspec++spec :: Spec+spec = return ()
+ tests/CoreSpec.hs view
@@ -0,0 +1,8 @@+{-# LANGUAGE OverloadedStrings #-}++module CoreSpec (spec) where++import Test.Hspec++spec :: Spec+spec = return ()
+ tests/JwtSpec.hs view
@@ -0,0 +1,76 @@+{-# LANGUAGE OverloadedStrings #-}+module JwtSpec (spec) where++--import qualified Data.Aeson as Aeson+import Data.Time.Clock (NominalDiffTime)+import Test.Hspec -- (Spec, describe, it, shouldBe, expectationFailure)+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck.Instances ()+import qualified Web.JWT as JWT+import qualified Data.Map.Strict as Map+import qualified Data.Text.Encoding as Text++import JwtAuth+import AccessControl+import OrphanInstances ()++spec :: Spec+spec = do+ describe "JWT" $ do++ let testAccess = IcepeakClaim+ [ AuthPath ["foo"] [ModeRead]+ , AuthPath ["bar", "baz"] [ModeRead, ModeWrite]+ ]++ let emptyClaim = JWT.JWTClaimsSet+ { JWT.iss = Nothing+ , JWT.sub = Nothing+ , JWT.aud = Nothing+ , JWT.exp = Nothing+ , JWT.nbf = Nothing+ , JWT.iat = Nothing+ , JWT.jti = Nothing+ , JWT.unregisteredClaims = JWT.ClaimsMap Map.empty+ }++ let testClaims = addIcepeakClaim testAccess emptyClaim++ let testSecret = JWT.HMACSecret "2o8357cEuc2o835cmsoei"++ let now = 12512 :: NominalDiffTime++ let joseHeader = JWT.JOSEHeader+ { JWT.typ = Just "JWT"+ , JWT.cty = Nothing+ , JWT.alg = Just JWT.HS256+ , JWT.kid = Nothing+ }++ it "should accept a valid token" $ do+ let token = Text.encodeUtf8 $ JWT.encodeSigned testSecret joseHeader testClaims+ extractClaim now testSecret token `shouldBe` Right testAccess++ it "should reject an expired token" $ do+ let Just expDate = JWT.numericDate $ now - 10+ claims = testClaims { JWT.exp = Just expDate }+ expiredToken = Text.encodeUtf8 $ JWT.encodeSigned testSecret joseHeader claims+ extractClaim now testSecret expiredToken `shouldBe` Left (VerificationError TokenExpired)++ it "should reject a token before its 'not before' date" $ do+ let Just nbfDate = JWT.numericDate $ now + 10+ claims = testClaims { JWT.nbf = Just nbfDate }+ nbfToken = Text.encodeUtf8 $ JWT.encodeSigned testSecret joseHeader claims+ extractClaim now testSecret nbfToken `shouldBe` Left (VerificationError TokenUsedTooEarly)++ it "should reject a token with wrong secret" $ do+ let claims = testClaims+ nbfToken = Text.encodeUtf8 $ JWT.encodeSigned testSecret joseHeader claims+ otherSecret = JWT.HMACSecret "dfhwcmo845cm8e5"+ extractClaim now otherSecret nbfToken `shouldBe` Left (VerificationError TokenSignatureInvalid)++ prop "should correctly encode and decode token" $ \icepeakClaim ->+ let claims = addIcepeakClaim icepeakClaim emptyClaim+ encoded = Text.encodeUtf8 $ JWT.encodeSigned testSecret joseHeader claims+ decoded = extractClaim now testSecret encoded+ in decoded == Right icepeakClaim
+ tests/OrphanInstances.hs view
@@ -0,0 +1,33 @@+module OrphanInstances where++import Data.Aeson (Value (..))+import Test.QuickCheck.Instances ()+import Test.QuickCheck.Arbitrary (Arbitrary (..))+import qualified Test.QuickCheck.Gen as Gen++import Store (Modification (..))+import AccessControl++instance Arbitrary AccessMode where+ arbitrary = Gen.elements [minBound..maxBound]++instance Arbitrary AuthPath where+ arbitrary = AuthPath <$> arbitrary <*> arbitrary++instance Arbitrary IcepeakClaim where+ arbitrary = IcepeakClaim <$> arbitrary++instance Arbitrary Modification where+ arbitrary = Gen.oneof+ [ Put <$> arbitrary <*> arbitrary+ , Delete <$> arbitrary+ ]++instance Arbitrary Value where+ arbitrary = Gen.oneof+ [ Object <$> Gen.scale (`div` 2) arbitrary+ , Array <$> Gen.scale (`div` 2) arbitrary+ , String <$> arbitrary+ , Bool <$> arbitrary+ , pure Null+ ]
+ tests/PersistenceSpec.hs view
@@ -0,0 +1,31 @@+{-# LANGUAGE OverloadedStrings #-}+module PersistenceSpec (spec) where++import Data.Foldable+import Test.Hspec+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck.Instances ()++import qualified Data.Aeson as Aeson+import qualified Data.ByteString.Lazy.Char8 as LBS8++import OrphanInstances ()+import Store (Modification (..))+import qualified Store++spec :: Spec+spec = do+ describe "Store.Modification" $ do+ prop "does not contain new lines when serialized" $ \op ->+ let jsonStr = Aeson.encode (op :: Modification)+ in '\n' `LBS8.notElem` jsonStr++ prop "round trips serialization" $ \op ->+ let jsonStr = Aeson.encode (op :: Modification)+ decoded = Aeson.decode jsonStr+ in Just op == decoded++ describe "Journaling" $ do+ prop "journal is idempotent" $ \ops initial ->+ let replay value = foldl' (flip Store.applyModification) value (ops :: [Modification])+ in replay initial == replay (replay initial)
+ tests/RequestSpec.hs view
@@ -0,0 +1,56 @@+{-# LANGUAGE OverloadedStrings #-}+module RequestSpec (spec) where++import Test.Hspec+import Test.Hspec.Wai++import HTTPMethodInvalid (canonicalizeHTTPMethods, limitHTTPMethods)++import qualified Network.HTTP.Types as HTTP+import qualified Network.Wai as Wai++import qualified Data.ByteString.Char8 as ByteString++spec :: Spec+spec = describe "Request method" $ with (pure app) $ do+ it "accepts method get" $ do+ (request HTTP.methodGet "/" [] "") `shouldRespondWith` 200+ it "accepts method post" $ do+ (request HTTP.methodPost "/" [] "") `shouldRespondWith` 200+ it "accepts method head" $ do+ (request HTTP.methodHead "/" [] "") `shouldRespondWith` 200+ it "accepts method put" $ do+ (request HTTP.methodPut "/" [] "") `shouldRespondWith` 200+ it "accepts method delete" $ do+ (request HTTP.methodDelete "/" [] "") `shouldRespondWith` 200+ it "accepts method trace" $ do+ (request HTTP.methodTrace "/" [] "") `shouldRespondWith` 200+ it "accepts method connect" $ do+ (request HTTP.methodConnect "/" [] "") `shouldRespondWith` 200+ it "accepts method options" $ do+ (request HTTP.methodOptions "/" [] "") `shouldRespondWith` 200+ it "accepts method patch" $ do+ (request HTTP.methodPatch "/" [] "") `shouldRespondWith` 200++ it "declines other methods yqus" $ do+ (request yqus "/" [] "") `shouldRespondWith` 400+ it "declines other methods badmethod" $ do+ (request badMethod "/" [] "") `shouldRespondWith` 400+ it "declines other methods invalid" $ do+ (request invalid "/" [] "") `shouldRespondWith` 400++app :: Wai.Application+app = (canonicalizeHTTPMethods . limitHTTPMethods) return200++return200 :: Wai.Application+return200 _ respond = respond $+ Wai.responseLBS HTTP.status200 [("Content-Type", "text/plain")] ""++yqus :: HTTP.Method+yqus = ByteString.pack "YQUS"++badMethod :: HTTP.Method+badMethod = ByteString.pack "BADMETHOD"++invalid :: HTTP.Method+invalid = ByteString.pack "INVALID"
+ tests/SocketSpec.hs view
@@ -0,0 +1,6 @@+module SocketSpec (spec) where++import Test.Hspec++spec :: Spec+spec = return ()
+ tests/Spec.hs view
@@ -0,0 +1,23 @@+import Test.Hspec++import qualified AccessControlSpec+import qualified ApiSpec+import qualified CoreSpec+import qualified JwtSpec+import qualified PersistenceSpec+import qualified RequestSpec+import qualified SocketSpec+import qualified StoreSpec+import qualified SubscriptionTreeSpec++main :: IO ()+main = hspec $ do+ AccessControlSpec.spec+ ApiSpec.spec+ CoreSpec.spec+ JwtSpec.spec+ PersistenceSpec.spec+ RequestSpec.spec+ SocketSpec.spec+ StoreSpec.spec+ SubscriptionTreeSpec.spec
+ tests/StoreSpec.hs view
@@ -0,0 +1,132 @@+{-# LANGUAGE OverloadedStrings #-}++module StoreSpec (spec) where++import Data.Aeson (Value (..))+import Test.Hspec (Spec, describe, it, shouldBe)+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck.Instances ()++import qualified Data.HashMap.Strict as HashMap++import OrphanInstances ()+import Store (Modification (..))+import qualified Store++spec :: Spec+spec = do+ describe "Store.insert" $ do++ it "creates an object when putting 'x' into Null" $+ let+ before = Null+ after = Object $ HashMap.singleton "x" (String "Robert")+ in+ Store.insert ["x"] (String "Robert") before `shouldBe` after++ it "overwrites a key when putting 'x' into {'x': ...}" $+ let+ before = Object $ HashMap.singleton "x" (String "Arian")+ after = Object $ HashMap.singleton "x" (String "Robert")+ in+ Store.insert ["x"] (String "Robert") before `shouldBe` after++ it "adds a key when putting 'x' into {'y': ...}" $+ let+ before = Object $ HashMap.singleton "y" (String "Arian")+ after = Object $ HashMap.fromList [("x", String "Robert"), ("y", String "Arian")]+ in+ Store.insert ["x"] (String "Robert") before `shouldBe` after++ it "creates a nested object when putting 'x/y' into Null" $+ let+ before = Null+ after = Object $ HashMap.singleton "x" $ Object $ HashMap.singleton "y" "Stefan"+ in+ Store.insert ["x", "y"] (String "Stefan") before `shouldBe` after++ it "updates a nested object when putting 'x/y' into {'x': {'y': ...}}" $+ let+ before = Object $ HashMap.singleton "x" $ Object $ HashMap.singleton "y" "Radek"+ after = Object $ HashMap.singleton "x" $ Object $ HashMap.singleton "y" "Stefan"+ in+ Store.insert ["x", "y"] (String "Stefan") before `shouldBe` after++ it "adds a nested key when putting 'x/y' into {'x': {'y': ...}, 'z': ...}" $+ let+ before = Object $ HashMap.fromList [("x", Object $ HashMap.singleton "y" "Nuno"), ("z", Null)]+ after = Object $ HashMap.fromList [("x", Object $ HashMap.singleton "y" "Stefan"), ("z", Null)]+ in+ Store.insert ["x", "y"] (String "Stefan") before `shouldBe` after++ describe "Store" $ do++ prop "returns None after (lookup . delete . insert) in Null" $ \ path value ->+ let+ lkupDelIns = Store.lookup path . Store.delete path . Store.insert path value+ in+ if path == []+ then lkupDelIns Null `shouldBe` (Just Null)+ else lkupDelIns Null `shouldBe` Nothing++ prop "returns None after (lookup . delete . insert) in anything" $ \ path value before ->+ let+ lkupDelIns = Store.lookup path . Store.delete path . Store.insert path value+ in+ if path == []+ then lkupDelIns before `shouldBe` (Just Null)+ else lkupDelIns before `shouldBe` Nothing++ describe "Store.applyModification" $ do++ it "creates an object when putting 'x' into Null" $+ let+ put = Put ["x"] (String "Robert")+ before = Null+ after = Object $ HashMap.singleton "x" (String "Robert")+ in+ Store.applyModification put before `shouldBe` after++ it "overwrites a key when putting 'x' into {'x': ...}" $+ let+ put = Put ["x"] (String "Robert")+ before = Object $ HashMap.singleton "x" (String "Arian")+ after = Object $ HashMap.singleton "x" (String "Robert")+ in+ Store.applyModification put before `shouldBe` after++ it "adds a key when putting 'x' into {'y': ...}" $+ let+ put = Put ["x"] (String "Robert")+ before = Object $ HashMap.singleton "y" (String "Arian")+ after = Object $ HashMap.fromList [("x", String "Robert"), ("y", String "Arian")]+ in+ Store.applyModification put before `shouldBe` after++ it "creates a nested object when putting 'x/y' into Null" $+ let+ put = Put ["x", "y"] (String "Stefan")+ before = Null+ after = Object $ HashMap.singleton "x" $ Object $ HashMap.singleton "y" "Stefan"+ in+ Store.applyModification put before `shouldBe` after++ it "updates a nested object when putting 'x/y' into {'x': {'y': ...}}" $+ let+ put = Put ["x", "y"] (String "Stefan")+ before = Object $ HashMap.singleton "x" $ Object $ HashMap.singleton "y" "Radek"+ after = Object $ HashMap.singleton "x" $ Object $ HashMap.singleton "y" "Stefan"+ in+ Store.applyModification put before `shouldBe` after++ it "adds a nested key when putting 'x/y' into {'x': {'y': ...}, 'z': ...}" $+ let+ put = Put ["x", "y"] (String "Stefan")+ before = Object $ HashMap.fromList [("x", Object $ HashMap.singleton "y" "Nuno"), ("z", Null)]+ after = Object $ HashMap.fromList [("x", Object $ HashMap.singleton "y" "Stefan"), ("z", Null)]+ in+ Store.applyModification put before `shouldBe` after++ prop "is idempotent" $ \op value ->+ let apply = Store.applyModification op+ in apply value == apply (apply value)
+ tests/SubscriptionTreeSpec.hs view
@@ -0,0 +1,114 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++module SubscriptionTreeSpec (spec) where++import Data.List (sortOn)+import Test.Hspec (Spec, describe, it, shouldBe)+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck.Instances ()++import qualified Data.Aeson as AE+import qualified Data.HashMap.Strict as HM++import Subscription (SubscriptionTree (..), broadcast', empty, subscribe, unsubscribe)++spec :: Spec+spec = do+ describe "SubscriptionTree" $ do++ it "adds a client listening to the root when calling subscribe with []" $ do+ let+ path = []+ conn_id = 1 :: Int+ conn = "dummy connection" :: String+ after = SubscriptionTree (HM.fromList [(conn_id, conn)]) HM.empty+ subscribe path conn_id conn empty `shouldBe` after++ it "adds a client listening to \"some\" when calling subscribe with [\"some\"]" $ do+ let+ path = ["some"]+ conn_id = 1 :: Int+ conn = "dummy connection" :: [Char]+ after = SubscriptionTree+ HM.empty+ (HM.fromList [("some", SubscriptionTree (HM.fromList [(conn_id,conn)]) HM.empty)])+ subscribe path conn_id conn empty `shouldBe` after++ it "adds two clients: ones listening to the root and one to \"some\"" $ do+ let+ root = []+ path = ["some"]+ conn_id = 1 :: Int+ conn_id2 = 2 :: Int+ conn = "dummy connection" :: [Char]+ conn2 = "dummy connection2" :: [Char]+ after = SubscriptionTree+ (HM.fromList [(conn_id, conn)])+ (HM.fromList [("some", SubscriptionTree (HM.fromList [(conn_id2,conn2)]) HM.empty)])+ subscribe root conn_id conn (subscribe path conn_id2 conn2 empty) `shouldBe` after++ prop "adding clients is commutative" $ \ lpath rpath (lid :: Int) ->+ let+ rid = lid + 1 -- The ids need to be distinct.+ lconn = "dummy connection left" :: [Char]+ rconn = "dummy connection right" :: [Char]+ ladd = subscribe lpath lid lconn+ radd = subscribe rpath rid rconn+ in+ (ladd . radd) empty `shouldBe` (radd . ladd) empty++ prop "adding and removing a client is identity" $ \ path (cid :: Int) ->+ let+ conn = "dummy connection" :: String+ subUnsub = unsubscribe path cid . subscribe path cid conn+ in+ subUnsub empty `shouldBe` empty++ it "removing a non-existing client does nothing" $ do+ let+ path = []+ conn_id = 1 :: Int+ unsubscribe path conn_id (empty :: SubscriptionTree Int String)`shouldBe` empty++ do+ let+ conn1 , conn2 , conn3 , conn4 :: Int+ conn1 = 1; conn2 = 2; conn3 = 3; conn4 = 4++ root = SubscriptionTree (HM.fromList [(conn1, conn1)])+ (HM.fromList [ ("foo", root_foo)+ , ("baz", root_baz) ])+ root_foo = SubscriptionTree (HM.fromList [(conn2, conn2)])+ (HM.fromList [("bar", root_foo_bar)])+ root_foo_bar = SubscriptionTree (HM.fromList [(conn3, conn3)]) HM.empty+ root_baz = SubscriptionTree (HM.fromList [(conn4, conn4)]) HM.empty++ value = AE.object ["foo" AE..= value_foo, "baz" AE..= value_baz]+ value_foo = AE.object ["bar" AE..= value_foo_bar]+ value_foo_bar = AE.Null+ value_baz = AE.object []++ broadcast'' path = sortOn fst $ broadcast' path value root++ it "notifies everyone on root updates" $ do+ broadcast'' []+ `shouldBe` [ (conn1, value)+ , (conn2, value_foo)+ , (conn3, value_foo_bar)+ , (conn4, value_baz)+ ]++ it "notifies parents and children about updates" $ do+ broadcast'' ["foo"]+ `shouldBe` [ (conn1, value)+ , (conn2, value_foo)+ , (conn3, value_foo_bar)+ ]++ it "notifies parents and children about updates" $ do+ broadcast'' ["foo", "bar"]+ `shouldBe` [ (conn1, value)+ , (conn2, value_foo)+ , (conn3, value_foo_bar)+ ]