diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,20 @@ +Copyright (c) 2014 Joe Nelson + +Permission is hereby granted, free of charge, to any person obtaining +a copy of this software and associated documentation files (the +"Software"), to deal in the Software without restriction, including +without limitation the rights to use, copy, modify, merge, publish, +distribute, sublicense, and/or sell copies of the Software, and to +permit persons to whom the Software is furnished to do so, subject to +the following conditions: + +The above copyright notice and this permission notice shall be included +in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. +IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, +TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE +SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/app/Main.hs b/app/Main.hs new file mode 100644 --- /dev/null +++ b/app/Main.hs @@ -0,0 +1,89 @@ +{-# LANGUAGE CPP #-} + +module Main where + + +import PostgREST.Config (AppConfig (..), + minimumPgVersion, + prettyVersion, + readOptions) +import PostgREST.DbStructure +import PostgRESTWS + +import Control.Monad +import Control.Monad.IO.Class (liftIO) +import Data.Monoid ((<>)) +import Data.String.Conversions (cs) +import qualified Hasql.Query as H +import qualified Hasql.Session as H +import qualified Hasql.Decoders as HD +import qualified Hasql.Encoders as HE +import qualified Hasql.Pool as P +import Network.Wai.Handler.Warp +import System.IO (BufferMode (..), + hSetBuffering, stderr, + stdin, stdout) +import Web.JWT (secret) + +#ifndef mingw32_HOST_OS +import System.Posix.Signals +import Control.Concurrent (myThreadId) +import Data.IORef +import Control.Exception.Base (throwTo, AsyncException(..)) +#endif +import qualified Database.PostgreSQL.LibPQ as PQ + +isServerVersionSupported :: H.Session Bool +isServerVersionSupported = do + ver <- H.query () pgVersion + return $ read (cs ver) >= minimumPgVersion + where + pgVersion = + H.statement "SHOW server_version_num" + HE.unit (HD.singleRow $ HD.value HD.text) True + +main :: IO () +main = do + hSetBuffering stdout LineBuffering + hSetBuffering stdin LineBuffering + hSetBuffering stderr NoBuffering + + conf <- readOptions + let port = configPort conf + pgSettings = cs (configDatabase conf) + appSettings = setPort port + . setServerName (cs $ "postgrest/" <> prettyVersion) + $ defaultSettings + + unless (secret "secret" /= configJwtSecret conf) $ + putStrLn "WARNING, running in insecure mode, JWT secret is the default value" + Prelude.putStrLn $ "Listening on port " ++ + (show $ configPort conf :: String) + + pool <- P.acquire (configPool conf, 10, pgSettings) + + result <- P.use pool $ do + supported <- isServerVersionSupported + unless supported $ error ( + "Cannot run in this PostgreSQL version, PostgREST needs at least " + <> show minimumPgVersion) + getDbStructure (cs $ configSchema conf) + + refDbStructure <- newIORef $ either (error.show) id result + notificationsCon <- PQ.connectdb pgSettings + +#ifndef mingw32_HOST_OS + tid <- myThreadId + forM_ [sigINT, sigTERM] $ \sig -> + void $ installHandler sig (Catch $ do + P.release pool + throwTo tid UserInterrupt + ) Nothing + + void $ installHandler sigHUP ( + Catch . void . P.use pool $ do + s <- getDbStructure (cs $ configSchema conf) + liftIO $ atomicWriteIORef refDbStructure s + ) Nothing +#endif + runSettings appSettings $ postgrestWsApp conf refDbStructure pool notificationsCon diff --git a/postgrest-ws.cabal b/postgrest-ws.cabal new file mode 100644 --- /dev/null +++ b/postgrest-ws.cabal @@ -0,0 +1,67 @@ +name: postgrest-ws +version: 0.1.0.0 +synopsis: Initial project template from stack +description: Please see README.md +homepage: https://github.com/diogob/postgrest-ws#readme +license: BSD3 +license-file: LICENSE +author: Author name here +maintainer: example@example.com +copyright: 2016 Author name here +category: Web +build-type: Simple +-- extra-source-files: +cabal-version: >=1.10 + +library + hs-source-dirs: src + ghc-options: -Wall + exposed-modules: PostgRESTWS + build-depends: base >= 4.7 && < 5 + , hasql-pool >= 0.4 && < 0.5 + , text >= 1.2 && < 2 + , wai >= 3.2 && < 4 + , websockets >= 0.9 && < 0.10 + , wai-websockets >= 3.0 && < 4 + , postgrest >= 0.3 && < 0.4 + , http-types >= 0.9 + , bytestring >= 0.10 + , postgresql-libpq + , time + , unordered-containers >= 0.2 + , postgresql-libpq >= 0.9 && < 1.0 + , aeson >= 0.11 + , string-conversions >= 0.4 + default-language: Haskell2010 + default-extensions: OverloadedStrings + +executable postgrest-ws + hs-source-dirs: app + main-is: Main.hs + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N + build-depends: base >= 4.7 && < 5 + , transformers >= 0.4 && < 0.5 + , string-conversions >= 0.4 && < 0.5 + , hasql >= 0.19 && < 0.20 + , hasql-pool >= 0.4 && < 0.5 + , warp >= 3.2 && < 4 + , unix >= 2.7 && < 3 + , jwt >= 0.7 && < 1 + , postgrest + , postgrest-ws + , postgresql-libpq >= 0.9 && < 1.0 + default-language: Haskell2010 + default-extensions: OverloadedStrings + +test-suite postgrest-ws-test + type: exitcode-stdio-1.0 + hs-source-dirs: test + main-is: Spec.hs + build-depends: base + , postgrest-ws + ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N + default-language: Haskell2010 + +source-repository head + type: git + location: https://github.com/diogob/postgrest-ws diff --git a/src/PostgRESTWS.hs b/src/PostgRESTWS.hs new file mode 100644 --- /dev/null +++ b/src/PostgRESTWS.hs @@ -0,0 +1,121 @@ +{-# LANGUAGE DeriveGeneric #-} + +module PostgRESTWS + ( postgrestWsApp + ) where + +import GHC.IORef +import qualified Hasql.Pool as H +import qualified Network.Wai as Wai +import qualified Network.Wai.Handler.WebSockets as WS +import qualified Network.WebSockets as WS +import PostgREST.App as PGR +import PostgREST.Config as PGR +import PostgREST.Types as PGR + +import qualified Data.Text as T +import qualified Data.Text.Encoding as T + +import Control.Monad (forever, void, when) +import qualified Data.HashMap.Strict as M +import Data.String.Conversions (cs) +import Data.Time.Clock.POSIX (getPOSIXTime) +import qualified Database.PostgreSQL.LibPQ as PQ +import PostgREST.Auth (jwtClaims) + +import qualified Data.Aeson as A +import qualified Data.ByteString as BS +import Data.ByteString.Lazy (toStrict) +import Data.Monoid + +import Control.Concurrent (forkIO, threadWaitReadSTM) +import GHC.Conc (atomically) + +import GHC.Generics + +data Message = Message + { userClaims :: A.Object + , payload :: T.Text + } deriving (Show, Eq, Generic) + +instance A.ToJSON Message + +postgrestWsApp :: PGR.AppConfig + -> IORef PGR.DbStructure + -> H.Pool + -> PQ.Connection + -> Wai.Application +postgrestWsApp conf refDbStructure pool pqCon = + WS.websocketsOr WS.defaultConnectionOptions wsApp $ postgrest conf refDbStructure pool + where + -- when the websocket is closed a ConnectionClosed Exception is triggered + -- this kills all children and frees resources for us + wsApp :: WS.ServerApp + wsApp pendingConn = do + time <- getPOSIXTime + let + claimsOrExpired = jwtClaims jwtSecret jwtToken time + case claimsOrExpired of + Left e -> rejectRequest e + Right claims -> do + -- role claim defaults to anon if not specified in jwt + -- We should accept only after verifying JWT + conn <- WS.acceptRequest pendingConn + -- each websocket needs its own listen connection to avoid + -- handling of multiple waiting threads in the same connection + when (hasRead claims) $ + void $ forkIO $ listenSession (channel claims) conf conn + -- all websockets share a single connection to NOTIFY + when (hasWrite claims) $ + forever $ notifySession (channel claims) claims pqCon conn + where + claimAsBS name cl = let A.String s = (cl M.! name) in T.encodeUtf8 s + channel = claimAsBS ("channel" :: T.Text) + mode = claimAsBS ("mode" :: T.Text) + hasRead cl = mode cl == "r" || mode cl == "rw" + hasWrite cl = mode cl == "w" || mode cl == "rw" + rejectRequest = WS.rejectRequest pendingConn . T.encodeUtf8 + jwtSecret = configJwtSecret conf + -- the first char in path is '/' the rest is the token + jwtToken = T.decodeUtf8 $ BS.drop 1 $ WS.requestPath $ WS.pendingRequest pendingConn + +-- private functions +-- Having both channel and claims as parameters seem redundant +-- But it allows the function to ignore the claims structure and the source +-- of the channel +notifySession :: BS.ByteString + -> A.Object + -> PQ.Connection + -> WS.Connection + -> IO () +notifySession channel claims pqCon wsCon = + WS.receiveData wsCon >>= (notify . jsonMsg) + where + notify msg = void $ PQ.exec pqCon ("NOTIFY " <> channel <> ", '" <> msg <> "'") + jsonMsg = toStrict . A.encode . Message claims . T.decodeUtf8 + +listenSession :: BS.ByteString + -> PGR.AppConfig + -> WS.Connection + -> IO () +listenSession channel conf wsCon = do + pqCon <- PQ.connectdb pgSettings + listen pqCon + waitForNotifications pqCon + where + waitForNotifications = forever . fetch + listen con = void $ PQ.exec con $ "LISTEN " <> channel + pgSettings = cs $ configDatabase conf + fetch con = do + mNotification <- PQ.notifies con + case mNotification of + Nothing -> do + mfd <- PQ.socket con + case mfd of + Nothing -> error "Error checking for PostgreSQL notifications" + Just fd -> do + (waitRead, _) <- threadWaitReadSTM fd + atomically waitRead + void $ PQ.consumeInput con + Just notification -> + WS.sendTextData wsCon $ PQ.notifyExtra notification diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,2 @@ +main :: IO () +main = putStrLn "Test suite not yet implemented"