diff --git a/data/xml/queries.xml b/data/xml/queries.xml
index c6e1d6345..37b668073 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 ea9f0a305..2e06e3415 100644
--- a/extra/dbwire/README.md
+++ b/extra/dbwire/README.md
@@ -23,6 +23,7 @@ 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 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 |
@@ -58,8 +59,11 @@ parameter binding, prepared statements, bulk load/`COPY`, or TLS. Notable per-pr
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. LOGIN7 is the
- Microsoft dialect (TDS 7.x); Sybase ASE speaks TDS 5.0 (LOGINREC + its own token dialect), which is not
- implemented yet - ASE `-d` goes through the pymssql/FreeTDS tier for now.
+ 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 4f3c4c32b..b064a9e65 100644
--- a/extra/dbwire/__init__.py
+++ b/extra/dbwire/__init__.py
@@ -11,8 +11,9 @@ sqlmap's direct ('-d') connection when no native driver (and no SQLAlchemy) is i
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. 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
diff --git a/extra/dbwire/sybase.py b/extra/dbwire/sybase.py
new file mode 100644
index 000000000..b0f424a7b
--- /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 ecc41bc9b..a028e631c 100644
--- a/extra/dbwire/tds.py
+++ b/extra/dbwire/tds.py
@@ -9,7 +9,7 @@ See the file 'LICENSE' for copying permission
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 - which is not implemented yet, so ASE still needs pymssql.
+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,
@@ -43,9 +43,9 @@ _STATUS_EOM = 0x01
def _u8(data, off):
return struct.unpack("...)
-VERSION = "1.10.8.31"
+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 cc984bce9..072d40dd1 100644
--- a/plugins/dbms/sybase/enumeration.py
+++ b/plugins/dbms/sybase/enumeration.py
@@ -145,7 +145,7 @@ class Enumeration(GenericEnumeration):
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 @@ class Enumeration(GenericEnumeration):
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 4816c21e9..ef4207fd4 100644
--- a/tests/test_datafiles.py
+++ b/tests/test_datafiles.py
@@ -79,6 +79,43 @@ class TestQueriesXmlCoverage(unittest.TestCase):
% (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 a1e8c740d..10ba672c4 100644
--- a/tests/test_dbwire.py
+++ b/tests/test_dbwire.py
@@ -40,6 +40,7 @@ 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
@@ -348,6 +349,158 @@ class TdsFramingTest(unittest.TestCase):
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("