tests.testScanAndResult.FullTest.testStopScan() - Code Metrics - Inspection of "Add new members for vt elements in vts dictionary" - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

Passed
Pull Request — master (#73)

by Juan José
created 2018-11-21 10:05 UTC
tests.testScanAndResult.FullTest.testStopScan() A

↳ Parent: tests.testScanAndResult
Complexity

Conditions
Size

Total Lines	17
Code Lines	12
Duplication

Lines	0
Ratio	0 %
Importance

Changes
Metric	Value
cc	1
eloc	12
nop	1
dl	0
loc	17
rs	9.8
c	0
b	0
f	0
# Copyright (C) 2015-2018 Greenbone Networks GmbH

#
# SPDX-License-Identifier: GPL-2.0-or-later
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

""" Test module for scan runs
"""

from __future__ import print_function

import time
import unittest
import xml.etree.ElementTree as ET

import defusedxml.lxml as secET
from defusedxml.common import EntitiesForbidden

from ospd.ospd import OSPDaemon, OSPDError

class Result(object):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
    def __init__(self, type_, **kwargs):
        self.result_type = type_
        self.host = ''
        self.name = ''
        self.value = ''
        self.port = ''
        self.test_id = ''
        self.severity = ''
        self.qod = ''
        for name, value in kwargs.items():
            setattr(self, name, value)


class DummyWrapper(OSPDaemon):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
    def __init__(self, results, checkresult=True):
        OSPDaemon.__init__(self, 'cert', 'key', 'ca')
        self.checkresult = checkresult
        self.results = results

    def check(self):
        return self.checkresults


    @staticmethod
    def get_custom_vt_as_xml_str(vt_id, custom):
        return '<mytest>static test</mytest>'

    @staticmethod
    def get_params_vt_as_xml_str(vt_id, vt_params):
        return ('<vt_param id="abc" type="string">'
                '<name>ABC</name><description>Test ABC</description><default>yes</default>'
                '</vt_param>'
                '<vt_param id="def" type="string">'
                '<name>DEF</name><description>Test DEF</description><default>no</default>'
                '</vt_param>')

    @staticmethod
    def get_refs_vt_as_xml_str(vt_id, vt_refs):
        response = ('<ref type="cve" id="CVE-2010-4480"/>' +
                    '<ref type="url" id="http://example.com"/>')
        return response

    @staticmethod
    def get_dependencies_vt_as_xml_str(vt_id, vt_dependencies):
        response = ('<dependency vt_id="1.3.6.1.4.1.25623.1.0.50282" />' +
                    '<dependency vt_id="1.3.6.1.4.1.25623.1.0.50283" />')

        return response

    @staticmethod
    def get_severities_vt_as_xml_str(vt_id, severities):
        response = ('<severity cvss_base="5.0" cvss_type="cvss_base_v2">' +
                    'AV:N/AC:L/Au:N/C:N/I:N/A:P</severity>')

        return response

    @staticmethod
    def get_detection_vt_as_xml_str(vt_id, detection=None,
                                    qod_type=None, qod=None):
        response = ('<detection qod_type="package">some detection</detection>')

        return response

    @staticmethod
    def get_summary_vt_as_xml_str(vt_id, summary):
        response = ('<summary>Some summary</summary>')

        return response

    @staticmethod
    def get_affected_vt_as_xml_str(vt_id, affected):
        response = ('<affected>Some affected</affected>')

        return response

    @staticmethod
    def get_impact_vt_as_xml_str(vt_id, impact):
        response = ('<impact>Some impact</impact>')

        return response

    @staticmethod
    def get_insight_vt_as_xml_str(vt_id, insight):
        response = ('<insight>Some insight</insight>')

        return response

    @staticmethod
    def get_solution_vt_as_xml_str(vt_id, solution, solution_type=None):
        response = ('<solution>Some solution</solution>')

        return response

    def exec_scan(self, scan_id, target):
        time.sleep(0.01)
        for res in self.results:
            if res.result_type == 'log':
                self.add_scan_log(scan_id, res.host or target, res.name, res.value, res.port)
            if res.result_type == 'error':
                self.add_scan_error(scan_id, res.host or target, res.name, res.value, res.port)
            elif res.result_type == 'host-detail':
                self.add_scan_host_detail(scan_id, res.host or target, res.name, res.value)
            elif res.result_type == 'alarm':
                self.add_scan_alarm(scan_id, res.host or target, res.name, res.value, res.port, res.test_id, res.severity, res.qod)

            else:
                raise ValueError(res.result_type)


class FullTest(unittest.TestCase):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
    # TODO: There should be a lot more assert in there !


    def testGetDefaultScannerParams(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        response = secET.fromstring(daemon.handle_command('<get_scanner_details />'))
        # The status of the response must be success (i.e. 200)
        self.assertEqual(response.get('status'), '200')
        # The response root element must have the correct name
        self.assertEqual(response.tag, 'get_scanner_details_response')
        # The response must contain a 'scanner_params' element
        self.assertIsNotNone(response.find('scanner_params'))

    def testGetDefaultHelp(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        response = secET.fromstring(daemon.handle_command('<help />'))
        self.assertEqual(response.get('status'), '200')
        response = secET.fromstring(daemon.handle_command('<help format="xml" />'))
        self.assertEqual(response.get('status'), '200')
        self.assertEqual(response.tag, 'help_response')

    def testGetDefaultScannerVersion(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        response = secET.fromstring(daemon.handle_command('<get_version />'))
        self.assertEqual(response.get('status'), '200')
        self.assertIsNotNone(response.find('protocol'))

    def testGetVTs_no_VT(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        response = secET.fromstring(daemon.handle_command('<get_vts />'))
        self.assertEqual(response.get('status'), '200')
        self.assertIsNotNone(response.find('vts'))

    def testGetVTs_single_VT(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4', 'A vulnerability test')
        response = secET.fromstring(daemon.handle_command('<get_vts />'))
        self.assertEqual(response.get('status'), '200')
        vts = response.find('vts')
        self.assertIsNotNone(vts.find('vt'))
        vt = vts.find('vt')

        self.assertEqual(vt.get('id'), '1.2.3.4')

    def testGetVTs_multiple_VTs(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4', 'A vulnerability test')
        daemon.add_vt('some id', 'Another vulnerability test')
        daemon.add_vt('123456789', 'Yet another vulnerability test')
        response = secET.fromstring(daemon.handle_command('<get_vts />'))
        self.assertEqual(response.get('status'), '200')
        vts = response.find('vts')
        self.assertIsNotNone(vts.find('vt'))

    def testGetVTs_multiple_VTs_with_custom(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4', 'A vulnerability test')
        daemon.add_vt('some id', 'Another vulnerability test with custom info', {'depencency': '1.2.3.4'})

        daemon.add_vt('123456789', 'Yet another vulnerability test')
        response = secET.fromstring(daemon.handle_command('<get_vts />'))


    def testGetVTs_VTs_with_params(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4', 'A vulnerability test', vt_params="a", custom="b")
        response = secET.fromstring(daemon.handle_command('<get_vts vt_id="1.2.3.4"></get_vts>'))
        # The status of the response must be success (i.e. 200)
        self.assertEqual(response.get('status'), '200')
        # The response root element must have the correct name
        self.assertEqual(response.tag, 'get_vts_response')
        # The response must contain a 'scanner_params' element
        self.assertIsNotNone(response.find('vts'))
        vt_params = response[0][0].findall('vt_params')
        self.assertEqual(1, len(vt_params))
        custom = response[0][0].findall('custom')
        self.assertEqual(1, len(custom))
        params = response.findall('vts/vt/vt_params/vt_param')
        self.assertEqual(2, len(params))

    def testGetVTs_VTs_with_refs(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4',
                      'A vulnerability test',
                      vt_params="a",
                      custom="b",
                      vt_refs="c")
        response = secET.fromstring(
            daemon.handle_command('<get_vts vt_id="1.2.3.4"></get_vts>'))
        # The status of the response must be success (i.e. 200)
        self.assertEqual(response.get('status'), '200')
        # The response root element must have the correct name
        self.assertEqual(response.tag, 'get_vts_response')
        # The response must contain a 'vts' element
        self.assertIsNotNone(response.find('vts'))
        vt_params = response[0][0].findall('vt_params')
        self.assertEqual(1, len(vt_params))
        custom = response[0][0].findall('custom')
        self.assertEqual(1, len(custom))
        refs = response.findall('vts/vt/vt_refs/ref')
        self.assertEqual(2, len(refs))

    def testGetVTs_VTs_with_dependencies(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4',
                      'A vulnerability test',
                      vt_params="a",
                      custom="b",
                      vt_dependencies="c",)
        response = secET.fromstring(
            daemon.handle_command('<get_vts vt_id="1.2.3.4"></get_vts>'))
        deps = response.findall('vts/vt/dependencies/dependency')
        self.assertEqual(2, len(deps))

    def testGetVTs_VTs_with_severities(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4',
                      'A vulnerability test',
                      vt_params="a",
                      custom="b",
                      severities="c",)
        response = secET.fromstring(
            daemon.handle_command('<get_vts vt_id="1.2.3.4"></get_vts>'))
        severity = response.findall('vts/vt/severities/severity')
        self.assertEqual(1, len(severity))

    def testGetVTs_VTs_with_detection_qodt(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4',
                      'A vulnerability test',
                      vt_params="a",
                      custom="b",
                      detection="c",
                      qod_t="d")
        response = secET.fromstring(
            daemon.handle_command('<get_vts vt_id="1.2.3.4"></get_vts>'))
        detection = response.findall('vts/vt/detection')
        self.assertEqual(1, len(detection))

    def testGetVTs_VTs_with_detection_qodv(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4',
                      'A vulnerability test',
                      vt_params="a",
                      custom="b",
                      detection="c",
                      qod_v="d")
        response = secET.fromstring(
            daemon.handle_command('<get_vts vt_id="1.2.3.4"></get_vts>'))
        detection = response.findall('vts/vt/detection')
        self.assertEqual(1, len(detection))

    def testGetVTs_VTs_with_summary(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4',
                      'A vulnerability test',
                      vt_params="a",
                      custom="b",
                      summary="c",)
        response = secET.fromstring(
            daemon.handle_command('<get_vts vt_id="1.2.3.4"></get_vts>'))
        summary = response.findall('vts/vt/summary')
        self.assertEqual(1, len(summary))

    def testGetVTs_VTs_with_impact(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4',
                      'A vulnerability test',
                      vt_params="a",
                      custom="b",
                      impact="c",)
        response = secET.fromstring(
            daemon.handle_command('<get_vts vt_id="1.2.3.4"></get_vts>'))
        impact = response.findall('vts/vt/impact')
        self.assertEqual(1, len(impact))

    def testGetVTs_VTs_with_affected(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4',
                      'A vulnerability test',
                      vt_params="a",
                      custom="b",
                      affected="c",)
        response = secET.fromstring(
            daemon.handle_command('<get_vts vt_id="1.2.3.4"></get_vts>'))
        affect = response.findall('vts/vt/affected')
        self.assertEqual(1, len(affect))

    def testGetVTs_VTs_with_insight(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4',
                      'A vulnerability test',
                      vt_params="a",
                      custom="b",
                      insight="c",)
        response = secET.fromstring(
            daemon.handle_command('<get_vts vt_id="1.2.3.4"></get_vts>'))
        insight = response.findall('vts/vt/insight')
        self.assertEqual(1, len(insight))

    def testGetVTs_VTs_with_solution(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        daemon.add_vt('1.2.3.4',
                      'A vulnerability test',
                      vt_params="a",
                      custom="b",
                      solution="c",
                      solution_t="d")
        response = secET.fromstring(
            daemon.handle_command('<get_vts vt_id="1.2.3.4"></get_vts>'))
        solution = response.findall('vts/vt/solution')
        self.assertEqual(1, len(solution))

    def testiScanWithError(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([
            Result('error', value='something went wrong'),
        ])

        response = secET.fromstring(daemon.handle_command('<start_scan target="localhost" ports="80, 443"><scanner_params /></start_scan>'))

        scan_id = response.findtext('id')
        finished = False
        while not finished:
            response = secET.fromstring(daemon.handle_command('<get_scans scan_id="%s" details="0"/>' % scan_id))

            scans = response.findall('scan')
            self.assertEqual(1, len(scans))
            scan = scans[0]
            if int(scan.get('progress')) != 100:
                self.assertEqual('0', scan.get('end_time'))
                time.sleep(.010)
            else:
                finished = True
        response = secET.fromstring(daemon.handle_command('<get_scans scan_id="%s"/>' % scan_id))
        response = secET.fromstring(daemon.handle_command('<get_scans />'))
        response = secET.fromstring(daemon.handle_command('<get_scans scan_id="%s" details="1"/>' % scan_id))

        self.assertEqual(response.findtext('scan/results/result'), 'something went wrong')

        response = secET.fromstring(daemon.handle_command('<delete_scan scan_id="%s" />' % scan_id))
        self.assertEqual(response.get('status'), '200')


    def testGetScanPop(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([
            Result('host-detail', value='Some Host Detail'),
        ])

        response = secET.fromstring(daemon.handle_command('<start_scan target="localhost" ports="80, 443"><scanner_params /></start_scan>'))

        scan_id = response.findtext('id')
        time.sleep(1)

        response = secET.fromstring(
            daemon.handle_command('<get_scans scan_id="%s"/>' % scan_id))
        self.assertEqual(response.findtext('scan/results/result'),
                         'Some Host Detail')

        response = secET.fromstring(
            daemon.handle_command(
                '<get_scans details="0" pop_results="1"/>'))
        self.assertEqual(response.findtext('scan/results/result'),
                         None)

        response = secET.fromstring(
            daemon.handle_command(
                '<get_scans scan_id="%s" pop_results="1"/>' % scan_id))
        self.assertEqual(response.findtext('scan/results/result'),
                         'Some Host Detail')

        response = secET.fromstring(
            daemon.handle_command(
                '<get_scans scan_id="%s" pop_results="1"/>' % scan_id))
        self.assertNotEqual(response.findtext('scan/results/result'),
                         'Some Host Detail')

        self.assertEqual(response.findtext('scan/results/result'),
                         None)

        while True:
            response = secET.fromstring(
                daemon.handle_command(
                    '<get_scans scan_id="%s" details="0"/>' % scan_id))
            scans = response.findall('scan')
            self.assertEqual(1, len(scans))
            scan = scans[0]
            if int(scan.get('progress')) == 100:
                break

        response = secET.fromstring(
            daemon.handle_command('<delete_scan scan_id="%s" />' % scan_id))
        self.assertEqual(response.get('status'), '200')


    def testStopScan(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        response = secET.fromstring(
            daemon.handle_command('<start_scan ' +
                                  'target="localhost" ports="80, 443">' +
                                  '<scanner_params /></start_scan>'))
        scan_id = response.findtext('id')

        # Depending on the sistem this test can end with a race condition
        # because the scanner is already stopped when the <stop_scan> command
        # is run.
        time.sleep(3)
        cmd = secET.fromstring('<stop_scan scan_id="%s" />' % scan_id)
        self.assertRaises(OSPDError, daemon.handle_stop_scan_command, cmd)

        cmd = secET.fromstring('<stop_scan />')
        self.assertRaises(OSPDError, daemon.handle_stop_scan_command, cmd)


    def testScanWithVTs(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        cmd = secET.fromstring('<start_scan ' +
                               'target="localhost" ports="80, 443">' +
                               '<scanner_params /><vt_selection /></start_scan>')
        self.assertRaises(OSPDError, daemon.handle_start_scan_command, cmd)

        # With one VT, without params
        response = secET.fromstring(
            daemon.handle_command('<start_scan ' +
                                  'target="localhost" ports="80, 443">' +
                                  '<scanner_params /><vt_selection><vt_single id="1.2.3.4" />' +
                                  '</vt_selection></start_scan>'))
        scan_id = response.findtext('id')
        time.sleep(0.01)
        self.assertEqual(daemon.get_scan_vts(scan_id), {'1.2.3.4': {}, 'vt_groups': []})
        self.assertNotEqual(daemon.get_scan_vts(scan_id), {'1.2.3.6': {}})

        # With out VTS
        response = secET.fromstring(
            daemon.handle_command('<start_scan ' +
                                  'target="localhost" ports="80, 443">' +
                                  '<scanner_params /></start_scan>'))
        scan_id = response.findtext('id')
        time.sleep(0.01)
        self.assertEqual(daemon.get_scan_vts(scan_id), {})

    def testScanWithVTs_and_param(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])

        # Raise because no vt_param id attribute
        cmd = secET.fromstring('<start_scan ' +
                               'target="localhost" ports="80, 443">' +
                               '<scanner_params /><vt_selection><vt_single id="1234">' +
                               '<vt_value>200</vt_value>' +
                               '</vt_single></vt_selection></start_scan>')
        self.assertRaises(OSPDError, daemon.handle_start_scan_command, cmd)

        # No error
        response = secET.fromstring(
            daemon.handle_command('<start_scan ' +
                                  'target="localhost" ports="80, 443">' +
                                  '<scanner_params /><vt_selection><vt_single id="1234">' +
                                  '<vt_value id="ABC">200' +
                                  '</vt_value></vt_single></vt_selection></start_scan>'))
        scan_id = response.findtext('id')
        time.sleep(0.01)
        self.assertEqual(daemon.get_scan_vts(scan_id),
                         {'1234': {'ABC': '200'}, 'vt_groups': []})


        # Raise because no vtgroup filter attribute
        cmd = secET.fromstring('<start_scan ' +
                               'target="localhost" ports="80, 443">' +
                               '<scanner_params /><vt_selection><vt_group/>' +
                               '</vt_selection></start_scan>')
        self.assertRaises(OSPDError, daemon.handle_start_scan_command, cmd)

        # No error
        response = secET.fromstring(
            daemon.handle_command('<start_scan ' +
                                  'target="localhost" ports="80, 443">' +
                                  '<scanner_params /><vt_selection>' +
                                  '<vt_group filter="a"/>' +
                                  '</vt_selection></start_scan>'))
        scan_id = response.findtext('id')
        time.sleep(0.01)
        self.assertEqual(daemon.get_scan_vts(scan_id),
                         {'vt_groups': ['a']})


    def testBillonLaughs(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        lol = ('<?xml version="1.0"?>' +
               '<!DOCTYPE lolz [' +
               ' <!ENTITY lol "lol">' +
               ' <!ELEMENT lolz (#PCDATA)>' +
               ' <!ENTITY lol1 "&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;&lol;">' +
               ' <!ENTITY lol2 "&lol1;&lol1;&lol1;&lol1;&lol1;&lol1;&lol1;&lol1;&lol1;&lol1;">' +
               ' <!ENTITY lol3 "&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;&lol2;">' +
               ' <!ENTITY lol4 "&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;&lol3;">' +
               ' <!ENTITY lol5 "&lol4;&lol4;&lol4;&lol4;&lol4;&lol4;&lol4;&lol4;&lol4;&lol4;">' +
               ' <!ENTITY lol6 "&lol5;&lol5;&lol5;&lol5;&lol5;&lol5;&lol5;&lol5;&lol5;&lol5;">' +
               ' <!ENTITY lol7 "&lol6;&lol6;&lol6;&lol6;&lol6;&lol6;&lol6;&lol6;&lol6;&lol6;">' +
               ' <!ENTITY lol8 "&lol7;&lol7;&lol7;&lol7;&lol7;&lol7;&lol7;&lol7;&lol7;&lol7;">' +
               ' <!ENTITY lol9 "&lol8;&lol8;&lol8;&lol8;&lol8;&lol8;&lol8;&lol8;&lol8;&lol8;">' +
               ']>')
        self.assertRaises(EntitiesForbidden, daemon.handle_command, lol)

    def testScanMultiTarget(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        response = secET.fromstring(
            daemon.handle_command('<start_scan>' +
                                  '<scanner_params /><vts><vt id="1.2.3.4" />' +
                                  '</vts>' +
                                  '<targets><target>' +
                                  '<hosts>localhosts</hosts>' +
                                  '<ports>80,443</ports>' +
                                  '</target>' +
                                  '<target><hosts>192.168.0.0/24</hosts>' +
                                  '<ports>22</ports></target></targets>' +
                                  '</start_scan>'))
        self.assertEqual(response.get('status'), '200')


    def testMultiTargetWithCredentials(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        response = secET.fromstring(
            daemon.handle_command('<start_scan>' +
                                  '<scanner_params /><vts><vt id="1.2.3.4" />' +
                                  '</vts>' +
                                  '<targets><target><hosts>localhosts</hosts>' +
                                  '<ports>80,443</ports></target><target>' +
                                  '<hosts>192.168.0.0/24</hosts><ports>22' +
                                  '</ports><credentials>' +
                                  '<credential type="up" service="ssh" port="22">' +
                                  '<username>scanuser</username>' +
                                  '<password>mypass</password>' +
                                  '</credential><credential type="up" service="smb">' +
                                  '<username>smbuser</username>' +
                                  '<password>mypass</password></credential>' +
                                  '</credentials>' +
                                  '</target></targets>' +
                                  '</start_scan>'))
        self.assertEqual(response.get('status'), '200')
        cred_dict = {'ssh': {'type': 'up', 'password':
                    'mypass', 'port': '22', 'username':

                    'scanuser'}, 'smb': {'type': 'up',

                    'password': 'mypass', 'username': 'smbuser'}}

        scan_id = response.findtext('id')
        response = daemon.get_scan_credentials(scan_id, "192.168.0.0/24")
        self.assertEqual(response, cred_dict)

    def testScanGetTarget(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        response = secET.fromstring(
            daemon.handle_command('<start_scan>' +
                                  '<scanner_params /><vts><vt id="1.2.3.4" />' +
                                  '</vts>' +
                                  '<targets><target>' +
                                  '<hosts>localhosts</hosts>' +
                                  '<ports>80,443</ports>' +
                                  '</target>' +
                                  '<target><hosts>192.168.0.0/24</hosts>' +
                                  '<ports>22</ports></target></targets>' +
                                  '</start_scan>'))
        scan_id = response.findtext('id')
        response = secET.fromstring(
            daemon.handle_command('<get_scans scan_id="%s"/>' % scan_id))
        scan_res = response.find('scan')
        self.assertEqual(scan_res.get('target'), 'localhosts,192.168.0.0/24')

    def testScanGetLegacyTarget(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])

        response = secET.fromstring(
            daemon.handle_command('<start_scan target="localhosts,192.168.0.0/24" ports="22">' +
                                  '<scanner_params /><vts><vt id="1.2.3.4" />' +
                                  '</vts>' +
                                  '</start_scan>'))
        scan_id = response.findtext('id')
        response = secET.fromstring(
            daemon.handle_command('<get_scans scan_id="%s"/>' % scan_id))
        scan_res = response.find('scan')
        self.assertEqual(scan_res.get('target'), 'localhosts,192.168.0.0/24')

    def testScanMultiTargetParallelWithError(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        cmd = secET.fromstring('<start_scan parallel="100a">' +
                               '<scanner_params />' +
                               '<targets><target>' +
                               '<hosts>localhosts</hosts>' +
                               '<ports>22</ports>' +
                               '</target></targets>' +
                               '</start_scan>')
        time.sleep(1)
        self.assertRaises(OSPDError, daemon.handle_start_scan_command, cmd)

    def testScanMultiTargetParallel100(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        response = secET.fromstring(
            daemon.handle_command('<start_scan parallel="100">' +
                                  '<scanner_params />' +
                                  '<targets><target>' +
                                  '<hosts>localhosts</hosts>' +
                                  '<ports>22</ports>' +
                                  '</target></targets>' +
                                  '</start_scan>'))
        time.sleep(1)
        self.assertEqual(response.get('status'), '200')

    def testProgress(self):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        daemon = DummyWrapper([])
        response = secET.fromstring(
            daemon.handle_command('<start_scan parallel="2">' +
                                  '<scanner_params />' +
                                  '<targets><target>' +
                                  '<hosts>localhost1</hosts>' +
                                  '<ports>22</ports>' +
                                  '</target><target>' +
                                  '<hosts>localhost2</hosts>' +
                                  '<ports>22</ports>' +
                                  '</target></targets>' +
                                  '</start_scan>'))
        scan_id = response.findtext('id')
        daemon.set_scan_target_progress(scan_id, 'localhost1', 75)
        daemon.set_scan_target_progress(scan_id, 'localhost2', 25)
        self.assertEqual(daemon.calculate_progress(scan_id), 50)

1			# Copyright (C) 2015-2018 Greenbone Networks GmbH
			0 ignored issues – show Coding Style Naming introduced 2018-10-19 19:46 UTC by Report Bug Copy Issue Report The name `testScanAndResult` does not conform to the module naming conventions (`(([a-z_][a-z0-9_]*)\|([A-Z][a-zA-Z0-9]+))$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
2			#
3			# SPDX-License-Identifier: GPL-2.0-or-later
4			#
5			# This program is free software; you can redistribute it and/or
6			# modify it under the terms of the GNU General Public License
7			# as published by the Free Software Foundation; either version 2
8			# of the License, or (at your option) any later version.
9			#
10			# This program is distributed in the hope that it will be useful,
11			# but WITHOUT ANY WARRANTY; without even the implied warranty of
12			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13			# GNU General Public License for more details.
14			#
15			# You should have received a copy of the GNU General Public License
16			# along with this program; if not, write to the Free Software
17			# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
18
19			""" Test module for scan runs
20			"""
21
22			from __future__ import print_function
23
24			import time
25			import unittest
26			import xml.etree.ElementTree as ET
			0 ignored issues – show Unused Code introduced 2018-11-09 11:28 UTC by Report Bug Copy Issue Report Unused xml.etree.ElementTree imported as ET Loading history...
27			import defusedxml.lxml as secET
28			from defusedxml.common import EntitiesForbidden
29
30			from ospd.ospd import OSPDaemon, OSPDError
31
32			class Result(object):
			0 ignored issues – show Coding Style introduced 2018-06-12 10:47 UTC by Report Bug Copy Issue Report This class should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history... best-practice introduced 2018-06-12 10:47 UTC by Report Bug Copy Issue Report Too many instance attributes (8/7) Loading history... Unused Code introduced 2018-06-12 10:47 UTC by Report Bug Copy Issue Report The variable `__class__` seems to be unused. Loading history...
33			def __init__(self, type_, **kwargs):
34			self.result_type = type_
35			self.host = ''
36			self.name = ''
37			self.value = ''
greenbone / ospd

Pull Request — master (#73)

tests.testScanAndResult.FullTest.testStopScan() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like
