Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 65 additions & 3 deletions vanir/detector_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
offline_directory_scanner /test/source
"""
import collections
import copy
import datetime
import functools
import inspect
Expand All @@ -23,7 +24,7 @@
import os
import sys
import textwrap
from typing import Any, Mapping, Sequence, Type, TypeVar
from typing import Any, Mapping, Sequence, Set, Type, TypeVar

from absl import app
from absl import flags
Expand All @@ -33,6 +34,7 @@
from vanir import detector_common_flags
from vanir import osv_client
from vanir import reporter
from vanir import vulnerability_manager

from vanir.scanners import scanner_base

Expand Down Expand Up @@ -160,6 +162,19 @@
</pre>
</td>
</tr>
<tr>
<th>Ignored CVEs</th>
<td>
{{ '<pre><b>Ignored CVEs by OSV ID filter</b></pre>' if ignored_cves['ignored_by_ID_filter']|length > 0 else '' }}
<pre>{% for cve in ignored_cves['ignored_by_ID_filter'] %}<nobr>{{ '%-15s' | format(cve) | replace(' ','&nbsp;')}}</nobr>&nbsp;<wbr>{% endfor %}
</pre>
{{ '<pre><b>Ignored CVEs by path filter</b></pre>' if ignored_cves['ignored_by_path_filter']|length > 0 else '' }}
{% for path in ignored_cves['ignored_by_path_filter'] %}
<pre><i>{{ path }}</i>&nbsp;<wbr>{% for cve in ignored_cves['ignored_by_path_filter'][path] %}<nobr>{{ '%-15s' | format(cve) | replace(' ','&nbsp;')}}</nobr>&nbsp;<wbr>{% endfor %}
</pre>
{% endfor %}
</td>
</tr>
</table>

<h3 onclick="toggle(this);" class="expand-toggle">
Expand Down Expand Up @@ -351,18 +366,28 @@ def _get_public_osv_url(osv_id: str) -> str:
return 'Not published.'
return osv_client.get_osv_url(osv_id)

def _get_cves_from_findings(findings: scanner_base.Findings, vuln_manager: vulnerability_manager)-> Set[str]:
"""Generate CVE ID set based on findings. """
signature_ids = [vuln.signature_id for vuln in findings]
cve_ids = set()
for signature_id in signature_ids:
cve_ids.update(vuln_manager.sign_id_to_cve_ids(signature_id))
return cve_ids


def _generate_json_report(
report_file_name: str,
report_book: reporter.ReportBook,
covered_cves: Sequence[str],
ignored_cves: Mapping[str, any],
) -> None:
"""Generates a JSON report based on the findings.

Args:
report_file_name: a JSON report file name to create.
report_book: a report book instance containing all reports.
covered_cves: a sequence of CVEs covered by this run.
ignored_cves: a mapping of CVEs ignored by this run.

Returns:
None
Expand Down Expand Up @@ -392,6 +417,7 @@ def _generate_json_report(
})
json_report['options'] = ' '.join(sys.argv[1:])
json_report['covered_cves'] = covered_cves
json_report['ignored_cves'] = ignored_cves
json_report['missing_patches'] = missing_patches
with open(report_file_name, 'w') as report_file:
json.dump(json_report, report_file, indent=4)
Expand All @@ -401,6 +427,7 @@ def _generate_html_report(
report_file_name: str,
report_book: reporter.ReportBook,
covered_cves: Sequence[str],
ignored_cves: Mapping[str, any],
stats: scanner_base.ScannedFileStats,
) -> None:
"""Generates a HTML file summarizing the report in a human-readable format.
Expand All @@ -409,6 +436,7 @@ def _generate_html_report(
report_file_name: a HTML report file name to create.
report_book: a report book instance containing all reports.
covered_cves: a sequence of CVEs covered by this run.
ignored_cves: a mapping of CVEs ignored by this run.
stats: |ScannedFileStats| object with scan result stats.

Returns:
Expand Down Expand Up @@ -455,6 +483,7 @@ def _generate_html_report(
html_report = template.render(
report_file_name=report_file_name,
covered_cves=covered_cves,
ignored_cves=ignored_cves,
unpatched_cves=report_book.unpatched_cves,
target_missing_patches=target_missing_patches,
non_target_missing_patches=non_target_missing_patches,
Expand Down Expand Up @@ -530,9 +559,28 @@ def main(argv: Sequence[str]) -> None:
[scanner_base.ShortFunctionFilter()]
+ list(detector_common_flags.generate_finding_filters_from_flags())
)

# Apply findings filters and collect filtered out CVE IDs
ignored_paths = collections.defaultdict(list)
findings = scanner_base.ShortFunctionFilter().filter(findings)
all_findings = copy.copy(findings)
all_findings_cve_ids = _get_cves_from_findings(all_findings, vuln_manager)

for finding_filter in finding_filters:
findings = finding_filter.filter(findings)
# Collect CVE IDs for vulnerabilities filtered out by PathPrefixFilter
if isinstance(finding_filter, scanner_base.PathPrefixFilter):
findings_filtered = finding_filter.filter(all_findings)
filtered_cve_ids = _get_cves_from_findings(findings_filtered, vuln_manager)
cve_ids_ignored = list(all_findings_cve_ids - filtered_cve_ids)
if cve_ids_ignored:
excluded_path = finding_filter._prefix
for cve_id in cve_ids_ignored:
ignored_paths[excluded_path].append(cve_id)

# sort CVE IDs in ignored paths dict
for path in ignored_paths:
ignored_paths[path] = sorted(ignored_paths[path])

report_book = reporter.ReportBook(
reporter.generate_reports(findings), vuln_manager
Expand All @@ -546,11 +594,25 @@ def main(argv: Sequence[str]) -> None:
)
covered_cves = sorted(set(covered_cves))

# Collect CVE IDs for vulnerabilities filtered out by OsvIdFilter
filtered_vuln_ids = [vuln.id for vuln in vuln_manager.get_vulnerabilities(ignore_filters=False)]
unfiltered_vuln_ids = [vuln.id for vuln in vuln_manager.get_vulnerabilities(ignore_filters=True)]
ignored_cve_ids = set()
for osv_id in detector_common_flags._OSV_ID_IGNORE_LIST.value:
if osv_id in unfiltered_vuln_ids and osv_id not in filtered_vuln_ids:
ignored_cve_ids.update(vuln_manager.osv_id_to_cve_ids(osv_id))

ignored_cves = {}
if ignored_cve_ids:
ignored_cves["ignored_by_ID_filter"] = sorted(ignored_cve_ids)
if ignored_paths:
ignored_cves["ignored_by_path_filter"] = ignored_paths

# Generate a machine-readable JSON report.
_generate_json_report(json_output_file_name, report_book, covered_cves)
_generate_json_report(json_output_file_name, report_book, covered_cves, ignored_cves)

# Generate a human-readable HTML report.
_generate_html_report(html_output_file_name, report_book, covered_cves, stats)
_generate_html_report(html_output_file_name, report_book, covered_cves, ignored_cves, stats)

# Generate a console output.
scanned_files = stats.analyzed_files + stats.skipped_files
Expand Down
7 changes: 7 additions & 0 deletions vanir/detector_runner_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -506,6 +506,13 @@ def test_main(self):
</pre>
</td>
</tr>
<tr>
<th>Ignored CVEs</th>
<td>
<pre>
</pre>
</td>
</tr>
</table>
<h3 onclick="toggle(this);" class="expand-toggle">
Missing Patches in Target Files (in 1 vuln)</h3>
Expand Down
5 changes: 4 additions & 1 deletion vanir/scanners/scanner_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -136,12 +136,15 @@ def __init__(self, prefix: str):
def filter(self, findings: Findings) -> Findings:
filtered_findings = {}
for sign, chunks in findings.items():
filtered_findings[sign] = list(
filtered_chunks = list(
filter(
lambda chunk: not chunk.target_file.startswith(self._prefix),
chunks,
)
)
# only add findings if chunks not empty
if filtered_chunks:
filtered_findings[sign] = filtered_chunks
return filtered_findings


Expand Down
5 changes: 3 additions & 2 deletions vanir/vulnerability_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -443,7 +443,7 @@ def add_vulnerability(
vuln: vulnerability.Vulnerability,
overwrite_older_duplicate: bool = False
):
"""Adds a new vulnearbility to the manager.
"""Adds a new vulnerability to the manager.

Args:
vuln: the Vulnerability object to be added.
Expand Down Expand Up @@ -826,7 +826,8 @@ def generate_from_managers(
vulnerabilities = []
vfilters = set()
for manager in managers:
vulnerabilities.extend(manager.vulnerabilities)
# Get all vulnerabilites for tracking purposes. The same filters are applied later anyway.
vulnerabilities.extend(manager.get_vulnerabilities(ignore_filters=True))
vfilters.update(manager.vulnerability_filters)
if vulnerability_filters:
vfilters.update(vulnerability_filters)
Expand Down