import json import traceback 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)) def test_signed_http_file_requires_explicit_provenance_urls(tmp_path): secret_file = tmp_path / "signed-urls.json" secret_file.write_text(json.dumps([ f"https://evidence.example.test/guide.md?token={SIGNED_CANARY}", ])) path = write_config(tmp_path, { "type": "http", "signed_urls_file": str(secret_file), }) with pytest.raises(ConfigError) as caught: load_config(path) assert "provenance" in str(caught.value).lower() assert_no_canaries(caught.value) def test_signed_http_file_rejects_explicit_null_provenance(tmp_path): secret_file = tmp_path / "signed-urls.json" secret_file.write_text(json.dumps(["https://evidence.example.test/guide.md"])) path = write_config(tmp_path, { "type": "http", "provenance_urls": None, "signed_urls_file": str(secret_file), }) with pytest.raises(ConfigError) as caught: load_config(path) assert "provenance" in str(caught.value).lower() def test_public_http_rejects_inline_query_bearing_transport_urls(tmp_path): path = write_config(tmp_path, { "type": "http", "urls": [f"https://evidence.example.test/guide.md?token={SIGNED_CANARY}"], }) with pytest.raises(ConfigError) as caught: load_config(path) assert "evidence.sources.0" in str(caught.value) assert_no_canaries(caught.value) @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_rejects_inline_credentials_and_never_discloses_them(tmp_path): path = write_config(tmp_path, { "type": "s3", "bucket": "clinical-evidence", "access_key": ACCESS_CANARY, "secret_key": SECRET_CANARY, "session_token": TOKEN_CANARY, }) with pytest.raises(ConfigError) as caught: load_config(path) assert "file" in str(caught.value).lower() assert_no_canaries(caught.value) assert_no_canaries("".join(traceback.format_exception(caught.value))) def test_s3_scalar_secret_files_are_bounded(tmp_path): access = tmp_path / "oversized-access-key" access.write_bytes(b"A" * (64 * 1024 + 1)) secret = tmp_path / "secret-key" secret.write_text("bounded-secret") path = write_config(tmp_path, { "type": "s3", "bucket": "clinical-evidence", "access_key_file": str(access), "secret_key_file": str(secret), }) with pytest.raises(ConfigError) as caught: load_config(path) assert "secret file" in str(caught.value).lower() assert "bounded-secret" not in str(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) assert_no_canaries("".join(traceback.format_exception(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)