added initial code to use authentication config for nessus (#267)

* added initial code to use authentication config for nessus

rapidast.py

@@ -65,6 +65,8 @@ def load_config(config_file_location: str) -> Dict[str, Any]:
     return yaml.safe_load(load_config_file(config_file_location))
 
 
+# pylint: disable=R0911
+# too many return statements
 def run_scanner(name, config, args, scan_exporter):
     """given the config `config`, runs scanner `name`.
     Returns:
@@ -94,8 +96,12 @@ def run_scanner(name, config, args, scan_exporter):
     # Part 1: create a instance based on configuration
     try:
         scanner = class_(config, name)
-    except OSError as excp:
-        logging.error(excp)
+    except OSError as e:
+        logging.error(f"Caught exception: {e}")
+        logging.error(f"Ignoring failed Scanner `{name}` of type `{typ}`")
+        return 1
+    except RuntimeError as e:
+        logging.error(f"Caught exception: {e}")
         logging.error(f"Ignoring failed Scanner `{name}` of type `{typ}`")
         return 1
 

scanners/nessus/nessus_none.py

@@ -5,6 +5,8 @@
 from dataclasses import field
 from os import listdir
 from os import path
+from typing import Any
+from typing import Dict
 from typing import List
 from typing import Optional
 
@@ -15,9 +17,16 @@
 from configmodel import RapidastConfigModel
 from scanners import RapidastScanner
 from scanners import State
+from scanners.authentication_factory import generic_authentication_factory
 from scanners.nessus.tools.convert_nessus_csv_to_sarif import convert_csv_to_sarif
 
 
+@dataclass
+class NessusAuthenticationConfig:
+    type: str
+    parameters: Dict[str, Any]
+
+
 @dataclass
 class NessusServerConfig:
     url: str
@@ -39,6 +48,7 @@ def targets_as_str(self) -> str:
 
 @dataclass
 class NessusConfig:
+    authentication: Optional[NessusAuthenticationConfig]
     server: NessusServerConfig
     scan: NessusScanConfig
 
@@ -62,10 +72,13 @@ def __init__(self, config: RapidastConfigModel, ident: str = "nessus"):
         nessus_config_section = config.subtree_to_dict(f"scanners.{ident}")
         if nessus_config_section is None:
             raise ValueError("'scanners.nessus' section not in config")
-        # self.auth_config = config.subtree_to_dict("general.authentication")  # creds for target hosts
+
         # XXX self.config is already a dict with raw config values
         self.cfg = dacite.from_dict(data_class=NessusConfig, data=nessus_config_section)
         self._sleep_interval: int = 10
+
+        self.authenticated = self.authentication_factory()
+
         self._connect()
 
     def _connect(self):
@@ -93,6 +106,20 @@ def scan_id(self) -> int:
             raise RuntimeError("scan_id is None")
         return self._scan_id
 
+    @generic_authentication_factory()
+    def authentication_factory(self):
+        """This is the default function, attached to error reporting"""
+        raise RuntimeError(
+            f"The authentication option is not supported. "
+            f"Input - type: {self.cfg.authentication.type}, params: {self.cfg.authentication.parameters}"
+        )
+
+    @authentication_factory.register(None)
+    def authentication_set_anonymous(self):
+        """No authentication: don't do anything"""
+        logging.info("Nessus scan not configured with any auth")
+        return False
+
     def setup(self):
         logging.debug(f"Creating new scan named {self.cfg.scan.folder}/{self.cfg.scan.name}")
         self._scan_id = self.nessus_client.new_scan(

tests/scanners/nessus/test_nessus.py

@@ -1,6 +1,7 @@
 from unittest.mock import Mock
 from unittest.mock import patch
 
+import pytest
 import requests
 
 import configmodel
@@ -22,3 +23,20 @@ def test_setup_nessus(self, mock_get, auth):
         test_nessus = Nessus(config=config)
         assert test_nessus is not None
         assert test_nessus.nessus_client is not None
+
+    @patch("py_nessus_pro.PyNessusPro._authenticate")
+    @patch("requests.Session.request")
+    def test_setup_nessus_auth(self, mock_get, auth):
+        # All this mocking is for PyNessusPro.__init__() which attempts to connect to Nessus
+        mock_get.return_value = Mock(spec=requests.Response)
+        mock_get.return_value.status_code = 200
+        mock_get.return_value.text = '{"token": "foo", "folders": []}'
+
+        config_data = rapidast.load_config("config/config-template-nessus.yaml")
+        config = configmodel.RapidastConfigModel(config_data)
+
+        authentication = {"type": "invalid", "parameters": {"name": "Authorizaiton", "value": "123"}}
+        config.set("scanners.nessus.authentication", authentication)
+
+        with pytest.raises(RuntimeError, match="The authentication option is not supported"):
+            test_nessus = Nessus(config=config)