Merge pull request #411 from jjnicola/port-list-validation

Port list validation

CHANGELOG.md

@@ -7,6 +7,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 ## [21.10] (unreleased)
 
 ### Added
+- Validate port list to be sent to openvas. [#411](https://github.com/greenbone/ospd-openvas/pull/411)
+
 ### Changed
 ### Removed
 ### Fixed

ospd_openvas/daemon.py

@@ -1258,7 +1258,7 @@ def exec_scan(self, scan_id: str):
 
         if not scan_prefs.prepare_ports_for_openvas():
             self.add_scan_error(
-                scan_id, name='', host='', value='No port list defined.'
+                scan_id, name='', host='', value='Invalid port list.'
             )
             do_not_launch = True
 

ospd_openvas/preferencehandler.py

@@ -32,6 +32,7 @@
 
 from ospd.scan import ScanCollection
 from ospd.ospd import BASE_SCANNER_PARAMS
+from ospd.network import valid_port_list
 from ospd_openvas.openvas import Openvas
 from ospd_openvas.db import KbDB
 from ospd_openvas.nvticache import NVTICache
@@ -528,6 +529,9 @@ def prepare_ports_for_openvas(self) -> str:
         """Get the port list from the scan collection and store the list
         in the kb."""
         ports = self.scan_collection.get_ports(self.scan_id)
+        if not valid_port_list(ports):
+            return False
+
         port_range = 'port_range|||%s' % ports
         self.kbdb.add_scan_preferences(self.scan_id, [port_range])
 

tests/test_preferencehandler.py

@@ -280,6 +280,17 @@ def test_set_ports(self, mock_kb):
             ['port_range|||80,443'],
         )
 
+    @patch('ospd_openvas.db.KbDB')
+    def test_set_ports_invalid(self, mock_kb):
+        w = DummyDaemon()
+
+        w.scan_collection.get_ports = MagicMock(return_value='2,-9,4')
+
+        p = PreferenceHandler('1234-1234', mock_kb, w.scan_collection, None)
+        p.scan_id = '456-789'
+        p.kbdb.add_scan_preferences = MagicMock()
+        self.assertFalse(p.prepare_ports_for_openvas())
+
     @patch('ospd_openvas.db.KbDB')
     def test_set_main_kbindex(self, mock_kb):
         w = DummyDaemon()