cve/tests/test_sync_exploits.py

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()