diff --git a/app/Icepeak/Main.hs b/app/Icepeak/Main.hs new file mode 100644 --- /dev/null +++ b/app/Icepeak/Main.hs @@ -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" diff --git a/app/IcepeakTokenGen/Main.hs b/app/IcepeakTokenGen/Main.hs new file mode 100644 --- /dev/null +++ b/app/IcepeakTokenGen/Main.hs @@ -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 diff --git a/icepeak.cabal b/icepeak.cabal new file mode 100644 --- /dev/null +++ b/icepeak.cabal @@ -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 diff --git a/src/AccessControl.hs b/src/AccessControl.hs new file mode 100644 --- /dev/null +++ b/src/AccessControl.hs @@ -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" diff --git a/src/Config.hs b/src/Config.hs new file mode 100644 --- /dev/null +++ b/src/Config.hs @@ -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) + ] diff --git a/src/Core.hs b/src/Core.hs new file mode 100644 --- /dev/null +++ b/src/Core.hs @@ -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 diff --git a/src/HTTPMethodInvalid.hs b/src/HTTPMethodInvalid.hs new file mode 100644 --- /dev/null +++ b/src/HTTPMethodInvalid.hs @@ -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" diff --git a/src/HttpServer.hs b/src/HttpServer.hs new file mode 100644 --- /dev/null +++ b/src/HttpServer.hs @@ -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 diff --git a/src/JwtAuth.hs b/src/JwtAuth.hs new file mode 100644 --- /dev/null +++ b/src/JwtAuth.hs @@ -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)] diff --git a/src/JwtMiddleware.hs b/src/JwtMiddleware.hs new file mode 100644 --- /dev/null +++ b/src/JwtMiddleware.hs @@ -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 @. +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) diff --git a/src/Logger.hs b/src/Logger.hs new file mode 100644 --- /dev/null +++ b/src/Logger.hs @@ -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 () diff --git a/src/Metrics.hs b/src/Metrics.hs new file mode 100644 --- /dev/null +++ b/src/Metrics.hs @@ -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 diff --git a/src/MetricsServer.hs b/src/MetricsServer.hs new file mode 100644 --- /dev/null +++ b/src/MetricsServer.hs @@ -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 diff --git a/src/Persistence.hs b/src/Persistence.hs new file mode 100644 --- /dev/null +++ b/src/Persistence.hs @@ -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' diff --git a/src/SentryLogging.hs b/src/SentryLogging.hs new file mode 100644 --- /dev/null +++ b/src/SentryLogging.hs @@ -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 diff --git a/src/Server.hs b/src/Server.hs new file mode 100644 --- /dev/null +++ b/src/Server.hs @@ -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 diff --git a/src/Store.hs b/src/Store.hs new file mode 100644 --- /dev/null +++ b/src/Store.hs @@ -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 diff --git a/src/Subscription.hs b/src/Subscription.hs new file mode 100644 --- /dev/null +++ b/src/Subscription.hs @@ -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) diff --git a/src/WebsocketServer.hs b/src/WebsocketServer.hs new file mode 100644 --- /dev/null +++ b/src/WebsocketServer.hs @@ -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 () diff --git a/tests/AccessControlSpec.hs b/tests/AccessControlSpec.hs new file mode 100644 --- /dev/null +++ b/tests/AccessControlSpec.hs @@ -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) diff --git a/tests/ApiSpec.hs b/tests/ApiSpec.hs new file mode 100644 --- /dev/null +++ b/tests/ApiSpec.hs @@ -0,0 +1,6 @@ +module ApiSpec (spec) where + +import Test.Hspec + +spec :: Spec +spec = return () diff --git a/tests/CoreSpec.hs b/tests/CoreSpec.hs new file mode 100644 --- /dev/null +++ b/tests/CoreSpec.hs @@ -0,0 +1,8 @@ +{-# LANGUAGE OverloadedStrings #-} + +module CoreSpec (spec) where + +import Test.Hspec + +spec :: Spec +spec = return () diff --git a/tests/JwtSpec.hs b/tests/JwtSpec.hs new file mode 100644 --- /dev/null +++ b/tests/JwtSpec.hs @@ -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 diff --git a/tests/OrphanInstances.hs b/tests/OrphanInstances.hs new file mode 100644 --- /dev/null +++ b/tests/OrphanInstances.hs @@ -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 + ] diff --git a/tests/PersistenceSpec.hs b/tests/PersistenceSpec.hs new file mode 100644 --- /dev/null +++ b/tests/PersistenceSpec.hs @@ -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) diff --git a/tests/RequestSpec.hs b/tests/RequestSpec.hs new file mode 100644 --- /dev/null +++ b/tests/RequestSpec.hs @@ -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" diff --git a/tests/SocketSpec.hs b/tests/SocketSpec.hs new file mode 100644 --- /dev/null +++ b/tests/SocketSpec.hs @@ -0,0 +1,6 @@ +module SocketSpec (spec) where + +import Test.Hspec + +spec :: Spec +spec = return () diff --git a/tests/Spec.hs b/tests/Spec.hs new file mode 100644 --- /dev/null +++ b/tests/Spec.hs @@ -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 diff --git a/tests/StoreSpec.hs b/tests/StoreSpec.hs new file mode 100644 --- /dev/null +++ b/tests/StoreSpec.hs @@ -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) diff --git a/tests/SubscriptionTreeSpec.hs b/tests/SubscriptionTreeSpec.hs new file mode 100644 --- /dev/null +++ b/tests/SubscriptionTreeSpec.hs @@ -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) + ]