Skip to content

Commit ed2e231

Browse files
pfcoperezpfcoperez
authored
Enable mTLS for PG sources (#4468)
This Pull Request allows users to add client authentication (PeerDB acting as client here) through mTLS by expanding the PG configuration contract to optionally include the client private certificate and its key. It also includes PeerDB UI changes. Part of: https://linear.app/clickhouse/issue/DBI-793
1 parent ecdbbfd commit ed2e231

22 files changed

Lines changed: 528 additions & 11 deletions

File tree

‎.env.example‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,11 @@ PG2_USER=postgres
4242
PG2_PASSWORD=postgres
4343
PG2_DATABASE=postgres
4444

45+
# Client certificate (mutual TLS) auth test, paths relative to repo root
46+
PG_MTLS_ROOT_CA_PATH=volumes/pg-tls/server.crt
47+
PG_MTLS_CLIENT_CERT_PATH=volumes/pg-tls/client.crt
48+
PG_MTLS_CLIENT_KEY_PATH=volumes/pg-tls/client.key
49+
4550
PEERDB_CATALOG_HOST=host.docker.internal
4651
PEERDB_CATALOG_PORT=9901
4752
PEERDB_CATALOG_USER=postgres

‎.github/workflows/flow.yml‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -323,6 +323,7 @@ jobs:
323323
openssl x509 -req -days 3650 -in ch-certs/client.csr -CA ch-certs/ca.crt -CAkey ch-certs/ca.key -CAcreateserial -out ch-certs/tls.crt
324324
325325
- name: create postgres extensions, increase logical replication limits, and setup catalog database
326+
# Also enables certificate-based client authentication (mTLS) on the catalog postgres instance.
326327
run: >
327328
docker exec "${{ job.services.catalog.id }}" apk add --no-cache build-base git &&
328329
docker exec "${{ job.services.catalog.id }}" git clone --branch v0.8.1 https://github.com/pgvector/pgvector.git /tmp/pgvector &&
@@ -333,6 +334,9 @@ jobs:
333334
-c "ALTER SYSTEM SET max_wal_senders=256;"
334335
-c "ALTER SYSTEM SET max_connections=2048;" &&
335336
(cat ./nexus/catalog/migrations/V{?,??}__* | docker exec -i "${{ job.services.catalog.id }}" psql -U postgres) &&
337+
docker cp volumes/pg-tls "${{ job.services.catalog.id }}:/tmp/pg-tls" &&
338+
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' &&
339+
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';" &&
336340
docker restart "${{ job.services.catalog.id }}"
337341
env:
338342
PGPASSWORD: postgres
@@ -639,6 +643,10 @@ jobs:
639643
PG2_USER: postgres
640644
PG2_PASSWORD: postgres
641645
PG2_DATABASE: postgres
646+
# Mutual-TLS client-certificate auth test (catalog has TLS enabled above)
647+
PG_MTLS_ROOT_CA_PATH: ${{ github.workspace }}/volumes/pg-tls/server.crt
648+
PG_MTLS_CLIENT_CERT_PATH: ${{ github.workspace }}/volumes/pg-tls/client.crt
649+
PG_MTLS_CLIENT_KEY_PATH: ${{ github.workspace }}/volumes/pg-tls/client.key
642650
PEERDB_SWITCHBOARD_ENABLED: "true"
643651
PEERDB_QUEUE_FORCE_TOPIC_CREATION: "true"
644652
ELASTICSEARCH_TEST_ADDRESS: http://localhost:9200

‎ancillary-docker-compose.yml‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,11 +11,13 @@ x-postgres-entrypoint: &postgres-entrypoint
1111
mkdir -p "$$DST"
1212
install -m 644 -o postgres -g postgres "$$SRC/server.crt" "$$DST/server.crt"
1313
install -m 600 -o postgres -g postgres "$$SRC/server.key" "$$DST/server.key"
14+
install -m 644 -o postgres -g postgres "$$SRC/client.crt" "$$DST/ca.crt"
1415
exec docker-entrypoint.sh postgres \
1516
-c ssl=on \
1617
-c ssl_cert_file=$$DST/server.crt \
17-
-c ssl_key_file=$$DST/server.key
18-
18+
-c ssl_key_file=$$DST/server.key \
19+
-c ssl_ca_file=$$DST/ca.crt
20+
#ssl_ca_file is used for client certificate verification
1921
services:
2022
mongodb:
2123
container_name: peerdb-mongodb

‎flow/connectors/mysql/cdc.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -303,6 +303,7 @@ func (c *MySqlConnector) startSyncer(ctx context.Context, env map[string]string)
303303
var err error
304304
tlsConfig, err = common.CreateTlsConfig(
305305
tls.VersionTLS12, c.config.RootCa, c.config.Host, c.config.TlsHost, c.config.SkipCertVerification,
306+
nil,
306307
)
307308
if err != nil {
308309
return nil, err

‎flow/connectors/mysql/mysql.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -187,6 +187,7 @@ func (c *MySqlConnector) connect(ctx context.Context) (*client.Conn, error) {
187187
if !c.config.DisableTls {
188188
config, err := common.CreateTlsConfig(
189189
tls.VersionTLS12, c.config.RootCa, c.config.Host, c.config.TlsHost, c.config.SkipCertVerification,
190+
nil,
190191
)
191192
if err != nil {
192193
return err

‎flow/connectors/postgres/postgres.go‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -186,8 +186,16 @@ func ParseConfig(connectionString string, pgConfig *protos.PostgresConfig) (*pgx
186186
shouldUseTls := internal.PGMustUseTlsConnection(pgConfig)
187187

188188
if shouldUseTls || pgConfig.RootCa != nil {
189+
var clientCert *common.ClientCertificate
190+
if clientTls := pgConfig.GetClientTls(); clientTls != nil {
191+
clientCert, err = common.NewClientCertificate(clientTls.GetCertificate(), clientTls.GetPrivateKey())
192+
if err != nil {
193+
return nil, err
194+
}
195+
}
189196
tlsConfig, err := common.CreateTlsConfig(
190-
tls.VersionTLS12, pgConfig.RootCa, connConfig.Host, pgConfig.TlsHost, pgConfig.SkipCertVerification)
197+
tls.VersionTLS12, pgConfig.RootCa, connConfig.Host, pgConfig.TlsHost, pgConfig.SkipCertVerification,
198+
clientCert)
191199
if err != nil {
192200
return nil, err
193201
}

‎flow/connectors/postgres/postgres_mtls_test.go‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
package connpostgres
2+
3+
import (
4+
"testing"
5+
6+
"github.com/jackc/pgx/v5/pgtype"
7+
"github.com/stretchr/testify/require"
8+
9+
"github.com/PeerDB-io/peerdb/flow/internal"
10+
)
11+
12+
// TestPostgresMutualTLSClientCertAuth validates client-certificate (mutual TLS) authentication
13+
// against a TLS-enabled Postgres whose ssl_ca_file trusts the test client certificate.
14+
func TestPostgresMutualTLSClientCertAuth(t *testing.T) {
15+
t.Parallel()
16+
17+
clientDN := func(t *testing.T, connector *PostgresConnector) pgtype.Text {
18+
t.Helper()
19+
var dn pgtype.Text
20+
require.NoError(t, connector.Conn().QueryRow(t.Context(),
21+
"SELECT client_dn FROM pg_stat_ssl WHERE pid = pg_backend_pid()").Scan(&dn))
22+
return dn
23+
}
24+
25+
t.Run("client certificate is presented and authenticated", func(t *testing.T) {
26+
t.Parallel()
27+
cfg := internal.GetMutualTLSPostgresConfigFromEnv()
28+
connector, err := NewPostgresConnector(t.Context(), nil, cfg)
29+
require.NoError(t, err)
30+
defer connector.Close()
31+
32+
dn := clientDN(t, connector)
33+
require.True(t, dn.Valid, "expected the server to record a verified client certificate")
34+
require.Contains(t, dn.String, "CN=postgres")
35+
})
36+
37+
t.Run("without a client certificate no client identity is recorded", func(t *testing.T) {
38+
t.Parallel()
39+
cfg := internal.GetMutualTLSPostgresConfigFromEnv()
40+
cfg.ClientTls = nil // Explicit client TLS configuration removal.
41+
connector, err := NewPostgresConnector(t.Context(), nil, cfg)
42+
require.NoError(t, err)
43+
defer connector.Close()
44+
45+
require.False(t, clientDN(t, connector).Valid, "did not expect a client certificate identity")
46+
})
47+
}

‎flow/connectors/postgres/postgres_tls_test.go‎

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,127 @@
1+
package connpostgres
2+
3+
import (
4+
"crypto/ecdsa"
5+
"crypto/elliptic"
6+
"crypto/rand"
7+
"crypto/x509"
8+
"crypto/x509/pkix"
9+
"encoding/pem"
10+
"math/big"
11+
"testing"
12+
"time"
13+
14+
"github.com/stretchr/testify/require"
15+
16+
"github.com/PeerDB-io/peerdb/flow/generated/protos"
17+
)
18+
19+
const testTLSConnString = "postgres://user@localhost:5432/testdb?sslmode=require"
20+
21+
// generateClientCertKey returns a self-signed client certificate and its private key, PEM-encoded.
22+
func generateClientCertKey(t *testing.T, commonName string) (string, string) {
23+
t.Helper()
24+
25+
priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
26+
require.NoError(t, err)
27+
28+
template := &x509.Certificate{
29+
SerialNumber: big.NewInt(1),
30+
Subject: pkix.Name{CommonName: commonName},
31+
NotBefore: time.Now().Add(-time.Hour),
32+
NotAfter: time.Now().Add(time.Hour),
33+
KeyUsage: x509.KeyUsageDigitalSignature,
34+
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
35+
}
36+
der, err := x509.CreateCertificate(rand.Reader, template, template, &priv.PublicKey, priv)
37+
require.NoError(t, err)
38+
certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
39+
40+
keyDER, err := x509.MarshalPKCS8PrivateKey(priv)
41+
require.NoError(t, err)
42+
keyPEM := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER})
43+
44+
return string(certPEM), string(keyPEM)
45+
}
46+
47+
func TestParseConfigClientTLS(t *testing.T) {
48+
certPEM, keyPEM := generateClientCertKey(t, "peerdb-client")
49+
50+
t.Run("client cert populates tls.Config.Certificates", func(t *testing.T) {
51+
connConfig, err := ParseConfig(testTLSConnString, &protos.PostgresConfig{
52+
Host: "localhost",
53+
Port: 5432,
54+
User: "user",
55+
Database: "testdb",
56+
RequireTls: true,
57+
ClientTls: &protos.ClientTlsConfig{
58+
Certificate: certPEM,
59+
PrivateKey: keyPEM,
60+
},
61+
})
62+
require.NoError(t, err)
63+
require.NotNil(t, connConfig.TLSConfig)
64+
require.Len(t, connConfig.TLSConfig.Certificates, 1)
65+
66+
leaf, err := x509.ParseCertificate(connConfig.TLSConfig.Certificates[0].Certificate[0])
67+
require.NoError(t, err)
68+
require.Equal(t, "peerdb-client", leaf.Subject.CommonName)
69+
})
70+
71+
t.Run("client cert is not applied when TLS is not enabled", func(t *testing.T) {
72+
connConfig, err := ParseConfig("postgres://user@localhost:5432/testdb", &protos.PostgresConfig{
73+
Host: "localhost",
74+
Port: 5432,
75+
User: "user",
76+
Database: "testdb",
77+
ClientTls: &protos.ClientTlsConfig{
78+
Certificate: certPEM,
79+
PrivateKey: keyPEM,
80+
},
81+
})
82+
require.NoError(t, err)
83+
require.Empty(t, connConfig.TLSConfig.Certificates)
84+
})
85+
86+
t.Run("no client cert leaves Certificates empty", func(t *testing.T) {
87+
connConfig, err := ParseConfig(testTLSConnString, &protos.PostgresConfig{
88+
Host: "localhost",
89+
Port: 5432,
90+
User: "user",
91+
Database: "testdb",
92+
RequireTls: true,
93+
})
94+
require.NoError(t, err)
95+
require.NotNil(t, connConfig.TLSConfig)
96+
require.Empty(t, connConfig.TLSConfig.Certificates)
97+
})
98+
99+
t.Run("invalid client cert/key pair is rejected", func(t *testing.T) {
100+
_, err := ParseConfig(testTLSConnString, &protos.PostgresConfig{
101+
Host: "localhost",
102+
Port: 5432,
103+
User: "user",
104+
Database: "testdb",
105+
RequireTls: true,
106+
ClientTls: &protos.ClientTlsConfig{
107+
Certificate: certPEM,
108+
PrivateKey: "not a valid key",
109+
},
110+
})
111+
require.Error(t, err)
112+
})
113+
114+
t.Run("client cert without private key is rejected", func(t *testing.T) {
115+
_, err := ParseConfig(testTLSConnString, &protos.PostgresConfig{
116+
Host: "localhost",
117+
Port: 5432,
118+
User: "user",
119+
Database: "testdb",
120+
RequireTls: true,
121+
ClientTls: &protos.ClientTlsConfig{
122+
Certificate: certPEM,
123+
},
124+
})
125+
require.Error(t, err)
126+
})
127+
}

‎flow/connectors/utils/aws.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -396,7 +396,7 @@ func CreateS3Client(ctx context.Context, credsProvider AWSCredentialsProvider) (
396396
rootCAs, tlsHost := credsProvider.GetTlsConfig()
397397
if rootCAs != nil || tlsHost != "" {
398398
// start with a clone of DefaultTransport so we keep http2, idle-conns, etc.
399-
tlsConfig, err := common.CreateTlsConfig(tls.VersionTLS13, rootCAs, tlsHost, tlsHost, tlsHost == "")
399+
tlsConfig, err := common.CreateTlsConfig(tls.VersionTLS13, rootCAs, tlsHost, tlsHost, tlsHost == "", nil)
400400
if err != nil {
401401
return nil, err
402402
}

‎flow/e2e/api_test.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1004,6 +1004,8 @@ func (s APITestSuite) TestSchemaEndpoints() {
10041004
switch config := peerInfo.Peer.Config.(type) {
10051005
case *protos.Peer_PostgresConfig:
10061006
require.Equal(s.t, "********", config.PostgresConfig.Password)
1007+
require.Equal(s.t, "********", config.PostgresConfig.ClientTls.Certificate)
1008+
require.Equal(s.t, "********", config.PostgresConfig.ClientTls.PrivateKey)
10071009
case *protos.Peer_MysqlConfig:
10081010
require.Equal(s.t, "********", config.MysqlConfig.Password)
10091011
case *protos.Peer_MongoConfig:

0 commit comments

Comments
 (0)