Fix 86 Pyright errors in security-critical paths (Phase 3)

src/python/controller/auto_queue.py

@@ -88,8 +88,8 @@ def add_listener(self, listener: IAutoQueuePersistListener):
 
     @classmethod
     @overrides(Persist)
-    def from_str(cls: "AutoQueuePersist", content: str) -> "AutoQueuePersist":
-        persist = AutoQueuePersist()
+    def from_str(cls: type["AutoQueuePersist"], content: str) -> "AutoQueuePersist":
+        persist = cls()
         try:
             dct = json.loads(content)
             pattern_list = dct[AutoQueuePersist.__KEY_PATTERNS]
@@ -356,6 +356,8 @@ def _is_auto_queue_enabled_for_file(self, file: ModelFile) -> bool:
         caller (process()) already gates on the global __enabled flag.
         """
         if self.__pair_auto_queue:
+            if file.pair_id is None:
+                return False
             return self.__pair_auto_queue.get(file.pair_id, False)
         return True
 
@@ -405,8 +407,8 @@ def __match(pattern: AutoQueuePattern, file: ModelFile) -> bool:
         :return:
         """
         # make the search case insensitive
-        pattern = pattern.pattern.lower()
+        pattern_str = pattern.pattern.lower()
         filename = file.name.lower()
         # 1. pattern match
         # 2. wildcard match
-        return pattern in filename or fnmatch.fnmatch(filename, pattern)
+        return pattern_str in filename or fnmatch.fnmatch(filename, pattern_str)

src/python/controller/extract/extract_process.py

@@ -1,10 +1,10 @@
 # Copyright 2017, Inderpreet Singh, All rights reserved.
 
-import datetime
 import logging
 import multiprocessing
 import queue
 import time
+from datetime import datetime
 
 from common import AppProcess, overrides
 
@@ -19,15 +19,15 @@ def __init__(self, timestamp: datetime, statuses: list[ExtractStatus]):
 
 
 class ExtractCompletedResult:
-    def __init__(self, timestamp: datetime, name: str, is_dir: bool, pair_id: str = None):
+    def __init__(self, timestamp: datetime, name: str, is_dir: bool, pair_id: str | None = None):
         self.timestamp = timestamp
         self.name = name
         self.is_dir = is_dir
         self.pair_id = pair_id
 
 
 class ExtractFailedResult:
-    def __init__(self, timestamp: datetime, name: str, is_dir: bool, pair_id: str = None):
+    def __init__(self, timestamp: datetime, name: str, is_dir: bool, pair_id: str | None = None):
         self.timestamp = timestamp
         self.name = name
         self.is_dir = is_dir
@@ -45,18 +45,16 @@ def __init__(
             self.completed_queue = completed_queue
             self.failed_queue = failed_queue
 
-        def extract_completed(self, name: str, is_dir: bool, pair_id: str = None):
+        def extract_completed(self, name: str, is_dir: bool, pair_id: str | None = None):
             self.logger.info("Extraction completed for {}".format(name))
             completed_result = ExtractCompletedResult(
-                timestamp=datetime.datetime.now(), name=name, is_dir=is_dir, pair_id=pair_id
+                timestamp=datetime.now(), name=name, is_dir=is_dir, pair_id=pair_id
             )
             self.completed_queue.put(completed_result)
 
-        def extract_failed(self, name: str, is_dir: bool, pair_id: str = None):
+        def extract_failed(self, name: str, is_dir: bool, pair_id: str | None = None):
             self.logger.error("Extraction failed for {}".format(name))
-            failed_result = ExtractFailedResult(
-                timestamp=datetime.datetime.now(), name=name, is_dir=is_dir, pair_id=pair_id
-            )
+            failed_result = ExtractFailedResult(timestamp=datetime.now(), name=name, is_dir=is_dir, pair_id=pair_id)
             self.failed_queue.put(failed_result)
 
     def __init__(self):
@@ -83,10 +81,12 @@ def run_init(self):
 
     @overrides(AppProcess)
     def run_cleanup(self):
+        assert self.__dispatch is not None
         self.__dispatch.stop()
 
     @overrides(AppProcess)
     def run_loop(self):
+        assert self.__dispatch is not None
         # Forward all the extract commands
         try:
             while True:
@@ -98,7 +98,7 @@ def run_loop(self):
                     # Report dispatch errors as failures so the controller
                     # can transition the file to EXTRACT_FAILED state
                     failed_result = ExtractFailedResult(
-                        timestamp=datetime.datetime.now(),
+                        timestamp=datetime.now(),
                         name=req.model_file.name,
                         is_dir=req.model_file.is_dir,
                         pair_id=req.pair_id,
@@ -109,7 +109,7 @@ def run_loop(self):
 
         # Queue the latest status
         statuses = self.__dispatch.status()
-        status_result = ExtractStatusResult(timestamp=datetime.datetime.now(), statuses=statuses)
+        status_result = ExtractStatusResult(timestamp=datetime.now(), statuses=statuses)
         self.__status_result_queue.put(status_result)
 
         time.sleep(ExtractProcess.__DEFAULT_SLEEP_INTERVAL_IN_SECS)

src/python/lftp/lftp.py

@@ -106,7 +106,7 @@ def __spawn_process(self):
         # Suppress DeprecationWarning from pexpect.spawn's internal forkpty call.
         with warnings.catch_warnings():
             warnings.filterwarnings("ignore", message=".*fork.*", category=DeprecationWarning)
-            self.__process = pexpect.spawn("/usr/bin/lftp", args, env=spawn_env)
+            self.__process = pexpect.spawn("/usr/bin/lftp", args, env=spawn_env)  # type: ignore[arg-type]
         self.__process.setwinsize(24, 10000)
         self.__process.expect(self.__expect_pattern)
         self.__setup()
@@ -122,7 +122,7 @@ def __restart_process(self):
         self.__spawn_process()
         # Replay cached settings
         for setting, value in self.__settings_cache.items():
-            self.__run_command("set {} {}".format(setting, value))
+            self.__run_command("set {} {}".format(setting, value))  # type: ignore[arg-type]
         self.__consecutive_timeouts = 0
 
     def __setup(self):
@@ -139,7 +139,7 @@ def __setup(self):
         # permission bits (e.g. 664 → remote) which would override our umask setting
         self.__set(Lftp.__SET_SFTP_SET_PERMISSIONS, "false")
 
-    def with_check_process(method: Callable):
+    def with_check_process(method: Callable):  # type: ignore[override]
         """
         Decorator that checks for a valid process before executing
         the decorated method. Attempts restart if process is dead.
@@ -181,7 +181,8 @@ def raise_pending_error(self):
             raise LftpError(error)
 
     @with_check_process
-    def __run_command(self, command: str):
+    def __run_command(self, command: str):  # type: ignore[arg-type]
+        assert self.__process is not None
         if self.__log_command_output:
             self.logger.debug("command: {}".format(command.encode("utf8", "surrogateescape")))
         self.__process.sendline(command)
@@ -204,14 +205,17 @@ def __run_command(self, command: str):
 
         # Success — reset consecutive timeout counter
         self.__consecutive_timeouts = 0
-        out = self.__process.before.decode("utf8", "replace")
+        before = self.__process.before
+        assert isinstance(before, bytes)
+        out = before.decode("utf8", "replace")
         out = out.strip()  # remove any CRs
 
         if self.__log_command_output:
             self.logger.debug("out ({} bytes):\n {}".format(len(out), out))
+            after_val = self.__process.after
             after = (
-                self.__process.after.decode("utf8", "replace").strip()
-                if self.__process.after != pexpect.TIMEOUT
+                after_val.decode("utf8", "replace").strip()  # type: ignore[union-attr]
+                if after_val != pexpect.TIMEOUT
                 else ""
             )
             self.logger.debug("after: {}".format(after))
@@ -227,13 +231,16 @@ def __run_command(self, command: str):
                 self.logger.warning("Lftp timeout while consuming error output")
                 self.__pending_error = error_out
                 return ""
-            out = self.__process.before.decode("utf8", "replace")
+            before = self.__process.before
+            assert isinstance(before, bytes)
+            out = before.decode("utf8", "replace")
             out = out.strip()  # remove any CRs
             if self.__log_command_output:
                 self.logger.debug("retry out ({} bytes):\n {}".format(len(out), out))
+                after_val = self.__process.after
                 after = (
-                    self.__process.after.decode("utf8", "replace").strip()
-                    if self.__process.after != pexpect.TIMEOUT
+                    after_val.decode("utf8", "replace").strip()  # type: ignore[union-attr]
+                    if after_val != pexpect.TIMEOUT
                     else ""
                 )
                 self.logger.debug("retry after: {}".format(after))
@@ -263,15 +270,15 @@ def __set(self, setting: str, value: str):
         :return:
         """
         self.__settings_cache[setting] = value
-        self.__run_command("set {} {}".format(setting, value))
+        self.__run_command("set {} {}".format(setting, value))  # type: ignore[arg-type]
 
     def __get(self, setting: str) -> str:
         """
         Get a setting from the lftp runtime
         :param setting:
         :return:
         """
-        out = self.__run_command("set -a | grep {}".format(setting))
+        out = self.__run_command("set -a | grep {}".format(setting))  # type: ignore[arg-type]
         m = re.search("set {} (.*)".format(setting), out)
         if not m or not m.group or not m.group(1):
             raise LftpError("Failed to get setting '{}'. Output: '{}'".format(setting, out))
@@ -473,7 +480,7 @@ def status(self) -> list[LftpJobStatus] | None:
         completion signals.
         :return:
         """
-        out = self.__run_command("jobs -v")
+        out = self.__run_command("jobs -v")  # type: ignore[arg-type]
         try:
             statuses = self.__job_status_parser.parse(out)
             self.__consecutive_status_errors = 0
@@ -486,7 +493,7 @@ def status(self) -> list[LftpJobStatus] | None:
                 raise
         return statuses
 
-    def queue(self, name: str, is_dir: bool, exclude_patterns: list = None):
+    def queue(self, name: str, is_dir: bool, exclude_patterns: list | None = None):
         """
         Queues a job for download
         This method may cause an exception to be generated in a later method call:
@@ -528,7 +535,7 @@ def escape(s: str) -> str:
         )
         command = " ".join(parts)
         self.logger.info("queue command: %s", command)
-        self.__run_command(command)
+        self.__run_command(command)  # type: ignore[arg-type]
 
     def kill(self, name: str) -> bool:
         """
@@ -538,7 +545,11 @@ def kill(self, name: str) -> bool:
         """
         # look for this name in the status list
         job_to_kill = None
-        for status in self.status():
+        statuses = self.status()
+        if statuses is None:
+            self.logger.debug("Kill failed - status unavailable for job '{}'".format(name))
+            return False
+        for status in statuses:
             if status.name == name:
                 job_to_kill = status
                 break
@@ -550,10 +561,10 @@ def kill(self, name: str) -> bool:
         #       in this case the wrong job may be killed, there's nothing we can do about it
         if job_to_kill.state == LftpJobStatus.State.RUNNING:
             self.logger.debug("Killing running job '{}'...".format(name))
-            self.__run_command("kill {}".format(job_to_kill.id))
+            self.__run_command("kill {}".format(job_to_kill.id))  # type: ignore[arg-type]
         elif job_to_kill.state == LftpJobStatus.State.QUEUED:
             self.logger.debug("Killing queued job '{}'...".format(name))
-            self.__run_command("queue --delete {}".format(job_to_kill.id))
+            self.__run_command("queue --delete {}".format(job_to_kill.id))  # type: ignore[arg-type]
         else:
             raise NotImplementedError("Unsupported state {}".format(str(job_to_kill.state)))
         return True
@@ -564,18 +575,19 @@ def kill_all(self):
         :return:
         """
         # empty the queue and kill running jobs
-        self.__run_command("queue -d *")
-        self.__run_command("kill all")
+        self.__run_command("queue -d *")  # type: ignore[arg-type]
+        self.__run_command("kill all")  # type: ignore[arg-type]
 
     def exit(self):
         """
         Exit the lftp instance. It cannot be used after being killed
         :return:
         """
         self.kill_all()
+        assert self.__process is not None
         self.__process.sendline("exit")
         self.__process.close(force=True)
 
     # Mark decorators as static (must be at end of class)
     # Source: https://stackoverflow.com/a/3422823
-    with_check_process = staticmethod(with_check_process)
+    with_check_process = staticmethod(with_check_process)  # type: ignore[arg-type]

src/python/ssh/sshcp.py

@@ -293,20 +293,24 @@ def __run_command(self, command: str, flags: str, args: str) -> bytes:
                     ]
                 )
                 if i > 0:
-                    before = sp.before.decode(errors="replace").strip() if sp.before != pexpect.EOF else ""
-                    after = sp.after.decode(errors="replace").strip() if sp.after != pexpect.EOF else ""
+                    before_val = sp.before
+                    after_val = sp.after
+                    before = before_val.decode(errors="replace").strip() if isinstance(before_val, bytes) else ""
+                    after = after_val.decode(errors="replace").strip() if isinstance(after_val, bytes) else ""
                     self.logger.warning("Command failed: '{} - {}'".format(before, after))
                 if i == 1:
                     error_msg = "Unknown error"
-                    if sp.before.decode(errors="replace").strip():
-                        error_msg += " - " + sp.before.decode(errors="replace").strip()
+                    before_val = sp.before
+                    if isinstance(before_val, bytes) and before_val.decode(errors="replace").strip():
+                        error_msg += " - " + before_val.decode(errors="replace").strip()
                     raise SshcpError(error_msg)
                 elif i == 3:
                     raise SshcpError("Bad hostname: {}".format(self.__host))
                 elif i in {2, 4}:
                     error_msg = "Connection refused by server"
-                    if sp.before.decode(errors="replace").strip():
-                        error_msg += " - " + sp.before.decode(errors="replace").strip()
+                    before_val = sp.before
+                    if isinstance(before_val, bytes) and before_val.decode(errors="replace").strip():
+                        error_msg += " - " + before_val.decode(errors="replace").strip()
                     raise SshcpError(error_msg)
                 sp.sendline(self.__password)
 
@@ -321,23 +325,29 @@ def __run_command(self, command: str, flags: str, args: str) -> bytes:
                 timeout=self.__TIMEOUT_SECS,
             )
             if i > 0:
-                before = sp.before.decode(errors="replace").strip() if sp.before != pexpect.EOF else ""
-                after = sp.after.decode(errors="replace").strip() if sp.after != pexpect.EOF else ""
+                before_val = sp.before
+                after_val = sp.after
+                before = before_val.decode(errors="replace").strip() if isinstance(before_val, bytes) else ""
+                after = after_val.decode(errors="replace").strip() if isinstance(after_val, bytes) else ""
                 self.logger.warning("Command failed: '{} - {}'".format(before, after))
             if i == 1:
                 raise SshcpError("Incorrect password")
             elif i == 3:
                 raise SshcpError("Bad hostname: {}".format(self.__host))
             elif i in {2, 4}:
                 error_msg = "Connection refused by server"
-                if sp.before.decode(errors="replace").strip():
-                    error_msg += " - " + sp.before.decode(errors="replace").strip()
+                before_val = sp.before
+                if isinstance(before_val, bytes) and before_val.decode(errors="replace").strip():
+                    error_msg += " - " + before_val.decode(errors="replace").strip()
                 raise SshcpError(error_msg)
 
             # Capture output attributes while sp is still open (close can clear them)
-            out_before = sp.before.decode(errors="replace").strip() if sp.before != pexpect.EOF else ""
-            out_after = sp.after.decode(errors="replace").strip() if sp.after != pexpect.EOF else ""
-            out_raw = sp.before.replace(b"\r\n", b"\n").strip()
+            before_val = sp.before
+            after_val = sp.after
+            out_before = before_val.decode(errors="replace").strip() if isinstance(before_val, bytes) else ""
+            out_after = after_val.decode(errors="replace").strip() if isinstance(after_val, bytes) else ""
+            assert isinstance(before_val, bytes)
+            out_raw = before_val.replace(b"\r\n", b"\n").strip()
 
         except pexpect.exceptions.TIMEOUT:
             elapsed = time.time() - start_time