tegwick
b181240b35
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a137ca-5c77-7a71-b9ae-23ad9b0b5684
188 lines
8.7 KiB
Python
188 lines
8.7 KiB
Python
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"})
|