SCMR ChangeServiceConfig2W - Modifying Service Failure Actions - Not working?

[dotnetassembly]

Hi, I am trying to use the SCMR library to change the Service Failure Actions (SC_ACTIONS) of a Windows Service. I can't figure how to do it. Can someone provide a working example of doing that or let me know if there is a bug with the library?

I followed the test_scmr.py script from impacket for my testing. However, I can't seem to get the right data structure with rpc_x_bad_stub_data error message

Thank you.

Configuration

impacket version: 0.13.0.dev0+20250919.210843.8426ec99
Python version: 3.12
Target OS: 6.12.38+kali-rt-amd64 #1 SMP PREEMPT_RT Kali 6.12.38-1kali1 (2025-08-12) x86_64 GNU/Linux

Debug Output With Command String

i.e.

python test.py
[*] Connecting to 192.168.31.125...
[+] Connected and bound to SCMR.
RChangeServiceConfig2WResponse
ErrorCode:                       0


Traceback (most recent call last):
  File "/home/user/test.py", line 137, in test_RChangeServiceConfig2W
    resp = dce.request(request)
           ^^^^^^^^^^^^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.8/lib/python3.12/site-packages/impacket/dcerpc/v5/rpcrt.py", line 1415, in request
    answer = self.recv()
             ^^^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.8/lib/python3.12/site-packages/impacket/dcerpc/v5/rpcrt.py", line 1883, in recv
    raise DCERPCException(rpc_status_codes[status_code])
impacket.dcerpc.v5.rpcrt.DCERPCException: rpc_x_bad_stub_data
rpc_x_bad_stub_data
.
----------------------------------------------------------------------
Ran 1 test in 0.061s

OK

Test Script

import sys
import socket
import time
import unittest
from struct import unpack

from impacket.dcerpc.v5 import scmr,transport 
from impacket.dcerpc.v5.ndr import NULL
from impacket.crypto import encryptSecret
from impacket.uuid import string_to_bin
from impacket import ntlm

class SCMRTests(unittest.TestCase):

    def get_service_handle(self,dce):
        lpMachineName = 'DUMMY\x00'
        lpDatabaseName = 'ServicesActive\x00'
        desiredAccess = scmr.SERVICE_START | scmr.SERVICE_STOP | scmr.SERVICE_CHANGE_CONFIG | scmr.SERVICE_QUERY_CONFIG | scmr.SERVICE_QUERY_STATUS | scmr.SERVICE_ENUMERATE_DEPENDENTS | scmr.SC_MANAGER_ENUMERATE_SERVICE
        resp = scmr.hROpenSCManagerW(dce, lpMachineName, lpDatabaseName, desiredAccess)
        scHandle = resp['lpScHandle']
        return scHandle

    def open_or_create_service(self, dce, scHandle, service_name, display_name, binary_path_name):

        try:
            desiredAccess = scmr.SERVICE_ALL_ACCESS
            resp = scmr.hROpenServiceW(dce, scHandle, service_name, desiredAccess)
            resp.dump()
            return resp['lpServiceHandle']
        except scmr.DCERPCSessionError as e:
            if e.get_error_code() != 0x424:
                raise

        dwDesiredAccess = scmr.SERVICE_ALL_ACCESS
        dwServiceType = scmr.SERVICE_WIN32_OWN_PROCESS
        dwStartType = scmr.SERVICE_DEMAND_START
        dwErrorControl = scmr.SERVICE_ERROR_NORMAL
        lpLoadOrderGroup = NULL
        lpdwTagId = NULL
        lpDependencies = NULL
        dwDependSize = 0
        lpServiceStartName = NULL
        lpPassword = NULL
        dwPwSize = 0
        resp = scmr.hRCreateServiceW(dce, scHandle, service_name, display_name, dwDesiredAccess,
                                        dwServiceType, dwStartType, dwErrorControl, binary_path_name,
                                        lpLoadOrderGroup, lpdwTagId, lpDependencies, dwDependSize,
                                        lpServiceStartName, lpPassword, dwPwSize)
        return resp['lpServiceHandle']


    def changeServiceAndQuery2(self, dce, info, changeDone):
        serviceHandle = info['hService']
        dwInfoLevel = info['Info']['Union']['tag']
        cbBuffSize = 0
        request = scmr.RQueryServiceConfig2W()
        request['hService'] = serviceHandle
        request['dwInfoLevel'] = dwInfoLevel
        request['cbBufSize'] = cbBuffSize
        try:
            resp = dce.request(request)
        except scmr.DCERPCSessionError as e:
            if str(e).find('ERROR_INSUFFICIENT_BUFFER') <= 0:
                raise
            else: 
                resp = e.get_packet()

        request['cbBufSize'] = resp['pcbBytesNeeded'] 
        resp = dce.request(request)
        arrayData = b''.join(resp['lpBuffer'])
        if dwInfoLevel == 1:
            self.assertEqual(arrayData[4:].decode('utf-16le'), changeDone)
        elif dwInfoLevel == 2:
            offset = unpack('<L', arrayData[4:][:4])[0]
            self.assertEqual(arrayData[offset:][:len(changeDone)*2].decode('utf-16le'), changeDone)
        elif dwInfoLevel == 3:
            self.assertEqual(unpack('<L', arrayData)[0], changeDone)
        elif dwInfoLevel == 4:
            self.assertEqual(unpack('<L', arrayData)[0], changeDone)
        elif dwInfoLevel == 5:
            self.assertEqual(unpack('<L', arrayData)[0], changeDone)
        elif dwInfoLevel == 6:
            changeDone = bytes(changeDone).decode('utf-16le')
            self.assertEqual(arrayData[4:].decode('utf-16le'), changeDone)
        elif dwInfoLevel == 7:
            self.assertEqual(unpack('<L', arrayData)[0], changeDone)

    def test_RChangeServiceConfig2W(self):
        # --- Connection Details ---
        target_ip = '192.168.x.x'
        username = 'Administrator'
        password = ''
        domain = 'WORKGROUP'

        # --- Main Logic ---
        dce = None
        # 1. Set up the RPC transport
        string_binding = f'ncacn_np:{target_ip}[\\pipe\\svcctl]'
        rpc_transport = transport.DCERPCTransportFactory(string_binding)
        rpc_transport.set_credentials(username, password, domain)
        dce = rpc_transport.get_dce_rpc()

        # 2. Connect and bind to the SCMR interface
        print(f"[*] Connecting to {target_ip}...")
        dce.connect()
        dce.bind(scmr.MSRPC_UUID_SCMR)
        print("[+] Connected and bound to SCMR.")

        # 3. Open the Service Control Manager
        scHandle = self.get_service_handle(dce)
        # 4. Open the target service
        newHandle = self.open_or_create_service(dce,scHandle, 'TESTSVC\x00', 'DisplayName\x00', 'binaryPath\x00')
        error = False
        try:
            request = scmr.RChangeServiceConfig2W()
            request['hService'] = newHandle
            request['Info']['dwInfoLevel'] = 1
            request['Info']['Union']['tag'] = 1
            request['Info']['Union']['psd']['lpDescription'] = 'betobeto\x00'
            resp = dce.request(request)
            resp.dump()
            self.changeServiceAndQuery2(dce, request, request['Info']['Union']['psd']['lpDescription'])

            action1 = scmr.SC_ACTION()
            action1['Type'] = scmr.SC_ACTION_RUN_COMMAND
            action1['Delay'] = 60000 

            actions_container = scmr.SC_ACTIONS()
            actions_container['Data'].append(action1)

            request['Info']['dwInfoLevel'] = 2
            request['Info']['Union']['tag'] = 2
            request['Info']['Union']['psfa']['lpRebootMsg'] = 'rebootMsg\00'
            request['Info']['Union']['psfa']['lpCommand'] = 'lpCommand\00'
            request['Info']['Union']['psfa']['cActions'] = 1
            request['Info']['Union']['psfa']['lpsaActions'] = actions_container
            resp = dce.request(request)
            resp.dump()
            self.changeServiceAndQuery2(dce, request, request['Info']['Union']['psfa']['lpsaActions'])


        except Exception as e:
            import traceback
            traceback.print_exc()
            print(e)
            error = True
            pass

        scmr.hRDeleteService(dce, newHandle)
        scmr.hRCloseServiceHandle(dce, newHandle)
        scmr.hRCloseServiceHandle(dce, scHandle)

# Process command-line arguments.
if __name__ == "__main__":
    unittest.main()

Additional context

Space for additional context, investigative results, suspected issue.