add -port to sql_console module

epinna · Jun 18, 2023 · 445bd88 · 445bd88
1 parent d7aa0b9
commit 445bd88
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 12 deletions.
diff --git a/modules/sql/console.py b/modules/sql/console.py
@@ -1,9 +1,11 @@
+import re
+
 import utils
 from core import messages
 from core.loggers import log
 from core.module import Module
 from core.vectors import PhpCode
-import re
+
 
 class Console(Module):
     """Execute SQL query or run console."""
@@ -22,23 +24,23 @@ def init(self):
         self.register_vectors(
             [
                 PhpCode(
-                    """mysqli_report(MYSQLI_REPORT_OFF);if($s=mysqli_connect('${host}','${user}','${passwd}')){$r=mysqli_query($s,'${query}');if($r){$f=mysqli_fetch_fields($r);foreach($f as $v){echo $v->name.'${linsep}';};echo '${colsep}';while($c=mysqli_fetch_row($r)){echo implode('${linsep}',$c);echo '${linsep}${colsep}';}};echo @mysqli_error($s);@mysqli_close($s);}echo '${errsep}'.@mysqli_connect_error();""",
+                    """mysqli_report(MYSQLI_REPORT_OFF);if($s=mysqli_connect('${host}:${port}','${user}','${passwd}')){$r=mysqli_query($s,'${query}');if($r){$f=mysqli_fetch_fields($r);foreach($f as $v){echo $v->name.'${linsep}';};echo '${colsep}';while($c=mysqli_fetch_row($r)){echo implode('${linsep}',$c);echo '${linsep}${colsep}';}};echo @mysqli_error($s);@mysqli_close($s);}echo '${errsep}'.@mysqli_connect_error();""",
                     name='mysql',
                 ),
                 PhpCode(
-                    """mysqli_report(MYSQLI_REPORT_OFF);if($s=mysqli_connect('${host}','${user}','${passwd}','${database}')){$r=mysqli_query($s,'${query}');if($r){$f=mysqli_fetch_fields($r);foreach($f as $v){echo $v->name.'${linsep}';};echo '${colsep}';while($c=mysqli_fetch_row($r)){echo implode('${linsep}',$c);echo '${linsep}${colsep}';}};echo @mysqli_error($s);@mysqli_close($s);}echo '${errsep}'.@mysqli_connect_error();""",
+                    """mysqli_report(MYSQLI_REPORT_OFF);if($s=mysqli_connect('${host}:${port}','${user}','${passwd}','${database}')){$r=mysqli_query($s,'${query}');if($r){$f=mysqli_fetch_fields($r);foreach($f as $v){echo $v->name.'${linsep}';};echo '${colsep}';while($c=mysqli_fetch_row($r)){echo implode('${linsep}',$c);echo '${linsep}${colsep}';}};echo @mysqli_error($s);@mysqli_close($s);}echo '${errsep}'.@mysqli_connect_error();""",
                     name='mysql_database',
                 ),
                 PhpCode(
                     """mysqli_report(MYSQLI_REPORT_OFF);$r=mysqli_query('${query}');if($r){while($c=mysqli_fetch_row($r)){foreach($c as $key=>$value){echo $value.'${linsep}';}echo '${colsep}';}};mysqli_close();echo '${errsep}'.@mysqli_connect_error().' '.@mysqli_error();""",
                     name="mysql_fallback"
                 ),
                 PhpCode(
-                    """if(pg_connect('host=${host} user=${user} password=${passwd}')){$r=pg_query('${query}');if($r){while($c=pg_fetch_row($r)){foreach($c as $key=>$value){echo $value.'${linsep}';}echo '${colsep}';}};pg_close();}echo '${errsep}'.@pg_last_error();""",
+                    """if(pg_connect('host=${host} port=${port} user=${user} password=${passwd}')){$r=pg_query('${query}');if($r){while($c=pg_fetch_row($r)){foreach($c as $key=>$value){echo $value.'${linsep}';}echo '${colsep}';}};pg_close();}echo '${errsep}'.@pg_last_error();""",
                     name="pgsql"
                 ),
                 PhpCode(
-                    """if(pg_connect('host=${host} user=${user} dbname=${database} password=${passwd}')){$r=pg_query('${query}');if($r){while($c=pg_fetch_row($r)){foreach($c as $key=>$value){echo $value.'${linsep}';}echo '${colsep}';}};pg_close();}echo '${errsep}'.@pg_last_error();""",
+                    """if(pg_connect('host=${host} port=${port} user=${user} dbname=${database} password=${passwd}')){$r=pg_query('${query}');if($r){while($c=pg_fetch_row($r)){foreach($c as $key=>$value){echo $value.'${linsep}';}echo '${colsep}';}};pg_close();}echo '${errsep}'.@pg_last_error();""",
                     name="pgsql_database"
                 ),
                 PhpCode(
@@ -54,6 +56,7 @@ def init(self):
             {'name': '-host', 'help': 'Db host (default: localhost)', 'nargs': '?', 'default': 'localhost'},
             {'name': '-dbms', 'help': 'Db type', 'choices': ('mysql', 'pgsql'), 'default': 'mysql'},
             {'name': '-database', 'help': 'Database name'},
+            {'name': '-port', 'help': 'Port number', 'type': int, 'default': 0},
             {'name': '-query', 'help': 'Execute a single query'},
             {'name': '-encoding', 'help': 'Db text encoding', 'default': 'utf-8'},
         ])
@@ -72,6 +75,10 @@ def _query(self, vector, args):
         # Escape ' in query strings
         self.args['query'] = self.args['query'].replace('\\', '\\\\').replace('\'', '\\\'')
 
+        # Set default port depending on selected dbms
+        if self.args['port'] <= 0:
+            self.args['port'] = '5432' if self.args['dbms'] == 'pgsql' else '3306'
+
         result = self.vectors.get_result(vector, args)
 
         # we wan't the result to be unicode, but depending on the source

diff --git a/tests/test_sql_console.py b/tests/test_sql_console.py
@@ -1,12 +1,14 @@
-from tests.base_test import BaseTest
+import subprocess
+import unittest
+
 from testfixtures import log_capture
+
+from core import messages
 from core import modules
 from core.sessions import SessionURL
-from core import messages
 from tests import config
-import unittest
-import subprocess
-import os
+from tests.base_test import BaseTest
+
 
 def setUpModule():
     try:
@@ -45,14 +47,18 @@ def test_wrongcommand(self, log_captured):
                                     messages.module_sql_console.check_credentials),
                          log_captured.records[-2].msg)
 
-
     def test_wronglogin(self):
-
         wrong_login = '-user bogus -passwd bogus -query "select \'A\';"'
 
         # Using run_cmdline to test the outputs
         self.assertIn('Access denied for user', self.run_cmdline(wrong_login)['error'])
 
+    def test_wrong_port(self):
+        wrong_port = ['-user', config.sql_user, '-passwd', config.sql_passwd, '-port', '1234', '-query', 'select 1234;']
+
+        # Using run_cmdline to test the outputs
+        self.assertIn('Cannot assign requested address', self.run_argv(wrong_port)['error'])
+
     def test_login(self):
 
         login = ['-user', config.sql_user, '-passwd', config.sql_passwd ]