Merge pull request #63 from fkie-cad/pre-commit-github-action

Pre commit GitHub action

.github/workflows/pre-commit.yaml

@@ -0,0 +1,17 @@
+name: pre-commit
+
+on:
+  pull_request:
+
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          # requites to grab the history of the PR
+          fetch-depth: 0
+      - uses: actions/setup-python@v3
+      - uses: pre-commit/[email protected]
+        with:
+          extra_args: --from-ref ${{ github.event.pull_request.base.sha }} --to-ref ${{ github.event.pull_request.head.sha }}

.github/workflows/tests.yml

@@ -2,11 +2,8 @@ name: Tests CI
 run-name: Tests CI
 on:
   pull_request:
-    branches:
-      - main
   push:
-    branches:
-      - main
+    branches: [main]
   workflow_dispatch:
 
 jobs:
@@ -29,4 +26,4 @@ jobs:
     - name: Installation
       run: python -m pip install ".[dev]"
     - name: Unit Tests
-      run: pytest -v ./tests
+      run: pytest -v -n 3 ./tests

honeypots/__init__.py

@@ -5,12 +5,9 @@
 from .helper import (
     is_privileged,
     clean_all,
-    close_port_wrapper,
     get_free_port,
-    get_running_servers,
-    kill_server_wrapper,
     kill_servers,
-    postgres_class,
+    PostgresClass,
     server_arguments,
     set_local_vars,
     setup_logger,
@@ -76,12 +73,9 @@
     "QVNCServer",
     "is_privileged",
     "clean_all",
-    "close_port_wrapper",
     "get_free_port",
-    "get_running_servers",
-    "kill_server_wrapper",
     "kill_servers",
-    "postgres_class",
+    "PostgresClass",
     "server_arguments",
     "set_local_vars",
     "setup_logger",

honeypots/__main__.py

@@ -168,7 +168,7 @@ def main(self):
 
         if self.options.list:
             for service in all_servers:
-                print(service)  # noqa: T201
+                print(service)
         elif self.options.kill:
             clean_all()
         elif self.options.chameleon and self.config_data is not None:
@@ -350,7 +350,7 @@ def _setup_logging(self) -> logging.Logger:
     def _start_sniffer(self, sniffer_filter, sniffer_interface):
         logger.info("[x] Starting sniffer")
         sniffer = QBSniffer(
-            filter=sniffer_filter,
+            filter_=sniffer_filter,
             interface=sniffer_interface,
             config=self.options.config,
         )

honeypots/base_server.py

@@ -1,13 +1,16 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+from contextlib import suppress
 from multiprocessing import Process
 from os import getenv
+from socket import AF_INET, SOCK_STREAM, socket
 from typing import Any
 from uuid import uuid4
 
+from psutil import process_iter, TimeoutExpired
+
 from honeypots.helper import (
-    close_port_wrapper,
     get_free_port,
     service_has_started,
     set_local_vars,
@@ -58,7 +61,24 @@ def __init__(self, **kwargs):
         self._server_process: Process | None = None
 
     def close_port(self):
-        return close_port_wrapper(self.NAME, self.ip, self.port, self.logs)
+        sock = socket(AF_INET, SOCK_STREAM)
+        sock.settimeout(2)
+        if sock.connect_ex((self.ip, self.port)) == 0:
+            for process in process_iter():
+                with suppress(Exception):
+                    for conn in process.connections(kind="inet"):
+                        if self.port == conn.laddr.port:
+                            process.terminate()
+                            try:
+                                process.wait(timeout=5)
+                            except TimeoutExpired:
+                                process.kill()
+        with suppress(OSError):
+            sock.bind((self.ip, self.port))
+            if sock.connect_ex((self.ip, self.port)) != 0:
+                return True
+        self.logger.error(f"[{self.NAME}]: Could not close port {self.port}")
+        return False
 
     def kill_server(self):
         if self._server_process:

honeypots/dhcp_server.py

@@ -11,7 +11,7 @@
 """
 
 from socket import inet_aton
-from struct import error as StructError, unpack
+import struct
 
 from twisted.internet import reactor
 from twisted.internet.protocol import DatagramProtocol
@@ -42,7 +42,7 @@ def payload(self, value, message):
                     siaddr,
                     giaddr,
                     chaddr,
-                ) = unpack("1s1s1s1s4s2s2s4s4s4s4s16s", message[:44])
+                ) = struct.unpack("1s1s1s1s4s2s2s4s4s4s4s16s", message[:44])
                 # op, htype, hlen, hops, xid, secs, flags, ciaddr
                 response = b"\x02\x01\x06\x00" + xid + b"\x00\x00\x00\x00\x00\x00\x00\x00"
                 # yiaddr, siaddr, giaddr, chaddr
@@ -69,27 +69,26 @@ def parse_options(self, raw):
                 tag_name = None
                 tag_size = None
                 tag = ""
-                for idx, b in enumerate(raw):
+                for b in raw:
                     if tag_name is None:
                         tag_name = b
                     elif tag_name is not None and tag_size is None:
                         tag_size = b
                         tag = ""
-                    else:
-                        if tag_size:
-                            tag_size -= 1
-                            tag += chr(b)
-                            if tag_size == 0:
-                                options.update({check_bytes(tag_name): check_bytes(tag)})
-                                tag_name = None
-                                tag_size = None
-                                tag = ""
+                    elif tag_size:
+                        tag_size -= 1
+                        tag += chr(b)
+                        if tag_size == 0:
+                            options.update({check_bytes(tag_name): check_bytes(tag)})
+                            tag_name = None
+                            tag_size = None
+                            tag = ""
                 return options
 
-            def datagramReceived(self, data, addr):
+            def datagramReceived(self, data, addr):  # noqa: N802
                 try:
-                    mac_address = unpack("!28x6s", data[:34])[0].hex(":")
-                except StructError:
+                    mac_address = struct.unpack("!28x6s", data[:34])[0].hex(":")
+                except struct.error:
                     mac_address = "None"
                 data = self.parse_options(data[240:])
                 data.update({"mac_address": mac_address})

honeypots/dns_server.py

@@ -36,17 +36,17 @@ def server_main(self):
         _q_s = self
 
         class CustomClientResolver(client.Resolver):
-            def queryUDP(self, queries, timeout=2):
+            def queryUDP(self, queries, timeout=2):  # noqa: N802
                 res = client.Resolver.queryUDP(self, queries, timeout)
 
-                def queryFailed(reason):
+                def queryFailed(reason):  # noqa: N802,ARG001
                     return defer.fail(error.DomainError())
 
                 res.addErrback(queryFailed)
                 return res
 
         class CustomDNSServerFactory(DNSServerFactory):
-            def gotResolverResponse(self, response, protocol, message, address):
+            def gotResolverResponse(self, response, protocol, message, address):  # noqa: N802
                 if address is None:
                     src_ip, src_port = "None", "None"
                 else:
@@ -64,7 +64,7 @@ def gotResolverResponse(self, response, protocol, message, address):
                 return super().gotResolverResponse(response, protocol, message, address)
 
         class CustomDnsUdpProtocol(dns.DNSDatagramProtocol):
-            def datagramReceived(self, data: bytes, addr: tuple[str, int]):
+            def datagramReceived(self, data: bytes, addr: tuple[str, int]):  # noqa: N802
                 _q_s.log(
                     {
                         "action": "connection",
@@ -82,15 +82,15 @@ def datagramReceived(self, data: bytes, addr: tuple[str, int]):
         reactor.listenTCP(self.port, self.factory, interface=self.ip)
         reactor.run()
 
-    def test_server(self, ip=None, port=None, domain=None):
+    def test_server(self, *_, domain=None, **__):
         with suppress(Exception):
             from dns.resolver import Resolver
 
             res = Resolver(configure=False)
             res.nameservers = [self.ip]
             res.port = self.port
             temp_domain = domain or "example.org"
-            r = res.query(temp_domain, "a")
+            res.resolve(temp_domain, "a")
 
 
 if __name__ == "__main__":