Merge branch 'smh-validate-hooks' into 'master'

Validate custom hook archive prior to accepting a transaction

Closes #5126

See merge request https://gitlab.com/gitlab-org/gitaly/-/merge_requests/5835

Merged-by: Justin Tobler <jtobler@gitlab.com>
Approved-by: Justin Tobler <jtobler@gitlab.com>
Approved-by: Toon Claes <toon@gitlab.com>
Co-authored-by: Sami Hiltunen <shiltunen@gitlab.com>

2 files changed, 82 insertions, 10 deletions

diff --git a/internal/gitaly/transaction_manager.go b/internal/gitaly/transaction_manager.go
index 82f2d6983..fd1b71284 100644
--- a/internal/gitaly/transaction_manager.go
+++ b/internal/gitaly/transaction_manager.go
@@ -188,10 +188,6 @@ func (mgr *TransactionManager) Begin(ctx context.Context) (_ *Transaction, retur
 	}
 
 	mgr.mutex.Lock()
-	if !mgr.repositoryExists {
-		mgr.mutex.Unlock()
-		return nil, ErrRepositoryNotFound
-	}
 
 	txn := &Transaction{
 		commit:   mgr.commit,
@@ -213,6 +209,7 @@ func (mgr *TransactionManager) Begin(ctx context.Context) (_ *Transaction, retur
 	openTransactionElement := mgr.openTransactions.PushBack(txn)
 
 	readReady := mgr.applyNotifications[txn.snapshot.ReadIndex]
+	repositoryExists := mgr.repositoryExists
 	mgr.mutex.Unlock()
 	if readReady == nil {
 		// The snapshot log entry is already applied if there is no notification channel for it.
@@ -261,6 +258,10 @@ func (mgr *TransactionManager) Begin(ctx context.Context) (_ *Transaction, retur
 	case <-mgr.ctx.Done():
 		return nil, ErrTransactionProcessingStopped
 	case <-readReady:
+		if !repositoryExists {
+			return nil, ErrRepositoryNotFound
+		}
+
 		return txn, nil
 	}
 }
@@ -513,6 +514,10 @@ func (mgr *TransactionManager) commit(ctx context.Context, transaction *Transact
 		return fmt.Errorf("setup staging repository: %w", err)
 	}
 
+	if err := mgr.stageHooks(ctx, transaction); err != nil {
+		return fmt.Errorf("stage hooks: %w", err)
+	}
+
 	if err := mgr.packObjects(ctx, transaction); err != nil {
 		return fmt.Errorf("pack objects: %w", err)
 	}
@@ -536,6 +541,30 @@ func (mgr *TransactionManager) commit(ctx context.Context, transaction *Transact
 	}
 }
 
+// stageHooks extracts the new hooks, if any, into <stagingDirectory>/custom_hooks. This is ensures the TAR
+// is valid prior to committing the transaction. The hooks files on the disk are also used to compute a vote
+// for Praefect.
+func (mgr *TransactionManager) stageHooks(ctx context.Context, transaction *Transaction) error {
+	if transaction.customHooksUpdate == nil || len(transaction.customHooksUpdate.CustomHooksTAR) == 0 {
+		return nil
+	}
+
+	if err := transaction.initStagingDirectory(); err != nil {
+		return fmt.Errorf("init staging directory: %w", err)
+	}
+
+	if err := repoutil.ExtractHooks(
+		ctx,
+		bytes.NewReader(transaction.customHooksUpdate.CustomHooksTAR),
+		transaction.stagingDirectory,
+		false,
+	); err != nil {
+		return fmt.Errorf("extract hooks: %w", err)
+	}
+
+	return nil
+}
+
 // setupStagingRepository sets a repository that is used to stage the transaction. The staging repository
 // has the quarantine applied so the objects are available for packing and verifying the references.
 func (mgr *TransactionManager) setupStagingRepository(ctx context.Context, transaction *Transaction) error {
diff --git a/internal/gitaly/transaction_manager_test.go b/internal/gitaly/transaction_manager_test.go
index f4817c3d2..47028edfc 100644
--- a/internal/gitaly/transaction_manager_test.go
+++ b/internal/gitaly/transaction_manager_test.go
@@ -287,7 +287,9 @@ func TestTransactionManager(t *testing.T) {
 		// Context is the context to use for the Commit call.
 		Context context.Context
 		// ExpectedError is the error that is expected to be returned when committing the transaction.
-		ExpectedError error
+		// If ExpectedError is a function with signature func(tb testing.TB, actualErr error), it is
+		// ran instead to asser the error.
+		ExpectedError any
 
 		// SkipVerificationFailures sets the verification failure handling for this commit.
 		SkipVerificationFailures bool
@@ -1105,6 +1107,24 @@ func TestTransactionManager(t *testing.T) {
 			},
 		},
 		{
+			desc: "rejects invalid custom hooks",
+			steps: steps{
+				StartManager{},
+				Begin{
+					TransactionID: 1,
+				},
+				Commit{
+					TransactionID: 1,
+					CustomHooksUpdate: &CustomHooksUpdate{
+						CustomHooksTAR: []byte("corrupted tar"),
+					},
+					ExpectedError: func(tb testing.TB, actualErr error) {
+						require.ErrorContains(tb, actualErr, "stage hooks: extract hooks: waiting for tar command completion: exit status")
+					},
+				},
+			},
+		},
+		{
 			desc: "reapplying custom hooks works",
 			steps: steps{
 				StartManager{
@@ -2912,7 +2932,7 @@ func TestTransactionManager(t *testing.T) {
 				t.Helper()
 
 				transactionManager.Stop()
-				managerRunning, err = checkManagerError(t, managerErr, transactionManager)
+				managerRunning, err = checkManagerError(t, ctx, managerErr, transactionManager)
 				require.NoError(t, err)
 				require.False(t, managerRunning)
 			}
@@ -2972,7 +2992,7 @@ func TestTransactionManager(t *testing.T) {
 					stopManager()
 				case AssertManager:
 					require.True(t, managerRunning, "test error: manager must be running for syncing")
-					managerRunning, err = checkManagerError(t, managerErr, transactionManager)
+					managerRunning, err = checkManagerError(t, ctx, managerErr, transactionManager)
 					require.ErrorIs(t, err, step.ExpectedError)
 				case Begin:
 					require.NotContains(t, openTransactions, step.TransactionID, "test error: transaction id reused in begin")
@@ -3045,7 +3065,17 @@ func TestTransactionManager(t *testing.T) {
 						commitCtx = step.Context
 					}
 
-					require.ErrorIs(t, transaction.Commit(commitCtx), step.ExpectedError)
+					commitErr := transaction.Commit(commitCtx)
+					switch expectedErr := step.ExpectedError.(type) {
+					case func(testing.TB, error):
+						expectedErr(t, commitErr)
+					case error:
+						require.ErrorIs(t, commitErr, expectedErr)
+					case nil:
+						require.NoError(t, commitErr)
+					default:
+						t.Fatalf("unexpected error type: %T", expectedErr)
+					}
 				case AsyncDeletion:
 					require.Contains(t, openTransactions, step.TransactionID, "test error: transaction committed before beginning it")
 
@@ -3088,7 +3118,7 @@ func TestTransactionManager(t *testing.T) {
 			}
 
 			if managerRunning {
-				managerRunning, err = checkManagerError(t, managerErr, transactionManager)
+				managerRunning, err = checkManagerError(t, ctx, managerErr, transactionManager)
 				require.NoError(t, err)
 			}
 
@@ -3139,7 +3169,7 @@ func TestTransactionManager(t *testing.T) {
 	}
 }
 
-func checkManagerError(t *testing.T, managerErrChannel chan error, mgr *TransactionManager) (bool, error) {
+func checkManagerError(t *testing.T, ctx context.Context, managerErrChannel chan error, mgr *TransactionManager) (bool, error) {
 	t.Helper()
 
 	testTransaction := &Transaction{
@@ -3170,6 +3200,19 @@ func checkManagerError(t *testing.T, managerErrChannel chan error, mgr *Transact
 		select {
 		case err := <-testTransaction.result:
 			require.Error(t, err, "test transaction is expected to error out")
+
+			// Begin a transaction to wait until the manager has applied all log entries currently
+			// committed. This ensures the disk state assertions run with all log entries fully applied
+			// to the repository.
+			if tx, err := mgr.Begin(ctx); err != nil {
+				// Since we already verified the manager was running by it processing the test transaction,
+				// the Begin call should succeed. The only expected error would be ErrRepositoryNotFound
+				// if the repository was deleted.
+				require.ErrorIs(t, err, ErrRepositoryNotFound)
+			} else {
+				require.NoError(t, tx.Rollback())
+			}
+
 			return true, nil
 		case managerErr, closeChannel = <-managerErrChannel:
 		}