Sourcery Starbot ⭐ refactored gosirys/sqlmap

extra/beep/beep.py

@@ -38,7 +38,7 @@ def _speaker_beep():
 
 # Reference: https://lists.gnu.org/archive/html/emacs-devel/2014-09/msg00815.html
 def _cygwin_beep(filename):
-    os.system("play-sound-file '%s' 2>/dev/null" % filename)
+    os.system(f"play-sound-file '{filename}' 2>/dev/null")
 
 def _mac_beep():
     import Carbon.Snd
@@ -51,7 +51,7 @@ def _win_wav_play(filename):
 
 def _linux_wav_play(filename):
     for _ in ("aplay", "paplay", "play"):
-        if not os.system("%s '%s' 2>/dev/null" % (_, filename)):
+        if not os.system(f"{_} '{filename}' 2>/dev/null"):
             return
 
     import ctypes
@@ -79,7 +79,9 @@ class struct_pa_sample_spec(ctypes.Structure):
 
     pa_stream = pa.pa_simple_new(None, filename, PA_STREAM_PLAYBACK, None, "playback", ctypes.byref(pa_sample_spec), None, None, ctypes.byref(error))
     if not pa_stream:
-        raise Exception("Could not create pulse audio stream: %s" % pa.strerror(ctypes.byref(error)))
+        raise Exception(
+            f"Could not create pulse audio stream: {pa.strerror(ctypes.byref(error))}"
+        )
 
     while True:
         latency = pa.pa_simple_get_latency(pa_stream, ctypes.byref(error))

extra/cloak/cloak.py

@@ -49,7 +49,7 @@ def decloak(inputFile=None, data=None):
     return data
 
 def main():
-    usage = '%s [-d] -i <input file> [-o <output file>]' % sys.argv[0]
+    usage = f'{sys.argv[0]} [-d] -i <input file> [-o <output file>]'
     parser = OptionParser(usage=usage, version='0.2')
 
     try:
@@ -69,20 +69,11 @@ def main():
         print('ERROR: the provided input file \'%s\' is non existent' % args.inputFile)
         sys.exit(1)
 
-    if not args.decrypt:
-        data = cloak(args.inputFile)
-    else:
-        data = decloak(args.inputFile)
-
+    data = cloak(args.inputFile) if not args.decrypt else decloak(args.inputFile)
     if not args.outputFile:
-        if not args.decrypt:
-            args.outputFile = args.inputFile + '_'
-        else:
-            args.outputFile = args.inputFile[:-1]
-
-    f = open(args.outputFile, 'wb')
-    f.write(data)
-    f.close()
+        args.outputFile = args.inputFile[:-1] if args.decrypt else f'{args.inputFile}_'
+    with open(args.outputFile, 'wb') as f:
+        f.write(data)
 
 if __name__ == '__main__':
     main()

extra/dbgtool/dbgtool.py

@@ -20,19 +20,18 @@ def convert(inputFile):
     fileSize = fileStat.st_size
 
     if fileSize > 65280:
-        print("ERROR: the provided input file '%s' is too big for debug.exe" % inputFile)
+        print(f"ERROR: the provided input file '{inputFile}' is too big for debug.exe")
         sys.exit(1)
 
     script = "n %s\nr cx\n" % os.path.basename(inputFile.replace(".", "_"))
     script += "%x\nf 0100 ffff 00\n" % fileSize
     scrString = ""
-    counter = 256
     counter2 = 0
 
     fp = open(inputFile, "rb")
     fileContent = fp.read()
 
-    for fileChar in fileContent:
+    for counter, fileChar in enumerate(fileContent, start=256):
         unsignedFileChar = fileChar if sys.version_info >= (3, 0) else ord(fileChar)
 
         if unsignedFileChar != 0:
@@ -47,8 +46,6 @@ def convert(inputFile):
             scrString = ""
             counter2 = 0
 
-        counter += 1
-
         if counter2 == 20:
             script += "%s\n" % scrString
             scrString = ""
@@ -60,7 +57,7 @@ def convert(inputFile):
 
 def main(inputFile, outputFile):
     if not os.path.isfile(inputFile):
-        print("ERROR: the provided input file '%s' is not a regular file" % inputFile)
+        print(f"ERROR: the provided input file '{inputFile}' is not a regular file")
         sys.exit(1)
 
     script = convert(inputFile)
@@ -74,7 +71,7 @@ def main(inputFile, outputFile):
         print(script)
 
 if __name__ == "__main__":
-    usage = "%s -i <input file> [-o <output file>]" % sys.argv[0]
+    usage = f"{sys.argv[0]} -i <input file> [-o <output file>]"
     parser = OptionParser(usage=usage, version="0.1")
 
     try:

extra/icmpsh/icmpsh_m.py

@@ -83,7 +83,7 @@ def main(src, dst):
             if sock in select.select([sock], [], [])[0]:
                 buff = sock.recv(4096)
 
-                if 0 == len(buff):
+                if len(buff) == 0:
                     # Socket remotely closed
                     sock.close()
                     sys.exit(0)
@@ -93,7 +93,11 @@ def main(src, dst):
                 icmppacket = ippacket.child()
 
                 # If the packet matches, report it to the user
-                if ippacket.get_ip_dst() == src and ippacket.get_ip_src() == dst and 8 == icmppacket.get_icmp_type():
+                if (
+                    ippacket.get_ip_dst() == src
+                    and ippacket.get_ip_src() == dst
+                    and icmppacket.get_icmp_type() == 8
+                ):
                     # Get identifier and sequence number
                     ident = icmppacket.get_icmp_id()
                     seq_id = icmppacket.get_icmp_seq()
@@ -136,8 +140,10 @@ def main(src, dst):
 
 if __name__ == '__main__':
     if len(sys.argv) < 3:
-        msg = 'missing mandatory options. Execute as root:\n'
-        msg += './icmpsh-m.py <source IP address> <destination IP address>\n'
+        msg = (
+            'missing mandatory options. Execute as root:\n'
+            + './icmpsh-m.py <source IP address> <destination IP address>\n'
+        )
         sys.stderr.write(msg)
         sys.exit(1)
 

extra/shutils/duplicates.py

@@ -11,7 +11,7 @@
 
 if __name__ == "__main__":
     if len(sys.argv) > 1:
-        items = list()
+        items = []
 
         with open(sys.argv[1], 'r') as f:
             for item in f:

extra/shutils/newlines.py

@@ -21,7 +21,7 @@ def check(filepath):
         print("no directory specified, defaulting to current working directory")
         BASE_DIRECTORY = os.getcwd()
 
-    print("looking for *.py scripts in subdirectories of '%s'" % BASE_DIRECTORY)
+    print(f"looking for *.py scripts in subdirectories of '{BASE_DIRECTORY}'")
     for root, dirs, files in os.walk(BASE_DIRECTORY):
         if any(_ in root for _ in ("extra", "thirdparty")):
             continue

extra/vulnserver/vulnserver.py

@@ -110,7 +110,19 @@ def do_REQUEST(self):
             if self.data.startswith('{') and self.data.endswith('}'):
                 params.update(json.loads(self.data))
             elif self.data.startswith('<') and self.data.endswith('>'):
-                params.update(dict((_[0], _[1].replace("&apos;", "'").replace("&quot;", '"').replace("&lt;", '<').replace("&gt;", '>').replace("&amp;", '&')) for _ in re.findall(r'name="([^"]+)" value="([^"]*)"', self.data)))
+                params.update(
+                    {
+                        _[0]: _[1]
+                        .replace("&apos;", "'")
+                        .replace("&quot;", '"')
+                        .replace("&lt;", '<')
+                        .replace("&gt;", '>')
+                        .replace("&amp;", '&')
+                        for _ in re.findall(
+                            r'name="([^"]+)" value="([^"]*)"', self.data
+                        )
+                    }
+                )
             else:
                 self.data = self.data.replace(';', '&')     # Note: seems that Python3 started ignoring parameter splitting with ';'
                 params.update(parse_qs(self.data))
@@ -125,16 +137,16 @@ def do_REQUEST(self):
                     name, value = part.split('=', 1)
                     params[name.strip()] = unquote_plus(value.strip())
 
-        for key in params:
-            if params[key] and isinstance(params[key], (tuple, list)):
+        for key, value_ in params.items():
+            if value_ and isinstance(params[key], (tuple, list)):
                 params[key] = params[key][-1]
 
         self.url, self.params = path, params
 
         if self.url == '/':
-            if not any(_ in self.params for _ in ("id", "query")):
+            if all(_ not in self.params for _ in ("id", "query")):
                 self.send_response(OK)
-                self.send_header("Content-type", "text/html; charset=%s" % UNICODE_ENCODING)
+                self.send_header("Content-type", f"text/html; charset={UNICODE_ENCODING}")
                 self.send_header("Connection", "close")
                 self.end_headers()
                 self.wfile.write(b"<!DOCTYPE html><html><head><title>vulnserver</title></head><body><h3>GET:</h3><a href='/?id=1'>link</a><hr><h3>POST:</h3><form method='post'>ID: <input type='text' name='id'><input type='submit' value='Submit'></form></body></html>")
@@ -143,44 +155,49 @@ def do_REQUEST(self):
 
                 try:
                     if self.params.get("echo", ""):
-                        output += "%s<br>" % self.params["echo"]
+                        output += f'{self.params["echo"]}<br>'
 
                     if self.params.get("reflect", ""):
-                        output += "%s<br>" % self.params.get("id")
+                        output += f'{self.params.get("id")}<br>'
 
                     with _lock:
                         if "query" in self.params:
                             _cursor.execute(self.params["query"])
                         elif "id" in self.params:
                             if "base64" in self.params:
-                                _cursor.execute("SELECT * FROM users WHERE id=%s LIMIT 0, 1" % base64.b64decode("%s===" % self.params["id"], altchars=self.params.get("altchars")).decode())
+                                _cursor.execute(
+                                    "SELECT * FROM users WHERE id=%s LIMIT 0, 1"
+                                    % base64.b64decode(
+                                        f'{self.params["id"]}===',
+                                        altchars=self.params.get("altchars"),
+                                    ).decode()
+                                )
                             else:
-                                _cursor.execute("SELECT * FROM users WHERE id=%s LIMIT 0, 1" % self.params["id"])
+                                _cursor.execute(f'SELECT * FROM users WHERE id={self.params["id"]} LIMIT 0, 1')
                         results = _cursor.fetchall()
 
                     output += "<b>SQL results:</b><br>\n"
 
                     if self.params.get("code", ""):
                         if not results:
                             code = INTERNAL_SERVER_ERROR
-                    else:
-                        if results:
-                            output += "<table border=\"1\">\n"
+                    elif results:
+                        output += "<table border=\"1\">\n"
 
-                            for row in results:
-                                output += "<tr>"
-                                for value in row:
-                                    output += "<td>%s</td>" % value
-                                output += "</tr>\n"
+                        for row in results:
+                            output += "<tr>"
+                            for value in row:
+                                output += f"<td>{value}</td>"
+                            output += "</tr>\n"
 
-                            output += "</table>\n"
-                        else:
-                            output += "no results found"
+                        output += "</table>\n"
+                    else:
+                        output += "no results found"
 
                     output += "</body></html>"
                 except Exception as ex:
                     code = INTERNAL_SERVER_ERROR
-                    output = "%s: %s" % (re.search(r"'([^']+)'", str(type(ex))).group(1), ex)
+                    output = f"""{re.search("'([^']+)'", str(type(ex))).group(1)}: {ex}"""
 
                 self.send_response(code)
 
@@ -208,8 +225,7 @@ def do_HEAD(self):
         self.do_REQUEST()
 
     def do_POST(self):
-        length = int(self.headers.get("Content-length", 0))
-        if length:
+        if length := int(self.headers.get("Content-length", 0)):
             data = self.rfile.read(length)
             data = unquote_plus(data.decode(UNICODE_ENCODING, "ignore"))
             self.data = data
@@ -221,12 +237,11 @@ def do_POST(self):
                 line += self.rfile.read(1)
                 if line.endswith(b'\n'):
                     if count % 2 == 1:
-                        current = line.rstrip(b"\r\n")
-                        if not current:
-                            break
-                        else:
+                        if current := line.rstrip(b"\r\n"):
                             data += current
 
+                        else:
+                            break
                     count += 1
                     line = b""
 

lib/controller/action.py

@@ -35,20 +35,21 @@ def action():
     if not Backend.getDbms() or not conf.dbmsHandler:
         htmlParsed = Format.getErrorParsedDBMSes()
 
-        errMsg = "sqlmap was not able to fingerprint the "
-        errMsg += "back-end database management system"
-
+        errMsg = (
+            "sqlmap was not able to fingerprint the "
+            + "back-end database management system"
+        )
         if htmlParsed:
             errMsg += ", but from the HTML error page it was "
             errMsg += "possible to determinate that the "
-            errMsg += "back-end DBMS is %s" % htmlParsed
+            errMsg += f"back-end DBMS is {htmlParsed}"
 
         if htmlParsed and htmlParsed.lower() in SUPPORTED_DBMS:
             errMsg += ". Do not specify the back-end DBMS manually, "
             errMsg += "sqlmap will fingerprint the DBMS for you"
         elif kb.nullConnection:
             errMsg += ". You can try to rerun without using optimization "
-            errMsg += "switch '%s'" % ("-o" if conf.optimize else "--null-connection")
+            errMsg += f"""switch '{"-o" if conf.optimize else "--null-connection"}'"""
 
         raise SqlmapUnsupportedDBMSException(errMsg)
 
@@ -184,8 +185,7 @@ def action():
 
     if conf.sqlQuery:
         for query in conf.sqlQuery.strip(';').split(';'):
-            query = query.strip()
-            if query:
+            if query := query.strip():
                 conf.dumper.sqlQuery(query, conf.dbmsHandler.sqlQuery(query))
 
     if conf.sqlShell: