diff --git a/main/Main.hs b/main/Main.hs index 888f76933..c5e73bab1 100644 --- a/main/Main.hs +++ b/main/Main.hs @@ -1,455 +1,48 @@ -{-# LANGUAGE CPP #-} -{-# LANGUAGE LambdaCase #-} -{-# LANGUAGE MultiWayIf #-} -{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE CPP #-} module Main (main) where -import qualified Data.Aeson as Aeson -import qualified Data.ByteString as BS -import qualified Data.ByteString.Lazy as LBS -import qualified Data.Map.Strict as M -import qualified Hasql.Connection as C -import qualified Hasql.Notifications as N -import qualified Hasql.Pool as P -import qualified Hasql.Transaction as HT -import qualified Hasql.Transaction.Sessions as HT +import qualified Data.Map.Strict as M -import Control.AutoUpdate (defaultUpdateSettings, mkAutoUpdate, - updateAction) -import Control.Retry (RetryStatus, capDelay, - exponentialBackoff, retrying, - rsPreviousDelay) -import Data.IORef (IORef, atomicWriteIORef, newIORef, - readIORef) -import Data.String (IsString (..)) -import Data.Text.IO (hPutStrLn) -import Data.Time.Clock (getCurrentTime) -import Network.Wai.Handler.Warp (defaultSettings, runSettings, - setHost, setPort, setServerName) -import System.CPUTime (getCPUTime) -import System.Environment (getEnvironment) -import System.IO (BufferMode (..), hSetBuffering) -import Text.Printf (hPrintf) +import System.IO (BufferMode (..), hSetBuffering) -import PostgREST.App (postgrest) -import PostgREST.CLI (CLI (..), Command (..), - readCLIShowHelp) -import PostgREST.Config (AppConfig (..), Environment, - configDbPoolTimeout', - dumpAppConfig, readAppConfig, - readDbUriFile, readSecretFile) -import PostgREST.DbStructure (DbStructure, getDbStructure, - getPgVersion) -import PostgREST.DbStructure.PgVersion (PgVersion (..), - minimumPgVersion) -import PostgREST.Error (PgError (PgError), - checkIsFatal, errorPayload) -import PostgREST.Query.Statements (dbSettingsStatement) -import PostgREST.Version (prettyVersion) +import qualified PostgREST.App as App +import qualified PostgREST.CLI as CLI -import qualified Data.Text as T - -import Protolude hiding (hPutStrLn, head, toS) -import Protolude.Conv (toS) +import PostgREST.Config (readPGRSTEnvironment) +import Protolude #ifndef mingw32_HOST_OS -import System.Posix.Signals -import UnixSocket +import qualified PostgREST.Unix as Unix #endif - --- | Current database connection status data ConnectionStatus -data ConnectionStatus - = NotConnected - | Connected PgVersion - | FatalConnectionError Text - deriving (Eq) - --- | Schema cache status -data SCacheStatus - = SCLoaded - | SCOnRetry - | SCFatalFail - --- | This is where everything starts. main :: IO () main = do - -- + setBuffering + hasPGRSTEnv <- not . M.null <$> readPGRSTEnvironment + opts <- CLI.readCLIShowHelp hasPGRSTEnv + CLI.main installSignalHandlers runAppInSocket opts + +installSignalHandlers :: App.SignalHandlerInstaller +#ifndef mingw32_HOST_OS +installSignalHandlers = Unix.installSignalHandlers +#else +installSignalHandlers _ = pass +#endif + +runAppInSocket :: App.SocketRunner +#ifndef mingw32_HOST_OS +runAppInSocket = Unix.runAppInSocket +#else +runAppInSocket _ _ _ _ = pass +#endif + +setBuffering :: IO () +setBuffering = do -- LineBuffering: the entire output buffer is flushed whenever a newline is -- output, the buffer overflows, a hFlush is issued or the handle is closed - -- - -- NoBuffering: output is written immediately and never stored in the buffer hSetBuffering stdout LineBuffering hSetBuffering stdin LineBuffering + -- NoBuffering: output is written immediately and never stored in the buffer hSetBuffering stderr NoBuffering - - -- read PGRST_ env variables - env <- readEnvironment - - -- read command/path from commad line - CLI{cliCommand, cliPath} <- readCLIShowHelp . not $ M.null env - - -- build the 'AppConfig' from the config file path and env vars - pathEnvConf <- either panic identity <$> readAppConfig mempty env cliPath Nothing Nothing - - -- read external files - dbUriFile <- readDbUriFile $ configDbUri pathEnvConf - secretFile <- readSecretFile $ configJwtSecret pathEnvConf - - -- add the external files to AppConfig - conf <- either panic identity <$> readAppConfig mempty env cliPath dbUriFile secretFile - - -- These are config values that can't be reloaded at runtime. Reloading some of them would imply restarting the web server. - let - host = configServerHost conf - port = configServerPort conf - maybeSocketAddr = configServerUnixSocket conf -#ifndef mingw32_HOST_OS - socketFileMode = configServerUnixSocketMode conf -#endif - dbUri = toS (configDbUri conf) - (dbChannelEnabled, dbChannel) = (configDbChannelEnabled conf, toS $ configDbChannel conf) - serverSettings = - setHost ((fromString . toS) host) -- Warp settings - . setPort port - . setServerName (toS $ "postgrest/" <> prettyVersion) $ - defaultSettings - poolSize = configDbPoolSize conf - poolTimeout = configDbPoolTimeout' conf - logLevel = configLogLevel conf - dbConfigEnabled = configDbConfig conf - - -- create connection pool with the provided settings, returns either a 'Connection' or a 'ConnectionError'. Does not throw. - pool <- P.acquire (poolSize, poolTimeout, dbUri) - - -- Used to sync the listener(NOTIFY reload) with the connectionWorker. No connection for the listener at first. Only used if dbChannelEnabled=true. - mvarConnectionStatus <- newEmptyMVar - - -- No schema cache at the start. Will be filled in by the connectionWorker - refDbStructure <- newIORef Nothing - - -- Helper ref to make sure just one connectionWorker can run at a time - refIsWorkerOn <- newIORef False - - -- Config that can change at runtime - refConf <- newIORef conf - - let - -- re-reads config file + db config - dbConfigReReader startingUp = when dbConfigEnabled $ - reReadConfig startingUp pool dbConfigEnabled env cliPath refConf dbUriFile secretFile - -- re-reads jwt-secret external file + config file + db config - fullConfigReReader = - reReadConfig False pool dbConfigEnabled env cliPath refConf - dbUriFile =<< -- db-uri external file could be re-read, but it doesn't make sense as db-uri is not reloadable - readSecretFile (configJwtSecret pathEnvConf) - - -- Override the config with config options from the db - -- TODO: the same operation is repeated on connectionWorker, ideally this would be done only once, but dump CmdDumpConfig needs it for tests. - dbConfigReReader True - - case cliCommand of - CmdDumpConfig -> - do - dumpedConfig <- dumpAppConfig <$> readIORef refConf - putStr dumpedConfig - exitSuccess - CmdDumpSchema -> - do - dumpedSchema <- dumpSchema pool =<< readIORef refConf - putStrLn dumpedSchema - exitSuccess - CmdRun -> - pass - - -- This is passed to the connectionWorker method so it can kill the main thread if the PostgreSQL's version is not supported. - mainTid <- myThreadId - - let connWorker = connectionWorker mainTid pool refConf refDbStructure refIsWorkerOn (dbChannelEnabled, mvarConnectionStatus) $ - dbConfigReReader False - - -- Sets the initial refDbStructure - connWorker - -#ifndef mingw32_HOST_OS - -- Only for systems with signals: - -- - -- releases the connection pool whenever the program is terminated, - -- see https://github.com/PostgREST/postgrest/issues/268 - forM_ [sigINT, sigTERM] $ \sig -> - void $ installHandler sig (Catch $ do - P.release pool - throwTo mainTid UserInterrupt - ) Nothing - - -- The SIGUSR1 signal updates the internal 'DbStructure' by running 'connectionWorker' exactly as before. - void $ installHandler sigUSR1 ( - Catch connWorker - ) Nothing - - -- Re-read the config on SIGUSR2 - void $ installHandler sigUSR2 ( - Catch fullConfigReReader - ) Nothing -#endif - - -- reload schema cache + config on NOTIFY - when dbChannelEnabled $ - listener dbUri dbChannel pool refConf refDbStructure mvarConnectionStatus connWorker fullConfigReReader - - -- ask for the OS time at most once per second - getTime <- mkAutoUpdate defaultUpdateSettings {updateAction = getCurrentTime} - - let postgrestApplication = - postgrest - logLevel - refConf - refDbStructure - pool - getTime - connWorker - -#ifndef mingw32_HOST_OS - -- run the postgrest application with user defined socket. Only for UNIX systems. - whenJust maybeSocketAddr $ - runAppInSocket serverSettings postgrestApplication socketFileMode -#endif - - -- run the postgrest application - whenNothing maybeSocketAddr $ do - putStrLn $ ("Listening on port " :: Text) <> show port - runSettings serverSettings postgrestApplication - -readEnvironment :: IO Environment -readEnvironment = getEnvironment <&> pgrst - where - pgrst env = M.filterWithKey (\k _ -> "PGRST_" `isPrefixOf` k) $ M.map T.pack $ M.fromList env - --- Time constants -_32s :: Int -_32s = 32000000 :: Int -- 32 seconds - -_1s :: Int -_1s = 1000000 :: Int -- 1 second - -{-| - The purpose of this worker is to obtain a healthy connection to pg and an up-to-date schema cache(DbStructure). - This method is meant to be called by multiple times by the same thread, but does nothing if - the previous invocation has not terminated. In all cases this method does not - halt the calling thread, the work is preformed in a separate thread. - - Note: 'atomicWriteIORef' is essentially a lazy semaphore that prevents two - threads from running 'connectionWorker' at the same time. - - Background thread that does the following : - 1. Tries to connect to pg server and will keep trying until success. - 2. Checks if the pg version is supported and if it's not it kills the main - program. - 3. Obtains the dbStructure. If this fails, it goes back to 1. --} -connectionWorker - :: ThreadId -- ^ Main thread id. Killed if pg version is unsupported - -> P.Pool -- ^ The pg connection pool - -> IORef AppConfig -- ^ mutable reference to AppConfig - -> IORef (Maybe DbStructure) -- ^ mutable reference to 'DbStructure' - -> IORef Bool -- ^ Used as a binary Semaphore - -> (Bool, MVar ConnectionStatus) -- ^ For interacting with the LISTEN channel - -> IO () - -> IO () -connectionWorker mainTid pool refConf refDbStructure refIsWorkerOn (dbChannelEnabled, mvarConnectionStatus) dbCfReader = do - isWorkerOn <- readIORef refIsWorkerOn - unless isWorkerOn $ do -- Prevents multiple workers to be running at the same time. Could happen on too many SIGUSR1s. - atomicWriteIORef refIsWorkerOn True - void $ forkIO work - where - work = do - putStrLn ("Attempting to connect to the database..." :: Text) - connected <- connectionStatus pool - when dbChannelEnabled $ - void $ tryPutMVar mvarConnectionStatus connected -- tryPutMVar doesn't lock the thread. It should always succeed since the worker is the only mvar producer. - case connected of - FatalConnectionError reason -> hPutStrLn stderr reason >> killThread mainTid -- Fatal error when connecting - NotConnected -> return () -- Unreachable because connectionStatus will keep trying to connect - Connected actualPgVersion -> do -- Procede with initialization - putStrLn ("Connection successful" :: Text) - dbCfReader -- this could be fail because the connection drops, but the loadSchemaCache will pick the error and retry again - scStatus <- loadSchemaCache pool actualPgVersion refConf refDbStructure - case scStatus of - SCLoaded -> pure () -- do nothing and proceed if the load was successful - SCOnRetry -> work -- retry - SCFatalFail -> killThread mainTid -- die if our schema cache query has an error - liftIO $ atomicWriteIORef refIsWorkerOn False - -{-| - Check if a connection from the pool allows access to the PostgreSQL database. - If not, the pool connections are released and a new connection is tried. - Releasing the pool is key for rapid recovery. Otherwise, the pool timeout would have to be reached for new healthy connections to be acquired. - Which might not happen if the server is busy with requests. No idle connection, no pool timeout. - - The connection tries are capped, but if the connection times out no error is thrown, just 'False' is returned. --} -connectionStatus :: P.Pool -> IO ConnectionStatus -connectionStatus pool = - retrying (capDelay _32s $ exponentialBackoff _1s) - shouldRetry - (const $ P.release pool >> getConnectionStatus) - where - getConnectionStatus :: IO ConnectionStatus - getConnectionStatus = do - pgVersion <- P.use pool getPgVersion - case pgVersion of - Left e -> do - let err = PgError False e - hPutStrLn stderr . toS $ errorPayload err - case checkIsFatal err of - Just reason -> return $ FatalConnectionError reason - Nothing -> return NotConnected - - Right version -> - if version < minimumPgVersion - then return . FatalConnectionError $ "Cannot run in this PostgreSQL version, PostgREST needs at least " <> pgvName minimumPgVersion - else return . Connected $ version - - shouldRetry :: RetryStatus -> ConnectionStatus -> IO Bool - shouldRetry rs isConnSucc = do - let delay = fromMaybe 0 (rsPreviousDelay rs) `div` _1s - itShould = NotConnected == isConnSucc - when itShould $ - putStrLn $ "Attempting to reconnect to the database in " <> (show delay::Text) <> " seconds..." - return itShould - --- | Load the DbStructure by using a connection from the pool. -loadSchemaCache :: P.Pool -> PgVersion -> IORef AppConfig -> IORef (Maybe DbStructure) -> IO SCacheStatus -loadSchemaCache pool actualPgVersion refConf refDbStructure = do - conf <- readIORef refConf - result <- P.use pool $ HT.transaction HT.ReadCommitted HT.Read $ getDbStructure (toList $ configDbSchemas conf) (configDbExtraSearchPath conf) actualPgVersion (configDbPreparedStatements conf) - case result of - Left e -> do - let err = PgError False e - putErr = hPutStrLn stderr . toS . errorPayload $ err - case checkIsFatal err of - Just _ -> do - hPutStrLn stderr ("A fatal error ocurred when loading the schema cache" :: Text) - putErr - hPutStrLn stderr ("This is probably a bug in PostgREST, please report it at https://github.com/PostgREST/postgrest/issues" :: Text) - return SCFatalFail - Nothing -> do - hPutStrLn stderr ("An error ocurred when loading the schema cache" :: Text) >> putErr - return SCOnRetry - - Right dbStructure -> do - atomicWriteIORef refDbStructure $ Just dbStructure - putStrLn ("Schema cache loaded" :: Text) - return SCLoaded - -{-| - Starts a dedicated pg connection to LISTEN for notifications. - When a NOTIFY - with an empty payload - is done, it refills the schema cache. - It uses the connectionWorker in case the LISTEN connection dies. --} -listener :: ByteString -> Text -> P.Pool -> IORef AppConfig -> IORef (Maybe DbStructure) -> MVar ConnectionStatus -> IO () -> IO () -> IO () -listener dbUri dbChannel pool refConf refDbStructure mvarConnectionStatus connWorker configLoader = start - where - start = do - connStatus <- takeMVar mvarConnectionStatus -- takeMVar makes the thread wait if the MVar is empty(until there's a connection). - case connStatus of - Connected actualPgVersion -> void $ forkFinally (do -- forkFinally allows to detect if the thread dies - dbOrError <- C.acquire dbUri - case dbOrError of - Right db -> do - putStrLn $ "Listening for notifications on the " <> dbChannel <> " channel" - let channelToListen = N.toPgIdentifier dbChannel - scLoader = void $ loadSchemaCache pool actualPgVersion refConf refDbStructure -- It's not necessary to check the loadSchemaCache success here. If the connection drops, the thread will die and proceed to recover below. - N.listen db channelToListen - N.waitForNotifications (\_ msg -> - if | BS.null msg -> scLoader -- reload the schema cache - | msg == "reload schema" -> scLoader -- reload the schema cache - | msg == "reload config" -> configLoader -- reload the config - | otherwise -> pure () -- Do nothing if anything else than an empty message is sent - ) db - _ -> die errorMessage) - (\_ -> do -- if the thread dies, we try to recover - putStrLn retryMessage - connWorker -- assume the pool connection was also lost, call the connection worker - start) -- retry the listener - _ -> - putStrLn errorMessage -- Should be unreachable. connectionStatus will retry until there's a connection. - errorMessage = "Could not listen for notifications on the " <> dbChannel <> " channel" :: Text - retryMessage = "Retrying listening for notifications on the " <> dbChannel <> " channel.." :: Text - --- | Re-reads the config plus config options from the db -reReadConfig :: Bool -> P.Pool -> Bool -> Environment -> Maybe FilePath -> IORef AppConfig -> Maybe Text -> Maybe BS.ByteString -> IO () -reReadConfig startingUp pool dbConfigEnabled env path refConf dbUriFile secretFile = do - dbSettings <- if dbConfigEnabled then loadDbSettings else pure [] - readAppConfig dbSettings env path dbUriFile secretFile >>= \case - Left err -> - if startingUp - then panic err -- die on invalid config if the program is starting up - else hPutStrLn stderr $ "Failed loading in-database config. " <> err - Right conf -> do - atomicWriteIORef refConf conf - if startingUp - then pass - else putStrLn ("In-database config loaded" :: Text) - where - loadDbSettings :: IO [(Text, Text)] - loadDbSettings = do - result <- P.use pool $ HT.transaction HT.ReadCommitted HT.Read $ HT.statement mempty dbSettingsStatement - case result of - Left e -> do - hPutStrLn stderr ("An error ocurred when trying to query database settings for the config parameters:\n" <> show e :: Text) - pure [] - Right x -> pure x - --- | Dump DbStructure schema to JSON -dumpSchema :: P.Pool -> AppConfig -> IO LBS.ByteString -dumpSchema pool conf = do - result <- - timeToStderr "Loaded schema in %.3f seconds" $ - P.use pool $ do - pgVersion <- getPgVersion - HT.transaction HT.ReadCommitted HT.Read $ - getDbStructure - (toList $ configDbSchemas conf) - (configDbExtraSearchPath conf) - pgVersion - (configDbPreparedStatements conf) - P.release pool - case result of - Left e -> do - hPutStrLn stderr $ "An error ocurred when loading the schema cache:\n" <> show e - exitFailure - Right dbStructure -> return $ Aeson.encode dbStructure - --- | Print the time taken to run an IO action to stderr with the given printf string -timeToStderr :: [Char] -> IO (Either a b) -> IO (Either a b) -timeToStderr fmtString a = - do - start <- getCPUTime - result <- a - end <- getCPUTime - let - duration :: Double - duration = fromIntegral (end - start) / picoseconds - when (isRight result) $ - hPrintf stderr (fmtString ++ "\n") duration - return result - - --- | 10^12 picoseconds per second -picoseconds :: Double -picoseconds = 1000000000000 - - --- Utility functions. -#ifndef mingw32_HOST_OS -whenJust :: Applicative f => Maybe a -> (a -> f ()) -> f () -whenJust (Just x) f = f x -whenJust Nothing _ = pass -#endif - -whenNothing :: Applicative f => Maybe a -> f () -> f () -whenNothing Nothing f = f -whenNothing _ _ = pass diff --git a/main/UnixSocket.hs b/main/UnixSocket.hs deleted file mode 100644 index 36359bc55..000000000 --- a/main/UnixSocket.hs +++ /dev/null @@ -1,40 +0,0 @@ -module UnixSocket ( - runAppInSocket -)where - -import Network.Socket (Family (AF_UNIX), - SockAddr (SockAddrUnix), Socket, - SocketType (Stream), bind, close, - defaultProtocol, listen, - maxListenQueue, socket) -import Network.Wai (Application) -import Network.Wai.Handler.Warp -import System.Directory (removeFile) -import System.IO.Error (isDoesNotExistError) -import System.Posix.Files (setFileMode) -import System.Posix.Types (FileMode) - -import Protolude - -createAndBindSocket :: FilePath -> FileMode -> IO Socket -createAndBindSocket socketFilePath socketFileMode = do - deleteSocketFileIfExist socketFilePath - sock <- socket AF_UNIX Stream defaultProtocol - bind sock $ SockAddrUnix socketFilePath - setFileMode socketFilePath socketFileMode - return sock - where - deleteSocketFileIfExist path = removeFile path `catch` handleDoesNotExist - handleDoesNotExist e - | isDoesNotExistError e = return () - | otherwise = throwIO e - --- run the postgrest application with user defined socket. -runAppInSocket :: Settings -> Application -> FileMode -> FilePath -> IO () -runAppInSocket settings app socketFileMode sockPath = do - sock <- createAndBindSocket sockPath socketFileMode - putStrLn $ ("Listening on unix socket " :: Text) <> show sockPath - listen sock maxListenQueue - runSettingsSocket settings sock app - -- clean socket up when done - close sock diff --git a/postgrest.cabal b/postgrest.cabal index 1ee117302..5b1829d3f 100644 --- a/postgrest.cabal +++ b/postgrest.cabal @@ -35,9 +35,11 @@ library NoImplicitPrelude hs-source-dirs: src exposed-modules: PostgREST.App + PostgREST.AppState PostgREST.Auth PostgREST.CLI PostgREST.Config + PostgREST.Config.Database PostgREST.Config.JSPath PostgREST.Config.Proxy PostgREST.ContentType @@ -61,12 +63,14 @@ library PostgREST.Request.Preferences PostgREST.Request.Types PostgREST.Version + PostgREST.Workers other-modules: Paths_postgrest build-depends: base >= 4.9 && < 4.15 , HTTP >= 4000.3.7 && < 4000.4 , Ranged-sets >= 0.3 && < 0.5 , aeson >= 1.4.7 && < 1.6 , ansi-wl-pprint >= 0.6.7 && < 0.7 + , auto-update >= 0.1.4 && < 0.2 , base64-bytestring >= 1 && < 1.3 , bytestring >= 0.10.8 && < 0.11 , case-insensitive >= 1.2 && < 1.3 @@ -81,6 +85,7 @@ library , gitrev >= 1.2 && < 1.4 , hasql >= 1.4 && < 1.5 , hasql-dynamic-statements == 0.3.1 + , hasql-notifications >= 0.1 && < 0.2 , hasql-pool >= 0.5 && < 0.6 , hasql-transaction >= 0.7.2 && < 1.1 , heredoc >= 0.2 && < 0.3 @@ -96,6 +101,7 @@ library , parsec >= 3.1.11 && < 3.2 , protolude >= 0.3 && < 0.4 , regex-tdfa >= 1.2.2 && < 1.4 + , retry >= 0.7.4 && < 0.9 , scientific >= 0.3.4 && < 0.4 , swagger2 >= 2.4 && < 2.7 , text >= 1.2.2 && < 1.3 @@ -107,6 +113,7 @@ library , wai-extra >= 3.0.19 && < 3.2 , wai-logger >= 2.3.2 , wai-middleware-static >= 0.8.1 && < 0.10 + , warp >= 3.2.12 && < 3.4 -- -fno-spec-constr may help keep compile time memory use in check, -- see https://gitlab.haskell.org/ghc/ghc/issues/16017#note_219304 -- -optP-Wno-nonportable-include-path @@ -122,6 +129,14 @@ library else ghc-options: -O2 + if !os(windows) + build-depends: + unix + , directory >= 1.2.6 && < 1.4 + , network >= 2.6 && < 3.2 + exposed-modules: + PostgREST.Unix + executable postgrest default-language: Haskell2010 default-extensions: OverloadedStrings @@ -129,25 +144,9 @@ executable postgrest hs-source-dirs: main main-is: Main.hs build-depends: base >= 4.9 && < 4.15 - , aeson >= 1.4.7 && < 1.6 - , auto-update >= 0.1.4 && < 0.2 - , base64-bytestring >= 1 && < 1.3 - , bytestring >= 0.10.8 && < 0.11 , containers >= 0.5.7 && < 0.7 - , directory >= 1.2.6 && < 1.4 - , either >= 4.4.1 && < 5.1 - , hasql >= 1.4 && < 1.5 - , hasql-pool >= 0.5 && < 0.6 - , hasql-transaction >= 0.7.2 && < 1.1 - , hasql-notifications >= 0.1 && < 0.2 - , network >= 2.6 && < 3.2 , postgrest , protolude >= 0.3 && < 0.4 - , retry >= 0.7.4 && < 0.9 - , text >= 1.2.2 && < 1.3 - , time >= 1.6 && < 1.11 - , wai >= 3.2.1 && < 3.3 - , warp >= 3.2.12 && < 3.4 ghc-options: -threaded -rtsopts "-with-rtsopts=-N -I2" -O2 -Werror -Wall -fwarn-identities -fno-spec-constr -optP-Wno-nonportable-include-path @@ -159,10 +158,6 @@ executable postgrest else ghc-options: -O2 - if !os(windows) - build-depends: unix - other-modules: UnixSocket - test-suite spec type: exitcode-stdio-1.0 default-language: Haskell2010 diff --git a/src/PostgREST/App.hs b/src/PostgREST/App.hs index 6fb885ccc..1066461c8 100644 --- a/src/PostgREST/App.hs +++ b/src/PostgREST/App.hs @@ -10,13 +10,21 @@ Some of its functionality includes: - Content Negotiation -} {-# LANGUAGE RecordWildCards #-} -module PostgREST.App (postgrest) where +module PostgREST.App + ( SignalHandlerInstaller + , SocketRunner + , postgrest + , run + ) where -import Control.Monad.Except (liftEither) -import Data.Either.Combinators (mapLeft) -import Data.IORef (IORef, readIORef) -import Data.List (union) -import Data.Time.Clock (UTCTime) +import Control.Monad.Except (liftEither) +import Data.Either.Combinators (mapLeft) +import Data.List (union) +import Data.String (IsString (..)) +import Data.Time.Clock (UTCTime) +import Network.Wai.Handler.Warp (defaultSettings, setHost, setPort, + setServerName) +import System.Posix.Types (FileMode) import qualified Data.ByteString.Char8 as BS8 import qualified Data.ByteString.Lazy as LBS @@ -29,7 +37,9 @@ import qualified Network.HTTP.Types.Header as HTTP import qualified Network.HTTP.Types.Status as HTTP import qualified Network.HTTP.Types.URI as HTTP import qualified Network.Wai as Wai +import qualified Network.Wai.Handler.Warp as Warp +import qualified PostgREST.AppState as AppState import qualified PostgREST.Auth as Auth import qualified PostgREST.DbStructure as DbStructure import qualified PostgREST.Error as Error @@ -41,6 +51,7 @@ import qualified PostgREST.RangeQuery as RangeQuery import qualified PostgREST.Request.ApiRequest as ApiRequest import qualified PostgREST.Request.DbRequestBuilder as ReqBuilder +import PostgREST.AppState (AppState) import PostgREST.Config (AppConfig (..), LogLevel (..)) import PostgREST.ContentType (ContentType (..)) @@ -64,6 +75,8 @@ import PostgREST.Request.Preferences (PreferCount (..), PreferParameters (..), PreferRepresentation (..)) import PostgREST.Request.Types (ReadRequest, fstFieldNames) +import PostgREST.Version (prettyVersion) +import PostgREST.Workers (connectionWorker, listener) import qualified PostgREST.ContentType as ContentType import qualified PostgREST.DbStructure.Proc as Proc @@ -83,27 +96,50 @@ type Handler = ExceptT Error type DbHandler = Handler SQL.Transaction +type SignalHandlerInstaller = AppState -> IO() + +type SocketRunner = Warp.Settings -> Wai.Application -> FileMode -> FilePath -> IO() + + +run :: SignalHandlerInstaller -> SocketRunner -> AppState -> IO () +run installHandlers runInSocket appState = do + conf@AppConfig{..} <- AppState.getConfig appState + connectionWorker appState -- Loads the initial DbStructure + installHandlers appState + -- reload schema cache + config on NOTIFY + when configDbChannelEnabled $ listener appState + + let app = postgrest configLogLevel appState (connectionWorker appState) + + case configServerUnixSocket of + Just socket -> + -- run the postgrest application with user defined socket. Only for UNIX systems. + runInSocket (serverSettings conf) app configServerUnixSocketMode socket + Nothing -> + do + putStrLn $ ("Listening on port " :: Text) <> show configServerPort + Warp.runSettings (serverSettings conf) app + +serverSettings :: AppConfig -> Warp.Settings +serverSettings AppConfig{..} = + defaultSettings + & setHost (fromString $ toS configServerHost) + & setPort configServerPort + & setServerName (toS $ "postgrest/" <> prettyVersion) -- | PostgREST application -postgrest - :: LogLevel - -> IORef AppConfig - -> IORef (Maybe DbStructure) - -> SQL.Pool - -> IO UTCTime - -> IO () -- ^ Lauch connection worker in a separate thread - -> Wai.Application -postgrest logLev refConf refDbStructure pool getTime connWorker = +postgrest :: LogLevel -> AppState.AppState -> IO () -> Wai.Application +postgrest logLev appState connWorker = Middleware.pgrstMiddleware logLev $ \req respond -> do - time <- getTime - conf <- readIORef refConf - maybeDbStructure <- readIORef refDbStructure + time <- AppState.getTime appState + conf <- AppState.getConfig appState + maybeDbStructure <- AppState.getDbStructure appState let eitherResponse :: IO (Either Error Wai.Response) eitherResponse = - runExceptT $ postgrestResponse conf maybeDbStructure pool time req + runExceptT $ postgrestResponse conf maybeDbStructure (AppState.getPool appState) time req response <- either Error.errorResponseFor identity <$> eitherResponse diff --git a/src/PostgREST/AppState.hs b/src/PostgREST/AppState.hs new file mode 100644 index 000000000..897f4898b --- /dev/null +++ b/src/PostgREST/AppState.hs @@ -0,0 +1,109 @@ +{-# LANGUAGE RecordWildCards #-} + +module PostgREST.AppState + ( AppState + , getConfig + , getDbStructure + , getIsWorkerOn + , getMainThreadId + , getPgVersion + , getPool + , getTime + , init + , initWithPool + , putConfig + , putDbStructure + , putIsWorkerOn + , putPgVersion + , releasePool + ) where + +import qualified Hasql.Pool as P + +import Control.AutoUpdate (defaultUpdateSettings, mkAutoUpdate, + updateAction) +import Data.IORef (IORef, atomicWriteIORef, newIORef, + readIORef) +import Data.Time.Clock (UTCTime, getCurrentTime) + +import PostgREST.Config (AppConfig (..)) +import PostgREST.DbStructure (DbStructure) +import PostgREST.DbStructure.PgVersion (PgVersion (..)) + +import Protolude hiding (toS) +import Protolude.Conv (toS) + + +data AppState = AppState + { statePool :: P.Pool -- | Connection pool, either a 'Connection' or a 'ConnectionError' + -- | Used to sync the listener(NOTIFY reload) with the connectionWorker. No + -- connection for the listener at first. Only used if dbChannelEnabled=true. + , statePgVersion :: MVar PgVersion + -- | No schema cache at the start. Will be filled in by the connectionWorker + , stateDbStructure :: IORef (Maybe DbStructure) + -- | Helper ref to make sure just one connectionWorker can run at a time + , stateIsWorkerOn :: IORef Bool + -- | Config that can change at runtime + , stateConf :: IORef AppConfig + , stateGetTime :: IO UTCTime + , stateMainThreadId :: ThreadId + } + +init :: AppConfig -> IO AppState +init conf = do + newPool <- initPool conf + initWithPool newPool conf + +initWithPool :: P.Pool -> AppConfig -> IO AppState +initWithPool newPool conf = + AppState newPool + <$> newEmptyMVar + <*> newIORef Nothing + <*> newIORef False + <*> newIORef conf + <*> mkAutoUpdate defaultUpdateSettings { updateAction = getCurrentTime } + <*> myThreadId + +initPool :: AppConfig -> IO P.Pool +initPool AppConfig{..} = + P.acquire (configDbPoolSize, configDbPoolTimeout, toS configDbUri) + +getPool :: AppState -> P.Pool +getPool = statePool + +releasePool :: AppState -> IO () +releasePool AppState{..} = P.release statePool >> throwTo stateMainThreadId UserInterrupt + +-- | As this IO action uses `takeMVar` internally, it will only return once +-- `statePgVersion` has been set using `putPgVersion`. This is currently used +-- to syncronize workers. +getPgVersion :: AppState -> IO PgVersion +getPgVersion = takeMVar . statePgVersion + +putPgVersion :: AppState -> PgVersion -> IO () +putPgVersion appState pgVer = void $ tryPutMVar (statePgVersion appState) pgVer + +getDbStructure :: AppState -> IO (Maybe DbStructure) +getDbStructure = readIORef . stateDbStructure + +putDbStructure :: AppState -> DbStructure -> IO () +putDbStructure appState structure = + atomicWriteIORef (stateDbStructure appState) $ Just structure + +getIsWorkerOn :: AppState -> IO Bool +getIsWorkerOn = readIORef . stateIsWorkerOn + +putIsWorkerOn :: AppState -> Bool -> IO () +putIsWorkerOn = atomicWriteIORef . stateIsWorkerOn + +getConfig :: AppState -> IO AppConfig +getConfig = readIORef . stateConf + +putConfig :: AppState -> AppConfig -> IO () +putConfig = atomicWriteIORef . stateConf + +getTime :: AppState -> IO UTCTime +getTime = stateGetTime + +getMainThreadId :: AppState -> ThreadId +getMainThreadId = stateMainThreadId diff --git a/src/PostgREST/CLI.hs b/src/PostgREST/CLI.hs index eab7b5c8a..13a76df8d 100644 --- a/src/PostgREST/CLI.hs +++ b/src/PostgREST/CLI.hs @@ -1,19 +1,73 @@ -{-# LANGUAGE QuasiQuotes #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE QuasiQuotes #-} +{-# LANGUAGE RecordWildCards #-} module PostgREST.CLI - ( CLI (..) + ( main + , CLI (..) , Command (..) , readCLIShowHelp ) where -import qualified Options.Applicative as O -import qualified Protolude.Conv as Conv +import qualified Data.Aeson as Aeson +import qualified Data.ByteString.Lazy as LBS +import qualified Hasql.Pool as P +import qualified Hasql.Transaction.Sessions as HT +import qualified Options.Applicative as O +import qualified Protolude.Conv as Conv +import Data.Text.IO (hPutStrLn) import Text.Heredoc (str) -import PostgREST.Version (prettyVersion) +import PostgREST.AppState (AppState) +import PostgREST.Config (AppConfig (..)) +import PostgREST.DbStructure (getDbStructure, getPgVersion) +import PostgREST.Version (prettyVersion) +import PostgREST.Workers (reReadConfig) -import Protolude +import qualified PostgREST.App as App +import qualified PostgREST.AppState as AppState +import qualified PostgREST.Config as Config +import Protolude hiding (hPutStrLn) + + +main :: App.SignalHandlerInstaller -> App.SocketRunner -> CLI -> IO () +main installSignalHandlers runAppInSocket CLI{cliCommand, cliPath} = do + conf@AppConfig{..} <- + either panic identity <$> Config.readAppConfig mempty cliPath Nothing + appState <- AppState.init conf + + -- Override the config with config options from the db + -- TODO: the same operation is repeated on connectionWorker, ideally this + -- would be done only once, but dump CmdDumpConfig needs it for tests. + when configDbConfig $ reReadConfig True appState + + exec cliCommand appState + where + exec :: Command -> AppState -> IO () + exec CmdDumpConfig appState = putStr . Config.toText =<< AppState.getConfig appState + exec CmdDumpSchema appState = putStrLn =<< dumpSchema appState + exec CmdRun appState = App.run installSignalHandlers runAppInSocket appState + +-- | Dump DbStructure schema to JSON +dumpSchema :: AppState -> IO LBS.ByteString +dumpSchema appState = do + AppConfig{..} <- AppState.getConfig appState + result <- + P.use (AppState.getPool appState) $ do + pgVersion <- getPgVersion + HT.transaction HT.ReadCommitted HT.Read $ + getDbStructure + (toList configDbSchemas) + configDbExtraSearchPath + pgVersion + configDbPreparedStatements + P.release $ AppState.getPool appState + case result of + Left e -> do + hPutStrLn stderr $ "An error ocurred when loading the schema cache:\n" <> show e + exitFailure + Right dbStructure -> return $ Aeson.encode dbStructure -- | Command line interface options data CLI = CLI diff --git a/src/PostgREST/Config.hs b/src/PostgREST/Config.hs index 21843b791..3b63078ee 100644 --- a/src/PostgREST/Config.hs +++ b/src/PostgREST/Config.hs @@ -7,6 +7,7 @@ Description : Manages PostgREST configuration type and parser. {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE RecordWildCards #-} {-# OPTIONS_GHC -fno-warn-type-defaults #-} module PostgREST.Config @@ -16,14 +17,12 @@ module PostgREST.Config , JSPathExp(..) , LogLevel(..) , Proxy(..) - , configDbPoolTimeout' - , dumpAppConfig + , toText , isMalformedProxyUri - , parseSecret , readAppConfig - , readDbUriFile - , readSecretFile + , readPGRSTEnvironment , toURI + , parseSecret ) where import qualified Crypto.JOSE.Types as JOSE @@ -47,7 +46,9 @@ import Data.List (lookup) import Data.List.NonEmpty (fromList, toList) import Data.Maybe (fromJust) import Data.Scientific (floatingOrInteger) +import Data.Time.Clock (NominalDiffTime) import Numeric (readOct, showOct) +import System.Environment (getEnvironment) import System.Posix.Types (FileMode) import PostgREST.Config.JSPath (JSPath, JSPathExp (..), pRoleClaimKey) @@ -58,15 +59,15 @@ import Protolude hiding (Proxy, toList, toS) import Protolude.Conv (toS) -data AppConfig = AppConfig { - configAppSettings :: [(Text, Text)] +data AppConfig = AppConfig + { configAppSettings :: [(Text, Text)] , configDbAnonRole :: Text , configDbChannel :: Text , configDbChannelEnabled :: Bool , configDbExtraSearchPath :: [Text] , configDbMaxRows :: Maybe Integer , configDbPoolSize :: Int - , configDbPoolTimeout :: Int + , configDbPoolTimeout :: NominalDiffTime , configDbPreRequest :: Maybe Text , configDbPreparedStatements :: Bool , configDbRootSpec :: Maybe Text @@ -75,6 +76,7 @@ data AppConfig = AppConfig { , configDbTxAllowOverride :: Bool , configDbTxRollbackAll :: Bool , configDbUri :: Text + , configFilePath :: Maybe FilePath , configJWKS :: Maybe JWKSet , configJwtAudience :: Maybe StringOrURI , configJwtRoleClaimKey :: JSPath @@ -89,10 +91,6 @@ data AppConfig = AppConfig { , configServerUnixSocketMode :: FileMode } -configDbPoolTimeout' :: (Fractional a) => AppConfig -> a -configDbPoolTimeout' = - fromRational . toRational . configDbPoolTimeout - data LogLevel = LogCrit | LogError | LogWarn | LogInfo instance Show LogLevel where @@ -102,10 +100,9 @@ instance Show LogLevel where show LogInfo = "info" -- | Dump the config -dumpAppConfig :: AppConfig -> Text -dumpAppConfig conf = - unlines $ (\(k, v) -> k <> " = " <> v) <$> - pgrstSettings ++ appSettings +toText :: AppConfig -> Text +toText conf = + unlines $ (\(k, v) -> k <> " = " <> v) <$> pgrstSettings ++ appSettings where -- apply conf to all pgrst settings pgrstSettings = (\(k, v) -> (k, v conf)) <$> @@ -115,7 +112,7 @@ dumpAppConfig conf = ,("db-extra-search-path", q . T.intercalate "," . configDbExtraSearchPath) ,("db-max-rows", maybe "\"\"" show . configDbMaxRows) ,("db-pool", show . configDbPoolSize) - ,("db-pool-timeout", show . configDbPoolTimeout) + ,("db-pool-timeout", show . floor . configDbPoolTimeout) ,("db-pre-request", q . fromMaybe mempty . configDbPreRequest) ,("db-prepared-statements", T.toLower . show . configDbPreparedStatements) ,("db-root-spec", q . fromMaybe mempty . configDbRootSpec) @@ -165,78 +162,69 @@ instance JustIfMaybe a a where instance JustIfMaybe a (Maybe a) where justIfMaybe a = Just a --- | Reads and parses the config and overrides its parameters from env vars, files or db settings. -readAppConfig :: [(Text, Text)] -> Environment -> Maybe FilePath -> Maybe Text -> Maybe B.ByteString -> IO (Either Text AppConfig) -readAppConfig dbSettings env optPath dbUriFile secretFile = do - -- Now read the actual config file - conf <- case optPath of - -- Both C.ParseError and IOError are shown here - Just cfgPath -> mapLeft show <$> (try $ C.load cfgPath :: IO (Either SomeException C.Config)) - -- if no filename provided, start with an empty map to read config from environment - Nothing -> return $ Right M.empty - - pure $ mapLeft ("Error in config: " <>) $ C.runParser parseConfig =<< conf +-- | Reads and parses the config and overrides its parameters from env vars, +-- files or db settings. +readAppConfig :: [(Text, Text)] -> Maybe FilePath -> Maybe Text -> IO (Either Text AppConfig) +readAppConfig dbSettings optPath prevDbUri = do + env <- readPGRSTEnvironment + -- if no filename provided, start with an empty map to read config from environment + conf <- maybe (return $ Right M.empty) loadConfig optPath + case C.runParser (parser optPath env dbSettings) =<< mapLeft show conf of + Left err -> + return . Left $ "Error in config " <> err + Right parsedConfig -> + Right <$> decodeLoadFiles parsedConfig where - parseConfig = - let pB64 = fromMaybe False <$> optWithAlias (optBool "jwt-secret-is-base64") - (optBool "secret-is-base64") - pSec = parseJwtSecret "jwt-secret" =<< pB64 - in - AppConfig - <$> parseAppSettings "app.settings" - <*> reqString "db-anon-role" - <*> (fromMaybe "pgrst" <$> optString "db-channel") - <*> (fromMaybe False <$> optBool "db-channel-enabled") - <*> (maybe ["public"] splitOnCommas <$> optValue "db-extra-search-path") - <*> optWithAlias (optInt "db-max-rows") - (optInt "max-rows") - <*> (fromMaybe 10 <$> optInt "db-pool") - <*> (fromMaybe 10 <$> optInt "db-pool-timeout") - <*> optWithAlias (optString "db-pre-request") - (optString "pre-request") - <*> (fromMaybe True <$> optBool "db-prepared-statements") - <*> optWithAlias (optString "db-root-spec") - (optString "root-spec") - <*> (fromList . splitOnCommas <$> reqWithAlias (optValue "db-schemas") - (optValue "db-schema") - "missing key: either db-schemas or db-schema must be set") - <*> (fromMaybe True <$> optBool "db-config") - <*> parseTxEnd "db-tx-end" snd - <*> parseTxEnd "db-tx-end" fst - <*> parseDbUri "db-uri" - <*> (fmap parseSecret <$> pSec) - <*> parseJwtAudience "jwt-aud" - <*> parseRoleClaimKey "jwt-role-claim-key" "role-claim-key" - <*> pSec - <*> pB64 - <*> parseLogLevel "log-level" - <*> parseOpenAPIServerProxyURI "openapi-server-proxy-uri" - <*> (maybe [] (fmap encodeUtf8 . splitOnCommas) <$> optValue "raw-media-types") - <*> (fromMaybe "!4" <$> optString "server-host") - <*> (fromMaybe 3000 <$> optInt "server-port") - <*> (fmap T.unpack <$> optString "server-unix-socket") - <*> parseSocketFileMode "server-unix-socket-mode" + -- Both C.ParseError and IOError are shown here + loadConfig :: FilePath -> IO (Either SomeException C.Config) + loadConfig = try . C.load - parseDbUri :: C.Key -> C.Parser C.Config Text - parseDbUri k = flip fromMaybe dbUriFile <$> reqString k - - parseJwtSecret :: C.Key -> Bool -> C.Parser C.Config (Maybe B.ByteString) - parseJwtSecret k isB64 = optString k >>= \case - Nothing -> pure Nothing - Just sec -> - let secStr = encodeUtf8 sec - secFile = fromMaybe secStr secretFile - -- replace because the JWT is actually base64url encoded which must be turned into just base64 before decoding. - replaceUrlChars = T.replace "_" "/" . T.replace "-" "+" . T.replace "." "=" - willBeFile = isPrefixOf "@" (toS secStr) && isNothing secretFile - in - if isB64 && not willBeFile -- don't decode in bas64 if the secret will be a file or it will err. The secFile will be filled with the file contents in a later stage. - then case B64.decode . encodeUtf8 . T.strip . replaceUrlChars $ decodeUtf8 secFile of - Left errMsg -> fail errMsg - Right bs -> pure $ Just bs - else pure $ Just secFile + decodeLoadFiles :: AppConfig -> IO AppConfig + decodeLoadFiles parsedConfig = + decodeJWKS <$> + (decodeSecret =<< readSecretFile =<< readDbUriFile prevDbUri parsedConfig) +parser :: Maybe FilePath -> Environment -> [(Text, Text)] -> C.Parser C.Config AppConfig +parser optPath env dbSettings = + AppConfig + <$> parseAppSettings "app.settings" + <*> reqString "db-anon-role" + <*> (fromMaybe "pgrst" <$> optString "db-channel") + <*> (fromMaybe False <$> optBool "db-channel-enabled") + <*> (maybe ["public"] splitOnCommas <$> optValue "db-extra-search-path") + <*> optWithAlias (optInt "db-max-rows") + (optInt "max-rows") + <*> (fromMaybe 10 <$> optInt "db-pool") + <*> (fromIntegral . fromMaybe 10 <$> optInt "db-pool-timeout") + <*> optWithAlias (optString "db-pre-request") + (optString "pre-request") + <*> (fromMaybe True <$> optBool "db-prepared-statements") + <*> optWithAlias (optString "db-root-spec") + (optString "root-spec") + <*> (fromList . splitOnCommas <$> reqWithAlias (optValue "db-schemas") + (optValue "db-schema") + "missing key: either db-schemas or db-schema must be set") + <*> (fromMaybe True <$> optBool "db-config") + <*> parseTxEnd "db-tx-end" snd + <*> parseTxEnd "db-tx-end" fst + <*> reqString "db-uri" + <*> pure optPath + <*> pure Nothing + <*> parseJwtAudience "jwt-aud" + <*> parseRoleClaimKey "jwt-role-claim-key" "role-claim-key" + <*> (fmap encodeUtf8 <$> optString "jwt-secret") + <*> (fromMaybe False <$> optWithAlias + (optBool "jwt-secret-is-base64") + (optBool "secret-is-base64")) + <*> parseLogLevel "log-level" + <*> parseOpenAPIServerProxyURI "openapi-server-proxy-uri" + <*> (maybe [] (fmap encodeUtf8 . splitOnCommas) <$> optValue "raw-media-types") + <*> (fromMaybe "!4" <$> optString "server-host") + <*> (fromMaybe 3000 <$> optInt "server-port") + <*> (fmap T.unpack <$> optString "server-unix-socket") + <*> parseSocketFileMode "server-unix-socket-mode" + where parseAppSettings :: C.Key -> C.Parser C.Config [(Text, Text)] parseAppSettings key = addFromEnv . fmap (fmap coerceText) <$> C.subassocs key C.value where @@ -371,37 +359,65 @@ readAppConfig dbSettings env optPath dbUriFile secretFile = do splitOnCommas (C.String s) = T.strip <$> T.splitOn "," s splitOnCommas _ = [] -{-| - Parse `jwt-secret` configuration option and turn into a JWKSet. +-- | Read the JWT secret from a file if configJwtSecret is actually a +-- filepath(has @ as its prefix). To check if the JWT secret is provided is +-- in fact a file path, it must be decoded as 'Text' to be processed. +readSecretFile :: AppConfig -> IO AppConfig +readSecretFile conf = + maybe (return conf) readSecret maybeFilename + where + maybeFilename = T.stripPrefix "@" . decodeUtf8 =<< configJwtSecret conf + readSecret filename = do + jwtSecret <- chomp <$> BS.readFile (toS filename) + return $ conf { configJwtSecret = Just jwtSecret } + chomp bs = fromMaybe bs (BS.stripSuffix "\n" bs) + +decodeSecret :: AppConfig -> IO AppConfig +decodeSecret conf@AppConfig{..} = + case (configJwtSecretIsBase64, configJwtSecret) of + (True, Just secret) -> + either fail (return . updateSecret) $ decodeB64 secret + _ -> return conf + where + updateSecret bs = conf { configJwtSecret = Just bs } + decodeB64 = B64.decode . encodeUtf8 . T.strip . replaceUrlChars . decodeUtf8 + replaceUrlChars = T.replace "_" "/" . T.replace "-" "+" . T.replace "." "=" + +-- | Parse `jwt-secret` configuration option and turn into a JWKSet. +-- +-- There are three ways to specify `jwt-secret`: text secret, JSON Web Key +-- (JWK), or JSON Web Key Set (JWKS). The first two are converted into a JWKSet +-- with one key and the last is converted as is. +decodeJWKS :: AppConfig -> AppConfig +decodeJWKS conf = + conf { configJWKS = parseSecret <$> configJwtSecret conf } - There are three ways to specify `jwt-secret`: text secret, JSON Web Key - (JWK), or JSON Web Key Set (JWKS). The first two are converted into a JWKSet - with one key and the last is converted as is. --} parseSecret :: ByteString -> JWKSet parseSecret bytes = fromMaybe (maybe secret (\jwk' -> JWT.JWKSet [jwk']) maybeJWK) maybeJWKSet - where - maybeJWKSet = JSON.decode (toS bytes) :: Maybe JWKSet - maybeJWK = JSON.decode (toS bytes) :: Maybe JWK - secret = JWT.JWKSet [JWT.fromKeyMaterial keyMaterial] - keyMaterial = JWT.OctKeyMaterial . JWT.OctKeyParameters $ JOSE.Base64Octets bytes - --- | Read the JWT secret from a file if configJwtSecret is actually a filepath(has @ as its prefix). --- | To check if the JWT secret is provided is in fact a file path, it must be decoded as 'Text' to be processed. -readSecretFile :: Maybe B.ByteString -> IO (Maybe B.ByteString) -readSecretFile mSecret = - case (T.stripPrefix "@" . decodeUtf8) =<< mSecret of - Nothing -> return Nothing - Just filename -> Just . chomp <$> BS.readFile (toS filename) where - chomp bs = fromMaybe bs (BS.stripSuffix "\n" bs) + maybeJWKSet = JSON.decode (toS bytes) :: Maybe JWKSet + maybeJWK = JSON.decode (toS bytes) :: Maybe JWK + secret = JWT.JWKSet [JWT.fromKeyMaterial keyMaterial] + keyMaterial = JWT.OctKeyMaterial . JWT.OctKeyParameters $ JOSE.Base64Octets bytes -- | Read database uri from a separate file if `db-uri` is a filepath. -readDbUriFile :: Text -> IO (Maybe Text) -readDbUriFile dbUri = case T.stripPrefix "@" dbUri of - Nothing -> return Nothing - Just filename -> Just . T.strip <$> readFile (toS filename) +readDbUriFile :: Maybe Text -> AppConfig -> IO AppConfig +readDbUriFile maybeDbUri conf = + case maybeDbUri of + Just prevDbUri -> + pure $ conf { configDbUri = prevDbUri } + Nothing -> + case T.stripPrefix "@" $ configDbUri conf of + Nothing -> return conf + Just filename -> do + dbUri <- T.strip <$> readFile (toS filename) + return $ conf { configDbUri = dbUri } type Environment = M.Map [Char] Text + +-- | Read environment variables that start with PGRST_ +readPGRSTEnvironment :: IO Environment +readPGRSTEnvironment = + M.map T.pack . M.fromList . filter (isPrefixOf "PGRST_" . fst) <$> getEnvironment diff --git a/src/PostgREST/Config/Database.hs b/src/PostgREST/Config/Database.hs new file mode 100644 index 000000000..beecb4f5a --- /dev/null +++ b/src/PostgREST/Config/Database.hs @@ -0,0 +1,55 @@ +{-# LANGUAGE QuasiQuotes #-} + +module PostgREST.Config.Database + ( loadDbSettings + ) where + +import qualified Hasql.Decoders as HD +import qualified Hasql.Encoders as HE +import qualified Hasql.Pool as P +import qualified Hasql.Statement as H +import qualified Hasql.Transaction as HT +import qualified Hasql.Transaction.Sessions as HT + +import Data.Text.IO (hPutStrLn) +import Text.InterpolatedString.Perl6 (q) + +import Protolude hiding (hPutStrLn) + + +loadDbSettings :: P.Pool -> IO [(Text, Text)] +loadDbSettings pool = do + result <- + P.use pool . HT.transaction HT.ReadCommitted HT.Read $ + HT.statement mempty dbSettingsStatement + case result of + Left e -> do + hPutStrLn stderr $ + "An error ocurred when trying to query database settings for the config parameters:\n" + <> show e + pure [] + Right x -> pure x + +-- | Get db settings from the connection role. Global settings will be overridden by database specific settings. +dbSettingsStatement :: H.Statement () [(Text, Text)] +dbSettingsStatement = H.Statement sql HE.noParams decodeSettings False + where + sql = [q| + with + role_setting as ( + select setdatabase, unnest(setconfig) as setting from pg_catalog.pg_db_role_setting + where setrole = current_user::regrole::oid + and setdatabase in (0, (select oid from pg_catalog.pg_database where datname = current_catalog)) + ), + kv_settings as ( + select setdatabase, split_part(setting, '=', 1) as k, split_part(setting, '=', 2) as value from role_setting + where setting like 'pgrst.%' + ) + select distinct on (key) replace(k, 'pgrst.', '') as key, value + from kv_settings + order by key, setdatabase desc; + |] + decodeSettings = HD.rowList $ (,) <$> column HD.text <*> column HD.text + +column :: HD.Value a -> HD.Row a +column = HD.column . HD.nonNullable diff --git a/src/PostgREST/Query/Statements.hs b/src/PostgREST/Query/Statements.hs index 025b01c26..1f01977a0 100644 --- a/src/PostgREST/Query/Statements.hs +++ b/src/PostgREST/Query/Statements.hs @@ -1,4 +1,3 @@ -{-# LANGUAGE QuasiQuotes #-} {-| Module : PostgREST.Query.Statements Description : PostgREST single SQL statements. @@ -15,7 +14,6 @@ module PostgREST.Query.Statements , createReadStatement , callProcStatement , createExplainStatement - , dbSettingsStatement ) where import qualified Data.Aeson as JSON @@ -24,14 +22,12 @@ import qualified Data.ByteString.Char8 as BS import qualified Hasql.Decoders as HD import qualified Hasql.DynamicStatements.Snippet as H import qualified Hasql.DynamicStatements.Statement as H -import qualified Hasql.Encoders as HE import qualified Hasql.Statement as H -import Control.Lens ((^?)) -import Data.Maybe (fromJust) -import Data.Text.Read (decimal) -import Network.HTTP.Types.Status (Status) -import Text.InterpolatedString.Perl6 (q) +import Control.Lens ((^?)) +import Data.Maybe (fromJust) +import Data.Text.Read (decimal) +import Network.HTTP.Types.Status (Status) import PostgREST.DbStructure.PgVersion (PgVersion) import PostgREST.Error (Error (..)) @@ -44,7 +40,6 @@ import PostgREST.Request.Preferences import Protolude hiding (toS) import Protolude.Conv (toS) - {-| The generic query result format used by API responses. The location header is represented as a list of strings containing variable bindings like @"k1=eq.42"@, or the empty list if there is no location header. @@ -199,27 +194,6 @@ decodeGucHeaders = first (const GucHeadersError) . JSON.eitherDecode . toS <$> H decodeGucStatus :: HD.Value (Either Error (Maybe Status)) decodeGucStatus = first (const GucStatusError) . fmap (Just . toEnum . fst) . decimal <$> HD.text --- | Get db settings from the connection role. Global settings will be overridden by database specific settings. -dbSettingsStatement :: H.Statement () [(Text, Text)] -dbSettingsStatement = H.Statement sql HE.noParams decodeSettings False - where - sql = [q| - with - role_setting as ( - select setdatabase, unnest(setconfig) as setting from pg_catalog.pg_db_role_setting - where setrole = current_user::regrole::oid - and setdatabase in (0, (select oid from pg_catalog.pg_database where datname = current_catalog)) - ), - kv_settings as ( - select setdatabase, split_part(setting, '=', 1) as k, split_part(setting, '=', 2) as value from role_setting - where setting like 'pgrst.%' - ) - select distinct on (key) replace(k, 'pgrst.', '') as key, value - from kv_settings - order by key, setdatabase desc; - |] - decodeSettings = HD.rowList $ (,) <$> column HD.text <*> column HD.text - column :: HD.Value a -> HD.Row a column = HD.column . HD.nonNullable diff --git a/src/PostgREST/Unix.hs b/src/PostgREST/Unix.hs new file mode 100644 index 000000000..9c742541d --- /dev/null +++ b/src/PostgREST/Unix.hs @@ -0,0 +1,61 @@ +module PostgREST.Unix + ( runAppInSocket + , installSignalHandlers + ) where + +import qualified Network.Socket as Socket +import qualified Network.Wai.Handler.Warp as Warp +import qualified System.Posix.Signals as Signals + +import Network.Wai (Application) +import System.Directory (removeFile) +import System.IO.Error (isDoesNotExistError) +import System.Posix.Files (setFileMode) +import System.Posix.Types (FileMode) + +import qualified PostgREST.AppState as AppState +import qualified PostgREST.Workers as Workers + +import Protolude + + +-- | Run the PostgREST application with user defined socket. +runAppInSocket :: Warp.Settings -> Application -> FileMode -> FilePath -> IO () +runAppInSocket settings app socketFileMode socketFilePath = do + sock <- createAndBindSocket + putStrLn $ ("Listening on unix socket " :: Text) <> show socketFilePath + Socket.listen sock Socket.maxListenQueue + Warp.runSettingsSocket settings sock app + Socket.close sock + where + createAndBindSocket = do + deleteSocketFileIfExist socketFilePath + sock <- Socket.socket Socket.AF_UNIX Socket.Stream Socket.defaultProtocol + Socket.bind sock $ Socket.SockAddrUnix socketFilePath + setFileMode socketFilePath socketFileMode + return sock + + deleteSocketFileIfExist path = + removeFile path `catch` handleDoesNotExist + + handleDoesNotExist e + | isDoesNotExistError e = return () + | otherwise = throwIO e + +-- | Set signal handlers, only for systems with signals +installSignalHandlers :: AppState.AppState -> IO () +installSignalHandlers appState = do + -- Releases the connection pool whenever the program is terminated, + -- see https://github.com/PostgREST/postgrest/issues/268 + install Signals.sigINT $ AppState.releasePool appState + install Signals.sigTERM $ AppState.releasePool appState + + -- The SIGUSR1 signal updates the internal 'DbStructure' by running + -- 'connectionWorker' exactly as before. + install Signals.sigUSR1 $ Workers.connectionWorker appState + + -- Re-read the config on SIGUSR2 + install Signals.sigUSR2 $ Workers.reReadConfig False appState + where + install signal handler = + void $ Signals.installHandler signal (Signals.Catch handler) Nothing diff --git a/src/PostgREST/Workers.hs b/src/PostgREST/Workers.hs new file mode 100644 index 000000000..2349a589b --- /dev/null +++ b/src/PostgREST/Workers.hs @@ -0,0 +1,243 @@ +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE RecordWildCards #-} + +module PostgREST.Workers + ( connectionWorker + , reReadConfig + , listener + ) where + +import qualified Data.ByteString as BS +import qualified Hasql.Connection as C +import qualified Hasql.Notifications as N +import qualified Hasql.Pool as P +import qualified Hasql.Transaction.Sessions as HT + +import Control.Retry (RetryStatus, capDelay, exponentialBackoff, + retrying, rsPreviousDelay) +import Data.Text.IO (hPutStrLn) + +import PostgREST.AppState (AppState) +import PostgREST.Config (AppConfig (..), readAppConfig) +import PostgREST.Config.Database (loadDbSettings) +import PostgREST.DbStructure (getDbStructure, getPgVersion) +import PostgREST.DbStructure.PgVersion (PgVersion (..), + minimumPgVersion) +import PostgREST.Error (PgError (PgError), + checkIsFatal, errorPayload) + +import qualified PostgREST.AppState as AppState + +import Protolude hiding (hPutStrLn, head, toS) +import Protolude.Conv (toS) + + +-- | Current database connection status data ConnectionStatus +data ConnectionStatus + = NotConnected + | Connected PgVersion + | FatalConnectionError Text + deriving (Eq) + +-- | Schema cache status +data SCacheStatus + = SCLoaded + | SCOnRetry + | SCFatalFail + +-- | The purpose of this worker is to obtain a healthy connection to pg and an +-- up-to-date schema cache(DbStructure). This method is meant to be called +-- multiple times by the same thread, but does nothing if the previous +-- invocation has not terminated. In all cases this method does not halt the +-- calling thread, the work is preformed in a separate thread. +-- +-- Background thread that does the following : +-- 1. Tries to connect to pg server and will keep trying until success. +-- 2. Checks if the pg version is supported and if it's not it kills the main +-- program. +-- 3. Obtains the dbStructure. If this fails, it goes back to 1. +connectionWorker :: AppState -> IO () +connectionWorker appState = do + isWorkerOn <- AppState.getIsWorkerOn appState + -- Prevents multiple workers to be running at the same time. Could happen on + -- too many SIGUSR1s. + unless isWorkerOn $ do + AppState.putIsWorkerOn appState True + void $ forkIO work + where + work = do + AppConfig{..} <- AppState.getConfig appState + putStrLn ("Attempting to connect to the database..." :: Text) + connected <- connectionStatus $ AppState.getPool appState + case connected of + FatalConnectionError reason -> + -- Fatal error when connecting + hPutStrLn stderr reason >> killThread (AppState.getMainThreadId appState) + NotConnected -> + -- Unreachable because connectionStatus will keep trying to connect + return () + Connected actualPgVersion -> do + when configDbChannelEnabled $ + -- tryPutMVar doesn't lock the thread. It should always succeed since + -- the worker is the only mvar producer. + AppState.putPgVersion appState actualPgVersion + -- Procede with initialization + putStrLn ("Connection successful" :: Text) + -- this could be fail because the connection drops, but the + -- loadSchemaCache will pick the error and retry again + when configDbConfig $ reReadConfig False appState + scStatus <- loadSchemaCache appState actualPgVersion + case scStatus of + SCLoaded -> + -- do nothing and proceed if the load was successful + return () + SCOnRetry -> + work + SCFatalFail -> + -- die if our schema cache query has an error + killThread $ AppState.getMainThreadId appState + AppState.putIsWorkerOn appState False + +-- | Check if a connection from the pool allows access to the PostgreSQL +-- database. If not, the pool connections are released and a new connection is +-- tried. Releasing the pool is key for rapid recovery. Otherwise, the pool +-- timeout would have to be reached for new healthy connections to be acquired. +-- Which might not happen if the server is busy with requests. No idle +-- connection, no pool timeout. +-- +-- The connection tries are capped, but if the connection times out no error is +-- thrown, just 'False' is returned. +connectionStatus :: P.Pool -> IO ConnectionStatus +connectionStatus pool = + retrying retrySettings shouldRetry $ + const $ P.release pool >> getConnectionStatus + where + retrySettings = capDelay delayMicroseconds $ exponentialBackoff backoffMicroseconds + delayMicroseconds = 32000000 -- 32 seconds + backoffMicroseconds = 1000000 -- 1 second + + getConnectionStatus :: IO ConnectionStatus + getConnectionStatus = do + pgVersion <- P.use pool getPgVersion + case pgVersion of + Left e -> do + let err = PgError False e + hPutStrLn stderr . toS $ errorPayload err + case checkIsFatal err of + Just reason -> + return $ FatalConnectionError reason + Nothing -> + return NotConnected + Right version -> + if version < minimumPgVersion then + return . FatalConnectionError $ + "Cannot run in this PostgreSQL version, PostgREST needs at least " + <> pgvName minimumPgVersion + else + return . Connected $ version + + shouldRetry :: RetryStatus -> ConnectionStatus -> IO Bool + shouldRetry rs isConnSucc = do + let + delay = fromMaybe 0 (rsPreviousDelay rs) `div` backoffMicroseconds + itShould = NotConnected == isConnSucc + when itShould . putStrLn $ + "Attempting to reconnect to the database in " + <> (show delay::Text) + <> " seconds..." + return itShould + +-- | Load the DbStructure by using a connection from the pool. +loadSchemaCache :: AppState -> PgVersion -> IO SCacheStatus +loadSchemaCache appState actualPgVersion = do + AppConfig{..} <- AppState.getConfig appState + result <- + P.use (AppState.getPool appState) . HT.transaction HT.ReadCommitted HT.Read $ + getDbStructure (toList configDbSchemas) configDbExtraSearchPath actualPgVersion configDbPreparedStatements + case result of + Left e -> do + let + err = PgError False e + putErr = hPutStrLn stderr . toS . errorPayload $ err + case checkIsFatal err of + Just _ -> do + hPutStrLn stderr "A fatal error ocurred when loading the schema cache" + putErr + hPutStrLn stderr $ + "This is probably a bug in PostgREST, please report it at " + <> "https://github.com/PostgREST/postgrest/issues" + return SCFatalFail + Nothing -> do + hPutStrLn stderr "An error ocurred when loading the schema cache" + putErr + return SCOnRetry + + Right dbStructure -> do + AppState.putDbStructure appState dbStructure + putStrLn ("Schema cache loaded" :: Text) + return SCLoaded + +-- | Starts a dedicated pg connection to LISTEN for notifications. When a +-- NOTIFY - with an empty payload - is done, it refills the schema +-- cache. It uses the connectionWorker in case the LISTEN connection dies. +listener :: AppState -> IO () +listener appState = do + AppConfig{..} <- AppState.getConfig appState + let dbChannel = toS configDbChannel + + -- AppState.getPgVersion makes the thread wait until the pgVersion has been + -- set by the connectionWorker + actualPgVersion <- AppState.getPgVersion appState + -- forkFinally allows to detect if the thread dies + void . flip forkFinally (handleFinally dbChannel) $ do + dbOrError <- C.acquire $ toS configDbUri + case dbOrError of + Right db -> do + putStrLn $ "Listening for notifications on the " <> dbChannel <> " channel" + N.listen db $ N.toPgIdentifier dbChannel + N.waitForNotifications (handleNotification actualPgVersion) db + _ -> + die $ "Could not listen for notifications on the " <> dbChannel <> " channel" + where + handleFinally dbChannel _ = do + -- if the thread dies, we try to recover + putStrLn $ "Retrying listening for notifications on the " <> dbChannel <> " channel.." + -- assume the pool connection was also lost, call the connection worker + connectionWorker appState + + -- retry the listener + listener appState + + handleNotification actualPgVersion _ msg + | BS.null msg = scLoader actualPgVersion -- reload the schema cache + | msg == "reload schema" = scLoader actualPgVersion -- reload the schema cache + | msg == "reload config" = reReadConfig False appState -- reload the config + | otherwise = pure () -- Do nothing if anything else than an empty message is sent + + scLoader actualPgVersion = + -- It's not necessary to check the loadSchemaCache success + -- here. If the connection drops, the thread will die and + -- proceed to recover below. + void $ loadSchemaCache appState actualPgVersion + +-- | Re-reads the config plus config options from the db +reReadConfig :: Bool -> AppState -> IO () +reReadConfig startingUp appState = do + AppConfig{..} <- AppState.getConfig appState + dbSettings <- + if configDbConfig then + loadDbSettings (AppState.getPool appState) + else + pure mempty + readAppConfig dbSettings configFilePath (Just configDbUri) >>= \case + Left err -> + if startingUp then + panic err -- die on invalid config if the program is starting up + else + hPutStrLn stderr $ "Failed loading in-database config. " <> err + Right newConf -> do + AppState.putConfig appState newConf + if startingUp then + pass + else + putStrLn ("In-database config loaded" :: Text) diff --git a/test/Main.hs b/test/Main.hs index 6e498e786..211ed47a5 100644 --- a/test/Main.hs +++ b/test/Main.hs @@ -3,13 +3,9 @@ module Main where import qualified Hasql.Pool as P import qualified Hasql.Transaction.Sessions as HT -import Control.AutoUpdate (defaultUpdateSettings, mkAutoUpdate, - updateAction) import Data.Function (id) import Data.List.NonEmpty (toList) -import Data.Time.Clock (getCurrentTime) -import Data.IORef import Test.Hspec import PostgREST.App (postgrest) @@ -20,6 +16,8 @@ import Protolude hiding (toList, toS) import Protolude.Conv (toS) import SpecHelper +import qualified PostgREST.AppState as AppState + import qualified Feature.AndOrParamsSpec import qualified Feature.AsymmetricJwtSpec import qualified Feature.AudienceJwtSecretSpec @@ -55,27 +53,35 @@ import qualified Feature.UpsertSpec main :: IO () main = do - getTime <- mkAutoUpdate defaultUpdateSettings { updateAction = getCurrentTime } - testDbConn <- getEnvVarWithDefault "PGRST_DB_URI" "postgres://postgrest_test@localhost/postgrest_test" pool <- P.acquire (3, 10, toS testDbConn) actualPgVersion <- either (panic.show) id <$> P.use pool getPgVersion - refDbStructure <- (newIORef . Just) =<< setupDbStructure pool (configDbSchemas $ testCfg testDbConn) (configDbExtraSearchPath $ testCfg testDbConn) actualPgVersion + baseDbStructure <- + loadDbStructure pool + (configDbSchemas $ testCfg testDbConn) + (configDbExtraSearchPath $ testCfg testDbConn) + actualPgVersion let -- For tests that run with the same refDbStructure app cfg = do - refConf <- newIORef $ cfg testDbConn - return ((), postgrest LogCrit refConf refDbStructure pool getTime $ pure ()) + appState <- AppState.initWithPool pool $ cfg testDbConn + AppState.putDbStructure appState baseDbStructure + return ((), postgrest LogCrit appState $ pure ()) -- For tests that run with a different DbStructure(depends on configSchemas) appDbs cfg = do - dbs <- (newIORef . Just) =<< setupDbStructure pool (configDbSchemas $ cfg testDbConn) (configDbExtraSearchPath $ cfg testDbConn) actualPgVersion - refConf <- newIORef $ cfg testDbConn - return ((), postgrest LogCrit refConf dbs pool getTime $ pure ()) + customDbStructure <- + loadDbStructure pool + (configDbSchemas $ cfg testDbConn) + (configDbExtraSearchPath $ cfg testDbConn) + actualPgVersion + appState <- AppState.initWithPool pool $ cfg testDbConn + AppState.putDbStructure appState customDbStructure + return ((), postgrest LogCrit appState $ pure ()) let withApp = app testCfg maxRowsApp = app testMaxRowsCfg @@ -198,5 +204,5 @@ main = do describe "Feature.RollbackForcedSpec" Feature.RollbackSpec.forced where - setupDbStructure pool schemas extraSearchPath ver = + loadDbStructure pool schemas extraSearchPath ver = either (panic.show) id <$> P.use pool (HT.transaction HT.ReadCommitted HT.Read $ getDbStructure (toList schemas) extraSearchPath ver True) diff --git a/test/SpecHelper.hs b/test/SpecHelper.hs index b2bfec2bc..fbbe53f4b 100644 --- a/test/SpecHelper.hs +++ b/test/SpecHelper.hs @@ -85,6 +85,7 @@ _baseCfg = let secret = Just $ encodeUtf8 "reallyreallyreallyreallyverysafe" in , configDbSchemas = fromList ["test"] , configDbConfig = False , configDbUri = mempty + , configFilePath = Nothing , configJWKS = parseSecret <$> secret , configJwtAudience = Nothing , configJwtRoleClaimKey = [JSPKey "role"]