Init commit

dns-server/.gitignore

@@ -0,0 +1,54 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+
+# C extensions
+*.so
+
+# Distribution / packaging
+bin/
+build/
+develop-eggs/
+dist/
+eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+.tox/
+.coverage
+.cache
+nosetests.xml
+coverage.xml
+
+# Translations
+*.mo
+
+# Mr Developer
+.mr.developer.cfg
+.project
+.pydevproject
+
+# Rope
+.ropeproject
+
+# Django stuff:
+*.log
+*.pot
+
+# Sphinx documentation
+docs/_build/
+
+.venv/
+venv/
+__pycache__/

dns-server/Makefile

@@ -0,0 +1,19 @@
+
+VENV = .venv
+PYTHON = $(VENV)/bin/python3
+
+build:
+ifeq (,$(wildcard $(VENV)))
+	@echo "Creating virtualenv..."
+	python3 -m venv $(VENV)
+endif
+
+run-server: build
+	@echo "Activating virtualenv..."
+	. $(VENV)/bin/activate
+	@echo "Running server..."
+	$(PYTHON) src/dns_server.py
+
+clean:
+	@echo "Cleaning virtualenv..."
+	rm -rf $(VENV)

dns-server/README.md

@@ -0,0 +1,50 @@
+# Information
+Full Name: Mohammad Amin Karami
+
+Student ID: 98105998
+
+# How to Run the Code
+The project is written in Python. To execute the code, simply run the following command in the console:
+
+```bash
+make run-server
+```
+
+Additionally, to clean up the virtual environment, you can run the following command in the console:
+
+```bash
+make clean
+```
+
+# Code Description
+The implementation is located in the `src` directory.
+
+In the `dns_database.py` file, there is a class responsible for reading from the `/etc/myhosts` file and storing the records related to the IP of each host in a dictionary. For example, for the record:
+
+93.184.216.34 example.com
+
+The dictionary will look like this:
+
+```py
+data = {
+    "example.com": "93.184.216.34"
+}
+```
+
+You can retrieve the IP address of a host using the `get_ip_of_domain` function if it exists.
+
+The `dns_server.py` file contains the implementation of the server, which is implemented in parallel using the `async.io` library.
+
+Packet handling implementation for the DNS server is done in `dns_packet`.
+
+The `Serializable` class is an interface for classes that require `parse` and `pack` methods, and this class is inherited by sections and the `DNSPacket` itself.
+
+Each packet has 5 sections, and each section is implemented in the `sections` directory.
+
+The `DNSPacket` class uses these sections and uses the `parse` and `pack` methods of each section for parsing and packing.
+
+The `helpers.py` file contains functions that make data extraction and creation easier. For example, the `convert_number_to_bit_string` function allows converting a number into a bit sequence of a specified length.
+
+The `constants.py` file contains Enums such as `RType`, which improves code readability.
+
+In the `sections` directory, each section related to the packet is implemented. Since `Answer`, `Authority`, and `Additional` are all of type RR, an `RR` class is also implemented, and these sections inherit from it.

dns-server/src/dns_database.py

@@ -0,0 +1,23 @@
+from typing import Dict, Optional
+
+
+class DNSDatabase(object):
+    def __init__(self, source_file: str = "/etc/myhosts") -> None:
+        self._source_file: str = source_file
+
+        self._data: Dict[str, str] = self._load_data(source_file)
+
+    def _load_data(self, source_file: str):
+        data: Dict[str, str] = {}
+        with open(source_file, "r") as data_file:
+            for record in data_file.readlines():
+                ip, host = record.split()
+                data[host.lower()] = ip
+
+        return data
+
+    def get_ip_of_domain(self, domain: str) -> Optional[str]:
+        return self._data.get(domain, None)
+
+
+dns_database = DNSDatabase(source_file="/etc/myhosts")

dns-server/src/dns_packet/__init__.py

@@ -0,0 +1,9 @@
+from .constants import RClass, RCode, RType
+from .dns_packet import DNSPacket
+
+__all__ = [
+    "DNSPacket",
+    "RType",
+    "RClass",
+    "RCode",
+]

dns-server/src/dns_packet/constants.py

@@ -0,0 +1,17 @@
+from enum import Enum
+
+
+class RType(int, Enum):
+    A = 1
+
+
+class RClass(int, Enum):
+    IN = 1
+
+
+class RCode(int, Enum):
+    NO_ERROR = 0
+
+
+class Opcode(int, Enum):
+    STANDART_QUERY = 0

dns-server/src/dns_packet/dns_packet.py

@@ -0,0 +1,48 @@
+from typing import List
+
+from dns_packet.sections import Additional, Answer, Authority, Header, Question
+from dns_packet.serializable import Serializable
+
+
+class DNSPacket(Serializable):
+    def __init__(self, header: Header) -> None:
+        self.header: Header = header
+        self.questions: List[Question] = []
+        self.answers: List[Answer] = []
+        self.authorities: List[Authority] = []
+        self.additionals: List[Additional] = []
+
+    def add_question(self, question: Question):
+        self.questions.append(question)
+        self._update_header()
+
+    def add_answer(self, answer: Answer):
+        self.answers.append(answer)
+        self._update_header()
+
+    def _update_header(self):
+        self.header.qdcount = len(self.questions)
+        self.header.ancount = len(self.answers)
+        self.header.nscount = len(self.authorities)
+        self.header.arcount = len(self.additionals)
+
+    @classmethod
+    def parse(cls, packet: bytes) -> "DNSPacket":
+        header, next_start = Header.parse(packet=packet)
+        dns_packet = DNSPacket(header=header)
+        question = Question.parse(packet=packet, starting_index=next_start)
+        dns_packet.add_question(question)
+
+        return dns_packet
+
+    def pack(self) -> bytes:
+        self._update_header()
+        data: bytes = self.header.pack()
+
+        for question in self.questions:
+            data += question.pack()
+
+        for answer in self.answers:
+            data += answer.pack()
+
+        return data

dns-server/src/dns_packet/helpers.py

@@ -0,0 +1,47 @@
+def convert_number_to_bit_string(number: int, bit_length: int) -> str:
+    bit_string = bin(number)[2:]
+
+    bit_string = bit_string.zfill(bit_length)
+
+    return bit_string
+
+
+def convert_bytes_to_bit_string(byte_sequence):
+    bit_string = "".join(format(byte, "08b") for byte in byte_sequence)
+
+    return bit_string
+
+
+def convert_bit_string_to_bytes(bit_string) -> bytes:
+    bit_string = bit_string.zfill((len(bit_string) + 7) // 8 * 8)
+
+    byte_string = bytes(
+        int(bit_string[i : i + 8], 2) for i in range(0, len(bit_string), 8)
+    )
+
+    return byte_string
+
+
+def convert_bytes_to_domain_name(bytes: bytes, starting_index: int):
+    domain_name = ""
+    i = starting_index
+    while i < len(bytes):
+        label_length = bytes[i]
+        if label_length == 0:
+            break
+        label = bytes[i + 1 : i + 1 + label_length].decode()
+        domain_name += label + "."
+        i += 1 + label_length
+
+    return domain_name.rstrip("."), i + 1
+
+
+def convert_domain_name_to_bytes(domain_name: str) -> bytes:
+    bytes_sequence = bytearray()
+    labels = domain_name.split(".")
+    for label in labels:
+        label_length = len(label)
+        bytes_sequence.append(label_length)
+        bytes_sequence.extend(label.encode())
+    bytes_sequence.append(0)
+    return bytes_sequence

dns-server/src/dns_packet/sections/__init__.py

@@ -0,0 +1,13 @@
+from .additional import Additional
+from .answer import Answer
+from .authority import Authority
+from .header import Header
+from .question import Question
+
+__all__ = [
+    "Additional",
+    "Answer",
+    "Authority",
+    "Header",
+    "Question",
+]

dns-server/src/dns_packet/sections/additional.py

@@ -0,0 +1,5 @@
+from .rr import RR
+
+
+class Additional(RR):
+    pass

dns-server/src/dns_packet/sections/answer.py

@@ -0,0 +1,5 @@
+from .rr import RR
+
+
+class Answer(RR):
+    pass

dns-server/src/dns_packet/sections/authority.py

@@ -0,0 +1,5 @@
+from .rr import RR
+
+
+class Authority(RR):
+    pass

dns-server/src/dns_packet/sections/header.py

@@ -0,0 +1,56 @@
+from dns_packet.constants import Opcode, RCode
+from dns_packet.helpers import (
+    convert_bit_string_to_bytes,
+    convert_bytes_to_bit_string,
+    convert_number_to_bit_string,
+)
+
+
+class Header(object):
+    def __init__(self, id: str, rd: int, rcode: int) -> None:
+        self.id = id
+        self.qr = 1
+        self.opcode = Opcode.STANDART_QUERY
+
+        self.aa = 0  # Not considered
+        self.tc = 0  # Not considered
+        self.rd = rd
+        self.ra = 0  # Not considered
+        self.z = 0  # Not considered
+
+        self.rcode = rcode
+
+        self.qdcount = 0
+        self.ancount = 0
+        self.nscount = 0
+        self.arcount = 0
+
+    @classmethod
+    def parse(cls, packet: bytes, starting_index=0):
+        end_index = 12
+        header = packet[starting_index:end_index]
+
+        bit_string = convert_bytes_to_bit_string(header)
+
+        id = bit_string[0:16]
+        rd = int(bit_string[23])
+
+        return cls(id=id, rd=rd, rcode=RCode.NO_ERROR), end_index
+
+    def pack(self) -> bytes:
+        bit_string: str = (
+            self.id
+            + convert_number_to_bit_string(number=self.qr, bit_length=1)
+            + convert_number_to_bit_string(number=self.opcode, bit_length=4)
+            + convert_number_to_bit_string(number=self.aa, bit_length=1)
+            + convert_number_to_bit_string(number=self.tc, bit_length=1)
+            + convert_number_to_bit_string(number=self.rd, bit_length=1)
+            + convert_number_to_bit_string(number=self.ra, bit_length=1)
+            + convert_number_to_bit_string(number=self.z, bit_length=3)
+            + convert_number_to_bit_string(number=self.rcode, bit_length=4)
+            + convert_number_to_bit_string(number=self.qdcount, bit_length=16)
+            + convert_number_to_bit_string(number=self.ancount, bit_length=16)
+            + convert_number_to_bit_string(number=self.nscount, bit_length=16)
+            + convert_number_to_bit_string(number=self.arcount, bit_length=16)
+        )
+        return convert_bit_string_to_bytes(bit_string)