Improved test for NDR + added timeout for debugger test (if pytest-timeout present)

samples/event_log/wevtutil.py

@@ -7,6 +7,7 @@ import struct
 import pickle
 import logging
 import binascii
+import traceback
 
 from io import BytesIO
 from collections import namedtuple
@@ -73,21 +74,21 @@ class RealtimeEventLogger(RealtimeEventLoggerBase):
             self.any_keywords |= keyword
 
         logging.debug("events KeywordsAny : 0x%x" % self.any_keywords)
-       
+
 
         # Create a custom realtime ETW trace for printing out events
         self.etw_trace = windows.system.etw.open_trace(self.etw_trace_name)
 
 
     def start_trace(self):
-
+        self.etw_trace.stop(soft=True)
         self.etw_trace.start()
 
         # We can't configure etw trace if it's not started previously
         self.etw_trace.enable_ex(
-            self.publisher_guid.to_string(), 
-            flags=0, 
-            level=0xff, 
+            self.publisher_guid.to_string(),
+            flags=0,
+            level=0xff,
             any_keyword=self.any_keywords
         )
 
@@ -113,7 +114,7 @@ class RealtimeEventLogger(RealtimeEventLoggerBase):
 
                 # Deserialize event.user_data based on the event_metadata xml template
                 message_data = event.user_data
-                message_params = self.parse_user_data(event_metadata.template, message_data)
+                message_params = self.parse_user_data(event_metadata, message_data)
                 logging.debug("message params : %s" %  message_params)
 
 
@@ -125,13 +126,27 @@ class RealtimeEventLogger(RealtimeEventLoggerBase):
                         raise
                     return
 
+                # import pdb;pdb.set_trace()
+                xx = tuple(event_log.ImprovedEVT_VARIANT.from_value(x.value) for x in message_params)
+                # xx = tuple(event_log.ImprovedEVT_VARIANT.from_value(x.value) for x in message_params)
+                res = ctypes.c_buffer(0x1000)
+                res_size = gdef.DWORD()
+                yolo_ptr = (gdef.EVT_VARIANT * len(xx))(*xx)
+                # import pdb;pdb.set_trace()
+                windows.winproxy.EvtFormatMessage(self.publisher.metadata, None, event_metadata.message_id, len(message_params), yolo_ptr, gdef.EvtFormatMessageId, 0x1000, ctypes.cast(res, gdef.LPCWSTR), res_size)
+                str = res[:res_size.value * 2].decode("utf-16-le")
+                print("=== MINE ===")
+                print(str)
+
+                print("=== REAL ===")
                 # "sprintf" the message using the event format message as well as the deserialized elements
                 event_message = self.format_event_log_message(template_message, message_params)
-                
                 print(event_message)
+                import pdb;pdb.set_trace()
 
         except Exception as unke:
             print("Unhandled exception in Evtlogger.process_event : %s" % unke)
+            traceback.print_tb(sys.exc_info()[2])
             sys.exit(0)  # Exiting on unknown error, since this is the only way to have some control
         finally:
             pass
@@ -181,10 +196,10 @@ class RealtimeEventLogger(RealtimeEventLoggerBase):
         elif in_type == "win:Binary":
             if not length:
                 raise ValueError(" param_in_type (%s) cannot be used with a null length value" % in_type)
-            
-            # TODO : we should return the raw bytes buffer, since get_param_str_format is too crude 
+
+            # TODO : we should return the raw bytes buffer, since get_param_str_format is too crude
             #        to properly display win:SocketAddress parameters
-            value = binascii.hexlify(stream.read(length)) 
+            value = binascii.hexlify(stream.read(length))
 
         elif in_type == "win:GUID":
             guid_data = struct.unpack("IHHBBBBBBBB", stream.read(16))
@@ -214,66 +229,51 @@ class RealtimeEventLogger(RealtimeEventLoggerBase):
 
         return PYTHON_FORMAT_DICT[param_out_type]
 
-    def parse_user_data(self, template, data):
-        """ 
+    def parse_user_data(self, event_metadata, data):
+        """
         Deserialize event.user_data based on the associated publisher's template.
         Return a list of ParsedElement(Name:string, Value:py_object, Type:py_type).
         """
-            
-        # xml.dom.minidom.parseString raise an error on parseString() if the xml template is empty
-        if not len(template):
+        stream = BytesIO(data)
+        event_items = event_metadata.event_data
+        params = []
+        if not event_items:
             return []
 
-        stream = BytesIO(data)
-        xmltemplate = xml.dom.minidom.parseString(template)
-        
-        params = []
         context = {} # saving parsed items for "count" elements
-
-        # xmltemplate.getElementsByTagName("data") return data node within <struct> decl, so we can't use it
-        direct_data_nodes = filter(lambda n: n.nodeType == 1 and n.tagName == "data", xmltemplate.childNodes[0].childNodes)
-
-        for (i,param_data) in enumerate(direct_data_nodes):
-
-            param_name = param_data.attributes["name"].value
-            param_in_type = param_data.attributes["inType"].value
-            param_out_type = param_data.attributes["outType"].value
-
+        for (i, param_data) in enumerate(event_items):
             # Some param are repeating, and "count" refers to the variable holding the number of repetitions
-            param_count = param_data.attributes.get("count", None)
-            if param_count != None:
-                param_count = context[param_count.value] # must be already set
-
-            # Some param (win:Binary) have a length attribute
-            param_length = param_data.attributes.get("length", None)
-            if param_length != None:
-                param_length = context[param_length.value] # must be already set
-            
-
-            # Parse element
-            if param_count != None:
-
-                # ignoring element with value count of 0
+            param_in_type = param_data["inType"]
+            param_out_type = param_data["outType"]
+            param_name = param_data["name"]
+            param_count = param_data.get("count", None)
+            if param_count:
+                param_count = context[param_count] # must be already set
                 if param_count == 0:
                     continue
 
-                value = [ self.parse_element(param_in_type, stream, param_length) for c in range(param_count) ]
+            # Some param (win:Binary) have a length attribute
+            param_length = param_data.get("length", None)
+            if param_length:
+                param_length = context[param_length] # must be already set
+
+            # Parse element
+            if param_count:
+                value = [self.parse_element(param_in_type, stream, param_length) for c in range(param_count)]
             else:
                 value = self.parse_element(param_in_type, stream, param_length)
 
-            # Get python string formating 
+            context[param_name] = value
             format_type = self.get_param_str_format(param_out_type)
 
-            context[param_name] = value
-
-            logging.debug(ParsedElement(i, param_name,value, format_type))
-            params.append(ParsedElement(i, param_name,value, format_type))
+            logging.debug(ParsedElement(i, param_name, value, format_type))
+            params.append(ParsedElement(i, param_name, value, format_type)) # Yield ?
 
         return params
 
     def lookup_event_metadata(self, publisher, event_id):
         matching_events_metadata = list(filter(lambda event_meta: event_meta.id == event_id, publisher.metadata.events_metadata))
-        
+
         if not len(matching_events_metadata):
             return None
 
@@ -283,28 +283,28 @@ class RealtimeEventLogger(RealtimeEventLoggerBase):
         return matching_events_metadata[0]
 
     def format_event_log_message(self, template, event_args):
-        
+
         py_template = ""
         last_span = (0,0)
-        
+
         # Convert message template to python string formating
         # e.g. : "ParseError: HResult: %1, Error: %2." into "ParseError: HResult: {arg0:x}, Error: {arg1:d}."
         pattern = re.compile(r"%(\d)")
         for match in re.finditer(pattern, template):
-        
+
             arg_id = int(match.groups()[0]) - 1 # event's template message index params from 1 to N, wtf
             span = match.span()
             str_format = ""
-        
+
             str_format = "{a%d:%s}" % (arg_id, event_args[arg_id].format)
-        
+
             py_template += template[last_span[1]:span[0]] + str_format
             last_span = span
-        
+
         py_template += template[last_span[1]:]
-        
+
         logging.debug(py_template)
-        
+
         # string formating using Python .format()
         events_kwargs =  {"a%d" % (x.index) : x.value for x in event_args}
         logging.debug(events_kwargs)
@@ -335,7 +335,7 @@ def format_channel_metadata(publisher_metadata, channel_metadata, args):
         "    flags: {channel.flags:d}",
         "    message: {channel_message:s}",
     ]).format(
-        channel=channel_metadata, 
+        channel=channel_metadata,
         channel_message=get_message(publisher_metadata, channel_metadata.message_id, args.gm)
     )
 
@@ -345,9 +345,9 @@ def format_level_metadata(publisher_metadata, level_metadata, args):
         "  level:",
         "    name: {level.name:s}",
         "    value: {level.value:d}",
-        "    message: {level_message:s}", 
+        "    message: {level_message:s}",
     ]).format(
-        level=level_metadata, 
+        level=level_metadata,
         level_message=get_message(publisher_metadata, level_metadata.message_id, args.gm)
     )
 
@@ -359,9 +359,9 @@ def format_opcode_metadata(publisher_metadata, opcode_metadata, args):
         "    value: {opcode.value:d}",
         #"      task: {opcode.task:d}",          # TODO
         #"      opcode: {opcode.task_value:d}",  # TODO
-        "    message: {opcode_message:s}", 
+        "    message: {opcode_message:s}",
     ]).format(
-        opcode=opcode_metadata, 
+        opcode=opcode_metadata,
         opcode_message=get_message(publisher_metadata, opcode_metadata.message_id, args.gm)
     )
 
@@ -371,10 +371,10 @@ def format_task_metadata(publisher_metadata, task_metadata, args):
         "  task:",
         "    name: {task.name:s}",
         "    value: {task.value:d}",
-        "    eventGUID: {task.event_guid:s}",   
-        "    message: {task_message:s}",        
+        "    eventGUID: {task.event_guid:s}",
+        "    message: {task_message:s}",
     ]).format(
-        task=task_metadata, 
+        task=task_metadata,
         task_message=get_message(publisher_metadata, task_metadata.message_id, args.gm)
     )
 
@@ -384,9 +384,9 @@ def format_keyword_metadata(publisher_metadata, keyword_metadata, args):
         "  keyword:",
         "    name: {keyword.name:s}",
         "    mask: {keyword.value:x}",
-        "    message: {keyword_message:s}",  
+        "    message: {keyword_message:s}",
     ]).format(
-        keyword=keyword_metadata, 
+        keyword=keyword_metadata,
         keyword_message=get_message(publisher_metadata, keyword_metadata.message_id, args.gm)
     )
 
@@ -401,21 +401,21 @@ def format_event_metadata(publisher_metadata, event_metadata, args):
         "    level: {event.level:d}",
         "    task: {event.task:d}",
         "    keywords: 0x{event.keyword:016x}",
-        "    message: {event_message:s}" 
+        "    message: {event_message:s}"
     ]).format(
-        event=event_metadata, 
+        event=event_metadata,
         event_message=get_message(publisher_metadata, event_metadata.message_id, args.gm)
     )
 
 def enum_publishers(args):
     """ enum-publishers verb implementation """
     manager = event_log.EvtlogManager()
-    for publisher in sorted(list(manager.publishers), key=lambda pub:pub.name.lower()):    
-        print(publisher.name)        
+    for publisher in sorted(list(manager.publishers), key=lambda pub:pub.name.lower()):
+        print(publisher.name)
 
 def get_publisher(args):
     """ get-publisher verb implementation """
-    
+
     manager = event_log.EvtlogManager()
     publisher = manager.open_publisher(args.publisher_name)
 
@@ -492,7 +492,7 @@ def get_publisher(args):
 
     publisher_infos += "\n"
     print(publisher_infos)
-    
+
 
 def main(args):
 
@@ -519,7 +519,7 @@ if __name__ == '__main__':
 
     # we can't express shorthands easily like ep for enum-publishers since only Python3's argparse surpport parser "aliases"
     enum_publishers_parser = action_parsers.add_parser("enum-publishers", help="enum-publishers verb")
-    
+
     get_publisher_parser = action_parsers.add_parser("get-publisher", help="get-publisher verb")
     get_publisher_parser.add_argument("publisher_name", type=str, help="registered publisher name")
     get_publisher_parser.add_argument("--ge", action="store_true", help="get event metadata")

samples/service/service_demo.py

@@ -6,7 +6,7 @@ for service in windows.system.services[:3]:
     print(" * {0}".format(service))
 print("")
 
-TARGET_SERVICE = "TapiSrv"
+TARGET_SERVICE = b"TapiSrv"
 print("Retriving service <{0}>".format(TARGET_SERVICE))
 service = windows.system.services[TARGET_SERVICE]
 print("{0}".format(service))

tests/test_debugger.py

@@ -4,6 +4,7 @@ import ctypes
 import os
 
 import windows
+import windows.debug
 import windows.generated_def as gdef
 import windows.native_exec.simple_x86 as x86
 import windows.native_exec.simple_x64 as x64
@@ -24,7 +25,9 @@ else:
 
 yolo = generate_pop_and_exit_fixtures([pop_proc_32, pop_proc_64], ids=["proc32dbg", "proc64dbg"], dwCreationFlags=gdef.CREATE_SUSPENDED)
 
+DEFAULT_DEBUGGER_TIMEOUT = 10
 
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_init_breakpoint_callback(proc32_64_debug):
     """Checking that the initial breakpoint call `on_exception`"""
     class MyDbg(windows.debug.Debugger):
@@ -41,6 +44,8 @@ def get_debug_process_ndll(proc):
     ntdll_addr = proc.query_memory(proc_pc).AllocationBase
     return windows.pe_parse.GetPEFile(ntdll_addr, target=proc)
 
+
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_simple_standard_breakpoint(proc32_64_debug):
     """Check that a standard Breakpoint method `trigger` is called with the correct informations"""
     class TSTBP(windows.debug.Breakpoint):
@@ -55,6 +60,7 @@ def test_simple_standard_breakpoint(proc32_64_debug):
     d.add_bp(TSTBP(LdrLoadDll))
     d.loop()
 
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_simple_hwx_breakpoint(proc32_64_debug):
     """Test that simple HXBP are trigger"""
 
@@ -71,7 +77,7 @@ def test_simple_hwx_breakpoint(proc32_64_debug):
     d.loop()
 
 
-
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_multiple_hwx_breakpoint(proc32_64_debug):
     """Checking that multiple succesives HXBP are properly triggered"""
     class TSTBP(windows.debug.HXBreakpoint):
@@ -103,6 +109,7 @@ def test_multiple_hwx_breakpoint(proc32_64_debug):
     assert TSTBP.COUNTER == 4
 
 
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_four_hwx_breakpoint_fail(proc32_64_debug):
     """Check that setting 4HXBP in the same thread fails"""
     # print("test_four_hwx_breakpoint_fail {0}".format(proc32_64_debug))
@@ -129,6 +136,7 @@ def test_four_hwx_breakpoint_fail(proc32_64_debug):
     assert "DRx" in e.value.args[0]
 
 
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_hwx_breakpoint_are_on_all_thread(proc32_64_debug):
     """Checking that HXBP without target are set on all threads"""
     class MyDbg(windows.debug.Debugger):
@@ -163,6 +171,7 @@ def test_hwx_breakpoint_are_on_all_thread(proc32_64_debug):
     assert TSTBP.COUNTER == 2
 
 
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 @pytest.mark.parametrize("bptype", [windows.debug.Breakpoint, windows.debug.HXBreakpoint])
 def test_simple_breakpoint_name_addr(proc32_64_debug, bptype):
     """Check breakpoint address resolution for format dll!api"""
@@ -184,6 +193,7 @@ def test_simple_breakpoint_name_addr(proc32_64_debug, bptype):
 
 from . import dbg_injection
 
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_hardware_breakpoint_name_addr(proc32_64_debug):
     """Check that name addr in HXBP are trigger in all threads"""
     class TSTBP(windows.debug.HXBreakpoint):
@@ -208,7 +218,7 @@ def test_hardware_breakpoint_name_addr(proc32_64_debug):
     # Code that will load wintrust !
     d.loop()
 
-
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_single_step(proc32_64_debug):
     """Check that BP/dbg can trigger single step and that instruction follows"""
     NB_SINGLE_STEP = 3
@@ -240,6 +250,7 @@ def test_single_step(proc32_64_debug):
     for i in range(NB_SINGLE_STEP):
         assert MyDbg.DATA[i] == addr + 1 + i
 
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 @pytest.mark.parametrize("bptype", [windows.debug.Breakpoint, windows.debug.HXBreakpoint])
 def test_single_step_from_bp(proc32_64_debug, bptype):
     """Check that HXBPBP/dbg can trigger single step"""
@@ -276,7 +287,7 @@ def test_single_step_from_bp(proc32_64_debug, bptype):
 
 # MEMBP
 
-
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_memory_breakpoint_write(proc32_64_debug):
     """Check MemoryBP WRITE"""
     class TSTBP(windows.debug.MemoryBreakpoint):
@@ -323,7 +334,7 @@ def test_memory_breakpoint_write(proc32_64_debug):
     # Used to verif we actually called the Breakpoints for the good addresses
     assert TSTBP.COUNTER == 2
 
-
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_memory_breakpoint_exec(proc32_64_debug):
     """Check MemoryBP EXEC"""
     NB_NOP_IN_PAGE = 3
@@ -354,6 +365,7 @@ def test_memory_breakpoint_exec(proc32_64_debug):
 # breakpoint remove
 import threading
 
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 @python_injection
 @pytest.mark.parametrize("bptype", [windows.debug.FunctionParamDumpHXBP, windows.debug.FunctionParamDumpBP])
 def test_standard_breakpoint_self_remove(proc32_64_debug, bptype):
@@ -382,6 +394,7 @@ def test_standard_breakpoint_self_remove(proc32_64_debug, bptype):
     assert data >= set([u"FILENAME1", u"FILENAME2"])
     assert u"FILENAME3" not in data
 
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 @python_injection
 @pytest.mark.parametrize("bptype", [windows.debug.FunctionParamDumpHXBP, windows.debug.FunctionParamDumpBP])
 def test_standard_breakpoint_remove(proc32_64_debug, bptype):
@@ -442,6 +455,7 @@ def get_generate_write_at_for_proc(target):
             return res.get_code()
     return generate_write_at
 
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_mem_breakpoint_remove(proc32_64_debug):
     data = []
     generate_read_at = get_generate_read_at_for_proc(proc32_64_debug)
@@ -469,7 +483,7 @@ def test_mem_breakpoint_remove(proc32_64_debug):
     d.loop()
     assert data == [data_addr, data_addr + 4]
 
-
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_mem_breakpoint_self_remove(proc32_64_debug):
     data = []
     generate_read_at = get_generate_read_at_for_proc(proc32_64_debug)
@@ -499,7 +513,7 @@ def test_mem_breakpoint_self_remove(proc32_64_debug):
     assert data == [data_addr, data_addr + 4]
 
 
-
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_read_write_bp_same_page(proc32_64_debug):
     data = []
     generate_read_at = get_generate_read_at_for_proc(proc32_64_debug)
@@ -536,7 +550,7 @@ def test_read_write_bp_same_page(proc32_64_debug):
 
     assert data == expected_result
 
-
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_exe_in_module_list(proc32_64_debug):
     class MyDbg(windows.debug.Debugger):
         def on_exception(self, exception):
@@ -550,7 +564,7 @@ def test_exe_in_module_list(proc32_64_debug):
     d = MyDbg(proc32_64_debug)
     d.loop()
 
-
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_bp_exe_by_name(proc32_64_debug):
     class TSTBP(windows.debug.Breakpoint):
         COUNTER = 0
@@ -574,6 +588,7 @@ def test_bp_exe_by_name(proc32_64_debug):
     assert TSTBP.COUNTER == 1
 
 
+@pytest.mark.timeout(DEFAULT_DEBUGGER_TIMEOUT)
 def test_keyboardinterrupt_when_bp_event(proc32_64_debug, monkeypatch):
     class ShouldNotTrigger(windows.debug.Breakpoint):
         COUNTER = 0

tests/test_ndr.py

@@ -6,6 +6,7 @@ from .pfwtest import *
 from tests.test_rpc import UACParameters
 
 # Memo: Padding byte is 'P' in ndr.py
+#   - UPTR value will be 0x01010101 * field_pos
 
 # 20 Bytes structures alignes on 4 butees
 DoubleDwordStructure = ndr.make_structure([ndr.NdrLong] * 5)
@@ -27,6 +28,30 @@ InternalAlignementStructure = ndr.make_structure([ndr.NdrShort, ndr.NdrByte, ndr
      #byte 	bfield5;
 # }InternalAlignementStructure;
 
+class ComplexAlignementStructure(ndr.NdrStructure):
+    MEMBERS = [
+        ndr.NdrByte,
+        ndr.NdrUniquePTR(ndr.NdrByte),
+        ndr.NdrUniquePTR(ndr.NdrLong),
+        ndr.NdrByte,
+        ndr.NdrUniquePTR(ndr.NdrHyper),
+        ndr.NdrUniquePTR(ndr.NdrByte)
+    ]
+
+# Struct with Pointer & Pointed alignement
+# Pack format ah follow (U for unique PTR bytes):
+target = "APPP\x01\x02\x03\x04\x05\x06\x07\x08\x44PPP\x09\x0a\x0b\x0c\x0d\x0e\x0f\x10BPPPCCCCPPPPEEEEEEEEF" + "X" * 100
+#    BPPPUUUUUUUUBPPPUUUUUUUUBPPPLLLLPPPPHHHHHHHHB
+# IDL Code:
+#   struct tstndr {
+#       byte x;
+#       [unique] byte* a;
+#       [unique] long* b;
+#       byte y;
+#       [unique] hyper* c;
+#       [unique] byte* d;
+#   };
+
 
 # NdrObject, Values, result
 NDR_PACK_TEST_CASE = [
@@ -69,9 +94,25 @@ NDR_PACK_TEST_CASE = [
         (0x0101, [0x4141, 0x42, 0x43434343, 0x44, 0x4545, 0x46], 0x4747),
         # Verified with an actual RPC server
         b"\x01\x01PPAABPCCCCDPEEFPGG"),
+    # Complex struct alignement
+    # A struct with both primitive field & pointer to primitiv field of various size
+    (ComplexAlignementStructure,
+        [0x41, 0x42, 0x43434343, 0x44, 0x4545454545454545, 0x46],
+        b"APPP\x02\x02\x02\x02\x03\x03\x03\x03\x44PPP\x05\x05\x05\x05\x06\x06\x06\x06BPPPCCCCEEEEEEEEF"),
+
+    # Complex struct alignement with nesting (due to uniqueptr at the start)
+    # The last hyper is not aligned/pad like previous test due to the leading UniquPTR
+    # This is the proof that NDR packing cannot be in "context-free" sub function and must share a state
+    # This test fails for now (0.6) and I don't know if I will implem the full NDR logic someday
+    (ndr.NdrUniquePTR(ComplexAlignementStructure),
+        [0x41, 0x42, 0x43434343, 0x44, 0x4545454545454545, 0x46],
+        b"\x01\x01\x01\x01APPP\x02\x02\x02\x02\x03\x03\x03\x03\x44PPP\x05\x05\x05\x05\x06\x06\x06\x06BPPPCCCCPPPPEEEEEEEEF")
 
 ]
 
+
+
+
 @pytest.mark.parametrize("ndrobj, values, result", NDR_PACK_TEST_CASE)
 def test_ndr_packing(ndrobj, values, result):
      assert ndrobj.pack(values) == result