diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,20 @@ # Changelog for pgmq-migration +## 0.7.0.0 -- 2026-10-26 + +### Bug Fixes + +- Append `0007-notify-only-registered-queues.sql`: reject physical partitions before + throttle updates and restrict missing-row fail-open to registered queue names. + Existing triggers adopt the correction without recreation. Ordinary queues retain + crash fail-open until an explicit enable/reconcile restores the throttle row. + Migrations 0001–0006 are unchanged. + +### Other Changes + +- Ephemeral PostgreSQL roots are created/validated as owner-only (`0700`), existing + owned roots are tightened, unsafe paths are rejected and registry contents survive. + ## 0.6.1.1 -- 2026-09-25 Accept the `pg-migrate` 1.2 series alongside 1.1: `pg-migrate`, `pg-migrate-embed` and diff --git a/migrations/0007-notify-only-registered-queues.sql b/migrations/0007-notify-only-registered-queues.sql new file mode 100644 --- /dev/null +++ b/migrations/0007-notify-only-registered-queues.sql @@ -0,0 +1,74 @@ +-- Notification fail-open only for a registered queue. +-- +-- Migration 0003 made pgmq.notify_queue_listeners() fail open when its throttle +-- row is missing, so crash recovery truncating the UNLOGGED throttle table +-- cannot silently stop delivery. That branch could not tell a row lost to a +-- crash from a missing leaf-name row. PostgreSQL clones a row trigger onto +-- every leaf partition, and inside the clone TG_TABLE_NAME is the partition +-- (q_pq_p0), so the extracted name (pq_p0) normally has no throttle row. On a +-- partitioned queue the fail-open therefore fired on every insert, unthrottled, +-- on a per-partition channel no listener on the documented channel receives. +-- +-- The fail-open now requires the extracted name to be a queue registered in +-- pgmq.meta. A relation guard before UPDATE rejects leaf partitions even +-- when their extracted names have alias metadata or throttle rows. Ordinary +-- queues keep crash recovery; a leaf partition, or a +-- mixed-case throttle row the trigger cannot match, behaves as upstream's +-- trigger does: no notification. Partitioned queues receive no insert +-- notifications on the hardened native install. Stock SQL is silent for +-- normal partitioned queues but does not guard manually aliased leaves. +-- +-- See docs/design/015-notification-delivery-contract.md and +-- docs/adr/notification-override-lineage-and-partitioned-queue-boundary.md. + +CREATE OR REPLACE FUNCTION pgmq.notify_queue_listeners() +RETURNS TRIGGER AS $$ +DECLARE + queue_name_extracted TEXT; -- Queue name extracted from trigger table name + updated_count INTEGER; -- Number of rows updated (0 or 1) +BEGIN + -- Metadata names do not prove that this is an ordinary queue relation. + IF EXISTS ( + SELECT 1 FROM pg_catalog.pg_class c + WHERE c.oid = TG_RELID AND c.relispartition + ) THEN + RETURN NEW; + END IF; + + queue_name_extracted := substring(TG_TABLE_NAME from 3); + + UPDATE pgmq.notify_insert_throttle + SET last_notified_at = clock_timestamp() + WHERE queue_name = queue_name_extracted + AND ( + throttle_interval_ms = 0 -- No throttling configured + OR clock_timestamp() - last_notified_at >= + (throttle_interval_ms * INTERVAL '1 millisecond') -- Throttle interval has elapsed + ); + + -- Check how many rows were updated (will be 0 or 1) + GET DIAGNOSTICS updated_count = ROW_COUNT; + + IF updated_count > 0 THEN + PERFORM PG_NOTIFY('pgmq.' || TG_TABLE_NAME || '.' || TG_OP, NULL); + ELSIF EXISTS ( + SELECT 1 FROM pgmq.meta m + WHERE m.queue_name = queue_name_extracted + ) + AND NOT EXISTS ( + SELECT 1 FROM pgmq.notify_insert_throttle nit + WHERE nit.queue_name = queue_name_extracted + ) THEN + -- Fail open only for a queue that could have had a throttle row: the + -- extracted name is registered in pgmq.meta, so a missing throttle row + -- can be crash truncation. Both + -- probes are unique-index lookups and run only when the UPDATE matched + -- nothing. The row is deliberately NOT re-inserted here (see 0003): the + -- configured interval is the crash's data loss, and the reconciler + -- restores it when explicitly invoked, typically at application startup. + PERFORM PG_NOTIFY('pgmq.' || TG_TABLE_NAME || '.' || TG_OP, NULL); + END IF; + +RETURN NEW; +END; +$$ LANGUAGE plpgsql; diff --git a/migrations/manifest b/migrations/manifest --- a/migrations/manifest +++ b/migrations/manifest @@ -4,3 +4,4 @@ 0004-upgrade-v1.12.0.sql 0005-upgrade-v1.13.0.sql 0006-preserve-partitioned-reentry-v1.13.0.sql +0007-notify-only-registered-queues.sql diff --git a/pgmq-migration.cabal b/pgmq-migration.cabal --- a/pgmq-migration.cabal +++ b/pgmq-migration.cabal @@ -1,6 +1,6 @@ cabal-version: 3.4 name: pgmq-migration -version: 0.6.1.1 +version: 0.7.0.0 synopsis: PGMQ schema migrations without PostgreSQL extension description: Installs the PGMQ schema into PostgreSQL without requiring the pgmq extension. @@ -75,16 +75,21 @@ type: exitcode-stdio-1.0 hs-source-dirs: test main-is: Main.hs + other-modules: + NotifyListener + RootPermissionsSpec + build-depends: base >=4.18 && <5, bytestring, containers, directory, - ephemeral-pg >=0.3.1.0, + ephemeral-pg >=0.3.1 && <0.4, hasql, pg-migrate, pg-migrate-import-hasql-migration, - pgmq-migration >=0.6 && <0.7, + pgmq-migration >=0.7 && <0.8, + postgresql-libpq >=0.10.1 && <0.12, tasty ^>=1.5, tasty-hunit ^>=0.10, text, diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -4,8 +4,9 @@ module Main (main) where import Control.Concurrent (forkIO, newEmptyMVar, putMVar, takeMVar) -import Control.Exception (SomeException, try) -import Control.Monad (filterM, forM_) +import Control.Exception (SomeException, finally, try) +import Control.Monad (filterM, forM_, unless) +import Data.Bits ((.&.)) import Data.ByteString qualified as ByteString import Data.Foldable (toList) import Data.List.NonEmpty (NonEmpty (..)) @@ -56,8 +57,11 @@ migrationNameText, planDescription, ) + -- MigrationDescription's 'migrationId' field would shadow the smart constructor -- of the same name imported above, so it is reached through this alias only. + +import Database.PostgreSQL.LibPQ qualified as PQ import Database.PostgreSQL.Migrate.Internal qualified as MigrateInternal import EphemeralPg ( Config (temporaryRoot), @@ -66,6 +70,7 @@ defaultConfig, withCachedConfig, ) +import EphemeralPg qualified as Pg import Hasql.Connection qualified as Connection import Hasql.Connection.Settings qualified as Settings import Hasql.Decoders qualified as Decoders @@ -74,6 +79,7 @@ import Hasql.Session qualified as Session import Hasql.Statement (preparable) import Hasql.Statement qualified as Statement +import NotifyListener (awaitNotify, withListener) import Pgmq.Migration qualified as Migration import Pgmq.Migration.History.HasqlMigration ( AlternativeHistoryPolicy (..), @@ -82,8 +88,12 @@ pgmqHasqlMigrationSourceConfig, pgmqHasqlMigrationSourceConfigWithPolicy, ) -import System.Directory (createDirectoryIfMissing, doesFileExist) +import RootPermissionsSpec qualified +import System.Directory (doesFileExist) import System.Environment (lookupEnv) +import System.IO.Error (catchIOError, isAlreadyExistsError) +import System.Posix.Directory qualified as Posix +import System.Posix.Files (deviceID, fileID, fileMode, fileOwner, getSymbolicLinkStatus, isDirectory, setFileMode) import System.Posix.User (getEffectiveUserID) import Test.Tasty (TestTree, defaultMain, testGroup) import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=)) @@ -103,11 +113,36 @@ uid <- getEffectiveUserID pure ("/tmp/ephpg-pgmq-hs-" <> show uid) +-- | Establish a real effective-uid-owned root without deleting its contents. +ensureOwnerOnlyRoot :: FilePath -> IO () +ensureOwnerOnlyRoot root = do + catchIOError (Posix.createDirectory root 0o700) $ \err -> + unless (isAlreadyExistsError err) (ioError err) + uid <- getEffectiveUserID + before <- getSymbolicLinkStatus root + unless (isDirectory before && fileOwner before == uid) $ + ioError (userError ("Unsafe ephemeral PostgreSQL root: " <> root)) + setFileMode root 0o700 + after <- getSymbolicLinkStatus root + unless + ( isDirectory after + && fileOwner after + == uid + && deviceID before + == deviceID after + && fileID before + == fileID after + && fileMode after + .&. 0o7777 + == 0o700 + ) + (ioError (userError ("Ephemeral PostgreSQL root changed or is not owner-only: " <> root))) + -- | Cached-startup configuration pinned to 'ephemeralRoot'. ephemeralConfig :: IO Config ephemeralConfig = do root <- ephemeralRoot - createDirectoryIfMissing True root + ensureOwnerOnlyRoot root pure defaultConfig {temporaryRoot = Last (Just root)} main :: IO () @@ -119,16 +154,17 @@ case connResult of Left err -> error $ "Failed to connect: " <> show err Right conn -> - defaultMain (tests connSettings conn) + defaultMain (tests (Pg.connectionString db) connSettings conn) case result of Left startErr -> error $ "Failed to start temp database: " <> show startErr Right () -> pure () -tests :: Settings.Settings -> Connection.Connection -> TestTree -tests settings conn = +tests :: Text -> Settings.Settings -> Connection.Connection -> TestTree +tests conninfo settings conn = testGroup "pgmq-migration" - [ testGroup + [ RootPermissionsSpec.tests ensureOwnerOnlyRoot, + testGroup "native definition" [ testCase "immutable history and upstream payload provenance" (testNativePayload conn), testCase "component pgmq lists the ledger in order and has no dependencies" testNativeComponent @@ -137,7 +173,8 @@ "native runner" [ testCase "fresh install applies once and is idempotent" (testNativeRunner settings conn), testCase "1.12 and 1.13 catalog convergence" (testConvergence conn), - testCase "populated partition upgrades and spill recovery" (testPartitions settings conn) + testCase "populated partition upgrades and spill recovery" (testPartitions settings conn), + testCase "existing triggers adopt notification correction on populated upgrade" (testNotifyUpgrade conninfo settings conn) ], testGroup "history import" @@ -155,7 +192,8 @@ forM_ [ ("0001-install-v1.11.0.sql", "faa9b8800005f80fbdf6a33d071183d2"), ("0002-schema-management-comment.sql", "f3010777d47ec6d4f7d92a179bcf09b8"), - ("0003-notify-crash-safety-and-locking.sql", "e0a66c96bd8811607440dbca0cb302ea") + ("0003-notify-crash-safety-and-locking.sql", "e0a66c96bd8811607440dbca0cb302ea"), + ("0007-notify-only-registered-queues.sql", "2892dc6622062d1530a3fa29c17039bb") ] $ \(name, digest) -> do bytes <- readMigration name @@ -183,7 +221,7 @@ testConvergence conn = forM_ [False, True] $ \latest -> do resetDb conn names <- nativeMigrationNames - forM_ (take (if latest then 6 else 4) names) $ \name -> + forM_ (take (if latest then length names else 4) names) $ \name -> readMigration (T.unpack name <> ".sql") >>= runSql conn . Text.decodeUtf8 upgraded <- schemaSnapshot conn resetDb conn @@ -339,6 +377,60 @@ resetDb conn runSql conn "DROP EXTENSION pg_partman CASCADE" +-- | Install six entries, preserve deployed trigger OIDs/data/settings, then apply 0007. +testNotifyUpgrade :: Text -> Settings.Settings -> Connection.Connection -> IO () +testNotifyUpgrade conninfo settings conn = do + required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN" + available <- queryText conn "SELECT EXISTS (SELECT FROM pg_available_extensions WHERE name='pg_partman')::text" + if available /= "true" + then if required then assertFailure "pg_partman required" else putStrLn "SKIPPED: notification upgrade requires pg_partman" + else flip finally (resetDb conn >> runSql conn "DROP EXTENSION IF EXISTS pg_partman CASCADE") $ do + resetDb conn + runSql conn "CREATE SCHEMA IF NOT EXISTS partman; CREATE EXTENSION IF NOT EXISTS pg_partman SCHEMA partman" + installPrefix settings 6 + runSql conn "SELECT pgmq.create('notify_ordinary'); SELECT pgmq.create_partitioned('notify_partition','10','100'); SELECT pgmq.enable_notify_insert('notify_ordinary',0); SELECT pgmq.enable_notify_insert('notify_partition',0)" + withListener conninfo "pgmq.q_notify_partition_p0.INSERT" $ \listener -> do + runSql conn "SELECT pgmq.send('notify_partition','{\"before\":true}')" + beforeNotify <- awaitNotify listener 20 + fmap PQ.notifyRelname beforeNotify @?= Just "pgmq.q_notify_partition_p0.INSERT" + runSql conn "SELECT pgmq.send('notify_ordinary','{\"before\":true}')" + before <- queryText conn notifyUpgradeSnapshot + plan <- nativePlan + report <- runMigrationPlan defaultRunOptions settings plan >>= either (assertFailure . show) pure + map outcome (toList (results report)) @?= replicate 6 AlreadyApplied <> [AppliedNow] + queryText conn notifyUpgradeSnapshot >>= (@?= before) + -- No enable/reconcile occurs: every parent and cloned trigger retains its OID. + runListen listener "pgmq.q_notify_partition.INSERT" + partitionResult <- PQ.exec listener "SELECT 'pgmq.' || c.relname || '.INSERT' FROM pg_inherits i JOIN pg_class c ON c.oid=i.inhrelid WHERE i.inhparent='pgmq.q_notify_partition'::regclass" + case partitionResult of + Nothing -> assertFailure "partition channel query failed" + Just res -> do + PQ.resultStatus res >>= (@?= PQ.TuplesOk) + rows <- PQ.ntuples res + forM_ [0 .. rows - 1] $ \row -> do + value <- PQ.getvalue res row 0 + maybe (assertFailure "null partition channel") (runListen listener . Text.decodeUtf8) value + forM_ [1 :: Int .. 3] $ \_ -> runSql conn "SELECT pgmq.send('notify_partition','{\"after\":true}')" + silent <- awaitNotify listener 10 + fmap PQ.notifyRelname silent @?= Nothing + assertSql conn "(SELECT count(*)=4 FROM pgmq.q_notify_partition) AND (SELECT last_notified_at=to_timestamp(0) FROM pgmq.notify_insert_throttle WHERE queue_name='notify_partition')" + withListener conninfo "pgmq.q_notify_ordinary.INSERT" $ \listener -> do + runSql conn "SELECT pgmq.send('notify_ordinary','{\"after\":true}')" + ordinary <- awaitNotify listener 20 + fmap PQ.notifyRelname ordinary @?= Just "pgmq.q_notify_ordinary.INSERT" + plan <- nativePlan + again <- runMigrationPlan defaultRunOptions settings plan >>= either (assertFailure . show) pure + map outcome (toList (results again)) @?= replicate 7 AlreadyApplied + where + runListen listener channel = do + result <- PQ.exec listener (Text.encodeUtf8 ("LISTEN \"" <> channel <> "\"")) + case result of + Nothing -> assertFailure "LISTEN failed" + Just res -> PQ.resultStatus res >>= (@?= PQ.CommandOk) + +notifyUpgradeSnapshot :: Text +notifyUpgradeSnapshot = "SELECT jsonb_build_object('data',(SELECT jsonb_agg(to_jsonb(m) ORDER BY source,msg_id) FROM (SELECT 'ordinary' source,msg_id,message FROM pgmq.q_notify_ordinary UNION ALL SELECT 'partition',msg_id,message FROM pgmq.q_notify_partition) m),'triggers',(SELECT jsonb_agg(jsonb_build_array(oid::text,tgrelid::text,tgfoid::text,tgparentid::text) ORDER BY oid) FROM pg_trigger WHERE tgname='trigger_notify_queue_insert_listeners'),'throttles',(SELECT jsonb_agg(to_jsonb(t) ORDER BY queue_name) FROM pgmq.notify_insert_throttle t))::text" + preservedDataSql :: Text preservedDataSql = "SELECT jsonb_agg(to_jsonb(m) ORDER BY source,msg_id)::text FROM (SELECT 'numeric' source,msg_id,message FROM pgmq.q_numeric_old UNION ALL SELECT 'archive',msg_id,message FROM pgmq.a_numeric_old UNION ALL SELECT 'time',msg_id,message FROM pgmq.q_time_old UNION ALL SELECT 'ordinary',msg_id,message FROM pgmq.q_ordinary_old) m" @@ -363,7 +455,8 @@ "0003-notify-crash-safety-and-locking", "0004-upgrade-v1.12.0", "0005-upgrade-v1.13.0", - "0006-preserve-partitioned-reentry-v1.13.0" + "0006-preserve-partitioned-reentry-v1.13.0", + "0007-notify-only-registered-queues" ] actual -> assertFailure ("unexpected native PGMQ plan: " <> show actual) diff --git a/test/NotifyListener.hs b/test/NotifyListener.hs new file mode 100644 --- /dev/null +++ b/test/NotifyListener.hs @@ -0,0 +1,48 @@ +{-# LANGUAGE OverloadedStrings #-} + +-- | Checked LISTEN transport for deployed-upgrade acceptance. +module NotifyListener (withListener, awaitNotify) where + +import Control.Concurrent (threadDelay) +import Control.Exception (bracket) +import Control.Monad (unless) +import Data.Text (Text) +import Data.Text qualified as T +import Data.Text.Encoding qualified as TE +import Database.PostgreSQL.LibPQ qualified as LibPQ +import Test.Tasty.HUnit (assertFailure, (@?=)) + +withListener :: Text -> Text -> (LibPQ.Connection -> IO a) -> IO a +withListener conninfo channel action = + bracket (LibPQ.connectdb (TE.encodeUtf8 conninfo)) LibPQ.finish $ \conn -> do + connStatus <- LibPQ.status conn + unless (connStatus == LibPQ.ConnectionOk) $ do + err <- LibPQ.errorMessage conn + assertFailure $ "libpq connection failed: " <> show err + -- The channel contains dots, so LISTEN needs the identifier double-quoted. + result <- LibPQ.exec conn (TE.encodeUtf8 ("LISTEN " <> quoteIdentifier channel)) + case result of + Nothing -> assertFailure "LISTEN returned no result" + Just res -> do + execStatus <- LibPQ.resultStatus res + unless (execStatus == LibPQ.CommandOk) $ + assertFailure ("LISTEN failed with " <> show execStatus) + action conn + +quoteIdentifier :: Text -> Text +quoteIdentifier ident = "\"" <> T.replace "\"" "\"\"" ident <> "\"" + +-- | Poll for a notification, 100 ms per attempt. +awaitNotify :: LibPQ.Connection -> Int -> IO (Maybe LibPQ.Notify) +awaitNotify conn attempts + | attempts <= 0 = pure Nothing + | otherwise = do + ok <- LibPQ.consumeInput conn + unless ok (assertFailure "libpq consumeInput failed") + LibPQ.status conn >>= (@?= LibPQ.ConnectionOk) + pending <- LibPQ.notifies conn + case pending of + Just n -> pure (Just n) + Nothing -> do + threadDelay 100_000 + awaitNotify conn (attempts - 1) diff --git a/test/RootPermissionsSpec.hs b/test/RootPermissionsSpec.hs new file mode 100644 --- /dev/null +++ b/test/RootPermissionsSpec.hs @@ -0,0 +1,63 @@ +-- | Check each suite's local root helper using private disposable fixtures. +module RootPermissionsSpec (tests) where + +import Control.Exception (IOException, bracket, try) +import Data.Bits ((.&.)) +import System.Directory (getTemporaryDirectory, removePathForcibly) +import System.Posix.Directory qualified as Posix +import System.Posix.Files (createSymbolicLink, fileMode, getSymbolicLinkStatus, isDirectory, setFileMode) +import System.Posix.Temp (mkdtemp) +import Test.Tasty (TestTree, testGroup) +import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=)) + +tests :: (FilePath -> IO ()) -> TestTree +tests ensure = + testGroup + "RootPermissionsSpec" + [ testCase "missing root is a real 0700 directory" $ private $ \parent -> do + let root = parent <> "/root" + ensure root + status <- getSymbolicLinkStatus root + assertBool "real directory" (isDirectory status) + fileMode status .&. 0o7777 @?= 0o700, + testCase "owned 0755 directory is tightened without losing contents" $ private $ \parent -> do + let root = parent <> "/root" + Posix.createDirectory root 0o755 + setFileMode root 0o755 + writeFile (root <> "/sentinel") "preserve" + ensure root + mode root >>= (@?= 0o700) + readFile (root <> "/sentinel") >>= (@?= "preserve"), + testCase "repeated call preserves contents and permissions" $ private $ \parent -> do + let root = parent <> "/root" + ensure root + writeFile (root <> "/sentinel") "preserve" + ensure root + mode root >>= (@?= 0o700) + readFile (root <> "/sentinel") >>= (@?= "preserve"), + testCase "regular file is rejected unchanged" $ private $ \parent -> do + let root = parent <> "/root" + writeFile root "preserve" + setFileMode root 0o644 + rejected (ensure root) + mode root >>= (@?= 0o644) + readFile root >>= (@?= "preserve"), + testCase "symlink is rejected without chmod of its target" $ private $ \parent -> do + let root = parent <> "/root" + target = parent <> "/target" + Posix.createDirectory target 0o755 + setFileMode target 0o755 + writeFile (target <> "/sentinel") "preserve" + createSymbolicLink target root + rejected (ensure root) + mode target >>= (@?= 0o755) + readFile (target <> "/sentinel") >>= (@?= "preserve") + ] + where + mode path = (.&. 0o7777) . fileMode <$> getSymbolicLinkStatus path + rejected action = do + result <- try action :: IO (Either IOException ()) + case result of + Left _ -> pure () + Right () -> assertFailure "unsafe root accepted" + private action = bracket (getTemporaryDirectory >>= \tmp -> mkdtemp (tmp <> "/pgmq-root-test-")) removePathForcibly action