A smart DNS proxy script in Python using SOCKS5

smart_dns.py

"""
SOCKS5 DNS Proxy
================

Description:
    This script runs a local DNS server that acts as a proxy, selectively
    routing DNS queries through any SOCKS5 proxy. It's a powerful tool for
    enhancing DNS privacy and bypassing internet censorship or sanctions.

Features:
    - Forwards DNS queries to an upstream server over a secure TCP connection.
    - Selectively routes queries for specified domains through a SOCKS5 proxy.
    - If no specific domains are listed, it routes ALL queries through the proxy.
    - Helps prevent DNS leak privacy issues.

Requirements:
    1. Python 3.x
    2. Required libraries: dnslib, PySocks
       Install them using pip:
       pip install dnslib PySocks
    3. A running SOCKS5 proxy service. This is not limited to Tor Browser.
       You can use any application that provides a SOCKS5 interface, such as
       V2RayNG or Nekoray. Just ensure you update the SOCKS5_PROXY_PORT in
       the configuration below to match your application's port.

Usage:
    1. Make sure your SOCKS5 proxy (e.g., Tor Browser, V2RayNG) is running.
    2. Run this script from your terminal: sudo python smart_dns.py
    3. Configure the DNS settings on your device(s):
       - For the computer running the script: Use 127.0.0.1 as the DNS server.
       - For other devices on the same network: Find this computer's local IP
         (e.g., 192.168.1.5), use it as the DNS server, and allow incoming
         traffic on UDP port 53 in this computer's firewall.

Troubleshooting:
    - OSError: [Errno 98] Address already in use:
      This error means another service (usually 'systemd-resolved' on Linux)
      is already using port 53. To fix it, stop the conflicting service.
      On Linux, you can typically use:
      sudo systemctl stop systemd-resolved

Tutorial: https://www.youtube.com/watch?v=fTi1h7lZHQI
"""

import socket
import socks
from dnslib import DNSRecord, DNSHeader, RCODE
from socketserver import BaseRequestHandler, UDPServer


# === Configuration ===
UPSTREAM_DNS_IP = '1.1.1.3'
UPSTREAM_DNS_PORT = 53
# The address of the local SOCKS5 proxy.
SOCKS5_PROXY_HOST = '127.0.0.1'
# Default SOCKS5 port for Tor Browser. Change this if you use another app.
SOCKS5_PROXY_PORT = 9150
# List of domain suffixes to resolve through the proxy.
# If empty, ALL domains will be resolved through the proxy.
DOMAINS_VIA_PROXY = []
LISTEN_PORT = 53


# === Helper Functions ===
def domain_matches(domain: str, patterns: list[str]) -> bool:
    """
    Determines if a domain should be resolved through the SOCKS5 proxy.
    """
    if not patterns:
        return True
    
    domain = domain.lower().rstrip('.')
    return any(domain == p or domain.endswith('.' + p) for p in patterns)


def query_dns_tcp(request_data: bytes, use_proxy: bool = False) -> bytes:
    """
    Forward a raw DNS query packet over TCP (optionally via SOCKS5).
    """
    if use_proxy:
        sock = socks.socksocket()
        sock.set_proxy(socks.SOCKS5, SOCKS5_PROXY_HOST, SOCKS5_PROXY_PORT)
    else:
        sock = socket.socket()

    try:
        sock.settimeout(10)
        sock.connect((UPSTREAM_DNS_IP, UPSTREAM_DNS_PORT))

        payload = len(request_data).to_bytes(2, 'big') + request_data
        sock.sendall(payload)

        resp_len_bytes = sock.recv(2)
        if len(resp_len_bytes) != 2:
            raise ConnectionAbortedError('Failed to receive response length.')
        
        resp_len = int.from_bytes(resp_len_bytes, 'big')
        response = sock.recv(resp_len)
        return response
    finally:
        sock.close()


def make_refused_response(request: DNSRecord) -> bytes:
    """Create a REFUSED DNS response for a given request."""
    reply = DNSRecord(
        DNSHeader(id=request.header.id, qr=1, aa=1, ra=1, rcode=RCODE.REFUSED),
        q=request.q
    )
    return reply.pack()


# === DNS Server Handler ===
class DNSHandler(BaseRequestHandler):
    """Handles incoming DNS queries and routes them appropriately."""
    def handle(self):
        data, sock = self.request
        try:
            request = DNSRecord.parse(data)
            domain = str(request.q.qname)
        except Exception as e:
            print(f'[!] Failed to parse request: {e}')
            return

        use_proxy = domain_matches(domain, DOMAINS_VIA_PROXY)
        print(f'[DNS] Query: {domain:<30} | Use Proxy: {use_proxy}')

        try:
            response_data = query_dns_tcp(data, use_proxy)
        except Exception as e:
            print(f'[!] Failed to resolve {domain}: {e}')
            response_data = make_refused_response(request)

        sock.sendto(response_data, self.client_address)


# === Main Entrypoint ===
def run_dns_server():
    """Starts the UDP DNS server."""
    # Bind to '0.0.0.0' to be accessible from the local network.
    server = UDPServer(('0.0.0.0', LISTEN_PORT), DNSHandler)
    print(f'[+] DNS server running on UDP port {LISTEN_PORT}...')
    print('[+] Press Ctrl+C to shut down.')
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print('\n[+] Shutting down...')
    finally:
        server.server_close()


if __name__ == '__main__':
    run_dns_server()