coulomb/prj-forgejo-org-refactor
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix access inventory evidence and reconcile closure coordination

Browse source 
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a11beb-4eef-7fc1-94a7-597130a01ae9
This commit is contained in:
tegwick 2026-10-09 00:23:04 +02:00
parent 490a89dca6
commit 57ad4abefc
9 changed files with 342 additions and 111 deletions
Download patch file Download diff file Expand all files Collapse all files

70
tests/test_access_consumers.py
View file

@ -1,9 +1,14 @@
import sys
import json
import os
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
from access_consumers import classify, runtime_consumers, target_repo # noqa: E402
from access_consumers import classify, runtime_consumers, scan_repo, target_repo # noqa: E402
class ClassifyTest(unittest.TestCase):
@ -18,6 +23,8 @@ class ClassifyTest(unittest.TestCase):
("README.md", "/coulomb/x", "see", "doc-link"),
("docs/evidence/run.json", "/coulomb/x", "x", "record"),
("Makefile", "/coulomb/x", "push", "build-tooling"),
("pyproject.toml", "/api/packages/coulomb/pypi/simple", "url: x", "package-consumer"),
("client.py", "/api/v1/packages/coulomb", "u = ...", "package-consumer"),
]
for f, p, line, want in cases:
self.assertEqual(classify(f, p, line), want, f)
@ -28,16 +35,73 @@ class ClassifyTest(unittest.TestCase):
self.assertEqual(target_repo("/v2/coulomb/flex-auth/manifests/x"), "coulomb/flex-auth")
self.assertEqual(target_repo("/coulomb/activity-core@sha256:ab"), "coulomb/activity-core")
self.assertIsNone(target_repo("/user/login"))
self.assertIsNone(target_repo("/api/packages/coulomb/pypi/simple"))
self.assertIsNone(target_repo("/api/v1/packages/coulomb"))
self.assertIsNone(target_repo("/api/v1/orgs/coulomb/repos"))
def test_runtime_anonymous_pull(self):
def test_missing_pull_secret_does_not_prove_anonymous_access(self):
rt = {"applications": [], "workloads": [{
"namespace": "ns", "kind": "Deployment", "name": "api", "image_pull_secret_names": [],
"images": ["forgejo.coulomb.social/coulomb/x@sha256:00"]}]}
(c,) = runtime_consumers(rt)
self.assertEqual(c["class"], "cluster-image-pull")
self.assertTrue(c["credential"].startswith("anonymous"))
self.assertTrue(c["credential"].startswith("unknown"))
self.assertEqual(c["target"], "coulomb/x")
def test_scan_excludes_sensitive_and_binary_files_without_values(self):
with tempfile.TemporaryDirectory() as d:
repo = Path(d)
(repo / "client.py").write_text('url = "https://user:hidden@forgejo.coulomb.social/coulomb/valid?token=hidden"\n')
(repo / ".env").write_text("forgejo.coulomb.social/coulomb/private-value\n")
(repo / "secret.sops.yaml").write_text("forgejo.coulomb.social/coulomb/encrypted-value\n")
(repo / "binary").write_bytes(b"\0forgejo.coulomb.social/coulomb/binary-value")
(repo / "link.py").symlink_to(repo / "client.py")
os.mkfifo(repo / "pipe")
diagnostics = []
hits = scan_repo(repo, diagnostics)
self.assertEqual([h["target"] for h in hits], ["coulomb/valid"])
serialized = json.dumps(hits)
for value in ["hidden", "private-value", "encrypted-value", "binary-value"]:
self.assertNotIn(value, serialized)
self.assertEqual({x["reason"] for x in diagnostics},
{"sensitive-or-encrypted", "binary-content", "symlink-not-followed", "non-regular-file"})
def test_cli_is_repeatable_and_keeps_legacy_host_separate(self):
with tempfile.TemporaryDirectory() as d:
root = Path(d)
repo = root / "reader"
repo.mkdir()
subprocess.run(["git", "init", "-q", str(repo)], check=True)
(repo / "client.py").write_text(
'a = "https://forgejo.coulomb.social/coulomb/x"\n'
'b = "https://gitea.coulomb.social/coulomb/x"\n')
runtime = root / "runtime.json"
runtime.write_text(json.dumps({"applications": [], "workloads": []}))
assignments = root / "assignments.csv"
assignments.write_text("repository,proposed_org\nx,netkingdom\n")
output = root / "output.json"
command = [sys.executable, str(Path(__file__).resolve().parents[1] / "tools/access_consumers.py"),
"--root", str(root), "--runtime", str(runtime),
"--assignments", str(assignments), "--output", str(output)]
subprocess.run(command, check=True, capture_output=True)
first = output.read_bytes()
subprocess.run(command, check=True, capture_output=True)
self.assertEqual(first, output.read_bytes())
report = json.loads(first)
self.assertEqual({c["host"] for c in report["consumers"]},
{"forgejo.coulomb.social", "gitea.coulomb.social"})
self.assertEqual(report["scan_error_count"], 0)
def test_unreadable_file_is_reported_without_raw_error(self):
with tempfile.TemporaryDirectory() as d:
repo = Path(d)
(repo / "client.py").write_text("placeholder")
diagnostics = []
with patch.object(Path, "read_text", side_effect=OSError("private error detail")):
self.assertEqual(scan_repo(repo, diagnostics), [])
self.assertEqual(diagnostics, [{"repository": repo.name,
"path": "client.py", "reason": "unreadable-file"}])
if __name__ == "__main__":
unittest.main()