158 lines
8.2 KiB
Python
158 lines
8.2 KiB
Python
"""Operator-side discovery reports new evidence without making it installable."""
|
|
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
MODULE = Path(__file__).resolve().parents[1] / "watch-upstream-releases.py"
|
|
spec = importlib.util.spec_from_file_location("upstream_watch_test", MODULE)
|
|
watch = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(watch)
|
|
|
|
|
|
def pin(name, kind="github", repo=None):
|
|
return SimpleNamespace(app_id=name, upstream={"kind": kind, "repo": repo or f"team/{name}"})
|
|
|
|
|
|
class UpstreamWatchTest(unittest.TestCase):
|
|
def test_new_release_and_published_advisory_report_once_without_installable_update(self):
|
|
pins = [pin("one"), pin("internal", "internal")]
|
|
def release(_pin, _offline):
|
|
return {"status": "BEHIND", "latest": "v1.2.3", "shipped": "v1.2.2"}
|
|
def advisory(_repo):
|
|
return {"GHSA-abcd-1234-efgh"}
|
|
first, notices = watch.discover(pins, {}, 4, release, advisory)
|
|
self.assertEqual([notice["type"] for notice in notices], ["upstream_release", "security_advisory"])
|
|
self.assertTrue(all(notice["action"] == "review_only" for notice in notices))
|
|
self.assertEqual(first["pending_review"], notices)
|
|
self.assertNotIn("internal", first["apps"])
|
|
second, notices = watch.discover(pins, first, 4, release, advisory)
|
|
self.assertEqual(notices, [])
|
|
self.assertEqual(second["apps"], first["apps"])
|
|
self.assertEqual(second["pending_review"], first["pending_review"])
|
|
|
|
def test_bounded_rotation_and_unavailable_advisory_preserve_last_seen(self):
|
|
pins = [pin("a"), pin("b"), pin("c")]
|
|
def release(_pin, _offline):
|
|
return {"status": "CURRENT", "latest": "1.0.0", "shipped": "1.0.0"}
|
|
first, _ = watch.discover(pins, {}, 1, release, lambda _: {"GHSA-abcd-1234-efgh"})
|
|
self.assertEqual(list(first["apps"]), ["a"])
|
|
second, _ = watch.discover(pins, first, 1, release, lambda _: set())
|
|
self.assertEqual(list(second["apps"]), ["a", "b"])
|
|
def unavailable(_repo):
|
|
raise OSError("rate limited")
|
|
rotated, notices = watch.discover(pins, {**second, "cursor": 0}, 1, release, unavailable)
|
|
self.assertEqual(notices, [])
|
|
self.assertEqual(rotated["apps"]["a"]["advisories"], ["GHSA-abcd-1234-efgh"])
|
|
|
|
def test_opaque_release_is_review_only_and_invalid_repositories_are_skipped(self):
|
|
pins = [pin("one", repo="team/one"), pin("bad", repo="evil/../../local")]
|
|
state, notices = watch.discover(pins, {}, 4,
|
|
lambda *_: {"status": "UNCOMPARABLE", "latest": "build-123", "shipped": "build-122"},
|
|
lambda _: set())
|
|
self.assertEqual(notices, [{"type": "upstream_release", "app": "one", "release": "build-123",
|
|
"shipped": "build-122", "status": "UNCOMPARABLE", "action": "review_only"}])
|
|
self.assertNotIn("bad", state["apps"])
|
|
|
|
def test_state_is_atomic_private_and_restorable(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "operator" / "state.json"
|
|
state = {"version": 1, "cursor": 1, "apps": {"one": {"release": "v2"}}}
|
|
watch.write_state(path, state)
|
|
self.assertEqual(json.loads(path.read_text()), state)
|
|
self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600)
|
|
self.assertEqual(stat.S_IMODE(path.parent.stat().st_mode), 0o700)
|
|
self.assertFalse(list(path.parent.glob(".upstream-*")))
|
|
|
|
def test_public_advisory_parser_rejects_unpublished_and_untrusted_ids(self):
|
|
with patch.object(watch.checker, "http_json", return_value=[
|
|
{"ghsa_id": "GHSA-abcd-1234-efgh", "state": "published"},
|
|
{"ghsa_id": "GHSA-other-1234-efgh", "state": "draft"},
|
|
{"ghsa_id": "../../private", "state": "published"},
|
|
]):
|
|
self.assertEqual(watch.published_github_advisories("team/project"), {"GHSA-abcd-1234-efgh"})
|
|
|
|
def test_offline_scan_never_fetches_release_or_advisory(self):
|
|
seen = []
|
|
def check(_pin, offline):
|
|
seen.append(offline)
|
|
return {"status": "SKIPPED", "latest": "", "shipped": "1.0.0"}
|
|
state, notices = watch.discover([pin("one")], {}, 4, check,
|
|
lambda _: self.fail("offline advisory fetch"), offline=True)
|
|
self.assertEqual(seen, [True])
|
|
self.assertEqual(notices, [])
|
|
self.assertEqual(state["cursor"], 0)
|
|
|
|
def test_authenticated_full_cycle_covers_all_58_in_one_run(self):
|
|
pins = [pin(f"app-{index:02}") for index in range(58)]
|
|
seen = []
|
|
def release(item, offline):
|
|
self.assertFalse(offline)
|
|
seen.append(item.app_id)
|
|
return {"status": "CURRENT", "latest": "1.0.0", "shipped": "1.0.0"}
|
|
state, notices = watch.discover(pins, {}, 100, release, lambda _: set())
|
|
self.assertEqual(len(seen), 58)
|
|
self.assertEqual(set(seen), {item.app_id for item in pins})
|
|
self.assertEqual(state["cursor"], 0)
|
|
self.assertEqual(len(state["apps"]), 58)
|
|
self.assertEqual(notices, [])
|
|
|
|
def test_rate_limit_preserves_prior_observation_and_marks_degraded(self):
|
|
previous = {"apps": {"one": {"release": "1.0.0", "advisories": ["GHSA-abcd-1234-efgh"]}}}
|
|
def unavailable(_repo):
|
|
raise watch.checker.RateLimited("rate limited")
|
|
state, notices = watch.discover([pin("one")], previous, 100,
|
|
lambda *_: {"status": "RATE-LIMITED", "latest": "", "shipped": "1.0.0"}, unavailable)
|
|
self.assertEqual(state["apps"]["one"], previous["apps"]["one"])
|
|
self.assertEqual(notices, [])
|
|
self.assertEqual({row["reason"] for row in state["degraded_sources"]},
|
|
{"release_source_unavailable", "advisory_source_unavailable"})
|
|
|
|
def test_default_mode_is_explicitly_degraded_without_token_and_full_with_token(self):
|
|
pins = [pin(f"app-{index:02}") for index in range(58)]
|
|
def exercise(token):
|
|
seen = []
|
|
def fake_discover(_pins, _previous, max_apps, **_kwargs):
|
|
seen.append(max_apps)
|
|
return {"pending_review": [], "degraded_sources": []}, []
|
|
with tempfile.TemporaryDirectory() as directory, patch.object(watch.checker, "load_apps", return_value=pins), \
|
|
patch.object(watch, "discover", side_effect=fake_discover), patch.object(watch, "write_state"), \
|
|
patch.object(sys, "argv", ["watch", "--state", str(Path(directory) / "state.json")]), \
|
|
patch.dict(os.environ, {"GITHUB_TOKEN": token} if token else {}, clear=True), \
|
|
patch.object(sys, "stdout", new_callable=io.StringIO) as output:
|
|
self.assertEqual(watch.main(), 0)
|
|
return seen, json.loads(output.getvalue())
|
|
limited, limited_output = exercise("")
|
|
full, full_output = exercise("test-token")
|
|
self.assertEqual(limited, [4])
|
|
self.assertEqual(limited_output["mode"], "degraded_no_github_token")
|
|
self.assertEqual(full, [100])
|
|
self.assertEqual(full_output["mode"], "authenticated_full_cycle")
|
|
self.assertEqual(full_output["scanned"], 58)
|
|
|
|
def test_sparse_or_untracked_checkout_fails_before_state_write(self):
|
|
for pins in ([], [pin("internal", "internal")]):
|
|
with self.subTest(pins=pins), tempfile.TemporaryDirectory() as directory, \
|
|
patch.object(watch.checker, "load_apps", return_value=pins), \
|
|
patch.object(watch, "write_state") as write, \
|
|
patch.object(watch, "discover") as discover, \
|
|
patch.object(sys, "argv", ["watch", "--state", str(Path(directory) / "state.json")]), \
|
|
patch.object(sys, "stderr", new_callable=io.StringIO):
|
|
with self.assertRaises(SystemExit) as exit_status:
|
|
watch.main()
|
|
self.assertEqual(exit_status.exception.code, 2)
|
|
write.assert_not_called()
|
|
discover.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|