mirror of
https://github.com/0xMarcio/cve.git
synced 2026-10-23 16:35:19 +00:00
148 lines
7.5 KiB
Python
148 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts'))
|
|
import releases
|
|
import sync_exploits as exploits
|
|
import sync_nuclei as nuclei
|
|
|
|
CVE = 'CVE-2020-1234'
|
|
URL = exploits.EDB_ENTRY + '100'
|
|
|
|
|
|
def csv_data(rows):
|
|
output = io.StringIO()
|
|
writer = csv.DictWriter(output, fieldnames=['id', 'codes', 'verified', 'date_published'])
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
return output.getvalue().encode()
|
|
|
|
|
|
class PublicationDateTests(unittest.TestCase):
|
|
def test_dates_are_calendar_dates_and_not_future_clocks(self):
|
|
for value in ['', 'unknown', '2020-02-30', '2020-1-1', '2020-01-01T00:00:00Z', '2099-01-01']:
|
|
self.assertIsNone(exploits.publication_date(value, today=date(2026, 10, 7)))
|
|
self.assertEqual(exploits.publication_date('2020-02-29', today=date(2026, 10, 7)), '2020-02-29')
|
|
|
|
def test_csv_keeps_dates_for_existing_associations_outside_selection_cap(self):
|
|
rows = [dict(id=n, codes=CVE, verified='1', date_published='2020-02-29') for n in range(100, 110)]
|
|
rows.append(dict(id=999, codes='', verified='0', date_published='2010-05-01'))
|
|
dates = {}
|
|
with patch.object(exploits, 'fetch', return_value=csv_data(rows)):
|
|
selected = exploits.exploitdb(dates)
|
|
self.assertEqual(len(selected[CVE]), exploits.PER_CVE)
|
|
self.assertNotIn(URL, selected[CVE])
|
|
self.assertEqual(dates[URL], '2020-02-29')
|
|
self.assertEqual(dates[exploits.EDB_ENTRY + '999'], '2010-05-01')
|
|
|
|
def test_legacy_urls_resolve_to_the_same_published_artifact(self):
|
|
dates = {URL: "2020-02-29"}
|
|
for url in [URL + "/", URL + "/?source=archive", "http://exploit-db.com/exploits/100"]:
|
|
self.assertEqual(exploits.artifact_publication_date(url, dates), "2020-02-29")
|
|
self.assertIsNone(exploits.artifact_publication_date("https://not-exploit-db.com/exploits/100", dates))
|
|
self.assertIsNone(exploits.artifact_publication_date("https://exploit-db.com/docs/100", dates))
|
|
|
|
def test_historical_dates_and_unknowns_preserve_observation_provenance(self):
|
|
with tempfile.TemporaryDirectory() as directory, patch.object(releases, 'DIRECTORY', Path(directory)):
|
|
old_url = exploits.EDB_ENTRY + '101'
|
|
future_url = exploits.EDB_ENTRY + '102'
|
|
ledger = releases.Ledger()
|
|
releases.record(ledger, CVE, old_url, observed_at='2026-10-01T00:00:00Z')
|
|
releases.save_ledger(ledger)
|
|
raw = csv_data([dict(id=100, codes=CVE, verified='1', date_published='2020-02-29'),
|
|
dict(id=101, codes=CVE, verified='1', date_published='2020-03-01'),
|
|
dict(id=102, codes=CVE, verified='1', date_published='2099-01-01')])
|
|
dates = {}
|
|
with patch.object(exploits, 'fetch', return_value=raw):
|
|
exploits.exploitdb(dates)
|
|
pairs = [(CVE, url) for url in [URL, old_url, future_url]]
|
|
with patch.object(releases, 'published_pairs', return_value=pairs):
|
|
summary = exploits.sync_release_dates(dates, import_mode=True)
|
|
current = releases.load_ledger()
|
|
historical = current[releases.key(CVE, URL)]
|
|
self.assertEqual((historical['basis'], historical['released']), ('source', '2020-02-29'))
|
|
self.assertTrue(historical['imported'])
|
|
self.assertNotIn('seen', historical)
|
|
existing = current[releases.key(CVE, old_url)]
|
|
self.assertEqual(existing['seen'], '2026-10-01T00:00:00Z')
|
|
self.assertNotIn('imported', existing)
|
|
future = current[releases.key(CVE, future_url)]
|
|
self.assertEqual((future['basis'], future['released']), ('unknown', None))
|
|
self.assertNotIn('seen', future)
|
|
self.assertEqual(summary['dated_edb_pairs'], 2)
|
|
self.assertEqual(releases.landed(current, now='2026-10-07T00:00:00Z'), [])
|
|
|
|
|
|
class SuccessfulSyncHookTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.cves = self.root / 'cves'
|
|
self.path = self.cves / '2020' / (CVE + '.md')
|
|
self.path.parent.mkdir(parents=True)
|
|
self.path.write_text('### Description\nFixture.\n')
|
|
|
|
def test_exploit_release_hook_runs_after_writes_not_on_failure_or_dry_run(self):
|
|
events = []
|
|
def recorded(*args, **kwargs):
|
|
self.assertIn(URL, self.path.read_text())
|
|
events.append('recorded')
|
|
return {}
|
|
with patch.object(exploits, 'CVES', self.cves), \
|
|
patch.object(exploits, 'exploitdb', return_value={CVE: [URL]}), \
|
|
patch.object(exploits, 'metasploit', return_value={}), \
|
|
patch.object(exploits, 'vulhub', return_value={}), \
|
|
patch.object(exploits, 'ensure_cve_entries', return_value=(set(), set())), \
|
|
patch.object(exploits, 'sync_release_dates', side_effect=recorded) as hook:
|
|
with patch.object(sys, 'argv', ['sync_exploits.py', '--dry-run']):
|
|
exploits.main()
|
|
hook.assert_not_called()
|
|
with patch.object(sys, 'argv', ['sync_exploits.py']):
|
|
exploits.main()
|
|
self.assertEqual(events, ['recorded'])
|
|
with patch.object(exploits, 'metasploit', side_effect=RuntimeError('source failed')), \
|
|
patch.object(sys, 'argv', ['sync_exploits.py']):
|
|
with self.assertRaisesRegex(RuntimeError, 'source failed'):
|
|
exploits.main()
|
|
self.assertEqual(events, ['recorded'])
|
|
|
|
def test_nuclei_records_unknown_release_only_after_successful_write(self):
|
|
url = 'https://github.com/projectdiscovery/nuclei-templates/blob/main/template.yaml'
|
|
def published():
|
|
self.assertIn(url, self.path.read_text())
|
|
self.assertTrue((self.root / 'nuclei.json').exists())
|
|
return [(CVE, url)]
|
|
with patch.object(nuclei, 'CVES', self.cves), patch.object(nuclei, 'OUTPUT', self.root / 'nuclei.json'), \
|
|
patch.object(nuclei, 'SOURCES', [('repo', 'tarball', 'blob', False)]), \
|
|
patch.object(nuclei, 'download', return_value=b''), \
|
|
patch.object(nuclei, 'collect', return_value={CVE: {'url': url}}), \
|
|
patch.object(nuclei, 'ensure_cve_entries', return_value=(set(), set())), \
|
|
patch.object(releases, 'DIRECTORY', self.root / 'releases'), \
|
|
patch.object(releases, 'published_pairs', side_effect=published) as pairs:
|
|
with patch.object(sys, 'argv', ['sync_nuclei.py', '--dry-run']):
|
|
self.assertEqual(nuclei.main(), 0)
|
|
pairs.assert_not_called()
|
|
with patch.object(sys, 'argv', ['sync_nuclei.py']):
|
|
self.assertEqual(nuclei.main(), 0)
|
|
row = releases.load_ledger()[releases.key(CVE, url)]
|
|
self.assertEqual((row['basis'], row['released']), ('unknown', None))
|
|
self.assertIn('seen', row)
|
|
with patch.object(nuclei, 'download', side_effect=RuntimeError('source failed')), \
|
|
patch.object(sys, 'argv', ['sync_nuclei.py']):
|
|
self.assertEqual(nuclei.main(), 1)
|
|
self.assertEqual(pairs.call_count, 1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|