fix: pass TLS parameters to all external connection points in modular mode

- Add setup_pg_ssl_env() to export libpq PGSSL* env vars; all child psql/pg_dump/pg_isready commands inherit TLS automatically

- Move TLS client key permission fix before external PG connections (was running after, defeating the fix)

- Enhance key fix to trigger on ownership mismatch (root-owned 0600 key unreadable by app user)

- Extract key fix to shared script (docker/init/00-fix-pg-ssl-key.sh) sourced by both entrypoints

- Default POSTGRES_PASSWORD to empty in modular mode (cert-only auth sends no spurious password)

- Propagate TLS OPTIONS to backup pg_dump/pg_restore subprocess calls via _get_pg_env()

- Pass SSL kwargs to dropdb management command's psycopg2.connect()

- Add REDIS_SSL_PARAMS to VOD proxy Redis connections missed in original TLS PR

- Add 8-scenario TLS integration test suite (PG mTLS, Redis TLS, verify-full, key fix, Celery, regression)

- Add 6 unit tests for _get_pg_env() and dropdb TLS parameter passing
This commit is contained in:
None 2026-03-30 20:19:27 -05:00
parent 344ff78c79
commit 763f42cfff
10 changed files with 1179 additions and 44 deletions

View file

@ -15,6 +15,13 @@ class Command(BaseCommand):
host = db_settings.get('HOST', 'localhost')
port = db_settings.get('PORT', 5432)
# Read TLS parameters from Django OPTIONS (populated when POSTGRES_SSL=true)
db_options = db_settings.get('OPTIONS', {})
ssl_kwargs = {}
for key in ('sslmode', 'sslrootcert', 'sslcert', 'sslkey'):
if key in db_options:
ssl_kwargs[key] = db_options[key]
self.stdout.write(self.style.WARNING(
f"WARNING: This will irreversibly drop the entire database '{db_name}'!"
))
@ -30,7 +37,7 @@ class Command(BaseCommand):
maintenance_db = 'postgres'
try:
self.stdout.write("Connecting to maintenance database...")
conn = psycopg2.connect(dbname=maintenance_db, user=user, password=password, host=host, port=port)
conn = psycopg2.connect(dbname=maintenance_db, user=user, password=password, host=host, port=port, **ssl_kwargs)
conn.autocommit = True
cur = conn.cursor()
self.stdout.write(f"Dropping database '{db_name}'...")

View file

@ -1,3 +1,5 @@
from unittest.mock import patch, MagicMock
from django.test import TestCase
from core.models import CoreSettings, DVR_SETTINGS_KEY, EPG_SETTINGS_KEY
@ -148,3 +150,74 @@ class EpgIgnoreListsTest(TestCase):
]:
self._set_epg_field_raw(field, "not a list")
self.assertEqual(getter(), [])
class DropDBCommandTlsTest(TestCase):
"""Verify dropdb management command passes TLS parameters to psycopg2."""
databases = []
_DB_WITH_TLS = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': 'testdb',
'USER': 'testuser',
'PASSWORD': 'testpass',
'HOST': 'localhost',
'PORT': 5432,
'OPTIONS': {
'sslmode': 'verify-full',
'sslrootcert': '/certs/ca.crt',
'sslcert': '/certs/client.crt',
'sslkey': '/certs/client.key',
},
}
}
_DB_NO_TLS = {
'default': {
'ENGINE': 'django.db.backends.postgresql',
'NAME': 'testdb',
'USER': 'testuser',
'PASSWORD': 'testpass',
'HOST': 'localhost',
'PORT': 5432,
}
}
@patch('core.management.commands.dropdb.psycopg2.connect')
@patch('core.management.commands.dropdb.connection')
@patch('builtins.input', return_value='yes')
def test_dropdb_passes_ssl_kwargs_when_tls_enabled(self, _inp, _conn, mock_connect):
mock_pg = MagicMock()
mock_connect.return_value = mock_pg
mock_pg.cursor.return_value = MagicMock()
with self.settings(DATABASES=self._DB_WITH_TLS):
from django.core.management import call_command
call_command('dropdb')
mock_connect.assert_called_once_with(
dbname='postgres', user='testuser', password='testpass',
host='localhost', port=5432,
sslmode='verify-full',
sslrootcert='/certs/ca.crt',
sslcert='/certs/client.crt',
sslkey='/certs/client.key',
)
@patch('core.management.commands.dropdb.psycopg2.connect')
@patch('core.management.commands.dropdb.connection')
@patch('builtins.input', return_value='yes')
def test_dropdb_no_ssl_kwargs_when_tls_disabled(self, _inp, _conn, mock_connect):
mock_pg = MagicMock()
mock_connect.return_value = mock_pg
mock_pg.cursor.return_value = MagicMock()
with self.settings(DATABASES=self._DB_NO_TLS):
from django.core.management import call_command
call_command('dropdb')
mock_connect.assert_called_once_with(
dbname='postgres', user='testuser', password='testpass',
host='localhost', port=5432,
)