🌐 Network Programming với Python

Network Programming là gì?

Network Programming cho phép Python apps giao tiếp qua mạng. Module socket cung cấp API low-level cho network communication.

Socket Basics

import socket

# Tạo socket
sock = socket.socket(
    socket.AF_INET,      # Address family (IPv4)
    socket.SOCK_STREAM   # Socket type (TCP)
)

# Socket types
# SOCK_STREAM = TCP (reliable, connection-oriented)
# SOCK_DGRAM = UDP (fast, connectionless)

TCP Client

import socket

def tcp_client(host: str, port: int, message: str) -> str:
    """Send message to TCP server and get response."""
    
    # Create socket
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    try:
        # Connect to server
        sock.connect((host, port))
        
        # Send data
        sock.send(message.encode())
        
        # Receive response
        response = sock.recv(4096)
        return response.decode()
    
    finally:
        sock.close()

# Usage
response = tcp_client("example.com", 80, "GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
print(response)

TCP Server

import socket
import threading

def handle_client(client_socket, address):
    """Handle client connection."""
    print(f"Connection from {address}")
    
    try:
        while True:
            # Receive data
            data = client_socket.recv(1024)
            if not data:
                break
            
            print(f"Received: {data.decode()}")
            
            # Send response
            response = f"Echo: {data.decode()}"
            client_socket.send(response.encode())
    
    finally:
        client_socket.close()
        print(f"Connection closed: {address}")

def tcp_server(host: str = "0.0.0.0", port: int = 9999):
    """Start TCP server."""
    
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
    server.bind((host, port))
    server.listen(5)
    
    print(f"Server listening on {host}:{port}")
    
    while True:
        client_socket, address = server.accept()
        
        # Handle client in new thread
        thread = threading.Thread(
            target=handle_client,
            args=(client_socket, address)
        )
        thread.start()

# Run server
# tcp_server()

UDP Client

import socket

def udp_client(host: str, port: int, message: str) -> str:
    """Send UDP packet and receive response."""
    
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.settimeout(5.0)
    
    try:
        # Send data (no connection needed)
        sock.sendto(message.encode(), (host, port))
        
        # Receive response
        data, addr = sock.recvfrom(4096)
        return data.decode()
    
    finally:
        sock.close()

# Example: DNS query (simplified)
# response = udp_client("8.8.8.8", 53, dns_query_bytes)

UDP Server

import socket

def udp_server(host: str = "0.0.0.0", port: int = 9999):
    """Start UDP server."""
    
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind((host, port))
    
    print(f"UDP Server listening on {host}:{port}")
    
    while True:
        data, addr = sock.recvfrom(1024)
        print(f"Received from {addr}: {data.decode()}")
        
        # Send response
        response = f"ACK: {data.decode()}"
        sock.sendto(response.encode(), addr)

HTTP Client (Low-level)

import socket
import ssl

def http_get(host: str, path: str = "/", port: int = 80, https: bool = False) -> str:
    """Make HTTP GET request."""
    
    # Create socket
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # Wrap with SSL for HTTPS
    if https:
        port = 443
        context = ssl.create_default_context()
        sock = context.wrap_socket(sock, server_hostname=host)
    
    try:
        sock.connect((host, port))
        
        # Build HTTP request
        request = f"GET {path} HTTP/1.1\r\n"
        request += f"Host: {host}\r\n"
        request += "Connection: close\r\n"
        request += "\r\n"
        
        sock.send(request.encode())
        
        # Receive response
        response = b""
        while True:
            chunk = sock.recv(4096)
            if not chunk:
                break
            response += chunk
        
        return response.decode()
    
    finally:
        sock.close()

# Usage
response = http_get("example.com", "/", https=True)
print(response[:500])

Banner Grabbing

import socket

def grab_banner(host: str, port: int, timeout: float = 2.0) -> str | None:
    """Grab service banner from port."""
    
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    
    try:
        sock.connect((host, port))
        
        # Some services send banner immediately
        banner = sock.recv(1024)
        
        if not banner:
            # Try sending something to trigger response
            sock.send(b"HEAD / HTTP/1.0\r\n\r\n")
            banner = sock.recv(1024)
        
        return banner.decode(errors='ignore').strip()
    
    except (socket.timeout, ConnectionRefusedError, OSError):
        return None
    
    finally:
        sock.close()

# Grab banners from common ports
ports = [21, 22, 25, 80, 443]
for port in ports:
    banner = grab_banner("scanme.nmap.org", port)
    if banner:
        print(f"Port {port}: {banner[:50]}")

DNS Resolution

import socket

def resolve_hostname(hostname: str) -> dict:
    """Resolve hostname to IP addresses."""
    
    result = {
        "hostname": hostname,
        "ipv4": [],
        "ipv6": []
    }
    
    try:
        # Get all address info
        infos = socket.getaddrinfo(hostname, None)
        
        for info in infos:
            family, _, _, _, addr = info
            ip = addr[0]
            
            if family == socket.AF_INET:
                if ip not in result["ipv4"]:
                    result["ipv4"].append(ip)
            elif family == socket.AF_INET6:
                if ip not in result["ipv6"]:
                    result["ipv6"].append(ip)
    
    except socket.gaierror as e:
        result["error"] = str(e)
    
    return result

# Usage
info = resolve_hostname("google.com")
print(f"IPv4: {info['ipv4']}")
print(f"IPv6: {info['ipv6']}")

# Reverse DNS
def reverse_dns(ip: str) -> str | None:
    """Reverse DNS lookup."""
    try:
        hostname, _, _ = socket.gethostbyaddr(ip)
        return hostname
    except socket.herror:
        return None

Network Scanner

import socket
from concurrent.futures import ThreadPoolExecutor

def check_port(host: str, port: int, timeout: float = 1.0) -> tuple:
    """Check if port is open."""
    
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    
    try:
        result = sock.connect_ex((host, port))
        return (port, result == 0)
    finally:
        sock.close()

def scan_ports(host: str, ports: list, threads: int = 100) -> list:
    """Scan multiple ports with threading."""
    
    open_ports = []
    
    with ThreadPoolExecutor(max_workers=threads) as executor:
        futures = [executor.submit(check_port, host, port) for port in ports]
        
        for future in futures:
            port, is_open = future.result()
            if is_open:
                open_ports.append(port)
    
    return sorted(open_ports)

# Usage
if __name__ == "__main__":
    target = "scanme.nmap.org"
    ports = list(range(1, 1025))
    
    print(f"Scanning {target}...")
    open_ports = scan_ports(target, ports)
    
    print(f"Open ports: {open_ports}")

Proxy SOCKS Support

import socks
import socket

# pip install PySocks

def connect_via_proxy(
    host: str, 
    port: int, 
    proxy_host: str = "127.0.0.1",
    proxy_port: int = 9050  # Tor default
) -> socket.socket:
    """Connect through SOCKS proxy."""
    
    sock = socks.socksocket()
    sock.set_proxy(socks.SOCKS5, proxy_host, proxy_port)
    sock.connect((host, port))
    
    return sock

# Global proxy (affects all sockets)
# socks.set_default_proxy(socks.SOCKS5, "127.0.0.1", 9050)
# socket.socket = socks.socksocket

Bước tiếp theo

Tiếp theo trong khóa học:

• Web Scraping: BeautifulSoup và requests
• Port Scanner: Xây dựng tool quét port hoàn chỉnh

---

💡 Pro tip: Luôn set timeout để tránh blocking vô hạn!