feat: bind evidence credentials through local files

2026-08-09 19:03:36 +02:00
parent c7a369f436
commit 7126b567b0
11 changed files with 839 additions and 18 deletions
@@ -384,3 +384,20 @@ evidence:
    legacy_source = build_evidence_sources(load_config(legacy))[0]
    assert isinstance(legacy_source, FilesystemEvidenceSource)
    assert legacy_source.root == (tmp_path / "curated").resolve()
+
+
+
+def test_safe_validation_formatter_keeps_location_and_type_without_rejected_input(tmp_path):
+    workspace = tmp_path / "workspace.yaml"
+    workspace.write_text("""
+dwh:
+  type: postgres_direct
+  connection: {database: analytics, schema: public, user: reader}
+""")
+
+    with pytest.raises(ConfigError) as caught:
+        load_config(workspace)
+
+    message = str(caught.value)
+    assert "dwh.postgres_direct.connection.password" in message
+    assert "missing" in message
@@ -0,0 +1,280 @@
+import json
+
+import pytest
+import yaml
+from pydantic import SecretStr
+from typer.testing import CliRunner
+
+from tht.adapters.evidence import HttpManifestEvidenceSource
+from tht.adapters.factory import build_evidence_sources
+from tht.cli import app
+from tht.config import ConfigError, load_config
+
+SIGNED_CANARY = "SIGNED-CANARY-QUERY"
+ACCESS_CANARY = "ACCESS-CANARY"
+SECRET_CANARY = "SECRET-CANARY"
+TOKEN_CANARY = "TOKEN-CANARY"
+ALL_CANARIES = (SIGNED_CANARY, ACCESS_CANARY, SECRET_CANARY, TOKEN_CANARY)
+
+
+def raw_runtime(source, *, vector=None):
+    value = {
+        "dwh": {
+            "type": "postgres_direct",
+            "connection": {
+                "database": "analytics", "schema": "public", "user": "reader",
+                "password": "not-a-canary",
+            },
+        },
+        "evidence": {"sources": [source]},
+    }
+    if vector is not None:
+        value["vector"] = vector
+    return value
+
+
+def write_config(tmp_path, source, *, vector=None):
+    path = tmp_path / "runtime.yaml"
+    path.write_text(yaml.safe_dump(raw_runtime(source, vector=vector)))
+    return path
+
+
+def assert_no_canaries(value):
+    text = str(value)
+    for canary in ALL_CANARIES:
+        assert canary not in text
+
+
+def test_filesystem_config_does_not_touch_a_declared_source_root(tmp_path):
+    missing = tmp_path / "deliberately-missing"
+    cfg = load_config(write_config(tmp_path, {"type": "filesystem", "root": str(missing)}))
+
+    assert cfg.evidence.sources[0].root == missing
+    assert cfg.evidence.sources[0].patterns == ["**/*.md"]
+    assert cfg.evidence.sources[0].max_bytes == 10 * 1024 * 1024
+    assert not missing.exists()
+
+
+def test_public_http_urls_are_secret_typed_without_adapter_construction(tmp_path):
+    cfg = load_config(write_config(tmp_path, {
+        "type": "http", "urls": ["https://evidence.example.test/guide.md"],
+    }))
+    source = cfg.evidence.sources[0]
+
+    assert isinstance(source.urls[0], SecretStr)
+    assert source.transport_urls() == ["https://evidence.example.test/guide.md"]
+    assert source.connect_timeout == 5
+    assert source.read_timeout == 30
+    assert source.max_bytes == 10 * 1024 * 1024
+    assert source.max_redirects == 5
+    assert source.allow_private_hosts is False
+    assert source.max_cache_bytes == 64 * 1024 * 1024
+
+
+def test_signed_http_file_resolves_in_memory_and_preserves_provenance_order(tmp_path):
+    signed = [
+        f"https://evidence.example.test/guide.md?token={SIGNED_CANARY}",
+        "https://evidence.example.test/runbook.md?signature=second",
+    ]
+    secret_file = tmp_path / "signed-urls.json"
+    secret_file.write_text(json.dumps(signed))
+    cfg = load_config(write_config(tmp_path, {
+        "type": "http",
+        "provenance_urls": [
+            "https://evidence.example.test/guide.md",
+            "https://evidence.example.test/runbook.md",
+        ],
+        "signed_urls_file": str(secret_file),
+        "connect_timeout": 7,
+        "read_timeout": 41,
+        "max_bytes": 1234,
+        "max_redirects": 2,
+        "allow_private_hosts": True,
+        "max_cache_bytes": 5678,
+    }))
+    source = cfg.evidence.sources[0]
+
+    assert all(isinstance(url, SecretStr) for url in source.urls)
+    assert source.transport_urls() == signed
+    assert source.provenance_urls == [
+        "https://evidence.example.test/guide.md",
+        "https://evidence.example.test/runbook.md",
+    ]
+    assert (source.connect_timeout, source.read_timeout) == (7, 41)
+    assert (source.max_bytes, source.max_redirects, source.max_cache_bytes) == (1234, 2, 5678)
+    assert source.allow_private_hosts is True
+    assert "signed_urls_file" not in repr(source)
+    assert_no_canaries(repr(cfg))
+    assert_no_canaries(cfg.model_dump_json())
+
+    adapter = build_evidence_sources(cfg)[0]
+    assert isinstance(adapter, HttpManifestEvidenceSource)
+    assert_no_canaries(repr(adapter))
+
+
+@pytest.mark.parametrize("contents", [
+    "{malformed", json.dumps({"url": "https://evidence.example.test/guide.md"}),
+    json.dumps([]), json.dumps(["https://evidence.example.test/guide.md", 3]),
+])
+def test_signed_http_rejects_malformed_non_list_empty_or_non_string_files(tmp_path, contents):
+    secret_file = tmp_path / "signed-urls.json"
+    secret_file.write_text(contents)
+    path = write_config(tmp_path, {
+        "type": "http",
+        "provenance_urls": ["https://evidence.example.test/guide.md"],
+        "signed_urls_file": str(secret_file),
+    })
+
+    with pytest.raises(ConfigError) as caught:
+        load_config(path)
+    assert "signed URL file" in str(caught.value)
+    assert_no_canaries(caught.value)
+
+
+def test_signed_http_rejects_missing_and_oversized_files_without_disclosure(tmp_path):
+    missing = tmp_path / "missing.json"
+    path = write_config(tmp_path, {
+        "type": "http",
+        "provenance_urls": ["https://evidence.example.test/guide.md"],
+        "signed_urls_file": str(missing),
+    })
+    with pytest.raises(ConfigError, match="signed URL file"):
+        load_config(path)
+
+    oversized = tmp_path / "oversized.json"
+    oversized.write_bytes(b"x" * (1024 * 1024 + 1))
+    path = write_config(tmp_path, {
+        "type": "http",
+        "provenance_urls": ["https://evidence.example.test/guide.md"],
+        "signed_urls_file": str(oversized),
+    })
+    with pytest.raises(ConfigError, match="signed URL file") as caught:
+        load_config(path)
+    assert_no_canaries(caught.value)
+
+
+@pytest.mark.parametrize("provenance,signed", [
+    (
+        ["https://evidence.example.test/a.md", "https://evidence.example.test/b.md"],
+        ["https://evidence.example.test/b.md?sig=1", "https://evidence.example.test/a.md?sig=2"],
+    ),
+    (["https://evidence.example.test/a.md"], [
+        "https://evidence.example.test/a.md?sig=1", "https://evidence.example.test/b.md?sig=2",
+    ]),
+    (["https://evidence.example.test/a.md"], ["https://evidence.example.test/b.md"]),
+    (["https://evidence.example.test/a.md"], [f"https://user:{SIGNED_CANARY}@evidence.example.test/a.md"]),
+    (
+        ["https://evidence.example.test/a.md", "https://evidence.example.test/a.md"],
+        ["https://evidence.example.test/a.md?sig=1", "https://evidence.example.test/a.md?sig=2"],
+    ),
+])
+def test_signed_http_rejects_reordered_extra_mismatch_userinfo_and_duplicate_provenance(
+    tmp_path, provenance, signed,
+):
+    secret_file = tmp_path / "signed-urls.json"
+    secret_file.write_text(json.dumps(signed))
+    path = write_config(tmp_path, {
+        "type": "http", "provenance_urls": provenance, "signed_urls_file": str(secret_file),
+    })
+
+    with pytest.raises(ConfigError) as caught:
+        load_config(path)
+    assert "evidence.sources.0" in str(caught.value)
+    assert_no_canaries(caught.value)
+
+
+def test_s3_ambient_and_static_file_credentials_are_secret_typed(tmp_path):
+    ambient = load_config(write_config(tmp_path, {
+        "type": "s3", "bucket": "clinical-evidence", "prefix": "published/",
+    })).evidence.sources[0]
+    assert ambient.access_key is None
+    assert ambient.secret_key is None
+    assert ambient.session_token is None
+    assert ambient.max_bytes == 10 * 1024 * 1024
+    assert ambient.max_objects == 10_000
+    assert ambient.max_pages == 100
+    assert ambient.page_size == 1000
+
+    files = {}
+    for name, canary in [
+        ("access_key", ACCESS_CANARY), ("secret_key", SECRET_CANARY),
+        ("session_token", TOKEN_CANARY),
+    ]:
+        path = tmp_path / name
+        path.write_text(canary)
+        files[f"{name}_file"] = str(path)
+    cfg = load_config(write_config(tmp_path, {
+        "type": "s3", "bucket": "clinical-evidence", "prefix": "published/",
+        **files,
+        "endpoint_url": "https://s3.example.test",
+        "region": "eu-west-1",
+        "trusted_endpoint": True,
+        "allow_private_endpoint": True,
+        "allow_insecure_endpoint": False,
+        "max_bytes": 222,
+        "max_objects": 33,
+        "max_pages": 4,
+        "page_size": 5,
+    }))
+    source = cfg.evidence.sources[0]
+    assert all(isinstance(value, SecretStr) for value in (
+        source.access_key, source.secret_key, source.session_token,
+    ))
+    assert (source.max_bytes, source.max_objects, source.max_pages, source.page_size) == (222, 33, 4, 5)
+    assert_no_canaries(repr(cfg))
+    assert_no_canaries(cfg.model_dump_json())
+
+
+def test_evidence_policy_defaults_non_defaults_and_unknown_keys(tmp_path):
+    default = load_config(write_config(tmp_path, {
+        "type": "filesystem", "root": str(tmp_path / "missing"),
+    }))
+    assert default.vector.max_chunk_chars == 4000
+    assert default.vector.retain_published_generations == 3
+
+    explicit = load_config(write_config(tmp_path, {
+        "type": "filesystem", "root": str(tmp_path / "missing"),
+        "patterns": ["docs/*.md"], "max_bytes": 99,
+    }, vector={"max_chunk_chars": 123, "retain_published_generations": 7}))
+    assert explicit.evidence.sources[0].patterns == ["docs/*.md"]
+    assert explicit.vector.max_chunk_chars == 123
+    assert explicit.vector.retain_published_generations == 7
+
+    for mutation in [
+        {"type": "filesystem", "root": str(tmp_path), "unknown": SIGNED_CANARY},
+        {"type": "http", "urls": ["https://evidence.example.test/a"], "unknown": SIGNED_CANARY},
+        {"type": "s3", "bucket": "bucket-name", "unknown": SIGNED_CANARY},
+    ]:
+        with pytest.raises(ConfigError) as caught:
+            load_config(write_config(tmp_path, mutation))
+        assert "extra_forbidden" in str(caught.value)
+        assert_no_canaries(caught.value)
+
+
+def test_validation_repr_cli_and_exception_output_never_disclose_transport_secrets(tmp_path):
+    secret_file = tmp_path / "signed-urls.json"
+    secret_file.write_text(json.dumps([
+        f"https://evidence.example.test/other.md?token={SIGNED_CANARY}",
+    ]))
+    path = write_config(tmp_path, {
+        "type": "http",
+        "provenance_urls": ["https://evidence.example.test/guide.md"],
+        "signed_urls_file": str(secret_file),
+    })
+    with pytest.raises(ConfigError) as caught:
+        load_config(path)
+    assert_no_canaries(caught.value)
+
+    valid_file = tmp_path / "valid-signed-urls.json"
+    valid_file.write_text(json.dumps([
+        f"https://evidence.example.test/guide.md?token={SIGNED_CANARY}",
+    ]))
+    valid = write_config(tmp_path, {
+        "type": "http",
+        "provenance_urls": ["https://evidence.example.test/guide.md"],
+        "signed_urls_file": str(valid_file),
+    })
+    result = CliRunner().invoke(app, ["config", "check", "--config", str(valid)])
+    assert result.exit_code == 0
+    assert_no_canaries(result.stdout)
+    assert_no_canaries(result.stderr)