anti/DECNET
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(tests): add live subprocess integration test suite for services

Browse Source 
Spins up each service's server.py in a real subprocess via a free ephemeral
port (PORT env var), connects with real protocol clients, and asserts both
correct protocol behavior and RFC 5424 log output.

- 44 live tests across 10 services: http, ftp, smtp, redis, mqtt,
  mysql, postgres, mongodb, pop3, imap
- Shared conftest.py: _ServiceProcess (bg reader thread + queue),
  free_port, live_service fixture, assert_rfc5424 helper
- PORT env var added to all 10 targeted server.py templates
- New pytest marker `live`; excluded from default addopts run
- requirements-live-tests.txt: flask, twisted + protocol clients
This commit is contained in:
anti
2026-04-12 01:34:16 -04:00
parent d63e396410
commit 662a5e43e8
24 changed files with 774 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
tests/live/__init__.py Normal file
View File

160
tests/live/conftest.py Normal file
View File

@@ -0,0 +1,160 @@
"""
Shared fixtures for live subprocess service tests.
Each fixture starts the real server.py in a subprocess, captures its stdout
(RFC 5424 syslog lines) via a background reader thread, polls the port for
readiness, yields (port, log_drain_fn), then tears down.
"""
import os
import queue
import re
import socket
import subprocess
import sys
import threading
import time
from collections.abc import Generator
from pathlib import Path
import pytest
_REPO_ROOT = Path(__file__).parent.parent.parent
_TEMPLATES = _REPO_ROOT / "templates"
# Prefer the project venv's Python (has Flask, Twisted, etc.) over system Python
_VENV_PYTHON = _REPO_ROOT / ".venv" / "bin" / "python"
_PYTHON = str(_VENV_PYTHON) if _VENV_PYTHON.exists() else sys.executable
# RFC 5424: <PRI>1 TIMESTAMP HOSTNAME APP-NAME - MSGID [SD] MSG?
# Use search (not match) so lines prefixed by Twisted timestamps are handled.
_RFC5424_RE = re.compile(r"<\d+>1 \S+ \S+ \S+ - \S+ ")
def _free_port() -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
def _wait_for_port(port: int, timeout: float = 8.0) -> bool:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
try:
socket.create_connection(("127.0.0.1", port), timeout=0.1).close()
return True
except OSError:
time.sleep(0.05)
return False
def _drain(q: queue.Queue, timeout: float = 2.0) -> list[str]:
"""Drain all lines from the log queue within *timeout* seconds."""
lines: list[str] = []
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
try:
lines.append(q.get(timeout=max(0.01, deadline - time.monotonic())))
except queue.Empty:
break
return lines
def assert_rfc5424(
lines: list[str],
*,
service: str | None = None,
event_type: str | None = None,
**fields: str,
) -> str:
"""
Assert that at least one line in *lines* is a valid RFC 5424 log entry
matching the given criteria. Returns the first matching line.
"""
for line in lines:
if not _RFC5424_RE.search(line):
continue
if service and f" {service} " not in line:
continue
if event_type and event_type not in line:
continue
if all(f'{k}="{v}"' in line or f"{k}={v}" in line for k, v in fields.items()):
return line
criteria = {"service": service, "event_type": event_type, **fields}
raise AssertionError(
f"No RFC 5424 line matching {criteria!r} found among {len(lines)} lines:\n"
+ "\n".join(f" {l!r}" for l in lines[:20])
)
class _ServiceProcess:
"""Manages a live service subprocess and its stdout log queue."""
def __init__(self, service: str, port: int):
template_dir = _TEMPLATES / service
env = {
**os.environ,
"NODE_NAME": "test-node",
"PORT": str(port),
"PYTHONPATH": str(template_dir),
"LOG_TARGET": "",
}
self._proc = subprocess.Popen(
[_PYTHON, str(template_dir / "server.py")],
cwd=str(template_dir),
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
env=env,
text=True,
)
self._q: queue.Queue = queue.Queue()
self._reader = threading.Thread(target=self._read_loop, daemon=True)
self._reader.start()
def _read_loop(self) -> None:
assert self._proc.stdout is not None
for line in self._proc.stdout:
self._q.put(line.rstrip("\n"))
def drain(self, timeout: float = 2.0) -> list[str]:
return _drain(self._q, timeout)
def stop(self) -> None:
self._proc.terminate()
try:
self._proc.wait(timeout=3)
except subprocess.TimeoutExpired:
self._proc.kill()
self._proc.wait()
@pytest.fixture
def live_service() -> Generator:
"""
Factory fixture: call live_service(service_name) to start a server.
Usage::
def test_foo(live_service):
port, drain = live_service("redis")
# connect to 127.0.0.1:port ...
lines = drain()
assert_rfc5424(lines, service="redis", event_type="auth")
"""
started: list[_ServiceProcess] = []
def _start(service: str) -> tuple[int, callable]:
port = _free_port()
svc = _ServiceProcess(service, port)
started.append(svc)
if not _wait_for_port(port):
svc.stop()
pytest.fail(f"Service '{service}' did not bind to port {port} within 8s")
# Flush startup noise before the test begins
svc.drain(timeout=0.3)
return port, svc.drain
yield _start
for svc in started:
svc.stop()

39
tests/live/test_ftp_live.py Normal file
View File

@@ -0,0 +1,39 @@
import ftplib
import pytest
from tests.live.conftest import assert_rfc5424
@pytest.mark.live
class TestFTPLive:
def test_banner_received(self, live_service):
port, drain = live_service("ftp")
ftp = ftplib.FTP()
ftp.connect("127.0.0.1", port, timeout=5)
welcome = ftp.getwelcome()
ftp.close()
assert "220" in welcome or "vsFTPd" in welcome or len(welcome) > 0
def test_login_logged(self, live_service):
port, drain = live_service("ftp")
ftp = ftplib.FTP()
ftp.connect("127.0.0.1", port, timeout=5)
try:
ftp.login("admin", "hunter2")
except ftplib.all_errors:
pass
finally:
ftp.close()
lines = drain()
assert_rfc5424(lines, service="ftp")
def test_connect_logged(self, live_service):
port, drain = live_service("ftp")
ftp = ftplib.FTP()
ftp.connect("127.0.0.1", port, timeout=5)
ftp.close()
lines = drain()
# At least one RFC 5424 line from the ftp service
rfc_lines = [l for l in lines if "<" in l and ">1 " in l and "ftp" in l]
assert rfc_lines, f"No ftp RFC 5424 lines found. stdout:\n" + "\n".join(lines[:15])

41
tests/live/test_http_live.py Normal file
View File

@@ -0,0 +1,41 @@
import pytest
import requests
from tests.live.conftest import assert_rfc5424
@pytest.mark.live
class TestHTTPLive:
def test_get_request_logged(self, live_service):
port, drain = live_service("http")
resp = requests.get(f"http://127.0.0.1:{port}/admin", timeout=5)
assert resp.status_code == 403
lines = drain()
assert_rfc5424(lines, service="http", event_type="request")
def test_server_header_set(self, live_service):
port, drain = live_service("http")
resp = requests.get(f"http://127.0.0.1:{port}/", timeout=5)
assert "Server" in resp.headers
assert resp.headers["Server"] != ""
def test_post_body_logged(self, live_service):
port, drain = live_service("http")
requests.post(
f"http://127.0.0.1:{port}/login",
data={"username": "admin", "password": "secret"},
timeout=5,
)
lines = drain()
# body field present in log line
assert any("body=" in l for l in lines if "request" in l), (
f"Expected 'body=' in request log line. Got:\n" + "\n".join(lines[:10])
)
def test_method_and_path_in_log(self, live_service):
port, drain = live_service("http")
requests.get(f"http://127.0.0.1:{port}/secret/file.txt", timeout=5)
lines = drain()
matched = assert_rfc5424(lines, service="http", event_type="request")
assert "GET" in matched or 'method="GET"' in matched
assert "/secret/file.txt" in matched or 'path="/secret/file.txt"' in matched

80
tests/live/test_imap_live.py Normal file
View File

@@ -0,0 +1,80 @@
import imaplib
import pytest
from tests.live.conftest import assert_rfc5424
@pytest.mark.live
class TestIMAPLive:
def test_banner_received(self, live_service):
port, drain = live_service("imap")
imap = imaplib.IMAP4("127.0.0.1", port)
welcome = imap.welcome.decode()
imap.logout()
assert "OK" in welcome
def test_connect_logged(self, live_service):
port, drain = live_service("imap")
imap = imaplib.IMAP4("127.0.0.1", port)
imap.logout()
lines = drain()
assert_rfc5424(lines, service="imap", event_type="connect")
def test_login_logged(self, live_service):
port, drain = live_service("imap")
imap = imaplib.IMAP4("127.0.0.1", port)
try:
imap.login("admin", "wrongpass")
except imaplib.IMAP4.error:
pass
lines = drain()
try:
imap.logout()
except Exception:
pass
lines += drain()
assert_rfc5424(lines, service="imap", event_type="auth")
def test_auth_success_logged(self, live_service):
port, drain = live_service("imap")
imap = imaplib.IMAP4("127.0.0.1", port)
imap.login("admin", "admin123") # valid cred from IMAP_USERS default
lines = drain()
imap.logout()
lines += drain()
matched = assert_rfc5424(lines, service="imap", event_type="auth")
assert "success" in matched, f"Expected auth success in log. Got:\n{matched!r}"
def test_auth_fail_logged(self, live_service):
port, drain = live_service("imap")
imap = imaplib.IMAP4("127.0.0.1", port)
try:
imap.login("hacker", "crackedpassword")
except imaplib.IMAP4.error:
pass # expected
lines = drain()
try:
imap.logout()
except Exception:
pass
lines += drain()
matched = assert_rfc5424(lines, service="imap", event_type="auth")
assert "failed" in matched, f"Expected auth failure in log. Got:\n{matched!r}"
def test_select_inbox_after_login(self, live_service):
port, drain = live_service("imap")
imap = imaplib.IMAP4("127.0.0.1", port)
imap.login("admin", "admin123")
status, data = imap.select("INBOX")
imap.logout()
assert status == "OK", f"SELECT INBOX failed: {data}"
def test_capability_command(self, live_service):
port, drain = live_service("imap")
imap = imaplib.IMAP4("127.0.0.1", port)
status, caps = imap.capability()
imap.logout()
assert status == "OK"
cap_str = b" ".join(caps).decode()
assert "IMAP4rev1" in cap_str

70
tests/live/test_mongodb_live.py Normal file
View File

@@ -0,0 +1,70 @@
import pytest
import pymongo
from tests.live.conftest import assert_rfc5424
@pytest.mark.live
class TestMongoDBLive:
def test_connect_succeeds(self, live_service):
port, drain = live_service("mongodb")
client = pymongo.MongoClient(
f"mongodb://127.0.0.1:{port}/",
serverSelectionTimeoutMS=5000,
connectTimeoutMS=5000,
)
# ismaster is handled — should not raise
client.admin.command("ismaster")
client.close()
def test_connect_logged(self, live_service):
port, drain = live_service("mongodb")
client = pymongo.MongoClient(
f"mongodb://127.0.0.1:{port}/",
serverSelectionTimeoutMS=5000,
connectTimeoutMS=5000,
)
try:
client.admin.command("ismaster")
except Exception:
pass
finally:
client.close()
lines = drain()
assert_rfc5424(lines, service="mongodb", event_type="connect")
def test_message_logged(self, live_service):
port, drain = live_service("mongodb")
client = pymongo.MongoClient(
f"mongodb://127.0.0.1:{port}/",
serverSelectionTimeoutMS=5000,
connectTimeoutMS=5000,
)
try:
client.admin.command("ismaster")
except Exception:
pass
finally:
client.close()
lines = drain()
assert_rfc5424(lines, service="mongodb", event_type="message")
def test_list_databases(self, live_service):
port, drain = live_service("mongodb")
client = pymongo.MongoClient(
f"mongodb://127.0.0.1:{port}/",
serverSelectionTimeoutMS=5000,
connectTimeoutMS=5000,
)
try:
# list_database_names triggers OP_MSG
client.list_database_names()
except Exception:
pass
finally:
client.close()
lines = drain()
# At least one message was exchanged
assert any("mongodb" in l for l in lines), (
"Expected at least one mongodb log line"
)

63
tests/live/test_mqtt_live.py Normal file
View File

@@ -0,0 +1,63 @@
import time
import pytest
import paho.mqtt.client as mqtt
from tests.live.conftest import assert_rfc5424
@pytest.mark.live
class TestMQTTLive:
def test_connect_accepted(self, live_service):
port, drain = live_service("mqtt")
connected = []
client = mqtt.Client(client_id="test-scanner")
client.on_connect = lambda c, u, f, rc: connected.append(rc)
client.connect("127.0.0.1", port, keepalive=5)
client.loop_start()
deadline = time.monotonic() + 5
while not connected and time.monotonic() < deadline:
time.sleep(0.05)
client.loop_stop()
client.disconnect()
assert connected and connected[0] == 0, f"Expected CONNACK rc=0, got {connected}"
def test_connect_logged(self, live_service):
port, drain = live_service("mqtt")
client = mqtt.Client(client_id="hax0r")
client.connect("127.0.0.1", port, keepalive=5)
client.loop_start()
time.sleep(0.3)
client.loop_stop()
client.disconnect()
lines = drain()
assert_rfc5424(lines, service="mqtt", event_type="auth")
def test_client_id_in_log(self, live_service):
port, drain = live_service("mqtt")
client = mqtt.Client(client_id="evil-scanner-9000")
client.connect("127.0.0.1", port, keepalive=5)
client.loop_start()
time.sleep(0.3)
client.loop_stop()
client.disconnect()
lines = drain()
matched = assert_rfc5424(lines, service="mqtt", event_type="auth")
assert "evil-scanner-9000" in matched, (
f"Expected client_id in log line. Got:\n{matched!r}"
)
def test_subscribe_logged(self, live_service):
port, drain = live_service("mqtt")
subscribed = []
client = mqtt.Client(client_id="sub-test")
client.on_subscribe = lambda c, u, mid, qos: subscribed.append(mid)
client.connect("127.0.0.1", port, keepalive=5)
client.loop_start()
time.sleep(0.2)
client.subscribe("plant/#")
time.sleep(0.3)
client.loop_stop()
client.disconnect()
lines = drain()
assert_rfc5424(lines, service="mqtt", event_type="subscribe")

65
tests/live/test_mysql_live.py Normal file
View File

@@ -0,0 +1,65 @@
import pytest
import pymysql
from tests.live.conftest import assert_rfc5424
@pytest.mark.live
class TestMySQLLive:
def test_handshake_received(self, live_service):
port, drain = live_service("mysql")
# Honeypot sends MySQL greeting then denies auth — OperationalError expected
try:
pymysql.connect(
host="127.0.0.1",
port=port,
user="root",
password="password",
connect_timeout=5,
)
except pymysql.err.OperationalError:
pass # expected: Access denied
def test_auth_logged(self, live_service):
port, drain = live_service("mysql")
try:
pymysql.connect(
host="127.0.0.1",
port=port,
user="admin",
password="hunter2",
connect_timeout=5,
)
except pymysql.err.OperationalError:
pass
lines = drain()
assert_rfc5424(lines, service="mysql", event_type="auth")
def test_username_in_log(self, live_service):
port, drain = live_service("mysql")
try:
pymysql.connect(
host="127.0.0.1",
port=port,
user="dbhacker",
password="letmein",
connect_timeout=5,
)
except pymysql.err.OperationalError:
pass
lines = drain()
matched = assert_rfc5424(lines, service="mysql", event_type="auth")
assert "dbhacker" in matched, (
f"Expected username in log line. Got:\n{matched!r}"
)
def test_connect_logged(self, live_service):
port, drain = live_service("mysql")
try:
pymysql.connect(
host="127.0.0.1", port=port, user="x", password="y", connect_timeout=5
)
except pymysql.err.OperationalError:
pass
lines = drain()
assert_rfc5424(lines, service="mysql", event_type="connect")

58
tests/live/test_pop3_live.py Normal file
View File

@@ -0,0 +1,58 @@
import poplib
import pytest
from tests.live.conftest import assert_rfc5424
@pytest.mark.live
class TestPOP3Live:
def test_banner_received(self, live_service):
port, drain = live_service("pop3")
pop = poplib.POP3("127.0.0.1", port)
welcome = pop.getwelcome().decode()
pop.quit()
assert "+OK" in welcome
def test_connect_logged(self, live_service):
port, drain = live_service("pop3")
pop = poplib.POP3("127.0.0.1", port)
pop.quit()
lines = drain()
assert_rfc5424(lines, service="pop3", event_type="connect")
def test_user_command_logged(self, live_service):
port, drain = live_service("pop3")
pop = poplib.POP3("127.0.0.1", port)
pop.user("admin")
pop.quit()
lines = drain()
assert_rfc5424(lines, service="pop3", event_type="command")
def test_auth_success_logged(self, live_service):
port, drain = live_service("pop3")
pop = poplib.POP3("127.0.0.1", port)
pop.user("admin")
pop.pass_("admin123") # valid cred from IMAP_USERS default
lines = drain()
pop.quit()
lines += drain()
matched = assert_rfc5424(lines, service="pop3", event_type="auth")
assert "success" in matched, f"Expected auth success in log. Got:\n{matched!r}"
def test_auth_fail_logged(self, live_service):
port, drain = live_service("pop3")
pop = poplib.POP3("127.0.0.1", port)
pop.user("root")
try:
pop.pass_("wrongpassword")
except poplib.error_proto:
pass # expected: -ERR Authentication failed
lines = drain()
try:
pop.quit()
except Exception:
pass
lines += drain()
matched = assert_rfc5424(lines, service="pop3", event_type="auth")
assert "failed" in matched, f"Expected auth failure in log. Got:\n{matched!r}"

75
tests/live/test_postgres_live.py Normal file
View File

@@ -0,0 +1,75 @@
import pytest
from tests.live.conftest import assert_rfc5424
@pytest.mark.live
class TestPostgresLive:
def test_handshake_received(self, live_service):
port, drain = live_service("postgres")
import psycopg2
try:
psycopg2.connect(
host="127.0.0.1",
port=port,
user="admin",
password="password",
dbname="production",
connect_timeout=5,
)
except psycopg2.OperationalError:
pass # expected: honeypot rejects auth
def test_startup_logged(self, live_service):
port, drain = live_service("postgres")
import psycopg2
try:
psycopg2.connect(
host="127.0.0.1",
port=port,
user="postgres",
password="secret",
dbname="postgres",
connect_timeout=5,
)
except psycopg2.OperationalError:
pass
lines = drain()
assert_rfc5424(lines, service="postgres", event_type="startup")
def test_username_in_log(self, live_service):
port, drain = live_service("postgres")
import psycopg2
try:
psycopg2.connect(
host="127.0.0.1",
port=port,
user="dbattacker",
password="cracked",
dbname="secrets",
connect_timeout=5,
)
except psycopg2.OperationalError:
pass
lines = drain()
matched = assert_rfc5424(lines, service="postgres", event_type="startup")
assert "dbattacker" in matched, (
f"Expected username in log line. Got:\n{matched!r}"
)
def test_auth_hash_logged(self, live_service):
port, drain = live_service("postgres")
import psycopg2
try:
psycopg2.connect(
host="127.0.0.1",
port=port,
user="root",
password="toor",
dbname="prod",
connect_timeout=5,
)
except psycopg2.OperationalError:
pass
lines = drain()
assert_rfc5424(lines, service="postgres", event_type="auth")

44
tests/live/test_redis_live.py Normal file
View File

@@ -0,0 +1,44 @@
import pytest
import redis
from tests.live.conftest import assert_rfc5424
@pytest.mark.live
class TestRedisLive:
def test_ping_responds(self, live_service):
port, drain = live_service("redis")
r = redis.Redis(host="127.0.0.1", port=port, socket_timeout=5)
assert r.ping() is True
def test_connect_logged(self, live_service):
port, drain = live_service("redis")
r = redis.Redis(host="127.0.0.1", port=port, socket_timeout=5)
r.ping()
lines = drain()
assert_rfc5424(lines, service="redis", event_type="connect")
def test_auth_logged(self, live_service):
port, drain = live_service("redis")
r = redis.Redis(
host="127.0.0.1", port=port, password="wrongpassword", socket_timeout=5
)
try:
r.ping()
except Exception:
pass
lines = drain()
assert_rfc5424(lines, service="redis", event_type="auth")
def test_command_logged(self, live_service):
port, drain = live_service("redis")
r = redis.Redis(host="127.0.0.1", port=port, socket_timeout=5)
r.execute_command("KEYS", "*")
lines = drain()
assert_rfc5424(lines, service="redis", event_type="command")
def test_keys_returns_bait_data(self, live_service):
port, drain = live_service("redis")
r = redis.Redis(host="127.0.0.1", port=port, socket_timeout=5)
keys = r.keys("*")
assert len(keys) > 0, "Expected bait keys in fake store"

39
tests/live/test_smtp_live.py Normal file
View File

@@ -0,0 +1,39 @@
import smtplib
import pytest
from tests.live.conftest import assert_rfc5424
@pytest.mark.live
class TestSMTPLive:
def test_banner_received(self, live_service):
port, drain = live_service("smtp")
with smtplib.SMTP("127.0.0.1", port, timeout=5) as s:
code, msg = s.ehlo("test.example.com")
assert code == 250
def test_ehlo_logged(self, live_service):
port, drain = live_service("smtp")
with smtplib.SMTP("127.0.0.1", port, timeout=5) as s:
s.ehlo("attacker.example.com")
lines = drain()
assert_rfc5424(lines, service="smtp", event_type="ehlo")
def test_auth_attempt_logged(self, live_service):
port, drain = live_service("smtp")
with smtplib.SMTP("127.0.0.1", port, timeout=5) as s:
s.ehlo("attacker.example.com")
try:
s.login("admin", "password123")
except smtplib.SMTPAuthenticationError:
pass # expected — honeypot rejects auth
lines = drain()
assert_rfc5424(lines, service="smtp", event_type="auth_attempt")
def test_connect_disconnect_logged(self, live_service):
port, drain = live_service("smtp")
with smtplib.SMTP("127.0.0.1", port, timeout=5) as s:
s.ehlo("scanner.example.com")
lines = drain()
assert_rfc5424(lines, service="smtp", event_type="connect")