Add tool to the repository

.gitignore

+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+.idea/

CensorBreaker/App.py

+import logging
+import time
+
+import dns
+
+from CensorBreaker.config import ConnectionConfig, EvasionConfig
+from CensorBreaker.enumerators.EvasionStategies import EvasionStrategyEnum
+from CensorBreaker.helper.Availability import AvailabilityCheck
+from CensorBreaker.messageHandler import MessageHandler
+from CensorBreaker.network.Proxy import DNSProxy
+
+
+class DNSCensorBreaker:
+    """
+    Main class of the tool to circumvent DNS censorship
+    :param timeout: Timeout in seconds until a listening socket times out.
+    :param connection_config: Connection config the tool should use.
+    :param evasion_config: Specifies the evasion techniques that should be used.
+    """
+
+    def __init__(self, timeout, connection_config: ConnectionConfig, evasion_config: EvasionConfig):
+        self.message_handler = None
+
+        self.timeout = timeout
+        self.connection_config = connection_config
+        self.evasion_config = evasion_config
+        self.dns_server = connection_config.dns_server
+
+        self.proxy = None
+
+        self._prepare_execution()
+
+    def execute(self):
+        """
+        Starts the tool and waits until DNSProxy is stopped.
+        """
+        self._print_startup_message()
+
+        self.message_handler = MessageHandler(self.connection_config, self.evasion_config)
+
+        self.proxy = DNSProxy()
+        self.proxy.start(self.timeout, self.connection_config, self.message_handler)
+
+        # wait until proxy started
+        while not self.proxy.is_running():
+            time.sleep(0.1)
+
+        # wait until proxy is not running anymore
+        while self.proxy.is_running():
+            time.sleep(0.5)
+
+    def _prepare_execution(self):
+        """
+        Called after initializing the tool. It checks the availability of the configuration and adapts it as needed.
+        """
+        availability = AvailabilityCheck()
+
+        # If dns server is provided with https://* get the IP first
+        if self.connection_config.dns_server is not None and not self.connection_config.dns_server.startswith(
+                "https://") and not dns.inet.is_address(self.connection_config.dns_server):
+            self.connection_config.dns_server = availability.get_ip_from_domain(self.connection_config.dns_server)
+
+        self.connection_config, self.evasion_config = availability.check_current_configuration(self.connection_config,
+                                                                                               self.evasion_config)
+
+    def _print_startup_message(self):
+        """
+        Prints the used configuration.
+        """
+        logging_string = (f"CensorBreaker started using configuration: \n"
+                          f"\tConnection Configuration:\n"
+                          f"\t\tHost: {self.connection_config.host},\n"
+                          f"\t\tPort: {self.connection_config.port},\n"
+                          f"\t\tDNS Server: {self.connection_config.dns_server},\n"
+                          f"\t\tServer Port: {self.connection_config.server_port}")
+
+        logging_string += "\n"
+
+        logging_string += f"\tEvasion Configuration:\n\t\tStrategy: {self.evasion_config.evasion_strategy}"
+
+        if self.evasion_config.evasion_strategy == EvasionStrategyEnum.Encrypted:
+            logging_string += f",\n\t\tEncryption Protocol: {self.evasion_config.encryption_protocol}"
+
+        logging.info(logging_string)
+
+    def stop(self):
+        """
+        Stops the DNSProxy so that the tool can exit.
+        """
+        self.proxy.stop()

CensorBreaker/config.py

+from enum import Enum
+
+import yaml
+
+from CensorBreaker.enumerators.EncryptionProtocol import EncryptionProtocolEnum
+from CensorBreaker.enumerators.EvasionStategies import EvasionStrategyEnum
+from CensorBreaker.exception.ConfigExceptions import KeyNotFoundException
+
+encrypted_dns_config_file = "configs/default_config.yml"
+
+_loaded_file = None
+
+
+def _load_file():
+    """
+    Loads the default_config.yml file and stores it.
+    """
+    global _loaded_file
+    with open(encrypted_dns_config_file, "r") as file:
+        _loaded_file = yaml.safe_load(file)
+
+
+def get_config():
+    """
+    Gets whole config.
+    """
+    global _loaded_file
+    if _loaded_file is None:
+        _load_file()
+    return _loaded_file
+
+def get_default_dns_server():
+    """
+    Gets the default_dns_server configuration.
+    """
+    config = get_config()
+    if config is not None and "default_dns_server" in config:
+        return config["default_dns_server"]
+    return None
+
+def get_default_dns_server_port():
+    """
+    Gets the default_dns_server_port configuration.
+    """
+    config = get_config()
+    if config is not None and "default_dns_server_port" in config:
+        return config["default_dns_server_port"]
+    return None
+
+
+def get_encryption_config():
+    """
+    Gets the default encrypted_dns configuration.
+    """
+    config = get_config()
+    if config is not None and "encrypted_dns" in config:
+        return config["encrypted_dns"]
+
+    return {}
+
+def get_availability_check_config():
+    """
+    Gets the default availability_check configuration.
+    """
+    config = get_config()
+    if config is not None and "availability_check" in config:
+        return config["availability_check"]
+
+    return {}
+
+
+def get_usability_protocol_order():
+    """
+    Gets the protocol order for availability check.
+    """
+    config = get_availability_check_config()
+    if "protocol_order" in config:
+        return config["protocol_order"]
+
+    return []
+
+
+def get_default_encryption_settings(namespace):
+    """
+    Gets the default encryption settings including default_host and default_port.
+    """
+    if isinstance(namespace, Enum):
+        namespace = namespace.name
+
+    config = get_encryption_config()
+    if config is not None and namespace in config and "default_host" in config[namespace] and "default_port" in config[namespace]:
+        return config[namespace]["default_host"], config[namespace]["default_port"]
+
+    raise KeyNotFoundException(namespace)
+
+
+def get_default_encryption_mode():
+    """
+    Gets the default encryption protocol mode.
+    """
+    config = get_encryption_config()
+
+    if "default_mode" in config:
+        return EncryptionProtocolEnum[config["default_mode"]]
+    return EncryptionProtocolEnum.DoQ
+
+
+def get_dns_server_list():
+    """
+    Gets the dns_server_list configuration.
+    """
+    config = get_availability_check_config()
+    if config is not None and "dns_server_list" in config:
+        return config["dns_server_list"]
+    return {}
+
+class EvasionConfig:
+    """
+    Defines the configuration of the evasion technique.
+    :param evasion_strategy: Includes the evasion strategy which should be used.
+    :param encryption_protocol: Includes the protocol which should be used for encrypted DNS.
+    """
+    def __init__(self, evasion_strategy: EvasionStrategyEnum, encryption_protocol: EncryptionProtocolEnum):
+        self.evasion_strategy = evasion_strategy
+        self.encryption_protocol = encryption_protocol
+
+    def __str__(self):
+        return f"EvasionConfig: {self.evasion_strategy}, {self.encryption_protocol}"
+
+class ConnectionConfig:
+    """
+    Defines the connection of the tool.
+    :param host: The host the tool is running on.
+    :param port: The port the host is reachable on.
+    :param dns_server: The IP of the DNS server that should be used for resolving domains.
+    :param server_port: (Optional) The port the DNS server is reachable on. When nothing is provided the standard port
+        53 is used.
+    """
+    def __init__(self, host: str, port: int, dns_server: str, server_port=None):
+        self.host = host
+        self.port = port
+        self.dns_server = dns_server
+        self.server_port = server_port
+
+    def __str__(self):
+        return f"ConnectionConfig: {self.host}, {self.port}, {self.dns_server}, {self.server_port}"

CensorBreaker/enumerators/EncryptionProtocol.py

+from enum import Enum
+
+
+class EncryptionProtocolEnum(Enum):
+    """
+    Defines possible encrypted DNS protocols. FindUsable is used for the availability check.
+    """
+    DoT = 0,
+    DoH = 1,
+    DoQ = 2,
+    FindUsable = 3
+
+    def __str__(self):
+        return self.name

CensorBreaker/enumerators/EvasionStategies.py

+from enum import Enum
+
+
+class EvasionStrategyEnum(Enum):
+    """
+    Defines possible evasion strategies. PreferEncrypted is used to check if an encrypted connection can be established.
+    If not, it falls back to GFWatch.
+    """
+    GFWatch = 0,
+    Encrypted = 1,
+    PreferEncrypted = 2,
+    TCPFragmentation = 3,
+
+    def __str__(self):
+        return self.name

CensorBreaker/evasion.py

+from abc import ABC, abstractmethod
+
+from CensorBreaker.enumerators.EncryptionProtocol import EncryptionProtocolEnum
+from CensorBreaker.exception.EncryptionException import UnknownEncryptionModeException
+from CensorBreaker.network.WrappedSocket import WrappedSocket, ClientSocket, TCPSocket
+
+import dns.query
+from dns.message import Message
+from dns.rdtypes.IN import A, AAAA
+
+from CensorBreaker.helper.GFWatch import GFWatchHelper
+
+import threading
+import CensorBreaker.config
+import logging
+import socket
+
+
+class EvasionStrategy(ABC):
+    """
+    Abstract class for all evasion strategies. When adding a new stategy it should extend from this class.
+    The working property should hold the current state of the strategy. When the strategy does not work anymore it
+    should be set to False.
+    """
+
+    @abstractmethod
+    def __init__(self):
+        self.__working = True
+
+    @property
+    def working(self):
+        """
+        Defines the working state of the strategy. When strategy is not working anymore this is set to False.
+        """
+        return self.__working
+
+    @working.setter
+    def working(self, value):
+        self.__working = value
+
+    @abstractmethod
+    def _start_execution_thread(self, client_socket: ClientSocket | TCPSocket, message: Message):
+        """
+        Basic execution method that is called when a strategy is started. This needs to be implemented for every
+        strategy.
+        :param client_socket: The socket which should be used to answer with the gathered response.
+        :param message: The message that should be sent to a resolver. It includes the DNS query for a domain.
+        """
+        pass
+
+    def execute(self, client_socket: ClientSocket | TCPSocket, message: Message):
+        """
+        Starts the strategy by executing a thread which executes the _start_execution_thread method.
+        :return: If the strategy is still working.
+        """
+        thread = threading.Thread(target=self._start_execution_thread, args=(client_socket, message))
+        thread.start()
+
+        return self.working
+
+
+class EncryptedDNS(EvasionStrategy):
+    """
+    Evasion strategy using encrypted DNS to bypass censorship.
+    :param host: (Optional) The IP of the DNS resolver that should be used for resolving. If None, default config is
+    used.
+    :param port: (Optional) The port of the DNS resolver that should be used for resolving. If None, default config is
+    used.
+    :param method: (Optional) The method that should be used  for resolving. If None, default encryption mode is used.
+    """
+
+    def __init__(self, host=None, port=None, method=None):
+        self.method = CensorBreaker.config.get_default_encryption_mode()
+
+        if method is not None:
+            self.method = method
+
+        self.host, self.port = CensorBreaker.config.get_default_encryption_settings(self.method.name)
+
+        if host is not None:
+            self.host = host
+        if port is not None:
+            self.port = port
+
+        self.working = True
+
+    def _start_execution_thread(self, client_socket: ClientSocket | TCPSocket, message: Message):
+        logging.debug(f"EncryptedDNS: Received message {message}")
+
+        # decide on encrypted protocol
+        if self.method == "DoH" or self.method == EncryptionProtocolEnum.DoH:
+            answer = self._get_doh_response(message)
+        elif self.method == "DoT" or self.method == EncryptionProtocolEnum.DoT:
+            answer = self._get_dot_response(message)
+        elif self.method == "DoQ" or self.method == EncryptionProtocolEnum.DoQ:
+            answer = self._get_doq_response(message)
+        else:
+            raise UnknownEncryptionModeException(self.method)
+
+        logging.debug(f"EncryptedDNS: Got answer {answer}")
+
+        # add data length to message
+        client_socket.send(answer.to_wire())
+
+    def _get_dot_response(self, message: Message) -> Message:
+        """
+        Gets the DNS response for a given message by using DoT.
+        :param message: The DNS query that should be queried.
+        """
+        return dns.query.tls(message, self.host, port=self.port)
+
+    def _get_doh_response(self, message: Message) -> Message:
+        """
+        Gets the DNS response for a given message by using DoH.
+        :param message: The DNS query that should be queried.
+        """
+        return dns.query.https(message, self.host, port=self.port)
+
+    def _get_doq_response(self, message: Message) -> Message:
+        """
+        Gets the DNS response for a given message by using DoQ.
+        :param message: The DNS query that should be queried.
+        """
+        # store message id because DoQ sets it to zero
+        msg_id = message.id
+        answer = dns.query.quic(message, self.host, port=self.port)
+
+        answer.id = msg_id
+        return answer
+
+
+class GFWatchComparison(EvasionStrategy):
+    """
+    Evasion strategy using the GFWatch comparison to detect injected responses and discard them.
+    :param dns_server: The IP of the DNS resolver that should be used for resolving.
+    :param port: (Optional) The port of the DNS resolver that is used for resolving the query. By default set to 53.
+    """
+
+    def __init__(self, dns_server, port=53):
+        self.dns_server = dns_server
+        self.port = port
+
+        # used for IP list check
+        self.gfwatch_helper = GFWatchHelper()
+        self.working = True
+
+    def execute_synchronous(self, message: Message) -> Message:
+        """
+        Main function for using this strategy.
+        Sends the given message to the defined DNS resolver and discards forged responses.
+        This can be used without starting an execution thread. It returns the correct response without censorship.
+        :param message: The DNS query that should be resolved.
+        :return: The DNS response that is gathered.
+        """
+        udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        udp_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        udp_sock = WrappedSocket(10, udp_sock)
+
+        # send query to dns server
+        dns.query.send_udp(udp_sock, message, (self.dns_server, self.port))
+        logging.debug(f"Sending to {self.dns_server}:{self.port}")
+
+        answer = self._receive_correct_answer(udp_sock)
+        return answer
+
+    def _start_execution_thread(self, client_socket: ClientSocket | TCPSocket, message: Message):
+        answer = self.execute_synchronous(message)
+
+        if answer is not None:
+            client_socket.send(answer)
+
+        # print(f"Thread finished!!")
+
+    def _receive_correct_answer(self, receiving_socket: WrappedSocket) -> Message | None:
+        """
+        Starts listening on given socket for incoming messages.
+        When a message is received, it checks whether it is forged or correct.
+        For safety reasons, the function only tries receiving a response 10 times before returning None.
+        :param receiving_socket: Socket that receives incoming DNS responses.
+        :return: Returns the DNS response if a correct response is received. None otherwise.
+        """
+        safety = 0
+        while safety < 10:
+            try:
+                message, address = receiving_socket.recvfrom(1024)
+
+                logging.debug(f"GFWatchComparison: received message {message} check if message is forged!")
+                if not self._is_response_forged(message):
+                    logging.debug(
+                        f"GFWatchComparison: Received correct dns answer: {dns.message.from_wire(message).answer}")
+                    return message
+                else:
+                    logging.debug(f"GFWatchComparison: Response is forged! Just wait for correct message!")
+                    logging.debug(f"GFWatchComparison: {dns.message.from_wire(message)}")
+            except TimeoutError:
+                logging.debug("Timeout occured in receiving socket")
+                receiving_socket.close()
+                break
+            except ConnectionResetError:
+                logging.debug(f"GFWatchComparison: Connection reset in receive correct answer!")
+                break
+
+            safety += 1
+
+        if safety >= 10:
+            logging.debug(f"GFWatchComparison: SAFETY Waited for response but did not receive one!")
+
+        return None
+
+    def _is_response_forged(self, raw_message: bytes) -> bool:
+        """
+        Checks if a given byte string is a forged DNS response.
+        Uses the GFWatch forged ip list.
+        :param raw_message: Byte string of the received DNS response.
+        :return: True if the response is forged, False otherwise.
+        """
+
+        dns_response = dns.message.from_wire(raw_message)
+        answer_section = dns_response.answer
+
+        forged = False
+        for rrset in answer_section:
+            for item in rrset:
+                # when an IPv4 response is found
+                if isinstance(item, A.A):
+                    forged = self.gfwatch_helper.is_forged_ipv4(item.address)
+                # when an IPv6 response is found
+                if isinstance(item, AAAA.AAAA):
+                    forged = self.gfwatch_helper.is_forged_ipv6(item.address)
+                # if one forged IP is found return True
+                # don't look at other records
+                if forged:
+                    return True
+        return forged
+
+
+class TCPFragmentation(EvasionStrategy):
+    """
+    Evasion config uses fragmentation of the DNS query on the TCP layer.
+    :param dns_server: The IP of the DNS server that should be used for resolving the DNS query.
+    :param port: (Optional) The port of the DNS resolver. By default set to 53.
+    """
+
+    def __init__(self, dns_server: str, port: int = 53):
+        self.dns_server = dns_server
+        self.port = port
+        self.gfwatch_helper = GFWatchHelper()
+        self.working = True
+
+    def _start_execution_thread(self, client_socket: ClientSocket | TCPSocket, message: Message):
+        tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        tcp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+
+        tcp_socket.connect((self.dns_server, self.port))
+
+        # Create TCP socket with fragmentation size of 18 bytes
+        wrapped_tcp_socket = TCPSocket(5, tcp_socket, 18)
+
+        wrapped_tcp_socket.send(message.to_wire())
+
+        response = b""
+        # try to receive response
+        while True:
+            try:
+                data = wrapped_tcp_socket.recv(1024)
+            except TimeoutError:
+                return
+            if not data:
+                wrapped_tcp_socket.try_close()
+                return
+
+            response += data
+            data_length = int.from_bytes(response[:2], "big")
+            if data_length > len(response[2:]):
+                continue
+            else:
+                response = response[2:]
+                break
+        wrapped_tcp_socket.try_close()
+
+        # check if forged DNS response is received
+        is_working = self._check_working(response)
+        if not is_working:
+            # set to not working if forged response is received
+            self.working = False
+
+        client_socket.send(response)
+
+    def _check_working(self, data: bytes) -> bool:
+        """
+        Checks if a received DNS response is forged to determine if the circumvention technique is still working.
+        :param data: Byte string of the received DNS response.
+        :return: True if the response is not forged by the GFW, False otherwise.
+        """
+        try:
+            message = dns.message.from_wire(data)
+            forged = False
+            for rrset in message.answer:
+                for item in rrset:
+                    if isinstance(item, A.A):
+                        forged = self.gfwatch_helper.is_forged_ipv4(item.address)
+                    if isinstance(item, AAAA.AAAA):
+                        forged = self.gfwatch_helper.is_forged_ipv6(item.address)
+                    if forged:
+                        return False
+            return True
+        except Exception:
+            return True

CensorBreaker/exception/AvailabilityException.py

+class NoIpForDomainException(Exception):
+    """
+    Exception thrown when no ip could be found for domain
+    """
+    def __init__(self, domain, *args, **kwargs):
+
+        self.domain = domain
+
+        self.message = f"No ip was found for given domain {domain}!"
+        super().__init__(self.message, args, kwargs)
+
+
+class ConfigurationUnusableException(Exception):
+    """
+    Exception thrown when the configuration given by arguments is not possible.
+    """
+    def __init__(self, *args, **kwargs):
+        self.message = (f"The given configuration is not usable, because the dns server is not reachable with the given"
+                        f" method!")
+        super().__init__(self.message, args, kwargs)
\ No newline at end of file

CensorBreaker/exception/ConfigExceptions.py

+
+class KeyNotFoundException(Exception):
+    """
+    Exception thrown when key not found in config
+    """
+    def __init__(self, key, *args, **kwargs):
+        self.key = key
+
+        self.message = f"Key {key} not found in config!"
+        super().__init__(self.message, args, kwargs)
\ No newline at end of file

CensorBreaker/exception/EncryptionException.py

+class UnknownEncryptionModeException(Exception):
+    """
+    Exception thrown when key is not in config
+    """
+    def __init__(self, mode, *args, **kwargs):
+
+        self.mode = mode
+
+        self.message = f"Encryption mode {mode} is unknown and cannot be executed!"
+        super().__init__(self.message, args, kwargs)
\ No newline at end of file

CensorBreaker/exception/ParserException.py

+class ParserException(Exception):
+    """
+    For exceptions during the parsing process
+    """
+    def __init__(self, *args, **kwargs):
+        super().__init__(args, kwargs)

CensorBreaker/helper/Availability.py

+import logging
+
+from dns.message import Message
+
+from CensorBreaker import config
+from CensorBreaker.enumerators.EncryptionProtocol import EncryptionProtocolEnum
+import dns.query
+import dns.resolver
+from dns.rdtypes.IN import A, AAAA
+
+from CensorBreaker.enumerators.EvasionStategies import EvasionStrategyEnum
+import CensorBreaker.evasion
+from CensorBreaker.exception.AvailabilityException import NoIpForDomainException, ConfigurationUnusableException
+from CensorBreaker.config import ConnectionConfig, EvasionConfig
+
+
+class AvailabilityCheck:
+    """
+    Class for checking the connection and evasion config for availability.
+    It does this by sending a not censored domain and checks if a response is received.
+    """
+    def __init__(self):
+        self.test_domain = "www.gov.cn"
+        self.test_timeout = 5
+
+    def check_current_configuration(self, connection_config: ConnectionConfig, evasion_config: EvasionConfig) \
+            -> (ConnectionConfig, EvasionConfig):
+        """
+        Main method for checking configurations.
+        :param connection_config: The connection config that should be used for the check.
+        :param evasion_config: The evasion config that should be used for the check.
+        :return: A tuple of an available connection and evasion config.
+        :raise ConfigurationUnusableException: When an encrypted connection is specified but not available.
+        """
+        # check encrypted DNS availability
+        if evasion_config.evasion_strategy == EvasionStrategyEnum.PreferEncrypted or evasion_config.evasion_strategy == EvasionStrategyEnum.Encrypted:
+            config_usable = False
+            # no specific connection is given, and we should find a usable encrypted server.
+            if evasion_config.encryption_protocol == EncryptionProtocolEnum.FindUsable:
+                server, port, protocol = self._find_usable_server()
+                if server is not None and protocol is not None:
+                    logging.debug(f"Found Usable connection {server}, {protocol}")
+                    connection_config.dns_server = server
+                    connection_config.server_port = port
+                    evasion_config.evasion_strategy = EvasionStrategyEnum.Encrypted
+                    evasion_config.encryption_protocol = protocol
+                    return connection_config, evasion_config
+                else:
+                    logging.warning(f"Could not find a working encrypted connection!")
+            else:
+                # check specific given encrypted connection
+                result = self.check_specific_connection(evasion_config.encryption_protocol,
+                                                        connection_config.dns_server)
+                if result:
+                    #encrypted DNS connection is not usable
+                    config_usable = False
+
+            if config_usable:
+                # if encrypted server is available set the strategy to encrypted.
+                if evasion_config.evasion_strategy == EvasionStrategyEnum.PreferEncrypted:
+                    evasion_config.evasion_strategy = EvasionStrategyEnum.Encrypted
+
+            else:
+                # when not encrypted server is available and we prefer encrypted fall back to GFWatchComparison
+                if evasion_config.evasion_strategy == EvasionStrategyEnum.PreferEncrypted:
+                    evasion_config.evasion_strategy = EvasionStrategyEnum.GFWatch
+                    logging.warning(
+                        "CensorBreaker: Encrypted DNS is not usable with the provided arguments! Fallback to GFWatch comparison.")
+                else:
+                    # If a specific encrypted DNS connection should be used but it is not available.
+                    raise ConfigurationUnusableException(evasion_config.evasion_strategy,
+                                                         evasion_config.encryption_protocol,
+                                                         connection_config.dns_server)
+        else:
+            # No encrypted DNS connection should be tested.
+            # Set the dns_server and server_port to default values in case they are not set within the parameters.
+            if connection_config.dns_server is None:
+                connection_config.dns_server = config.get_default_dns_server()
+            if connection_config.server_port is None:
+                connection_config.server_port = config.get_default_dns_server_port()
+        return connection_config, evasion_config
+
+    def _find_usable_server(self) -> (str, int, EncryptionProtocolEnum):
+        """
+        Finds a usable encrypted DNS resolver.
+        :return: A tuple of IP, port, and encrypted DNS protocol
+        """
+        protocol_order = config.get_usability_protocol_order()
+
+        dns_server_list = config.get_dns_server_list()
+
+        for protocol in protocol_order:
+            p = EncryptionProtocolEnum[protocol]
+            if protocol in dns_server_list:
+                for server in dns_server_list[protocol]:
+                    result = self.check_specific_connection(p, server)
+                    if result:
+                        return server, None, p
+
+        return None, None, None
+
+    def check_specific_connection(self, method: EncryptionProtocolEnum, server: str) -> bool:
+        """
+        Checks one specific encrypted connection for the given protocol and server.
+        :param method: The protocol that should be checked.
+        :param server: The IP of the server that should be checked.
+        :return: True if connection is available and False otherwise.
+        """
+        if method == EncryptionProtocolEnum.DoT:
+            return self._check_DoT(server)
+        elif method == EncryptionProtocolEnum.DoH:
+            return self._check_DoH(server)
+        elif method == EncryptionProtocolEnum.DoQ:
+            return self._check_DoQ(server)
+        else:
+            return False
+
+    def _check_DoT(self, server: str) -> bool:
+        """
+        Checks DoT connection for the given server.
+        :param server: The IP of the server that should be checked.
+        :return: True if server is available, False otherwise.
+        """
+        try:
+            message = self._get_dummy_message()
+            response = dns.query.tls(message, server, self.test_timeout)
+            return True
+        except Exception as e:
+            print(str(e))
+            return False
+
+    def _check_DoH(self, server: str) -> bool:
+        """
+        Checks DoH connection for the given server.
+        :param server: The IP of the server that should be checked.
+        :return: True if server is available, False otherwise.
+        """
+        try:
+            message = self._get_dummy_message()
+            response = dns.query.https(message, server, self.test_timeout)
+            return True
+        except Exception as e:
+            print(str(e))
+            return False
+
+    def _check_DoQ(self, server: str) -> bool:
+        """
+        Checks DoQ connection for the given server.
+        :param server: The IP of the server that should be checked.
+        :return: True if server is available, False otherwise.
+        """
+        try:
+            message = self._get_dummy_message()
+            response = dns.query.quic(message, server, self.test_timeout)
+            return True
+        except Exception as e:
+            print(str(e))
+            return False
+
+    def _get_dummy_message(self) -> Message:
+        """
+        Gets a DNS query for the test domain.
+        :return: The dummy DNS query.
+        """
+        return dns.message.make_query(self.test_domain, dns.rdatatype.A)
+
+    def get_ip_from_domain(self, domain: str, dns_server=None, port=None) -> str:
+        """
+        Sends a DNS query for the given domain to resolve the IP address.
+        :param domain: The domain that should be queried.
+        :param dns_server: (Optional) The DNS server that should be used for the query. By default, it uses the standard
+        DNS resolver from the PC configuration.
+        :param port: (Optional) The port of the DNS server that should be used for the query. By default, it is set to
+        53.
+        :return: IP of the domain.
+        :raise NoIpForDomainException: If no IP could be gathered for the given domain.
+        """
+        if dns_server is None:
+            dns_resolver = dns.resolver.Resolver()
+            dns_server = dns_resolver.nameservers[0]
+
+        if port is None:
+            port = 53
+        gfwatch = CensorBreaker.evasion.GFWatchComparison(dns_server, port)
+
+        message = dns.message.make_query(domain, dns.rdatatype.A)
+
+        dns_response = gfwatch.execute_synchronous(message)
+        dns_response = dns.message.from_wire(dns_response)
+        answer_section = dns_response.answer
+
+        for rrset in answer_section:
+            for item in rrset:
+                if isinstance(item, A.A) or isinstance(item, AAAA.AAAA):
+                    return item.address
+
+        raise NoIpForDomainException(domain)

CensorBreaker/helper/GFWatch.py

+import socket
+
+
+class GFWatchHelper:
+    """
+    Used to check IP addresses if they are forged by the GFW.
+    """
+
+    _ip4_addresses = {}
+    _ip6_addresses = {}
+
+    def __init__(self):
+        # loads the files
+        self._load_ipv4()
+        self._load_ipv6()
+
+    def is_forged_ipv4(self, address: str) -> bool:
+        """
+        Checks if the given IPv4 address is forged.
+        :return: True if IP is forged, False otherwise.
+        """
+        return address in self._ip4_addresses and self._ip4_addresses[address]
+
+    def is_forged_ipv6(self, address: str) -> bool:
+        """
+        Checks if the given IPv6 address is forged.
+        :return: True if IP is forged, False otherwise.
+        """
+        try:
+            # check if given IP is actually an IPv6 address
+            # and pack the IP to make it comparable
+            ip = socket.inet_pton(socket.AF_INET6, address)
+        except:
+            return False
+        return ip in self._ip6_addresses and self._ip6_addresses[ip]
+
+    def _load_ipv4(self):
+        """
+        Loads the forged IPv4 addresses into the class.
+        """
+        with open("resources/forged.ipv4", "r") as f:
+            lines = f.readlines()
+        f.close()
+        for line in lines:
+            ip = line.split("|")[1]
+            self._ip4_addresses[ip] = True
+
+    def _load_ipv6(self):
+        """
+        Loads the forged IPv6 addresses into the class.
+        """
+        with open("resources/forged.ipv6", "r") as f:
+            lines = f.readlines()
+        f.close()
+        for line in lines:
+            ip = line.split("|")[1]
+            try:
+                # pack IPv6 address so that it is comparable later
+                ip = socket.inet_pton(socket.AF_INET6, ip)
+            except:
+                continue
+            self._ip6_addresses[ip] = True

CensorBreaker/messageHandler.py

+import logging
+
+import dns
+from dns import query as DNSQuery
+
+from .config import EvasionConfig, ConnectionConfig
+from .enumerators.EvasionStategies import EvasionStrategyEnum
+from .evasion import EncryptedDNS, TCPFragmentation
+from .evasion import GFWatchComparison
+from .network.WrappedSocket import ClientSocket, WrappedSocket
+
+
+class MessageHandler:
+    """
+    Main class for handling incoming DNS queries.
+    It automatically uses the selected strategy and falls back to GFWatchComparison if the used method is not working
+    anymore.
+    :param connection_config: The connection config of the tool, which defines the used DNS server.
+    :param evasion_config: The evasion config that should be used. Needed for choosing the used strategy.
+    """
+
+    def __init__(self, connection_config: ConnectionConfig, evasion_config: EvasionConfig):
+        self.connection_config = connection_config
+
+        self.evasion_config = evasion_config
+
+        if evasion_config.evasion_strategy == EvasionStrategyEnum.GFWatch:
+            self.strategy = GFWatchComparison(connection_config.dns_server, connection_config.server_port)
+
+        if evasion_config.evasion_strategy == EvasionStrategyEnum.Encrypted:
+            self.strategy = EncryptedDNS(connection_config.dns_server, connection_config.server_port,
+                                         evasion_config.encryption_protocol)
+
+        if evasion_config.evasion_strategy == EvasionStrategyEnum.TCPFragmentation:
+            self.strategy = TCPFragmentation(connection_config.dns_server, connection_config.server_port)
+
+    def handle_message(self, client_socket: ClientSocket | WrappedSocket, raw_message: bytes):
+        """
+        Main function for handling an incoming message. This should be called whenever a DNS query should be resolved
+        using a circumvention method.
+        :param client_socket: The socket that is used to send an answer back. Just call send on this socket to respond.
+        :param raw_message: The byte string of the received DNS message.
+        """
+        message = dns.message.from_wire(raw_message)
+        if self.evasion_config.evasion_strategy == EvasionStrategyEnum.PreferEncrypted:
+            # The evasion strategy PreferEncrypted should not appear here. There must be something wrong with the
+            # availability check!
+            logging.error(f"MessageHandler: Something went wrong with availability checking. "
+                          f"Got PreferEncrypted as strategy!")
+        else:
+            # execute strategy
+            is_working = self.strategy.execute(client_socket, message)
+
+            # fallback to GFWatchComparison if strategy is not working anymore.
+            # In case GFWatchComparison is not working anymore change this to a strategy that is most likely to succeed.
+            if not is_working:
+                logging.warning(f"Current strategy {self.evasion_config.evasion_strategy} is not working anymore! "
+                                f"Fallback to GFWatch comparison!")
+                self.strategy = GFWatchComparison(self.connection_config.dns_server, self.connection_config.server_port)

CensorBreaker/network/Proxy.py

+import logging
+import socket
+import threading
+from abc import ABC
+
+from CensorBreaker.config import ConnectionConfig
+from CensorBreaker.messageHandler import MessageHandler
+from CensorBreaker.network.WrappedSocket import WrappedSocket, ClientSocket, TCPSocket
+
+
+class AbstractProxy(ABC):
+    """
+    Abstract class for connection proxies. Should be extended when adding new connection classes.
+    :param timeout: The timeout of the underlying socket.
+    :param connection_config: The defined connection config used for defining the address the socket should listen on.
+    :param message_handler: The message handler object every received message is passed to.
+    """
+
+    def __init__(self, timeout: int, connection_config: ConnectionConfig, message_handler: MessageHandler):
+        self.connection_config = connection_config
+        self.message_handler = message_handler
+        self.timeout = timeout
+
+        self._running = False
+
+    @staticmethod
+    def debug(message, prefix=""):
+        logging.debug(f"{prefix}: {message}")
+
+    def stop(self):
+        """
+        Stops the execution of the proxy manually.
+        """
+        self._running = False
+
+    def is_running(self) -> bool:
+        """
+        Checks if the proxy is currently running.
+        :return: True if proxy still running, False otherwise.
+        """
+        return self._running
+
+
+class TCPProxy(AbstractProxy):
+    """
+    Class establishes a TCP endpoint with the given connection config.
+    It receives messages and sends it to the message handler.
+    :param timeout: The timeout of the underlying UDP socket.
+    :param connection_config: The defined connection config used for defining the address the socket should listen on.
+    :param message_handler: The message handler object every received message is passed to.
+    """
+    def __init__(self, timeout: int, connection_config: ConnectionConfig, message_handler: MessageHandler):
+        super().__init__(timeout, connection_config, message_handler)
+        self.tcp_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.tcp_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+
+    def _receive(self, client_socket: WrappedSocket, address: tuple):
+        """
+        Receives messages on given socket and forwards it to the message handler.
+        Tries to receive until the TCPProxy is stopped manually or TCP connection is finished.
+        :param client_socket: Socket to receive data from.
+        :param address: The connection address of the TCP connection.
+        """
+        dns_data = b""
+        while self._running:
+            try:
+                data = client_socket.recv(4096)
+                data_length = int.from_bytes(data[:2], "big")
+                data = data[2:]
+                dns_data += data
+                if data_length > len(dns_data):
+                    continue
+                if not data:
+                    AbstractProxy.debug("No data received! -> Close receiving socket.", "TCPProxy")
+                    client_socket.try_close()
+                    break
+                else:
+                    self.message_handler.handle_message(client_socket, data)
+            except BrokenPipeError as e:
+                AbstractProxy.debug(f"{str(address)}: Forwarding broken with {e}", "TCPProxy")
+                client_socket.close()
+                self._running = False
+            except TimeoutError as e:
+                AbstractProxy.debug(f"{str(address)}: Timeout occurred for TCP socket", "TCPProxy")
+            except ConnectionResetError:
+                AbstractProxy.debug(f"{str(address)}: Connection was reset! Stop listening on socket!", "TCPProxy")
+                break
+            except Exception as e:
+                AbstractProxy.debug(f"{str(address)}: Exception while forwarding: {e} {type(e)}", "TCPProxy")
+                client_socket.close()
+                self._running = False
+
+        client_socket.try_close()
+        logging.info(f"TCP: Closed connection")
+
+    def _accept_socket_connection(self):
+        """
+        Accepts new TCP connections and starts a new thread using the _receive method for receiving messages on the
+        established connection.
+        This accepts new connections until TCPProxy is manually stopped.
+        """
+        self.tcp_server.settimeout(self.timeout)
+        while self._running:  # listen for incoming connections
+            try:
+                client_socket, address = self.tcp_server.accept()
+                wrapped_socket = TCPSocket(self.timeout, client_socket)
+                AbstractProxy.debug(f"TCPProxy: request from {address[0]}:{address[1]}", "TCPProxy")
+                # spawn new thread to handle incoming connection
+                threading.Thread(target=self._receive, args=(wrapped_socket, address)).start()
+            except TimeoutError:
+                AbstractProxy.debug("TCP socket timeout out while accepting connection", "TCPProxy")
+            except Exception as e:
+                logging.info("Error occurred while accepting TCP connection!" + str(e))
+                break
+        AbstractProxy.debug(f"TCPProxy: Done waiting for connections", "TCPProxy")
+
+    def start(self):
+        """
+        Starts the TCPProxy by binding to the given address and accepting connections with calling
+        _accept_socket_connection.
+        """
+        self.tcp_server.bind((self.connection_config.host, self.connection_config.port))
+        logging.info(f"Receiving TCP messages on {self.connection_config.host}:{self.connection_config.port}")
+        self._running = True
+        # opening server socket
+        self.tcp_server.listen()
+        # spawn a thread for each socket
+        self._accept_socket_connection()
+
+
+class UDPProxy(AbstractProxy):
+    """
+    Class establishes a UDP endpoint with the given connection config.
+    It receives messages and sends it to the message handler.
+    :param timeout: The timeout of the underlying UDP socket.
+    :param connection_config: The defined connection config used for defining the address the socket should listen on.
+    :param message_handler: The message handler object every received message is passed to.
+    """
+    def __init__(self, timeout: int, connection_config: ConnectionConfig, message_handler: MessageHandler):
+        super().__init__(timeout, connection_config, message_handler)
+
+        self.udp_server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self.udp_server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+
+        self.udp_client_address = None
+
+    def _receive(self, client_socket: WrappedSocket):
+        """
+        Receives messages on given socket and forwards it to the message handler.
+        Tries to receive until the UDPProxy is stopped manually.
+        :param client_socket: Socket to receive data from.
+        """
+        AbstractProxy.debug("Start receive UDP: ")
+        while self._running:
+            try:
+                data, address = client_socket.recvfrom(1024)
+
+                AbstractProxy.debug(f"UDPProxy: request from {address[0]}:{address[1]}", "UDPProxy")
+
+                AbstractProxy.debug(f"Received Data: {data}", "UDPProxy")
+                if not data:
+                    AbstractProxy.debug("Connection closed, closing proxy socket", "UDPProxy")
+                    client_socket.try_close()
+                    break
+                else:
+                    new_client_socket = ClientSocket(client_socket, address)
+                    self.message_handler.handle_message(new_client_socket, data)
+
+            except BrokenPipeError as e:
+                AbstractProxy.debug(f"Forwarding broken with {e}", "UDPProxy")
+                client_socket.close()
+                self._running = False
+            except TimeoutError as e:
+                AbstractProxy.debug(f"Timeout occurred for UDP socket", "UDPProxy")
+            except ConnectionResetError:
+                AbstractProxy.debug(f"Connection was reset! Receive next message", "UDPProxy")
+            except Exception as e:
+                AbstractProxy.debug(f"Exception while forwarding: {e} {type(e)}", "UDPProxy")
+                client_socket.close()
+                self._running = False
+
+        logging.info(f"UDP: Closed connection")
+
+    def start(self):
+        """
+        Starts the UDPProxy by binding the socket and starting the _receive method.
+        """
+        self.udp_server.bind((self.connection_config.host, self.connection_config.port))
+        logging.info(f"Receiving UDP messages on {self.connection_config.host}:{self.connection_config.port}")
+        self._running = True
+        # start listening
+        udp_client_socket = WrappedSocket(self.timeout, self.udp_server)
+        self._receive(udp_client_socket)
+
+
+class DNSProxy:
+    """
+    This class establishes a possible UDP and TCP endpoint that can be used to send DNS queries.
+    """
+    def __init__(self):
+        self.udp_proxy = None
+        self.tcp_proxy = None
+        self._udp_thread = None
+        self._tcp_thread = None
+
+    def is_running(self) -> bool:
+        """
+        Checks if DNSProxy is running. It checks if either the UDP or the TCP proxy is stopped. When one proxy is
+        stopped the DNSProxy is considered as stopped.
+        :return: If the UDP and TCP proxies are running return True, False if one proxy is stopped.
+        """
+        return (self.udp_proxy is not None and self.udp_proxy.is_running()
+                or self.tcp_proxy is not None and self.tcp_proxy.is_running())
+
+    def start(self, timeout: int, connection_config: ConnectionConfig, message_handler: MessageHandler):
+        """
+        Starts the DNSProxy by starting two thread for the UDP and TCP connection.
+        :param timeout: The timeout that is used for TCP and UDP sockets.
+        :param connection_config: The connection config of the tool to define the host and port the proxies should
+        listen on.
+        :param message_handler: The message handler object that is used for handling incoming DNS messages.
+        """
+        if self._udp_thread is not None or self._tcp_thread is not None:
+            return
+        self.udp_proxy = UDPProxy(timeout, connection_config, message_handler)
+
+        self.tcp_proxy = TCPProxy(timeout, connection_config, message_handler)
+
+        self._udp_thread = threading.Thread(target=self.udp_proxy.start)
+        self._udp_thread.start()
+
+        self._tcp_thread = threading.Thread(target=self.tcp_proxy.start)
+        self._tcp_thread.start()
+
+    def stop(self):
+        """
+        Stops receiving incoming messages by stopping the UDP and TCP proxies.
+        """
+        if self.udp_proxy is None and self.tcp_proxy is None:
+            return
+
+        if self.udp_proxy is not None:
+            self.udp_proxy.stop()
+
+        if self.tcp_proxy is not None:
+            self.tcp_proxy.stop()
+
+        if self.udp_proxy is not None:
+            self._udp_thread.join()
+            self._udp_thread = None
+
+        if self.tcp_proxy is not None:
+            self._tcp_thread.join()
+            self._tcp_thread = None

CensorBreaker/network/WrappedSocket.py

+import socket
+from time import time
+from typing import Tuple, Any
+
+from CensorBreaker.exception.ParserException import ParserException
+
+
+class WrappedSocket:
+    """
+    Wraps a socket with useful utility functions.
+    """
+
+    def __init__(self, timeout: int, _socket: socket.socket):
+        self.timeout = timeout
+        self.buffer = b''
+        self.socket = _socket
+        self.socket.settimeout(timeout)
+
+    def read(self, size: int) -> bytes:
+        """
+        Reads specified amount of data from socket. Blocks until amount of data received or timeout.
+        :param size: Data to read.
+        :return: Read data
+        """
+        while len(self.buffer) < size:
+            self.buffer += self.socket.recv(4096)
+        _res = self.buffer[:size]
+        self.buffer = self.buffer[size:]
+        return _res
+
+    def read_until(self, until: list[bytes], max_len: int = 100) -> bytes:
+        """
+        Returns all bytes from the socket until and including the given bytes. Also cancels after timeout
+        :param until: Bytes until which to receive
+        :param max_len: Max length until which to search
+        :return: Read data
+        """
+        start_time = time()
+        while len(list(filter(lambda x: x in self.buffer, until))) == 0:
+            if len(self.buffer) > max_len:
+                raise ParserException(f"Exceeded max length of {max_len} bytes")
+            if time() - start_time > self.timeout:
+                raise ParserException(f"Exceeded timeout of {self.timeout}s")
+            self.buffer += self.socket.recv(4096)
+        until = list(filter(lambda x: x in self.buffer, until))[0]
+        index = self.buffer.index(until) + len(until)
+        _res = self.buffer[:index]
+        self.buffer = self.buffer[index:]
+        return _res
+
+    def peek(self, size: int) -> bytes:
+        """
+        Similar to read, but keeps data in buffer.
+        """
+        while len(self.buffer) < size:
+            self.buffer += self.socket.recv(4096)
+        return self.buffer[:size]
+
+    def recv(self, size: int, *args, **kwargs) -> bytes:
+        """
+        Works similar to recv of the wrapped socket. Prepends any bytes still buffered.
+        :param size: Size of the buffer to read into.
+        :return: Bytes read from the socket
+        """
+        if len(self.buffer) > 0:
+            _res = self.buffer
+            self.buffer = b''
+        else:
+            _res = self.socket.recv(size, *args, **kwargs)
+        return _res
+
+    def send(self, data: bytes, *args, **kwargs) -> int:
+        """
+        Wraps send() of the wrapped socket. Split into tcp fragments if given as value.
+        :return: Return value of the wrapped socket's send method
+        """
+        return self.socket.send(data, *args, **kwargs)
+
+    def sendto(self, message: bytes, address: tuple) -> int:
+        """
+        Wraps sendto() of the wrapped socket.
+        :return: Return value of the wrapped socket's sendto method
+        """
+        return self.socket.sendto(message, address)
+
+    def recvfrom(self, bufsize: int) -> tuple[bytes, Any]:
+        """
+        Wraps recvfrom() of the wrapped socket.
+        :return: Return value of the wrapped socket's recvfrom method
+        """
+        return self.socket.recvfrom(bufsize)
+
+    def close(self):
+        """
+        Closes the underlying socket.
+        """
+        self.socket.close()
+
+    def try_close(self):
+        """
+        Tries to close the underlying socket. If that fails, we ignore the error.
+        """
+        try:
+            self.socket.close()
+        except:
+            pass
+
+    def inject(self, content: bytes):
+        """
+        Injects bytes to the front of the buffer. Can be used to write back read data.
+        :param content: the bytes to prepend
+        :return: None
+        """
+        self.buffer = content + self.buffer
+
+
+class TCPSocket(WrappedSocket):
+    """
+    TCP socket that is used for sending DNS messages.
+    It can use TCP fragmentation to send messages.
+    Further, it automatically adds the needed length of the message in front of the message.
+    :param timeout: The timeout of the underlying socket.
+    :param _socket: The socket that should be wrapped.
+    :param tcp_fragment_size: (Optional) Defines the size of fragments when TCP fragmentation is used. By default set to
+    0, which disables TCP fragmentation.
+    """
+
+    def __init__(self, timeout: int, _socket: socket.socket, tcp_fragment_size=0):
+        self.tcp_frag_size = tcp_fragment_size
+        super().__init__(timeout, _socket)
+
+    def send(self, data: bytes, *args, **kwargs) -> int:
+        answer_len = len(data)
+        byte_answer = answer_len.to_bytes(2, "big") + data
+
+        if self.tcp_frag_size <= 0:
+            # Just send message normally
+            super().send(byte_answer, *args, **kwargs)
+        else:
+            # split into fragments and send each separately
+            fragments = (byte_answer[i:i + self.tcp_frag_size] for i in range(0, len(byte_answer), self.tcp_frag_size))
+            for fragment in fragments:
+                self.socket.send(fragment, *args, **kwargs)
+                self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 0)
+                self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+
+
+class ClientSocket:
+    """
+    UDP socket that is used for sending messages back to the sender.
+    It wraps the given wrapped socket and the senders address. Therefore, only send needs to be called to automatically
+    send the message back to its address without needing to know the address.
+    :param wrapped_socket: The socket the class should use to send messages.
+    :param address: The address the send method should send messages to.
+    """
+
+    def __init__(self, wrapped_socket: WrappedSocket, address: tuple):
+        self.wrapped_socket = wrapped_socket
+        self.address = address
+
+    def send(self, message):
+        self.wrapped_socket.sendto(message, self.address)