"""Verify the published research snapshot using retained public source files. Run: python3 reproduce_counts.py (Python 3 standard library only, no network). Keep data.json, research.json, monthly.csv and source-manifest.json beside this script, and all retained source files in raw/. NVD response totalResults values are API aggregates; saved responses contain one sample CVE. CVE Program counts remain a separate quarterly reference. """ import calendar import csv import hashlib import json import math from collections import Counter from pathlib import Path from urllib.parse import parse_qs, urlparse ROOT = Path(__file__).resolve().parent def read_json(name): return json.loads((ROOT / name).read_text(encoding='utf-8')) def require(condition, message): if not condition: raise ValueError(message) def equal(actual, expected, label): require(actual == expected, '{}: {} != {}'.format(label, actual, expected)) manifest = read_json('source-manifest.json') for item in manifest['files']: path = (ROOT / item['path']).resolve() require(ROOT in path.parents, 'Manifest path must stay inside snapshot folder') require(path.is_file(), 'Missing snapshot file: ' + item['path']) equal(hashlib.sha256(path.read_bytes()).hexdigest(), item['sha256'], item['path'] + ' SHA-256') data = read_json('data.json') research = read_json('research.json') expected = research['dataResearch'] cutoff = data['retrievedAt'] equal(cutoff, research['cutoff'], 'Research cutoff') with (ROOT / 'raw/kev_official_raw.csv').open(encoding='utf-8', newline='') as handle: rows = list(csv.DictReader(handle)) equal(len({row['cveID'] for row in rows}), len(rows), 'Unique KEV CVE IDs') equal(len(rows), data['officialProvenance']['kev']['rows'], 'KEV source rows') equal(max(row['dateAdded'] for row in rows), data['kevSnapshotLatestEntry'], 'KEV latest date') counts = Counter(row['dateAdded'][:7] for row in rows if '2024-01-01' <= row['dateAdded'] <= cutoff) quarters = [] official = read_json('raw/cve_official_metrics_raw.json') for year in reversed(official['publishedCveRecords']): if int(year['year']) < 2024: continue for item in year['data']: if item['value'] == 'TBA': continue q = int(item['quarter']) kev = sum(counts['{}-{:02d}'.format(year['year'], m)] for m in range((q - 1) * 3 + 1, q * 3 + 1)) quarters.append({'quarter': '{}-Q{}'.format(year['year'], q), 'cvePublished': int(item['value'].replace(',', '')), 'kevAdded': kev}) equal(quarters, expected['quarterly'], 'Research quarterly reference') equal(quarters, data['quarterlyOfficial'], 'Chart quarterly reference') for row in expected['monthly']: equal(counts[row['month']], row['kevAdded'], 'Research KEV ' + row['month']) equal(row['partial'], row['month'] > data['completeThrough'], 'Partial-month flag ' + row['month']) series = data['series'] year, month = map(int, data['completeThrough'].split('-')) month_count = (year - 2024) * 12 + month months = ['{}-{:02d}'.format(2024 + n // 12, n % 12 + 1) for n in range(month_count)] equal([row['month'] for row in series], months, 'Contiguous complete-month window') snapshots = read_json('raw/nvd/manifest.json')['records'] snapshot_by_key = {item['key']: item for item in snapshots} equal(list(snapshot_by_key), months + ['2025-10-01-21', '2026-10-01-21'], 'NVD snapshot windows') equal([item['rawFile'] for item in snapshots], data['officialProvenance']['nvd']['files'], 'NVD source-file inventory') equal(snapshots, manifest['snapshotSources'], 'Manifest NVD query provenance') equal(snapshots, research['dataRefresh']['snapshotSources'], 'Research NVD query provenance') for item in snapshots: raw = read_json(item['rawFile']) equal(hashlib.sha256((ROOT / item['rawFile']).read_bytes()).hexdigest(), item['sha256'], 'NVD query digest ' + item['key']) equal(raw['format'], 'NVD_CVE', 'NVD response format') equal(raw['totalResults'], item['totalResults'], 'NVD aggregate ' + item['key']) require(isinstance(raw['totalResults'], int) and raw['totalResults'] >= 0, 'Invalid NVD aggregate') parsed = urlparse(item['url']) equal(parsed.netloc, 'services.nvd.nist.gov', 'Official NVD host') params = parse_qs(parsed.query, keep_blank_values=True) require('noRejected' in params, 'NVD query must exclude rejected records') equal(params['resultsPerPage'], ['1'], 'NVD aggregate page size') equal(params['pubStartDate'][0] + 'Z', item['pubStartDate'], 'NVD query start') equal(params['pubEndDate'][0] + 'Z', item['pubEndDate'], 'NVD query end') equal(item['pubStartDate'], item['month'] + '-01T00:00:00.000Z', 'UTC range start') y, m = map(int, item['month'].split('-')) day = 21 if item['partial'] else calendar.monthrange(y, m)[1] equal(item['pubEndDate'], '{}-{:02d}T23:59:59.999Z'.format(item['month'], day), 'UTC range end') cumulative = 0 for row in series: y, m = row['month'].split('-') equal(row['partial'], False, 'Main-chart complete month ' + row['month']) equal(snapshot_by_key[row['month']]['totalResults'], row['cvePublished'], 'Monthly CVE ' + row['month']) equal(counts[row['month']], row['kevAdded'], 'Monthly KEV ' + row['month']) require(all(isinstance(row[key], int) and row[key] >= 0 for key in ('cvePublished', 'kevAdded')), 'Counts must be non-negative integers') cumulative += row['cvePublished'] equal(cumulative, row['cveCumulativeSince2024'], 'Cumulative CVE ' + row['month']) partial = data['partialMonth'] equal(partial['partial'], True, 'October partial flag') equal(partial['month'], '2026-10', 'Partial month') equal(partial['from'], '2026-10-01', 'Partial start') equal(partial['through'], '2026-10-21', 'Partial end') equal(partial['cvePublished'], snapshot_by_key['2026-10-01-21']['totalResults'], 'Partial NVD CVE') equal(partial['kevAdded'], sum(partial['from'] <= row['dateAdded'] <= partial['through'] for row in rows), 'Partial CISA KEV') equal(partial['cveCumulativeSince2024'], cumulative + partial['cvePublished'], 'Partial cumulative') equal(expected['monthly'], series + [partial], 'Research complete and partial rows') monthly_comparison = data['monthlyComparison'] equal(monthly_comparison['mode'], 'month-to-date', 'Comparison mode') equal(monthly_comparison['timezone'], 'UTC', 'Comparison timezone') equal(monthly_comparison, expected['monthlyComparison'], 'Research comparison') for label, y in [('previous', 2025), ('current', 2026)]: period = monthly_comparison[label] equal(period['from'], '{}-10-01'.format(y), 'Matched October start') equal(period['through'], '{}-10-21'.format(y), 'Matched October end') equal(period['cvePublished'], snapshot_by_key['{}-10-01-21'.format(y)]['totalResults'], 'Matched NVD count') equal(period['kevAdded'], sum(period['from'] <= row['dateAdded'] <= period['through'] for row in rows), 'Matched CISA count') for metric, key in [('cvePublished', 'cveYoYPercent'), ('kevAdded', 'kevYoYPercent')]: growth = (monthly_comparison['current'][metric] / monthly_comparison['previous'][metric] - 1) * 100 require(math.isclose(growth, monthly_comparison[key], rel_tol=1e-12), 'Monthly comparison mismatch: ' + key) with (ROOT / 'monthly.csv').open(encoding='utf-8', newline='') as handle: csv_rows = list(csv.DictReader(handle)) equal(len(csv_rows), len(series) + 1, 'CSV complete and partial row count') for csv_row, row in zip(csv_rows, series + [partial]): equal(csv_row['month'], row['month'], 'CSV month') for key in ('cvePublished', 'kevAdded', 'cveCumulativeSince2024'): equal(int(csv_row[key]), row[key], 'CSV ' + key + ' ' + row['month']) equal(csv_row['partial'].lower(), str(row['partial']).lower(), 'CSV partial flag') y, m = map(int, row['month'].split('-')) through = row.get('through', '{}-{:02d}'.format(row['month'], calendar.monthrange(y, m)[1])) equal(csv_row['through'], through, 'CSV observation boundary') def total(start, end, metric): return sum(row[metric] for row in series if start <= row['month'] <= end) comparison = { 'janSep2025CVE': total('2025-01', '2025-09', 'cvePublished'), 'janSep2026CVE': total('2026-01', '2026-09', 'cvePublished'), 'janSep2025KEV': total('2025-01', '2025-09', 'kevAdded'), 'janSep2026KEV': total('2026-01', '2026-09', 'kevAdded'), 'trailing12MonthsCVE': sum(row['cvePublished'] for row in series[-12:]), 'previous12MonthsCVE': sum(row['cvePublished'] for row in series[-24:-12]), } comparison['cveYoYPercent'] = (comparison['janSep2026CVE'] / comparison['janSep2025CVE'] - 1) * 100 comparison['kevYoYPercent'] = (comparison['janSep2026KEV'] / comparison['janSep2025KEV'] - 1) * 100 comparison['trailing12MonthsGrowthPercent'] = (comparison['trailing12MonthsCVE'] / comparison['previous12MonthsCVE'] - 1) * 100 for key, value in comparison.items(): require(math.isclose(value, data['comparisons'][key], rel_tol=1e-12), 'Historical comparison mismatch: ' + key) windows = [] for model in data['models']: release_month = model['date'][:7] if release_month not in months: continue index = months.index(release_month) before, after = series[max(0, index - 3):index], series[index + 1:index + 4] complete = len(before) == len(after) == 3 result = {'model': model['name'], 'completeWindow': complete} if complete: result['before'] = [row['month'] for row in before] result['after'] = [row['month'] for row in after] for metric in ('cvePublished', 'kevAdded'): before_mean = sum(row[metric] for row in before) / 3 after_mean = sum(row[metric] for row in after) / 3 result[metric] = {'beforeMean': before_mean, 'afterMean': after_mean, 'changePercent': (after_mean / before_mean - 1) * 100 if before_mean else None} windows.append(result) print(json.dumps({ 'verified': True, 'retrievedAt': cutoff, 'sha256FilesChecked': len(manifest['files']), 'completeMonths': len(series), 'completeThrough': data['completeThrough'], 'kevRawRows': len(rows), 'officialQuartersChecked': len(quarters), 'nvdQuerySnapshotsChecked': len(snapshots), 'partialMonth': partial, 'monthlyComparison': monthly_comparison, 'comparisons': comparison, 'modelWindows': windows, }, ensure_ascii=False, indent=2)) print('PASS: retained snapshots, hashes, full and partial monthly/CSV/cumulative values and equal-period comparisons agree.')