如何让Python开发的adBot接收PowerShell命令？

我在虚拟机中运行名为adBot的.exe程序，目前它可通过adClient脚本接收命令，在Active Directory中创建用户，这部分代码运行良好。但我希望让它能够接收PowerShell命令，当前使用rpyc和subprocess库实现连接。

adBot 代码 #

#adBot
import argparse
import rpyc
from rpyc.utils.server import ThreadedServer
import datetime
import subprocess
date_time = datetime.datetime.now()

class MonitorService(rpyc.Service):
    def on_connect(self, conn):
        print("Conectado {}".format(date_time))
    def on_disconnect(self, conn):
        print("Desconectado {}\n".format(date_time))
    def exposed_run_command(self,command):
        try:
            output = subprocess.check_output(command,shell=True) 
            print(output)
        except subprocess.CalledProcessError as Error:
            print(Error.returncode)
            print(Error.output)
def main():
    parser = argparse.ArgumentParser(description="Active Directory Bot")
    parser.add_argument("-port",type=int,help="Enter Custom Port Number")
    args = parser.parse_args()
    port = args.port
    print("Puerto Recuperado: " + str(port))
    if not port:
        port = 18821 
    print("Puerto Usado: "+ str(port))
 
    t = ThreadedServer(MonitorService,port=port)
    t.start()

AdClient 代码 #

#AdClient
import rpyc
DISABLE_user = 'disable'
ENABLE_user = 'enable'
DELETE_user = 'delete'
ADD_group = 'add'
REMOVE_group = 'remove'
AD_IP = "172.19.128.3"  # IP Directorio Activo
LOCAL = "localhost"
MV_SERVER = "192.168.74.128"  # IP  maquina Virtual
AD_BOT_PORT = 18821
domain_controller = "dc=TEST,dc=LOCAL"
users_ou = f"ou=ALL,ou=Employee,{domain_controller}"
groups_ou = f"ou=Employee_Groups,{domain_controller}"

def enviarComando(comando):
 
    try:
        con = rpyc.connect(MV_SERVER, AD_BOT_PORT)
        con.root.run_command(comando)
        print("ok")
    except Exception as ERR:
        print(f"Error en enviar comando {ERR}")
    except TimeoutError as TIMEOUT:
        print(f"Error en tiempo de espera: {TIMEOUT}")

def crearUsuario(nombres, apellidos,desc, oficina, numTelf, numCasa, mobileNum, titulo, departamento, compania, dirCasa, active=False):

    if active:
        deshabilitar = "no"
    else:
        deshabilitar = "yes"
    descripcion = desc+f" USUARIO CREADO EN EL DIRECTORIO ACTIVO {datetime.datetime.now()}"
    displayName = nombres + " " + apellidos
    defaultPassword = "P@tito.123"
    username = cda.defineUsername(nombres, apellidos)
    upn = cda.defineUPN(username)
    email = cda.defineCorreo(nombres,apellidos)
    dn = f'"cn={username},{users_ou}"'
    country = "EC"
    grupo = f'"cn=All,{groups_ou}"'# All grupo ya creado
    comando = 'dsadd user ' \
              f'{dn} ' \
              f'-samid "{username}" ' \
              f'-upn "{upn}" ' \
              f'-fn "{nombres}" ' \
              f'-ln "{apellidos}" ' \
              f'-display "{displayName}" ' \
              f'-empid "{NULL}" ' \
              f'-pwd {defaultPassword} ' \
              f'-desc "{descripcion}" ' \
              f'-office "{oficina}" ' \
              f'-tel "{numTelf}" ' \
              f'-email "{email}" ' \
              f'-hometel "{numCasa}" ' \
              f'-pager "{NULL}" ' \
              f'-mobile "{mobileNum}" ' \
              f'-fax "{NULL}" ' \
              f'-iptel "{NULL}" ' \
              f'-webpg "{NULL}" ' \
              f'-title "{titulo}" ' \
              f'-dept "{departamento}" ' \
              f'-company "{compania}" ' \
              f'-disabled {deshabilitar} ' \
              '-pwdneverexpires yes ' \
              '-mustchpwd yes ' \
              f'-memberof {grupo} ' \
              '-acctexpires never ' \
              ''
    #comando = "New-ADUser -name 'sales user' -GivenName Sales -Surname USER -SamAccountName SalesUser -UserPrincipalName saleuser@corp.lan"
    print(comando)
    enviarComando(comando)

要让adBot支持PowerShell命令，核心是确保subprocess调用PowerShell解释器而非默认的cmd，以下是具体修改方式：

1. 优化服务端命令执行逻辑 #

修改exposed_run_command方法，明确调用PowerShell并处理编码问题，避免输出乱码：

def exposed_run_command(self, command):
    try:
        # 指定调用PowerShell，输出采用UTF-8编码
        output = subprocess.check_output(
            ["powershell", "-Command", command],
            encoding='utf-8',
            stderr=subprocess.STDOUT
        )
        print(output)
        return output  # 返回执行结果给客户端，方便后续处理
    except subprocess.CalledProcessError as error:
        print(f"命令执行失败，返回码: {error.returncode}")
        print(f"错误输出: {error.output}")
        return error.output

2. 客户端发送PowerShell命令 #

直接构造PowerShell命令发送即可，比如修改crearUsuario中的示例命令：

# 替换原注释的命令为以下内容
comando = 'New-ADUser -name "sales user" -GivenName Sales -Surname USER -SamAccountName SalesUser -UserPrincipalName saleuser@corp.lan'
enviarComando(comando)

如果未修改服务端默认调用逻辑，客户端需要显式指定调用PowerShell：

comando = 'powershell -Command "New-ADUser -name \'sales user\' -GivenName Sales -Surname USER -SamAccountName SalesUser -UserPrincipalName saleuser@corp.lan"'
enviarComando(comando)

3. 注意事项 #

• 确保虚拟机环境已安装Active Directory PowerShell模块，保证AD命令可正常执行
• 命令中的引号需正确转义，避免语法错误
• 建议服务端返回执行结果，客户端可根据返回值判断命令执行状态

内容的提问来源于stack exchange，提问作者Cuervo 616