HL7 server

README.md

@@ -5,7 +5,7 @@
 ## Why honeypots package is very powerful?
 The honeypots respond back, non-blocking, can be used as objects, or called directly with the in-built auto-configure scripts! Also, they are easy to set up and customize; it takes 1-2 seconds to spin a honeypot up. You can spin up multiple instances with the same type. For easy integration, the output can be logged to a Postgres database, file[s], terminal, or Syslog.
 
-This honeypots package is the only package that contains all the following: 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.
+This honeypots package is the only package that contains all the following: dhcp, dns, elastic, ftp, hl7, 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.
 
 Honeypots is in the awesome [telekom security T-Pot project!](https://github.com/telekom-security/tpotce)
 
@@ -254,6 +254,11 @@ qsshserver.kill_server()
     - Lib: Twisted.ftp
     - Logs: ip, port, username and password (default)
     - Options: Capture all threat actor commands and data (available)
+- HL7Server
+    - Server: HL7 
+    - Port: 2575/tcp
+    - Lib: socketserver
+    - Logs: ip, port and data
 - QHTTPProxyServer
     - Server: HTTP Proxy
     - Port: 8080/tcp

honeypots/__init__.py

@@ -9,10 +9,10 @@
     kill_servers,
     PostgresClass,
     server_arguments,
-    set_local_vars,
     setup_logger,
     set_up_error_logging,
 )
+from .hl7_server import HL7Server
 from .http_proxy_server import QHTTPProxyServer
 from .http_server import QHTTPServer
 from .https_server import QHTTPSServer
@@ -41,6 +41,7 @@
 from .vnc_server import QVNCServer
 
 __all__ = [
+    "HL7Server",
     "QSniffer",
     "QDHCPServer",
     "QDNSServer",
@@ -77,7 +78,6 @@
     "kill_servers",
     "PostgresClass",
     "server_arguments",
-    "set_local_vars",
     "setup_logger",
     "set_up_error_logging",
 ]

honeypots/__main__.py

@@ -7,19 +7,18 @@
 from argparse import ArgumentParser, SUPPRESS, Namespace
 from atexit import register
 from functools import wraps
-from json import JSONDecodeError, loads
 from os import geteuid
-from pathlib import Path
 from signal import alarm, SIGALRM, SIGINT, signal, SIGTERM, SIGTSTP
 from subprocess import Popen
 from time import sleep
-from typing import Any
+from typing import Any, Type, TYPE_CHECKING
 from uuid import uuid4
 
 from netifaces import ifaddresses, AF_INET, AF_LINK, interfaces
 from psutil import net_io_counters, Process
 
 from honeypots import (
+    HL7Server,
     QSniffer,
     QDHCPServer,
     QDNSServer,
@@ -50,17 +49,23 @@
     QSSHServer,
     QTelnetServer,
     QVNCServer,
-    is_privileged,
     clean_all,
-    setup_logger,
+    is_privileged,
     set_up_error_logging,
+    setup_logger,
 )
+from honeypots.helper import load_config
+
+if TYPE_CHECKING:
+    from honeypots.base_server import BaseServer
+
 
 all_servers = {
     "dhcp": QDHCPServer,
     "dns": QDNSServer,
     "elastic": QElasticServer,
     "ftp": QFTPServer,
+    "hl7": HL7Server,
     "httpproxy": QHTTPProxyServer,
     "https": QHTTPSServer,
     "http": QHTTPServer,
@@ -141,10 +146,10 @@ def handle_timeout(signum, frame):  # noqa: ARG001
 
 
 @timeout(5)
-def server_timeout(obj, name):
+def server_timeout(server: BaseServer, name: str):
     try:
         logger.info(f"Start testing {name}")
-        obj.test_server()
+        server.test_server()
     except TimeoutError:
         logging.error(f"Timeout during test {name}")
     logger.info(f"Done testing {name}")
@@ -154,7 +159,7 @@ class HoneypotsManager:
     def __init__(self, options: Namespace, server_args: dict[str, str | int]):
         self.options = options
         self.server_args = server_args
-        self.config_data = self._load_config() if self.options.config else {}
+        self.config_data: dict = load_config(self.options.config) if self.options.config else {}
         self.auto = options.auto if geteuid() != 0 else False
         self.honeypots: list[tuple[Any, str, bool]] = []
 
@@ -177,22 +182,6 @@ def main(self):
                 self._set_up_sniffer()
             self._set_up_honeypots()
 
-    def _load_config(self):
-        config_path = Path(self.options.config)
-        if not config_path.is_file():
-            logger.error(f'Config file "{config_path}" not found')
-            sys.exit(1)
-        try:
-            config_data = loads(config_path.read_text())
-            logger.info(f"Successfully loaded config file {config_path}")
-            return config_data
-        except FileNotFoundError:
-            logger.error(f"Unable to load config file: File {config_path} not found")
-            sys.exit(1)
-        except JSONDecodeError as error:
-            logger.error(f"Unable to parse config file as JSON: {error}")
-            sys.exit(1)
-
     def _set_up_honeypots(self):  # noqa: C901
         register(_exit_handler)
         if self.options.termination_strategy == "input":
@@ -257,11 +246,11 @@ def _start_configured_servers(self):
     def _start_server(self, service: str, auto: bool | None = None):
         if auto is None:
             auto = self.auto
-        server_class = all_servers.get(service.lower())
+        server_class: Type[BaseServer] = all_servers.get(service.lower().replace("_", ""))
         if not server_class:
             logger.warning(f"Skipping unknown service {service}")
             return
-        server = server_class(**self.server_args)
+        server = server_class(**self.server_args, config=self.config_data)
         if not self.options.test:
             status = server.run_server(process=True, auto=auto)
         else:
@@ -333,7 +322,7 @@ def _setup_logging(self) -> logging.Logger:
             logger.info(f"[x] Setup Logger {uuid} with a db, drop is {drop}")
         else:
             drop = True
-        return setup_logger("main", uuid, self.options.config, drop)
+        return setup_logger("main", uuid, self.config_data, drop)
 
     def _set_up_sniffer(self):
         sniffer_filter = self.config_data.get("sniffer_filter")
@@ -464,15 +453,15 @@ def _parse_args() -> tuple[Namespace, dict[str, str | int]]:
         "--password", help="Override the password", metavar="", default=""
     )
     arg_parser_optional.add_argument(
+        "--options", type=str, help="Extra options", metavar="", default=""
+    )
+    arg_parser_optional_2 = arg_parser.add_argument_group("General options")
+    arg_parser_optional_2.add_argument(
         "--config",
         help="Use a config file for honeypots settings",
         metavar="",
         default="",
     )
-    arg_parser_optional.add_argument(
-        "--options", type=str, help="Extra options", metavar="", default=""
-    )
-    arg_parser_optional_2 = arg_parser.add_argument_group("General options")
     arg_parser_optional_2.add_argument(
         "--termination-strategy",
         help="Determines the strategy to terminate by",

honeypots/base_server.py

@@ -29,7 +29,7 @@ def __init__(self, **kwargs):
         self.auto_disabled = False
         self.process = None
         self.uuid = f"honeypotslogger_{self.__class__.__name__}_{str(uuid4())[:8]}"
-        self.config = kwargs.get("config", "")
+        self.config: dict = kwargs.get("config", {})
         if self.config:
             self.logs = setup_logger(self.__class__.__name__, self.uuid, self.config)
             set_local_vars(self, self.config)
@@ -88,11 +88,15 @@ def kill_server(self):
             except TimeoutError:
                 self._server_process.kill()
 
+    def test_server(self, **_):
+        self.logger.warning(f"Test method of {self.NAME} is not implemented")
+
     def server_is_alive(self) -> bool:
         return self._server_process and self._server_process.is_alive()
 
     @abstractmethod
     def server_main(self):
+        # main server loop goes here
         pass
 
     def run_server(self, process: bool = False, auto: bool = False) -> bool | None:

honeypots/dhcp_server.py

@@ -17,7 +17,7 @@
 from twisted.internet.protocol import DatagramProtocol
 
 from honeypots.base_server import BaseServer
-from honeypots.helper import check_bytes, server_arguments
+from honeypots.helper import check_bytes, run_single_server
 
 
 class QDHCPServer(BaseServer):
@@ -107,14 +107,6 @@ def datagramReceived(self, data, addr):  # noqa: N802
         )
         reactor.run()
 
-    def test_server(self, ip=None, port=None):
-        pass
-
 
 if __name__ == "__main__":
-    parsed = server_arguments()
-    if parsed.docker or parsed.aws or parsed.custom:
-        qdhcpserver = QDHCPServer(
-            ip=parsed.ip, port=parsed.port, options=parsed.options, config=parsed.config
-        )
-        qdhcpserver.run_server()
+    run_single_server(QDHCPServer)

honeypots/dns_server.py

@@ -19,9 +19,7 @@
 from twisted.names.server import DNSServerFactory
 
 from honeypots.base_server import BaseServer
-from honeypots.helper import (
-    server_arguments,
-)
+from honeypots.helper import run_single_server
 
 
 class QDNSServer(BaseServer):
@@ -94,7 +92,4 @@ def test_server(self, *_, domain=None, **__):
 
 
 if __name__ == "__main__":
-    parsed = server_arguments()
-    if parsed.docker or parsed.aws or parsed.custom:
-        qdnsserver = QDNSServer(ip=parsed.ip, port=parsed.port, config=parsed.config)
-        qdnsserver.run_server()
+    run_single_server(QDNSServer)

honeypots/elastic_server.py

@@ -20,9 +20,9 @@
 
 from honeypots.base_server import BaseServer
 from honeypots.helper import (
-    server_arguments,
     create_certificate,
     check_bytes,
+    run_single_server,
 )
 
 
@@ -349,14 +349,4 @@ def test_server(self, ip=None, port=None, username=None, password=None):
 
 
 if __name__ == "__main__":
-    parsed = server_arguments()
-    if parsed.docker or parsed.aws or parsed.custom:
-        qelasticserver = QElasticServer(
-            ip=parsed.ip,
-            port=parsed.port,
-            username=parsed.username,
-            password=parsed.password,
-            options=parsed.options,
-            config=parsed.config,
-        )
-        qelasticserver.run_server()
+    run_single_server(QElasticServer)

honeypots/ftp_server.py

@@ -33,8 +33,8 @@
 
 from honeypots.base_server import BaseServer
 from honeypots.helper import (
-    server_arguments,
     check_bytes,
+    run_single_server,
 )
 
 
@@ -178,14 +178,4 @@ def test_server(self, ip=None, port=None, username=None, password=None):
 
 
 if __name__ == "__main__":
-    parsed = server_arguments()
-    if parsed.docker or parsed.aws or parsed.custom:
-        ftpserver = QFTPServer(
-            ip=parsed.ip,
-            port=parsed.port,
-            username=parsed.username,
-            password=parsed.password,
-            options=parsed.options,
-            config=parsed.config,
-        )
-        ftpserver.run_server()
+    run_single_server(QFTPServer)