如何在自定义HTTP监听器中解析授权头的用户名密码实现身份验证

你当前遇到的连接异常并非携带Authorization头被主动拒绝，而是现有服务端代码仅完成了TCP连接握手，没有接收、解析HTTP请求，也没有返回合法的HTTP响应，导致客户端判定请求失败。要实现Basic Auth身份校验，可按以下逻辑修改服务端：

实现思路 #

• 客户端连接建立后，先读取完整的HTTP请求头内容
• 从请求头中匹配Authorization字段，Basic Auth的标准格式为Authorization: Basic <base64编码字符串>，编码字符串为用户名:密码拼接后的base64编码结果
• 对编码内容做base64解码后按:拆分，即可得到明文的用户名和密码
• 身份校验完成后返回符合HTTP规范的响应内容

完整可运行服务端代码 #

import socket
import base64

serverSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serverPort = 8080
# 新增端口复用配置，避免重启服务时报端口占用
serverSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serverSocket.bind(("127.0.0.1", serverPort))
serverSocket.listen(1)

# 预设合法的用户名密码
VALID_USER = "user"
VALID_PASS = "pass"

while True:
    print('Ready to serve...')
    try:
        connectionSocket, addr = serverSocket.accept()
        # 读取HTTP请求数据，单次读取1024字节足够覆盖普通GET请求头
        request_data = connectionSocket.recv(1024).decode("utf-8")
        # 拆分请求行和请求头
        headers = request_data.split("\r\n")
        auth_header = None
        for header in headers:
            if header.lower().startswith("authorization:"):
                auth_header = header
                break
        
        auth_pass = False
        username = ""
        if auth_header:
            # 提取Basic标识后的编码内容
            auth_value = auth_header.split(" ", 2)[-1]
            # 尝试base64解码并拆分用户名密码
            try:
                decode_str = base64.b64decode(auth_value).decode("utf-8")
                username, password = decode_str.split(":", 1)
                if username == VALID_USER and password == VALID_PASS:
                    auth_pass = True
            except:
                # 解码失败或格式错误，直接判定为认证失败
                pass
        
        # 构造返回HTTP响应
        if auth_pass:
            response = f"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nWelcome {username}!"
        else:
            # 认证失败返回401，触发客户端弹窗或重传认证信息
            response = "HTTP/1.1 401 Unauthorized\r\nWWW-Authenticate: Basic realm=\"Access to the site\"\r\nContent-Type: text/plain\r\n\r\nAuthentication required"
        
        connectionSocket.sendall(response.encode("utf-8"))
        # 短连接模式，响应发送完成后关闭连接
        connectionSocket.close()
    except Exception as e:
        print(f"Socket error occured for 127.0.0.1 {serverPort}: {e}")

测试说明 #

修改完服务端代码后启动服务，再运行你提供的客户端请求代码，服务端可正常解析出Authorization头中的用户名和密码，匹配成功后返回200响应，匹配失败返回401响应。

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