vulnerablecode/vulnerabilities/tests/test_apache_kafka.py at 353c2ac32421f605130f4756113aaf201d9cdc2b · aboutcode-org/vulnerablecode · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
#
# Copyright (c) nexB Inc. and others. All rights reserved.
# VulnerableCode is a trademark of nexB Inc.
# SPDX-License-Identifier: Apache-2.0
# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
# See https://github.com/aboutcode-org/vulnerablecode for support or download.
# See https://aboutcode.org for more information about nexB OSS projects.
#

import json
import os
from unittest.mock import patch

import pytest
from packageurl import PackageURL

from vulnerabilities.importer import AdvisoryData
from vulnerabilities.importers.apache_kafka import ApacheKafkaImporter
from vulnerabilities.improvers.default import DefaultImprover
from vulnerabilities.improvers.valid_versions import ApacheKafkaImprover
from vulnerabilities.tests import util_tests

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

TEST_DATA = os.path.join(
    BASE_DIR,
    "test_data/apache_kafka",
)


def test_to_advisory():
    with open(os.path.join(TEST_DATA, "cve-list-2022-12-06.html")) as f:
        raw_data = f.read()
    advisories = ApacheKafkaImporter().to_advisory(raw_data)
    result = [data.to_dict() for data in advisories]

    expected_file = os.path.join(TEST_DATA, f"to-advisory-apache_kafka-expected.json")
    util_tests.check_results_against_json(result, expected_file)


@patch("vulnerabilities.improvers.valid_versions.ApacheKafkaImprover.get_package_versions")
def test_apache_tomcat_improver(mock_response):
    advisory_file = os.path.join(TEST_DATA, f"to-advisory-apache_kafka-expected.json")
    with open(advisory_file) as exp:
        advisories = [AdvisoryData.from_dict(adv) for adv in (json.load(exp))]
    mock_response.return_value = [
        "1.1.0",
        "1.1.1",
        "1.1.2",
        "1.1.3",
        "1.1.4",
        "1.1.5",
        "1.1.6",
        "1.1.7",
        "1.1.8",
    ]
    improvers = [ApacheKafkaImprover(), DefaultImprover()]
    result = []
    for improver in improvers:
        for advisory in advisories:
            inference = [data.to_dict() for data in improver.get_inferences(advisory)]
            result.extend(inference)
    expected_file = os.path.join(TEST_DATA, f"apache-kafka-improver-expected.json")
    util_tests.check_results_against_json(result, expected_file)


def to_advisory_changed_cve():
    with open(os.path.join(TEST_DATA, "cve-list-changed-cve.html")) as f:
        raw_data = f.read()
    advisories = ApacheKafkaImporter().to_advisory(raw_data)


def to_advisory_changed_versions_affected():
    with open(os.path.join(TEST_DATA, "cve-list-changed-versions-affected.html")) as f:
        raw_data = f.read()
    advisories = ApacheKafkaImporter().to_advisory(raw_data)


def to_advisory_changed_fixed_versions():
    with open(os.path.join(TEST_DATA, "cve-list-changed-fixed-versions.html")) as f:
        raw_data = f.read()
    advisories = ApacheKafkaImporter().to_advisory(raw_data)


@pytest.fixture
def mock_apache_kafka_advisory_page(monkeypatch):
    test_file = os.path.join(TEST_DATA, "cve-list-2022-12-06.html")
    with open(test_file, "rb") as f:
        html_content = f.read()

    def mock_fetch_advisory_page(self):
        return html_content

    monkeypatch.setattr(ApacheKafkaImporter, "fetch_advisory_page", mock_fetch_advisory_page)


def test_package_first_mode_all_advisories(monkeypatch, mock_apache_kafka_advisory_page):
    purl = PackageURL(type="apache", name="kafka")
    importer = ApacheKafkaImporter(purl=purl)
    advisories = list(importer.advisory_data())

    assert len(advisories) == 6


def test_package_first_mode_version_affected(monkeypatch, mock_apache_kafka_advisory_page):
    purl = PackageURL(type="apache", name="kafka", version="2.8.0")
    importer = ApacheKafkaImporter(purl=purl)
    advisories = list(importer.advisory_data())

    assert any(
        any(
            ap.affected_version_range and "2.8.0" in ap.affected_version_range
            for ap in adv.affected_packages
        )
        for adv in advisories
    )


def test_package_first_mode_version_not_affected(monkeypatch, mock_apache_kafka_advisory_page):
    purl = PackageURL(type="apache", name="kafka", version="3.3.0")
    importer = ApacheKafkaImporter(purl=purl)
    advisories = list(importer.advisory_data())
    assert advisories == []


def test_package_first_mode_wrong_package(monkeypatch, mock_apache_kafka_advisory_page):
    purl = PackageURL(type="apache", name="notkafka")
    importer = ApacheKafkaImporter(purl=purl)
    advisories = list(importer.advisory_data())
    assert advisories == []