Python Socket如何接收完整字节数据?现有代码优化咨询
First, let's recap your original code—this is a solid starting point for raw socket communication, perfect for understanding low-level network behavior in pen testing:
import socket import ssl def http_socket(domain='www.google.com', port=80): client = socket.socket() host = socket.gethostbyname(domain) client.connect((host, port)) client.sendall("GET /\r\n") response = client.recv(10000) return response
The core issue with client.recv(10000) is that TCP is a stream-based protocol, not message-based. This means recv() only grabs whatever bytes are available in the buffer at that moment—up to 10k here. If the server sends more data (which most modern HTTP responses do), you'll only get a partial chunk. For pen testing, this is critical because incomplete responses can hide error codes, misconfiguration details, or sensitive data you need to analyze.
How to Fix It: Receive the Full Response
To get the complete response, we need to handle two common HTTP response types: those with a Content-Length header, and chunked responses (marked with Transfer-Encoding: chunked). Here's a modified version of your code that keeps raw socket logic front and center, while addressing both cases:
import socket import ssl def http_socket(domain='www.google.com', port=80): client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) host = socket.gethostbyname(domain) client.connect((host, port)) # Send a properly formatted HTTP/1.1 request (Host header is required for modern servers) request = f"GET / HTTP/1.1\r\nHost: {domain}\r\nConnection: close\r\n\r\n" client.sendall(request.encode('utf-8')) full_response = b"" while True: chunk = client.recv(4096) # Smaller buffer for granular control over incoming data if not chunk: break # Server closed the connection, no more data full_response += chunk # Check if we've received the full header section to determine when to stop reading if b"\r\n\r\n" in full_response: headers_part, body_part = full_response.split(b"\r\n\r\n", 1) # Case 1: Response uses Content-Length header content_length = None for header in headers_part.decode('utf-8').split("\r\n"): if header.lower().startswith('content-length:'): content_length = int(header.split(':')[1].strip()) break if content_length is not None and len(body_part) >= content_length: break # Case 2: Response uses chunked encoding elif b"Transfer-Encoding: chunked" in headers_part.lower(): # Chunked responses end with 0\r\n\r\n if b"\r\n0\r\n\r\n" in full_response: break client.close() return full_response
Key Socket Mechanics for Pen Testing
- TCP Streaming Logic: Unlike UDP, TCP doesn't treat data as discrete messages. You have to keep reading until you confirm you've received all content. This is non-negotiable for pen testing—when you're sending custom payloads or analyzing raw traffic, you can't assume one
recv()call gets everything. - HTTP Compliance: I added the
Hostheader (required for virtual hosting) andConnection: close(tells the server to shut the connection after sending the response, making it easier to detect when we're done). Understanding these small details helps you avoid getting blocked or receiving incomplete responses during testing. - Buffer Control: Using
4096instead of10000gives you more granular control over how you process incoming chunks. This is useful if you're scanning for specific patterns (like hidden headers or error messages) as data arrives. - Manual Header Parsing: By parsing headers yourself instead of relying on libraries, you get full visibility into exactly what the server is sending. This is critical for identifying misconfigurations—like servers leaking internal IPs or using outdated encoding schemes.
Bonus: HTTPS Support for Pen Testing
Since you imported ssl but didn't use it, here's how to adapt the code for HTTPS endpoints (essential for testing modern web apps):
def https_socket(domain='www.google.com', port=443): context = ssl.create_default_context() with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: with context.wrap_socket(sock, server_hostname=domain) as client: host = socket.gethostbyname(domain) client.connect((host, port)) request = f"GET / HTTP/1.1\r\nHost: {domain}\r\nConnection: close\r\n\r\n" client.sendall(request.encode('utf-8')) full_response = b"" while True: chunk = client.recv(4096) if not chunk: break full_response += chunk return full_response
This lets you inspect the SSL handshake and raw encrypted traffic (paired with tools like Wireshark) to test for TLS misconfigurations or vulnerabilities.
内容的提问来源于stack exchange,提问作者Sam B.

