Merge pull request #46 from jstucke/refactoring

add CI tests

.github/workflows/tests.yml

@@ -0,0 +1,32 @@
+name: Tests CI
+run-name: Tests CI
+on:
+  pull_request:
+    branches:
+      - main
+  push:
+    branches:
+      - main
+  workflow_dispatch:
+
+jobs:
+  tests:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.8", "3.9", "3.10", "3.11"]
+      fail-fast: false
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Display Python version
+      run: python -c "import sys; print(sys.version)"
+    - name: Install dependencies
+      run: python -m pip install --upgrade pip setuptools wheel
+    - name: Installation
+      run: python -m pip install ".[test]"
+    - name: Unit Tests
+      run: pytest -v ./tests

.gitignore

@@ -0,0 +1,67 @@
+### Python template
+__pycache__/
+
+# 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
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# IDEs
+.spyderproject
+.spyproject
+.idea/
+.vscode/
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/

honeypots/dhcp_server.py

@@ -9,22 +9,19 @@
 //  contributors list qeeqbox/honeypots/graphs/contributors
 //  -------------------------------------------------------------
 '''
-
 from warnings import filterwarnings
 filterwarnings(action='ignore', module='.*OpenSSL.*')
 filterwarnings(action='ignore', module='.*socket.*')
 
 from twisted.internet.protocol import DatagramProtocol
 from twisted.internet import reactor
-from time import time
 from twisted.python import log as tlog
-from struct import unpack
+from struct import unpack, error as StructError
 from socket import inet_aton
 from subprocess import Popen
 from os import path, getenv
 from honeypots.helper import close_port_wrapper, get_free_port, kill_server_wrapper, server_arguments, setup_logger, disable_logger, set_local_vars, check_if_server_is_running
 from uuid import uuid4
-from contextlib import suppress
 
 
 class QDHCPServer():
@@ -95,7 +92,10 @@ def parse_options(self, raw):
                 return options
 
             def datagramReceived(self, data, addr):
-                mac_address = unpack('!28x6s', data[:34])[0].hex(':')
+                try:
+                    mac_address = unpack('!28x6s', data[:34])[0].hex(':')
+                except StructError:
+                    mac_address = "None"
                 data = self.parse_options(data[240:])
                 data.update({'mac_address': mac_address})
                 _q_s.logs.info({'server': 'dhcp_server', 'action': 'query', 'status': 'success', 'src_ip': addr[0], 'src_port': addr[1], 'dest_ip': _q_s.ip, 'dest_port': _q_s.port, 'data': data})

honeypots/https_server.py

@@ -177,7 +177,7 @@ def check_bytes(string):
                     request.responseHeaders.addRawHeader('Server', _q_s.mocking_server)
 
                 if request.method == b'GET' or request.method == b'POST':
-                    _q_s.logs.info({'server': 'https_server', 'action': request.method.decode(), 'src_ip': request.client_ip, 'src_port': request.getClientAddress().port, 'dest_ip': _q_s.ip, 'dest_port': _q_s.port})
+                    _q_s.logs.info({'server': 'https_server', 'action': request.method.decode(), 'src_ip': client_ip, 'src_port': request.getClientAddress().port, 'dest_ip': _q_s.ip, 'dest_port': _q_s.port})
 
                 if request.method == b'GET':
                     if request.uri == b'/login.html':

setup.py

@@ -4,37 +4,44 @@
     long_description = f.read()
 
 setup(
-    name='honeypots',
-    author='QeeqBox',
-    author_email='[email protected]',
+    name="honeypots",
+    author="QeeqBox",
+    author_email="[email protected]",
     description=r"30 different honeypots in one package! (dhcp, dns, elastic, ftp, http proxy, https proxy, http, https, imap, ipp, irc, ldap, memcache, mssql, mysql, ntp, oracle, pjl, pop3, postgres, rdp, redis, sip, smb, smtp, snmp, socks5, ssh, telnet, vnc)",
     long_description=long_description,
-    version='0.64',
+    version="0.64",
     license="AGPL-3.0",
-    license_files=('LICENSE'),
+    license_files=("LICENSE"),
     url="https://github.com/qeeqbox/honeypots",
-    packages=['honeypots'],
-    entry_points={
-        "console_scripts": [
-            'honeypots=honeypots.__main__:main_logic'
-        ]
-    },
+    packages=["honeypots"],
+    entry_points={"console_scripts": ["honeypots=honeypots.__main__:main_logic"]},
     include_package_data=True,
     install_requires=[
-        'twisted==21.7.0',
-        'psutil==5.9.0',
-        'psycopg2-binary==2.9.3',
-        'pycrypto==2.6.1',
-        'requests==2.28.2',
-        'requests[socks]==2.28.2',
-        'impacket==0.9.24',
-        'paramiko==3.1.0',
-        'scapy==2.4.5',
-        'service_identity==21.1.0',
-        'netifaces==0.11.0'
+        "twisted==21.7.0",
+        "psutil==5.9.0",
+        "psycopg2-binary==2.9.3",
+        "pycryptodome==3.19.0",
+        "requests==2.28.2",
+        "requests[socks]==2.28.2",
+        "impacket==0.9.24",
+        "paramiko==3.1.0",
+        "scapy==2.4.5",
+        "service_identity==21.1.0",
+        "netifaces==0.11.0",
     ],
     extras_require={
-        'test': ['redis', 'mysql-connector', 'elasticsearch', 'pymssql', 'ldap3', 'pysnmp']
+        "test": [
+            "redis",
+            "mysql-connector",
+            "dnspython==2.4.2",
+            "elasticsearch",
+            "pymssql",
+            "ldap3",
+            "pysnmplib",
+            "pytest",
+            "redis",
+            "vncdotool",
+        ]
     },
-    python_requires='>=3.5'
+    python_requires=">=3.8",
 )

tests/conftest.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import json
+from contextlib import contextmanager
+from multiprocessing import Process
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from typing import Iterator
+
+import pytest
+
+from .utils import IP, PASSWORD, USERNAME
+
+
+@contextmanager
+def config_for_testing(custom_config: dict) -> Iterator[Path]:
+    with TemporaryDirectory() as tmp_dir:
+        config = Path(tmp_dir) / "config.json"
+        logs_output_dir = Path(tmp_dir) / "logs"
+        logs_output_dir.mkdir()
+        testing_config = {
+            "logs": "file,terminal,json",
+            "logs_location": str(logs_output_dir.absolute()),
+            **custom_config,
+        }
+        config.write_text(json.dumps(testing_config))
+        yield config
+
+
+@pytest.fixture
+def server_logs(request):
+    custom_config = request.param.get("custom_config", {})
+    with config_for_testing(custom_config) as config_file:
+        _server = request.param["server"](
+            ip=IP,
+            port=request.param["port"],
+            username=USERNAME,
+            password=PASSWORD,
+            options="",
+            config=str(config_file.absolute()),
+        )
+        server_process = Process(target=_server.run_server)
+        server_process.start()
+        yield config_file.parent / "logs"
+        server_process.terminate()
+        server_process.join()

tests/test_dhcp_server.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from time import sleep
+
+import pytest
+
+from honeypots import QDHCPServer
+from .utils import (
+    connect_to,
+    EXPECTED_KEYS,
+    IP,
+    load_logs_from_file,
+)
+
+PORT = "50067"
+
+
+@pytest.mark.parametrize(
+    "server_logs",
+    [{"server": QDHCPServer, "port": PORT}],
+    indirect=True,
+)
+def test_dhcp_server(server_logs):
+    sleep(1)  # give the server some time to start
+
+    with connect_to(IP, PORT, udp=True) as connection:
+        connection.send(b"\x03" * 240)
+
+    sleep(1)  # give the server process some time to write logs
+
+    logs = load_logs_from_file(server_logs)
+
+    assert len(logs) == 1
+    (query,) = logs
+    assert all(k in query for k in EXPECTED_KEYS)
+    assert query["action"] == "query"
+    assert query["status"] == "success"
+    assert query["data"] == {"mac_address": "03:03:03:03:03:03"}

tests/test_dns_server.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from time import sleep
+
+import pytest
+from dns.resolver import Resolver
+
+from honeypots import QDNSServer
+from .utils import (
+    assert_connect_is_logged,
+    IP,
+    load_logs_from_file,
+)
+
+PORT = "50053"
+
+
+@pytest.mark.parametrize(
+    "server_logs",
+    [{"server": QDNSServer, "port": PORT}],
+    indirect=True,
+)
+def test_dns_server(server_logs):
+    sleep(1)  # give the server some time to start
+
+    resolver = Resolver(configure=False)
+    resolver.nameservers = [IP]
+    resolver.port = int(PORT)
+    domain = "example.org"
+    response = resolver.resolve(domain, "a")
+
+    sleep(1)  # give the server process some time to write logs
+
+    logs = load_logs_from_file(server_logs)
+
+    assert len(logs) == 2
+    connect, query = logs
+    assert_connect_is_logged(connect, PORT)
+
+    assert query["action"] == "query"
+    assert "data" in query
+    assert "<A address=" in query["data"]
+
+    assert domain in response.canonical_name.to_text()

tests/test_elastic_server.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from contextlib import suppress
+from time import sleep
+
+import pytest
+from elasticsearch import Elasticsearch, NotFoundError
+
+from honeypots import QElasticServer
+from .utils import (
+    assert_connect_is_logged,
+    assert_login_is_logged,
+    IP,
+    load_logs_from_file,
+    PASSWORD,
+    USERNAME,
+)
+
+PORT = "59200"
+
+
+@pytest.mark.parametrize(
+    "server_logs",
+    [{"server": QElasticServer, "port": PORT}],
+    indirect=True,
+)
+def test_elastic_server(server_logs):
+    sleep(1)  # give the server some time to start
+
+    with suppress(NotFoundError):
+        elastic = Elasticsearch(
+            [f"https://{IP}:{PORT}"],
+            basic_auth=(USERNAME, PASSWORD),
+            verify_certs=False,
+        )
+        elastic.search(index="test", body={}, size=99)
+
+    sleep(1)  # give the server process some time to write logs
+
+    logs = load_logs_from_file(server_logs)
+
+    assert len(logs) == 2
+    connect, login = logs
+    assert_connect_is_logged(connect, PORT)
+    assert_login_is_logged(login)