Merge branch 'master' of https://github.com/Neo23x0/Loki

.github/workflows/lint_python.yml

@@ -8,7 +8,7 @@ jobs:
       - uses: actions/setup-python@v4
       - run: pip install --upgrade pip setuptools wheel
       - run: pip install black codespell mypy pytest ruff safety
-      - run: ruff --format=github --ignore=E501,E701,E713,E722,F401,F403,F405,F841 --line-length=263 .
+      - run: ruff --output-format=github --ignore=E501,E701,E713,E722,F401,F403,F405,F841 --line-length=263 .
       - run: black --check . || true
       - run: codespell --ignore-words-list="datas" --skip="./.git/*"
       - run: pip install -r requirements.txt

lib/doublepulsar.py

@@ -210,7 +210,7 @@ def check_ip_rdp(self):
                 return False, "Status Unknown - Response received but length was %d not 288" % (len(ping_response))
 
             s.close()
-        except socket.error as e:
+        except socket.error:
             return False, "No presence of DOUBLEPULSAR RDP implant"
 
 

lib/helpers.py

@@ -8,15 +8,14 @@
 import sys
 import hashlib
 import string
-import struct
 import traceback
 import os
 import re
 import psutil
 try:
     from StringIO import StringIO
 except ImportError:
-    from io import StringIO
+    pass
 import netaddr
 import platform
 import time
@@ -66,7 +65,7 @@ def generateHashes(filedata):
         sha1.update(filedata)
         sha256.update(filedata)
         return md5.hexdigest(), sha1.hexdigest(), sha256.hexdigest()
-    except Exception as e:
+    except Exception:
         traceback.print_exc()
         return 0, 0, 0
 
@@ -75,7 +74,7 @@ def getPlatformFull():
     type_info = ""
     try:
         type_info = "%s PROC: %s ARCH: %s" % ( " ".join(platform.win32_ver()), platform.processor(), " ".join(platform.architecture()))
-    except Exception as e:
+    except Exception:
         type_info = " ".join(platform.win32_ver())
     return type_info
 
@@ -87,7 +86,7 @@ def setNice(logger):
         logger.log("INFO", "Init", "Setting LOKI process with PID: %s to priority IDLE" % pid)
         p.nice(psutil.IDLE_PRIORITY_CLASS)
         return 1
-    except Exception as e:
+    except Exception:
         if logger.debug:
             traceback.print_exc()
         logger.log("ERROR", "Init", "Error setting nice value of THOR process")
@@ -103,7 +102,7 @@ def getExcludedMountpoints():
             if not options[0].startswith("/dev/"):
                 if not options[1] == "/":
                     excludes.append(options[1])
-    except Exception as e:
+    except Exception:
         print ("Error while reading /etc/mtab")
     finally:
         mtab.close()
@@ -174,7 +173,7 @@ def get_file_type(filePath, filetype_sigs, max_filetype_magics, logger):
             if res == bytes.fromhex(sig):
                 return filetype_sigs[sig]
         return "UNKNOWN"
-    except Exception as e:
+    except Exception:
         if logger.debug:
             traceback.print_exc()
         return "UNKNOWN"
@@ -187,10 +186,10 @@ def removeNonAscii(s, stripit=False):
             printable = set(string.printable)
             filtered_string = filter(lambda x: x in printable, s.decode('utf-8'))
             nonascii = ''.join(filtered_string)
-        except Exception as e:
+        except Exception:
             traceback.print_exc()
             nonascii = s.hex()
-    except Exception as e:
+    except Exception:
         traceback.print_exc()
         pass
 
@@ -203,7 +202,7 @@ def removeNonAsciiDrop(s):
         # Generate a new string without disturbing characters
         printable = set(string.printable)
         nonascii = filter(lambda x: x in printable, s)
-    except Exception as e:
+    except Exception:
         traceback.print_exc()
         pass
     return nonascii
@@ -220,7 +219,7 @@ def getAge(filePath):
         # Accessed
         atime=stats.st_atime
 
-    except Exception as e:
+    except Exception:
         # traceback.print_exc()
         return (0, 0, 0)
 
@@ -232,7 +231,7 @@ def getAgeString(filePath):
     timestring = ""
     try:
         timestring = "CREATED: %s MODIFIED: %s ACCESSED: %s" % ( time.ctime(ctime), time.ctime(mtime), time.ctime(atime) )
-    except Exception as e:
+    except Exception:
         timestring = "CREATED: not_available MODIFIED: not_available ACCESSED: not_available"
     return timestring
 
@@ -269,7 +268,7 @@ def _kill_process_after_a_timeout(pid):
         watchdog.cancel() # if it's still waiting to run
         success = not kill_check.isSet()
         kill_check.clear()
-    except Exception as e:
+    except Exception:
         traceback.print_exc()
 
     return output, returnCode

lib/lokilogger.py

@@ -13,7 +13,6 @@
 import logging
 from logging import handlers
 import socket
-from .helpers import removeNonAsciiDrop
 
 __version__ = '0.51.0'
 
@@ -101,7 +100,7 @@ def log(self, mes_type, module, message):
         # to stdout
         try:
             self.log_to_stdout(message, mes_type)
-        except Exception as e:
+        except Exception:
             print ("Cannot print certain characters to command line - see log file for full unicode encoded log line")
             self.log_to_stdout(message, mes_type)
 
@@ -174,7 +173,7 @@ def log_to_stdout(self, message, mes_type):
                 else:
                     sys.stdout.write("%s%s\b\b%s %s%s%s%s\n" % (reset_all, base_color, mes_type, message, Back.BLACK,Fore.WHITE,Style.NORMAL))
 
-            except Exception as e:
+            except Exception:
                 if self.debug:
                     traceback.print_exc()
                     sys.exit(1)
@@ -188,7 +187,7 @@ def log_to_file(self, message, mes_type, module):
                     logfile.write(self.Format(self.FILE_CSV, u"{0},{1},{2},{3},{4}{5}", getSyslogTimestamp(), self.hostname, mes_type, module, message, self.linesep))
                 else:
                     logfile.write(self.Format(self.FILE_LINE, u"{0} {1} LOKI: {2}: MODULE: {3} MESSAGE: {4}{5}", getSyslogTimestamp(), self.hostname, mes_type.title(), module, message, self.linesep))
-        except Exception as e:
+        except Exception:
             if self.debug:
                 traceback.print_exc()
                 sys.exit(1)

lib/pesieve.py

@@ -3,7 +3,6 @@
 # PE-Sieve Integration by @hasherezade
 
 import os
-import sys
 import json
 import traceback
 
@@ -64,12 +63,12 @@ def scan(self, pid, pesieveshellc = False):
             return results
         try:
             results_raw = json.loads(output)
-            results = results_raw["scan_report"]["scanned"]["modified"]
-            #results = results_raw["scanned"]["modified"]
-        except ValueError as v:
+            #results = results_raw["scan_report"]["scanned"]["modified"]
+            results = results_raw["scanned"]["modified"]
+        except ValueError:
             traceback.print_exc()
             self.logger.log("DEBUG", "PESieve", "Couldn't parse the JSON output.")
-        except Exception as e:
+        except Exception:
             traceback.print_exc()
             self.logger.log("ERROR", "PESieve", "Something went wrong during PE-Sieve scan.")
         return results

lib/vuln_checker.py

@@ -5,7 +5,6 @@
 # Different vulnerability checks
 
 import subprocess
-import traceback
 
 class VulnChecker():
 
@@ -27,11 +26,11 @@ def check_sam_readable(self):
         output = b''
         try:
             output += subprocess.check_output([r'icacls.exe', r'C:\Windows\System32\config\sam'], stderr=subprocess.STDOUT)
-        except subprocess.CalledProcessError as e:
+        except subprocess.CalledProcessError:
             pass
         try:
             output += subprocess.check_output([r'icacls.exe', r'C:\Windows\SysNative\config\sam'], stderr=subprocess.STDOUT)
-        except subprocess.CalledProcessError as e:
+        except subprocess.CalledProcessError:
             pass
         # Check the output
         try:
@@ -45,6 +44,6 @@ def check_sam_readable(self):
                 return True
             else:
                 self.logger.log("DEBUG", "VulnChecker", "SAM Database isn't readable by every user.")
-        except UnicodeDecodeError as e:
+        except UnicodeDecodeError:
             self.logger.log("ERROR", "VulnChecker", "Unicode decode error in SAM check")
         return False

loki-upgrader.py

@@ -24,10 +24,8 @@
 # Win32 Imports
 if _platform == "win32":
     try:
-        import wmi
         import win32api
-        from win32com.shell import shell
-    except Exception as e:
+    except Exception:
         platform = "linux"  # crazy guess
 
 
@@ -65,7 +63,7 @@ def needs_update(sig_url):
                 file.write(sha)
                 changed=True
         return changed
-    except Exception as e:
+    except Exception:
         return True
 
 
@@ -94,7 +92,7 @@ def update_signatures(self, clean=False):
                     try:
                         self.logger.log("INFO", "Upgrader", "Downloading %s ..." % sig_url)
                         response = urlopen(sig_url)
-                    except Exception as e:
+                    except Exception:
                         if self.debug:
                             traceback.print_exc()
                         self.logger.log("ERROR", "Upgrader", "Error downloading the signature database - "
@@ -111,7 +109,7 @@ def update_signatures(self, clean=False):
                             fullOutDir = os.path.join(sigDir, outDir)
                             if not os.path.exists(fullOutDir):
                                 os.makedirs(fullOutDir)
-                    except Exception as e:
+                    except Exception:
                         if self.debug:
                             traceback.print_exc()
                         self.logger.log("ERROR", "Upgrader", "Error while creating the signature-base directories")
@@ -157,7 +155,7 @@ def update_signatures(self, clean=False):
                             target.close()
                             source.close()
 
-                    except Exception as e:
+                    except Exception:
                         if self.debug:
                             traceback.print_exc()
                         self.logger.log("ERROR", "Upgrader", "Error while extracting the signature files from the download "
@@ -166,7 +164,7 @@ def update_signatures(self, clean=False):
                 else:
                     self.logger.log("INFO", "Upgrader", "%s is up to date." % sig_url)
 
-        except Exception as e:
+        except Exception:
             if self.debug:
                 traceback.print_exc()
             return False
@@ -185,7 +183,7 @@ def update_loki(self):
                 zip_url = data['assets'][0]['browser_download_url']
                 self.logger.log("INFO", "Upgrader", "Downloading latest release %s ..." % zip_url)
                 response_zip = urlopen(zip_url)
-            except Exception as e:
+            except Exception:
                 if self.debug:
                     traceback.print_exc()
                 self.logger.log("ERROR", "Upgrader", "Error downloading the loki update - check your Internet connection")
@@ -208,7 +206,7 @@ def update_loki(self):
                         if not os.path.exists(os.path.dirname(targetFile)):
                             if os.path.dirname(targetFile) != '':
                                 os.makedirs(os.path.dirname(targetFile))
-                    except Exception as e:
+                    except Exception:
                         if self.debug:
                             self.logger.log("DEBUG", "Upgrader", "Cannot create dir name '%s'" % os.path.dirname(targetFile))
                             traceback.print_exc()
@@ -221,19 +219,19 @@ def update_loki(self):
                             if self.debug:
                                 self.logger.log("DEBUG", "Upgrader", "Successfully extracted '%s'" % targetFile)
                         target.close()
-                    except Exception as e:
+                    except Exception:
                         self.logger.log("ERROR", "Upgrader", "Cannot extract '%s'" % targetFile)
                         if self.debug:
                             traceback.print_exc()
 
-            except Exception as e:
+            except Exception:
                 if self.debug:
                     traceback.print_exc()
                 self.logger.log("ERROR", "Upgrader",
                                 "Error while extracting the signature files from the download package")
                 sys.exit(1)
 
-        except Exception as e:
+        except Exception:
             if self.debug:
                 traceback.print_exc()
             return False
@@ -253,7 +251,7 @@ def get_application_path():
         #if args.debug:
         #    logger.log("DEBUG", "Init", "Application Path: %s" % application_path)
         return application_path
-    except Exception as e:
+    except Exception:
         print("Error while evaluation of application path")
         traceback.print_exc()