diff --git a/pysap/SAPNI.py b/pysap/SAPNI.py index 9c13891..17a6ced 100644 --- a/pysap/SAPNI.py +++ b/pysap/SAPNI.py @@ -25,7 +25,7 @@ from socketserver import BaseRequestHandler, ThreadingMixIn, TCPServer # External imports from scapy.fields import LenField -from scapy.packet import Packet, Raw +from scapy.packet import Packet, Raw, raw from scapy.supersocket import socket, StreamSocket # Custom imports from pysap.utils import Worker @@ -145,7 +145,11 @@ def recv(self): # Decode the packet payload according to the base class defined packet = SAPNI(nidata) if self.basecls: - packet.decode_payload_as(self.basecls) + if callable(self.basecls) and not isinstance(self.basecls, type): + cls = self.basecls(packet, raw(packet.payload)) + else: + cls = self.basecls + packet.decode_payload_as(cls) elif keep_alive_request: packet = SAPNI(length=nilength) / Raw(nidata[4:]) return packet diff --git a/pysap/SAPNWRFC.py b/pysap/SAPNWRFC.py new file mode 100644 index 0000000..eec2a2c --- /dev/null +++ b/pysap/SAPNWRFC.py @@ -0,0 +1,448 @@ +# encoding: utf-8 +# pysap - Python library for crafting SAP's network protocols packets +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License +# as published by the Free Software Foundation; either version 2 +# of the License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# Author: +# Martin Gallo (@martingalloar) +# Code contributed by SecureAuth to the OWASP CBAS project +# + +"""SAP NetWeaver RFC (NWRFC) protocol constants and TLV parser. + +The NetWeaver RFC SDK (libsapnwrfc) is the modern SAP client library used by +pyrfc, SAP JCo, SAP .NET Connector, and the SAP ABAP kernel's own outbound +RFC stack. It differs from the classic SAPRFC/CPIC framing (see +:mod:`pysap.SAPRFC`) in two key areas: + +1. **Different magic bytes**: NWRFC frames begin with ``\\x06\\xcb\\x02\\x00`` + rather than the classic ``\\x06\\x03\\x02\\x00``. + +2. **TLV encoding**: All fields (username, password, function module name, + client, language, hostname, …) are carried in a Type-Length-Value (TLV) + stream using 2-byte big-endian tags and 2-byte big-endian lengths. + Values are encoded in **UTF-16LE**. + +Both frame types are transported inside SAP NI framing (4-byte big-endian +payload length prefix) on the same RFC ports (default: TCP/3300). + +TLV tag assignments +------------------- +The tag-to-semantic mapping below was determined by analysis of live packet +captures from multiple NWRFC SDK clients (pyrfc 3.x, npl_rfc, SAP ABAP +kernel). Tags not listed here are observed in captures but their semantics +are not yet confirmed. + +.. code-block:: text + + Tag Semantic Encoding Size (bytes, typ./max.) + ------ ----------------------- ---------- ----------------------- + 0x0006 RFC destination name UTF-16LE 2 – 40 + 0x0007 Client IP address UTF-16LE 4 – 32 + 0x0008 Server hostname UTF-16LE 8 – 120 + 0x0009 Username (tertiary) UTF-16LE 4 – 48 + 0x0100 Program / task name UTF-16LE 10 – 120 + 0x0102 Function module name UTF-16LE 8 – 80 + 0x0111 Username (primary) UTF-16LE 4 – 48 + 0x0114 SAP client number UTF-16LE 2 – 8 ("001", "100", …) + 0x0117 ab_scramble password binary 6 – 84 [4B LE seed][UTF-16LE] + 0x0119 Username (secondary) UTF-16LE 4 – 48 + 0x0152 Language key UTF-16LE 2 – 4 ("E", "D", …) + 0x0201 RFC param name UTF-16LE 2 – 120 + 0x0203 RFC param value UTF-16LE variable + +Password field (tag 0x0117) +--------------------------- +The password TLV value is a raw binary field, **not** a UTF-16LE string:: + + [ 4 bytes little-endian seed ][ N×2 bytes ab_scrambled UTF-16LE payload ] + +Total field length = 4 + (password_length × 2). Minimum length is 6 bytes +(seed + one UTF-16LE character). Use :func:`pysap.utils.crypto.rfc.ab_descramble` +with ``encoding="utf-16-le"`` to recover the plaintext. + +Hostname / SID extraction +------------------------- +The server hostname value (tag ``0x0008``) frequently encodes the SAP System +ID (SID) in the pattern ``__``, for example +``sapnw752_NPL_00``. The regex :data:`NWRFC_SID_RE` matches this pattern. + +References +---------- +* Protocol reverse-engineered from libsapnwrfc (SAP NetWeaver RFC SDK). +* TLV tag assignments confirmed from live PCAP captures. +* See also: :mod:`pysap.SAPRFC` for classic RFC/CPIC framing. +""" + +# Standard imports +import html +import re +import struct + + +# ── Frame identification ────────────────────────────────────────────────────── + +#: Magic bytes that identify an NWRFC frame (bytes 0-3 of the NI payload). +#: Classic SAPRFC frames use ``b'\\x06\\x03\\x02\\x00'`` instead. +NWRFC_MAGIC = b'\x06\xcb\x02\x00' + +#: Magic bytes for classic SAPRFC/CPIC frames, shown here for comparison. +SAPRFC_MAGIC = b'\x06\x03\x02\x00' + + +# ── TLV structure ───────────────────────────────────────────────────────────── + +#: Size of a TLV header in bytes (2-byte tag + 2-byte length, both big-endian). +NWRFC_TLV_HEADER_SIZE = 4 + +#: Confirmed NWRFC TLV tag assignments. +#: Keys are integer tag values; values are human-readable semantic names. +#: All string values are encoded in UTF-16LE unless noted otherwise. +NWRFC_TAGS = { + 0x0006: "dest", # RFC destination name + 0x0007: "ip", # Client IP address (text form) + 0x0008: "hostname", # Server hostname; contains _SID_NN pattern + 0x0009: "username", # SAP username (tertiary tag) + 0x0100: "program", # Program / task handler name + 0x0102: "function_module", # RFC function module name + 0x0111: "username", # SAP username (primary logon tag) + 0x0114: "client", # SAP client number ("001", "100", …) + 0x0117: "password", # ab_scramble field: [4B seed][UTF-16LE bytes] + 0x0119: "username", # SAP username (secondary tag) + 0x0152: "language", # Language key ("E", "D", …) + 0x0201: "param_name", # RFC call parameter name + 0x0203: "param_value", # RFC call parameter value +} + +#: Username tags, in priority order (highest priority first). +#: When multiple username tags are present the highest-priority non-empty +#: value should be used. +NWRFC_USERNAME_TAGS = (0x0111, 0x0119, 0x0009) + +#: Per-tag minimum and maximum *value* length in bytes (before decoding). +#: Used to reject spurious TLV matches during scanning. +NWRFC_TAG_CONSTRAINTS = { + 0x0006: (2, 40), # dest + 0x0007: (4, 32), # ip + 0x0008: (4, 120), # hostname + 0x0009: (4, 48), # username (tertiary) + 0x0100: (10, 120), # program + 0x0102: (8, 80), # function module + 0x0111: (4, 48), # username (primary) + 0x0114: (2, 8), # client + 0x0117: (6, 84), # password (4-byte seed + scrambled UTF-16LE) + 0x0119: (4, 48), # username (secondary) + 0x0152: (2, 4), # language + 0x0201: (2, 120), # param_name + # 0x0203 (param_value): no constraint, value length varies widely + # (scalars to large XML table fragments). +} + +#: Minimum byte count of the ab_scramble seed prefix inside a password field. +NWRFC_PASSWORD_SEED_SIZE = 4 + +#: Regex that extracts the SAP System ID (SID) from a server hostname value. +#: Matches the ``_SID_NN`` suffix, e.g. ``sapnw752_NPL_00`` → ``NPL``. +NWRFC_SID_RE = re.compile(r'_([A-Z][A-Z0-9]{2})_\d{2}(?:\s|$)') + + +# ── TLV parser ──────────────────────────────────────────────────────────────── + +def parse_tlv(data, tags=None): + """Parse NWRFC TLV fields from a raw byte buffer. + + Scans *data* for TLV entries whose 2-byte big-endian tag appears in + *tags* (or in :data:`NWRFC_TAGS` when *tags* is ``None``). Length + constraints from :data:`NWRFC_TAG_CONSTRAINTS` are enforced to suppress + false positives; unrecognised or out-of-range entries are skipped. + + The buffer does **not** need to start at a TLV boundary — the function + searches forward byte-by-byte, which handles frames where the TLV stream + is preceded by other header fields. + + Args: + data (bytes): Raw frame payload (NI payload, without the 4-byte NI + length header). + tags (collection | None): Iterable of integer tag values to extract. + Defaults to all keys in :data:`NWRFC_TAGS`. + + Yields: + tuple[int, bytes]: ``(tag, raw_value)`` pairs in the order they + appear in *data*. The caller is responsible for decoding *raw_value* + (typically ``raw_value.decode("utf-16-le").rstrip("\\x00")`` for + string fields, or passing to + :func:`pysap.utils.crypto.rfc.ab_descramble` for the password tag). + + Example:: + + >>> from pysap.SAPNWRFC import parse_tlv, NWRFC_TAGS + >>> for tag, value in parse_tlv(frame_data): + ... print(hex(tag), NWRFC_TAGS.get(tag), value) + """ + if tags is None: + tags = set(NWRFC_TAGS) + else: + tags = set(tags) + + total = len(data) + off = 0 + + while off + NWRFC_TLV_HEADER_SIZE <= total: + tag = struct.unpack_from(">H", data, off)[0] + length = struct.unpack_from(">H", data, off + 2)[0] + + if tag not in tags: + off += 1 + continue + + constraints = NWRFC_TAG_CONSTRAINTS.get(tag) + if constraints: + min_len, max_len = constraints + if not (min_len <= length <= max_len): + off += 1 + continue + + end = off + NWRFC_TLV_HEADER_SIZE + length + if end > total: + off += 1 + continue + + yield tag, data[off + NWRFC_TLV_HEADER_SIZE: end] + off = end + + +def find_tlv_field_by_marker(data, marker, search_start=0): + """Find a TLV field by its 2-byte tag, scanning byte-by-byte. + + This is an alternative to :func:`parse_tlv` for buffers where the field + order varies across NWRFC SDK versions and :func:`parse_tlv`'s + sequential scan may miss a field that is preceded by an unrecognised + or misaligned entry. It scans for any 4-byte delimiter of the form + ``[2-byte end-of-prev][2-byte tag]`` followed by a 2-byte big-endian + length, i.e.:: + + [end-of-prev (2)][tag (2)][length (2)][value (length)] + + Args: + data (bytes): Raw frame payload to scan. + marker (int | bytes): Tag to search for, either as an integer (e.g. + ``0x0117``) or as 2 raw bytes (e.g. ``b"\\x01\\x17"``). + search_start (int): Offset to start scanning from. + + Returns: + tuple[bytes | None, int]: ``(value, end_offset)`` if found, where + *end_offset* is the offset just past the value (suitable as + *search_start* for a subsequent call); otherwise ``(None, + search_start)``. + """ + if isinstance(marker, int): + marker = struct.pack(">H", marker) + + idx = search_start + while idx < len(data) - 7: + if data[idx + 2:idx + 4] == marker: + length = struct.unpack(">H", data[idx + 4:idx + 6])[0] + end = idx + 6 + length + if length > 0 and end <= len(data): + return data[idx + 6:end], end + idx += 1 + return None, search_start + + +def find_tlv_field_by_padd(data, padd): + """Find a TLV field by its full 4-byte padding marker. + + The 4-byte *padd* value encodes ``[end-of-prev-field (2)][tag (2)]`` and + only matches when the immediately preceding field is the one the caller + expects, making this more precise than :func:`find_tlv_field_by_marker` + when the field order is known and fixed. + + Layout:: + + [padd (4)][length (2, big-endian)][value (length)] + + Args: + data (bytes): Raw frame payload to scan. + padd (bytes): 4-byte padding marker to search for. + + Returns: + bytes | None: The value bytes, or ``None`` if not found or invalid. + """ + idx = data.find(padd) + if idx < 0: + return None + offset = idx + len(padd) + if offset + 2 > len(data): + return None + length = struct.unpack(">H", data[offset:offset + 2])[0] + if length == 0 or offset + 2 + length > len(data): + return None + return data[offset + 2:offset + 2 + length] + + +def decode_string(raw): + """Decode a UTF-16LE TLV value, stripping null and space padding. + + Args: + raw (bytes): Raw TLV value bytes. + + Returns: + str: Decoded string, or an empty string if decoding fails. + """ + try: + return raw.decode("utf-16-le").rstrip("\x00 ") + except (UnicodeDecodeError, ValueError): + return "" + + +def decode_value(raw): + """Decode a TLV value of unknown encoding, trying UTF-16LE then ASCII. + + Most NWRFC TLV values are UTF-16LE (see :func:`decode_string`), but some + fields observed in F_SAP_SEND bodies (e.g. CPIC marker fields on older + SDK versions) are plain ASCII. This heuristically picks UTF-16LE when + *raw* "looks like" UTF-16LE (every other byte is ``0x00``, as is the + case for ASCII-range characters encoded that way), falling back to + ASCII (with replacement of invalid bytes) otherwise. + + Args: + raw (bytes | None): Raw TLV value bytes. + + Returns: + str | None: Decoded string with null/space padding stripped, or + ``None`` if *raw* is ``None``. + """ + if raw is None: + return None + + if len(raw) >= 4 and raw[1:2] == b"\x00" and raw[3:4] == b"\x00": + try: + return raw.decode("utf-16-le").strip("\x00 ") + except (UnicodeDecodeError, ValueError): + pass + + return raw.decode("ascii", errors="replace").strip("\x00 ") + + +# ── RFC call payload extraction ───────────────────────────────────────────── + +#: Pattern for valid RFC parameter names: alphanumeric plus "_" and "/" +#: (used to reject spurious 0x0201 matches that aren't real parameter names). +_RFC_PARAM_NAME_RE = re.compile(r"^[A-Za-z0-9_/]+$") + + +def extract_rfc_params(raw): + """Extract RFC function call parameters from an F_SAP_SEND body. + + Scans for ``0x0201`` (param_name) TLVs carrying a valid identifier, then + reads the immediately following ``0x0203`` (param_value) TLV via + :func:`find_tlv_field_by_marker`. + + Args: + raw (bytes): Raw F_SAP_SEND body to scan. + + Returns: + dict[str, str]: Mapping of parameter name to decoded value, in the + order they appear in *raw*. + """ + name_min, name_max = NWRFC_TAG_CONSTRAINTS[0x0201] + name_tag = struct.pack(">H", 0x0201) + + params = {} + idx = 0 + while idx < len(raw) - 7: + if raw[idx + 2:idx + 4] == name_tag: + name_len = struct.unpack(">H", raw[idx + 4:idx + 6])[0] + name_end = idx + 6 + name_len + if name_min <= name_len <= name_max and name_end <= len(raw): + name = decode_value(raw[idx + 6:name_end]) + if name and _RFC_PARAM_NAME_RE.match(name): + val, _ = find_tlv_field_by_marker(raw, 0x0203, name_end) + if val: + decoded = decode_value(val) + if decoded is not None: + params[name] = decoded + idx += 1 + return params + + +#: Matches top-level ``...`` blocks (and nested ```` blocks) +#: in an RFC call's XML-serialised table/structure parameters. +_XML_TAG_RE = re.compile(r"<([A-Z_/][A-Z0-9_/]*)>(.*?)", re.DOTALL) +_XML_ITEM_RE = re.compile(r"(.*?)", re.DOTALL) + + +def extract_xml_data(raw): + """Extract XML-encoded parameters and table rows from an RFC call body. + + The NetWeaver RFC SDK serialises table/structure parameters as ASCII XML + fragments embedded in the F_SAP_SEND body, e.g.:: + + value + base64== + + Args: + raw (bytes): Raw F_SAP_SEND body. Everything before the first ``<`` + is treated as binary and ignored; the remainder is decoded as + ASCII (XML is always ASCII in NWRFC). + + Returns: + dict[str, str | list]: Mapping of parameter name to value, where + value is either a plain string (scalar) or a list of rows (table + parameter). Table rows are dicts of sub-field name to value, except + rows of a single unnamed field which are returned as plain strings. + A sub-field whose value itself contains ```` entries (nested + table, e.g. ABAP source lines) is returned as a list of strings — + but note rows containing a *nested* ````-bearing sub-field are + not split correctly, as the non-greedy outer ```` match ends + at the first inner ````. + """ + result = {} + raw_bytes = bytes(raw) + + xml_start = raw_bytes.find(b"<") + if xml_start == -1: + return result + + try: + xml_region = raw_bytes[xml_start:].decode("ascii", errors="replace") + except Exception: + return result + + for m in _XML_TAG_RE.finditer(xml_region): + tag, content = m.group(1), m.group(2) + if len(tag) < 2: + continue + content = content.strip() + + if "" in content: + rows = [] + for item_m in _XML_ITEM_RE.finditer(content): + item_body = item_m.group(1).strip() + if "<" in item_body: + row = {} + for fld in _XML_TAG_RE.finditer(item_body): + fname, fval = fld.group(1), fld.group(2) + if "" in fval: + row[fname] = [html.unescape(li.group(1)) + for li in _XML_ITEM_RE.finditer(fval)] + else: + row[fname] = html.unescape(fval.strip()) + if row: + rows.append(row) + else: + rows.append(html.unescape(item_body)) + if rows: + result[tag] = rows + else: + result[tag] = html.unescape(content) + + return result diff --git a/pysap/SAPRFC.py b/pysap/SAPRFC.py index 584deca..9f2906d 100644 --- a/pysap/SAPRFC.py +++ b/pysap/SAPRFC.py @@ -910,7 +910,7 @@ class SAPRFC(PacketNoPadded): # answer from Anon GW ConditionalField(StrFixedLenField("anon_repl_sign", b'\x05\x00\x00\x00', 4), lambda pkt: pkt.version == 0x6 and 'F_V_RECEIVE' == str(pkt.vector)), - ConditionalField(PacketListField("repl", None, DEF_FIELDS, length_from=lambda pkt: pkt.codepage_size2), lambda pkt: pkt.version == 0x6 and 'F_V_RECEIVE' in str(pkt.vector) and pkt.codepage_size2 > 0), + ConditionalField(PacketListField("repl", None, DEF_FIELDS, length_from=lambda pkt: pkt.codepage_size2), lambda pkt: pkt.version == 0x6 and 'F_V_RECEIVE' in str(pkt.vector) and (pkt.codepage_size2 or 0) > 0), ConditionalField(ShortField("cpic_packet_size", 0x0), lambda pkt: pkt.version == 0x6 and 'F_V_SEND_DATA' in str(pkt.vector)), # chipik ConditionalField(IntField("rfc_packet_size", 0x0), lambda pkt: pkt.version == 0x6 and 'F_V_SEND_DATA' in str(pkt.vector)), # chipik diff --git a/pysap/utils/crypto/rfc.py b/pysap/utils/crypto/rfc.py new file mode 100644 index 0000000..0ff75ec --- /dev/null +++ b/pysap/utils/crypto/rfc.py @@ -0,0 +1,170 @@ +# encoding: utf-8 +# pysap - Python library for crafting SAP's network protocols packets +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License +# as published by the Free Software Foundation; either version 2 +# of the License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# Author: +# Martin Gallo (@martingalloar) +# Code contributed by SecureAuth to the OWASP CBAS project +# + +"""SAP RFC ab_scramble password obfuscation. + +SAP's RFC layer uses a symmetric XOR-based obfuscation scheme called +*ab_scramble* to encode passwords on the wire in unencrypted (non-SNC) +RFC/CPIC connections. The scheme is **not** a cryptographic hash — it is +fully reversible given only the ciphertext. + +Wire format +----------- +The password field is always:: + + [ 4-byte little-endian seed ][ obfuscated password bytes ] + +The seed is chosen by the client at connection time. + +Encoding variants +----------------- +Two client-side encodings are observed in practice: + +* **ASCII** (classic SAPRFC / pyrfc): the plaintext password is encoded as + ASCII bytes before obfuscation. The descrambled result is decoded with + ``"ascii"``. + +* **UTF-16LE** (NetWeaver RFC SDK / npl_rfc / SAP JCo): the plaintext + password is encoded as UTF-16LE before obfuscation. The descrambled + result must be decoded with ``"utf-16-le"``. Callers must pass + ``encoding="utf-16-le"`` explicitly. + +References +---------- +* Reverse-engineered from libsapnwrfc (SAP NWRFC SDK). +* Confirmed against live PCAP captures of pyrfc and npl_rfc clients. +""" + +# Standard imports +import os +import struct + + +# 64-byte XOR lookup table extracted from libsapnwrfc (ab_scramble routine). +# This table is the sole secret of the scheme; it is static and embedded in +# every SAP client SDK that implements RFC. +_TABLE = bytes([ + 0xf0, 0xed, 0x53, 0xb8, 0x32, 0x44, 0xf1, 0xf8, + 0x76, 0xc6, 0x79, 0x59, 0xfd, 0x4f, 0x13, 0xa2, + 0xc1, 0x51, 0x95, 0xec, 0x54, 0x83, 0xc2, 0x34, + 0x77, 0x49, 0x43, 0xa2, 0x7d, 0xe2, 0x65, 0x96, + 0x5e, 0x53, 0x98, 0x78, 0x9a, 0x17, 0xa3, 0x3c, + 0xd3, 0x83, 0xa8, 0xb8, 0x29, 0xfb, 0xdc, 0xa5, + 0x55, 0xd7, 0x02, 0x77, 0x84, 0x13, 0xac, 0xdd, + 0xf9, 0xb8, 0x31, 0x16, 0x61, 0x0e, 0x6d, 0xfa, +]) + +# Minimum field size: 4-byte seed + at least 1 password byte +_MIN_FIELD_SIZE = 5 + +# Maximum observed password length (SAP enforces 40 chars; UTF-16LE doubles bytes) +_MAX_PASSWORD_BYTES = 80 + + +def ab_descramble(raw, encoding="ascii"): + """Descramble an SAP RFC ab_scramble password field. + + Args: + raw (bytes): Full password field — ``[4-byte LE seed][obfuscated bytes]``. + encoding (str): Character encoding of the obfuscated payload. + Use ``"ascii"`` (default) for classic SAPRFC / pyrfc clients. + Use ``"utf-16-le"`` for NetWeaver RFC SDK (npl_rfc, JCo) clients. + + Returns: + str: Plaintext password. + + Raises: + ValueError: If *raw* is shorter than :data:`_MIN_FIELD_SIZE` bytes. + UnicodeDecodeError: If the descrambled bytes cannot be decoded with + the given *encoding*. + + Example:: + + >>> from pysap.utils.crypto.rfc import ab_descramble + >>> ab_descramble(bytes.fromhex("a3b7e05a3384be74606be2de")) + 'secret' + """ + if len(raw) < _MIN_FIELD_SIZE: + raise ValueError( + "ab_scramble field too short: need at least {} bytes, got {}".format( + _MIN_FIELD_SIZE, len(raw) + ) + ) + + seed = struct.unpack_from(">> from pysap.utils.crypto.rfc import ab_scramble, ab_descramble + >>> raw = ab_scramble("secret", seed=0x12345678) + >>> ab_descramble(raw) + 'secret' + """ + if seed is None: + seed = struct.unpack("> 5)) & 0xFFFFFFFF + start_idx = (tmp ^ ((seed << 1) & 0xFFFFFFFF)) & 0xFFFFFFFF + + for i in range(len(data)): + tidx = (start_idx + i) & 0x3F + sval = ((seed * i * i) - i) & 0xFFFFFFFF + data[i] ^= _TABLE[tidx] ^ (sval & 0xFF) diff --git a/pysap/utils/fields.py b/pysap/utils/fields.py index 4fa86b8..a40408d 100644 --- a/pysap/utils/fields.py +++ b/pysap/utils/fields.py @@ -154,19 +154,48 @@ def randval(self): class StrFixedLenPaddedField(StrFixedLenField): """Packet field that has a fixed length and is padded with a given character. + + The optional *encoding* parameter enables transparent decoding of fields + that carry multi-byte character encodings in a fixed-width byte buffer. + This is required for SAP NetWeaver RFC SDK (NWRFC) connections, where + fixed-length string fields are encoded in **UTF-16LE** rather than ASCII. + + When *encoding* is set: + + * :meth:`getfield` attempts to decode the raw bytes with *encoding* and + strips trailing null and space padding from the decoded string. If + decoding fails the raw bytes are returned unchanged (graceful fallback). + * :meth:`addfield` encodes a ``str`` value with *encoding*, padding with + NUL bytes, before writing. """ - __slots__ = ["length_from", "padd"] + __slots__ = ["length_from", "padd", "encoding"] - def __init__(self, name, default, length=None, length_from=None, padd=" "): + def __init__(self, name, default, length=None, length_from=None, padd=" ", encoding=None): StrFixedLenField.__init__(self, name, default, length, length_from) self.padd = padd.encode() if isinstance(padd, str) else padd + self.encoding = encoding def getfield(self, pkt, s): l = self.length_from(pkt) - return s[l:], self.m2i(pkt, s[:l]) + raw = s[:l] + if self.encoding: + try: + decoded = raw.decode(self.encoding).rstrip("\x00 ") + return s[l:], decoded + except (UnicodeDecodeError, AttributeError): + pass + return s[l:], self.m2i(pkt, raw) def addfield(self, pkt, s, val): l = self.length_from(pkt) + if self.encoding and isinstance(val, str): + # Pad with NUL bytes regardless of self.padd so that trailing + # padding can be safely stripped after decoding. (Padding with + # the default ASCII space b" " would produce 0x2020 dagger chars + # in UTF-16LE, which rstrip("\x00 ") would not remove.) + encoded = val.encode(self.encoding) + padded = (encoded + b'\x00' * l)[:l] + return s + padded if val is None: val = b"" if isinstance(val, str): diff --git a/tests/fields_test.py b/tests/fields_test.py index 479c410..e5b24be 100644 --- a/tests/fields_test.py +++ b/tests/fields_test.py @@ -107,6 +107,30 @@ def test_str_fixed_len_padded_field_roundtrip(self): self.assertEqual(remaining, b"") self.assertEqual(value, b"ab ") + def test_str_fixed_len_padded_field_utf16le_roundtrip(self): + pkt = DummyLengthPacket(8) + field = StrFixedLenPaddedField("value", b"", length_from=lambda pkt: pkt.size, + encoding="utf-16-le") + + raw = field.addfield(pkt, b"", "AB") + self.assertEqual(raw, b"A\x00B\x00\x00\x00\x00\x00") + + remaining, value = field.getfield(pkt, raw) + self.assertEqual(remaining, b"") + self.assertEqual(value, "AB") + + def test_str_fixed_len_padded_field_utf16le_non_ascii(self): + pkt = DummyLengthPacket(8) + field = StrFixedLenPaddedField("value", b"", length_from=lambda pkt: pkt.size, + encoding="utf-16-le") + + raw = field.addfield(pkt, b"", "Ä1") + self.assertEqual(raw, "Ä1".encode("utf-16-le") + b"\x00\x00\x00\x00") + + remaining, value = field.getfield(pkt, raw) + self.assertEqual(remaining, b"") + self.assertEqual(value, "Ä1") + def test_str_null_fixed_len_padded_field_getfield(self): pkt = DummyLengthPacket(4) field = StrNullFixedLenPaddedField("value", b"", length_from=lambda pkt: pkt.size, padd=" ") diff --git a/tests/sapni_test.py b/tests/sapni_test.py index c15edd5..f332d00 100755 --- a/tests/sapni_test.py +++ b/tests/sapni_test.py @@ -150,6 +150,52 @@ class SomeClass(Packet): self.stop_server() + def test_sapnistreamsocket_base_cls_dispatcher(self): + """Test SAPNIStreamSocket dispatching the base class dynamically based + on the raw payload (e.g. NWRFC vs classic SAPRFC magic bytes)""" + self.start_server(self.test_address, self.test_port, SAPNITestHandler) + + class ClassA(Packet): + fields_desc = [StrField("text", None)] + + class ClassB(Packet): + fields_desc = [StrField("text", None)] + + def dispatcher(packet, payload): + if payload.startswith(b"AAAA"): + return ClassA + return ClassB + + sock = socket.socket() + sock.connect((self.test_address, self.test_port)) + + self.client = SAPNIStreamSocket(sock, base_cls=dispatcher) + packet = self.client.sr(Raw(b"AAAA" + self.test_string)) + self.client.close() + + self.assertIn(SAPNI, packet) + self.assertIn(ClassA, packet) + self.assertNotIn(ClassB, packet) + self.assertEqual(packet[ClassA].text, b"AAAA" + self.test_string) + + self.stop_server() + + self.start_server(self.test_address, self.test_port + 1, SAPNITestHandler) + + sock = socket.socket() + sock.connect((self.test_address, self.test_port + 1)) + + self.client = SAPNIStreamSocket(sock, base_cls=dispatcher) + packet = self.client.sr(Raw(self.test_string)) + self.client.close() + + self.assertIn(SAPNI, packet) + self.assertIn(ClassB, packet) + self.assertNotIn(ClassA, packet) + self.assertEqual(packet[ClassB].text, self.test_string) + + self.stop_server() + def test_sapnistreamsocket_getnisocket(self): """Test SAPNIStreamSocket get nisocket class method""" self.start_server(self.test_address, self.test_port, SAPNITestHandler) diff --git a/tests/sapnwrfc_test.py b/tests/sapnwrfc_test.py new file mode 100644 index 0000000..c6b6905 --- /dev/null +++ b/tests/sapnwrfc_test.py @@ -0,0 +1,326 @@ +# encoding: utf-8 +# pysap - Python library for crafting SAP's network protocols packets +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License +# as published by the Free Software Foundation; either version 2 +# of the License, or (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# Author: +# Martin Gallo (@martingalloar) +# Code contributed by SecureAuth to the OWASP CBAS project + +# Standard imports +import struct +import unittest + +# Custom imports +from pysap.SAPNWRFC import (NWRFC_MAGIC, SAPRFC_MAGIC, NWRFC_TAGS, + NWRFC_TAG_CONSTRAINTS, NWRFC_USERNAME_TAGS, + NWRFC_SID_RE, parse_tlv, decode_string, decode_value, + find_tlv_field_by_marker, find_tlv_field_by_padd, + extract_rfc_params, extract_xml_data) +from pysap.utils.crypto.rfc import ab_scramble, ab_descramble + + +def build_tlv(tag, value): + return struct.pack(">HH", tag, len(value)) + value + + +class PySAPNWRFCTest(unittest.TestCase): + + def test_magic_bytes(self): + self.assertEqual(NWRFC_MAGIC, b"\x06\xcb\x02\x00") + self.assertEqual(SAPRFC_MAGIC, b"\x06\x03\x02\x00") + self.assertNotEqual(NWRFC_MAGIC, SAPRFC_MAGIC) + + def test_parse_tlv_single_field(self): + value = "100".encode("utf-16-le") + data = build_tlv(0x0114, value) + + tags = list(parse_tlv(data)) + + self.assertEqual(tags, [(0x0114, value)]) + self.assertEqual(decode_string(tags[0][1]), "100") + + def test_parse_tlv_multiple_fields(self): + client = build_tlv(0x0114, "001".encode("utf-16-le")) + user = build_tlv(0x0111, "DEVELOPER".encode("utf-16-le")) + lang = build_tlv(0x0152, "E".encode("utf-16-le")) + data = client + user + lang + + tags = dict(parse_tlv(data)) + + self.assertEqual(decode_string(tags[0x0114]), "001") + self.assertEqual(decode_string(tags[0x0111]), "DEVELOPER") + self.assertEqual(decode_string(tags[0x0152]), "E") + + def test_parse_tlv_filters_by_tags(self): + client = build_tlv(0x0114, "001".encode("utf-16-le")) + user = build_tlv(0x0111, "DEVELOPER".encode("utf-16-le")) + data = client + user + + tags = dict(parse_tlv(data, tags=[0x0111])) + + self.assertNotIn(0x0114, tags) + self.assertEqual(decode_string(tags[0x0111]), "DEVELOPER") + + def test_parse_tlv_skips_out_of_range_length(self): + # Length constraint for client (0x0114) is (2, 8), so a 1-byte value + # must be skipped. + bogus = build_tlv(0x0114, b"\x00") + valid = build_tlv(0x0114, "001".encode("utf-16-le")) + data = bogus + valid + + tags = list(parse_tlv(data)) + + self.assertEqual(len(tags), 1) + self.assertEqual(decode_string(tags[0][1]), "001") + + def test_parse_tlv_handles_unaligned_data(self): + # The TLV stream is preceded by some other header bytes that don't + # form a valid TLV entry; the scanner should skip forward byte by + # byte until it finds a recognised tag. + prefix = b"\x00\x01\x02\x03" + valid = build_tlv(0x0152, "E".encode("utf-16-le")) + data = prefix + valid + + tags = list(parse_tlv(data)) + + self.assertEqual(len(tags), 1) + self.assertEqual(tags[0][0], 0x0152) + self.assertEqual(decode_string(tags[0][1]), "E") + + def test_decode_string_strips_padding(self): + raw = "DEVELOPER\x00\x00".encode("utf-16-le") + self.assertEqual(decode_string(raw), "DEVELOPER") + + def test_decode_string_invalid_returns_empty(self): + self.assertEqual(decode_string(b"\x00"), "") + + def test_decode_value_none(self): + self.assertIsNone(decode_value(None)) + + def test_decode_value_utf16le(self): + raw = "DEVELOPER\x00\x00".encode("utf-16-le") + self.assertEqual(decode_value(raw), "DEVELOPER") + + def test_decode_value_ascii(self): + raw = b"DEVELOPER\x00" + self.assertEqual(decode_value(raw), "DEVELOPER") + + def test_decode_value_short_ascii(self): + # Too short to be detected as UTF-16LE (< 4 bytes); falls back to ASCII. + self.assertEqual(decode_value(b"001"), "001") + + def test_username_tags_priority_order(self): + self.assertEqual(NWRFC_USERNAME_TAGS, (0x0111, 0x0119, 0x0009)) + for tag in NWRFC_USERNAME_TAGS: + self.assertEqual(NWRFC_TAGS[tag], "username") + + def test_sid_regex_extraction(self): + match = NWRFC_SID_RE.search("sapnw752_NPL_00") + self.assertIsNotNone(match) + self.assertEqual(match.group(1), "NPL") + + def test_sid_regex_no_match(self): + self.assertIsNone(NWRFC_SID_RE.search("sapnw752")) + + def test_tag_constraints_cover_all_tags(self): + # 0x0203 (param_value) has no length constraint: RFC parameter + # values vary widely in size (scalars to large XML table fragments). + for tag in NWRFC_TAGS: + if tag == 0x0203: + self.assertNotIn(tag, NWRFC_TAG_CONSTRAINTS) + else: + self.assertIn(tag, NWRFC_TAG_CONSTRAINTS) + + def test_parse_tlv_dest_and_ip(self): + dest = build_tlv(0x0006, "BACKEND".encode("utf-16-le")) + ip = build_tlv(0x0007, "10.0.0.1".encode("utf-16-le")) + data = dest + ip + + tags = dict(parse_tlv(data)) + + self.assertEqual(decode_string(tags[0x0006]), "BACKEND") + self.assertEqual(decode_string(tags[0x0007]), "10.0.0.1") + + def test_parse_tlv_param_name_and_value(self): + name = build_tlv(0x0201, "IV_NAME".encode("utf-16-le")) + value = build_tlv(0x0203, "Hello, World!".encode("utf-16-le")) + data = name + value + + tags = dict(parse_tlv(data)) + + self.assertEqual(decode_string(tags[0x0201]), "IV_NAME") + self.assertEqual(decode_string(tags[0x0203]), "Hello, World!") + + def test_find_tlv_field_by_marker_int_tag(self): + # [end-of-prev (2)][tag (2)][length (2)][value] + value = "DEVELOPER".encode("utf-16-le") + data = b"\x00\x00" + struct.pack(">H", 0x0111) + struct.pack(">H", len(value)) + value + + found, end = find_tlv_field_by_marker(data, 0x0111) + + self.assertEqual(found, value) + self.assertEqual(end, len(data)) + + def test_find_tlv_field_by_marker_bytes_tag_and_search_start(self): + value1 = "001".encode("utf-16-le") + value2 = "100".encode("utf-16-le") + entry1 = b"\x00\x00" + b"\x01\x14" + struct.pack(">H", len(value1)) + value1 + entry2 = b"\x00\x00" + b"\x01\x14" + struct.pack(">H", len(value2)) + value2 + data = entry1 + entry2 + + found1, end1 = find_tlv_field_by_marker(data, b"\x01\x14") + self.assertEqual(found1, value1) + + found2, end2 = find_tlv_field_by_marker(data, b"\x01\x14", search_start=end1) + self.assertEqual(found2, value2) + + def test_find_tlv_field_by_marker_not_found(self): + data = b"\x00\x00\x01\x11\x00\x00" + found, end = find_tlv_field_by_marker(data, 0x0117) + self.assertIsNone(found) + self.assertEqual(end, 0) + + def test_find_tlv_field_by_padd(self): + value = "001".encode("utf-16-le") + padd = b"\x00\x00\x01\x14" + data = b"garbage" + padd + struct.pack(">H", len(value)) + value + + found = find_tlv_field_by_padd(data, padd) + + self.assertEqual(found, value) + + def test_find_tlv_field_by_padd_not_found(self): + self.assertIsNone(find_tlv_field_by_padd(b"\x00\x00\x00\x00", b"\x01\x14\x01\x17")) + + def test_parse_tlv_param_value_large(self): + # 0x0203 has no length constraint, so large values must parse fine. + big = ("" + "X" * 5000 + "").encode("utf-16-le") + data = build_tlv(0x0203, big) + + tags = dict(parse_tlv(data)) + + self.assertEqual(decode_string(tags[0x0203]), big.decode("utf-16-le")) + + +class PySAPNWRFCExtractTest(unittest.TestCase): + + def _param_field(self, tag, value): + return b"\x00\x00" + struct.pack(">H", tag) + struct.pack(">H", len(value)) + value + + def test_extract_rfc_params_single(self): + name = "IV_GUID".encode("utf-16-le") + value = "Hello, World!".encode("utf-16-le") + data = self._param_field(0x0201, name) + self._param_field(0x0203, value) + + params = extract_rfc_params(data) + + self.assertEqual(params, {"IV_GUID": "Hello, World!"}) + + def test_extract_rfc_params_multiple(self): + data = (self._param_field(0x0201, "IV_A".encode("utf-16-le")) + + self._param_field(0x0203, "1".encode("utf-16-le")) + + self._param_field(0x0201, "IV_B".encode("utf-16-le")) + + self._param_field(0x0203, "2".encode("utf-16-le"))) + + params = extract_rfc_params(data) + + self.assertEqual(params, {"IV_A": "1", "IV_B": "2"}) + + def test_extract_rfc_params_skips_invalid_name(self): + # Name containing non-identifier characters should be skipped. + bogus_name = "\x01\x02\x03\x04".encode("utf-16-le") + data = self._param_field(0x0201, bogus_name) + self._param_field(0x0203, "x".encode("utf-16-le")) + + params = extract_rfc_params(data) + + self.assertEqual(params, {}) + + def test_extract_rfc_params_no_value(self): + name = "IV_GUID".encode("utf-16-le") + data = self._param_field(0x0201, name) + + params = extract_rfc_params(data) + + self.assertEqual(params, {}) + + def test_extract_xml_data_scalar(self): + raw = b"junkabc123==" + + result = extract_xml_data(raw) + + self.assertEqual(result, {"IV_GUID": "abc123=="}) + + def test_extract_xml_data_table_structured_rows(self): + raw = (b"" + b"value1" + b"value2" + b"") + + result = extract_xml_data(raw) + + self.assertEqual(result, {"IT_MODULE": [{"FIELD": "value1"}, {"FIELD": "value2"}]}) + + def test_extract_xml_data_table_scalar_rows(self): + raw = b"line1line2" + + result = extract_xml_data(raw) + + self.assertEqual(result, {"IT_LINES": ["line1", "line2"]}) + + def test_extract_xml_data_html_unescape(self): + raw = b"a & b < c" + + result = extract_xml_data(raw) + + self.assertEqual(result, {"IV_TEXT": "a & b < c"}) + + def test_extract_xml_data_no_xml(self): + self.assertEqual(extract_xml_data(b"\x06\xcb\x02\x00binary"), {}) + + +class PySAPNWRFCPasswordTest(unittest.TestCase): + + def test_ab_scramble_roundtrip_ascii(self): + raw = ab_scramble("secret", seed=0x12345678) + self.assertEqual(ab_descramble(raw), "secret") + + def test_ab_scramble_roundtrip_utf16le(self): + raw = ab_scramble("secret", seed=0x12345678, encoding="utf-16-le") + self.assertEqual(ab_descramble(raw, encoding="utf-16-le"), "secret") + + def test_ab_scramble_random_seed(self): + raw1 = ab_scramble("secret") + raw2 = ab_scramble("secret") + # Random seeds should (almost certainly) differ, producing different + # ciphertexts for the same plaintext. + self.assertNotEqual(raw1, raw2) + self.assertEqual(ab_descramble(raw1), "secret") + self.assertEqual(ab_descramble(raw2), "secret") + + def test_ab_descramble_too_short_raises(self): + with self.assertRaises(ValueError): + ab_descramble(b"\x00\x00\x00") + + def test_nwrfc_password_tlv_roundtrip(self): + # Simulate an NWRFC password TLV (tag 0x0117): 4-byte seed + + # ab_scrambled UTF-16LE password. + scrambled = ab_scramble("s3cr3t!", seed=0xdeadbeef, encoding="utf-16-le") + data = build_tlv(0x0117, scrambled) + + tags = dict(parse_tlv(data)) + + self.assertIn(0x0117, tags) + self.assertEqual(ab_descramble(tags[0x0117], encoding="utf-16-le"), "s3cr3t!") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/saprfc_test.py b/tests/saprfc_test.py index bc28c15..268563c 100644 --- a/tests/saprfc_test.py +++ b/tests/saprfc_test.py @@ -15,6 +15,7 @@ # Martin Gallo (@martingalloar) # Code contributed by SecureAuth to the OWASP CBAS project +import struct import sys import unittest @@ -65,6 +66,42 @@ def test_rfc_version_six_appc_roundtrip(self): self.assertEqual(parsed.version, 6) self.assertEqual(parsed.func_type, 0xca) + def test_rfc_f_sap_send_receive_without_codepage(self): + # F_SAP_SEND (0xcb) with vector=F_V_RECEIVE and info3 lacking + # GW_WITH_CODE_PAGE leaves codepage_size2 as None. Dissection used to + # raise TypeError comparing None > 0 for the "repl" condition. + raw = bytearray() + raw += bytes([0x06]) # version + raw += bytes([0xcb]) # func_type + raw += bytes([0x03]) # protocol + raw += bytes([0x00]) # mode + raw += struct.pack("!H", 0) # uid + raw += struct.pack("!H", 0) # gw_id + raw += struct.pack("!H", 0) # err_len + raw += bytes([0x00]) # info2 + raw += bytes([0x01]) # trace_level + raw += struct.pack("!I", 0) # time + raw += bytes([0x00]) # info3 - no GW_WITH_CODE_PAGE + raw += struct.pack("!i", -1) # timeout + raw += bytes([0x00]) # info4 + raw += struct.pack("!I", 0) # seq_no + raw += struct.pack("!H", 0) # sap_param_len + raw += bytes([0x00]) # padd_appc + raw += struct.pack("!H", 0) # info - no SYNC_CPIC_FUNCTION + raw += bytes([0x08]) # vector - F_V_RECEIVE + raw += struct.pack("!I", 0) # appc_rc + raw += struct.pack("!I", 0) # sap_rc + raw += b"\x00" * 8 # conv_id + raw += b"\x05\x00\x00\x00" # anon_repl_sign for F_V_RECEIVE + raw += b"PAYLOADDATA1234" + + packet = SAPRFC(bytes(raw)) + + self.assertEqual(packet.version, 6) + self.assertEqual(packet.func_type, 0xcb) + self.assertIsNone(packet.codepage_size2) + self.assertIsNone(packet.repl) + def suite(): loader = unittest.TestLoader()