Merge pull request #12 from cowrie/claude/c17-c23-comparison-UEqFB

Author: micheloosterhof

modules/python/dionaea/__init__.py

@@ -170,7 +170,7 @@ def load_config_from_files(filename_patterns: list[str]) -> list[dict[str, Any]]
                         e.problem  # type: ignore[attr-defined]
                     )
                     if e.context is not None:  # type: ignore[attr-defined]
-                        logger.debug("Parser(context): %s" % e.context)  # type: ignore[attr-defined]
+                        logger.debug(f"Parser(context): {e.context}")  # type: ignore[attr-defined]
                 else:
                     logger.error("Unknown error while parsing config file '%s'", filename)
 

modules/python/dionaea/cmd.py

@@ -90,7 +90,7 @@ def execute(self, cmd, args):
 
     def cmd_ECHO(self, args):
         out = " ".join(args) + '\n'
-        logger.debug("echo %s" % (out))
+        logger.debug(f"echo {out}")
         return out,None
 
     def cmd_FTP(self, args):
@@ -140,11 +140,11 @@ def cmd_FTP(self, args):
         for i in range(len(lines)):
             line = lines[i]
             logger.debug(f"FTP PARSER STATE {state}")
-            logger.debug("FTP CMD LINE: %s" % (line) )
+            logger.debug(f"FTP CMD LINE: {line}")
             args = line.split()
             if len(args) == 0:
                 continue
-            logger.debug("FTP CMD ARGS: %s" % (args) )
+            logger.debug(f"FTP CMD ARGS: {args}")
             if state == 'NEXT_IS_SOMETHING':
                 if args[0] == 'open':
                     if len(args) == 1:
@@ -181,8 +181,7 @@ def cmd_FTP(self, args):
                             i.con = self
                         elif hasattr(self, 'con') and isinstance(self.con, connection):
                             i.con = self.con
-                        i.url = "ftp://%s:%s@%s:%i/%s" % (
-                            user,passwd,host,port,dfile)
+                        i.url = f"ftp://{user}:{passwd}@{host}:{port}/{dfile}"
                         i.ftpmode = ftpmode
                         i.report()
                 elif args[0] == 'cd':
@@ -228,20 +227,18 @@ def cmd_FTP(self, args):
                     elif hasattr(self, 'con') and isinstance(self.con, connection):
                         i.con = self.con
                     i.ftpmode = ftpmode
-                    i.url = "ftp://%s:%s@%s:%i/%s" % (user,
-                                                      passwd,host,port,dfile)
+                    i.url = f"ftp://{user}:{passwd}@{host}:{port}/{dfile}"
                     i.report()
 
             elif state == 'NEXT_IS_PATH':
                 if len(args) == 1:
                     fpath = args[0]
                     state = 'NEXT_IS_SOMETHING'
-        logger.info("ftp://%s:%s@%s:%i/%s/%s" %
-                    (user,passwd,host,port,fpath,dfile))
+        logger.info(f"ftp://{user}:{passwd}@{host}:{port}/{fpath}/{dfile}")
         return out,None
 
     def cmd_TFTP(self, args):
-        logger.debug("TFTP %s" % (args) )
+        logger.debug(f"TFTP {args}")
         if len(args) != 4:
             logger.warning("TFTP command invalid number of args: expected 4, got %d", len(args))
             return "foo","error, invalid number of args"
@@ -388,7 +385,7 @@ def handle_origin(self, parent):
 class cmdshellhandler(ihandler):
 
     def __init__(self, path, config=None):
-        logger.debug("%s ready!" % (self.__class__.__name__))
+        logger.debug(f"{self.__class__.__name__} ready!")
         ihandler.__init__(self, path)
 
     def handle_incident(self, icd):

modules/python/dionaea/emu.py

@@ -26,7 +26,7 @@ def start(cls, config: dict[str, Any] | None = None) -> 'emuprofilehandler':
 class emuprofilehandler(ihandler):
 
     def __init__(self, path: str, config: dict[str, Any] | None = None) -> None:
-        logger.debug("%s ready!" % (self.__class__.__name__))
+        logger.debug(f"{self.__class__.__name__} ready!")
         ihandler.__init__(self, path)
 
     def handle_incident(self, icd: incident) -> None:
@@ -39,7 +39,7 @@ def handle_incident(self, icd: incident) -> None:
             con = None
         p = json.loads(p)
 #        print(p)
-        logger.info("profiledump %s" % (p))
+        logger.info(f"profiledump {p}")
         state = "NONE"
         host = None
         port = None
@@ -51,7 +51,7 @@ def handle_incident(self, icd: incident) -> None:
                     state = "SOCKET"
                 if api['call'] == 'URLDownloadToFile':
                     url = api['args'][1]
-                    logger.debug("download file %s" % (url))
+                    logger.debug(f"download file {url}")
                     i = incident("dionaea.download.offer")
                     i.set("url", url)
                     if con is not None:
@@ -88,7 +88,7 @@ def handle_incident(self, icd: incident) -> None:
 
             elif state == "ACCEPT":
                 if api['call'] == 'CreateProcess':
-                    logger.debug("bindshell host %s port %s"  % (host, port) )
+                    logger.debug(f"bindshell host {host} port {port}")
                     i = incident("dionaea.service.shell.listen")
                     assert port is not None  # For mypy
                     i.set("port", int(port))
@@ -98,8 +98,7 @@ def handle_incident(self, icd: incident) -> None:
 
             elif state == "CONNECT":
                 if api['call'] == 'CreateProcess':
-                    logger.debug(
-                        "connectbackshell host %s port %s"  % (host, port) )
+                    logger.debug(f"connectbackshell host {host} port {port}")
                     i = incident("dionaea.service.shell.connect")
                     assert port is not None  # For mypy
                     i.set("port", int(port))
@@ -112,8 +111,7 @@ def handle_incident(self, icd: incident) -> None:
                 if api['call'] == 'connect':
                     host = api['args'][1]['sin_addr']['s_addr']
                     port = api['args'][1]['sin_port']
-                    logger.debug(
-                        "connectbackshell host %s port %s"  % (host, port) )
+                    logger.debug(f"connectbackshell host {host} port {port}")
                     i = incident("dionaea.service.shell.connect")
                     assert port is not None  # For mypy
                     i.set("port", int(port))

modules/python/dionaea/emu_scripts/handler.py

@@ -4,7 +4,7 @@
 #
 # SPDX-License-Identifier: GPL-2.0-or-later
 
-from typing import Any, Pattern
+from typing import Any
 import logging
 import re
 
@@ -20,9 +20,9 @@ def __init__(self, config: dict[str, Any] | None = None) -> None:
             self._config = config
 
         self.min_match_count: int = 0
-        self._regex_detect: list[Pattern[bytes]] = []
+        self._regex_detect: list[re.Pattern[bytes]] = []
 
-        self._regex_url: Pattern[bytes] = re.compile(
+        self._regex_url: re.Pattern[bytes] = re.compile(
             br"(?P<url>(http|ftp|https)://([\w_-]+(?:(?:\.[\w_-]+)+))([\w.,@?^=%&:/~+#-]*[\w@?^=%&/~+#-])?)"
         )
 
@@ -52,7 +52,7 @@ def __init__(self, config: dict[str, Any] | None = None) -> None:
         if isinstance(config, dict):
             self._config = config
 
-        self._regex_url: Pattern[bytes] = re.compile(
+        self._regex_url: re.Pattern[bytes] = re.compile(
             br"(?P<url>(http|ftp|https)://([\w_-]+(?:(?:\.[\w_-]+)+))([\w.,@?^=%&:/~+#-]*[\w@?^=%&/~+#-])?)"
         )
 

modules/python/dionaea/exception.py

@@ -35,9 +35,7 @@ def __init__(self, connection: Any = None, error_id: int | None = None) -> None:
 
 class ConnectionDNSTimeout(DionaeaConnectionError):
     def __str__(self) -> str:
-        return "Timeout resolving the hostname/domain: %s" % (
-            self.connection.remote.hostname
-        )
+        return f"Timeout resolving the hostname/domain: {self.connection.remote.hostname}"
 
 
 class ConnectionUnreachable(DionaeaConnectionError):
@@ -46,17 +44,12 @@ def __str__(self) -> str:
         if hostname is None or hostname == "":
             hostname = self.connection.remote.host
 
-        return "Could not connect to host(s): %s:%d" % (
-            hostname,
-            self.connection.remote.port
-        )
+        return f"Could not connect to host(s): {hostname}:{self.connection.remote.port}"
 
 
 class ConnectionNoSuchDomain(DionaeaConnectionError):
     def __str__(self) -> str:
-        return "Could not resolve the domain: %s" % (
-            self.connection.remote.hostname
-        )
+        return f"Could not resolve the domain: {self.connection.remote.hostname}"
 
 
 class ConnectionTooMany(DionaeaConnectionError):
@@ -67,6 +60,4 @@ def __str__(self) -> str:
 class ConnectionUnknownError(DionaeaConnectionError):
     def __str__(self) -> str:
         assert self.error_id is not None  # For mypy
-        return "Unknown error occurred: error_id=%d" % (
-            self.error_id
-        )
+        return f"Unknown error occurred: error_id={self.error_id}"

modules/python/dionaea/fail2ban.py

@@ -30,7 +30,7 @@ def start(cls, config: dict[str, Any] | None = None) -> 'fail2banhandler | None'
 
 class fail2banhandler(ihandler):
     def __init__(self, config: dict[str, Any] | None = None) -> None:
-        logger.debug("%s ready!" % (self.__class__.__name__))
+        logger.debug(f"{self.__class__.__name__} ready!")
         ihandler.__init__(self, "*")
         if config is None:
             config = {}

modules/python/dionaea/ftp.py

@@ -191,7 +191,7 @@ def reply(self, name, **kwargs):
         self.sendline(msg.format(**kwargs))
 
     def handle_origin(self, parent):
-        logger.debug("setting basedir to %s" % parent.basedir)
+        logger.debug(f"setting basedir to {parent.basedir}")
         self.basedir = parent.basedir
 
     def handle_established(self):
@@ -231,7 +231,7 @@ def handle_io_in(self, data: bytes) -> int:
         return lastsep
 
     def processcmd(self, cmd, args):
-        logger.debug("cmd '%s'" % cmd)
+        logger.debug(f"cmd '{cmd}'")
         arguments = [i.decode('latin-1') for i in args]
 
         i = incident("dionaea.modules.python.ftp.command")
@@ -307,9 +307,9 @@ def ftp_PORT(self, address):
             self.dtp.close()
             self.dtp = None
         addr = list(map(int, address.split(',')))
-        ip = '%d.%d.%d.%d' % tuple(addr[:4])
+        ip = f'{addr[0]}.{addr[1]}.{addr[2]}.{addr[3]}'
         port = addr[4] << 8 | addr[5]
-        logger.debug("PORT cmd for port %i" % port)
+        logger.debug(f"PORT cmd for port {port}")
         if self.remote.host != ip and "::ffff:" + self.remote.host != ip:
             logger.warning("Potential FTP Bounce Scan detected")
             return None
@@ -582,7 +582,7 @@ def handle_error(self, err):
 
     def send_list(self, p, rm):
         def ls(f, r):
-            logger.debug("stat %s" % f)
+            logger.debug(f"stat {f}")
             name = f[r:]
             s=os.stat(f)
             size = s.st_size
@@ -598,41 +598,26 @@ def formatMode(mode):
 
             def formatDate(mtime):
                 now = time.gmtime()
-                info = {
-                    'month': mtime.tm_mon,
-                    'day': mtime.tm_mday,
-                    'year': mtime.tm_year,
-                    'hour': mtime.tm_hour,
-                    'minute': mtime.tm_min
-                }
                 if now.tm_year != mtime.tm_year:
-                    return '%(month)s %(day)02d %(year)5d' % info
+                    return f'{mtime.tm_mon} {mtime.tm_mday:02d} {mtime.tm_year:5d}'
                 else:
-                    return '%(month)s %(day)02d %(hour)02d:%(minute)02d' % info
-
-            format = (
-                '%(directory)s%(permissions)s%(hardlinks)4d '
-                '%(owner)-9s %(group)-9s %(size)15d %(date)12s '
-                '%(name)s'
+                    return f'{mtime.tm_mon} {mtime.tm_mday:02d} {mtime.tm_hour:02d}:{mtime.tm_min:02d}'
+
+            dir_char = 'd' if directory else '-'
+            date_str = formatDate(time.gmtime(modified))
+            return (
+                f'{dir_char}{formatMode(permissions)}{hardlinks:4d} '
+                f'{owner:<9} {group:<9} {size:15d} {date_str:>12} '
+                f'{name}'
             )
-            return format % {
-                'directory': directory and 'd' or '-',
-                'permissions': formatMode(permissions),
-                'hardlinks': hardlinks,
-                'owner': owner,
-                'group': group,
-                'size': size,
-                'date': formatDate(time.gmtime(modified)),
-                'name': name
-            }
 
         self.mode = 'list'
         if os.path.isdir(p):
             self.data = [ls(os.path.join(p,f), rm) for f in os.listdir(p)]
         elif os.path.isfile(p):
             self.data = [ls(p)]
 
-        logger.debug("p %s len %i" % (p, len(self.data)) )
+        logger.debug(f"p {p} len {len(self.data)}")
         if len(self.data) > 0:
             self.off = 0
             self.off = self.off + 1

modules/python/dionaea/hpfeeds.py

@@ -267,7 +267,7 @@ def __init__(self, path, config=None):
         try:
             reconnect_timeout = float(reconnect_timeout)
         except (TypeError, ValueError):
-            logger.warning("Unable to convert value '%s' for reconnect timeout to float" % reconnect_timeout)
+            logger.warning(f"Unable to convert value '{reconnect_timeout}' for reconnect timeout to float")
             reconnect_timeout = self.default_reconnect_timeout
 
         port = config.get("port")
@@ -276,7 +276,7 @@ def __init__(self, path, config=None):
         try:
             port = int(port)
         except (TypeError, ValueError):
-            logger.warning("Unable to convert value '%s' for port to int" % port)
+            logger.warning(f"Unable to convert value '{port}' for port to int")
             port = self.default_port
 
         self.client = hpclient(
@@ -341,56 +341,47 @@ def handle_incident(self, i):
     def handle_incident_dionaea_connection_tcp_listen(self, icd):
         self.connection_publish(icd, 'listen')
         con=icd.con
-        logger.info("listen connection on %s:%i" %
-            (con.remote.host, con.remote.port))
+        logger.info(f"listen connection on {con.remote.host}:{con.remote.port}")
 
     def handle_incident_dionaea_connection_tls_listen(self, icd):
         self.connection_publish(icd, 'listen')
         con=icd.con
-        logger.info("listen connection on %s:%i" %
-            (con.remote.host, con.remote.port))
+        logger.info(f"listen connection on {con.remote.host}:{con.remote.port}")
 
     def handle_incident_dionaea_connection_tcp_connect(self, icd):
         self.connection_publish(icd, 'connect')
         con=icd.con
-        logger.info("connect connection to %s/%s:%i from %s:%i" %
-            (con.remote.host, con.remote.hostname, con.remote.port, self._ownip(icd), con.local.port))
+        logger.info(f"connect connection to {con.remote.host}/{con.remote.hostname}:{con.remote.port} from {self._ownip(icd)}:{con.local.port}")
 
     def handle_incident_dionaea_connection_tls_connect(self, icd):
         self.connection_publish(icd, 'connect')
         con=icd.con
-        logger.info("connect connection to %s/%s:%i from %s:%i" %
-            (con.remote.host, con.remote.hostname, con.remote.port, self._ownip(icd), con.local.port))
+        logger.info(f"connect connection to {con.remote.host}/{con.remote.hostname}:{con.remote.port} from {self._ownip(icd)}:{con.local.port}")
 
     def handle_incident_dionaea_connection_udp_connect(self, icd):
         self.connection_publish(icd, 'connect')
         con=icd.con
-        logger.info("connect connection to %s/%s:%i from %s:%i" %
-            (con.remote.host, con.remote.hostname, con.remote.port, self._ownip(icd), con.local.port))
+        logger.info(f"connect connection to {con.remote.host}/{con.remote.hostname}:{con.remote.port} from {self._ownip(icd)}:{con.local.port}")
 
     def handle_incident_dionaea_connection_tcp_accept(self, icd):
         self.connection_publish(icd, 'accept')
         con=icd.con
-        logger.info("accepted connection from  %s:%i to %s:%i" %
-            (con.remote.host, con.remote.port, self._ownip(icd), con.local.port))
+        logger.info(f"accepted connection from  {con.remote.host}:{con.remote.port} to {self._ownip(icd)}:{con.local.port}")
 
     def handle_incident_dionaea_connection_tls_accept(self, icd):
         self.connection_publish(icd, 'accept')
         con=icd.con
-        logger.info("accepted connection from %s:%i to %s:%i" %
-            (con.remote.host, con.remote.port, self._ownip(icd), con.local.port))
+        logger.info(f"accepted connection from {con.remote.host}:{con.remote.port} to {self._ownip(icd)}:{con.local.port}")
 
     def handle_incident_dionaea_connection_tcp_reject(self, icd):
         self.connection_publish(icd, 'reject')
         con=icd.con
-        logger.info("reject connection from %s:%i to %s:%i" %
-            (con.remote.host, con.remote.port, self._ownip(icd), con.local.port))
+        logger.info(f"reject connection from {con.remote.host}:{con.remote.port} to {self._ownip(icd)}:{con.local.port}")
 
     def handle_incident_dionaea_connection_tcp_pending(self, icd):
         self.connection_publish(icd, 'pending')
         con=icd.con
-        logger.info("pending connection from %s:%i to %s:%i" %
-            (con.remote.host, con.remote.port, self._ownip(icd), con.local.port))
+        logger.info(f"pending connection from {con.remote.host}:{con.remote.port} to {self._ownip(icd)}:{con.local.port}")
 
     def handle_incident_dionaea_download_complete_unique(self, i):
         self.handle_incident_dionaea_download_complete_again(i)