added ipv6 support; untested

enumerators/NetworkType.py

@@ -0,0 +1,13 @@
+from enum import Enum
+
+
+class NetworkType(Enum):
+    """
+    Modes the proxy can operate in
+    """
+    IPV4 = "IPv4"
+    IPV6 = "IPv6",
+    DUAL_STACK = "IPv4+IPv6"
+
+    def __str__(self):
+        return self.name

main.py

@@ -3,6 +3,7 @@
 
 import argparse
 
+from enumerators.NetworkType import NetworkType
 from enumerators.ProxyMode import ProxyMode
 from network.Proxy import Proxy
 from network.NetworkAddress import NetworkAddress
@@ -30,7 +31,7 @@ def list_of_modes(arg):
     parser.add_argument('--disabled_modes', type=list_of_modes,
                         choices=ProxyMode,
                         default=[],
-                        help='List of proxy modes to ignore. By default, all none are disabled. Hence, all are enabled')
+                        help='List of proxy modes to ignore. By default, none are disabled. Hence, all are enabled')
 
     parser.add_argument('--timeout', type=int,
                         default=120,
@@ -62,6 +63,14 @@ def list_of_modes(arg):
                         default=None,
                         help='DNS server ip for DNS over TLS')
 
+    parser.add_argument('--force_ipv6', type=bool,
+                        default=False,
+                        help='Force IPv6 when connecting to other servers')
+
+    parser.add_argument('--force_ipv4', type=bool,
+                        default=False,
+                        help='Force IPv4 when connecting to other servers')
+
     parser.add_argument_group('Forward proxy arguments')
 
     parser.add_argument('--forward_proxy_host', type=str,
@@ -103,13 +112,23 @@ def main():
     if args.forward_proxy_port is not None:
         forward_proxy = NetworkAddress(args.forward_proxy_host, args.forward_proxy_port)
 
+    if args.force_ipv4 and args.force_ipv6:
+        logging.debug("Cannot force both ipv4 and ipv6.")
+        exit()
+    elif args.force_ipv4:
+        domain_resolution = NetworkType.IPV4
+    elif args.force_ipv6:
+        domain_resolution = NetworkType.IPV6
+    else:
+        domain_resolution = NetworkType.DUAL_STACK
+
     if args.forward_proxy_mode in [ProxyMode.HTTP, ProxyMode.SNI] and args.forward_proxy_mode != args.proxy_mode:
         logging.debug("Forward proxy modes HTTP and SNI only usable if proxy mode is HTTP or SNI respectively.")
         exit()
 
     proxy = Proxy(server_address, args.timeout, args.record_frag, args.tcp_frag, args.frag_size,
-                  args.dot_resolver, args.disabled_modes, forward_proxy, args.forward_proxy_mode,
-                  args.forward_proxy_resolve_address)
+                  args.dot_resolver, domain_resolution, args.disabled_modes, forward_proxy,
+                  args.forward_proxy_mode, args.forward_proxy_resolve_address)
     proxy.start()
 
 

network/ConnectionHandler.py

@@ -1,6 +1,7 @@
 import logging
 import socket
 
+from enumerators.NetworkType import NetworkType
 from enumerators.ProxyMode import ProxyMode
 from exception.ParserException import ParserException
 from network.DomainResolver import DomainResolver
@@ -11,7 +12,7 @@
 from network.protocols.socksv4 import Socksv4
 from network.protocols.socksv5 import Socksv5
 from network.protocols.tls import Tls
-from util.Util import is_valid_ipv4_address
+from util.Util import is_valid_ip_address, is_valid_ipv4_address, is_valid_ipv6_address
 from util.constants import STANDARD_SOCKET_RECEIVE_SIZE, TLS_1_0_HEADER, TLS_1_2_HEADER, \
     TLS_1_1_HEADER, SOCKSv4_HEADER, SOCKSv5_HEADER
 
@@ -29,6 +30,7 @@ def __init__(self,
                  tcp_frag: bool,
                  frag_size: int,
                  dot_ip: str,
+                 network_type: NetworkType,
                  disabled_modes: list[ProxyMode],
                  forward_proxy: NetworkAddress,
                  forward_proxy_mode: ProxyMode,
@@ -41,6 +43,7 @@ def __init__(self,
         self.tcp_frag = tcp_frag
         self.frag_size = frag_size
         self.dot_ip = dot_ip
+        self.network_type = network_type
         self.disabled_modes = disabled_modes
         self.forward_proxy = forward_proxy
         self.forward_proxy_mode = forward_proxy_mode
@@ -65,7 +68,7 @@ def handle(self):
             return
 
         # resolve domain if no forward proxy or the forward proxy needs a resolved address
-        if (not is_valid_ipv4_address(final_server_address.host)) and \
+        if (not is_valid_ip_address(final_server_address.host)) and \
                 (self.forward_proxy_resolve_address or self.forward_proxy is None):
             if self.dot_ip:
                 host = DomainResolver.resolve_over_dot(final_server_address.host, self.dot_ip)
@@ -81,6 +84,17 @@ def handle(self):
             target_address = (self.forward_proxy.host, self.forward_proxy.port)
             self.debug(f"Using forward proxy {target_address}")
 
+        # check if derived ip clashes with ip preferences
+        if self.network_type == NetworkType.IPV4 and not is_valid_ipv4_address(target_address[0]):
+            self.error(f"Resolved address {final_server_address.host} is not a valid ipv4 address. Stopping!")
+            return
+        elif self.network_type == NetworkType.IPV6 and not is_valid_ipv6_address(target_address[0]):
+            self.error(f"Resolved address {final_server_address.host} is not a valid ipv6 address. Stopping!")
+            return
+        elif not is_valid_ip_address(target_address[0]):
+            self.error(f"Resolved address {final_server_address.host} is not a valid ip address. Stopping!")
+            return
+
         # open socket to server
         try:
             server_socket = socket.create_connection(target_address)

network/DomainResolver.py

@@ -14,7 +14,7 @@ def resolve_plain(domain: str) -> str:
         """
         Resolves the given domain to an ip address using the system's DNS resolver.
         """
-        return socket.gethostbyname(domain)
+        return socket.getaddrinfo(domain, None)[0][4][0]
 
     @staticmethod
     def resolve_over_dot(domain: str, dns_resolver: str) -> str:
@@ -28,6 +28,8 @@ def resolve_over_dot(domain: str, dns_resolver: str) -> str:
         if not domain.is_absolute():
             domain = domain.concatenate(dns.name.root)
 
+        # for now assume that all servers still have ipv6
+        # TODO: make this configurable
         query = dns.message.make_query(domain, dns.rdatatype.A)
         query.flags |= dns.flags.AD
         query.find_rrset(query.additional, dns.name.root, 65535,

network/Proxy.py

@@ -2,11 +2,11 @@
 import socket
 import threading
 
+from enumerators.NetworkType import NetworkType
 from enumerators.ProxyMode import ProxyMode
 from network.ConnectionHandler import ConnectionHandler
 from network.NetworkAddress import NetworkAddress
 from network.WrappedSocket import WrappedSocket
-from util.constants import TLS_1_0_HEADER, TLS_1_2_HEADER, TLS_1_1_HEADER, STANDARD_SOCKET_RECEIVE_SIZE
 
 
 class Proxy:
@@ -20,6 +20,7 @@ def __init__(self, address: NetworkAddress,
                  tcp_frag: bool = False,
                  frag_size: int = 20,
                  dot_ip: str = "8.8.4.4",
+                 network_type: NetworkType = NetworkType.DUAL_STACK,
                  disabled_modes: list[ProxyMode] = None,
                  forward_proxy: NetworkAddress = None,
                  forward_proxy_mode: ProxyMode = ProxyMode.HTTPS,
@@ -34,6 +35,7 @@ def __init__(self, address: NetworkAddress,
         self.frag_size = frag_size
         # whether to use dot for domain resolution
         self.dot_ip = dot_ip
+        self.network_type = network_type
         self.disabled_modes = disabled_modes
         if self.disabled_modes is None:
             self.disabled_modes = []
@@ -53,6 +55,7 @@ def handle(self, client_socket: WrappedSocket, address: NetworkAddress):
             self.tcp_frag,
             self.frag_size,
             self.dot_ip,
+            self.network_type,
             self.disabled_modes,
             self.forward_proxy,
             self.forward_proxy_mode,

network/protocols/socksv4.py

@@ -1,8 +1,10 @@
+import logging
+
 from exception.ParserException import ParserException
 from network.NetworkAddress import NetworkAddress
 from network.WrappedSocket import WrappedSocket
 from util.constants import SOCKSv4_HEADER
-from util.Util import is_valid_ipv4_address
+from util.Util import is_valid_ipv4_address, is_valid_ipv6_address
 
 
 class Socksv4:
@@ -44,14 +46,18 @@ def read_socks4(connection_socket: WrappedSocket) -> tuple[str, int]:
 
     @staticmethod
     def socks4_request(server_address: NetworkAddress):
-        if not is_valid_ipv4_address(server_address.host):
-            # Socksv4a domain encoding
+        if is_valid_ipv4_address(server_address.host):
+            # Socksv4 ip encoding
             return SOCKSv4_HEADER + b'\x01' + server_address.port.to_bytes(2, byteorder='big') + \
-                    "0.0.0.1".encode("'utf-8") + b'\x00' + server_address.host.encode('utf-8') + b'\x00'
+                bytes([int(i) for i in server_address.host.split('.')]) + b'\x00'
+            # ipv6 not supported in SOCKSv4a
+        elif is_valid_ipv6_address(server_address.host):
+            logging.warning("IPv6 not supported by SOCKSv4, use domain encoding or SOCKSv5 instead")
+            raise ValueError("IPv6 not supported by SOCKSv4, use domain encoding or SOCKSv5 instead")
         else:
-            # Socksv4 ip encoding
+            # Socksv4a domain encoding
             return SOCKSv4_HEADER + b'\x01' + server_address.port.to_bytes(2, byteorder='big') + \
-               bytes([int(i) for i in server_address.host.split('.')]) + b'\x00'
+                "0.0.0.1".encode("'utf-8") + b'\x00' + server_address.host.encode('utf-8') + b'\x00'
 
     @staticmethod
     def socks4_ok() -> bytes:

network/protocols/socksv5.py

@@ -4,7 +4,7 @@
 from network.NetworkAddress import NetworkAddress
 from network.WrappedSocket import WrappedSocket
 from util.constants import SOCKSv5_HEADER
-from util.Util import is_valid_ipv4_address
+from util.Util import is_valid_ipv4_address, is_valid_ipv6_address
 
 
 class Socksv5:
@@ -86,16 +86,22 @@ def socks5_auth_methods() -> bytes:
 
     @staticmethod
     def socks5_request(server_address: NetworkAddress):
-        if not is_valid_ipv4_address(server_address.host):
+        # ipv4
+        if is_valid_ipv4_address(server_address.host):
+            address = b'\0x01' + socket.inet_aton(server_address.host)
+        # ipv6
+        elif is_valid_ipv6_address(server_address.host):
+            address = b'\0x04' + socket.inet_pton(socket.AF_INET6, server_address.host)
+        # domain
+        else:
             domain = server_address.host.encode('utf-8')
             address = b'\0x03' + len(domain).to_bytes() + domain
-        else:
-            address = b'\0x01' + socket.inet_aton(server_address.host)
         return (SOCKSv5_HEADER + b'\x01' + Socksv5.RESERVED_BYTE + address
                 + server_address.port.to_bytes(2, byteorder='big'))
 
     @staticmethod
     def socks5_ok(connection_socket: WrappedSocket) -> bytes:
+        # TODO: only ipv4
         host_ip, host_port = connection_socket.socket.getsockname()
         host_address = b'\x01' + socket.inet_aton(host_ip)
         return SOCKSv5_HEADER + b'\x00' + Socksv5.RESERVED_BYTE + host_address + host_port.to_bytes(2, byteorder='big')

util/Util.py

@@ -1,14 +1,41 @@
 import socket
 
+from enumerators.NetworkType import NetworkType
 
-def is_valid_ipv4_address(ip_address: str) -> bool:
+
+def is_valid_ipv4_address(ip_address: str, network_type: NetworkType = NetworkType.DUAL_STACK) -> bool:
     """
     Returns whether the given string is a valid ipv4 address.
     :param ip_address: String to check for ipv4 validity
+    :param network_type: Network type to check for
     :return: Whether the given string is a valid ip address
     """
     try:
         socket.inet_aton(ip_address)
-        return True
+        return network_type == NetworkType.IPV4 or network_type == NetworkType.DUAL_STACK
+    except socket.error:
+        return False
+
+
+def is_valid_ipv6_address(ip_address: str, network_type: NetworkType = NetworkType.DUAL_STACK) -> bool:
+    """
+    Returns whether the given string is a valid ipv4 address.
+    :param ip_address: String to check for ipv4 validity
+    :param network_type: Network type to check for
+    :return: Whether the given string is a valid ip address
+    """
+    try:
+        socket.inet_pton(socket.AF_INET6, ip_address)
+        return network_type == NetworkType.IPV6 or network_type == NetworkType.DUAL_STACK
     except socket.error:
         return False
+
+
+def is_valid_ip_address(ip_address: str, network_type: NetworkType = NetworkType.DUAL_STACK) -> bool:
+    """
+    Returns whether the given string is a valid ip address.
+    :param ip_address: String to check for ipv validity
+    :param network_type: Network type to check for
+    :return: Whether the given string is a valid ip address
+    """
+    return is_valid_ipv4_address(ip_address, network_type) or is_valid_ipv6_address(ip_address, network_type)