e2e mTLS validation

Author: pfcoperez

.env.example

@@ -35,6 +35,11 @@ PG2_USER=postgres
 PG2_PASSWORD=postgres
 PG2_DATABASE=postgres
 
+# Client certificate (mutual TLS) auth test, paths relative to repo root
+PG_MTLS_ROOT_CA_PATH=volumes/pg-tls/server.crt
+PG_MTLS_CLIENT_CERT_PATH=volumes/pg-tls/client.crt
+PG_MTLS_CLIENT_KEY_PATH=volumes/pg-tls/client.key
+
 PEERDB_CATALOG_HOST=host.docker.internal
 PEERDB_CATALOG_PORT=9901
 PEERDB_CATALOG_USER=postgres

.github/workflows/flow.yml

@@ -321,6 +321,10 @@ jobs:
           -c "ALTER SYSTEM SET max_wal_senders=256;"
           -c "ALTER SYSTEM SET max_connections=2048;" &&
           (cat ./nexus/catalog/migrations/V{?,??}__* | docker exec -i "${{ job.services.catalog.id }}" psql -U postgres) &&
+          # Adds support for certificate based client authentication (mTLS) to the catalog postgres instance.
+          docker cp volumes/pg-tls "${{ job.services.catalog.id }}:/tmp/pg-tls" &&
+          docker exec "${{ job.services.catalog.id }}" sh -c 'install -d -o postgres -g postgres /var/lib/postgresql/tls && install -m 644 -o postgres -g postgres /tmp/pg-tls/server.crt /var/lib/postgresql/tls/server.crt && install -m 600 -o postgres -g postgres /tmp/pg-tls/server.key /var/lib/postgresql/tls/server.key && install -m 644 -o postgres -g postgres /tmp/pg-tls/client.crt /var/lib/postgresql/tls/ca.crt' &&
+          docker exec "${{ job.services.catalog.id }}" psql -U postgres -c "ALTER SYSTEM SET ssl=on;" -c "ALTER SYSTEM SET ssl_cert_file='/var/lib/postgresql/tls/server.crt';" -c "ALTER SYSTEM SET ssl_key_file='/var/lib/postgresql/tls/server.key';" -c "ALTER SYSTEM SET ssl_ca_file='/var/lib/postgresql/tls/ca.crt';" &&
           docker restart "${{ job.services.catalog.id }}"
         env:
           PGPASSWORD: postgres
@@ -627,6 +631,10 @@ jobs:
           PG2_USER: postgres
           PG2_PASSWORD: postgres
           PG2_DATABASE: postgres
+          # Mutual-TLS client-certificate auth test (catalog has TLS enabled above)
+          PG_MTLS_ROOT_CA_PATH: ${{ github.workspace }}/volumes/pg-tls/server.crt
+          PG_MTLS_CLIENT_CERT_PATH: ${{ github.workspace }}/volumes/pg-tls/client.crt
+          PG_MTLS_CLIENT_KEY_PATH: ${{ github.workspace }}/volumes/pg-tls/client.key
           PEERDB_SWITCHBOARD_ENABLED: "true"
           PEERDB_QUEUE_FORCE_TOPIC_CREATION: "true"
           ELASTICSEARCH_TEST_ADDRESS: http://localhost:9200

flow/connectors/postgres/postgres_mtls_test.go

@@ -0,0 +1,52 @@
+package connpostgres
+
+import (
+	"testing"
+
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/stretchr/testify/require"
+
+	"github.com/PeerDB-io/peerdb/flow/internal"
+)
+
+// TestPostgresMutualTLSClientCertAuth validates client-certificate (mutual TLS) authentication
+// against a TLS-enabled Postgres whose ssl_ca_file trusts the test client certificate.
+func TestPostgresMutualTLSClientCertAuth(t *testing.T) {
+	t.Parallel()
+
+	if _, ok := internal.GetMutualTLSPostgresConfigFromEnv(); !ok {
+		t.Skip("mutual-TLS Postgres fixtures not configured; " +
+			"set PG_MTLS_CLIENT_CERT_PATH, PG_MTLS_CLIENT_KEY_PATH and PG_MTLS_ROOT_CA_PATH")
+	}
+
+	clientDN := func(t *testing.T, connector *PostgresConnector) pgtype.Text {
+		t.Helper()
+		var dn pgtype.Text
+		require.NoError(t, connector.Conn().QueryRow(t.Context(),
+			"SELECT client_dn FROM pg_stat_ssl WHERE pid = pg_backend_pid()").Scan(&dn))
+		return dn
+	}
+
+	t.Run("client certificate is presented and authenticated", func(t *testing.T) {
+		t.Parallel()
+		cfg, _ := internal.GetMutualTLSPostgresConfigFromEnv()
+		connector, err := NewPostgresConnector(t.Context(), nil, cfg)
+		require.NoError(t, err)
+		defer connector.Close()
+
+		dn := clientDN(t, connector)
+		require.True(t, dn.Valid, "expected the server to record a verified client certificate")
+		require.Contains(t, dn.String, "CN=postgres")
+	})
+
+	t.Run("without a client certificate no client identity is recorded", func(t *testing.T) {
+		t.Parallel()
+		cfg, _ := internal.GetMutualTLSPostgresConfigFromEnv()
+		cfg.ClientTls = nil // Explicit client TLS configuration removal.
+		connector, err := NewPostgresConnector(t.Context(), nil, cfg)
+		require.NoError(t, err)
+		defer connector.Close()
+
+		require.False(t, clientDN(t, connector).Valid, "did not expect a client certificate identity")
+	})
+}

flow/internal/test_env.go

@@ -1,8 +1,10 @@
 package internal
 
 import (
+	"errors"
 	"fmt"
 	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"testing"
@@ -12,15 +14,39 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 )
 
+// repoRoot returns the repository root (the directory containing .env.example) so test fixtures can
+// be referenced by paths relative to it, independent of a test's working directory.
+func repoRoot() (string, error) {
+	dir, err := os.Getwd()
+	if err != nil {
+		return "", err
+	}
+	for {
+		if _, err := os.Stat(filepath.Join(dir, ".env.example")); err == nil {
+			return dir, nil
+		}
+		parent := filepath.Dir(dir)
+		if parent == dir {
+			return "", errors.New("could not locate repository root (.env.example)")
+		}
+		dir = parent
+	}
+}
+
 func loadRootCAFromEnv(envVar string) *string {
 	path := os.Getenv(envVar)
 	if path == "" {
 		return nil
 	}
+	if !filepath.IsAbs(path) {
+		if root, err := repoRoot(); err == nil {
+			path = filepath.Join(root, path)
+		}
+	}
 	// #nosec G703 -- This is test code
 	pem, err := os.ReadFile(path)
 	if err != nil {
-		panic(fmt.Sprintf("%s=%q: failed to read root CA: %v", envVar, path, err))
+		panic(fmt.Sprintf("%s=%q: failed to read PEM file: %v", envVar, path, err))
 	}
 	s := string(pem)
 	return &s
@@ -46,6 +72,31 @@ func GetAncillaryPostgresConfigFromEnv() *protos.PostgresConfig {
 	}
 }
 
+// GetMutualTLSPostgresConfigFromEnv builds a Postgres config that authenticates with a client
+// certificate over TLS. Returns (nil, false) unless PG_MTLS_CLIENT_CERT_PATH,
+// PG_MTLS_CLIENT_KEY_PATH and PG_MTLS_ROOT_CA_PATH are all set.
+func GetMutualTLSPostgresConfigFromEnv() (*protos.PostgresConfig, bool) {
+	clientCert := loadRootCAFromEnv("PG_MTLS_CLIENT_CERT_PATH")
+	clientKey := loadRootCAFromEnv("PG_MTLS_CLIENT_KEY_PATH")
+	rootCA := loadRootCAFromEnv("PG_MTLS_ROOT_CA_PATH")
+	if clientCert == nil || clientKey == nil || rootCA == nil {
+		return nil, false
+	}
+	return &protos.PostgresConfig{
+		Host:       GetEnvString("PG_HOST", "localhost"),
+		Port:       uint32(getEnvUint[uint16]("PG_PORT", 5432)),
+		User:       GetEnvString("PG_USER", "postgres"),
+		Password:   GetEnvString("PG_PASSWORD", "postgres"),
+		Database:   GetEnvString("PG_DATABASE", "postgres"),
+		RequireTls: true,
+		RootCa:     rootCA,
+		ClientTls: &protos.ClientTlsConfig{
+			Certificate: *clientCert,
+			PrivateKey:  *clientKey,
+		},
+	}, true
+}
+
 func GetSecondaryPostgresConfigFromEnv() *protos.PostgresConfig {
 	return &protos.PostgresConfig{
 		Host:       GetEnvString("PG2_HOST", "localhost"),