From 16b1420a8e743c04b1d9fbe71af7c63694b09cd4 Mon Sep 17 00:00:00 2001 From: Jamie Strandboge Date: Tue, 24 Mar 2026 13:39:00 -0500 Subject: [PATCH 1/7] fix: black for setup.py --- setup.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/setup.py b/setup.py index 58f685d..f41253b 100644 --- a/setup.py +++ b/setup.py @@ -1,8 +1,10 @@ from setuptools import setup + if __name__ == "__main__": setup( scripts=[ # bash "bin/cve-edit", "bin/cve-nfu", - ]) + ] + ) From c8e4c2c3899c661ff299ad063c30f724e2ad7a73 Mon Sep 17 00:00:00 2001 From: Jamie Strandboge Date: Tue, 24 Mar 2026 10:53:20 -0500 Subject: [PATCH 2/7] feat: improve SQL schema and operations --- cvelib/sql.py | 632 ++++++++++++++++++++++++++++++++++++++++---- tests/test_sql.py | 650 +++++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 1196 insertions(+), 86 deletions(-) diff --git a/cvelib/sql.py b/cvelib/sql.py index 569f49f..22d7bec 100644 --- a/cvelib/sql.py +++ b/cvelib/sql.py @@ -9,7 +9,7 @@ import sqlite3 import sys import textwrap -from typing import Dict, List, Tuple +from typing import Dict, List, Tuple, Union from cvelib.common import ( _experimental, @@ -19,7 +19,9 @@ verifyDate, ) import cvelib.cve +import cvelib.github import cvelib.pkg +import cvelib.scan class CVEdb(object): @@ -31,17 +33,6 @@ def __del__(self): if hasattr(self, "conn"): self.conn.close() - # TODO cves: - # - add/break out scans - # Later cves: - # - break out references - # - break out mitigation - # - break out bugs - # - break out discoveredBy - # - break out assignedTo - # Later pkgs - # - add/break out tags - # - add/break out patches def create_tables(self): """Create all the tables""" cursor = self.conn.cursor() @@ -53,13 +44,10 @@ def create_tables(self): 'closeDate' DATE, 'publicDate' DATE, 'crd' DATE, - 'references' TEXT, 'description' TEXT, 'notes' TEXT, 'mitigation' TEXT, - 'bugs' TEXT, 'priority' TEXT, - 'discoveredBy' TEXT, 'assignedTo' TEXT, 'cvss' TEXT ) @@ -74,15 +62,140 @@ def create_tables(self): 'candidate' TEXT NOT NULL, 'status' TEXT NOT NULL, 'when' TEXT, - 'priority' TEXT, PRIMARY KEY ('product', 'where', 'software', 'modifier', 'candidate') ) """) - def insert_into_cves(self, cve: cvelib.cve.CVE): + cursor.execute(""" +CREATE TABLE 'cve_references' ( + 'candidate' TEXT NOT NULL, + 'reference' TEXT NOT NULL, + PRIMARY KEY ('candidate', 'reference') +) +""") + + cursor.execute(""" +CREATE TABLE 'cve_bugs' ( + 'candidate' TEXT NOT NULL, + 'bug' TEXT NOT NULL, + PRIMARY KEY ('candidate', 'bug') +) +""") + + cursor.execute(""" +CREATE TABLE 'cve_discovered_by' ( + 'candidate' TEXT NOT NULL, + 'discoverer' TEXT NOT NULL, + PRIMARY KEY ('candidate', 'discoverer') +) +""") + + cursor.execute(""" +CREATE TABLE 'ghas_dependabot' ( + 'candidate' TEXT NOT NULL, + 'dependency' TEXT NOT NULL, + 'detectedIn' TEXT NOT NULL, + 'advisory' TEXT NOT NULL, + 'severity' TEXT NOT NULL, + 'status' TEXT NOT NULL, + 'url' TEXT NOT NULL, + PRIMARY KEY ('candidate', 'dependency', 'detectedIn', 'advisory', 'url') +) +""") + + cursor.execute(""" +CREATE TABLE 'ghas_secret' ( + 'candidate' TEXT NOT NULL, + 'secret' TEXT NOT NULL, + 'detectedIn' TEXT NOT NULL, + 'severity' TEXT NOT NULL, + 'status' TEXT NOT NULL, + 'url' TEXT NOT NULL, + PRIMARY KEY ('candidate', 'secret', 'detectedIn', 'url') +) +""") + + cursor.execute(""" +CREATE TABLE 'ghas_code' ( + 'candidate' TEXT NOT NULL, + 'description' TEXT NOT NULL, + 'detectedIn' TEXT NOT NULL, + 'severity' TEXT NOT NULL, + 'status' TEXT NOT NULL, + 'url' TEXT NOT NULL, + PRIMARY KEY ('candidate', 'description', 'detectedIn', 'url') +) +""") + + cursor.execute(""" +CREATE TABLE 'scan_oci' ( + 'candidate' TEXT NOT NULL, + 'component' TEXT NOT NULL, + 'detectedIn' TEXT NOT NULL, + 'advisory' TEXT NOT NULL, + 'versionAffected' TEXT NOT NULL, + 'versionFixed' TEXT NOT NULL, + 'severity' TEXT NOT NULL, + 'status' TEXT NOT NULL, + 'url' TEXT NOT NULL, + PRIMARY KEY ('candidate', 'component', 'detectedIn', 'advisory', 'url') +) +""") + + cursor.execute(""" +CREATE TABLE 'pkg_patches' ( + 'product' TEXT, + 'where' TEXT, + 'software' TEXT NOT NULL, + 'modifier' TEXT, + 'candidate' TEXT NOT NULL, + 'patch' TEXT NOT NULL, + PRIMARY KEY ('product', 'where', 'software', 'modifier', 'candidate', 'patch') +) +""") + + cursor.execute(""" +CREATE TABLE 'pkg_tags' ( + 'product' TEXT, + 'where' TEXT, + 'software' TEXT NOT NULL, + 'modifier' TEXT, + 'candidate' TEXT NOT NULL, + 'tagKey' TEXT NOT NULL, + 'tag' TEXT NOT NULL, + PRIMARY KEY ('product', 'where', 'software', 'modifier', 'candidate', 'tagKey', 'tag') +) +""") + + cursor.execute(""" +CREATE TABLE 'pkg_priorities' ( + 'product' TEXT, + 'where' TEXT, + 'software' TEXT NOT NULL, + 'modifier' TEXT, + 'candidate' TEXT NOT NULL, + 'priorityKey' TEXT NOT NULL, + 'priority' TEXT NOT NULL, + PRIMARY KEY ('product', 'where', 'software', 'modifier', 'candidate', 'priorityKey') +) +""") + + cursor.execute(""" +CREATE TABLE 'pkg_close_dates' ( + 'product' TEXT, + 'where' TEXT, + 'software' TEXT NOT NULL, + 'modifier' TEXT, + 'candidate' TEXT NOT NULL, + 'closeDateKey' TEXT NOT NULL, + 'closeDate' DATE NOT NULL, + PRIMARY KEY ('product', 'where', 'software', 'modifier', 'candidate', 'closeDateKey') +) +""") + + def insert_into_cves(self, cve: cvelib.cve.CVE, commit: bool = True): """Insert a CVE into the database""" cursor = self.conn.cursor() - # Insert using parameterized queries cursor.execute( """ INSERT INTO 'cves' ( @@ -91,16 +204,13 @@ def insert_into_cves(self, cve: cvelib.cve.CVE): 'closeDate', 'publicDate', 'crd', - 'references', 'description', 'notes', 'mitigation', - 'bugs', 'priority', - 'discoveredBy', 'assignedTo', 'cvss' - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( cve.candidate, @@ -108,24 +218,22 @@ def insert_into_cves(self, cve: cvelib.cve.CVE): convertCveDateToISO8601(cve.closeDate, cve.candidate), convertCveDateToISO8601(cve.publicDate, cve.candidate), convertCveDateToISO8601(cve.crd, cve.candidate), - " \n".join(cve.references), " \n".join(cve.description), " \n".join(cve.notes), " \n".join(cve.mitigation), - " \n".join(cve.bugs), cve.priority, - cve.discoveredBy, cve.assignedTo, cve.cvss, ), ) + if commit: + self.conn.commit() - self.conn.commit() - - def insert_into_pkgs(self, candidate: str, pkg: cvelib.pkg.CvePkg): + def insert_into_pkgs( + self, candidate: str, pkg: cvelib.pkg.CvePkg, commit: bool = True + ): """Insert a pkg into the database""" cursor = self.conn.cursor() - # Insert using parameterized queries cursor.execute( """ INSERT INTO 'pkgs' ( @@ -135,9 +243,8 @@ def insert_into_pkgs(self, candidate: str, pkg: cvelib.pkg.CvePkg): 'modifier', 'candidate', 'status', - 'when', - 'priority' - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + 'when' + ) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( pkg.product, @@ -147,33 +254,365 @@ def insert_into_pkgs(self, candidate: str, pkg: cvelib.pkg.CvePkg): candidate, pkg.status, pkg.when, - ( - "" - if pkg.software not in pkg.priorities - else pkg.priorities[pkg.software] - ), ), ) + if commit: + self.conn.commit() - self.conn.commit() + def insert_into_cve_references( + self, candidate: str, references: List[str], commit: bool = True + ): + """Insert CVE references into the database""" + cursor = self.conn.cursor() + for ref in references: + ref = ref.strip() + if ref: + cursor.execute( + """ + INSERT INTO 'cve_references' ( + 'candidate', 'reference' + ) VALUES (?, ?) + """, + (candidate, ref), + ) + if commit: + self.conn.commit() + + def insert_into_cve_bugs( + self, candidate: str, bugs: List[str], commit: bool = True + ): + """Insert CVE bugs into the database""" + cursor = self.conn.cursor() + for bug in bugs: + bug = bug.strip() + if bug: + cursor.execute( + """ + INSERT INTO 'cve_bugs' ( + 'candidate', 'bug' + ) VALUES (?, ?) + """, + (candidate, bug), + ) + if commit: + self.conn.commit() + + def insert_into_cve_discovered_by( + self, candidate: str, discoveredBy: str, commit: bool = True + ): + """Insert CVE discoveredBy into the database""" + cursor = self.conn.cursor() + for discoverer in discoveredBy.split(","): + discoverer = discoverer.strip() + if discoverer: + cursor.execute( + """ + INSERT INTO 'cve_discovered_by' ( + 'candidate', 'discoverer' + ) VALUES (?, ?) + """, + (candidate, discoverer), + ) + if commit: + self.conn.commit() + + def insert_into_ghas_dependabot( + self, + candidate: str, + dep: cvelib.github.GHDependabot, + commit: bool = True, + ): + """Insert a GHAS dependabot alert into the database""" + # OR IGNORE: some retired CVEs have truly identical entries (all + # fields match including url=unavailable). Parse-time checks catch + # meaningful duplicates; this silently drops identical rows. + cursor = self.conn.cursor() + cursor.execute( + """ + INSERT OR IGNORE INTO 'ghas_dependabot' ( + 'candidate', + 'dependency', + 'detectedIn', + 'advisory', + 'severity', + 'status', + 'url' + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + candidate, + dep.dependency, + dep.detectedIn, + dep.advisory, + dep.severity, + dep.status, + dep.url, + ), + ) + if commit: + self.conn.commit() + + def insert_into_ghas_secret( + self, + candidate: str, + sec: cvelib.github.GHSecret, + commit: bool = True, + ): + """Insert a GHAS secret alert into the database""" + # OR IGNORE: see insert_into_ghas_dependabot comment + cursor = self.conn.cursor() + cursor.execute( + """ + INSERT OR IGNORE INTO 'ghas_secret' ( + 'candidate', + 'secret', + 'detectedIn', + 'severity', + 'status', + 'url' + ) VALUES (?, ?, ?, ?, ?, ?) + """, + ( + candidate, + sec.secret, + sec.detectedIn, + sec.severity, + sec.status, + sec.url, + ), + ) + if commit: + self.conn.commit() + + def insert_into_ghas_code( + self, + candidate: str, + code: cvelib.github.GHCode, + commit: bool = True, + ): + """Insert a GHAS code scanning alert into the database""" + # OR IGNORE: see insert_into_ghas_dependabot comment + cursor = self.conn.cursor() + cursor.execute( + """ + INSERT OR IGNORE INTO 'ghas_code' ( + 'candidate', + 'description', + 'detectedIn', + 'severity', + 'status', + 'url' + ) VALUES (?, ?, ?, ?, ?, ?) + """, + ( + candidate, + code.description, + code.detectedIn, + code.severity, + code.status, + code.url, + ), + ) + if commit: + self.conn.commit() + + def insert_into_ghas( + self, + candidate: str, + ghas_item: object, + commit: bool = True, + ): + """Insert a GHAS alert into the appropriate table""" + if isinstance(ghas_item, cvelib.github.GHDependabot): + self.insert_into_ghas_dependabot(candidate, ghas_item, commit=commit) + elif isinstance(ghas_item, cvelib.github.GHSecret): + self.insert_into_ghas_secret(candidate, ghas_item, commit=commit) + elif isinstance(ghas_item, cvelib.github.GHCode): + self.insert_into_ghas_code(candidate, ghas_item, commit=commit) + else: + warn("unsupported GHAS type: %s" % type(ghas_item).__name__) + + def insert_into_scan_oci( + self, candidate: str, oci: cvelib.scan.ScanOCI, commit: bool = True + ): + """Insert a scan OCI report into the database""" + cursor = self.conn.cursor() + cursor.execute( + """ + INSERT INTO 'scan_oci' ( + 'candidate', + 'component', + 'detectedIn', + 'advisory', + 'versionAffected', + 'versionFixed', + 'severity', + 'status', + 'url' + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + candidate, + oci.component, + oci.detectedIn, + oci.advisory, + oci.versionAffected, + oci.versionFixed, + oci.severity, + oci.status, + oci.url, + ), + ) + if commit: + self.conn.commit() + + def insert_into_pkg_patches( + self, candidate: str, pkg: cvelib.pkg.CvePkg, commit: bool = True + ): + """Insert package patches into the database""" + cursor = self.conn.cursor() + for patch in pkg.patches: + cursor.execute( + """ + INSERT INTO 'pkg_patches' ( + 'product', 'where', 'software', 'modifier', + 'candidate', 'patch' + ) VALUES (?, ?, ?, ?, ?, ?) + """, + ( + pkg.product, + pkg.where, + pkg.software, + pkg.modifier, + candidate, + patch, + ), + ) + if commit: + self.conn.commit() + + def insert_into_pkg_tags( + self, candidate: str, pkg: cvelib.pkg.CvePkg, commit: bool = True + ): + """Insert package tags into the database""" + cursor = self.conn.cursor() + for tagKey, tagVals in pkg.tags.items(): + for tag in tagVals: + cursor.execute( + """ + INSERT INTO 'pkg_tags' ( + 'product', 'where', 'software', 'modifier', + 'candidate', 'tagKey', 'tag' + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + pkg.product, + pkg.where, + pkg.software, + pkg.modifier, + candidate, + tagKey, + tag, + ), + ) + if commit: + self.conn.commit() + + def insert_into_pkg_priorities( + self, candidate: str, pkg: cvelib.pkg.CvePkg, commit: bool = True + ): + """Insert package priorities into the database""" + cursor = self.conn.cursor() + for priKey, priVal in pkg.priorities.items(): + cursor.execute( + """ + INSERT INTO 'pkg_priorities' ( + 'product', 'where', 'software', 'modifier', + 'candidate', 'priorityKey', 'priority' + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + pkg.product, + pkg.where, + pkg.software, + pkg.modifier, + candidate, + priKey, + priVal, + ), + ) + if commit: + self.conn.commit() + + def insert_into_pkg_close_dates( + self, candidate: str, pkg: cvelib.pkg.CvePkg, commit: bool = True + ): + """Insert package close dates into the database""" + cursor = self.conn.cursor() + for cdKey, cdVal in pkg.closeDates.items(): + cursor.execute( + """ + INSERT INTO 'pkg_close_dates' ( + 'product', 'where', 'software', 'modifier', + 'candidate', 'closeDateKey', 'closeDate' + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + pkg.product, + pkg.where, + pkg.software, + pkg.modifier, + candidate, + cdKey, + cdVal, + ), + ) + if commit: + self.conn.commit() def get_schema(self) -> List: """Get database schema""" cursor = self.conn.cursor() - cursor.execute('SELECT sql FROM sqlite_master WHERE type="table"') + cursor.execute('SELECT sql FROM sqlite_master WHERE type="table" ORDER BY name') result = cursor.fetchall() return result + def commit(self): + """Commit the current transaction""" + self.conn.commit() + def execute_query(self, q: str) -> List: - """Execute query""" - # XXX: make this robust - if not q.startswith("SELECT "): - print("Only support SELECT") + """Execute a read-only query using set_authorizer()""" + + def _readOnlyAuthorizer(action, arg1, arg2, dbname, trigger): + """Only allow read operations""" + _ = arg1 # for pyright + _ = arg2 # for pyright + _ = dbname # for pyright + _ = trigger # for pyright + + # SQLITE_SELECT: allows the SELECT statement itself + # SQLITE_READ: allows reading individual columns (fired per column) + # SQLITE_FUNCTION: allows SQL functions (COUNT, COALESCE, etc) + allowed = { + sqlite3.SQLITE_SELECT, + sqlite3.SQLITE_READ, + sqlite3.SQLITE_FUNCTION, + } + if action in allowed: + return sqlite3.SQLITE_OK + return sqlite3.SQLITE_DENY + + self.conn.set_authorizer(_readOnlyAuthorizer) + try: + cursor = self.conn.cursor() + cursor.execute(q) + results = cursor.fetchall() + except sqlite3.DatabaseError as e: + print("Query error: %s" % e) return [] - cursor = self.conn.cursor() - # XXX: this is trusting - cursor.execute(q) - results = cursor.fetchall() + finally: + self.conn.set_authorizer(None) return results @@ -287,7 +726,60 @@ def main_cve_query(): description="Query cve database with SQL", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=textwrap.dedent("""\ -cve-query ... +Example queries: + # Packages affected by a CVE + cve-query -q "SELECT * FROM pkgs WHERE candidate = 'CVE-2023-1234'" + + # Dismissed dependabot alerts for the 'lodash' dependency + cve-query -q "SELECT candidate, status FROM ghas_dependabot + WHERE dependency = 'lodash' AND status LIKE 'dismissed%%'" + + # CVEs for 'go' opened between dates with priority >= medium + cve-query -q "SELECT DISTINCT c.candidate, COALESCE(pp.priority, c.priority) as priority + FROM pkgs p + JOIN cves c ON p.candidate = c.candidate + LEFT JOIN pkg_priorities pp ON pp.candidate = p.candidate + AND pp.product = p.product AND pp.'where' = p.'where' + AND pp.software = p.software AND pp.modifier = p.modifier + AND pp.priorityKey = p.software + WHERE p.software = 'go' AND c.openDate BETWEEN '2025-01-01' AND '2025-12-31' + AND COALESCE(pp.priority, c.priority) IN ('medium', 'high', 'critical')" + + # Software affected by a particular GHSA + cve-query -q "SELECT DISTINCT p.software FROM ghas_dependabot g + JOIN pkgs p ON g.candidate = p.candidate + WHERE g.advisory = 'https://github.com/advisories/GHSA-35jh-r3h4-6jhm'" + + # Count open CVEs by priority + cve-query -q "SELECT c.priority, COUNT(DISTINCT c.candidate) as count + FROM cves c JOIN pkgs p ON c.candidate = p.candidate + WHERE p.status IN ('needs-triage', 'needed', 'pending') + GROUP BY c.priority ORDER BY count DESC" + + # Open scan_oci vulnerabilities by severity + cve-query -q "SELECT s.severity, COUNT(*) as count FROM scan_oci s + WHERE s.status IN ('needs-triage', 'needed') + GROUP BY s.severity ORDER BY count DESC" + + # CVEs open longer than 90 days + cve-query -q "SELECT c.candidate, c.openDate, c.priority FROM cves c + JOIN pkgs p ON c.candidate = p.candidate + WHERE p.status IN ('needs-triage', 'needed', 'pending') + AND c.openDate < DATE('now', '-90 days') + GROUP BY c.candidate ORDER BY c.openDate" + + # Top discoverers by CVE count + cve-query -q "SELECT d.discoverer, COUNT(DISTINCT d.candidate) as count + FROM cve_discovered_by d GROUP BY d.discoverer + ORDER BY count DESC LIMIT 10" + + # Find CVEs referencing a specific bug + cve-query -q "SELECT c.candidate, c.priority, c.openDate FROM cves c + JOIN cve_bugs b ON c.candidate = b.candidate + WHERE b.bug = 'https://github.com/org/repo/issues/NNN'" + + # Show database schema + cve-query --show-schema """), ) @@ -365,9 +857,25 @@ def main_cve_query(): untriagedOk=True, filter_tag="-limit-report", # XXX: don't hardcode this ): - db.insert_into_cves(cve) + # For performance, commit=False and commit everything at the end + db.insert_into_cves(cve, commit=False) + db.insert_into_cve_references(cve.candidate, cve.references, commit=False) + db.insert_into_cve_bugs(cve.candidate, cve.bugs, commit=False) + db.insert_into_cve_discovered_by( + cve.candidate, cve.discoveredBy, commit=False + ) + for ghas_item in cve.ghas: + db.insert_into_ghas(cve.candidate, ghas_item, commit=False) + for scan in cve.scan_reports: + db.insert_into_scan_oci(cve.candidate, scan, commit=False) for pkg in cve.pkgs: - db.insert_into_pkgs(cve.candidate, pkg) + db.insert_into_pkgs(cve.candidate, pkg, commit=False) + db.insert_into_pkg_patches(cve.candidate, pkg, commit=False) + db.insert_into_pkg_tags(cve.candidate, pkg, commit=False) + db.insert_into_pkg_priorities(cve.candidate, pkg, commit=False) + db.insert_into_pkg_close_dates(cve.candidate, pkg, commit=False) + + db.commit() # an indicator to show that this is intended only for queries if dbname != ":memory:": @@ -377,6 +885,30 @@ def main_cve_query(): res = db.get_schema() for r in res: print(r[0]) + print( + "\n-- Relationships:\n" + "--\n" + "-- cve_references, cve_bugs, cve_discovered_by: join to cves on\n" + "-- candidate\n" + "--\n" + "-- ghas_dependabot, ghas_secret, ghas_code: join to cves on\n" + "-- candidate\n" + "--\n" + "-- scan_oci: join to cves on candidate\n" + "--\n" + "-- pkgs: join to cves on candidate\n" + "--\n" + "-- pkg_patches, pkg_tags, pkg_priorities, pkg_close_dates: join to\n" + "-- pkgs on (product, where, software, modifier, candidate)\n" + "--\n" + "-- Note: pkg_priorities.priorityKey typically matches pkgs.software.\n" + "-- When joining to get effective priority, use:\n" + "-- LEFT JOIN pkg_priorities pp ON pp.candidate = p.candidate\n" + "-- AND pp.product = p.product AND pp.'where' = p.'where'\n" + "-- AND pp.software = p.software AND pp.modifier = p.modifier\n" + "-- AND pp.priorityKey = p.software\n" + "-- Then: COALESCE(pp.priority, cves.priority) as priority" + ) elif args.query or args.query_file: sql: str if args.query_file: diff --git a/tests/test_sql.py b/tests/test_sql.py index 8e6d857..7aed199 100644 --- a/tests/test_sql.py +++ b/tests/test_sql.py @@ -1,4 +1,4 @@ -"""test_common.py: tests for common.py module""" +"""test_sql.py: tests for sql.py module""" # # SPDX-License-Identifier: MIT @@ -9,7 +9,9 @@ import cvelib.common import cvelib.cve +import cvelib.github import cvelib.pkg +import cvelib.scan import cvelib.sql import tests.testutil @@ -282,21 +284,18 @@ def test_insert_into_cves(self): self.cursor.execute("SELECT * FROM cves WHERE candidate=?", (cve.candidate,)) res = self.cursor.fetchone() - self.assertEqual(14, len(res)) + self.assertEqual(11, len(res)) self.assertEqual(cve.candidate, res[0]) self.assertEqual(cve.openDate, res[1]) self.assertEqual(cve.closeDate, res[2]) self.assertEqual(cve.publicDate, res[3]) self.assertEqual(cve.crd, res[4]) - self.assertEqual(" \n".join(cve.references), res[5]) - self.assertEqual(" \n".join(cve.description), res[6]) - self.assertEqual(" \n".join(cve.notes), res[7]) - self.assertEqual(" \n".join(cve.mitigation), res[8]) - self.assertEqual(" \n".join(cve.bugs), res[9]) - self.assertEqual(cve.priority, res[10]) - self.assertEqual(cve.discoveredBy, res[11]) - self.assertEqual(cve.assignedTo, res[12]) - self.assertEqual(cve.cvss, res[13]) + self.assertEqual(" \n".join(cve.description), res[5]) + self.assertEqual(" \n".join(cve.notes), res[6]) + self.assertEqual(" \n".join(cve.mitigation), res[7]) + self.assertEqual(cve.priority, res[8]) + self.assertEqual(cve.assignedTo, res[9]) + self.assertEqual(cve.cvss, res[10]) def test_insert_into_pkgs(self): """Test insert_into_pkgs()""" @@ -311,7 +310,7 @@ def test_insert_into_pkgs(self): self.cursor.execute("SELECT * FROM pkgs WHERE candidate=?", (cve_cand1,)) res = self.cursor.fetchone() - self.assertEqual(8, len(res)) + self.assertEqual(7, len(res)) self.assertEqual(pkg.product, res[0]) self.assertEqual(pkg.where, res[1]) self.assertEqual(pkg.software, res[2]) @@ -319,20 +318,16 @@ def test_insert_into_pkgs(self): self.assertEqual(cve_cand1, res[4]) self.assertEqual(pkg.status, res[5]) self.assertEqual(pkg.when, res[6]) - self.assertEqual("", res[7]) cve_cand2 = "CVE-2023-NNN2" pkg = cvelib.pkg.parse("upstream_baz: needed") - pkg_pri_override = "low" - pkg.setPriorities([("baz", pkg_pri_override), ("other", "critical")]) db.insert_into_pkgs(cve_cand2, pkg) self.cursor.execute("SELECT * FROM pkgs WHERE candidate=?", (cve_cand2,)) res = self.cursor.fetchone() - self.assertEqual(8, len(res)) + self.assertEqual(7, len(res)) self.assertEqual(pkg.software, res[2]) self.assertEqual(cve_cand2, res[4]) - self.assertEqual(pkg_pri_override, res[7]) def test_get_schema(self): """Test get_schema()""" @@ -343,28 +338,24 @@ def test_get_schema(self): self.cursor = self.conn.cursor() res = db.get_schema() - self.assertEqual(2, len(res)) + self.assertEqual(13, len(res)) - # XXX: brittle - exp0 = """CREATE TABLE 'cves' ( + exp_cves = """CREATE TABLE 'cves' ( 'candidate' TEXT PRIMARY KEY NOT NULL, 'openDate' DATE, 'closeDate' DATE, 'publicDate' DATE, 'crd' DATE, - 'references' TEXT, 'description' TEXT, 'notes' TEXT, 'mitigation' TEXT, - 'bugs' TEXT, 'priority' TEXT, - 'discoveredBy' TEXT, 'assignedTo' TEXT, 'cvss' TEXT )""" - self.assertEqual(exp0, res[0][0]) + self.assertEqual(exp_cves, res[3][0]) - exp1 = """CREATE TABLE 'pkgs' ( + exp_pkgs = """CREATE TABLE 'pkgs' ( 'product' TEXT, 'where' TEXT, 'software' TEXT NOT NULL, @@ -372,10 +363,28 @@ def test_get_schema(self): 'candidate' TEXT NOT NULL, 'status' TEXT NOT NULL, 'when' TEXT, - 'priority' TEXT, PRIMARY KEY ('product', 'where', 'software', 'modifier', 'candidate') )""" - self.assertEqual(exp1, res[1][0]) + self.assertEqual(exp_pkgs, res[11][0]) + + # Verify all table names are present + table_names = [r[0].split("'")[1] for r in res] + for exp_table in [ + "cves", + "pkgs", + "cve_references", + "cve_bugs", + "cve_discovered_by", + "ghas_dependabot", + "ghas_secret", + "ghas_code", + "scan_oci", + "pkg_patches", + "pkg_tags", + "pkg_priorities", + "pkg_close_dates", + ]: + self.assertIn(exp_table, table_names) def test_execute_query(self): """Test execute_query()""" @@ -395,21 +404,451 @@ def test_execute_query(self): cve.closeDate, cve.publicDate, cve.crd, - " \n".join(cve.references), " \n".join(cve.description), " \n".join(cve.notes), " \n".join(cve.mitigation), - " \n".join(cve.bugs), cve.priority, - cve.discoveredBy, cve.assignedTo, cve.cvss, ) self.assertEqual(1, len(res)) self.assertEqual(exp, res[0]) - # invalid - res = db.execute_query("UPDATE...") + # invalid - write operations are denied by authorizer + with tests.testutil.capturedOutput() as (output, error): + res = db.execute_query("DELETE FROM cves") + self.assertEqual(0, len(res)) + self.assertIn("Query error:", output.getvalue()) + + # malformed SQL + with tests.testutil.capturedOutput() as (output, error): + res = db.execute_query("UPDATE...") + self.assertEqual(0, len(res)) + self.assertIn("Query error:", output.getvalue()) + + # case-insensitive select works + res = db.execute_query("select * from 'cves'") + self.assertEqual(1, len(res)) + self.assertEqual(exp, res[0]) + + # INSERT denied + with tests.testutil.capturedOutput() as (output, error): + res = db.execute_query( + "INSERT INTO cves (candidate) VALUES ('CVE-2023-HACK')" + ) + self.assertEqual(0, len(res)) + self.assertIn("Query error:", output.getvalue()) + + # DROP denied + with tests.testutil.capturedOutput() as (output, error): + res = db.execute_query("DROP TABLE cves") + self.assertEqual(0, len(res)) + self.assertIn("Query error:", output.getvalue()) + + # verify data is unchanged after denied operations + res = db.execute_query("SELECT COUNT(*) FROM cves") + self.assertEqual(1, res[0][0]) + + def test_insert_into_cve_references(self): + """Test insert_into_cve_references()""" + db = cvelib.sql.CVEdb(":memory:") + db.create_tables() + self.conn = db.conn + self.cursor = self.conn.cursor() + + cand = "CVE-2023-0001" + refs = ["https://ref1", "https://ref2"] + db.insert_into_cve_references(cand, refs) + + self.cursor.execute("SELECT * FROM cve_references WHERE candidate=?", (cand,)) + res = self.cursor.fetchall() + self.assertEqual(2, len(res)) + self.assertEqual((cand, "https://ref1"), res[0]) + self.assertEqual((cand, "https://ref2"), res[1]) + + # empty list + db.insert_into_cve_references("CVE-2023-0002", []) + self.cursor.execute( + "SELECT * FROM cve_references WHERE candidate=?", ("CVE-2023-0002",) + ) + res = self.cursor.fetchall() + self.assertEqual(0, len(res)) + + # whitespace-only entries are skipped + db.insert_into_cve_references("CVE-2023-0003", [" ", "https://ref3"]) + self.cursor.execute( + "SELECT * FROM cve_references WHERE candidate=?", ("CVE-2023-0003",) + ) + res = self.cursor.fetchall() + self.assertEqual(1, len(res)) + self.assertEqual("https://ref3", res[0][1]) + + def test_insert_into_cve_bugs(self): + """Test insert_into_cve_bugs()""" + db = cvelib.sql.CVEdb(":memory:") + db.create_tables() + self.conn = db.conn + self.cursor = self.conn.cursor() + + cand = "CVE-2023-0001" + bugs = ["https://bug1", "https://bug2"] + db.insert_into_cve_bugs(cand, bugs) + + self.cursor.execute("SELECT * FROM cve_bugs WHERE candidate=?", (cand,)) + res = self.cursor.fetchall() + self.assertEqual(2, len(res)) + self.assertEqual((cand, "https://bug1"), res[0]) + self.assertEqual((cand, "https://bug2"), res[1]) + + # empty list + db.insert_into_cve_bugs("CVE-2023-0002", []) + self.cursor.execute( + "SELECT * FROM cve_bugs WHERE candidate=?", ("CVE-2023-0002",) + ) + res = self.cursor.fetchall() + self.assertEqual(0, len(res)) + + def test_insert_into_cve_discovered_by(self): + """Test insert_into_cve_discovered_by()""" + db = cvelib.sql.CVEdb(":memory:") + db.create_tables() + self.conn = db.conn + self.cursor = self.conn.cursor() + + cand = "CVE-2023-0001" + db.insert_into_cve_discovered_by(cand, "Alice, Bob") + + self.cursor.execute( + "SELECT * FROM cve_discovered_by WHERE candidate=?", (cand,) + ) + res = self.cursor.fetchall() + self.assertEqual(2, len(res)) + self.assertEqual((cand, "Alice"), res[0]) + self.assertEqual((cand, "Bob"), res[1]) + + # single discoverer + db.insert_into_cve_discovered_by("CVE-2023-0002", "Charlie") + self.cursor.execute( + "SELECT * FROM cve_discovered_by WHERE candidate=?", ("CVE-2023-0002",) + ) + res = self.cursor.fetchall() + self.assertEqual(1, len(res)) + self.assertEqual(("CVE-2023-0002", "Charlie"), res[0]) + + # empty string + db.insert_into_cve_discovered_by("CVE-2023-0003", "") + self.cursor.execute( + "SELECT * FROM cve_discovered_by WHERE candidate=?", ("CVE-2023-0003",) + ) + res = self.cursor.fetchall() + self.assertEqual(0, len(res)) + + def test_insert_into_ghas_dependabot(self): + """Test insert_into_ghas_dependabot()""" + db = cvelib.sql.CVEdb(":memory:") + db.create_tables() + self.conn = db.conn + self.cursor = self.conn.cursor() + + cand = "CVE-2023-0001" + dep = cvelib.github.GHDependabot( + { + "dependency": "lodash", + "detectedIn": "package-lock.json", + "advisory": "https://github.com/advisories/GHSA-test-1234-5678", + "severity": "high", + "status": "needs-triage", + "url": "https://github.com/org/repo/security/dependabot/1", + } + ) + db.insert_into_ghas_dependabot(cand, dep) + + self.cursor.execute("SELECT * FROM ghas_dependabot WHERE candidate=?", (cand,)) + res = self.cursor.fetchall() + self.assertEqual(1, len(res)) + self.assertEqual(cand, res[0][0]) + self.assertEqual("lodash", res[0][1]) + self.assertEqual("package-lock.json", res[0][2]) + self.assertEqual("https://github.com/advisories/GHSA-test-1234-5678", res[0][3]) + self.assertEqual("high", res[0][4]) + self.assertEqual("needs-triage", res[0][5]) + self.assertEqual("https://github.com/org/repo/security/dependabot/1", res[0][6]) + + def test_insert_into_ghas_secret(self): + """Test insert_into_ghas_secret()""" + db = cvelib.sql.CVEdb(":memory:") + db.create_tables() + self.conn = db.conn + self.cursor = self.conn.cursor() + + cand = "CVE-2023-0001" + sec = cvelib.github.GHSecret( + { + "secret": "github_personal_access_token", + "detectedIn": "config.yml", + "severity": "critical", + "status": "needs-triage", + "url": "https://github.com/org/repo/security/secret-scanning/1", + } + ) + db.insert_into_ghas_secret(cand, sec) + + self.cursor.execute("SELECT * FROM ghas_secret WHERE candidate=?", (cand,)) + res = self.cursor.fetchall() + self.assertEqual(1, len(res)) + self.assertEqual(cand, res[0][0]) + self.assertEqual("github_personal_access_token", res[0][1]) + self.assertEqual("config.yml", res[0][2]) + self.assertEqual("critical", res[0][3]) + self.assertEqual("needs-triage", res[0][4]) + self.assertEqual( + "https://github.com/org/repo/security/secret-scanning/1", res[0][5] + ) + + def test_insert_into_ghas_code(self): + """Test insert_into_ghas_code()""" + db = cvelib.sql.CVEdb(":memory:") + db.create_tables() + self.conn = db.conn + self.cursor = self.conn.cursor() + + cand = "CVE-2023-0001" + code = cvelib.github.GHCode( + { + "description": "SQL injection vulnerability", + "detectedIn": "src/app.py", + "severity": "high", + "status": "needs-triage", + "url": "https://github.com/org/repo/security/code-scanning/1", + } + ) + db.insert_into_ghas_code(cand, code) + + self.cursor.execute("SELECT * FROM ghas_code WHERE candidate=?", (cand,)) + res = self.cursor.fetchall() + self.assertEqual(1, len(res)) + self.assertEqual(cand, res[0][0]) + self.assertEqual("SQL injection vulnerability", res[0][1]) + self.assertEqual("src/app.py", res[0][2]) + self.assertEqual("high", res[0][3]) + self.assertEqual("needs-triage", res[0][4]) + self.assertEqual( + "https://github.com/org/repo/security/code-scanning/1", res[0][5] + ) + + def test_insert_into_ghas(self): + """Test insert_into_ghas()""" + db = cvelib.sql.CVEdb(":memory:") + db.create_tables() + self.conn = db.conn + self.cursor = self.conn.cursor() + + cand = "CVE-2023-0001" + + dep = cvelib.github.GHDependabot( + { + "dependency": "lodash", + "detectedIn": "package-lock.json", + "advisory": "https://github.com/advisories/GHSA-test-1234-5678", + "severity": "high", + "status": "needs-triage", + "url": "https://github.com/org/repo/security/dependabot/1", + } + ) + db.insert_into_ghas(cand, dep) + self.cursor.execute("SELECT COUNT(*) FROM ghas_dependabot") + self.assertEqual(1, self.cursor.fetchone()[0]) + + sec = cvelib.github.GHSecret( + { + "secret": "github_personal_access_token", + "detectedIn": "config.yml", + "severity": "critical", + "status": "needs-triage", + "url": "https://github.com/org/repo/security/secret-scanning/1", + } + ) + db.insert_into_ghas(cand, sec) + self.cursor.execute("SELECT COUNT(*) FROM ghas_secret") + self.assertEqual(1, self.cursor.fetchone()[0]) + + code = cvelib.github.GHCode( + { + "description": "SQL injection vulnerability", + "detectedIn": "src/app.py", + "severity": "high", + "status": "needs-triage", + "url": "https://github.com/org/repo/security/code-scanning/1", + } + ) + db.insert_into_ghas(cand, code) + self.cursor.execute("SELECT COUNT(*) FROM ghas_code") + self.assertEqual(1, self.cursor.fetchone()[0]) + + def test_insert_into_ghas_unsupported_type(self): + """Test insert_into_ghas() with unsupported type""" + db = cvelib.sql.CVEdb(":memory:") + db.create_tables() + + cand = "CVE-2023-0001" + with mock.patch("cvelib.sql.warn") as mock_warn: + db.insert_into_ghas(cand, "not a ghas object") # type: ignore[arg-type] + mock_warn.assert_called_once_with("unsupported GHAS type: str") + + def test_insert_into_scan_oci(self): + """Test insert_into_scan_oci()""" + db = cvelib.sql.CVEdb(":memory:") + db.create_tables() + self.conn = db.conn + self.cursor = self.conn.cursor() + + cand = "CVE-2023-0001" + oci = cvelib.scan.ScanOCI( + { + "component": "libssl3", + "detectedIn": "myimage@sha256:abc123", + "advisory": "https://security.example.com/CVE-2023-0001", + "version": "3.0.2-0ubuntu1.6", + "fixedBy": "3.0.2-0ubuntu1.7", + "severity": "high", + "status": "needs-triage", + "url": "https://quay.io/repository/org/myimage/manifest/sha256:abc123", + } + ) + db.insert_into_scan_oci(cand, oci) + + self.cursor.execute("SELECT * FROM scan_oci WHERE candidate=?", (cand,)) + res = self.cursor.fetchall() + self.assertEqual(1, len(res)) + self.assertEqual(cand, res[0][0]) + self.assertEqual("libssl3", res[0][1]) + self.assertEqual("myimage@sha256:abc123", res[0][2]) + self.assertEqual("https://security.example.com/CVE-2023-0001", res[0][3]) + self.assertEqual("3.0.2-0ubuntu1.6", res[0][4]) + self.assertEqual("3.0.2-0ubuntu1.7", res[0][5]) + self.assertEqual("high", res[0][6]) + self.assertEqual("needs-triage", res[0][7]) + self.assertEqual( + "https://quay.io/repository/org/myimage/manifest/sha256:abc123", + res[0][8], + ) + + def test_insert_into_pkg_patches(self): + """Test insert_into_pkg_patches()""" + db = cvelib.sql.CVEdb(":memory:") + db.create_tables() + self.conn = db.conn + self.cursor = self.conn.cursor() + + cand = "CVE-2023-0001" + pkg = cvelib.pkg.parse("upstream_foo: needed") + pkg.setPatches( + [ + "upstream: https://example.com/patch1", + "vendor: https://example.com/patch2", + ], + False, + ) + db.insert_into_pkg_patches(cand, pkg) + + self.cursor.execute("SELECT * FROM pkg_patches WHERE candidate=?", (cand,)) + res = self.cursor.fetchall() + self.assertEqual(2, len(res)) + self.assertEqual("upstream: https://example.com/patch1", res[0][5]) + self.assertEqual("vendor: https://example.com/patch2", res[1][5]) + + # empty patches + pkg2 = cvelib.pkg.parse("upstream_bar: needed") + db.insert_into_pkg_patches("CVE-2023-0002", pkg2) + self.cursor.execute( + "SELECT * FROM pkg_patches WHERE candidate=?", ("CVE-2023-0002",) + ) + res = self.cursor.fetchall() + self.assertEqual(0, len(res)) + + def test_insert_into_pkg_tags(self): + """Test insert_into_pkg_tags()""" + db = cvelib.sql.CVEdb(":memory:") + db.create_tables() + self.conn = db.conn + self.cursor = self.conn.cursor() + + cand = "CVE-2023-0001" + pkg = cvelib.pkg.parse("upstream_foo: needed") + pkg.setTags([("foo", "apparmor hardlink-restriction")]) + db.insert_into_pkg_tags(cand, pkg) + + self.cursor.execute("SELECT * FROM pkg_tags WHERE candidate=?", (cand,)) + res = self.cursor.fetchall() + self.assertEqual(2, len(res)) + self.assertEqual("foo", res[0][5]) + self.assertEqual("apparmor", res[0][6]) + self.assertEqual("foo", res[1][5]) + self.assertEqual("hardlink-restriction", res[1][6]) + + # empty tags + pkg2 = cvelib.pkg.parse("upstream_bar: needed") + db.insert_into_pkg_tags("CVE-2023-0002", pkg2) + self.cursor.execute( + "SELECT * FROM pkg_tags WHERE candidate=?", ("CVE-2023-0002",) + ) + res = self.cursor.fetchall() + self.assertEqual(0, len(res)) + + def test_insert_into_pkg_priorities(self): + """Test insert_into_pkg_priorities()""" + db = cvelib.sql.CVEdb(":memory:") + db.create_tables() + self.conn = db.conn + self.cursor = self.conn.cursor() + + cand = "CVE-2023-0001" + pkg = cvelib.pkg.parse("upstream_foo: needed") + pkg.setPriorities([("foo", "high"), ("other", "low")]) + db.insert_into_pkg_priorities(cand, pkg) + + self.cursor.execute("SELECT * FROM pkg_priorities WHERE candidate=?", (cand,)) + res = self.cursor.fetchall() + self.assertEqual(2, len(res)) + # Check both rows exist (order may vary by dict iteration) + priorities = {r[5]: r[6] for r in res} + self.assertEqual("high", priorities["foo"]) + self.assertEqual("low", priorities["other"]) + + # empty priorities + pkg2 = cvelib.pkg.parse("upstream_bar: needed") + db.insert_into_pkg_priorities("CVE-2023-0002", pkg2) + self.cursor.execute( + "SELECT * FROM pkg_priorities WHERE candidate=?", ("CVE-2023-0002",) + ) + res = self.cursor.fetchall() + self.assertEqual(0, len(res)) + + def test_insert_into_pkg_close_dates(self): + """Test insert_into_pkg_close_dates()""" + db = cvelib.sql.CVEdb(":memory:") + db.create_tables() + self.conn = db.conn + self.cursor = self.conn.cursor() + + cand = "CVE-2023-0001" + pkg = cvelib.pkg.parse("upstream_foo: needed") + pkg.setCloseDates([("foo", "2023-06-01")]) + db.insert_into_pkg_close_dates(cand, pkg) + + self.cursor.execute("SELECT * FROM pkg_close_dates WHERE candidate=?", (cand,)) + res = self.cursor.fetchall() + self.assertEqual(1, len(res)) + self.assertEqual("foo", res[0][5]) + self.assertEqual("2023-06-01", res[0][6]) + + # empty closeDates + pkg2 = cvelib.pkg.parse("upstream_bar: needed") + db.insert_into_pkg_close_dates("CVE-2023-0002", pkg2) + self.cursor.execute( + "SELECT * FROM pkg_close_dates WHERE candidate=?", ("CVE-2023-0002",) + ) + res = self.cursor.fetchall() self.assertEqual(0, len(res)) @mock.patch( @@ -427,8 +866,23 @@ def test_main_cve_query_show_schema(self): cvelib.sql.main_cve_query() self.assertEqual("", error.getvalue().strip()) - self.assertIn("CREATE TABLE 'cves'", output.getvalue()) - self.assertIn("CREATE TABLE 'pkgs'", output.getvalue()) + out = output.getvalue() + for table in [ + "cves", + "pkgs", + "cve_references", + "cve_bugs", + "cve_discovered_by", + "ghas_dependabot", + "ghas_secret", + "ghas_code", + "scan_oci", + "pkg_patches", + "pkg_tags", + "pkg_priorities", + "pkg_close_dates", + ]: + self.assertIn("CREATE TABLE '%s'" % table, out) @mock.patch( "sys.argv", @@ -664,5 +1118,129 @@ def test_main_cve_query_with_cve_data(self): cvelib.sql.main_cve_query() self.assertEqual("", error.getvalue().strip()) - self.assertIn("CREATE TABLE 'cves'", output.getvalue()) - self.assertIn("CREATE TABLE 'pkgs'", output.getvalue()) + out = output.getvalue() + self.assertIn("CREATE TABLE 'cves'", out) + self.assertIn("CREATE TABLE 'pkgs'", out) + + def test_main_cve_query_with_ghas_data(self): + """Test main_cve_query() - with GHAS data""" + _, cveDirs = self._setup_temp_config() + + cve_data = self._mock_cve_file("CVE-2023-8888") + cve_data["GitHub-Advanced-Security"] = ( + "\n" + " - type: dependabot\n" + " dependency: lodash\n" + " detectedIn: package-lock.json\n" + " advisory: https://github.com/advisories/GHSA-test-1234-5678\n" + " severity: high\n" + " status: needs-triage\n" + " url: https://github.com/org/repo/security/dependabot/1\n" + " - type: secret-scanning\n" + " secret: github_personal_access_token\n" + " detectedIn: config.yml\n" + " severity: critical\n" + " status: needs-triage\n" + " url: https://github.com/org/repo/security/secret-scanning/1\n" + " - type: code-scanning\n" + " description: SQL injection\n" + " detectedIn: src/app.py\n" + " severity: high\n" + " status: needs-triage\n" + " url: https://github.com/org/repo/security/code-scanning/1" + ) + cve_content = tests.testutil.cveContentFromDict(cve_data) + cve_fn = os.path.join(cveDirs["active"], "CVE-2023-8888") + with open(cve_fn, "w") as fp: + fp.write(cve_content) + + with mock.patch( + "sys.argv", + [ + "cve-query", + "--query", + "SELECT COUNT(*) FROM ghas_dependabot", + ], + ): + with tests.testutil.capturedOutput() as (output, error): + cvelib.sql.main_cve_query() + + self.assertEqual("", error.getvalue().strip()) + self.assertIn("1", output.getvalue()) + + def test_main_cve_query_with_ghas_data_unavailable(self): + """Test main_cve_query() - with GHAS data using unavailable URLs""" + _, cveDirs = self._setup_temp_config() + + cve_data = self._mock_cve_file("CVE-2023-8889") + cve_data["GitHub-Advanced-Security"] = ( + "\n" + " - type: dependabot\n" + " dependency: foo\n" + " detectedIn: go.sum\n" + " advisory: https://github.com/advisories/GHSA-a\n" + " severity: high\n" + " status: needed\n" + " url: unavailable\n" + " - type: dependabot\n" + " dependency: bar\n" + " detectedIn: go.sum\n" + " advisory: https://github.com/advisories/GHSA-b\n" + " severity: medium\n" + " status: needed\n" + " url: unavailable" + ) + cve_content = tests.testutil.cveContentFromDict(cve_data) + cve_fn = os.path.join(cveDirs["active"], "CVE-2023-8889") + with open(cve_fn, "w") as fp: + fp.write(cve_content) + + with mock.patch( + "sys.argv", + [ + "cve-query", + "--query", + "SELECT COUNT(*) FROM ghas_dependabot", + ], + ): + with tests.testutil.capturedOutput() as (output, error): + cvelib.sql.main_cve_query() + + self.assertEqual("", error.getvalue().strip()) + self.assertIn("2", output.getvalue()) + + def test_main_cve_query_with_scan_data(self): + """Test main_cve_query() - with scan report data""" + _, cveDirs = self._setup_temp_config() + + cve_data = self._mock_cve_file("CVE-2023-7777") + cve_data["Scan-Reports"] = ( + "\n" + " - type: oci\n" + " component: libssl3\n" + " detectedIn: Distro 1.0\n" + " advisory: https://www.cve.org/CVERecord?id=CVE-2023-7777\n" + " version: 3.0.2\n" + " fixedBy: 3.0.3\n" + " severity: high\n" + " status: needs-triage\n" + " url: https://quay.io/repository/org/myimage/manifest/sha256:abc123" + ) + cve_content = tests.testutil.cveContentFromDict(cve_data) + cve_fn = os.path.join(cveDirs["active"], "CVE-2023-7777") + with open(cve_fn, "w") as fp: + fp.write(cve_content) + + with mock.patch( + "sys.argv", + [ + "cve-query", + "--query", + "SELECT COUNT(*) FROM scan_oci", + ], + ): + with tests.testutil.capturedOutput() as (output, error): + cvelib.sql.main_cve_query() + + self.assertEqual("", error.getvalue().strip()) + self.assertIn("1", output.getvalue()) From 3cce778dffb3912c2b68cdd3c2174032dcab29c2 Mon Sep 17 00:00:00 2001 From: Jamie Strandboge Date: Tue, 24 Mar 2026 14:36:17 -0500 Subject: [PATCH 3/7] fix(check-syntax): reject duplicate entries at parse time Detect duplicates in setDiscoveredBy(), GHAS parse(), scan parse(), and setTags() raising CveException. --- cvelib/cve.py | 7 +++++ cvelib/github.py | 13 ++++++--- cvelib/pkg.py | 2 ++ cvelib/scan.py | 12 ++++++++- cvelib/sql.py | 2 +- tests/test_cve.py | 6 +++++ tests/test_github.py | 55 ++++++++++++++++++++++++++++++------- tests/test_pkg.py | 1 + tests/test_report.py | 4 +-- tests/test_scan.py | 64 +++++++++++++++++++++++++++++++++++++------- 10 files changed, 139 insertions(+), 27 deletions(-) diff --git a/cvelib/cve.py b/cvelib/cve.py index 73741ed..ee2f528 100644 --- a/cvelib/cve.py +++ b/cvelib/cve.py @@ -298,6 +298,13 @@ def setPriority(self, s: str) -> None: def setDiscoveredBy(self, s: str) -> None: """Set Discovered-by""" + seen: set = set() + for d in s.split(","): + d = d.strip() + if d != "": + if d in seen: + raise CveException("duplicate discoverer '%s'" % d) + seen.add(d) self.discoveredBy = s self.data["Discovered-by"] = self.discoveredBy diff --git a/cvelib/github.py b/cvelib/github.py index def52de..0e281f9 100644 --- a/cvelib/github.py +++ b/cvelib/github.py @@ -340,21 +340,28 @@ def parse(s: str) -> List[Union[GHDependabot, GHSecret, GHCode]]: raise CveException("invalid yaml:\n'%s'" % s) ghas: List[Union[GHDependabot, GHSecret, GHCode]] = [] + seen_urls: set = set() for item in yml: if "type" not in item: raise CveException("invalid GHAS document: 'type' missing for item") + obj: Union[GHDependabot, GHSecret, GHCode] if item["type"] == "dependabot": - ghas.append(GHDependabot(item)) + obj = GHDependabot(item) elif item["type"] == "secret-scanning": - ghas.append(GHSecret(item)) + obj = GHSecret(item) elif item["type"] == "code-scanning": - ghas.append(GHCode(item)) + obj = GHCode(item) else: raise CveException( "invalid GHAS document: unknown GHAS type '%s'" % item["type"] ) + if obj.url != "unavailable" and obj.url in seen_urls: + raise CveException("duplicate GHAS url '%s'" % obj.url) + seen_urls.add(obj.url) + ghas.append(obj) + return ghas diff --git a/cvelib/pkg.py b/cvelib/pkg.py index b58bbbc..cc46add 100644 --- a/cvelib/pkg.py +++ b/cvelib/pkg.py @@ -166,6 +166,8 @@ def setTags(self, tagList: List[Tuple[str, str]]) -> None: t = t.strip() if not rePatterns["pkg-tags"].search(t): raise CveException("invalid tag '%s'" % t) + if t in self.tags[tagKey]: + raise CveException("duplicate tag '%s'" % t) self.tags[tagKey].append(t) def setPriorities(self, priorityList: List[Tuple[str, str]]) -> None: diff --git a/cvelib/scan.py b/cvelib/scan.py index 5269a82..a859207 100644 --- a/cvelib/scan.py +++ b/cvelib/scan.py @@ -267,17 +267,27 @@ def parse(s: str) -> List[ScanOCI]: raise CveException("invalid yaml:\n'%s'" % s) mans: List[ScanOCI] = [] + seen_keys: set = set() for item in yml: if "type" not in item: raise CveException("invalid Scan-Reports document: 'type' missing for item") if item["type"] == "oci": - mans.append(ScanOCI(item)) + obj = ScanOCI(item) else: raise CveException( "invalid Scan-Reports document: unknown type '%s'" % item["type"] ) + key = (obj.component, obj.detectedIn, obj.advisory, obj.url) + if key in seen_keys: + raise CveException( + "duplicate scan entry '%s' '%s' '%s' '%s'" + % (obj.component, obj.detectedIn, obj.advisory, obj.url) + ) + seen_keys.add(key) + mans.append(obj) + return mans diff --git a/cvelib/sql.py b/cvelib/sql.py index 22d7bec..7407d36 100644 --- a/cvelib/sql.py +++ b/cvelib/sql.py @@ -9,7 +9,7 @@ import sqlite3 import sys import textwrap -from typing import Dict, List, Tuple, Union +from typing import Dict, List, Tuple from cvelib.common import ( _experimental, diff --git a/tests/test_cve.py b/tests/test_cve.py index 6056e3a..507d036 100644 --- a/tests/test_cve.py +++ b/tests/test_cve.py @@ -916,6 +916,12 @@ def test_setData(self): cvelib.cve.CVE().setData(hdrs) self.assertEqual("duplicate bug 'http://1'", str(context.exception)) + hdrs = self._mockHeaders(self._cve_template()) + hdrs["Discovered-by"] = "Alice (alice), Alice (alice)" + with self.assertRaises(cvelib.common.CveException) as context: + cvelib.cve.CVE().setData(hdrs) + self.assertEqual("duplicate discoverer 'Alice (alice)'", str(context.exception)) + def test_setDataPatchesKeys(self): """Test setData() - Patches_""" tsts = [ diff --git a/tests/test_github.py b/tests/test_github.py index 4174ae4..b6ceae0 100644 --- a/tests/test_github.py +++ b/tests/test_github.py @@ -877,16 +877,35 @@ def _getValidYaml(self): def test_parse(self): """Test parse()""" tsts = [ - # valid - (self._getValidYaml(), None), - ("", None), + # valid (yaml, expected_count, expected_error) + (self._getValidYaml(), 3, None), + ("", 0, None), + ( + """ - type: dependabot + dependency: foo + detectedIn: go.sum + advisory: https://github.com/advisories/GHSA-a + severity: medium + status: needed + url: unavailable + - type: dependabot + dependency: bar + detectedIn: go.sum + advisory: https://github.com/advisories/GHSA-b + severity: medium + status: needed + url: unavailable""", + 2, + None, + ), # invalid - (None, "invalid yaml:\n'None'"), - ("bad", "invalid GHAS document: 'type' missing for item"), + (None, 0, "invalid yaml:\n'None'"), + ("bad", 0, "invalid GHAS document: 'type' missing for item"), ( """ - type: other foo: bar baz: norf""", + 0, "invalid GHAS document: unknown GHAS type 'other'", ), ( @@ -897,17 +916,33 @@ def test_parse(self): severity: medium status: needed url: https://github.com/bar/baz/security/dependabot/1""", + 0, "invalid yaml: uses unquoted 'dependency: @...'", ), + ( + """ - type: dependabot + dependency: foo + detectedIn: go.sum + advisory: https://github.com/advisories/GHSA-a + severity: medium + status: needed + url: https://github.com/bar/baz/security/dependabot/1 + - type: dependabot + dependency: bar + detectedIn: go.sum + advisory: https://github.com/advisories/GHSA-b + severity: medium + status: needed + url: https://github.com/bar/baz/security/dependabot/1""", + 0, + "duplicate GHAS url 'https://github.com/bar/baz/security/dependabot/1'", + ), ] - for s, expErr in tsts: + for s, expLen, expErr in tsts: if expErr is None: res = cvelib.github.parse(s) - if s == "": - self.assertEqual(0, len(res)) - else: - self.assertEqual(3, len(res)) + self.assertEqual(expLen, len(res)) else: with self.assertRaises(cvelib.common.CveException) as context: cvelib.github.parse(s) diff --git a/tests/test_pkg.py b/tests/test_pkg.py index ed231e1..0e707ae 100644 --- a/tests/test_pkg.py +++ b/tests/test_pkg.py @@ -605,6 +605,7 @@ def test_setTags(self): [("test-key", "apparmor"), ("test-key_3", "pie bad")], "invalid tag 'bad'", ), + ([("test-key", "apparmor pie apparmor")], "duplicate tag 'apparmor'"), # ([("test-key", "")], "invalid tag 'bad'"), ] pkg = cvelib.pkg.CvePkg("git", "foo", "needed") diff --git a/tests/test_report.py b/tests/test_report.py index 666db2a..259c503 100644 --- a/tests/test_report.py +++ b/tests/test_report.py @@ -1000,10 +1000,10 @@ def _write_cve(cve_fn, d): - type: dependabot dependency: foo detectedIn: go.sum - advisory: https://github.com/advisories/GHSA-b + advisory: https://github.com/advisories/GHSA-d severity: high status: needed - url: https://github.com/org/bar/security/dependabot/2 + url: https://github.com/org/bar/security/dependabot/4 - type: dependabot dependency: corge detectedIn: go.sum diff --git a/tests/test_scan.py b/tests/test_scan.py index b509ff9..4c9e119 100644 --- a/tests/test_scan.py +++ b/tests/test_scan.py @@ -517,27 +517,71 @@ def test_diff(self): def test_parse(self): """Test parse()""" tsts = [ - # valid - (self._getValidYaml(), None), - ("", None), + # valid (yaml, expected_count, expected_error) + (self._getValidYaml(), 3, None), + ("", 0, None), + ( + """ - type: oci + component: foo + detectedIn: Distro 1.0 + advisory: https://www.cve.org/CVERecord?id=CVE-2023-0001 + version: 1.2.2 + fixedBy: 1.2.3 + severity: medium + status: needed + url: https://blah.com/BAR-a + - type: oci + component: bar + detectedIn: Distro 1.0 + advisory: https://www.cve.org/CVERecord?id=CVE-2023-0002 + version: 2.3.3 + fixedBy: 2.3.4 + severity: medium + status: needed + url: https://blah.com/BAR-a""", + 2, + None, + ), # invalid - (None, "invalid yaml:\n'None'"), - ("bad", "invalid Scan-Reports document: 'type' missing for item"), + (None, 0, "invalid yaml:\n'None'"), + ("bad", 0, "invalid Scan-Reports document: 'type' missing for item"), ( """ - type: other foo: bar baz: norf""", + 0, "invalid Scan-Reports document: unknown type 'other'", ), + ( + """ - type: oci + component: foo + detectedIn: Distro 1.0 + advisory: https://www.cve.org/CVERecord?id=CVE-2023-0001 + version: 1.2.2 + fixedBy: 1.2.3 + severity: medium + status: needed + url: https://blah.com/BAR-a + - type: oci + component: foo + detectedIn: Distro 1.0 + advisory: https://www.cve.org/CVERecord?id=CVE-2023-0001 + version: 2.3.3 + fixedBy: 2.3.4 + severity: medium + status: needed + url: https://blah.com/BAR-a""", + 0, + "duplicate scan entry 'foo' 'Distro 1.0'" + " 'https://www.cve.org/CVERecord?id=CVE-2023-0001'" + " 'https://blah.com/BAR-a'", + ), ] - for s, expErr in tsts: + for s, expLen, expErr in tsts: if expErr is None: res = cvelib.scan.parse(s) - if s == "": - self.assertEqual(0, len(res)) - else: - self.assertEqual(3, len(res)) + self.assertEqual(expLen, len(res)) else: with self.assertRaises(cvelib.common.CveException) as context: cvelib.scan.parse(s) From f33eb06688b572578a6de915e9b34137deae15f0 Mon Sep 17 00:00:00 2001 From: Jamie Strandboge Date: Tue, 24 Mar 2026 16:28:16 -0500 Subject: [PATCH 4/7] chore(sql): output header with csv --- cvelib/sql.py | 18 ++++++++----- tests/test_sql.py | 66 +++++++++++++++++++++++++++++++++++------------ 2 files changed, 62 insertions(+), 22 deletions(-) diff --git a/cvelib/sql.py b/cvelib/sql.py index 7407d36..4b908b7 100644 --- a/cvelib/sql.py +++ b/cvelib/sql.py @@ -581,7 +581,7 @@ def commit(self): """Commit the current transaction""" self.conn.commit() - def execute_query(self, q: str) -> List: + def execute_query(self, q: str) -> Tuple[List[str], List]: """Execute a read-only query using set_authorizer()""" def _readOnlyAuthorizer(action, arg1, arg2, dbname, trigger): @@ -608,12 +608,15 @@ def _readOnlyAuthorizer(action, arg1, arg2, dbname, trigger): cursor = self.conn.cursor() cursor.execute(q) results = cursor.fetchall() + columns = ( + [desc[0] for desc in cursor.description] if cursor.description else [] + ) except sqlite3.DatabaseError as e: print("Query error: %s" % e) - return [] + return [], [] finally: self.conn.set_authorizer(None) - return results + return columns, results def parse_dsn(dsn: str) -> Tuple: @@ -706,12 +709,15 @@ def convertCveDateToISO8601(cve_date: str, candidate: str) -> str: return iso_date -def print_results(res: List[Tuple], format: str) -> None: +def print_results(res: List[Tuple], format: str, columns: List[str]) -> None: if format == "raw": for r in res: print(r) else: # default to csv try: + if columns: + # use \r\n to match csv.writer line endings + sys.stdout.write("#%s\r\n" % ",".join(columns)) csv.writer(sys.stdout).writerows(res) except BrokenPipeError: # pragma: nocover pass @@ -918,11 +924,11 @@ def main_cve_query(): sql = fp.read() else: sql = args.query - res = db.execute_query(sql) + columns, res = db.execute_query(sql) supported_formats: List[str] = ["csv", "raw"] if args.output_format not in supported_formats: error( "Unsupported output format '%s'. Please use: %s" % (args.output_format, ", ".join(supported_formats)) ) - print_results(res, format=args.output_format) + print_results(res, format=args.output_format, columns=columns) diff --git a/tests/test_sql.py b/tests/test_sql.py index 7aed199..2ba3fcf 100644 --- a/tests/test_sql.py +++ b/tests/test_sql.py @@ -258,17 +258,30 @@ def test_convertCveDateToISO8601(self): def test_print_results(self): """Test print_results""" tsts = [ - # list, format, exp - ([], "csv", ""), - ([("foo", "bar")], "csv", "foo,bar"), - ([("foo", "bar"), ("baz", "quz")], "csv", "foo,bar\r\nbaz,quz"), - ([], "raw", ""), - ([("foo", "bar")], "raw", "('foo', 'bar')"), - ([("foo", "bar"), ("baz", "quz")], "raw", "('foo', 'bar')\n('baz', 'quz')"), + # list, format, columns, exp + ([], "csv", [], ""), + ([], "csv", ["a", "b"], "#a,b"), + ([("foo", "bar")], "csv", [], "foo,bar"), + ([("foo", "bar")], "csv", ["a", "b"], "#a,b\r\nfoo,bar"), + ( + [("foo", "bar"), ("baz", "quz")], + "csv", + ["a", "b"], + "#a,b\r\nfoo,bar\r\nbaz,quz", + ), + ([], "raw", [], ""), + ([("foo", "bar")], "raw", [], "('foo', 'bar')"), + ([("foo", "bar")], "raw", ["a", "b"], "('foo', 'bar')"), + ( + [("foo", "bar"), ("baz", "quz")], + "raw", + [], + "('foo', 'bar')\n('baz', 'quz')", + ), ] - for lst, fmt, exp in tsts: + for lst, fmt, cols, exp in tsts: with tests.testutil.capturedOutput() as (output, error): - cvelib.sql.print_results(lst, fmt) + cvelib.sql.print_results(lst, fmt, cols) self.assertEqual("", error.getvalue().strip()) self.assertEqual(exp, output.getvalue().strip()) @@ -397,7 +410,7 @@ def test_execute_query(self): db.insert_into_cves(cve) query = "SELECT * FROM 'cves'" - res = db.execute_query(query) + columns, res = db.execute_query(query) exp = ( cve.candidate, cve.openDate, @@ -413,40 +426,61 @@ def test_execute_query(self): ) self.assertEqual(1, len(res)) self.assertEqual(exp, res[0]) + self.assertEqual( + [ + "candidate", + "openDate", + "closeDate", + "publicDate", + "crd", + "description", + "notes", + "mitigation", + "priority", + "assignedTo", + "cvss", + ], + columns, + ) # invalid - write operations are denied by authorizer with tests.testutil.capturedOutput() as (output, error): - res = db.execute_query("DELETE FROM cves") + columns, res = db.execute_query("DELETE FROM cves") self.assertEqual(0, len(res)) + self.assertEqual(0, len(columns)) self.assertIn("Query error:", output.getvalue()) # malformed SQL with tests.testutil.capturedOutput() as (output, error): - res = db.execute_query("UPDATE...") + columns, res = db.execute_query("UPDATE...") self.assertEqual(0, len(res)) + self.assertEqual(0, len(columns)) self.assertIn("Query error:", output.getvalue()) # case-insensitive select works - res = db.execute_query("select * from 'cves'") + columns, res = db.execute_query("select * from 'cves'") self.assertEqual(1, len(res)) self.assertEqual(exp, res[0]) + self.assertTrue(len(columns) > 0) # INSERT denied with tests.testutil.capturedOutput() as (output, error): - res = db.execute_query( + columns, res = db.execute_query( "INSERT INTO cves (candidate) VALUES ('CVE-2023-HACK')" ) self.assertEqual(0, len(res)) + self.assertEqual(0, len(columns)) self.assertIn("Query error:", output.getvalue()) # DROP denied with tests.testutil.capturedOutput() as (output, error): - res = db.execute_query("DROP TABLE cves") + columns, res = db.execute_query("DROP TABLE cves") self.assertEqual(0, len(res)) + self.assertEqual(0, len(columns)) self.assertIn("Query error:", output.getvalue()) # verify data is unchanged after denied operations - res = db.execute_query("SELECT COUNT(*) FROM cves") + columns, res = db.execute_query("SELECT COUNT(*) FROM cves") self.assertEqual(1, res[0][0]) def test_insert_into_cve_references(self): From 2980576b992d90c79ca9660b88ca5e2d55ebf47f Mon Sep 17 00:00:00 2001 From: Jamie Strandboge Date: Tue, 24 Mar 2026 16:35:07 -0500 Subject: [PATCH 5/7] feat(sql): support --output-format json --- cvelib/sql.py | 7 +++++-- tests/test_sql.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/cvelib/sql.py b/cvelib/sql.py index 4b908b7..ba5e344 100644 --- a/cvelib/sql.py +++ b/cvelib/sql.py @@ -5,6 +5,7 @@ import argparse import csv import datetime +import json import os import sqlite3 import sys @@ -710,7 +711,9 @@ def convertCveDateToISO8601(cve_date: str, candidate: str) -> str: def print_results(res: List[Tuple], format: str, columns: List[str]) -> None: - if format == "raw": + if format == "json": + print(json.dumps([dict(zip(columns, row)) for row in res])) + elif format == "raw": for r in res: print(r) else: # default to csv @@ -925,7 +928,7 @@ def main_cve_query(): else: sql = args.query columns, res = db.execute_query(sql) - supported_formats: List[str] = ["csv", "raw"] + supported_formats: List[str] = ["csv", "json", "raw"] if args.output_format not in supported_formats: error( "Unsupported output format '%s'. Please use: %s" diff --git a/tests/test_sql.py b/tests/test_sql.py index 2ba3fcf..fb257d2 100644 --- a/tests/test_sql.py +++ b/tests/test_sql.py @@ -269,6 +269,15 @@ def test_print_results(self): ["a", "b"], "#a,b\r\nfoo,bar\r\nbaz,quz", ), + ([], "json", [], "[]"), + ([], "json", ["a", "b"], "[]"), + ([("foo", "bar")], "json", ["a", "b"], '[{"a": "foo", "b": "bar"}]'), + ( + [("foo", "bar"), ("baz", "quz")], + "json", + ["a", "b"], + '[{"a": "foo", "b": "bar"}, {"a": "baz", "b": "quz"}]', + ), ([], "raw", [], ""), ([("foo", "bar")], "raw", [], "('foo', 'bar')"), ([("foo", "bar")], "raw", ["a", "b"], "('foo', 'bar')"), @@ -994,6 +1003,26 @@ def test_main_cve_query_raw_output(self): self.assertEqual("", error.getvalue().strip()) self.assertIn("(0,)", output.getvalue()) + @mock.patch( + "sys.argv", + [ + "cve-query", + "--query", + "SELECT COUNT(*) FROM cves", + "--output-format", + "json", + ], + ) + def test_main_cve_query_json_output(self): + """Test main_cve_query() - json output format""" + self._setup_temp_config() + + with tests.testutil.capturedOutput() as (output, error): + cvelib.sql.main_cve_query() + + self.assertEqual("", error.getvalue().strip()) + self.assertEqual('[{"COUNT(*)": 0}]', output.getvalue().strip()) + @mock.patch( "sys.argv", [ From 321bf04aa43942c0ba15cd555d0adf89c98fe4d2 Mon Sep 17 00:00:00 2001 From: Jamie Strandboge Date: Tue, 24 Mar 2026 16:38:30 -0500 Subject: [PATCH 6/7] fix(sql): replace newlines with two character '\n' --- cvelib/sql.py | 14 +++++++++++++- tests/test_sql.py | 8 ++++++++ 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/cvelib/sql.py b/cvelib/sql.py index ba5e344..7991fc1 100644 --- a/cvelib/sql.py +++ b/cvelib/sql.py @@ -721,7 +721,19 @@ def print_results(res: List[Tuple], format: str, columns: List[str]) -> None: if columns: # use \r\n to match csv.writer line endings sys.stdout.write("#%s\r\n" % ",".join(columns)) - csv.writer(sys.stdout).writerows(res) + # flatten embedded newlines so each row is one terminal line + flat = [ + tuple( + ( + str(v).replace("\r\n", "\\n").replace("\n", "\\n") + if isinstance(v, str) + else v + ) + for v in row + ) + for row in res + ] + csv.writer(sys.stdout).writerows(flat) except BrokenPipeError: # pragma: nocover pass diff --git a/tests/test_sql.py b/tests/test_sql.py index fb257d2..b54379c 100644 --- a/tests/test_sql.py +++ b/tests/test_sql.py @@ -269,6 +269,14 @@ def test_print_results(self): ["a", "b"], "#a,b\r\nfoo,bar\r\nbaz,quz", ), + # csv flattens embedded newlines + ([("foo\nbar", "baz")], "csv", ["a", "b"], "#a,b\r\nfoo\\nbar,baz"), + ( + [("foo\r\nbar", "baz")], + "csv", + ["a", "b"], + "#a,b\r\nfoo\\nbar,baz", + ), ([], "json", [], "[]"), ([], "json", ["a", "b"], "[]"), ([("foo", "bar")], "json", ["a", "b"], '[{"a": "foo", "b": "bar"}]'), From 2088186001b351792c4a11eb7d893dbc6739291a Mon Sep 17 00:00:00 2001 From: Jamie Strandboge Date: Tue, 24 Mar 2026 16:55:19 -0500 Subject: [PATCH 7/7] feat(sql): support --output-format=markdown|markdown-full --- cvelib/sql.py | 67 ++++++++++++++++++++++++++++++++++++++++++----- tests/test_sql.py | 60 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 120 insertions(+), 7 deletions(-) diff --git a/cvelib/sql.py b/cvelib/sql.py index 7991fc1..28f8c15 100644 --- a/cvelib/sql.py +++ b/cvelib/sql.py @@ -710,9 +710,61 @@ def convertCveDateToISO8601(cve_date: str, candidate: str) -> str: return iso_date +def _flattenNewlines(v: str) -> str: + """Replace embedded newlines with literal \\n""" + return v.replace("\r\n", "\\n").replace("\n", "\\n") + + +def _markdownEscape(v: str) -> str: + """Escape pipe characters for markdown table cells""" + return v.replace("|", "\\|") + + +def _printMarkdown(res: List[Tuple], columns: List[str], ellipsize: bool) -> None: + """Print results as a markdown table""" + max_len: int = 40 + + def _fmt(v) -> str: + s = _markdownEscape(_flattenNewlines(str(v))) + if ellipsize and len(s) > max_len: + return s[: max_len - 3] + "..." + return s + + if ellipsize: + # pre-format all cells so we can compute column widths + formatted: List[List[str]] = [[_fmt(v) for v in row] for row in res] + ncols: int = len(columns) + widths: List[int] = [0] * ncols + for i, col in enumerate(columns): + widths[i] = len(col) + for row in formatted: + for i, cell in enumerate(row): + if len(cell) > widths[i]: + widths[i] = len(cell) + + hdr = ( + "| " + " | ".join(c.ljust(widths[i]) for i, c in enumerate(columns)) + " |" + ) + sep = "| " + " | ".join("-" * widths[i] for i in range(ncols)) + " |" + print(hdr) + print(sep) + for row in formatted: + print( + "| " + " | ".join(row[i].ljust(widths[i]) for i in range(ncols)) + " |" + ) + else: + if columns: + print("| %s |" % " | ".join(columns)) + print("| %s |" % " | ".join("---" for _ in columns)) + for row in res: + print("| %s |" % " | ".join(_fmt(v) for v in row)) + + def print_results(res: List[Tuple], format: str, columns: List[str]) -> None: if format == "json": print(json.dumps([dict(zip(columns, row)) for row in res])) + elif format in ("markdown", "markdown-full"): + _printMarkdown(res, columns, ellipsize=(format == "markdown")) elif format == "raw": for r in res: print(r) @@ -724,12 +776,7 @@ def print_results(res: List[Tuple], format: str, columns: List[str]) -> None: # flatten embedded newlines so each row is one terminal line flat = [ tuple( - ( - str(v).replace("\r\n", "\\n").replace("\n", "\\n") - if isinstance(v, str) - else v - ) - for v in row + _flattenNewlines(str(v)) if isinstance(v, str) else v for v in row ) for row in res ] @@ -940,7 +987,13 @@ def main_cve_query(): else: sql = args.query columns, res = db.execute_query(sql) - supported_formats: List[str] = ["csv", "json", "raw"] + supported_formats: List[str] = [ + "csv", + "json", + "markdown", + "markdown-full", + "raw", + ] if args.output_format not in supported_formats: error( "Unsupported output format '%s'. Please use: %s" diff --git a/tests/test_sql.py b/tests/test_sql.py index b54379c..a9dc1c5 100644 --- a/tests/test_sql.py +++ b/tests/test_sql.py @@ -286,6 +286,43 @@ def test_print_results(self): ["a", "b"], '[{"a": "foo", "b": "bar"}, {"a": "baz", "b": "quz"}]', ), + # markdown with ellipsize and column alignment + ([], "markdown", ["a", "b"], "| a | b |\n| - | - |"), + ( + [("foo", "bar")], + "markdown", + ["a", "b"], + "| a | b |\n| --- | --- |\n| foo | bar |", + ), + # markdown ellipsizes long values + ( + [("x" * 50, "bar")], + "markdown", + ["a", "b"], + "| a%s | b |\n| %s | --- |\n| %s... | bar |" + % (" " * 39, "-" * 40, "x" * 37), + ), + # markdown flattens newlines + ( + [("foo\nbar", "baz")], + "markdown", + ["a", "b"], + "| a%s | b |\n| %s | --- |\n| foo\\nbar | baz |" % (" " * 7, "-" * 8), + ), + # markdown escapes pipes + ( + [("a|b", "c")], + "markdown", + ["a", "b"], + "| a%s | b |\n| %s | - |\n| a\\|b | c |" % (" " * 3, "-" * 4), + ), + # markdown-full does not ellipsize + ( + [("x" * 50, "bar")], + "markdown-full", + ["a", "b"], + "| a | b |\n| --- | --- |\n| %s | bar |" % ("x" * 50), + ), ([], "raw", [], ""), ([("foo", "bar")], "raw", [], "('foo', 'bar')"), ([("foo", "bar")], "raw", ["a", "b"], "('foo', 'bar')"), @@ -1031,6 +1068,29 @@ def test_main_cve_query_json_output(self): self.assertEqual("", error.getvalue().strip()) self.assertEqual('[{"COUNT(*)": 0}]', output.getvalue().strip()) + @mock.patch( + "sys.argv", + [ + "cve-query", + "--query", + "SELECT COUNT(*) FROM cves", + "--output-format", + "markdown", + ], + ) + def test_main_cve_query_markdown_output(self): + """Test main_cve_query() - markdown output format""" + self._setup_temp_config() + + with tests.testutil.capturedOutput() as (output, error): + cvelib.sql.main_cve_query() + + self.assertEqual("", error.getvalue().strip()) + out = output.getvalue().strip() + self.assertIn("| COUNT(*) |", out) + self.assertIn("| -------- |", out) + self.assertIn("| 0 |", out) + @mock.patch( "sys.argv", [