Packaging ; improve terminal output

Author: Yann-P

pyproject.toml

@@ -5,12 +5,17 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "security-overview"
 version = "0.1.0"
+description = "Overview of GitHub security advisories across orgs"
+readme = "README.md"
 requires-python = ">=3.10"
 dependencies = ["httpx", "trio"]
 
 [project.optional-dependencies]
 dev = ["pytest", "pytest-trio", "respx"]
 
+[project.scripts]
+security-overview = "security_overview.cli:run"
+
 [tool.setuptools.packages.find]
 where = ["."]
 include = ["security_overview*"]

security_overview/cli.py

@@ -5,7 +5,7 @@
 
 from . import render_md, render_terminal
 from .constants import ALL_STATES
-from .fetch import check_token, fetch, fetch_first_pull
+from .fetch import check_token, fetch, fetch_pulls
 
 RENDERERS = {
     "terminal": render_terminal,
@@ -72,8 +72,15 @@ async def main():
         }
         async with trio.open_nursery() as nursery:
             for fork_url in fork_urls:
-                nursery.start_soon(fetch_first_pull, client, fork_url, pull_results)
+                nursery.start_soon(fetch_pulls, client, fork_url, pull_results)
 
     for org in args.orgs:
-        print(renderer.render_org(org, states, results, pull_results=pull_results, redact=args.redact))
+        out = renderer.render_org(org, states, results, pull_results=pull_results, redact=args.redact)
+        if out:
+            print(out)
     print()
+
+
+def run():
+    """Synchronous console-script entry point."""
+    trio.run(main)

security_overview/fetch.py

@@ -14,20 +14,19 @@ def check_token():
     return token
 
 
-async def fetch_first_pull(client, fork_html_url, results):
+async def fetch_pulls(client, fork_html_url, results):
     path = fork_html_url.split("github.com/", 1)[-1].rstrip("/")
     url = f"{GITHUB_API}/repos/{path}/pulls"
-    params = {"per_page": 1, "state": "all", "sort": "created", "direction": "asc"}
+    params = {"per_page": 100, "state": "all", "sort": "created", "direction": "asc"}
     try:
         req = client.build_request("GET", url, params=params)
         resp = await client.send(req)
         if resp.status_code == 200:
-            pulls = resp.json()
-            results[fork_html_url] = pulls[0] if pulls else None
+            results[fork_html_url] = resp.json()
         else:
-            results[fork_html_url] = None
+            results[fork_html_url] = []
     except httpx.HTTPError:
-        results[fork_html_url] = None
+        results[fork_html_url] = []
 
 
 async def fetch(client, org, state, results):

security_overview/render_md.py

@@ -1,4 +1,4 @@
-from .render_common import days_ago, group_by_repo, pr_state
+from .render_common import days_ago, group_by_repo
 
 
 def render_org(org, states, results_by_key, pull_results=None, redact=False):
@@ -9,14 +9,13 @@ def render_org(org, states, results_by_key, pull_results=None, redact=False):
     lines = [f"\n## {'REDACTED ORG' if redact else org}"]
 
     if not advisories:
-        lines.append("_no advisories_")
-        return "\n".join(lines)
+        return ""
 
     for repo, items in group_by_repo(advisories):
         items = sorted(items, key=lambda a: a.get("updated_at", ""), reverse=True)
         lines.append(f"\n### {'REDACTED REPO' if redact else repo}\n")
-        lines.append("| Age | State | Advisory | Title | CVE | PR |")
-        lines.append("|-----|-------|----------|-------|-----|----|")
+        lines.append("| Age | State | Advisory | Title | CVE | PRs |")
+        lines.append("|-----|-------|----------|-------|-----|-----|")
         for advisory in items:
             ghsa_id = advisory.get("ghsa_id", "") if not redact else "GHSA-xxxx-yyyy-zzzz"
             url = advisory.get("html_url", "")
@@ -26,11 +25,10 @@ def render_org(org, states, results_by_key, pull_results=None, redact=False):
             cve = advisory.get("cve_id") or ""
             fork = advisory.get("private_fork")
             fork_html_url = fork.get("html_url") if fork else None
-            first_pull = (pull_results or {}).get(fork_html_url) if fork_html_url else None
-            ps = pr_state(first_pull)
+            pulls = (pull_results or {}).get(fork_html_url, []) if fork_html_url else []
             d = days_ago(date)
             advisory_cell = f"[{ghsa_id}]({url})" if (url and not redact) else ghsa_id
-            pr_cell = f"[PR {ps}]({first_pull['html_url']})" if first_pull else ""
+            pr_cell = "PRs: " + ", ".join(f"[#{p['number']}]({p['html_url']})" for p in pulls) if pulls else ""
             lines.append(f"| {d}d | {state_val} | {advisory_cell} | {title} | {cve} | {pr_cell} |")
 
     return "\n".join(lines)

security_overview/render_terminal.py

@@ -1,6 +1,13 @@
+import shutil
+
 from .constants import BOLD, GREY, RESET, PR_STATE_COLORS, STATE_COLORS
 from .render_common import age_t, days_ago, group_by_repo, plasma_rgb, pr_state
 
+# Visible chars on the advisory line excluding title:
+#   4(connector) + 3(badge) + 2 + 9(state) + 2 = 20
+_FIXED_OVERHEAD = 20
+_NVD_URL = "https://nvd.nist.gov/vuln/detail/{}"
+
 
 def _link(text, target):
     return f"\033[4m\033]8;;{target}\033\\{text}\033]8;;\033\\\033[24m"
@@ -10,6 +17,22 @@ def _color(text, c):
     return f"{c}{text}{RESET}"
 
 
+def _dim(text):
+    return _color(text, GREY)
+
+
+def _fmt_age(days):
+    if days < 7:
+        s = f"{days}d"
+    elif days < 30:
+        s = f"{days // 7}w"
+    elif days < 365:
+        s = f"{days // 30}m"
+    else:
+        s = f"{days // 365}y"
+    return s.rjust(3)
+
+
 def _age_badge(date_str, text):
     t = age_t(date_str) if date_str else 0.0
     r, g, b = plasma_rgb(t)
@@ -20,42 +43,64 @@ def _age_badge(date_str, text):
 
 
 def render_org(org, states, results_by_key, pull_results=None, redact=False):
+    term_w = shutil.get_terminal_size(fallback=(120, 24)).columns
+    title_w = max(10, term_w - _FIXED_OVERHEAD)
     advisories = []
     for state in states:
         advisories.extend(results_by_key.get((org, state), []))
 
     lines = [f"\n{BOLD}=== {org if not redact else 'REDACTED ORG'} ==={RESET}"]
 
     if not advisories:
-        if not redact:
-            lines.append("  (no advisories)")
-        return "\n".join(lines)
+        return ""
 
+    first_repo = True
     for repo, items in group_by_repo(advisories):
         items = sorted(items, key=lambda a: a.get("updated_at", ""), reverse=True)
+        if not items:
+            continue
+        if not first_repo:
+            lines.append("")
+        first_repo = False
+
         lines.append(f"{BOLD}{repo if not redact else 'REDACTED REPO'}{RESET}")
-        for advisory in items:
-            id_ = advisory.get("ghsa_id", "") if not redact else "GHSA-xxxx-yyyy-zzzz"
+        for i, advisory in enumerate(items):
+            is_last = i == len(items) - 1
+            connector = _dim("└── " if is_last else "├── ")
+            # prefix for sub-lines: continue the vertical bar only if more advisories follow
+            sub_prefix = "    " if is_last else "│   "
+
             url = advisory.get("html_url", "")
             date = advisory.get("updated_at", "")
-            title = (advisory.get("summary") or "")[:40].ljust(40)
             state_val = advisory.get("state") or "?"
             state_str = _color(state_val.ljust(9), STATE_COLORS.get(state_val, ""))
             cve = advisory.get("cve_id") or ""
-            cve_str = _color(cve[:15].ljust(15), GREY)
             fork = advisory.get("private_fork")
             fork_html_url = fork.get("html_url") if fork else None
-            first_pull = (pull_results or {}).get(fork_html_url) if fork_html_url else None
-            ps = pr_state(first_pull)
-            pull_str = (
-                _color(_link("PR " + ps.ljust(6), first_pull["html_url"]), PR_STATE_COLORS.get(ps, GREY))
-                if first_pull else " " * 9
-            )
+            pulls = (pull_results or {}).get(fork_html_url, []) if fork_html_url else []
+            title = (advisory.get("summary") or "")[:title_w]
             d = days_ago(date)
-            badge = _age_badge(date, (str(d) + "d").rjust(5))
+            badge = _age_badge(date, _fmt_age(d))
+
             if redact:
-                lines.append(f" {badge}\t{id_}  {state_str}")
+                lines.append(f"{connector}{badge}  {state_str}")
             else:
-                lines.append(f" {badge}\t{state_str}  {_color(_link(id_, url), GREY)}  {title}  {cve_str}  {pull_str}")
+                lines.append(f"{connector}{badge}  {state_str}  {_link(title, url)}")
+
+                sub_items = []
+                if cve:
+                    sub_items.append(_link(cve, _NVD_URL.format(cve)))
+                if pulls:
+                    pr_items = []
+                    for p in pulls:
+                        ps = pr_state(p)
+                        label = "opened" if ps == "open" else ps
+                        pr_items.append(_color(_link(f"#{p['number']} ({label})", p["html_url"]), PR_STATE_COLORS.get(ps, GREY)))
+                    sub_items.append("PRs: " + ", ".join(pr_items))
+
+                for j, sub_item in enumerate(sub_items):
+                    is_last_sub = j == len(sub_items) - 1
+                    sub_connector = _dim(sub_prefix + ("└── " if is_last_sub else "├── "))
+                    lines.append(f"{sub_connector}{sub_item}")
 
     return "\n".join(lines)

tests/test_render_md.py

@@ -26,7 +26,7 @@ def test_repo_heading():
 
 def test_table_header():
     out = render_md.render_org(ORG, STATES, RESULTS)
-    assert "| Age | State | Advisory | Title | CVE | PR |" in out
+    assert "| Age | State | Advisory | Title | CVE | PRs |" in out
 
 
 def test_linked_ghsa():
@@ -41,7 +41,7 @@ def test_cve_in_row():
 
 def test_no_advisories():
     out = render_md.render_org(ORG, STATES, {})
-    assert "_no advisories_" in out
+    assert out == ""
 
 
 def test_redact_hides_org():
@@ -64,11 +64,27 @@ def test_pr_link():
         }]
     }
     pull_results = {
-        "https://github.com/acme/repo-x-private": {
-            "html_url": "https://github.com/acme/repo-x-private/pull/1",
-            "state": "open",
-            "merged_at": None,
-        }
+        "https://github.com/acme/repo-x-private": [
+            {
+                "number": 1,
+                "html_url": "https://github.com/acme/repo-x-private/pull/1",
+                "state": "open",
+                "merged_at": None,
+            },
+            {
+                "number": 3,
+                "html_url": "https://github.com/acme/repo-x-private/pull/3",
+                "state": "open",
+                "merged_at": None,
+            },
+        ]
     }
     out = render_md.render_org(ORG, STATES, results, pull_results=pull_results)
-    assert "[PR open](" in out
+    assert "[#1](https://github.com/acme/repo-x-private/pull/1)" in out
+    assert "[#3](https://github.com/acme/repo-x-private/pull/3)" in out
+    assert "PRs:" in out
+
+
+def test_table_header_pr_column():
+    out = render_md.render_org(ORG, STATES, RESULTS)
+    assert "| PRs |" in out

tests/test_render_terminal.py

@@ -24,14 +24,60 @@ def test_repo_name():
     assert "repo-x" in out
 
 
-def test_ghsa_id_present():
+def test_title_linked():
     out = render_terminal.render_org(ORG, STATES, RESULTS)
-    assert "GHSA-1234-5678-abcd" in out
+    assert "Remote code execution via evil input" in out
+    assert ADVISORY["html_url"] in out
 
 
 def test_no_advisories():
     out = render_terminal.render_org(ORG, STATES, {})
-    assert "(no advisories)" in out
+    assert out == ""
+
+
+def test_tree_connector_single():
+    out = render_terminal.render_org(ORG, STATES, RESULTS)
+    assert "└── " in out
+
+
+def test_tree_connector_multiple():
+    results = {
+        (ORG, "draft"): [
+            ADVISORY,
+            {**ADVISORY, "ghsa_id": "GHSA-aaaa-bbbb-cccc", "updated_at": "2024-05-01T00:00:00Z"},
+        ]
+    }
+    out = render_terminal.render_org(ORG, STATES, results)
+    assert "├── " in out
+    assert "└── " in out
+
+
+def test_cve_on_sub_line():
+    out = render_terminal.render_org(ORG, STATES, RESULTS)
+    lines = out.splitlines()
+    cve_line = next(l for l in lines if "CVE-2024-9999" in l)
+    assert "└── " in cve_line
+    assert "nvd.nist.gov" in cve_line
+
+
+def test_pr_on_sub_line():
+    results = {
+        (ORG, "draft"): [{
+            **ADVISORY,
+            "private_fork": {"html_url": "https://github.com/acme/repo-x-private"},
+        }]
+    }
+    pull_results = {
+        "https://github.com/acme/repo-x-private": [
+            {"number": 7, "html_url": "https://github.com/acme/repo-x-private/pull/7",
+             "state": "open", "merged_at": None},
+        ]
+    }
+    out = render_terminal.render_org(ORG, STATES, results, pull_results=pull_results)
+    lines = out.splitlines()
+    sub_line = next(l for l in lines if "#7" in l)
+    assert "opened" in sub_line
+    assert "└── " in sub_line
 
 
 def test_redact_hides_org():
@@ -45,7 +91,7 @@ def test_redact_hides_repo():
     assert "repo-x" not in out
 
 
-def test_redact_masks_ghsa():
+def test_redact_hides_title():
     out = render_terminal.render_org(ORG, STATES, RESULTS, redact=True)
-    assert "GHSA-xxxx-yyyy-zzzz" in out
-    assert "GHSA-1234-5678-abcd" not in out
+    assert "Remote code execution" not in out
+    assert ADVISORY["html_url"] not in out