diff --git a/data/xml/queries.xml b/data/xml/queries.xml
index c6e1d6345bb..37b6680730c 100644
--- a/data/xml/queries.xml
+++ b/data/xml/queries.xml
@@ -598,11 +598,11 @@
-
+
-
+
diff --git a/extra/dbwire/README.md b/extra/dbwire/README.md
index cfd37dceba3..2e06e3415de 100644
--- a/extra/dbwire/README.md
+++ b/extra/dbwire/README.md
@@ -22,7 +22,8 @@ A wire protocol is shared across a whole family of products, so one client serve
|-----------------|---------------------|---------|
| `postgres.py` | PostgreSQL v3 | PostgreSQL, CockroachDB, CrateDB, Redshift, Greenplum, Vertica |
| `mysql.py` | MySQL client/server | MySQL, MariaDB, TiDB, Aurora (MySQL), Percona |
-| `tds.py` | TDS | Microsoft SQL Server, Sybase |
+| `tds.py` | TDS 7.x | Microsoft SQL Server |
+| `sybase.py` | TDS 5.0 | Sybase / SAP ASE |
| `firebird.py` | Firebird wire | Firebird 3 / 4 / 5 |
| `cubrid.py` | CUBRID CAS | CUBRID |
| `clickhouse.py` | HTTP (TabSeparated) | ClickHouse and HTTP-compatible forks |
@@ -57,7 +58,12 @@ parameter binding, prepared statements, bulk load/`COPY`, or TLS. Notable per-pr
`caching_sha2_password` authentication over a plaintext connection requires RSA/TLS and is not
supported; use a `mysql_native_password` account for the dependency-free path.
- **TDS** - cleartext login only. Servers that force encryption (for example Azure SQL Database)
- require TLS and are not supported here; the native driver or SQLAlchemy tier covers those.
+ require TLS and are not supported here; the native driver or SQLAlchemy tier covers those. LOGIN7 is the
+ Microsoft dialect (TDS 7.x) and does not reach Sybase - see the next bullet.
+- **TDS 5.0 (Sybase)** - cleartext LOGINREC login. The client declares no optional capabilities, so the
+ server converts anything exotic (`date`/`time`/`bigdatetime`, wide formats) down to the baseline types
+ before sending it. Packet size is whatever the login negotiates. Sends and expects UTF-8, so a server
+ running any character set is decoded correctly.
- **Firebird** - SRP-256 (and SRP) authentication with ChaCha20 or RC4 wire encryption, as required by
default on Firebird 3 and later. Legacy (pre-SRP) authentication is not implemented.
- **CUBRID** - cleartext login over the CAS broker protocol. Large objects (BLOB/CLOB) are returned as
diff --git a/extra/dbwire/__init__.py b/extra/dbwire/__init__.py
index e809c201ddb..b064a9e652a 100644
--- a/extra/dbwire/__init__.py
+++ b/extra/dbwire/__init__.py
@@ -11,10 +11,13 @@
Design note: connectors speak a *wire protocol*, not a product, so a single client covers the whole
compatible family - e.g. the PostgreSQL client also serves CockroachDB, CrateDB, Redshift and Greenplum;
-a MySQL client serves MariaDB/TiDB/Aurora; a TDS client serves MSSQL/Sybase. Each module exposes a small
-PEP 249 (DB-API 2.0) subset (connect(), Connection.cursor()/commit()/close(), Cursor.execute()/fetchall()).
+a MySQL client serves MariaDB/TiDB/Aurora. Where a family split the protocol, so does the client: tds.py
+speaks Microsoft's TDS 7.x and sybase.py the TDS 5.0 that ASE kept. Each module exposes a small PEP 249
+(DB-API 2.0) subset (connect(), Connection.cursor()/commit()/close(), Cursor.execute()/fetchall()).
"""
+import socket
+
__version__ = "0.1"
apilevel = "2.0"
@@ -71,6 +74,21 @@ def connection_lost(ex):
return OperationalError("connection lost (%s)" % ex)
+def handshake_done(sock):
+ """
+ Drop the connect deadline, once the login exchange is over.
+
+ connect_timeout has to stay armed THROUGH the handshake, not just the TCP connect: a peer that
+ accepts the connection and then says nothing (a wrong port, a silent proxy, a dropping firewall) is
+ perfectly alive as far as keepalive() below is concerned, so an unbounded login read waits forever.
+ A query is the opposite case - see keepalive().
+ """
+
+ try:
+ sock.settimeout(None)
+ except Exception:
+ pass
+
def keepalive(sock):
"""
Ask the kernel to probe an idle connection, so a peer that dies without a FIN is eventually detected.
@@ -80,12 +98,29 @@ def keepalive(sock):
failure being guarded against. Best-effort - the options are not portable everywhere.
"""
- import socket as _socket
-
try:
- sock.setsockopt(_socket.SOL_SOCKET, _socket.SO_KEEPALIVE, 1)
+ sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
for name, value in (("TCP_KEEPIDLE", 60), ("TCP_KEEPINTVL", 10), ("TCP_KEEPCNT", 5)):
- if hasattr(_socket, name):
- sock.setsockopt(_socket.IPPROTO_TCP, getattr(_socket, name), value)
+ if hasattr(socket, name):
+ sock.setsockopt(socket.IPPROTO_TCP, getattr(socket, name), value)
except Exception:
pass
+
+def recvn(sock, n):
+ """
+ Read exactly n bytes off `sock`, or raise - every wire protocol here is framed, so a short read is a
+ desynchronized stream, not a smaller message.
+
+ Shared because it was five identical copies: a fix to the recv loop has to land once, not per module.
+ """
+
+ buf = b""
+ while len(buf) < n:
+ try:
+ chunk = sock.recv(n - len(buf))
+ except (socket.error, OSError) as ex:
+ raise connection_lost(ex)
+ if not chunk:
+ raise InterfaceError("connection closed by server")
+ buf += chunk
+ return buf
diff --git a/extra/dbwire/cubrid.py b/extra/dbwire/cubrid.py
index 750d7578b59..aea26d480a8 100644
--- a/extra/dbwire/cubrid.py
+++ b/extra/dbwire/cubrid.py
@@ -24,7 +24,9 @@
from extra.dbwire import NotSupportedError
from extra.dbwire import OperationalError
from extra.dbwire import connection_lost
+from extra.dbwire import handshake_done
from extra.dbwire import keepalive
+from extra.dbwire import recvn
from extra.dbwire import ProgrammingError
_MAGIC = b"CUBRK"
@@ -132,6 +134,10 @@ def double(self):
v = struct.unpack_from(">d", self._buf, self._off)[0]; self._off += 8; return v
def raw(self, n):
+ # a slice past the end returns short data *silently*, which would hand a truncated value to the
+ # caller (the fixed-width readers above raise struct.error instead)
+ if n < 0 or n > self.remaining():
+ raise InterfaceError("CAS response too short (wanted %d of %d bytes)" % (n, self.remaining()))
v = self._buf[self._off:self._off + n]; self._off += n; return bytes(v)
def skip(self, n):
@@ -257,23 +263,13 @@ def _safe_close(self):
self._sock = None
def _recvn(self, n):
- buf = b""
- while len(buf) < n:
- try:
- chunk = self._sock.recv(n - len(buf))
- except (socket.error, OSError) as ex:
- raise connection_lost(ex)
- if not chunk:
- raise InterfaceError("connection closed by server")
- buf += chunk
- return buf
+ return recvn(self._sock, n)
def _open(self):
# broker handshake (may redirect to a dedicated CAS worker port), then cleartext OPEN_DATABASE login
try:
sock = socket.create_connection((self._host, self._port), timeout=self._timeout)
keepalive(sock)
- sock.settimeout(None)
sock.sendall(_MAGIC + struct.pack(">BB", _CLIENT_JDBC, _CAS_VERSION) + b"\x00\x00\x00")
self._sock = sock
(port,) = struct.unpack(">i", self._recvn(4))
@@ -283,21 +279,25 @@ def _open(self):
self._safe_close()
sock = socket.create_connection((self._host, port), timeout=self._timeout)
keepalive(sock)
- sock.settimeout(None)
self._sock = sock
+
+ login = self._fixed(self._database, 32) + self._fixed(self._user, 32) + self._fixed(self._password, 32)
+ login += b"\x00" * 532 # 512 extended-info + 20 reserved
+ self._sock.sendall(login)
+ reader = self._read_response()
+ reader.int() # response_code (>=0; errors already raised in _read_response)
+ broker = reader.raw(8)
+ self._protocol_version = bytearray(broker)[4] & 0x3f
+ # enable auto-commit so each statement is independent (avoids the CAS keep-connection handshake dance)
+ self._call(_Writer(_FC_SET_DB_PARAMETER).arg_int(_PARAM_AUTO_COMMIT).arg_int(1))
except (socket.error, socket.timeout) as ex:
self._safe_close()
raise OperationalError("could not connect to '%s:%s' (%s)" % (self._host, self._port, ex))
+ except Exception: # a rejected login (or connection_lost() out of _recvn) is still ours to close
+ self._safe_close()
+ raise
- login = self._fixed(self._database, 32) + self._fixed(self._user, 32) + self._fixed(self._password, 32)
- login += b"\x00" * 532 # 512 extended-info + 20 reserved
- self._sock.sendall(login)
- reader = self._read_response()
- reader.int() # response_code (>=0; errors already raised in _read_response)
- broker = reader.raw(8)
- self._protocol_version = bytearray(broker)[4] & 0x3f
- # enable auto-commit so each statement is independent (avoids the CAS keep-connection handshake dance)
- self._call(_Writer(_FC_SET_DB_PARAMETER).arg_int(_PARAM_AUTO_COMMIT).arg_int(1))
+ handshake_done(self._sock)
@staticmethod
def _fixed(value, length):
@@ -343,9 +343,9 @@ def _raise(errno, message):
text = message.lower()
if any(k in text for k in ("unique", "duplicate", "foreign key", "constraint violat")):
raise IntegrityError(message)
- if any(k in text for k in ("syntax", "unknown class", "does not exist", "not found", "before ' '")):
- raise ProgrammingError(message)
- if any(k in text for k in ("cast", "conversion", "overflow", "truncat")):
+ if any(k in text for k in ("syntax", "unknown class", "does not exist", "not found", "before '")):
+ raise ProgrammingError(message) # CUBRID points at the offending token as: before ' ,'y')'
+ if any(k in text for k in ("cast", "coerce", "conversion", "overflow", "truncat")):
raise DataError(message)
raise ProgrammingError(message)
@@ -371,9 +371,15 @@ def _execute(self, handle, reader):
reader.byte() # is_updatable
columns = self._parse_columns(reader, reader.int())
+ # args: handle, flag, max_col_size, max_row, binds, fetch_flag, auto_commit, forward_only_cursor,
+ # cache_time, query_timeout. auto_commit makes the CAS worker commit the statement and end the
+ # transaction - without it DML is rolled back when the connection drops, whatever _open() asked
+ # SET_DB_PARAMETER for. It must stay off for a SELECT: ending the transaction there invalidates the
+ # request handle the paged _fetch_remaining() still reads from.
+ select = stmt_type == _STMT_SELECT
exec_writer = (_Writer(_FC_EXECUTE).arg_int(handle).arg_byte(0).arg_int(0).arg_int(0)
- .arg_null().arg_byte(1 if stmt_type == _STMT_SELECT else 0)
- .arg_byte(0).arg_byte(1).arg_cache_time().arg_int(0))
+ .arg_null().arg_byte(1 if select else 0)
+ .arg_byte(0 if select else 1).arg_byte(1).arg_cache_time().arg_int(0))
reader = self._call(exec_writer)
total = reader.int()
diff --git a/extra/dbwire/firebird.py b/extra/dbwire/firebird.py
index a7a56da2e51..b0db9f6e1d2 100644
--- a/extra/dbwire/firebird.py
+++ b/extra/dbwire/firebird.py
@@ -27,7 +27,9 @@
from extra.dbwire import NotSupportedError
from extra.dbwire import OperationalError
from extra.dbwire import connection_lost
+from extra.dbwire import handshake_done
from extra.dbwire import keepalive
+from extra.dbwire import recvn
# operation codes
_op_connect = 1
@@ -142,6 +144,8 @@
_GDS_DATA = frozenset((335544321,))
_GDS_WARNING = 335544434
+_MAX_MESSAGE_LENGTH = 0x40000000 # cap on a wire-supplied length, to bound a hostile/corrupt stream
+
# SRP-6a group used by Firebird (fixed 1024-bit prime, generator 2)
_SRP_N = int("E67D2E994B2F900C3F41F08F5BB2627ED0D49EE1FE767A52EFCD565CD6E768812C3E1E9CE8F0A8BEA6CB13CD29DDE"
"BF7A96D4A93B55D488DF099A15C89DCB0640738EB2CBDD9A8F7BAB561AB1B0DC1C6CDABF303264A08D1BCA932D1F"
@@ -294,18 +298,13 @@ def send(self, data):
raise connection_lost(ex)
def _recv_raw(self, n):
- buf = b""
- while len(buf) < n:
- try:
- chunk = self._sock.recv(n - len(buf))
- except (socket.error, OSError) as ex:
- raise connection_lost(ex)
- if not chunk:
- raise InterfaceError("connection closed by server")
- buf += chunk
- return buf
+ return recvn(self._sock, n)
def recv(self, n, align=False):
+ # every length here comes off the wire (response buffers, status strings, per-value lengths): a
+ # negative one would silently return short data, a huge one would read until memory ran out
+ if n < 0 or n > _MAX_MESSAGE_LENGTH:
+ raise InterfaceError("invalid Firebird length (%d)" % n)
total = n + ((4 - n % 4) % 4) if align else n
data = self._recv_raw(total)
if self._rc:
@@ -324,6 +323,38 @@ def close(self):
except Exception:
pass
+def _parse_response(wire):
+ head = wire.recv(16)
+ handle = struct.unpack("!i", head[:4])[0]
+ object_id = head[4:12]
+ buf = wire.recv(struct.unpack("!i", head[12:16])[0], align=True)
+ _check_status(wire)
+ return handle, object_id, buf
+
+def _check_status(wire):
+ gds, message = set(), ""
+ n = wire.recv_int()
+ while n != _isc_arg_end:
+ if n == _isc_arg_gds:
+ gds_code = wire.recv_int()
+ if gds_code:
+ gds.add(gds_code)
+ elif n == _isc_arg_number:
+ message += " %d" % wire.recv_int()
+ elif n in (_isc_arg_string, _isc_arg_interpreted, _isc_arg_sql_state):
+ s = wire.recv(wire.recv_int(), align=True)
+ if n != _isc_arg_sql_state:
+ message += " " + s.decode("utf-8", "replace")
+ n = wire.recv_int()
+ if gds:
+ message = ("(remote) firebird error %s%s" % (sorted(gds), message)).strip()
+ if gds & _GDS_INTEGRITY:
+ raise IntegrityError(message)
+ if gds & _GDS_DATA:
+ raise DataError(message)
+ if _GDS_WARNING not in gds:
+ raise OperationalError(message)
+
def _pack_int(v):
return struct.pack("!i", v)
@@ -455,39 +486,7 @@ def _response(self):
op = self._wire.recv_int()
if op != _op_response:
raise OperationalError("unexpected Firebird operation %d" % op)
- return self._parse_response()
-
- def _parse_response(self):
- head = self._wire.recv(16)
- handle = struct.unpack("!i", head[:4])[0]
- object_id = head[4:12]
- buf = self._wire.recv(struct.unpack("!i", head[12:16])[0], align=True)
- self._check_status()
- return handle, object_id, buf
-
- def _check_status(self):
- gds, message = set(), ""
- n = self._wire.recv_int()
- while n != _isc_arg_end:
- if n == _isc_arg_gds:
- gds_code = self._wire.recv_int()
- if gds_code:
- gds.add(gds_code)
- elif n == _isc_arg_number:
- message += " %d" % self._wire.recv_int()
- elif n in (_isc_arg_string, _isc_arg_interpreted, _isc_arg_sql_state):
- s = self._wire.recv(self._wire.recv_int(), align=True)
- if n != _isc_arg_sql_state:
- message += " " + s.decode("utf-8", "replace")
- n = self._wire.recv_int()
- if gds:
- message = ("(remote) firebird error %s%s" % (sorted(gds), message)).strip()
- if gds & _GDS_INTEGRITY:
- raise IntegrityError(message)
- if gds & _GDS_DATA:
- raise DataError(message)
- if _GDS_WARNING not in gds:
- raise OperationalError(message)
+ return _parse_response(self._wire)
# ---- query ----
@@ -626,7 +625,7 @@ def _fetch(self, stmt, columns):
op = self._wire.recv_int()
if op != _op_fetch_response:
if op == _op_response:
- self._parse_response()
+ _parse_response(self._wire)
raise OperationalError("unexpected Firebird operation %d during fetch" % op)
status = self._wire.recv_int()
count = self._wire.recv_int()
@@ -759,7 +758,6 @@ def connect(host=None, port=3050, user=None, password=None, database=None, conne
try:
sock = socket.create_connection((host or "localhost", int(port or 3050)), timeout=connect_timeout)
keepalive(sock)
- sock.settimeout(None)
except (socket.error, socket.timeout) as ex:
raise OperationalError("could not connect to '%s:%s' (%s)" % (host, port, ex))
@@ -776,6 +774,7 @@ def connect(host=None, port=3050, user=None, password=None, database=None, conne
_authenticate(wire, user, password, public_key, private_key)
connection = Connection(wire, filename, user, password)
_attach(connection, wire, user)
+ handshake_done(sock)
except (DatabaseError, InterfaceError):
wire.close()
raise
@@ -799,7 +798,7 @@ def _authenticate(wire, user, password, public_key, private_key):
if op == _op_reject:
raise OperationalError("Firebird connection rejected")
if op == _op_response:
- Connection(wire, b"", user, password)._parse_response() # will raise the server error
+ _parse_response(wire) # will raise the server error
raise OperationalError("Firebird connection rejected")
wire.recv(12) # accept block: protocol version / architecture / type (not needed once lazy-send is off)
@@ -852,7 +851,7 @@ def _read_response(wire, user, password):
raise OperationalError("Firebird authentication failed")
if op != _op_response:
raise OperationalError("unexpected Firebird operation %d during login" % op)
- return Connection(wire, b"", user, password)._parse_response()[2]
+ return _parse_response(wire)[2]
def _attach(connection, wire, user):
dpb = bytearray([_isc_dpb_version1])
diff --git a/extra/dbwire/monetdb.py b/extra/dbwire/monetdb.py
index 90dd4451fda..0652811cff1 100644
--- a/extra/dbwire/monetdb.py
+++ b/extra/dbwire/monetdb.py
@@ -23,32 +23,28 @@
from extra.dbwire import NotSupportedError
from extra.dbwire import OperationalError
from extra.dbwire import connection_lost
+from extra.dbwire import handshake_done
from extra.dbwire import keepalive
+from extra.dbwire import recvn
from extra.dbwire import ProgrammingError
_MAX_BLOCK = 0xffff >> 1
-
-def _recvn(sock, n):
- buf = b""
- while len(buf) < n:
- try:
- chunk = sock.recv(n - len(buf))
- except (socket.error, OSError) as ex:
- raise connection_lost(ex)
- if not chunk:
- raise InterfaceError("connection closed by server")
- buf += chunk
- return buf
+_MAX_MESSAGE_LENGTH = 0x40000000 # cap on a (re-assembled) response, to bound a hostile/corrupt stream
def _getblock(sock):
- out = b""
+ # the block length is a 15-bit field, so bounding IT is pointless - a peer that never sets the last-flag
+ # simply streams blocks forever. Bound the accumulated response instead (as the other wire modules do).
+ chunks, total = [], 0
while True:
- (header,) = struct.unpack("> 1, header & 1
- out += _recvn(sock, length)
+ total += length
+ if total > _MAX_MESSAGE_LENGTH:
+ raise InterfaceError("backend message too large (%d bytes)" % total)
+ chunks.append(recvn(sock, length))
if last:
break
- return out.decode("utf-8", "replace")
+ return b"".join(chunks).decode("utf-8", "replace")
def _putblock(sock, text):
data = text.encode("utf-8")
@@ -195,7 +191,6 @@ def connect(host=None, port=50000, user=None, password=None, database=None, conn
try:
sock = socket.create_connection((host, port), timeout=connect_timeout)
keepalive(sock)
- sock.settimeout(None)
except (socket.error, socket.timeout) as ex:
raise OperationalError("could not connect to '%s:%s' (%s)" % (host, port, ex))
@@ -212,7 +207,6 @@ def connect(host=None, port=50000, user=None, password=None, database=None, conn
host, port, database = m.group(1), int(m.group(2)), m.group(3) or database
sock = socket.create_connection((host, port), timeout=connect_timeout)
keepalive(sock)
- sock.settimeout(None)
continue # merovingian proxy redirect: keep reading the next challenge on this socket
if block[0] == "!":
raise OperationalError("(remote) %s" % block[1:].strip())
@@ -238,4 +232,9 @@ def connect(host=None, port=50000, user=None, password=None, database=None, conn
except (socket.error, socket.timeout) as ex:
connection.close()
raise OperationalError("connection error: %s" % ex)
+ except Exception: # e.g. connection_lost() out of _putblock/_getblock: the socket is still ours to close
+ connection.close()
+ raise
+
+ handshake_done(sock)
return connection
diff --git a/extra/dbwire/mysql.py b/extra/dbwire/mysql.py
index b7d119610b9..e6a923aa5df 100644
--- a/extra/dbwire/mysql.py
+++ b/extra/dbwire/mysql.py
@@ -23,7 +23,9 @@
from extra.dbwire import NotSupportedError
from extra.dbwire import OperationalError
from extra.dbwire import connection_lost
+from extra.dbwire import handshake_done
from extra.dbwire import keepalive
+from extra.dbwire import recvn
from extra.dbwire import ProgrammingError
# capability flags
@@ -62,31 +64,19 @@ def _cstring(data, off):
return data[off:], len(data)
return data[off:end], end + 1
-def _recvn(sock, n):
- buf = b""
- while len(buf) < n:
- try:
- chunk = sock.recv(n - len(buf))
- except (socket.error, OSError) as ex:
- raise connection_lost(ex)
- if not chunk:
- raise InterfaceError("connection closed by server")
- buf += chunk
- return buf
-
def _read_packet(sock):
- header = _recvn(sock, 4)
+ header = recvn(sock, 4)
length = struct.unpack(" _MAX_MESSAGE_LENGTH:
raise InterfaceError("backend message too large (%d bytes)" % total)
- payload += _recvn(sock, length)
+ payload += recvn(sock, length)
return seq, payload
def _send_packet(sock, seq, payload):
@@ -295,7 +285,6 @@ def connect(host=None, port=3306, user=None, password=None, database=None, conne
try:
sock = socket.create_connection((host or "localhost", int(port or 3306)), timeout=connect_timeout)
keepalive(sock)
- sock.settimeout(None)
except (socket.error, socket.timeout) as ex:
raise OperationalError("could not connect to '%s:%s' (%s)" % (host, port, ex))
@@ -352,6 +341,7 @@ def connect(host=None, port=3306, user=None, password=None, database=None, conne
_send_packet(sock, seq + 1, response)
_finish_auth(sock, password or "", plugin or "mysql_native_password", salt)
+ handshake_done(sock)
except (DatabaseError, InterfaceError):
_safe_close(sock)
raise
@@ -363,12 +353,16 @@ def connect(host=None, port=3306, user=None, password=None, database=None, conne
# SET NAMES: reset collation_connection to the server's default (the fixed handshake collation 45 =
# utf8mb4_general_ci otherwise clashes with MySQL 8's utf8mb4_0900_ai_ci columns -> 'illegal mix of
# collations' 1271 in a UNION/CONCAT); results stay utf8mb4 so the utf-8 decode is unchanged. autocommit=1
- # so DML persists even if the server default is autocommit=0. Both best-effort (one-time, at connect).
- for setup in ("SET NAMES utf8mb4", "SET autocommit=1"):
- try:
- connection._query(setup)
- except Exception:
- pass
+ # so DML persists even if the server default is autocommit=0. Best-effort (one-time, at connect) - but the
+ # charset has to land on *something*, else rows decode as utf-8 that never was: pre-5.5.3 servers have no
+ # utf8mb4, and 'utf8' (3-byte) is the fallback every 4.1+ server does have.
+ for alternatives in (("SET NAMES utf8mb4", "SET NAMES utf8"), ("SET autocommit=1",)):
+ for setup in alternatives:
+ try:
+ connection._query(setup)
+ break
+ except Exception:
+ pass
return connection
def _safe_close(sock):
diff --git a/extra/dbwire/postgres.py b/extra/dbwire/postgres.py
index 8402095abfd..30d36ca8bb4 100644
--- a/extra/dbwire/postgres.py
+++ b/extra/dbwire/postgres.py
@@ -29,7 +29,9 @@
from extra.dbwire import NotSupportedError
from extra.dbwire import OperationalError
from extra.dbwire import connection_lost
+from extra.dbwire import handshake_done
from extra.dbwire import keepalive
+from extra.dbwire import recvn
from extra.dbwire import ProgrammingError
_PROTOCOL_VERSION = 196608 # 3.0
@@ -70,24 +72,12 @@ def _xor(a, b):
return b"".join(chr(ord(x) ^ ord(y)) for x, y in zip(a, b))
return bytes(x ^ y for x, y in zip(a, b))
-def _recvn(sock, n):
- buf = b""
- while len(buf) < n:
- try:
- chunk = sock.recv(n - len(buf))
- except (socket.error, OSError) as ex:
- raise connection_lost(ex)
- if not chunk:
- raise InterfaceError("connection closed by server")
- buf += chunk
- return buf
-
def _read_message(sock):
- mtype = _recvn(sock, 1)
- (length,) = struct.unpack("!I", _recvn(sock, 4))
+ mtype = recvn(sock, 1)
+ (length,) = struct.unpack("!I", recvn(sock, 4))
if length < 4 or length > _MAX_MESSAGE_LENGTH:
raise InterfaceError("invalid backend message length (%d)" % length)
- return mtype, _recvn(sock, length - 4)
+ return mtype, recvn(sock, length - 4)
def _send(sock, mtype, payload):
try:
@@ -313,7 +303,6 @@ def connect(host=None, port=5432, user=None, password=None, database=None, conne
try:
sock = socket.create_connection((host or "localhost", int(port or 5432)), timeout=connect_timeout)
keepalive(sock)
- sock.settimeout(None)
except (socket.error, socket.timeout) as ex:
raise OperationalError("could not connect to '%s:%s' (%s)" % (host, port, ex))
@@ -321,9 +310,9 @@ def connect(host=None, port=5432, user=None, password=None, database=None, conne
for key, value in (("user", user or ""), ("database", database or user or ""), ("client_encoding", "UTF8")):
params += key.encode("ascii") + b"\x00" + ("%s" % value).encode("utf-8") + b"\x00"
params += b"\x00"
- _send(sock, b"", struct.pack("!I", _PROTOCOL_VERSION) + params)
try:
+ _send(sock, b"", struct.pack("!I", _PROTOCOL_VERSION) + params) # StartupMessage
_authenticate(sock, user, password)
while True: # drain until ReadyForQuery (ParameterStatus/BackendKeyData/NoticeResponse)
mtype, payload = _read_message(sock)
@@ -331,6 +320,7 @@ def connect(host=None, port=5432, user=None, password=None, database=None, conne
_raise_server_error_as_operational(payload)
if mtype == b"Z":
break
+ handshake_done(sock)
except Exception: # any setup failure (DB-API or otherwise) must still close the socket
try:
sock.close()
diff --git a/extra/dbwire/sybase.py b/extra/dbwire/sybase.py
new file mode 100644
index 00000000000..b0f424a7bf0
--- /dev/null
+++ b/extra/dbwire/sybase.py
@@ -0,0 +1,429 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+"""
+
+"""
+Minimal pure-python Sybase ASE client speaking TDS 5.0 (stdlib only, no pymssql/FreeTDS).
+
+Sybase and Microsoft share the 8-byte TDS packet framing (reused from tds.py) and part company at the
+login: ASE speaks TDS 5.0, which means a fixed-layout LOGINREC instead of LOGIN7, a capability
+negotiation, SQL sent as a LANGUAGE token in a normal packet, and results described by ROWFMT rather than
+COLMETADATA. Cleartext login only. Read-oriented for sqlmap: execute() takes a fully-formed query string,
+binary values come back as bytes (sqlmap hex-encodes them).
+
+Two encodings look like their Microsoft namesakes and are not: DECIMAL/NUMERIC carries a big-endian
+magnitude with 1 meaning *negative* (Microsoft is little-endian with 1 meaning positive), and DONE reports
+a 4-byte row count (Microsoft uses 8).
+"""
+
+import os
+import re
+import socket
+import struct
+
+from extra.dbwire import DatabaseError
+from extra.dbwire import DataError
+from extra.dbwire import IntegrityError
+from extra.dbwire import InterfaceError
+from extra.dbwire import NotSupportedError
+from extra.dbwire import OperationalError
+from extra.dbwire import handshake_done
+from extra.dbwire import keepalive
+from extra.dbwire import ProgrammingError
+
+# the packet framing is byte-identical to the Microsoft dialect - only what travels inside it differs
+from extra.dbwire.tds import _decode_datetime
+from extra.dbwire.tds import _decode_money
+from extra.dbwire.tds import _decode_smalldatetime
+from extra.dbwire.tds import _read_message
+from extra.dbwire.tds import _send_message
+
+_PKT_LOGIN = 0x02
+_PKT_NORMAL = 0x0f
+_LOGIN_CHUNK = 504 # the login travels in 512-byte packets: the server's buffer is not negotiated yet
+_PACKET_SIZE = 2048 # ASE's default 'max network packet size'; a bigger request is negotiated down
+_HEADER_SIZE = 8
+
+# tokens
+_TOKEN_LANGUAGE = 0x21
+_TOKEN_ROWFMT = 0xee
+_TOKEN_ROWFMT2 = 0x61
+_TOKEN_ROW = 0xd1
+_TOKEN_CAPABILITY = 0xe2
+_TOKEN_ENVCHANGE = 0xe3
+_TOKEN_EED = 0xe5
+_TOKEN_RETURNSTATUS = 0x79
+_TOKEN_LOGOUT = 0x71
+_TOKEN_DONE = frozenset((0xfd, 0xfe, 0xff)) # DONE / DONEPROC / DONEINPROC
+_DONE_COUNT = 0x0010 # DONE status bit: the row count is meaningful
+_ENV_PACKET_SIZE = 4 # ENVCHANGE type: the server's answer to the requested size
+
+# SYB* datatype codes. Fixed-length types carry no per-value length; everything else is length-prefixed,
+# and a zero length means NULL (TDS 5 cannot tell an empty string from NULL - neither can ASE itself).
+_T_INT1, _T_BIT, _T_INT2, _T_INT4, _T_INT8 = 0x30, 0x32, 0x34, 0x38, 0xbf
+_T_REAL, _T_FLT8, _T_MONEY, _T_MONEY4 = 0x3b, 0x3e, 0x3c, 0x7a
+_T_DATETIME, _T_DATETIME4, _T_DATE, _T_TIME = 0x3d, 0x3a, 0x31, 0x33
+_T_INTN, _T_FLTN, _T_MONEYN, _T_DATETIMN, _T_DATEN, _T_TIMEN = 0x26, 0x6d, 0x6e, 0x6f, 0x7b, 0x93
+_T_DECIMAL, _T_NUMERIC = 0x6a, 0x6c
+_T_CHAR, _T_VARCHAR, _T_BINARY, _T_VARBINARY, _T_NVARCHAR = 0x2f, 0x27, 0x2d, 0x25, 0x67
+_T_TEXT, _T_IMAGE, _T_UNITEXT = 0x23, 0x22, 0xae
+_T_LONGBINARY, _T_LONGCHAR = 0xe1, 0xaf
+_T_BIGDATETIME, _T_BIGTIME = 0xbb, 0xbc
+_T_BIGDATETIMEN, _T_BIGTIMEN = 0xbd, 0xbe
+_T_VOID = 0x1f
+
+_FIXED_LENGTH = {
+ _T_INT1: 1, _T_BIT: 1, _T_INT2: 2, _T_INT4: 4, _T_INT8: 8, _T_REAL: 4, _T_FLT8: 8,
+ _T_MONEY: 8, _T_MONEY4: 4, _T_DATETIME: 8, _T_DATETIME4: 4, _T_DATE: 4, _T_TIME: 4,
+ _T_BIGDATETIME: 8, _T_BIGTIME: 8, _T_VOID: 0,
+}
+_BLOB_TYPES = frozenset((_T_TEXT, _T_IMAGE, _T_UNITEXT)) # textptr + timestamp + 4-byte length
+_LONG_TYPES = frozenset((_T_LONGBINARY, _T_LONGCHAR)) # 4-byte length
+_BINARY_TYPES = frozenset((_T_BINARY, _T_VARBINARY, _T_IMAGE))
+_DECIMAL_TYPES = frozenset((_T_DECIMAL, _T_NUMERIC))
+# univarchar/unichar/unitext are carried as LONGBINARY/UNITEXT holding UTF-16 - only the user type says so
+_UNICODE_USERTYPES = frozenset((34, 35, 36))
+# a client that does not claim the date/time/bigdatetime capabilities gets those columns converted to a
+# plain datetime, which is lossless but would render a date as '... 00:00:00'. The user type survives the
+# conversion, so it is what says how much of the value the column actually holds.
+_USERTYPE_DATE = 37
+_USERTYPE_TIME = frozenset((38, 49))
+
+_IDENTIFIER = re.compile(r"^[A-Za-z_#][A-Za-z0-9_#$]*$")
+
+def _u8(data, off):
+ return struct.unpack(" the server default)
+ out += b"\x00" * 13 # notify-on-language-change, security & HA fields
+ out += _login_string(charset, 30) # client character set
+ out += b"\x00" # notify on character-set change
+ out += _login_string("%d" % packetsize, 6) # network packet size, as text
+ out += b"\x00" * 4 # spare
+
+ # capability negotiation: request bits say what the client can do, response bits what it wants back.
+ # All-zero is the baseline dialect - the wide/streaming formats a fallback client has no use for are
+ # exactly the ones it must not claim, or the server starts answering in them.
+ capability = struct.pack("<3B", _TOKEN_CAPABILITY, 32, 0)
+ capability += struct.pack("<2B", 1, 14) + b"\x00" * 14
+ capability += struct.pack("<2B", 2, 14) + b"\x00" * 14
+ return out + capability
+
+class _Column(object):
+ __slots__ = ("name", "type", "size", "scale", "usertype")
+
+def _parse_rowfmt(data, off, length, codec):
+ end = off + length
+ count = _u16(data, off); off += 2
+ columns = []
+ for _ in range(count):
+ col = _Column()
+ namelen = _u8(data, off); off += 1
+ col.name = data[off:off + namelen].decode(codec, "replace"); off += namelen
+ off += 1 # status (nullability - the value length says it)
+ col.usertype = _i32(data, off); off += 4
+ col.type = _u8(data, off); off += 1
+ col.size, col.scale = _FIXED_LENGTH.get(col.type, 0), 0
+ if col.type in _DECIMAL_TYPES:
+ col.size = _u8(data, off)
+ col.scale = _u8(data, off + 2) # precision (off + 1) is not needed to decode
+ off += 3
+ elif col.type in _BLOB_TYPES:
+ col.size = _i32(data, off); off += 4
+ off += 2 + _u16(data, off) # the blob's table name
+ elif col.type in _LONG_TYPES:
+ col.size = _i32(data, off); off += 4
+ elif col.type not in _FIXED_LENGTH:
+ col.size = _u8(data, off); off += 1
+ off += 1 + _u8(data, off) # locale
+ columns.append(col)
+ if off != end:
+ raise InterfaceError("malformed ROWFMT token (%d columns did not fill %d bytes)" % (count, length))
+ return columns, off
+
+def _decode_numeric(raw, scale):
+ # sign byte then a BIG-endian magnitude, and 1 means NEGATIVE (the Microsoft encoding is the mirror
+ # image of this: little-endian, 1 == positive)
+ magnitude = 0
+ for b in bytearray(raw[1:]):
+ magnitude = (magnitude << 8) | b
+ value = -magnitude if bytearray(raw)[0] else magnitude
+ if scale:
+ text = "%0*d" % (scale + 1, abs(value))
+ return ("-" if value < 0 else "") + text[:-scale] + "." + text[-scale:]
+ return "%d" % value
+
+def _decode_date(raw):
+ import datetime
+ return "%s" % (datetime.date(1900, 1, 1) + datetime.timedelta(days=_i32(raw, 0)))
+
+def _decode_time(raw):
+ import datetime
+ ticks = struct.unpack(" NULL
+ off += ptrlen + 8 # text pointer + timestamp
+ length = _i32(data, off); off += 4
+ raw, off = data[off:off + length], off + length
+ elif col.type in _LONG_TYPES:
+ length = _i32(data, off); off += 4
+ if not length:
+ return None, off
+ raw, off = data[off:off + length], off + length
+ else:
+ length = _u8(data, off); off += 1
+ if not length:
+ return None, off
+ raw, off = data[off:off + length], off + length
+ return _decode_value(col, raw, codec), off
+
+def _eed(data, off, codec):
+ # EED: number(4) state(1) class(1) sqlstate, status(1) transtate(2) message, server, procedure, line
+ number = _i32(data, off)
+ severity = _u8(data, off + 5)
+ pos = off + 6
+ pos += 1 + _u8(data, pos) # SQLSTATE
+ pos += 1 + 2 # status + transaction state
+ length = _u16(data, pos); pos += 2
+ return number, severity, data[pos:pos + length].decode(codec, "replace").strip()
+
+def _raise_server_error(number, message):
+ if number in (2601, 2615, 2627, 1105, 546, 547): # duplicate key/row, constraint, foreign key
+ raise IntegrityError(message)
+ if number in (247, 249, 257, 260, 264, 512, 8115): # conversion, overflow, arithmetic
+ raise DataError(message)
+ raise ProgrammingError(message)
+
+class Cursor(object):
+ def __init__(self, connection):
+ self.connection = connection
+ self.description = None
+ self.rowcount = -1
+ self._rows = []
+ self._pos = 0
+
+ def execute(self, query, params=None):
+ if params is not None:
+ raise NotSupportedError("parameter binding is not supported; pass a fully-formed query string")
+ self.description, self.rowcount, self._rows, self._pos = None, -1, [], 0
+ self.description, self._rows, affected = self.connection._query(query)
+ self.rowcount = len(self._rows) if self.description is not None else (affected if affected is not None else -1)
+ return self
+
+ def fetchall(self):
+ retVal = self._rows[self._pos:]
+ self._pos = len(self._rows)
+ return retVal
+
+ def fetchone(self):
+ if self._pos >= len(self._rows):
+ return None
+ retVal = self._rows[self._pos]
+ self._pos += 1
+ return retVal
+
+ def close(self):
+ self._rows = []
+
+class Connection(object):
+ def __init__(self, sock, codec):
+ self._sock = sock
+ self._codec = codec
+ # a packet larger than the size agreed at login is not an error the server reports - it drops the
+ # connection, so every send has to be chopped to what the login negotiated (ENVCHANGE 4 below)
+ self._chunk = _PACKET_SIZE - _HEADER_SIZE
+
+ def cursor(self):
+ return Cursor(self)
+
+ def commit(self):
+ pass # sqlmap issues autonomous statements; ASE is in unchained (auto-commit) mode by default
+
+ def rollback(self):
+ pass
+
+ def close(self):
+ try:
+ _send_message(self._sock, _PKT_NORMAL, struct.pack("<2B", _TOKEN_LOGOUT, 0))
+ except Exception:
+ pass
+ try:
+ self._sock.close()
+ except Exception:
+ pass
+
+ def _query(self, query):
+ # LANGUAGE token: length covers the status byte and the (single-byte encoded) statement text
+ text = query.encode(self._codec, "replace")
+ token = struct.pack(" 10: # 10 and below are informational (e.g. 'Changed database')
+ error = (number, "(remote) %s" % message)
+ off += 2 + length
+ elif token == _TOKEN_ENVCHANGE:
+ length = _u16(data, off)
+ if _u8(data, off + 2) == _ENV_PACKET_SIZE:
+ size = data[off + 4:off + 4 + _u8(data, off + 3)]
+ self._chunk = max(_LOGIN_CHUNK, int(size) - _HEADER_SIZE)
+ off += 2 + length
+ elif token == _TOKEN_RETURNSTATUS:
+ off += 4
+ elif token == _TOKEN_ROWFMT2:
+ raise NotSupportedError("the server answered with the wide ROWFMT2 format")
+ else:
+ off += 2 + _u16(data, off) # every other TDS 5 token is 2-byte length prefixed
+ if error is not None:
+ if login:
+ raise OperationalError(error[1])
+ _raise_server_error(*error)
+ return description, rows, affected
+
+def connect(host=None, port=5000, user=None, password=None, database=None, connect_timeout=None, **kwargs):
+ charset, codec = kwargs.get("charset", "utf8"), "utf-8"
+ try:
+ sock = socket.create_connection((host or "localhost", int(port or 5000)), timeout=connect_timeout)
+ keepalive(sock)
+ except (socket.error, socket.timeout) as ex:
+ raise OperationalError("could not connect to '%s:%s' (%s)" % (host, port, ex))
+
+ connection = Connection(sock, codec)
+ try:
+ login = _loginrec("dbwire", user or "", password or "", "dbwire", host or "localhost", charset, _PACKET_SIZE)
+ _send_message(sock, _PKT_LOGIN, login, _LOGIN_CHUNK)
+ connection._read_response(login=True)
+ handshake_done(sock)
+ if database:
+ if not _IDENTIFIER.match(database):
+ raise ProgrammingError("unsupported database name %r" % database)
+ connection._query("USE %s" % database)
+ except (DatabaseError, InterfaceError):
+ connection.close()
+ raise
+ except Exception as ex:
+ connection.close()
+ raise OperationalError("Sybase login failed (%s)" % ex)
+ return connection
diff --git a/extra/dbwire/tds.py b/extra/dbwire/tds.py
index 3cad7d70af7..a028e631c6a 100644
--- a/extra/dbwire/tds.py
+++ b/extra/dbwire/tds.py
@@ -6,7 +6,10 @@
"""
"""
-Minimal pure-python TDS (Tabular Data Stream) client for Microsoft SQL Server / Sybase (stdlib only).
+Minimal pure-python TDS (Tabular Data Stream) client for Microsoft SQL Server (stdlib only).
+
+LOGIN7 / TDS 7.4 is the Microsoft dialect. Sybase ASE speaks TDS 5.0 - same 8-byte packet framing, but a
+LOGINREC login and its own token/type dialect - and lives in sybase.py, which reuses the framing below.
Cleartext login only (TDS pre-login encryption negotiated to NOT_SUP); a server that forces encryption
would need TLS-in-TDS which is out of scope for the dependency-free client. Implements PRELOGIN, LOGIN7,
@@ -23,7 +26,9 @@
from extra.dbwire import NotSupportedError
from extra.dbwire import OperationalError
from extra.dbwire import connection_lost
+from extra.dbwire import handshake_done
from extra.dbwire import keepalive
+from extra.dbwire import recvn
from extra.dbwire import ProgrammingError
_MAX_MESSAGE_LENGTH = 0x40000000
@@ -38,21 +43,9 @@
def _u8(data, off):
return struct.unpack("BBH", header[:4])
if length < 8:
raise InterfaceError("invalid TDS packet length (%d)" % length)
total += length - 8
if total > _MAX_MESSAGE_LENGTH:
raise InterfaceError("TDS message exceeds the maximum allowed length (%d bytes)" % _MAX_MESSAGE_LENGTH)
- chunks.append(_recvn(sock, length - 8))
+ chunks.append(recvn(sock, length - 8))
if status & _STATUS_EOM:
break
return b"".join(chunks)
@@ -160,7 +153,10 @@ def _login7(sock, user, password, database, hostname="dbwire", appname="dbwire")
# ---- token stream + type decoding --------------------------------------------------------------------
-def _read_us_varchar(data, off):
+def _read_b_varchar(data, off):
+ # B_VARCHAR: 1-byte length in UCS-2 *characters* (not bytes), then that many 16-bit units. COLMETADATA's
+ # ColName is a B_VARCHAR (MS-TDS 2.2.7.4) - identifiers cap at 128 chars and an unaliased expression
+ # column comes back with an empty name, so the one-byte count cannot overflow.
(n,) = struct.unpack("...)
-VERSION = "1.10.8.29"
+VERSION = "1.10.8.32"
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)
diff --git a/plugins/dbms/sybase/enumeration.py b/plugins/dbms/sybase/enumeration.py
index cc984bce978..072d40dd168 100644
--- a/plugins/dbms/sybase/enumeration.py
+++ b/plugins/dbms/sybase/enumeration.py
@@ -145,7 +145,7 @@ def getTables(self, bruteForce=None):
for db in dbs:
for blind in blinds:
- query = rootQuery.inband.query % db
+ query = rootQuery.inband.query % ((db,) * 7)
retVal = pivotDumpTable("(%s) AS %s" % (query, kb.aliasName), ['%s.name' % kb.aliasName], blind=blind, alias=kb.aliasName)
if retVal:
@@ -276,7 +276,7 @@ def getColumns(self, onlyColNames=False, colTuple=None, bruteForce=None, dumpMod
logger.info(infoMsg)
for blind in blinds:
- query = rootQuery.inband.query % (conf.db, conf.db, conf.db, conf.db, conf.db, conf.db, conf.db, unsafeSQLIdentificatorNaming(tbl))
+ query = rootQuery.inband.query % (conf.db, conf.db, conf.db, conf.db, conf.db, safeSQLIdentificatorNaming(tbl, True))
retVal = pivotDumpTable("(%s) AS %s" % (query, kb.aliasName), ['%s.name' % kb.aliasName, '%s.usertype' % kb.aliasName], blind=blind, alias=kb.aliasName)
if retVal:
diff --git a/tests/test_datafiles.py b/tests/test_datafiles.py
index 4816c21e944..ef4207fd4d6 100644
--- a/tests/test_datafiles.py
+++ b/tests/test_datafiles.py
@@ -79,6 +79,43 @@ def test_column_comment_queries_format_with_three_args(self):
% (dbms.get("value"), query, ex))
+ def test_tds_table_queries_carry_the_owner(self):
+ # Regression: safeSQLIdentificatorNaming() prepends DEFAULT_MSSQL_SCHEMA ('dbo.') to any MSSQL or
+ # Sybase table name that has no dot in it. A query that projects the bare name therefore
+ # makes every table owned by anyone but dbo unreachable: --columns/--count/--dump all end up asking
+ # for 'db.dbo.'. Both dialects must join sysusers and project '.'.
+ tree = ET.parse(os.path.join(ROOT, "data", "xml", "queries.xml"))
+ seen = 0
+ for dbms in tree.findall(".//dbms"):
+ if dbms.get("value") not in ("Microsoft SQL Server", "Sybase"):
+ continue
+ for node in dbms.iter("tables"):
+ query = (node.find("inband") if node.find("inband") is not None else node).get("query") or ""
+ if not query:
+ continue
+ seen += 1
+ self.assertIn("sysusers", query, msg="%s does not resolve the table owner" % dbms.get("value"))
+ self.assertIn("+'.'+", query, msg="%s does not qualify the table with its owner" % dbms.get("value"))
+ self.assertEqual(seen, 2)
+
+ def test_sybase_catalog_queries_take_the_argument_count_the_plugin_passes(self):
+ # plugins/dbms/sybase/enumeration.py formats these with ((db,) * 7) and (db, db, db, db, db, table).
+ # pivotDumpTable() then selects '.name' (and '.usertype'), so those output column
+ # names have to survive the projection.
+ tree = ET.parse(os.path.join(ROOT, "data", "xml", "queries.xml"))
+ sybase = [_ for _ in tree.findall(".//dbms") if _.get("value") == "Sybase"][0]
+ tables = sybase.find("tables").find("inband").get("query")
+ columns = sybase.find("columns").find("inband").get("query")
+ self.assertEqual(tables.count("%s"), 7)
+ self.assertEqual(columns.count("%s"), 6)
+ self.assertIn("AS name", tables)
+ self.assertIn("usertype", columns)
+ # the table has to be resolved by owner, not by a bare sysobjects.name match: two owners can hold
+ # a same-named table, and matching on the name alone returned both tables' columns merged
+ self.assertIn("object_id(", columns)
+ self.assertNotIn("sysobjects.name='%s'", columns)
+
+
class TestErrorsXmlCompile(unittest.TestCase):
def test_all_error_regexes_compile(self):
tree = ET.parse(os.path.join(ROOT, "data", "xml", "errors.xml"))
diff --git a/tests/test_dbwire.py b/tests/test_dbwire.py
index 57dd32a8880..10ba672c499 100644
--- a/tests/test_dbwire.py
+++ b/tests/test_dbwire.py
@@ -30,11 +30,17 @@
bootstrap()
import extra.dbwire as dbwire
+from extra.dbwire import clickhouse as _clickhouse
from extra.dbwire import connection_lost
+from extra.dbwire import cubrid as _cubrid
+from extra.dbwire import firebird as _firebird
from extra.dbwire import http_origin
+from extra.dbwire import monetdb as _monetdb
from extra.dbwire import mysql as _mysql
from extra.dbwire import postgres as _postgres
from extra.dbwire import presto as _presto
+from extra.dbwire import recvn
+from extra.dbwire import sybase as _sybase
from extra.dbwire import tds as _tds
@@ -44,6 +50,7 @@ class FakeSocket(object):
def __init__(self, inbound=b""):
self.inbound = bytearray(inbound)
self.sent = bytearray()
+ self.timeouts = []
self.closed = False
def feed(self, data):
@@ -59,8 +66,8 @@ def recv(self, count):
def sendall(self, data):
self.sent.extend(data)
- def settimeout(self, _value):
- pass
+ def settimeout(self, value):
+ self.timeouts.append(value)
def setsockopt(self, *_args):
pass
@@ -177,20 +184,23 @@ def recv(count):
self.assertRaises(dbwire.OperationalError, _postgres._authenticate, sock, "user", "secret")
-class MysqlCapabilityTest(unittest.TestCase):
- def _handshake(self, server_caps):
- payload = b"\x0a" + b"8.0.0-fake\x00" + struct.pack("> 16) & 0xffff)
- payload += struct.pack("> 16) & 0xffff)
+ payload += struct.pack("BBHHBB", 4, 1 if eom else 0, len(body) + 8, 0, 0, 0) + body
@@ -276,6 +349,158 @@ def test_done_without_the_count_flag_is_not_a_row_count(self):
self.assertIsNone(_tds._parse_tokens(sock)[2])
+class SybaseLoginTest(unittest.TestCase):
+ """The TDS 5.0 LOGINREC is one fixed-layout record: every field sits at a byte offset the server counts
+ on, so a field that changes width silently shifts the credentials into the reserved area."""
+
+ def _loginrec(self):
+ return _sybase._loginrec("dbwire", "tester", "guest1234", "dbwire", "srv", "utf8", 2048)
+
+ def test_field_offsets_match_the_fixed_layout(self):
+ login = self._loginrec()
+ self.assertEqual(len(login), 568 + 35, "LOGINREC is not the fixed 568 bytes plus a CAPABILITY token")
+ for offset, size, expected in ((0, 30, b"dbwire"), (31, 30, b"tester"), (62, 30, b"guest1234"),
+ (140, 30, b"dbwire"), (171, 30, b"srv"), (462, 10, b"dbwire"),
+ (525, 30, b"utf8"), (557, 6, b"2048")):
+ self.assertEqual(login[offset:offset + size].rstrip(b"\x00"), expected, offset)
+ self.assertEqual(login[offset + size:offset + size + 1], struct.pack("BBHHBB", 4, 1, len(blob) + 8, 0, 0, 0) + blob
+ return _sybase.Connection(Replay(packet), "utf-8")._read_response()
+
+ def test_rowfmt_and_row(self):
+ blob = (b"\xee\x19\x00\x02\x00\x03one\x10\x07\x00\x00\x00\x38\x00"
+ b"\x03txt\x10\x02\x00\x00\x00\x27\x03\x00"
+ b"\xd1\x01\x00\x00\x00\x03abc"
+ b"\xfd\x10\x00\x02\x00\x01\x00\x00\x00")
+ description, rows, affected = self._response(blob)
+ self.assertEqual([_[0] for _ in description], ["one", "txt"])
+ self.assertEqual(rows, [("1", "abc")])
+ self.assertEqual(affected, 1)
+
+ def test_done_row_count_is_four_bytes(self):
+ """Microsoft widened DoneRowCount to 8 bytes; ASE did not, so reading 8 here eats the next token."""
+
+ description, rows, affected = self._response(b"\xfd\x10\x00\x02\x00\x2a\x00\x00\x00")
+ self.assertIsNone(description)
+ self.assertEqual(affected, 42)
+ self.assertIsNone(self._response(b"\xfd\x00\x00\x02\x00\x2a\x00\x00\x00")[2],
+ "a row count without the DONE_COUNT status bit is meaningless")
+
+ def test_server_error_is_raised_and_informational_message_is_not(self):
+ def eed(number, severity, message):
+ body = (struct.pack("BBHHBB", 4, 1, len(blob) + 8, 0, 0, 0) + blob), "utf-8")
+ self.assertEqual(connection._chunk, _sybase._PACKET_SIZE - 8)
+ connection._read_response()
+ self.assertEqual(connection._chunk, 8192 - 8)
+
+
+class SybaseDecoderTest(unittest.TestCase):
+ def _column(self, dtype, scale=0, usertype=0):
+ col = _sybase._Column()
+ col.name, col.type, col.size, col.scale, col.usertype = "c", dtype, 0, scale, usertype
+ return col
+
+ def test_numeric_is_the_mirror_image_of_the_microsoft_encoding(self):
+ """ASE sends a big-endian magnitude with 1 meaning negative; Microsoft sends little-endian with 1
+ meaning positive. Sharing one decoder between the two silently returns wrong numbers."""
+
+ for raw, scale, expected in ((b"\x01\x00\x00\xbcaN", 3, "-12345.678"),
+ (b"\x00\x00\x00'\x0f", 2, "99.99"),
+ (b"\x00\x00\x00\x00\x01", 0, "1"),
+ (b"\x01\x00\x00\x00\x01", 0, "-1")):
+ self.assertEqual(_sybase._decode_numeric(raw, scale), expected, raw)
+ self.assertNotEqual(_sybase._decode_numeric(b"\x01\x00\x00\xbcaN", 3),
+ _tds._decode_numeric(b"\x01\x00\x00\xbcaN", 3))
+
+ def test_date_and_time_columns_survive_the_servers_conversion(self):
+ """Unclaimed date/time capabilities make ASE send both as a plain datetime; only the user type still
+ says how much of it the column holds."""
+
+ raw = struct.pack("iBiBi", 0, 0, 0, 0, 0))
+
+ connection._call = _call
+ prepare = _cubrid._Reader(struct.pack(">iBiBi", 0, stmt_type, 0, 0, 0))
+ connection._execute(1, prepare)
+
+ args, off = [], 1 # skip the function code, then walk [len(4)][value] args
+ payload = sent[0]
+ while off < len(payload):
+ (length,) = struct.unpack(">i", payload[off:off + 4])
+ args.append(payload[off + 4:off + 4 + length])
+ off += 4 + length
+ return bytearray(args[6])[0] # handle, flag, max_col_size, max_row, binds, fetch, auto_commit
+
+ def test_dml_is_committed_by_the_execute_request(self):
+ self.assertEqual(self._auto_commit_byte(_cubrid._STMT_SELECT + 1), 1)
+
+ def test_select_does_not_end_the_transaction(self):
+ self.assertEqual(self._auto_commit_byte(_cubrid._STMT_SELECT), 0)
+
+
+class BoundedReadTest(unittest.TestCase):
+ """A length taken off the wire is attacker/corruption controlled: unchecked, it either reads until
+ memory runs out or (on a short buffer) hands back silently truncated data."""
+
+ def test_cubrid_short_response_is_not_silently_truncated(self):
+ reader = _cubrid._Reader(b"AB")
+ self.assertRaises(dbwire.InterfaceError, reader.raw, 8)
+ self.assertRaises(dbwire.InterfaceError, reader.raw, -1)
+
+ def test_firebird_rejects_an_out_of_range_length(self):
+ wire = _firebird._Wire(FakeSocket())
+ self.assertRaises(dbwire.InterfaceError, wire.recv, -1)
+ self.assertRaises(dbwire.InterfaceError, wire.recv, _firebird._MAX_MESSAGE_LENGTH + 1)
+
+ def test_monetdb_unterminated_block_stream_is_bounded(self):
+ """The MAPI block length is a 15-bit field, so only the accumulated response can be bounded."""
+
+ block = struct.pack(" never ends
+ sock = FakeSocket(block * 32)
+ original = sock.recv
+
+ def recv(count):
+ if not sock.inbound:
+ sock.feed(block * 32)
+ return original(count)
+
+ sock.recv = recv
+ saved = _monetdb._MAX_MESSAGE_LENGTH
+ try:
+ _monetdb._MAX_MESSAGE_LENGTH = 100000
+ self.assertRaises(dbwire.InterfaceError, _monetdb._getblock, sock)
+ finally:
+ _monetdb._MAX_MESSAGE_LENGTH = saved
+
+
+class DecoderTest(unittest.TestCase):
+ """Byte fixtures for the socket-free decoders, taken from what the real servers put on the wire."""
+
+ def test_tds_column_name_is_a_character_count(self):
+ """COLMETADATA ColName is B_VARCHAR - a length in UCS-2 characters (MS-TDS 2.2.7.4), so the count
+ byte is followed by twice as many bytes."""
+
+ data = struct.pack("'"; a coercion failure is a DataError."""
+
+ for message, expected in (("(remote) Syntax: In line 1, column 1 before ' 1'", dbwire.ProgrammingError),
+ ("(remote) Unknown class \"public.t\".", dbwire.ProgrammingError),
+ ("(remote) Cannot coerce value of domain \"character\" to domain \"integer\".",
+ dbwire.DataError),
+ ("(remote) unique constraint violated", dbwire.IntegrityError)):
+ self.assertRaises(expected, _cubrid.Connection._raise, -494, message)
+
+
class HelperTest(unittest.TestCase):
def test_socket_failure_maps_into_the_dbapi_hierarchy(self):
"""Callers of a PEP 249 driver only catch Error and its subclasses."""
@@ -327,8 +677,17 @@ def test_http_origin_brackets_a_literal_ipv6_host(self):
self.assertEqual(http_origin("[fe80::1]", 8123), "http://[fe80::1]:8123")
self.assertEqual(http_origin(None, 8123), "http://localhost:8123")
+ def test_recvn_reads_exactly_n_bytes_across_chunks(self):
+ """Shared by every socket module: a framed protocol has no notion of a short read."""
+
+ sock = FakeSocket(b"ABCDEFGH")
+ original = sock.recv
+ sock.recv = lambda _count: original(3) # dribble the stream 3 bytes at a time
+ self.assertEqual(recvn(sock, 8), b"ABCDEFGH")
+ self.assertRaises(dbwire.InterfaceError, recvn, FakeSocket(b"AB"), 4)
+
def test_every_module_exposes_the_dbapi_surface(self):
- for name in ("postgres", "mysql", "tds", "firebird", "cubrid", "monetdb", "clickhouse", "presto"):
+ for name in ("postgres", "mysql", "tds", "sybase", "firebird", "cubrid", "monetdb", "clickhouse", "presto"):
module = __import__("extra.dbwire.%s" % name, fromlist=["connect"])
self.assertTrue(callable(getattr(module, "connect", None)), name)