From aaf77902f6fee453ffb1a6afc378b22ce48f650a Mon Sep 17 00:00:00 2001 From: steve-chavez Date: Sat, 15 Apr 2023 14:19:24 -0500 Subject: [PATCH] feat: isolation level for roles/functions --- CHANGELOG.md | 3 ++ src/PostgREST/App.hs | 56 ++++++++++++--------- src/PostgREST/Config/Database.hs | 8 +-- src/PostgREST/Query.hs | 8 ++- src/PostgREST/SchemaCache.hs | 5 +- src/PostgREST/SchemaCache/Routine.hs | 5 +- test/io/fixtures.sql | 36 +++++++++++++- test/io/test_io.py | 73 ++++++++++++++++++++++++++++ 8 files changed, 158 insertions(+), 36 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9f52af408..cfdd1ee68 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,6 +37,9 @@ This project adheres to [Semantic Versioning](http://semver.org/). `ALTER ROLE anon SET statement_timeout TO '5s'` will result in that `statement_timeout` getting applied for that role. - Works when switching roles when a JWT is sent - Settings can be reloaded with `NOTIFY pgrst, 'reload config'`. + - #2468, Configurable transaction isolation level with `default_transaction_isolation` - @steve-chavez + - Can be set per function `create function .. set default_transaction_isolation = 'repeatable read'` + - Or per role `alter role .. set default_transaction_isolation = 'serializable'` ### Fixed diff --git a/src/PostgREST/App.hs b/src/PostgREST/App.hs index c48dd0330..f44192d73 100644 --- a/src/PostgREST/App.hs +++ b/src/PostgREST/App.hs @@ -27,6 +27,7 @@ import Network.Wai.Handler.Warp (defaultSettings, setHost, setPort, setServerName) import System.Posix.Types (FileMode) +import qualified Data.HashMap.Strict as HM import qualified Hasql.Pool as SQL import qualified Hasql.Transaction.Sessions as SQL import qualified Network.Wai as Wai @@ -44,16 +45,17 @@ import qualified PostgREST.Query as Query import qualified PostgREST.Response as Response import qualified PostgREST.Workers as Workers -import PostgREST.ApiRequest (Action (..), ApiRequest (..), - Mutation (..), Target (..)) -import PostgREST.AppState (AppState) -import PostgREST.Auth (AuthResult (..)) -import PostgREST.Config (AppConfig (..)) -import PostgREST.Config.PgVersion (PgVersion (..)) -import PostgREST.Error (Error) -import PostgREST.Query (DbHandler) -import PostgREST.SchemaCache (SchemaCache (..)) -import PostgREST.Version (prettyVersion) +import PostgREST.ApiRequest (Action (..), ApiRequest (..), + Mutation (..), Target (..)) +import PostgREST.AppState (AppState) +import PostgREST.Auth (AuthResult (..)) +import PostgREST.Config (AppConfig (..)) +import PostgREST.Config.PgVersion (PgVersion (..)) +import PostgREST.Error (Error) +import PostgREST.Query (DbHandler) +import PostgREST.SchemaCache (SchemaCache (..)) +import PostgREST.SchemaCache.Routine (Routine (..)) +import PostgREST.Version (prettyVersion) import Protolude hiding (Handler) @@ -152,11 +154,11 @@ postgrestResponse appState conf@AppConfig{..} maybeSchemaCache pgVer authResult@ Response.optionalRollback conf apiRequest $ handleRequest authResult conf appState (Just authRole /= configDbAnonRole) configDbPreparedStatements pgVer apiRequest sCache -runDbHandler :: AppState.AppState -> SQL.Mode -> Bool -> Bool -> DbHandler b -> Handler IO b -runDbHandler appState mode authenticated prepared handler = do +runDbHandler :: AppState.AppState -> Maybe Text -> SQL.Mode -> Bool -> Bool -> DbHandler b -> Handler IO b +runDbHandler appState isoLvl mode authenticated prepared handler = do dbResp <- lift $ do let transaction = if prepared then SQL.transaction else SQL.unpreparedTransaction - res <- AppState.usePool appState . transaction SQL.ReadCommitted mode $ runExceptT handler + res <- AppState.usePool appState . transaction (toIsolationLevel isoLvl) mode $ runExceptT handler whenLeft res (\case SQL.AcquisitionTimeoutUsageError -> AppState.debounceLogAcquisitionTimeout appState -- this can happen rapidly for many requests, so we debounce _ -> pure ()) @@ -167,42 +169,48 @@ runDbHandler appState mode authenticated prepared handler = do mapLeft (Error.PgError authenticated) dbResp liftEither resp + where + toIsolationLevel = \case + Nothing -> SQL.ReadCommitted + Just "repeatable read" -> SQL.RepeatableRead + Just "serializable" -> SQL.Serializable + _ -> SQL.ReadCommitted handleRequest :: AuthResult -> AppConfig -> AppState.AppState -> Bool -> Bool -> PgVersion -> ApiRequest -> SchemaCache -> Handler IO Wai.Response handleRequest AuthResult{..} conf appState authenticated prepared pgVer apiReq@ApiRequest{..} sCache = case (iAction, iTarget) of (ActionRead headersOnly, TargetIdent identifier) -> do wrPlan <- liftEither $ Plan.wrappedReadPlan identifier conf sCache apiReq - resultSet <- runQuery (Plan.wrTxMode wrPlan) $ Query.readQuery wrPlan conf apiReq + resultSet <- runQuery roleIsoLvl (Plan.wrTxMode wrPlan) $ Query.readQuery wrPlan conf apiReq return $ Response.readResponse headersOnly identifier apiReq resultSet (ActionMutate MutationCreate, TargetIdent identifier) -> do mrPlan <- liftEither $ Plan.mutateReadPlan MutationCreate apiReq identifier conf sCache - resultSet <- runQuery (Plan.mrTxMode mrPlan) $ Query.createQuery mrPlan apiReq conf + resultSet <- runQuery roleIsoLvl (Plan.mrTxMode mrPlan) $ Query.createQuery mrPlan apiReq conf return $ Response.createResponse identifier mrPlan apiReq resultSet (ActionMutate MutationUpdate, TargetIdent identifier) -> do mrPlan <- liftEither $ Plan.mutateReadPlan MutationUpdate apiReq identifier conf sCache - resultSet <- runQuery (Plan.mrTxMode mrPlan) $ Query.updateQuery mrPlan apiReq conf + resultSet <- runQuery roleIsoLvl (Plan.mrTxMode mrPlan) $ Query.updateQuery mrPlan apiReq conf return $ Response.updateResponse apiReq resultSet (ActionMutate MutationSingleUpsert, TargetIdent identifier) -> do mrPlan <- liftEither $ Plan.mutateReadPlan MutationSingleUpsert apiReq identifier conf sCache - resultSet <- runQuery (Plan.mrTxMode mrPlan) $ Query.singleUpsertQuery mrPlan apiReq conf + resultSet <- runQuery roleIsoLvl (Plan.mrTxMode mrPlan) $ Query.singleUpsertQuery mrPlan apiReq conf return $ Response.singleUpsertResponse apiReq resultSet (ActionMutate MutationDelete, TargetIdent identifier) -> do mrPlan <- liftEither $ Plan.mutateReadPlan MutationDelete apiReq identifier conf sCache - resultSet <- runQuery (Plan.mrTxMode mrPlan) $ Query.deleteQuery mrPlan apiReq conf + resultSet <- runQuery roleIsoLvl (Plan.mrTxMode mrPlan) $ Query.deleteQuery mrPlan apiReq conf return $ Response.deleteResponse apiReq resultSet (ActionInvoke invMethod, TargetProc identifier _) -> do cPlan <- liftEither $ Plan.callReadPlan identifier conf sCache apiReq invMethod - resultSet <- runQuery (Plan.crTxMode cPlan) $ Query.invokeQuery (Plan.crProc cPlan) cPlan apiReq conf pgVer + resultSet <- runQuery (roleIsoLvl <|> pdIsoLvl (Plan.crProc cPlan))(Plan.crTxMode cPlan) $ Query.invokeQuery (Plan.crProc cPlan) cPlan apiReq conf pgVer return $ Response.invokeResponse invMethod (Plan.crProc cPlan) apiReq resultSet (ActionInspect headersOnly, TargetDefaultSpec tSchema) -> do - oaiResult <- runQuery Plan.inspectPlanTxMode $ Query.openApiQuery sCache pgVer conf tSchema + oaiResult <- runQuery roleIsoLvl Plan.inspectPlanTxMode $ Query.openApiQuery sCache pgVer conf tSchema return $ Response.openApiResponse headersOnly oaiResult conf sCache iSchema iNegotiatedByProfile (ActionInfo, TargetIdent identifier) -> @@ -220,8 +228,10 @@ handleRequest AuthResult{..} conf appState authenticated prepared pgVer apiReq@A -- TODO Refactor the Action/Target types to remove this line throwError $ Error.ApiRequestError ApiRequestTypes.NotFound where - runQuery mode query = - runDbHandler appState mode authenticated prepared $ do - Query.setPgLocals conf authClaims authRole apiReq pgVer + roleSettings = fromMaybe mempty (HM.lookup authRole $ configRoleSettings conf) + roleIsoLvl = decodeUtf8 <$> HM.lookup "default_transaction_isolation" roleSettings + runQuery isoLvl mode query = + runDbHandler appState isoLvl mode authenticated prepared $ do + Query.setPgLocals conf authClaims authRole (HM.toList roleSettings) apiReq pgVer Query.runPreReq conf query diff --git a/src/PostgREST/Config/Database.hs b/src/PostgREST/Config/Database.hs index 8a85c5743..9f64f1d70 100644 --- a/src/PostgREST/Config/Database.hs +++ b/src/PostgREST/Config/Database.hs @@ -25,7 +25,7 @@ import Text.InterpolatedString.Perl6 (q) import Protolude -type RoleSettings = (HM.HashMap ByteString [(ByteString, ByteString)]) +type RoleSettings = (HM.HashMap ByteString (HM.HashMap ByteString ByteString)) queryPgVersion :: Bool -> Session PgVersion queryPgVersion prepared = statement mempty $ pgVersionStatement prepared @@ -75,7 +75,7 @@ queryRoleSettings prepared = transaction SQL.ReadCommitted SQL.Read $ SQL.statement mempty $ roleSettingsStatement prepared roleSettingsStatement :: Bool -> SQL.Statement () RoleSettings -roleSettingsStatement = SQL.Statement sql HE.noParams decodeSettings +roleSettingsStatement = SQL.Statement sql HE.noParams decodeRoleSettings where sql = [q| with @@ -89,14 +89,14 @@ roleSettingsStatement = SQL.Statement sql HE.noParams decodeSettings SELECT rolname, substr(setting, 1, strpos(setting, '=') - 1) as key, - substr(setting, strpos(setting, '=') + 1) as value + lower(substr(setting, strpos(setting, '=') + 1)) as value FROM role_setting ) select rolname, array_agg(row(key, value)) from kv_settings group by rolname; |] - decodeSettings = HM.fromList . map (bimap encodeUtf8 ((encodeUtf8 *** encodeUtf8) <$>)) <$> HD.rowList aRow + decodeRoleSettings = HM.fromList . map (bimap encodeUtf8 (HM.fromList . ((encodeUtf8 *** encodeUtf8) <$>))) <$> HD.rowList aRow aRow :: HD.Row (Text, [(Text, Text)]) aRow = (,) <$> column HD.text <*> compositeArrayColumn ((,) <$> compositeField HD.text <*> compositeField HD.text) diff --git a/src/PostgREST/Query.hs b/src/PostgREST/Query.hs index 2464f2c49..b72a0c772 100644 --- a/src/PostgREST/Query.hs +++ b/src/PostgREST/Query.hs @@ -235,9 +235,9 @@ optionalRollback AppConfig{..} ApiRequest{iPreferences=Preferences{..}} = do configDbTxAllowOverride && preferTransaction == Just Rollback -- | Runs local (transaction scoped) GUCs for every request. -setPgLocals :: AppConfig -> KM.KeyMap JSON.Value -> BS.ByteString -> +setPgLocals :: AppConfig -> KM.KeyMap JSON.Value -> BS.ByteString -> [(ByteString, ByteString)] -> ApiRequest -> PgVersion -> DbHandler () -setPgLocals AppConfig{..} claims role req actualPgVersion = lift $ +setPgLocals AppConfig{..} claims role roleSettings req actualPgVersion = lift $ SQL.statement mempty $ SQL.dynamicallyParameterized ("select " <> intercalateSnippet ", " (searchPathSql : roleSql ++ roleSettingsSql ++ claimsSql ++ [methodSql, pathSql] ++ headersSql ++ cookiesSql ++ appSettingsSql)) HD.noResult configDbPreparedStatements @@ -254,9 +254,7 @@ setPgLocals AppConfig{..} claims role req actualPgVersion = lift $ then setConfigLocal "request.jwt.claim." <$> [(toUtf8 $ K.toText c, toUtf8 $ unquoted v) | (c,v) <- KM.toList claims] else [setConfigLocal mempty ("request.jwt.claims", LBS.toStrict $ JSON.encode claims)] roleSql = [setConfigLocal mempty ("role", role)] - roleSettingsSql = if null configRoleSettings - then mempty - else setConfigLocal mempty <$> fromMaybe mempty (HM.lookup role configRoleSettings) + roleSettingsSql = setConfigLocal mempty <$> roleSettings appSettingsSql = setConfigLocal mempty <$> (join bimap toUtf8 <$> configAppSettings) searchPathSql = let schemas = pgFmtIdentList (iSchema req : configDbExtraSearchPath) in diff --git a/src/PostgREST/SchemaCache.hs b/src/PostgREST/SchemaCache.hs index 90f69e3f0..9ac260d93 100644 --- a/src/PostgREST/SchemaCache.hs +++ b/src/PostgREST/SchemaCache.hs @@ -250,6 +250,7 @@ decodeFuncs = <*> column HD.bool) <*> (parseVolatility <$> column HD.char) <*> column HD.bool + <*> nullableColumn HD.text addKey :: Routine -> (QualifiedIdentifier, Routine) addKey pd = (QualifiedIdentifier (pdSchema pd) (pdName pd), pd) @@ -339,7 +340,8 @@ funcsSqlQuery pgVer = [q| ) AS rettype_is_composite, bt.oid <> bt.base as rettype_is_composite_alias, p.provolatile, - p.provariadic > 0 as hasvariadic + p.provariadic > 0 as hasvariadic, + lower((regexp_split_to_array((regexp_split_to_array(config, '='))[2], ','))[1]) AS transaction_isolation_level FROM pg_proc p LEFT JOIN arguments a ON a.oid = p.oid JOIN pg_namespace pn ON pn.oid = p.pronamespace @@ -348,6 +350,7 @@ funcsSqlQuery pgVer = [q| JOIN pg_namespace tn ON tn.oid = t.typnamespace LEFT JOIN pg_class comp ON comp.oid = t.typrelid LEFT JOIN pg_description as d ON d.objoid = p.oid + LEFT JOIN LATERAL unnest(proconfig) config ON config like 'default_transaction_isolation%' WHERE t.oid <> 'trigger'::regtype AND COALESCE(a.callable, true) |] <> (if pgVer >= pgVersion110 then "AND prokind = 'f'" else "AND NOT (proisagg OR proiswindow)") diff --git a/src/PostgREST/SchemaCache/Routine.hs b/src/PostgREST/SchemaCache/Routine.hs index 568d734d9..caae214d1 100644 --- a/src/PostgREST/SchemaCache/Routine.hs +++ b/src/PostgREST/SchemaCache/Routine.hs @@ -48,6 +48,7 @@ data Routine = Function , pdReturnType :: RetType , pdVolatility :: FuncVolatility , pdHasVariadic :: Bool + , pdIsoLvl :: Maybe Text } deriving (Eq, Generic, JSON.ToJSON) @@ -61,10 +62,10 @@ data RoutineParam = RoutineParam -- Order by least number of params in the case of overloaded functions instance Ord Routine where - Function schema1 name1 des1 prms1 rt1 vol1 hasVar1 `compare` Function schema2 name2 des2 prms2 rt2 vol2 hasVar2 + Function schema1 name1 des1 prms1 rt1 vol1 hasVar1 iso1 `compare` Function schema2 name2 des2 prms2 rt2 vol2 hasVar2 iso2 | schema1 == schema2 && name1 == name2 && length prms1 < length prms2 = LT | schema2 == schema2 && name1 == name2 && length prms1 > length prms2 = GT - | otherwise = (schema1, name1, des1, prms1, rt1, vol1, hasVar1) `compare` (schema2, name2, des2, prms2, rt2, vol2, hasVar2) + | otherwise = (schema1, name1, des1, prms1, rt1, vol1, hasVar1, iso1) `compare` (schema2, name2, des2, prms2, rt2, vol2, hasVar2, iso2) -- | A map of all procs, all of which can be overloaded(one entry will have more than one Routine). -- | It uses a HashMap for a faster lookup. diff --git a/test/io/fixtures.sql b/test/io/fixtures.sql index 3d66492e2..f96237c74 100644 --- a/test/io/fixtures.sql +++ b/test/io/fixtures.sql @@ -8,7 +8,13 @@ ALTER ROLE :USER SET pgrst.db_anon_role = 'postgrest_test_anonymous'; CREATE ROLE postgrest_test_author; -GRANT postgrest_test_anonymous, postgrest_test_author TO :USER; +CREATE ROLE postgrest_test_serializable; +alter role postgrest_test_serializable set default_transaction_isolation = 'serializable'; + +CREATE ROLE postgrest_test_repeatable_read; +alter role postgrest_test_repeatable_read set default_transaction_isolation = 'REPEATABLE READ'; + +GRANT postgrest_test_anonymous, postgrest_test_author, postgrest_test_serializable, postgrest_test_repeatable_read TO :USER; CREATE SCHEMA v1; GRANT USAGE ON SCHEMA v1 TO postgrest_test_anonymous; @@ -108,3 +114,31 @@ begin alter role current_user set statement_timeout = %L; $$, timeout); end $_$ volatile language plpgsql ; + +create table items as select x as id from generate_series(1,5) x; + +create view items_w_isolation_level as +select + id, + current_setting('transaction_isolation', true) as isolation_level +from items; + +grant all on items_w_isolation_level to postgrest_test_anonymous, postgrest_test_repeatable_read, postgrest_test_serializable; + +create function default_isolation_level() +returns text as $$ + select current_setting('transaction_isolation', true); +$$ +language sql; + +create function serializable_isolation_level() +returns text as $$ + select current_setting('transaction_isolation', true); +$$ +language sql set default_transaction_isolation = 'serializable'; + +create function repeatable_read_isolation_level() +returns text as $$ + select current_setting('transaction_isolation', true); +$$ +language sql set default_transaction_isolation = 'REPEATABLE READ'; diff --git a/test/io/test_io.py b/test/io/test_io.py index b4a0c8e5e..a6f49b286 100644 --- a/test/io/test_io.py +++ b/test/io/test_io.py @@ -873,6 +873,79 @@ def test_role_settings(defaultenv): assert response.text == '"10s"' +def test_isolation_level(defaultenv): + "isolation_level should be set per role and per function" + + env = { + **defaultenv, + "PGRST_JWT_SECRET": SECRET, + } + + with run(env=env) as postgrest: + # default isolation level for postgrest_test_anonymous + response = postgrest.session.get( + "/items_w_isolation_level?select=isolation_level&limit=1" + ) + assert response.text == '[{"isolation_level":"read committed"}]' + + # isolation level for postgrest_test_repeatable_read on GET + headers = jwtauthheader({"role": "postgrest_test_repeatable_read"}, SECRET) + response = postgrest.session.get( + "/items_w_isolation_level?select=isolation_level&limit=1", headers=headers + ) + assert response.text == '[{"isolation_level":"repeatable read"}]' + + # isolation level for postgrest_test_serializable on POST + headers = jwtauthheader({"role": "postgrest_test_serializable"}, SECRET) + headers["Prefer"] = "return=representation" + response = postgrest.session.post( + "/items_w_isolation_level?select=isolation_level", + json={"id": "666"}, + headers=headers, + ) + assert response.text == '[{"isolation_level":"serializable"}]' + + # isolation level for postgrest_test_serializable on PATCH + headers = jwtauthheader({"role": "postgrest_test_serializable"}, SECRET) + headers["Prefer"] = "return=representation" + response = postgrest.session.patch( + "/items_w_isolation_level?select=isolation_level&id=eq.666", + json={"id": "666"}, + headers=headers, + ) + assert response.text == '[{"isolation_level":"serializable"}]' + + # isolation level for postgrest_test_serializable on DELETE + headers = jwtauthheader({"role": "postgrest_test_serializable"}, SECRET) + headers["Prefer"] = "return=representation" + response = postgrest.session.delete( + "/items_w_isolation_level?select=isolation_level&id=eq.666", headers=headers + ) + assert response.text == '[{"isolation_level":"serializable"}]' + + # default isolation level for function + response = postgrest.session.get("/rpc/default_isolation_level") + assert response.text == '"read committed"' + + # changes with role isolation level + headers = jwtauthheader({"role": "postgrest_test_repeatable_read"}, SECRET) + response = postgrest.session.get( + "/rpc/default_isolation_level", headers=headers + ) + assert response.text == '"repeatable read"' + + # isolation level can be set per function + response = postgrest.session.get("/rpc/serializable_isolation_level") + assert response.text == '"serializable"' + response = postgrest.session.get("/rpc/repeatable_read_isolation_level") + assert response.text == '"repeatable read"' + + # isolation level for a function overrides the role isolation level + headers = jwtauthheader({"role": "postgrest_test_repeatable_read"}, SECRET) + response = postgrest.session.get("/rpc/serializable_isolation_level") + assert response.text == '"serializable"' + + # TODO: This test fails now because of https://github.com/PostgREST/postgrest/pull/2122 # The stack size of 1K(-with-rtsopts=-K1K) is not enough and this fails with "stack overflow" # A stack size of 200K seems to be enough for succeess