Skip to content

Commit ded3e5b

Browse files
colinmegillclaude
andauthored
import: give the job daemon a queue DSN its parser can read (#3057)
The detached production import requires its queue login to reach the database with libpq verify-full TLS, so QUEUE_DATABASE_URL carries sslmode=verify-full and a pinned sslrootcert. The importer started the Rust job daemon with that same URL, which the daemon's tokio-postgres parser rejects ("transport refused: unparseable DSN"). The existing tests use a loopback DSN, so they never hit it. Under POLIS_JOBS_TRANSPORT=tls the daemon now gets the same postgresql:// URL with only sslmode (verify-full or require) and sslrootcert removed, which the daemon reproduces: built-in roots off, only the pinned CA trusted, hostname checked, host restricted to an exact allowlist. Every other span is kept byte for byte, and the derived DSN is never logged. Keyword/value DSNs and TLS policy parameters the daemon does not reproduce (client certificates, CRLs, protocol bounds and so on) are refused rather than dropped. The Python-side verify-full requirement is unchanged. Tests: Python unit tests on the exact production DSN shape, encoded keys, odd passwords and the refusals, and a queue-rs test that the derived DSN parses under TLS while the libpq form is refused. The participant-facing request path is not touched. Claude-Session: https://claude.ai/code/session_018NVGBuYCk4EZmiz4csUv9s Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent aac2651 commit ded3e5b

4 files changed

Lines changed: 262 additions & 2 deletions

File tree

‎delphi/polismath/delphi_storage/production_import.py‎

Lines changed: 70 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import sys
2020
import time
2121
import uuid
22+
from urllib.parse import unquote, urlsplit, urlunsplit
2223

2324
from . import codec, legacy_import as legacy, scan_pacing
2425
from .embedding_spool import MAX_DIMS, MAX_SPOOL_BYTES, write_spool
@@ -748,6 +749,65 @@ def specification(bundle, report_ids):
748749
seed=0, config=config, mode='full', memory_bytes=4294967296, work_units=1), inputs=[], max_attempts=3)])
749750

750751

752+
# libpq TLS policy parameters the Rust job daemon does NOT reproduce. Its transport
753+
# verifies the server with verify-full against POLIS_JOBS_CA_FILE plus a hostname and
754+
# exact host-allowlist check and a TLS 1.2 floor (queue-rs jobs/transport.rs), but it
755+
# loads no CRL, no client identity, no GSS policy and no requested min/max protocol
756+
# version. Silently dropping one of these would hand the daemon a weaker policy than
757+
# the import's own connection uses, so the importer refuses a queue URL that carries
758+
# one rather than quietly change it. `sslmode`/`sslrootcert` are handled separately
759+
# (removed, because the daemon reproduces exactly what they ask for); `sslnegotiation`
760+
# is left in place, since the pinned tokio-postgres parser accepts and applies it.
761+
_DAEMON_UNREPRODUCED_TLS_PARAMS = frozenset({
762+
'sslcert', 'sslkey', 'sslpassword', 'sslcrl', 'sslcrldir',
763+
'sslcompression', 'sslsni', 'sslcertmode', 'sslkeylogfile',
764+
'ssl_min_protocol_version', 'ssl_max_protocol_version',
765+
'requiressl', 'gssencmode', 'gsslib', 'krbsrvname',
766+
})
767+
768+
769+
def _daemon_queue_url(dsn):
770+
"""The queue DSN the Rust daemon can parse. Production always uses the URL form
771+
(postgresql://...), which is the only form supported here; a keyword/value DSN is
772+
refused rather than rewritten, because its grammar (quoting, escapes, spaces around
773+
`=`) cannot be edited safely by token splitting. Only the two libpq TLS parameters
774+
the daemon reproduces from its own configuration are removed -- `sslmode` (verify-
775+
full or require) and `sslrootcert`, replaced by its verify-full CA/hostname/allowlist
776+
checks. Any other libpq TLS policy parameter the daemon does not reproduce is
777+
refused, not dropped. Every surviving span (user, the percent-encoded password,
778+
host, port, dbname and each kept query token) is preserved byte for byte; only
779+
parameter names and the `sslmode` value are percent-decoded for comparison, matching
780+
how tokio-postgres reads keys, and nothing is re-encoded. The result still carries
781+
the password: never log it."""
782+
if not dsn.startswith(('postgresql://', 'postgres://')):
783+
raise ValueError('queue database URL must be a postgresql:// URL')
784+
# urlsplit silently deletes a raw tab, newline or carriage return (even inside the
785+
# password or a retained value) and treats a raw '#' as the start of a fragment,
786+
# which would hide query parameters from the checks below. libpq reads all four as
787+
# ordinary bytes, so a caller must percent-encode them (%09/%0A/%0D/%23); the
788+
# encoded forms are passed through and preserved. Refuse the raw bytes instead of
789+
# handing the daemon a different password or a dropped parameter.
790+
if any(ch in dsn for ch in '\t\n\r#'):
791+
raise ValueError('queue database URL must percent-encode tab, newline, carriage return and #')
792+
parts = urlsplit(dsn)
793+
kept = []
794+
for token in parts.query.split('&'):
795+
if not token:
796+
continue
797+
raw_key, _, raw_value = token.partition('=')
798+
key = unquote(raw_key).lower()
799+
if key == 'sslmode':
800+
if unquote(raw_value).lower() in ('verify-full', 'require'):
801+
continue # the daemon forces its own verified TLS instead
802+
raise ValueError('unexpected sslmode in queue database URL')
803+
if key == 'sslrootcert':
804+
continue # reproduced by POLIS_JOBS_CA_FILE
805+
if key in _DAEMON_UNREPRODUCED_TLS_PARAMS:
806+
raise ValueError('queue database URL carries a TLS policy parameter the daemon does not reproduce')
807+
kept.append(token)
808+
return urlunsplit(parts._replace(query='&'.join(kept)))
809+
810+
751811
def import_all(bundles, reports, environment, timeout, scope):
752812
"""Run the actual daemon in a dedicated import environment; never publish."""
753813
import psycopg2
@@ -758,9 +818,17 @@ def import_all(bundles, reports, environment, timeout, scope):
758818
raise ValueError('invalid import scope')
759819
dsn = os.environ['QUEUE_DATABASE_URL']
760820
from psycopg2.extensions import parse_dsn
761-
if os.environ.get('POLIS_JOBS_TRANSPORT', 'tls') != 'loopback' and parse_dsn(dsn).get('sslmode') != 'verify-full':
821+
transport = os.environ.get('POLIS_JOBS_TRANSPORT', 'tls')
822+
if transport != 'loopback' and parse_dsn(dsn).get('sslmode') != 'verify-full':
762823
raise ValueError('verified queue database TLS required')
763-
worker_env = {**os.environ, 'QUEUE_ENV': environment, 'POLIS_JOBS_ENABLED': '1',
824+
# The daemon parses QUEUE_DATABASE_URL with tokio-postgres, which cannot read the
825+
# libpq verify-full/sslrootcert parameters this import requires. Under tls it
826+
# verifies the server from its own POLIS_JOBS_* configuration, so hand it the same
827+
# URL with exactly those two parameters removed (and any TLS policy parameter it
828+
# cannot reproduce refused). Other transports keep the DSN as-is.
829+
queue_url = _daemon_queue_url(dsn) if transport == 'tls' else dsn
830+
worker_env = {**os.environ, 'QUEUE_DATABASE_URL': queue_url,
831+
'QUEUE_ENV': environment, 'POLIS_JOBS_ENABLED': '1',
764832
'POLIS_JOBS_WORKER_CLASS': 'delphi', 'POLIS_JOBS_STAGES': 'graph_narrative',
765833
'POLIS_JOBS_PYTHON': sys.executable, 'DELPHI_APP_PATH': '/app',
766834
'POLIS_JOBS_WORK_DIR': '/import/work', 'POLIS_JOBS_JOURNAL_DIR': '/import/journal'}

‎delphi/scripts/report_dynamo_import.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,8 +67,12 @@
6767
'source identity mapping changed',
6868
'source key schema mismatch',
6969
'source table disappeared during scan',
70+
'queue database URL carries a TLS policy parameter the daemon does not reproduce',
71+
'queue database URL must be a postgresql:// URL',
72+
'queue database URL must percent-encode tab, newline, carriage return and #',
7073
'unapproved missing source table',
7174
'unexpected published import',
75+
'unexpected sslmode in queue database URL',
7276
'unquarantined source row',
7377
'verified queue database TLS required',
7478
'verified source database TLS required',

‎delphi/tests/test_production_import.py‎

Lines changed: 165 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -361,3 +361,168 @@ def test_flag_off_materialize_is_unchanged(tmp_path):
361361
assert 'embeddings_table' not in manifest and 'embeddings' not in manifest
362362
assert manifest['families'][FAMILY]['rows'] == 2
363363
assert codec.decode_family((tmp_path/'off'/plain[0]['digest']/(FAMILY+'.jsonl')).read_bytes())[1] == rows
364+
365+
366+
def test_daemon_queue_url_strips_only_libpq_tls_params():
367+
"""The exact production QUEUE_DATABASE_URL shape: import_worker login, verify-full
368+
and a pinned CA bundle. The daemon DSN drops only the libpq TLS parameters the
369+
Rust parser rejects and keeps the user, password, host, port and dbname."""
370+
from psycopg2.extensions import parse_dsn
371+
# 44-char password, letters+digits only (the genpw alphabet, URL-safe with no encoding).
372+
password = ('aZ09' * 11)
373+
dsn = ('postgresql://import_worker:%s@db.example.internal:5432/polis'
374+
'?sslmode=verify-full&sslrootcert=/ca/global-bundle.pem' % password)
375+
derived = imp._daemon_queue_url(dsn)
376+
assert 'sslmode' not in derived and 'sslrootcert' not in derived
377+
# Every stripped parameter is gone; everything else is byte-for-byte unchanged.
378+
assert derived == 'postgresql://import_worker:%s@db.example.internal:5432/polis' % password
379+
parsed = parse_dsn(derived)
380+
assert parsed['user'] == 'import_worker' and parsed['password'] == password
381+
assert parsed['host'] == 'db.example.internal' and parsed['port'] == '5432'
382+
assert parsed['dbname'] == 'polis'
383+
assert 'sslmode' not in parsed and 'sslrootcert' not in parsed
384+
385+
386+
def test_daemon_queue_url_preserves_percent_encoding_and_other_params():
387+
"""A percent-encoded password byte and a non-TLS parameter tokio-postgres accepts
388+
survive unchanged; only the TLS keys are dropped."""
389+
from psycopg2.extensions import parse_dsn
390+
dsn = ('postgresql://import_worker:p%40ss%2Fword@db.example.internal:5432/polis'
391+
'?sslmode=verify-full&sslrootcert=/ca/global-bundle.pem'
392+
'&application_name=import&connect_timeout=10')
393+
derived = imp._daemon_queue_url(dsn)
394+
assert derived == ('postgresql://import_worker:p%40ss%2Fword@db.example.internal:5432/polis'
395+
'?application_name=import&connect_timeout=10')
396+
parsed = parse_dsn(derived)
397+
assert parsed['password'] == 'p@ss/word' # the percent-encoding round-trips
398+
assert parsed['application_name'] == 'import' and parsed['connect_timeout'] == '10'
399+
400+
401+
def test_daemon_queue_url_odd_password_in_url_form_is_byte_preserved():
402+
"""An odd password in URL form (percent-encoded quotes, spaces and reserved bytes)
403+
is kept byte for byte; only the TLS keys are dropped."""
404+
from psycopg2.extensions import parse_dsn
405+
# Encodes the literal password: 'a b'@=&/
406+
encoded = "%27a%20b%27%40%3D%26%2F"
407+
dsn = ('postgresql://import_worker:%s@db.example.internal:5432/polis'
408+
'?sslmode=verify-full&sslrootcert=/ca/global-bundle.pem' % encoded)
409+
derived = imp._daemon_queue_url(dsn)
410+
assert derived == 'postgresql://import_worker:%s@db.example.internal:5432/polis' % encoded
411+
# The userinfo bytes were never decoded or re-encoded; libpq recovers the original.
412+
assert parse_dsn(derived)['password'] == "'a b'@=&/"
413+
414+
415+
def test_daemon_queue_url_recognizes_percent_encoded_tls_keys():
416+
"""tokio-postgres percent-decodes parameter names, so an encoded sslmode/sslrootcert
417+
key must be recognized and stripped, not passed through as an unknown name."""
418+
# ssl%6Dode -> sslmode ; sslroot%63ert -> sslrootcert
419+
dsn = ('postgresql://import_worker:pw@db.example.internal:5432/polis'
420+
'?ssl%6Dode=verify-full&sslroot%63ert=/ca/global-bundle.pem&application_name=import')
421+
derived = imp._daemon_queue_url(dsn)
422+
assert derived == ('postgresql://import_worker:pw@db.example.internal:5432/polis'
423+
'?application_name=import')
424+
425+
426+
def test_daemon_queue_url_drops_duplicate_tls_params_keeps_tls_text_in_values():
427+
"""Repeated TLS keys are all removed; a non-TLS value that merely looks like a TLS
428+
parameter (even percent-encoded) is left untouched."""
429+
dsn = ('postgresql://import_worker:pw@db.example.internal:5432/polis'
430+
'?sslmode=verify-full&sslrootcert=/ca/a.pem&sslrootcert=/ca/b.pem'
431+
'&application_name=sslmode%3Dverify-full')
432+
derived = imp._daemon_queue_url(dsn)
433+
assert derived == ('postgresql://import_worker:pw@db.example.internal:5432/polis'
434+
'?application_name=sslmode%3Dverify-full')
435+
436+
437+
def test_daemon_queue_url_preserves_sslnegotiation():
438+
"""sslnegotiation is accepted and applied by the pinned tokio-postgres parser, so it
439+
is preserved rather than stripped under the 'remove only what the daemon reproduces'
440+
contract."""
441+
dsn = ('postgresql://import_worker:pw@db.example.internal:5432/polis'
442+
'?sslmode=verify-full&sslrootcert=/ca/global-bundle.pem&sslnegotiation=direct')
443+
derived = imp._daemon_queue_url(dsn)
444+
assert derived == ('postgresql://import_worker:pw@db.example.internal:5432/polis'
445+
'?sslnegotiation=direct')
446+
447+
448+
@pytest.mark.parametrize('dsn', [
449+
'host=db.example.internal port=5432 user=import_worker dbname=polis sslmode=verify-full',
450+
"host=db.example.internal user=import_worker password='a b' sslmode=verify-full",
451+
'user=import_worker dbname=polis sslmode = verify-full sslrootcert = /ca/global-bundle.pem',
452+
])
453+
def test_daemon_queue_url_refuses_keyword_dsns(dsn):
454+
"""Keyword/value DSNs are refused, not rewritten: their grammar cannot be edited
455+
safely by token splitting, and production never uses them."""
456+
with pytest.raises(ValueError, match='postgresql'):
457+
imp._daemon_queue_url(dsn)
458+
459+
460+
@pytest.mark.parametrize('param', [
461+
'sslcrl=/ca/revoked.pem',
462+
'sslcrldir=/ca/crls',
463+
'ssl_min_protocol_version=TLSv1.3',
464+
'ssl_max_protocol_version=TLSv1.3',
465+
'sslcert=/ca/client.pem',
466+
'sslkey=/ca/client.key',
467+
'sslpassword=secret',
468+
'gssencmode=require',
469+
'sslcompression=1',
470+
'ssl_%6Din_protocol_version=TLSv1.3', # encoded 'm' -> ssl_min_protocol_version
471+
])
472+
def test_daemon_queue_url_refuses_unreproduced_tls_params(param):
473+
"""A TLS policy parameter the daemon does not reproduce is refused, never silently
474+
dropped (which would hand the daemon a weaker policy than the import requires)."""
475+
dsn = ('postgresql://import_worker:pw@db.example.internal:5432/polis'
476+
'?sslmode=verify-full&sslrootcert=/ca/global-bundle.pem&' + param)
477+
with pytest.raises(ValueError, match='does not reproduce'):
478+
imp._daemon_queue_url(dsn)
479+
480+
481+
@pytest.mark.parametrize('raw', ['\t', '\n', '\r', '#'], ids=['tab', 'lf', 'cr', 'hash'])
482+
def test_daemon_queue_url_refuses_raw_tab_newline_cr_and_hash_in_password(raw):
483+
"""urlsplit deletes a raw tab/newline/CR and treats a raw '#' as a fragment start,
484+
either of which would change the password libpq sees. Refuse the raw byte and make
485+
the caller percent-encode it."""
486+
dsn = ('postgresql://import_worker:a' + raw + 'b@db.example.internal:5432/polis'
487+
'?sslmode=verify-full&sslrootcert=/ca/global-bundle.pem')
488+
with pytest.raises(ValueError, match='percent-encode'):
489+
imp._daemon_queue_url(dsn)
490+
491+
492+
def test_daemon_queue_url_refuses_raw_hash_that_would_hide_a_policy_param():
493+
"""A raw '#' in a query value would push the parameters after it into urlsplit's
494+
fragment, bypassing the TLS-policy refusal below. Refuse it."""
495+
dsn = ('postgresql://import_worker:pw@db.example.internal:5432/polis'
496+
'?sslmode=verify-full&sslrootcert=/ca/global-bundle.pem'
497+
'&application_name=a#b&ssl_min_protocol_version=TLSv1.3')
498+
with pytest.raises(ValueError, match='percent-encode'):
499+
imp._daemon_queue_url(dsn)
500+
501+
502+
@pytest.mark.parametrize('raw', ['\t', '\n', '\r'], ids=['tab', 'lf', 'cr'])
503+
def test_daemon_queue_url_refuses_raw_whitespace_in_a_retained_value(raw):
504+
"""The same raw-byte refusal applies to a retained (non-TLS) query value, not only
505+
the password."""
506+
dsn = ('postgresql://import_worker:pw@db.example.internal:5432/polis'
507+
'?sslmode=verify-full&sslrootcert=/ca/global-bundle.pem'
508+
'&application_name=a' + raw + 'b')
509+
with pytest.raises(ValueError, match='percent-encode'):
510+
imp._daemon_queue_url(dsn)
511+
512+
513+
def test_daemon_queue_url_preserves_percent_encoded_control_bytes():
514+
"""The percent-encoded counterparts (%09 tab, %0A LF, %0D CR, %23 '#') pass through
515+
untouched in both the password and a retained value, and libpq recovers the literal
516+
bytes."""
517+
from psycopg2.extensions import parse_dsn
518+
pw = 'a%09b%0Ac%0Dd%23e' # tab, LF, CR, '#'
519+
appname = 'x%09y%23z'
520+
dsn = ('postgresql://import_worker:' + pw + '@db.example.internal:5432/polis'
521+
'?sslmode=verify-full&sslrootcert=/ca/global-bundle.pem'
522+
'&application_name=' + appname)
523+
derived = imp._daemon_queue_url(dsn)
524+
assert derived == ('postgresql://import_worker:' + pw + '@db.example.internal:5432/polis'
525+
'?application_name=' + appname)
526+
parsed = parse_dsn(derived)
527+
assert parsed['password'] == 'a\tb\nc\rd#e'
528+
assert parsed['application_name'] == 'x\ty#z'

‎queue-rs/src/jobs/transport.rs‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -200,4 +200,27 @@ mod tests {
200200
assert!(parse("postgresql://u@127.0.0.1/d", &TransportKind::Local).is_err());
201201
assert!(parse("host=relative/dir user=u", &TransportKind::Local).is_err());
202202
}
203+
204+
#[test]
205+
fn tls_accepts_the_importer_derived_dsn_and_rejects_the_libpq_tls_form() {
206+
// The production queue login reaches the database with libpq verify-full and a
207+
// pinned CA bundle; the importer hands this parser the same DSN with those two
208+
// parameters stripped, because the daemon verifies the server itself.
209+
let kind = TransportKind::Tls {
210+
ca_file: "/ca/global-bundle.pem".into(),
211+
host_allowlist: vec!["db.example.internal".into()],
212+
};
213+
let derived = "postgresql://import_worker:aZ09aZ09aZ09@db.example.internal:5432/polis";
214+
let config = parse(derived, &kind);
215+
// It parses, the host is allowed, and TLS is still demanded (verification is
216+
// done by the TLS connector, built-in roots off).
217+
assert!(matches!(
218+
config.map(|c| c.get_ssl_mode()),
219+
Ok(SslMode::Require)
220+
));
221+
// The original libpq form is what tokio-postgres cannot read: it is refused.
222+
let original = "postgresql://import_worker:aZ09aZ09aZ09@db.example.internal:5432/polis\
223+
?sslmode=verify-full&sslrootcert=/ca/global-bundle.pem";
224+
assert!(parse(original, &kind).is_err());
225+
}
203226
}

0 commit comments

Comments
 (0)