diff --git a/.github/getTrending.py b/.github/getTrending.py
index 7509817b90a..3d73d9a7c34 100644
--- a/.github/getTrending.py
+++ b/.github/getTrending.py
@@ -36,6 +36,11 @@ MIN_STARS = 2
# short window and a hard cap.
LANDED_DAYS = 10
LANDED_ROWS = 10
+# Candidates are judged newest push first, one gate query at a time, until no
+# later push can place. The limit only matters when a flood of empty
+# placeholders keeps the rows from filling.
+LANDED_BATCH = 20
+LANDED_LIMIT = 200
NVD_API = "https://services.nvd.nist.gov/rest/json/cves/2.0?cveId="
# NVD allows five requests per thirty seconds unauthenticated, and only a
# couple of rows per run ever need one.
@@ -61,13 +66,18 @@ SEARCH_CTA_URL = f"https://raw.githubusercontent.com/{SLUG}/{SEARCH_CTA_REV}/doc
KEV_MARK = f'
'
-def search(query: str) -> list[dict]:
- """Every repository a search exposes, most recently updated first."""
+def search(query: str) -> tuple[int, list[dict]]:
+ """How many repositories a search matches, and every one it exposes, most
+ recently updated first."""
headers = {"Accept": "application/vnd.github+json", "User-Agent": USER_AGENT}
+ # The token authenticates the API only; it is never placed in the URL,
+ # written to disk, or printed, so it cannot leak through logs or the commit.
token = github_token()
if token:
headers["Authorization"] = f"Bearer {token}"
found: list[dict] = []
+ seen: set[str] = set()
+ total = 0
for page in range(1, SEARCH_LIMIT // SEARCH_PAGE + 1):
# The REST API reads sort and order. The website's s and o are ignored
# and leave the results in best-match order.
@@ -83,7 +93,10 @@ def search(query: str) -> list[dict]:
with request.urlopen(request.Request(url, headers=headers), timeout=30) as response:
payload = json.load(response)
except error.HTTPError as problem:
- if problem.code not in (403, 422, 503) or attempt == 2:
+ if problem.code not in {403, 429, 500, 502, 503, 504} or attempt == 2:
+ raise
+ except (error.URLError, TimeoutError, json.JSONDecodeError):
+ if attempt == 2:
raise
else:
# A search that runs out of time still answers, with whatever
@@ -93,11 +106,18 @@ def search(query: str) -> list[dict]:
if attempt == 2:
raise RuntimeError(f"GitHub returned incomplete results for {query}")
time.sleep(5 * (attempt + 1))
+ total = int(payload.get("total_count") or 0)
items = payload.get("items") or []
- found.extend(items)
- if not items or len(found) >= min(int(payload.get("total_count") or 0), SEARCH_LIMIT):
+ for repo in items:
+ # Results shift between pages while they are read, so a repository
+ # can come back twice; GitHub names are case-insensitive.
+ name = str(repo.get("full_name") or "").lower()
+ if name and name not in seen:
+ seen.add(name)
+ found.append(repo)
+ if not items or len(found) >= min(total, SEARCH_LIMIT):
break
- return found
+ return total, found
def nvd_description(cve: str) -> str:
@@ -156,7 +176,7 @@ def just_landed(token: str) -> list[dict]:
since = cutoff.isoformat()
# Landed means created in the window. An old repository with a new commit
# has not landed; a new PoC for a CVE from any year has.
- rows = search(f"CVE in:name created:>={since}")
+ _, rows = search(f"CVE in:name created:>={since}")
candidates, seen = [], set()
for repo in rows:
@@ -172,29 +192,34 @@ def just_landed(token: str) -> list[dict]:
seen.add(name.lower())
candidates.append(repo)
candidates.sort(key=lambda repo: str(repo.get("pushed_at") or ""), reverse=True)
- # Only the freshest are worth two API round trips each.
- candidates = candidates[:LANDED_ROWS * 4]
- searched = len(candidates)
- candidates, paths = qualifying_repositories(candidates, token)
- names = [str(repo["full_name"]) for repo in candidates]
- shipped = code_pushed(names, token, paths)
-
- fresh, hollow = [], 0
- for repo in candidates:
- name = str(repo["full_name"])
- pushed = shipped.get(name)
- if not pushed:
- hollow += 1
- continue
- if pushed[:10] < since:
- continue
- repo["_shipped"] = pushed
- fresh.append(repo)
- fresh.sort(key=lambda repo: repo["_shipped"], reverse=True)
+ pool = candidates[:LANDED_LIMIT]
+ fresh: list[dict] = []
+ judged = rejected = hollow = 0
+ for start in range(0, len(pool), LANDED_BATCH):
+ # An artifact date never follows its repository's last push, so once the
+ # last row is at least as new as the next push, nothing further down
+ # can place.
+ if len(fresh) >= LANDED_ROWS and (
+ str(pool[start].get("pushed_at") or "") <= fresh[LANDED_ROWS - 1]["_shipped"]
+ ):
+ break
+ batch = pool[start : start + LANDED_BATCH]
+ judged += len(batch)
+ accepted, paths = qualifying_repositories(batch, token)
+ rejected += len(batch) - len(accepted)
+ shipped = code_pushed([str(repo["full_name"]) for repo in accepted], token, paths)
+ for repo in accepted:
+ pushed = shipped.get(str(repo["full_name"]))
+ if not pushed:
+ hollow += 1
+ elif pushed[:10] >= since:
+ repo["_shipped"] = pushed
+ fresh.append(repo)
+ fresh.sort(key=lambda repo: repo["_shipped"], reverse=True)
print(
- f"just landed: {len(fresh)} qualified, "
- f"{searched - len(candidates)} rejected by artifact gate, "
+ f"just landed: {judged} of {len(candidates)} candidates judged, "
+ f"{len(fresh)} qualified, {rejected} rejected by artifact gate, "
f"{hollow} without an artifact commit"
)
return fresh[:LANDED_ROWS]
@@ -208,49 +233,7 @@ def search_year(year: int, since: str) -> tuple[int, list[dict]]:
[f'"CVE-{year}" in:name', f"stars:>{MIN_STARS}", f"pushed:>{since}"]
+ [f"language:{language}" for language in LANGUAGES]
)
- headers = {"Accept": "application/vnd.github+json", "User-Agent": USER_AGENT}
- # The token authenticates the API only; it is never placed in the URL,
- # written to disk, or printed, so it cannot leak through logs or the commit.
- token = github_token()
- if token:
- headers["Authorization"] = f"Bearer {token}"
-
- repositories: list[dict] = []
- seen: set[str] = set()
- total = 0
- for page in range(1, SEARCH_LIMIT // SEARCH_PAGE + 1):
- url = SEARCH_URL + "?" + parse.urlencode({
- "q": query,
- "s": "updated",
- "o": "desc",
- "per_page": SEARCH_PAGE,
- "page": page,
- })
- for attempt in range(3):
- try:
- with request.urlopen(
- request.Request(url, headers=headers), timeout=30
- ) as response:
- payload = json.load(response)
- break
- except error.HTTPError as exc:
- if exc.code not in {403, 429, 500, 502, 503, 504} or attempt == 2:
- raise
- except (error.URLError, TimeoutError, json.JSONDecodeError):
- if attempt == 2:
- raise
- time.sleep(5 * (attempt + 1))
-
- total = int(payload.get("total_count") or 0)
- found = payload.get("items") or []
- for repo in found:
- name = str(repo.get("full_name") or "")
- if name and name not in seen:
- seen.add(name)
- repositories.append(repo)
- if not found or len(repositories) >= min(total, SEARCH_LIMIT):
- break
- return total, repositories
+ return search(query)
def github_token() -> str:
@@ -259,6 +242,7 @@ def github_token() -> str:
def graphql(query: str, token: str) -> dict:
body = json.dumps({"query": query}).encode("utf-8")
+ aliases = set(re.findall(r"\b(r\d+)\s*:\s*repository\s*\(", query))
headers = {
"Accept": "application/json",
"Content-Type": "application/json",
@@ -272,13 +256,31 @@ def graphql(query: str, token: str) -> dict:
request.Request(GRAPHQL_URL, data=body, headers=headers, method="POST"),
timeout=45,
) as response:
- return json.load(response).get("data") or {}
+ payload = json.load(response)
except error.HTTPError as exc:
if exc.code not in {403, 429, 500, 502, 503, 504} or attempt == 2:
raise
except (error.URLError, TimeoutError, json.JSONDecodeError):
if attempt == 2:
raise
+ else:
+ data = payload.get("data") or {}
+ errors = payload.get("errors") or []
+ # Only a confirmed missing repository is safe to skip. Nested
+ # lookup failures and partial responses must not remove good rows.
+ failed = []
+ for item in errors:
+ path = item.get("path") or []
+ if not (
+ item.get("type") == "NOT_FOUND" and len(path) == 1
+ and path[0] in aliases and path[0] in data and data[path[0]] is None
+ ):
+ failed.append(item)
+ if not failed and aliases.issubset(data):
+ return data
+ if attempt == 2:
+ reason = (failed[0].get("message") or failed[0]) if failed else "missing repository data"
+ raise RuntimeError(f"GitHub GraphQL failed: {reason}")
time.sleep(4 * (attempt + 1))
return {}
@@ -458,7 +460,7 @@ def repository_summary(repo: dict, cve: str) -> str:
remainder,
)
if not remainder or placeholder:
- return nvd_description(cve) or summary
+ return cell(nvd_description(cve)) or summary
return summary
@@ -647,7 +649,7 @@ Advisory rows are `[URL, NVD reference tags]`.
| [ExploitDB](https://gitlab.com/exploit-database/exploitdb) | Archived exploits, mapped by their own CVE column |
| [Metasploit](https://github.com/rapid7/metasploit-framework) | Modules, best ranked first |
| [Vulhub](https://github.com/vulhub/vulhub) | Runnable vulnerable environments and reproduction steps |
-| [afrog](https://github.com/zan8in/afrog), [Vulnerability](https://github.com/tzwlhack/Vulnerability), [0day](https://github.com/helloexp/0day), [xray](https://github.com/chaitin/xray) | CVE-specific templates, code and reproduction guides inside multi-CVE repositories |
+| [afrog](https://github.com/zan8in/afrog), [Vulnerability](https://github.com/tzwlhack/Vulnerability), [0day](https://github.com/helloexp/0day), [xray](https://github.com/chaitin/xray), [Google Security Research](https://github.com/google/security-research), [GitHub Security Lab](https://github.com/github/securitylab) | CVE-specific templates, code and reproduction guides inside multi-CVE repositories |
| [EPSS](https://www.first.org/epss/) | Daily exploitation probability from FIRST |
| [CISA KEV](https://www.cisa.gov/known-exploited-vulnerabilities-catalog) | What is being exploited in the wild |
| [NVD](https://nvd.nist.gov/) | CVSS assessments and tagged vendor, third-party, patch and mitigation references |
@@ -657,7 +659,7 @@ Advisory rows are `[URL, NVD reference tags]`.
| Job | Cadence | Picks up |
| --- | --- | --- |
-| [Trending sweep](.github/workflows/hot_cves.yml) | hourly | Front-page repositories and prior-hour candidates added to the searchable index |
+| [Trending sweep](.github/workflows/hot_cves.yml) | hourly | Front-page repositories, also added to the searchable index |
| [CVE sync](.github/workflows/sync_cve_pocs.yml) | daily | New CVEs, CNA references and recently pushed GitHub repositories for every CVE year |
| [Metadata sync](.github/workflows/sync_metadata.yml) | daily plus weekly full pass | CVSS, advisories, rejected records and current CISA KEV status |
| [Nuclei sync](.github/workflows/sync_nuclei.yml) | daily | New templates and rating changes |
diff --git a/.github/workflows/hot_cves.yml b/.github/workflows/hot_cves.yml
index ef7c296d7ad..f89ed97cbd2 100644
--- a/.github/workflows/hot_cves.yml
+++ b/.github/workflows/hot_cves.yml
@@ -39,11 +39,14 @@ jobs:
run: python3 scripts/update_cves.py --lookback-days 1 --years 2 --skip-cvelist
- name: Rebuild the trending PoC table
+ id: trending
+ continue-on-error: true
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: python3 .github/getTrending.py
- name: Ingest the current candidates
+ if: steps.trending.outcome == 'success'
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: python3 scripts/sync_pocingithub.py --input index/trending.json
@@ -101,3 +104,7 @@ jobs:
env:
GH_TOKEN: ${{ github.token }}
run: gh workflow run site.yml --ref main
+
+ - name: Report trending refresh failure
+ if: steps.trending.outcome == 'failure'
+ run: exit 1
diff --git a/README.md b/README.md
index a4504b982c8..6f8ba7fc1f5 100644
--- a/README.md
+++ b/README.md
@@ -2,7 +2,7 @@
-[](https://github.com/0xMarcio/pocindex/commits/main) [](https://github.com/0xMarcio/pocindex/actions/workflows/hot_cves.yml) [](https://pocindex.io/) [](https://pocindex.io/) [](https://github.com/0xMarcio/pocindex/stargazers)
+[](https://github.com/0xMarcio/pocindex/commits/main) [](https://github.com/0xMarcio/pocindex/actions/workflows/hot_cves.yml) [](https://pocindex.io/) [](https://pocindex.io/) [](https://github.com/0xMarcio/pocindex/stargazers)
@@ -14,15 +14,15 @@
| Stars | Updated | Repository | Description |
| --- | --- | --- | --- |
-| 2⭐ | 49m ago | [CVE-2026-103648](https://github.com/EterNullSec/CVE-2026-103648) | CVE-2026-103648 / Path Traversal in image-downloader 4.3.0 / CVSS 9.1 Critical / CWE-22 / By EterNullSec |
+| 2⭐ | 1h ago | [CVE-2026-103648](https://github.com/EterNullSec/CVE-2026-103648) | CVE-2026-103648 / Path Traversal in image-downloader 4.3.0 / CVSS 9.1 Critical / CWE-22 / By EterNullSec |
| 0⭐ | 1h ago | [CVE-2026-57967](https://github.com/c0dem4sters/CVE-2026-57967) | CVE-2026-57967 PoC - Apache ActiveMQ Artemis unauthenticated session hijacking via SESSION_REATTACH… |
-| 0⭐ | 2h ago | [CVE-2026-21589-PoC-Exploit](https://github.com/tc4dy/CVE-2026-21589-PoC-Exploit) | CVE-2026-21589 - Atlassian Data Center Unauth Arbitrary File Read (CVSS 9.3) / Red/Blue Team suite. 2 tools:… |
+| 0⭐ | 3h ago | [CVE-2026-21589-PoC-Exploit](https://github.com/tc4dy/CVE-2026-21589-PoC-Exploit) | CVE-2026-21589 - Atlassian Data Center Unauth Arbitrary File Read (CVSS 9.3) / Red/Blue Team suite. 2 tools:… |
| 0⭐ | 3h ago | [Zenphoto-1.4.1.4-CVE-2011-4825-RCE](https://github.com/XavLimSG/Zenphoto-1.4.1.4-CVE-2011-4825-RCE) | Python 3 reverse-shell exploit for Zenphoto <= 1.4.1.4 / CVE-2011-4825 (EDB-18083) |
| 1⭐ | 5h ago | [CVE-2026-3888-snap-confine-privilege-escalation](https://github.com/AlanNewberry/CVE-2026-3888-snap-confine-privilege-escalation) | Exploit para CVE-2026-3888: race condition en snap-confine con hijack del loader para escalar a root en Linux. |
| 5⭐ | 5h ago | [CVE-2026-21589](https://github.com/MarcusProgram/CVE-2026-21589) | h3. Summary This is a vulnerability in Bitbucket Data Center, Confluence Data Center, Jira Service Management… |
-| 0⭐ | 6h ago | [CVE-2026-67401](https://github.com/hitechcloud-vietnam/CVE-2026-67401) | PoC for CVE-2026-67401 - cPanel/WHM EmailTrack SQL injection leading to arbitrary file write and root RCE.… |
+| 0⭐ | 7h ago | [CVE-2026-67401](https://github.com/hitechcloud-vietnam/CVE-2026-67401) | PoC for CVE-2026-67401 - cPanel/WHM EmailTrack SQL injection leading to arbitrary file write and root RCE.… |
| 0⭐ | 7h ago | 
[cve-2026-41940-PoC](https://github.com/hitechcloud-vietnam/cve-2026-41940-PoC) | A tool for exploiting CVE-2026-41940, a critical authentication bypass in cPanel & WHM (CVSS 10.0), allowing… |
-| 0⭐ | 8h ago | [CVE-2025-34069](https://github.com/cppghoul/CVE-2025-34069) | Kerio Control 9.4.5 Auth Bypass PoC |
+| 0⭐ | 9h ago | [CVE-2025-34069](https://github.com/cppghoul/CVE-2025-34069) | Kerio Control 9.4.5 Auth Bypass PoC |
| 2⭐ | 10h ago | [watchTowr-vs-Atlassian-CVE-2026-21589](https://github.com/watchtowrlabs/watchTowr-vs-Atlassian-CVE-2026-21589) | h3. Summary This is a vulnerability in Bitbucket Data Center, Confluence Data Center, Jira Service Management… |
## Trending in 2026
@@ -142,7 +142,7 @@ Advisory rows are `[URL, NVD reference tags]`.
| [ExploitDB](https://gitlab.com/exploit-database/exploitdb) | Archived exploits, mapped by their own CVE column |
| [Metasploit](https://github.com/rapid7/metasploit-framework) | Modules, best ranked first |
| [Vulhub](https://github.com/vulhub/vulhub) | Runnable vulnerable environments and reproduction steps |
-| [afrog](https://github.com/zan8in/afrog), [Vulnerability](https://github.com/tzwlhack/Vulnerability), [0day](https://github.com/helloexp/0day), [xray](https://github.com/chaitin/xray) | CVE-specific templates, code and reproduction guides inside multi-CVE repositories |
+| [afrog](https://github.com/zan8in/afrog), [Vulnerability](https://github.com/tzwlhack/Vulnerability), [0day](https://github.com/helloexp/0day), [xray](https://github.com/chaitin/xray), [Google Security Research](https://github.com/google/security-research), [GitHub Security Lab](https://github.com/github/securitylab) | CVE-specific templates, code and reproduction guides inside multi-CVE repositories |
| [EPSS](https://www.first.org/epss/) | Daily exploitation probability from FIRST |
| [CISA KEV](https://www.cisa.gov/known-exploited-vulnerabilities-catalog) | What is being exploited in the wild |
| [NVD](https://nvd.nist.gov/) | CVSS assessments and tagged vendor, third-party, patch and mitigation references |
@@ -152,7 +152,7 @@ Advisory rows are `[URL, NVD reference tags]`.
| Job | Cadence | Picks up |
| --- | --- | --- |
-| [Trending sweep](.github/workflows/hot_cves.yml) | hourly | Front-page repositories and prior-hour candidates added to the searchable index |
+| [Trending sweep](.github/workflows/hot_cves.yml) | hourly | Front-page repositories, also added to the searchable index |
| [CVE sync](.github/workflows/sync_cve_pocs.yml) | daily | New CVEs, CNA references and recently pushed GitHub repositories for every CVE year |
| [Metadata sync](.github/workflows/sync_metadata.yml) | daily plus weekly full pass | CVSS, advisories, rejected records and current CISA KEV status |
| [Nuclei sync](.github/workflows/sync_nuclei.yml) | daily | New templates and rating changes |
diff --git a/docs/logic.js b/docs/logic.js
index 19ea28a3596..832ae352e8e 100644
--- a/docs/logic.js
+++ b/docs/logic.js
@@ -183,7 +183,9 @@ function pocText(entry) {
function entryLinks(entry) {
if (entry._allLinks === undefined) {
- entry._allLinks = uniqueSourceLinks(POC_FIELDS.flatMap(field => entry[field] || []));
+ entry._allLinks = uniqueSourceLinks(
+ POC_FIELDS.flatMap(field => entry[field] || []),
+ POC_FIELDS.slice(1).flatMap(field => entry[field] || []));
}
return entry._allLinks;
}
@@ -804,7 +806,7 @@ function rankedLinks(entry) {
...(entry.nuclei || []), ...(entry.msf || []), ...(entry.edb || []), ...(entry.vulhub || []),
...(entry.collections || []),
...scored.map(item => item.url)
- ]);
+ ], POC_FIELDS.slice(1).flatMap(field => entry[field] || []));
return entry._ranked;
}
@@ -829,16 +831,25 @@ function advisoryRow(item) {
'';
}
-function normalizedLink(url) {
+function normalizedLink(url, preservePath = false) {
const parsed = repoFromUrl(String(url || ''));
- if (parsed) return `github:${parsed.owner.toLowerCase()}/${parsed.repo.toLowerCase()}`;
+ if (parsed) {
+ const key = `github:${parsed.owner.toLowerCase()}/${parsed.repo.toLowerCase()}`;
+ if (!preservePath) return key;
+ const location = new URL(url);
+ const path = location.pathname.split('/').filter(Boolean).slice(2).join('/');
+ return `${key}/${path}${location.search}${location.hash}`;
+ }
return String(url || '').replace(/\/$/, '');
}
-function uniqueSourceLinks(urls) {
+function uniqueSourceLinks(urls, curated = []) {
const seen = new Set();
+ const artifacts = new Set(curated);
+ const sources = new Set(curated.map(url => normalizedLink(url)));
return urls.filter(url => {
- const key = normalizedLink(url);
+ if (!artifacts.has(url) && sources.has(normalizedLink(url))) return false;
+ const key = normalizedLink(url, artifacts.has(url));
if (seen.has(key)) return false;
seen.add(key);
return true;
@@ -859,7 +870,7 @@ function resultRow(entry) {
: '';
const advisoryOpen = state.advisoryOpen.has(id);
- const pocLinks = new Set(entryLinks(entry).map(normalizedLink));
+ const pocLinks = new Set(entryLinks(entry).map(url => normalizedLink(url)));
const advisories = ((metadata[id] || {}).advisories || [])
.filter(item => Array.isArray(item) && !pocLinks.has(normalizedLink(item[0])));
const shownAdvisories = advisoryOpen ? advisories : advisories.slice(0, ADVISORY_PREVIEW);
diff --git a/scripts/audit_poc_links.py b/scripts/audit_poc_links.py
index 3b0dc43ef9d..c53cbbf9885 100644
--- a/scripts/audit_poc_links.py
+++ b/scripts/audit_poc_links.py
@@ -326,6 +326,8 @@ def refresh_kev(*, dry_run: bool) -> int:
vulnerabilities = payload["vulnerabilities"]
if type(payload.get("count")) is not int or payload["count"] != len(vulnerabilities):
raise ValueError("KEV count does not match its vulnerability list")
+ if not vulnerabilities:
+ raise ValueError("Empty KEV catalogue")
entries = {}
for item in vulnerabilities:
if not isinstance(item, dict):
diff --git a/scripts/build_site.py b/scripts/build_site.py
index cbe595868f5..844b1f68174 100644
--- a/scripts/build_site.py
+++ b/scripts/build_site.py
@@ -245,11 +245,11 @@ def poc_link_rank(url: str, cve_id: str) -> tuple[int, int]:
return 4, len(url)
-def dedupe_source_links(urls: Iterable[str], cve_id: str) -> List[str]:
+def dedupe_source_links(urls: Iterable[str], cve_id: str, *, preserve_paths: bool = False) -> List[str]:
selected: Dict[str, tuple[int, str]] = {}
order: Dict[str, int] = {}
for index, url in enumerate(urls):
- key = source_key(url)
+ key = link_key(url) if preserve_paths else source_key(url)
order.setdefault(key, index)
current = selected.get(key)
if current is None or poc_link_rank(url, cve_id) < poc_link_rank(current[1], cve_id):
@@ -358,6 +358,7 @@ def build_cve_list(blacklist: Collection[str]) -> tuple[List[Dict[str, object]],
field: dedupe_source_links(
collect_links(sections.get(header, ""), blacklist=blacklist),
cve_id,
+ preserve_paths=True,
)
for header, field in CURATED
}
diff --git a/tests/test_audit_poc_links.py b/tests/test_audit_poc_links.py
index 9c3327f6c01..133ff602f9c 100644
--- a/tests/test_audit_poc_links.py
+++ b/tests/test_audit_poc_links.py
@@ -94,6 +94,7 @@ class KevRefreshTests(unittest.TestCase):
def test_incomplete_or_invalid_catalogue_preserves_stored_flags(self) -> None:
row = self.row()
invalid = [
+ {"count": 0, "vulnerabilities": []},
{"count": 2, "vulnerabilities": [row]},
{"count": 2, "vulnerabilities": [row, row]},
{"count": 1, "vulnerabilities": [self.row("NOT-A-CVE")]},
@@ -110,19 +111,22 @@ class KevRefreshTests(unittest.TestCase):
audit_poc_links, "http_json", return_value=payload
):
audit_poc_links.refresh_kev(dry_run=False)
- self.assertEqual(path.read_text(), old)
+ self.assertEqual(path.read_bytes(), old.encode("utf-8"))
- def test_complete_catalogue_can_remove_entries(self) -> None:
+ def test_complete_nonempty_catalogue_can_add_and_remove_entries(self) -> None:
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "kev.json"
- for rows in ([self.row()], []):
- path.write_text('{"CVE-2026-10000":["2026-01-01",0]}\n', encoding="utf-8")
- with self.subTest(rows=rows), patch.object(audit_poc_links, "KEV_FILE", path), patch.object(
- audit_poc_links, "http_json", return_value={"count": len(rows), "vulnerabilities": rows}
- ):
- self.assertEqual(audit_poc_links.refresh_kev(dry_run=False), len(rows))
- expected = {"CVE-2026-12345": ["2026-10-06", 1]} if rows else {}
- self.assertEqual(json.loads(path.read_text()), expected)
+ path.write_text(json.dumps({
+ "CVE-2026-10000": ["2026-01-01", 0], "CVE-2026-12345": ["2026-01-02", 0],
+ }), encoding="utf-8")
+ rows = [self.row(), self.row("CVE-2026-12346")]
+ with patch.object(audit_poc_links, "KEV_FILE", path), patch.object(
+ audit_poc_links, "http_json", return_value={"count": len(rows), "vulnerabilities": rows}
+ ):
+ self.assertEqual(audit_poc_links.refresh_kev(dry_run=False), len(rows))
+ self.assertEqual(json.loads(path.read_text()), {
+ "CVE-2026-12345": ["2026-10-06", 1], "CVE-2026-12346": ["2026-10-06", 1],
+ })
def test_failed_atomic_replace_preserves_catalogue(self) -> None:
with tempfile.TemporaryDirectory() as directory:
diff --git a/tests/test_get_trending.py b/tests/test_get_trending.py
index 791d4f59c66..36aff3d9cad 100644
--- a/tests/test_get_trending.py
+++ b/tests/test_get_trending.py
@@ -4,6 +4,7 @@ import importlib.util
import io
import json
import re
+import sys
import tempfile
import unittest
from contextlib import redirect_stdout
@@ -16,9 +17,14 @@ ROOT = Path(__file__).resolve().parents[1]
SPEC = importlib.util.spec_from_file_location("getTrending", str(ROOT / ".github" / "getTrending.py"))
trending = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(trending)
+sys.path.insert(0, str(ROOT / "scripts"))
+
+import sync_collections
NOW = datetime.now(timezone.utc)
CVE = f"CVE-{NOW.year}-21589"
+ALIAS = re.compile(r'(r\d+): repository\(owner: "[^"]*", name: "([^"]*)"\)')
+EMPTY = {"readmeMd": {"text": "Soon"}, "root": {"entries": [{"name": "README.md", "type": "blob"}]}}
def stamp(days: float) -> str:
@@ -44,7 +50,7 @@ def gate(query: str, token: str) -> dict:
"readmeMd": {"text": f"Proof of concept for {name}"},
"root": {"entries": [{"name": "exploit.py", "type": "blob"}]},
}
- for alias, name in re.findall(r'(r\d+): repository\(owner: "[^"]*", name: "([^"]*)"\)', query)
+ for alias, name in ALIAS.findall(query)
}
@@ -52,6 +58,38 @@ def shipped(names: list[str], token: str, paths: dict | None = None) -> dict[str
return {name: stamp(0.2) for name in names}
+def run_landed(rows: list[dict], dates: dict[str, str]) -> tuple[list[str], list[str]]:
+ """Run the landed lane over search rows. A repository with a code date in
+ dates carries an exploit; any other is an empty placeholder the real gate
+ rejects. Returns the listed names and every name the gate was asked about."""
+ asked: list[str] = []
+
+ def graphql(query: str, token: str) -> dict:
+ found = gate(query, token)
+ for alias, name in ALIAS.findall(query):
+ asked.append(name)
+ if name not in dates:
+ found[alias] = EMPTY
+ return found
+
+ def pushed(names: list[str], token: str, paths: dict | None = None) -> dict[str, str]:
+ return {name: dates[name.split("/", 1)[1]] for name in names}
+
+ with patch.multiple(trending, search=lambda query: (len(rows), rows), graphql=graphql, code_pushed=pushed), \
+ redirect_stdout(io.StringIO()):
+ listed = trending.just_landed("")
+ return [found["name"] for found in listed], asked
+
+
+def pushed_rows(owner: str, first: int, count: int, created: float, newest: float) -> list[dict]:
+ """Repositories created days ago, last pushed from newest days ago, older by
+ a quarter hour each."""
+ return [
+ dict(repo(f"{owner}/CVE-2025-{first + i}", stamp(created)), pushed_at=stamp(newest + 0.01 * i))
+ for i in range(count)
+ ]
+
+
class JustLandedTests(unittest.TestCase):
def test_lists_new_repositories_for_any_cve_year_once(self) -> None:
rows = [
@@ -65,9 +103,9 @@ class JustLandedTests(unittest.TestCase):
]
queries: list[str] = []
- def search(query: str) -> list[dict]:
+ def search(query: str) -> tuple[int, list[dict]]:
queries.append(query)
- return rows
+ return len(rows), rows
with patch.multiple(trending, search=search, graphql=gate, code_pushed=shipped), \
redirect_stdout(io.StringIO()):
@@ -81,6 +119,33 @@ class JustLandedTests(unittest.TestCase):
self.assertIn("created:>=", queries[0])
self.assertNotRegex(queries[0], r"pushed:|CVE-\d{4}")
+ def test_empty_placeholders_cannot_crowd_out_fresh_pocs(self) -> None:
+ empty = pushed_rows("spam", 5000, 45, created=1, newest=0.01)
+ real = pushed_rows("acme", 1000, 12, created=3, newest=1)
+ listed, asked = run_landed(empty + real, {row["name"]: row["pushed_at"] for row in real})
+ self.assertEqual(listed, [row["name"] for row in real[:trending.LANDED_ROWS]])
+ self.assertEqual(len(asked), len(empty + real))
+
+ def test_judging_stops_once_no_later_push_can_place(self) -> None:
+ rows = pushed_rows("acme", 1000, 30, created=3, newest=0.01)
+ listed, asked = run_landed(rows, {row["name"]: row["pushed_at"] for row in rows})
+ self.assertEqual(listed, [row["name"] for row in rows[:trending.LANDED_ROWS]])
+ self.assertEqual(len(asked), trending.LANDED_BATCH)
+
+ def test_a_later_push_that_can_still_place_is_judged(self) -> None:
+ # The newest pushes only touched paperwork; their code is days old.
+ touched = pushed_rows("acme", 1000, 20, created=6, newest=0.01)
+ newer = pushed_rows("acme", 2000, 10, created=3, newest=1)
+ dates = {row["name"]: stamp(5) for row in touched}
+ dates.update({row["name"]: row["pushed_at"] for row in newer})
+ listed, asked = run_landed(touched + newer, dates)
+ self.assertEqual(listed, [row["name"] for row in newer])
+ self.assertEqual(len(asked), 30)
+
+ def test_judging_is_capped(self) -> None:
+ rows = pushed_rows("spam", 5000, trending.LANDED_LIMIT + 50, created=3, newest=0.001)
+ self.assertEqual(run_landed(rows, {}), ([], [row["name"] for row in rows[:trending.LANDED_LIMIT]]))
+
class MainTests(unittest.TestCase):
def test_new_trending_repository_is_in_both_sections_and_written_once(self) -> None:
@@ -96,7 +161,7 @@ class MainTests(unittest.TestCase):
figures=lambda: {"total_cves": 3, "with_pocs": 2, "kev": 1},
known_exploited=lambda: {CVE},
search_year=lambda year, since: (2, [dict(both), dict(older)]) if year == NOW.year else (0, []),
- search=lambda query: [dict(fresh), dict(both), dict(older)],
+ search=lambda query: (3, [dict(fresh), dict(both), dict(older)]),
graphql=gate,
code_pushed=shipped,
), redirect_stdout(io.StringIO()) as log:
@@ -118,6 +183,17 @@ class MainTests(unittest.TestCase):
)
self.assertIn("(3 repositories, 1 known-exploited)", log.getvalue())
+ def test_every_ingested_collection_is_credited(self) -> None:
+ for source in (*sync_collections.SOURCES, *sync_collections.TREE_SOURCES):
+ self.assertIn(f"](https://github.com/{source[1]})", trending.FOOTER)
+
+ def test_cve_summary_is_cleaned_like_repository_text(self) -> None:
+ with patch.object(trending, "nvd_description", return_value="Path traversal \N{EM DASH} remote | unauthenticated"):
+ self.assertEqual(
+ trending.repository_summary({"description": CVE}, CVE),
+ "Path traversal - remote / unauthenticated",
+ )
+
class ArtifactDateTests(unittest.TestCase):
def test_commit_clock_cannot_postdate_the_repository_push(self) -> None:
@@ -138,6 +214,56 @@ class ArtifactDateTests(unittest.TestCase):
self.assertEqual(trending.time_ago(stamp(-0.1)), "just now")
+class GraphQLTests(unittest.TestCase):
+ def answer(self, payload: dict):
+ return lambda req, timeout: io.BytesIO(json.dumps(payload).encode())
+
+ def query(self) -> str:
+ return "query { " + " ".join(
+ trending.repository_alias(index, name, "name")
+ for index, name in enumerate(["owner/missing", "owner/healthy"])
+ ) + " }"
+
+ def test_partial_operational_failures_retry_then_abort(self) -> None:
+ for error in (
+ {"type": "INTERNAL", "path": ["r0"]},
+ {"type": "RATE_LIMITED", "path": ["r0"]},
+ {"type": "NOT_FOUND", "path": ["r0", "root"]},
+ ):
+ payload = {"data": {"r0": None, "r1": {"name": "healthy"}}, "errors": [error]}
+ with self.subTest(error=error), patch.object(trending.time, "sleep"), patch.object(
+ trending.request, "urlopen", side_effect=self.answer(payload)
+ ) as fetch:
+ with self.assertRaises(RuntimeError):
+ trending.graphql(self.query(), "test")
+ self.assertEqual(fetch.call_count, 3)
+
+ def test_missing_alias_is_not_a_deleted_repository(self) -> None:
+ with patch.object(trending.time, "sleep"), patch.object(
+ trending.request, "urlopen", side_effect=self.answer({"data": {"r1": {"name": "healthy"}}})
+ ) as fetch:
+ with self.assertRaises(RuntimeError):
+ trending.graphql(self.query(), "test")
+ self.assertEqual(fetch.call_count, 3)
+
+ def test_failed_query_is_not_read_as_repositories_without_code(self) -> None:
+ limited = {"errors": [{"type": "RATE_LIMITED", "message": "API rate limit exceeded"}]}
+ with patch.object(trending.time, "sleep"), patch.object(trending.request, "urlopen", self.answer(limited)):
+ with self.assertRaisesRegex(RuntimeError, "rate limit"):
+ trending.code_pushed(["owner/repo"], "test", {"owner/repo": ["exploit.py"]})
+
+ def test_deleted_repository_leaves_only_that_one_out(self) -> None:
+ kept = {"pushedAt": stamp(1), "defaultBranchRef": {"target": {"p0": {"nodes": []}}}}
+ payload = {
+ "data": {"r0": None, "r1": kept},
+ "errors": [
+ {"type": "NOT_FOUND", "path": ["r0"], "message": "Could not resolve to a Repository"},
+ ],
+ }
+ with patch.object(trending.request, "urlopen", self.answer(payload)):
+ self.assertEqual(trending.graphql(self.query(), "test"), {"r0": None, "r1": kept})
+
+
class SearchTests(unittest.TestCase):
def replies(self, *payloads: dict) -> tuple[list[str], object]:
urls: list[str] = []
@@ -156,7 +282,8 @@ class SearchTests(unittest.TestCase):
{"total_count": 101, "incomplete_results": False, "items": [{"full_name": "a/100"}]},
)
with patch.object(trending.request, "urlopen", urlopen):
- self.assertEqual(len(trending.search("CVE in:name")), 101)
+ total, found = trending.search("CVE in:name")
+ self.assertEqual((total, len(found)), (101, 101))
params = [parse_qs(urlparse(url).query) for url in urls]
self.assertEqual(
[(p["sort"], p["order"], p["page"]) for p in params],
@@ -169,11 +296,30 @@ class SearchTests(unittest.TestCase):
with patch.object(trending.time, "sleep"):
_, urlopen = self.replies(partial, complete)
with patch.object(trending.request, "urlopen", urlopen):
- self.assertEqual(trending.search("CVE in:name"), [{"full_name": "a/b"}])
+ self.assertEqual(trending.search("CVE in:name"), (1, [{"full_name": "a/b"}]))
_, urlopen = self.replies(partial, partial, partial)
with patch.object(trending.request, "urlopen", urlopen), self.assertRaises(RuntimeError):
trending.search("CVE in:name")
+ def test_year_search_keeps_its_filters_and_lists_each_repository_once(self) -> None:
+ urls, urlopen = self.replies(
+ {"total_count": 101, "incomplete_results": False,
+ "items": [{"full_name": f"a/{n}"} for n in range(100)]},
+ # An update during the read shifts the pages; the boundary row comes back.
+ {"total_count": 101, "incomplete_results": False,
+ "items": [{"full_name": "A/99"}, {"full_name": "a/100"}]},
+ )
+ with patch.object(trending.request, "urlopen", urlopen):
+ total, found = trending.search_year(2026, "2026-07-08")
+ self.assertEqual((total, len(found), found[-1]["full_name"]), (101, 101, "a/100"))
+ params = parse_qs(urlparse(urls[0]).query)
+ self.assertEqual((params["sort"], params["order"]), (["updated"], ["desc"]))
+ self.assertNotIn("s", params)
+ self.assertEqual(
+ params["q"][0].split()[:4],
+ ['"CVE-2026"', "in:name", "stars:>2", "pushed:>2026-07-08"],
+ )
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_search.js b/tests/test_search.js
index 316bc554a2f..8cfd9e8acce 100644
--- a/tests/test_search.js
+++ b/tests/test_search.js
@@ -56,6 +56,29 @@ test('repository security advisories have GHSA filtering without repository star
assert.equal(vm.runInContext(`repoFromUrl(${JSON.stringify(url)})`, context), null);
});
+test('curated artifact paths survive search rendering without duplicate repository roots', async t => {
+ const { context } = startPage(t, { trending: new Promise(() => {}) });
+ await new Promise(setImmediate);
+ const base = 'https://github.com/google/security-research';
+ const variants = ['lts', 'mitigation'].map(kind => `${base}/tree/master/pocs/linux/kernelctf/CVE-2024-26642_${kind}`);
+ const template = 'https://github.com/projectdiscovery/nuclei-templates/blob/main/http/Test.yaml?version=One#Request';
+ const entry = { cve: 'CVE-2024-26642', poc: [base, 'https://github.com/example/poc', 'https://github.com/EXAMPLE/POC/blob/main/exploit.py'],
+ collections: [...variants, variants[0].replace('google/security-research', 'GOOGLE/Security-Research')],
+ nuclei: [template, template.replace('Test.yaml', 'test.yaml')] };
+ context.variantEntry = entry;
+ vm.runInContext(`metadata = { 'CVE-2024-26642': { advisories: [[${JSON.stringify(base)}, ['Exploit']]] } };`, context);
+ for (const fn of ['entryLinks', 'rankedLinks']) {
+ const links = Array.from(vm.runInContext(`${fn}(variantEntry)`, context));
+ assert.equal(links.length, 5);
+ for (const url of [...variants, template, template.replace('Test.yaml', 'test.yaml')]) assert.ok(links.includes(url), url);
+ assert.ok(!links.includes(base));
+ }
+ const html = vm.runInContext('resultRow(variantEntry)', context);
+ for (const url of variants) assert.ok(html.includes(`href="${url}"`));
+ assert.doesNotMatch(html, /CREDIBLE ADVISORIES/);
+ assert.equal(vm.runInContext(`normalizedLink(${JSON.stringify(variants[0])}) === normalizedLink(${JSON.stringify(base)})`, context), true);
+});
+
test('a stalled trending feed does not block a bookmarked search', async t => {
const { elements, calls } = startPage(t, { search: '?q=widget', trending: new Promise(() => {}) });
await new Promise(setImmediate);
diff --git a/tests/test_update_cves.py b/tests/test_update_cves.py
index 6daf8f1caea..93b52973796 100644
--- a/tests/test_update_cves.py
+++ b/tests/test_update_cves.py
@@ -459,6 +459,29 @@ ui.run(port=8080)
class PublishedLinkDeduplicationTests(unittest.TestCase):
+ def test_curated_artifact_variants_survive_publication(self) -> None:
+ cve = "CVE-2024-26642"
+ base = "https://github.com/google/security-research"
+ variants = [f"{base}/tree/master/pocs/linux/kernelctf/{cve}_{kind}" for kind in ("lts", "mitigation")]
+ template = "https://github.com/projectdiscovery/nuclei-templates/blob/main/http/Test.yaml?version=One#Request"
+ with tempfile.TemporaryDirectory() as directory:
+ root = Path(directory)
+ path = root / "2024" / f"{cve}.md"
+ path.parent.mkdir()
+ path.write_text(
+ "### Description\nA vulnerability.\n\n#### Github\n- " + base
+ + "\n\n#### Collections\n" + "\n".join(f"- {url}" for url in [*variants, variants[0].replace("google/security-research", "GOOGLE/Security-Research")])
+ + "\n\n#### Nuclei\n" + "\n".join(f"- {url}" for url in [template, template.replace("Test.yaml", "test.yaml")]) + "\n",
+ encoding="utf-8",
+ )
+ with patch.object(build_site, "CVES", root), patch.object(build_site, "load_metadata", return_value={}), \
+ patch.object(build_site, "load_dates", return_value={}), \
+ patch.object(build_site, "load_verified_references", return_value=set()):
+ entries, _ = build_site.build_cve_list(set())
+ self.assertEqual(entries[0]["collections"], variants)
+ self.assertEqual(entries[0]["nuclei"], [template, template.replace("Test.yaml", "test.yaml")])
+ self.assertEqual(entries[0]["poc"], [])
+
def test_published_links_require_absolute_http_urls(self) -> None:
valid = [
"https://Example.com/PoC.py?Version=One#Example",