tegwick
57ad4abefc
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11beb-4eef-7fc1-94a7-597130a01ae9
107 lines
5.7 KiB
Python
107 lines
5.7 KiB
Python
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, scan_repo, target_repo # noqa: E402
|
|
|
|
|
|
class ClassifyTest(unittest.TestCase):
|
|
def test_kinds(self):
|
|
cases = [
|
|
("posts/x.yaml", "/coulomb/pr-hall-of-helix/raw/branch/main/a.png", "url: x", "raw-fetch"),
|
|
("src/client.py", "/api/v1/repos/coulomb/x", "u = ...", "api-client"),
|
|
(".forgejo/workflows/image.yaml", "/coulomb/x", "uses: x", "ci-action"),
|
|
("go.mod", "/coulomb/x", "require x", "package-consumer"),
|
|
("k8s/app.yaml", "/coulomb/x.git", " repoURL: https://...", "gitops-source"),
|
|
("chart/values.yaml", "/coulomb/x", " repository: forgejo...", "image-pull"),
|
|
("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)
|
|
|
|
def test_target_repo(self):
|
|
self.assertEqual(target_repo("/coulomb/key-cape.git"), "coulomb/key-cape")
|
|
self.assertEqual(target_repo("/api/v1/repos/coulomb/state-hub/issues"), "coulomb/state-hub")
|
|
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_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("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()
|