unit tests: smtp and pop3: test for capture_commands

tests/test_pop3_server.py

@@ -17,11 +17,18 @@
 )
 
 PORT = "50110"
+SERVER_CONFIG = {
+    "honeypots": {
+        "pop3": {
+            "options": ["capture_commands"],
+        },
+    }
+}
 
 
 @pytest.mark.parametrize(
     "server_logs",
-    [{"server": QPOP3Server, "port": PORT}],
+    [{"server": QPOP3Server, "port": PORT, "custom_config": SERVER_CONFIG}],
     indirect=True,
 )
 def test_pop3_server(server_logs):
@@ -36,7 +43,12 @@ def test_pop3_server(server_logs):
 
     logs = load_logs_from_file(server_logs)
 
-    assert len(logs) == 2
-    connect, login = logs
+    assert len(logs) == 4
+    connect, cmd1, cmd2, login = logs
     assert_connect_is_logged(connect, PORT)
     assert_login_is_logged(login)
+
+    assert cmd1["action"] == "command"
+    assert cmd1["data"] == {"args": "testing", "cmd": "USER"}
+    assert cmd2["action"] == "command"
+    assert cmd2["data"] == {"args": "testing", "cmd": "PASS"}

tests/test_smtp_server.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from base64 import b64decode
 from smtplib import SMTP
 from time import sleep
 
@@ -16,11 +17,25 @@
 )
 
 PORT = "50025"
+SERVER_CONFIG = {
+    "honeypots": {
+        "smtp": {
+            "options": ["capture_commands"],
+        },
+    }
+}
+EXPECTED_DATA = [
+    {"arg": "FROM:<fromtest>", "command": "MAIL", "data": "None"},
+    {"arg": "TO:<totest>", "command": "RCPT", "data": "None"},
+    {"arg": "None", "command": "DATA", "data": "None"},
+    {"arg": "None", "command": "NOTHING", "data": "None"},
+    {"arg": "None", "command": "QUIT", "data": "None"},
+]
 
 
 @pytest.mark.parametrize(
     "server_logs",
-    [{"server": QSMTPServer, "port": PORT}],
+    [{"server": QSMTPServer, "port": PORT, "custom_config": SERVER_CONFIG}],
     indirect=True,
 )
 def test_smtp_server(server_logs):
@@ -36,7 +51,13 @@ def test_smtp_server(server_logs):
 
     logs = load_logs_from_file(server_logs)
 
-    assert len(logs) == 2
-    connect, login = logs
+    assert len(logs) == 8
+    connect, auth, login, *additional = logs
     assert_connect_is_logged(connect, PORT)
     assert_login_is_logged(login)
+
+    assert auth["data"]["command"] == "AUTH"
+    assert b64decode(auth["data"]["data"]).decode() == f"\x00{USERNAME}\x00{PASSWORD}"
+
+    for entry, expected in zip(additional, EXPECTED_DATA):
+        assert entry

tests/test_ssh_server.py

@@ -12,14 +12,7 @@
 SERVER_CONFIG = {
     "honeypots": {
         "ssh": {
-            "backup_count": 10,
-            "ip": IP,
-            "log_file_name": "ssh.jsonl",
-            "max_bytes": 10000,
             "options": ["capture_commands"],
-            "password": PASSWORD,
-            "port": str(PORT),
-            "username": USERNAME,
         },
     }
 }