Fix related to the #6171 (WebSocket Host port)

Author: stamparm

lib/core/common.py

@@ -5140,7 +5140,7 @@ def getHostHeader(url):
     if url:
         parsed = _urllib.parse.urlparse(url)
         retVal = parsed.netloc
-        defaultPort = {"http": 80, "https": 443}.get(parsed.scheme.lower())
+        defaultPort = {"http": 80, "https": 443, "ws": 80, "wss": 443}.get(parsed.scheme.lower())
 
         # Keep non-default ports, including 80 on HTTPS and 443 on HTTP.
         if defaultPort is not None and retVal.endswith(':%d' % defaultPort):

lib/core/settings.py

@@ -19,7 +19,7 @@
 from thirdparty import six
 
 # sqlmap version (<major>.<minor>.<month>.<monthly commit>)
-VERSION = "1.10.11.2"
+VERSION = "1.10.11.3"
 TYPE = "dev" if VERSION.count('.') > 2 and VERSION.split('.')[-1] != '0' else "stable"
 TYPE_COLORS = {"dev": 33, "stable": 90, "pip": 34}
 VERSION_STRING = "sqlmap/%s#%s" % ('.'.join(VERSION.split('.')[:-1]) if VERSION.count('.') > 2 and VERSION.split('.')[-1] == '0' else VERSION, TYPE)

lib/request/websocket.py

@@ -73,7 +73,7 @@ def connect(self, url, header=None, cookie=None):
         self.sock.settimeout(self._timeout)
 
         hostport = "[%s]" % host if ":" in host else host       # bracket IPv6 literals
-        if port not in (80, 443):
+        if port != (443 if secure else 80):    # Note: only the scheme's own default port gets omitted
             hostport = "%s:%d" % (hostport, port)
 
         key = getText(base64.b64encode(os.urandom(16)))

tests/test_websocket.py

@@ -133,6 +133,35 @@ def recv(self, n):
         ws = _client(); ws.sock = _RaisingSock(_ssl.SSLError("decryption failed"))
         self.assertRaises(_ssl.SSLError, ws.recv)
 
+    def test_handshake_host_port(self):
+        # Note: only the scheme's own default port is omitted (port 443 used to be dropped for 'ws://' too)
+        import lib.request.websocket as W
+
+        class _Sent(Exception):
+            pass
+
+        class _Capture(_FakeSock):
+            def settimeout(self, value):
+                pass
+
+            def sendall(self, data):
+                raise _Sent(data)
+
+        original = W.socket.create_connection
+        try:
+            W.socket.create_connection = lambda *args, **kwargs: _Capture()
+            for url, host in (("ws://example.com:443/x", "example.com:443"), ("ws://example.com:80/x", "example.com"), ("ws://example.com:8080/x", "example.com:8080")):
+                ws = WebSocket.__new__(WebSocket)
+                ws._timeout = None
+                try:
+                    ws.connect(url)
+                except _Sent as ex:
+                    self.assertIn(("Host: %s\r\n" % host).encode(), ex.args[0])
+                else:
+                    self.fail("handshake was not sent")
+        finally:
+            W.socket.create_connection = original
+
 
 if __name__ == "__main__":
     unittest.main()