import hashlib import io import json from email.message import Message from urllib.request import HTTPSHandler, build_opener from urllib.response import addinfourl import pytest import yaml from reuse_surface import federation, private_source_verify as verifier def _digest(text): return hashlib.sha256(text.encode()).hexdigest() @pytest.fixture def packets(tmp_path, monkeypatch): """Fake owner-consented bodies; no estate credential or network transport.""" approved, sources, bodies = [], [], {} for repo in sorted(verifier.REPOSITORIES): cap_id = f"capability.fixture.{repo}" commit = "a" * 40 path = f"registry/capabilities/{cap_id}.md" metadata = { "id": cap_id, "name": "Fixture", "summary": "Fixture metadata", "owner": repo, "status": "draft", "maturity": {"discovery": {"current": "D3", "target": "D5"}, "availability": {"current": "A2", "target": "A5"}}, "external_evidence": {"completeness": {"level": "C1"}, "reliability": {"level": "R0"}}, } entry = "---\n" + yaml.safe_dump(metadata) + "---\nPrivate fixture body\n" index = yaml.safe_dump({"capabilities": [{ "id": cap_id, "name": "Fixture", "summary": "Fixture metadata", "vector": "D3 / A2 / C1 / R0", "domain": "infotech", "status": "draft", "owner": repo, "path": path, }]}) index_url = verifier._raw_url(repo, commit, "registry/indexes/capabilities.yaml") bodies[index_url] = index bodies[verifier._raw_url(repo, commit, path)] = entry sources.append({"repo": repo, "capability_ids": [cap_id], "consented_commit": commit, "native_raw_url": index_url, "index_sha256": _digest(index), "auth_env": verifier.AUTH_ENV}) approved.append({"repo": repo, "capability_ids": [cap_id], "commit": commit, "files": [ {"path": "registry/indexes/capabilities.yaml", "sha256": _digest(index)}, {"path": path, "sha256": _digest(entry)}, ]}) disclosure = json.dumps({"id": "fixture", "revision": 1, "sources": approved}) consent = json.dumps({"consent_status": "approved", "proposal_sha256": _digest(disclosure), "proposal_id": "fixture", "proposal_revision": 1, "repositories": sorted(verifier.REPOSITORIES)}) proposal = json.dumps({"sources": sources}) paths = [] for name, text in (("proposal", proposal), ("disclosure", disclosure), ("consent", consent)): path = tmp_path / f"{name}.json" path.write_text(text) paths.append(path) monkeypatch.setattr(verifier, f"{name.upper()}_SHA256", _digest(text)) return paths, bodies def _transport(monkeypatch, bodies, mutate=None): calls = [] class FakeResponse: def __init__(self, body): self.body = body def __enter__(self): return self def __exit__(self, *args): return False def read(self): return self.body.encode() class FakeOpener: def open(self, req, timeout): assert req.get_header("Authorization") == "Bearer fixture-secret" assert timeout == 30 calls.append(req.full_url) body = bodies[req.full_url] return FakeResponse(mutate(req.full_url, body) if mutate else body) monkeypatch.setenv(verifier.AUTH_ENV, "fixture-secret") monkeypatch.setenv("REUSE_SURFACE_AUTH_SOURCE_ALLOWLIST", json.dumps({verifier.AUTH_ENV: list(bodies)})) monkeypatch.setattr(federation.urllib.request, "build_opener", lambda handler: FakeOpener()) # Deliberately prohibit composition/cache resolution; direct fetch must work. monkeypatch.setattr(federation, "resolve_source_index_path", lambda *a, **k: pytest.fail("cache path used")) return calls def test_exact_consented_native_index_and_entries_no_cache(packets, monkeypatch): paths, bodies = packets calls = _transport(monkeypatch, bodies) receipt = verifier.verify(*paths) assert receipt["all_verified"] assert not receipt["cache_used"] assert len(calls) == 14 assert all("?ref=" + "a" * 40 in url for url in calls) output = json.dumps(receipt) assert "fixture-secret" not in output assert "Private fixture body" not in output @pytest.mark.parametrize("suffix", [".yaml", ".md"]) def test_changed_unconsented_body_fails(packets, monkeypatch, suffix): paths, bodies = packets _transport(monkeypatch, bodies, lambda url, text: text + "changed" if suffix in url else text) receipt = verifier.verify(*paths) assert not receipt["all_verified"] assert all(not row["verified"] for row in receipt["sources"]) def test_packet_tampering_rejected_before_fetch(packets, monkeypatch): paths, _ = packets paths[0].write_text(paths[0].read_text() + " ") monkeypatch.setattr(federation, "fetch_remote_index_text", lambda *a: pytest.fail("network before consent checks")) assert verifier.verify(*paths)["error"] == "packet_scope_or_schema_validation_failed" def test_missing_auth_fails_before_transport(packets, monkeypatch): paths, bodies = packets monkeypatch.setenv("REUSE_SURFACE_AUTH_SOURCE_ALLOWLIST", json.dumps({verifier.AUTH_ENV: list(bodies)})) monkeypatch.delenv(verifier.AUTH_ENV, raising=False) monkeypatch.setattr(federation.urllib.request, "build_opener", lambda *a: pytest.fail("transport without auth")) receipt = verifier.verify(*paths) assert not receipt["all_verified"] assert all(not row["verified"] for row in receipt["sources"]) def test_error_output_never_embeds_secret_or_body(packets, monkeypatch): paths, bodies = packets monkeypatch.setenv("REUSE_SURFACE_AUTH_SOURCE_ALLOWLIST", json.dumps({verifier.AUTH_ENV: list(bodies)})) def failure(*args): raise ValueError("fixture-secret Private fixture body") monkeypatch.setattr(federation, "fetch_remote_index_text", failure) output = json.dumps(verifier.verify(*paths)) assert "fixture-secret" not in output assert "Private fixture body" not in output def test_broadened_runtime_reader_scope_fails_before_fetch(packets, monkeypatch): paths, bodies = packets monkeypatch.setenv("REUSE_SURFACE_AUTH_SOURCE_ALLOWLIST", json.dumps({verifier.AUTH_ENV: list(bodies) + ["https://attacker.invalid/"]})) monkeypatch.setattr(federation, "fetch_remote_index_text", lambda *a: pytest.fail("broadened scope used")) assert not verifier.verify(*paths)["all_verified"] @pytest.mark.parametrize("target", ["https://attacker.invalid/stolen", "https://forgejo.coulomb.social/other/repo"]) def test_native_auth_redirect_never_replays_header(monkeypatch, target): calls = [] class FakeHTTPS(HTTPSHandler): def https_open(self, req): calls.append(req.full_url) headers = Message() headers["Location"] = target result = addinfourl(io.BytesIO(b""), headers, req.full_url, 302) result.msg = "Found" return result # Use real urllib opener/redirect/error processors and fake HTTPS transport. opener = build_opener(FakeHTTPS(), federation._RejectAuthenticatedRedirect()) monkeypatch.setattr(federation.urllib.request, "build_opener", lambda *a: opener) monkeypatch.setenv("FIXTURE_AUTH", "fixture-secret") monkeypatch.setenv("REUSE_SURFACE_AUTH_SOURCE_ALLOWLIST", json.dumps({"FIXTURE_AUTH": ["https://forgejo.coulomb.social/approved"]})) with pytest.raises(ConnectionError, match="redirect refused"): federation.fetch_remote_index_text("https://forgejo.coulomb.social/approved", {"repo": "fixture", "auth_env": "FIXTURE_AUTH"}) assert calls == ["https://forgejo.coulomb.social/approved"] def test_authenticated_http_rejected_before_transport(monkeypatch): monkeypatch.setenv("FIXTURE_AUTH", "fixture-secret") monkeypatch.setattr(federation.urllib.request, "build_opener", lambda *a: pytest.fail("cleartext transport")) with pytest.raises(ValueError, match="HTTPS"): federation.fetch_remote_index_text("http://forgejo.coulomb.social/approved", {"repo": "fixture", "auth_env": "FIXTURE_AUTH"}) @pytest.mark.parametrize("config", [None, "broken-json", "[]", '{"FIXTURE_AUTH":"https://attacker.invalid/"}', '{"OTHER_AUTH":["https://attacker.invalid/"]}']) def test_unapproved_auth_target_rejected_before_credential_lookup(monkeypatch, config): if config is None: monkeypatch.delenv("REUSE_SURFACE_AUTH_SOURCE_ALLOWLIST", raising=False) else: monkeypatch.setenv("REUSE_SURFACE_AUTH_SOURCE_ALLOWLIST", config) monkeypatch.setattr(federation, "_auth_headers", lambda *a: pytest.fail("unauthorized credential lookup")) with pytest.raises(ValueError, match="runtime-authorized"): federation.fetch_remote_index_text("https://attacker.invalid/", {"repo": "fixture", "auth_env": "FIXTURE_AUTH"})