diff --git a/secator/config.py b/secator/config.py index bea53e099..ab09f3b25 100644 --- a/secator/config.py +++ b/secator/config.py @@ -201,6 +201,7 @@ class MongodbAddon(StrictModel): 'is_false_positive', 'is_acknowledged', 'verified', + 'status', 'tags', ] @@ -216,6 +217,7 @@ class SqliteAddon(StrictModel): 'is_false_positive', 'is_acknowledged', 'verified', + 'status', 'tags', ] diff --git a/secator/definitions.py b/secator/definitions.py index fd58fab18..a23d3daf9 100644 --- a/secator/definitions.py +++ b/secator/definitions.py @@ -142,6 +142,7 @@ SOURCES = 'sources' STORED_RESPONSE_PATH = 'stored_response_path' STATE = 'state' +STATUS = 'status' STATUS_CODE = 'status_code' STRING = 'str' TAGS = 'tags' diff --git a/secator/hooks/_dedup.py b/secator/hooks/_dedup.py index 0675eb8b9..0a4d1d5cb 100644 --- a/secator/hooks/_dedup.py +++ b/secator/hooks/_dedup.py @@ -1,6 +1,20 @@ # secator/hooks/_dedup.py +def _is_unset(field, value): + """Return True if `value` should be treated as "empty" for copy-forward purposes. + + For most fields, emptiness is the generic falsy check (`not value`). The `status` + field (Vulnerability) is special: its default value `'NEW'` is truthy but means + "untouched", so we treat `''` / `None` / `'NEW'` as unset. This lets a prior + `ACKNOWLEDGED` / `FIXED` status carry forward onto a re-found main whose status is + still the default `'NEW'`, while a never-touched vuln stays `'NEW'`. + """ + if field == 'status': + return not value or str(value).strip().upper() == 'NEW' + return not value + + def compute_duplicate_updates(workspace_findings, untagged_findings, copy_fields=None): """Compute duplicate-tagging updates for a set of findings (backend-agnostic). @@ -35,10 +49,12 @@ def compute_duplicate_updates(workspace_findings, untagged_findings, copy_fields if not hasattr(previous_item, field): continue value_prev = getattr(previous_item, field) - if not value_prev: + # Nothing meaningful to carry forward (handles `status='NEW'` as unset too). + if _is_unset(field, value_prev): continue value_curr = getattr(item, field, None) - if not value_curr and field not in copied_fields: + # Copy only onto an "empty" current value; for `status`, `'NEW'` counts as empty. + if _is_unset(field, value_curr) and field not in copied_fields: copied_fields[field] = value_prev related_ids = [] diff --git a/secator/output_types/vulnerability.py b/secator/output_types/vulnerability.py index 1a66c424f..922e6c3d0 100644 --- a/secator/output_types/vulnerability.py +++ b/secator/output_types/vulnerability.py @@ -2,7 +2,9 @@ from dataclasses import dataclass, field from typing import List -from secator.definitions import CONFIDENCE, CVSS_SCORE, EXTRA_DATA, ID, MATCHED_AT, NAME, REFERENCE, SEVERITY, TAGS +from secator.definitions import ( + CONFIDENCE, CVSS_SCORE, EXTRA_DATA, ID, MATCHED_AT, NAME, REFERENCE, SEVERITY, STATUS, TAGS +) from secator.output_types import OutputType from secator.utils import rich_to_ansi, rich_escape as _s, format_object, trim_string @@ -30,6 +32,7 @@ class Vulnerability(OutputType): verified: bool = field(default=False, compare=False) is_false_positive: bool = field(default=False, compare=False) is_acknowledged: bool = field(default=False, compare=False) + status: str = field(default='NEW', compare=False) tags: list = field(default_factory=list, compare=False) _source: str = field(default='', repr=True, compare=False) _type: str = field(default='vulnerability', repr=True) @@ -40,7 +43,8 @@ class Vulnerability(OutputType): _duplicate: bool = field(default=False, repr=True, compare=False) _related: list = field(default_factory=list, compare=False) - _table_fields = [MATCHED_AT, SEVERITY, CONFIDENCE, NAME, ID, CVSS_SCORE, TAGS, EXTRA_DATA, REFERENCE] + _table_fields = [MATCHED_AT, SEVERITY, CONFIDENCE, NAME, ID, CVSS_SCORE, STATUS, TAGS, EXTRA_DATA, REFERENCE] + STATUSES = ('NEW', 'ACKNOWLEDGED', 'FIXED') _sort_by = ('confidence_nb', 'severity_nb', 'matched_at', 'cvss_score') @staticmethod @@ -66,6 +70,11 @@ def __post_init__(self): self.severity_nb = severity_map.get(self.severity, 6) self.confidence_nb = severity_map[self.confidence] + # Normalize status: coerce empty / None / unknown values to 'NEW', uppercase. + # Allowed values are NEW / ACKNOWLEDGED / FIXED (see STATUSES). + status = (self.status or '').strip().upper() + self.status = status if status in self.STATUSES else 'NEW' + def __rich__(self): data = self.extra_data diff --git a/tests/unit/test_dedup.py b/tests/unit/test_dedup.py new file mode 100644 index 000000000..b3428cea0 --- /dev/null +++ b/tests/unit/test_dedup.py @@ -0,0 +1,58 @@ +import unittest + +from secator.hooks._dedup import compute_duplicate_updates +from secator.output_types import Vulnerability + + +def _vuln(uuid, status='NEW', verified=False, **kwargs): + return Vulnerability( + name='CVE-2025-53020', + id='CVE-2025-53020', + matched_at='host:80', + status=status, + verified=verified, + _uuid=uuid, + **kwargs, + ) + + +class TestComputeDuplicateUpdates(unittest.TestCase): + + def test_status_carried_forward_onto_new_main(self): + """A prior ACKNOWLEDGED main carries onto a re-found main whose status is NEW.""" + prev = _vuln('prev', status='ACKNOWLEDGED') + new = _vuln('new', status='NEW') + updates = compute_duplicate_updates([prev], [new], copy_fields=['status']) + assert updates['new']['status'] == 'ACKNOWLEDGED' + + def test_status_fixed_not_overwritten(self): + """A new main that is already FIXED keeps its value (FIXED is not 'unset').""" + prev = _vuln('prev', status='ACKNOWLEDGED') + new = _vuln('new', status='FIXED') + updates = compute_duplicate_updates([prev], [new], copy_fields=['status']) + assert 'status' not in updates['new'] + + def test_prior_new_status_not_carried(self): + """A prior status of NEW is treated as unset and is not carried forward.""" + prev = _vuln('prev', status='NEW') + new = _vuln('new', status='NEW') + updates = compute_duplicate_updates([prev], [new], copy_fields=['status']) + assert 'status' not in updates['new'] + + def test_non_status_field_keeps_not_value_semantics(self): + """Generic fields still use the `not value` emptiness check.""" + # Prior verified=True copies onto new verified=False (falsy -> empty). + prev = _vuln('prev', verified=True) + new = _vuln('new', verified=False) + updates = compute_duplicate_updates([prev], [new], copy_fields=['verified']) + assert updates['new']['verified'] is True + + # Prior verified=False is empty -> nothing to copy. + prev2 = _vuln('prev2', verified=False) + new2 = _vuln('new2', verified=True) + updates2 = compute_duplicate_updates([prev2], [new2], copy_fields=['verified']) + assert 'verified' not in updates2['new2'] + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/unit/test_output_types.py b/tests/unit/test_output_types.py index 614f77b42..77569ed08 100644 --- a/tests/unit/test_output_types.py +++ b/tests/unit/test_output_types.py @@ -42,6 +42,32 @@ def test_merge_with_exclude_fields(self): assert vuln1.name == 'CVE-2025-53020' +class TestVulnerabilityStatus(unittest.TestCase): + + def test_status_defaults_to_new(self): + vuln = Vulnerability(name='CVE-2025-53020') + assert vuln.status == 'NEW' + + def test_status_empty_coerces_to_new(self): + assert Vulnerability(name='CVE-2025-53020', status='').status == 'NEW' + assert Vulnerability(name='CVE-2025-53020', status=None).status == 'NEW' + + def test_status_unknown_coerces_to_new(self): + assert Vulnerability(name='CVE-2025-53020', status='bogus').status == 'NEW' + + def test_status_valid_values_preserved_and_uppercased(self): + assert Vulnerability(name='CVE-2025-53020', status='ACKNOWLEDGED').status == 'ACKNOWLEDGED' + assert Vulnerability(name='CVE-2025-53020', status='fixed').status == 'FIXED' + assert Vulnerability(name='CVE-2025-53020', status=' new ').status == 'NEW' + + def test_status_does_not_affect_equality(self): + # Same identity (name/id/matched_at) but different status must still be equal (dedup-safe). + vuln1 = Vulnerability(name='CVE-2025-53020', id='CVE-2025-53020', matched_at='host:80', status='NEW') + vuln2 = Vulnerability(name='CVE-2025-53020', id='CVE-2025-53020', matched_at='host:80', status='FIXED') + assert vuln1 == vuln2 + assert vuln1._compare_key() == vuln2._compare_key() + + class TestErrorRich(unittest.TestCase): def test_error_rich_with_node_id(self):