如何用Python Socket实现远程电脑的实时命令执行？（Python3.7.4）

Hey there! Since you already have a solid grasp of Python socket basics—text transfer, port forwarding, and WLAN byte transmission—building a remote real-time terminal that mimics local os.system() is totally within your reach. Let’s walk through a practical implementation tailored for Python 3.7.4.

Core Concept #

The idea is straightforward:

1. Client: Sends command strings (e.g., ls, ipconfig) to the remote server.
2. Server: Receives the command, executes it locally, captures both stdout and stderr, then sends the output back to the client.
3. Loop: Maintain a continuous connection so you can input commands and get real-time results until you choose to exit.

We’ll use subprocess instead of os.system() because it lets us capture command output (not just exit codes), which is essential for a terminal-like experience.

Step 1: Server Implementation #

This script runs on the remote machine (the one you want to control) and listens for incoming commands.

import socket
import subprocess
import sys

def execute_command(cmd: str) -> bytes:
    """Execute a command and return its stdout/stderr as bytes"""
    try:
        # Use Popen to capture output; shell=True supports complex commands
        process = subprocess.Popen(
            cmd,
            shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,  # Merge stderr into stdout for unified output
            stdin=subprocess.PIPE
        )
        # Read all output (works for non-interactive commands)
        output, _ = process.communicate()
        return output
    except Exception as e:
        return f"Error executing command: {str(e)}".encode('utf-8')

def start_server(host: str = '0.0.0.0', port: int = 9999) -> None:
    # Create TCP socket
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # Allow reusing the port quickly after shutdown
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    try:
        server_socket.bind((host, port))
        server_socket.listen(5)
        print(f"Server listening on {host}:{port}")
        
        # Accept incoming connection
        client_socket, client_addr = server_socket.accept()
        print(f"Connected to {client_addr[0]}:{client_addr[1]}")
        
        while True:
            # Receive command from client (adjust buffer size if needed)
            cmd = client_socket.recv(1024).decode('utf-8').strip()
            
            if not cmd or cmd.lower() == 'exit':
                print("Client disconnected.")
                break
            
            print(f"Executing command: {cmd}")
            output = execute_command(cmd)
            
            # Send output safely: first send length, then the data
            # This avoids "sticky packet" issues with large outputs
            output_len = len(output)
            client_socket.sendall(output_len.to_bytes(4, byteorder='big'))
            client_socket.sendall(output)
            
    except Exception as e:
        print(f"Server error: {str(e)}")
    finally:
        client_socket.close()
        server_socket.close()

if __name__ == "__main__":
    start_server()

Step 2: Client Implementation #

This script runs on your local machine and acts as the terminal interface.

import socket
import sys

def start_client(host: str, port: int = 9999) -> None:
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    try:
        client_socket.connect((host, port))
        print(f"Connected to remote server {host}:{port}")
        print("Type 'exit' to quit.\n")
        
        while True:
            # Get user input command
            cmd = input("$ ").strip()
            
            if cmd.lower() == 'exit':
                client_socket.send(cmd.encode('utf-8'))
                break
            
            if not cmd:
                continue
            
            # Send command to server
            client_socket.send(cmd.encode('utf-8'))
            
            # Receive output length first
            len_data = client_socket.recv(4)
            if not len_data:
                print("Server disconnected unexpectedly.")
                break
            
            output_len = int.from_bytes(len_data, byteorder='big')
            # Receive the full output in chunks
            output = b""
            while len(output) < output_len:
                chunk = client_socket.recv(min(1024, output_len - len(output)))
                if not chunk:
                    break
                output += chunk
            
            # Print result, handle encoding errors gracefully
            print(output.decode('utf-8', errors='replace'))
            
    except Exception as e:
        print(f"Client error: {str(e)}")
    finally:
        client_socket.close()

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: python client.py <server_ip>")
        sys.exit(1)
    start_client(sys.argv[1])

Key Notes for Python 3.7.4 #

• Encoding: We use utf-8 for string/byte conversion, which works for most cases. Adjust the encoding if you need to handle non-UTF-8 outputs (e.g., legacy systems).
• Subprocess Compatibility: subprocess.Popen behaves consistently in 3.7.4; we avoid newer shorthands like capture_output=True to keep it explicit and compatible.
• Socket Stability: The SO_REUSEADDR option prevents "address already in use" errors when restarting the server quickly.

Enhancements for a Better Terminal Experience #

• Real-Time Output: For long-running commands (e.g., ping -t), modify the server to read output line-by-line with process.stdout.readline() and send each line immediately instead of waiting for the command to finish.
• Command Injection Safety: If sharing this terminal with others, avoid shell=True and use shell=False with a list of command arguments (e.g., ["ls", "-l"]). For personal use, shell=True is fine for complex commands.
• Interactive Commands: Tools like vim or top won’t work with this basic setup—look into the pty module on the server side to support interactive apps.

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