From 92c2ee47eaaee37354c0d406acfd00cac5488af2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Thu, 13 Aug 2026 14:41:00 +0200 Subject: [PATCH] Adding support for Sybase into dbwire --- data/xml/queries.xml | 4 +- extra/dbwire/README.md | 8 +- extra/dbwire/__init__.py | 5 +- extra/dbwire/sybase.py | 429 +++++++++++++++++++++++++++++ extra/dbwire/tds.py | 8 +- lib/core/dicts.py | 24 +- lib/core/settings.py | 2 +- plugins/dbms/sybase/enumeration.py | 4 +- tests/test_datafiles.py | 37 +++ tests/test_dbwire.py | 155 ++++++++++- 10 files changed, 661 insertions(+), 15 deletions(-) create mode 100644 extra/dbwire/sybase.py 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("