Panos error handling (#749)

* Handle errors in panos node

Signed-off-by: Claudio Usai <[email protected]>

* Node: fix call of not-implemented func in case of failed discovery

Signed-off-by: Claudio Usai <[email protected]>

Author: claudious96

suzieq/poller/worker/nodes/node.py

@@ -693,7 +693,9 @@ async def _init_ssh(self, init_dev_data=True, use_lock=True) -> None:
                         self.ssh_ready.release()
 
             # Release here because init_dev_data uses this lock as well
-            if init_dev_data:
+            # We need to be sure that devtype is set, otherwise the
+            # _fetch_dev_data function is not implemented and will raise.
+            if init_dev_data and self.devtype:
                 await self._fetch_init_dev_data()
 
     @abstractmethod
@@ -1904,13 +1906,11 @@ async def _fetch_init_dev_data(self):
                         password=self.password, known_hosts=None) as conn:
                     async with conn.create_process() as process:
                         process.stdin.write("show system info\n")
-                        recv_chars = False
                         output = ""
-                        if not recv_chars:
-                            output += await process.stdout.read(1)
+                        output += await process.stdout.read(1)
                         try:
                             await asyncio.wait_for(
-                                process.wait_closed(), timeout=0.01)
+                                process.wait_closed(), timeout=0.1)
                         except asyncio.TimeoutError:
                             pass
 
@@ -1927,22 +1927,38 @@ async def _fetch_init_dev_data(self):
             )
             if self.api_key is None:
                 await self.get_api_key()
+        except asyncssh.misc.PermissionDenied:
+            self.logger.error(
+                f'Permission denied for {self.address}:{self.port}')
+            self._retry -= 1
         except Exception:
-            pass
+            self.logger.warning(
+                f'Cannot parse device data from {self.address}:{self.port}')
 
     async def get_api_key(self):
         """Authenticate to get the api key needed in all cmd requests"""
         url = f"https://{self.address}:{self.port}/api/?type=keygen&user=" \
             f"{self.username}&password={self.password}"
 
         async with self.cmd_pacer(self.per_cmd_auth):
+            if not self._retry:
+                return
             async with self._session.get(url, timeout=self.connect_timeout) \
                     as response:
                 status, xml = response.status, await response.text()
                 if status == 200:
                     data = xmltodict.parse(xml)
                     self.api_key = data["response"]["result"]["key"]
-                # need to manage errors
+                    # reset retry count, just in case.
+                    self._retry = self._max_retries_on_auth_fail
+                elif status == 403:
+                    self.logger.error('Invalid credentials, could not get api '
+                                      f'key for {self.address}:{self.port}.')
+                    self._retry -= 1
+                else:
+                    self.logger.error('Unknown error, could not get '
+                                      'api key for '
+                                      f'{self.address}:{self.port}.')
 
     async def _parse_init_dev_data(self, output, cb_token) -> None:
         """Parse the uptime command output"""
@@ -1995,6 +2011,12 @@ async def _init_rest(self):
                     )
                     if self.api_key is None:
                         await self.get_api_key()
+                    # If the api_key is still None we can't gather any data.
+                    # Ensure that the connection pool is closed and set it to
+                    # None so that _rest_gather can fail gracefully.
+                    if self.api_key is None:
+                        self._session.close()
+                        self._session = None
                 except Exception as e:
                     self.logger.error(
                         f'{self.transport}://{self.hostname}:{self.port}, '
@@ -2011,12 +2033,15 @@ async def _rest_gather(self, service_callback, cmd_list, cb_token,
 
         now = int(datetime.now(tz=timezone.utc).timestamp() * 1000)
 
-        port = self.port or 443
-
-        url = f"https://{self.address}:{port}/api/"
+        url = f"https://{self.address}:{self.port}/api/"
 
         status = 200  # status OK
 
+        # if there's no session we have failed to get init dev data
+        if not self._session and self._retry:
+            self._fetch_init_dev_data()
+
+        # if there's still no session, we need to create an error
         if not self._session:
             for cmd in cmd_list:
                 result.append(self._create_error(cmd))
@@ -2030,19 +2055,26 @@ async def _rest_gather(self, service_callback, cmd_list, cb_token,
                     async with self._session.get(
                             url_cmd, timeout=timeout) as response:
                         status, xml = response.status, await response.text()
-                        json_out = json.dumps(
-                            xmltodict.parse(xml))
-
-                        result.append({
-                            "status": status,
-                            "timestamp": now,
-                            "cmd": cmd,
-                            "devtype": self.devtype,
-                            "namespace": self.nsname,
-                            "hostname": self.hostname,
-                            "address": self.address,
-                            "data": json_out,
-                        })
+                        if status == 200:
+                            json_out = json.dumps(
+                                xmltodict.parse(xml))
+                            self.logger.info(f"{cmd} {status}")
+                            result.append({
+                                "status": status,
+                                "timestamp": now,
+                                "cmd": cmd,
+                                "devtype": self.devtype,
+                                "namespace": self.nsname,
+                                "hostname": self.hostname,
+                                "address": self.address,
+                                "data": json_out,
+                            })
+                        else:
+                            result.append(self._create_error(cmd))
+                            self.logger.error(
+                                f'{self.transport}://{self.hostname}:'
+                                f'{self.port}: Command {cmd} failed with '
+                                f'status {response.status}')
             except Exception as e:
                 self.current_exception = e
                 for cmd in cmd_list:

suzieq/poller/worker/services/interfaces.py

@@ -1051,5 +1051,6 @@ def _clean_panos_data(self, processed_data, _):
                     entry["type"] = "subinterface"
 
         # remove mtu data cmd
-        processed_data.pop(0)
+        if len(processed_data) > 1:
+            processed_data.pop(0)
         return processed_data