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

feat(mail): operator-tunable IMAP/POP3 email seed (DEBT-026)

Browse Source 
IMAP_EMAIL_SEED / POP3_EMAIL_SEED accept a directory (rglob *.eml +
*.json) or a single .json/.eml. Loaded entries CONCATENATE with the
hardcoded _BAIT_EMAILS — additive to the realism-engine emailgen
output rather than replacing it. JSON dicts require from_addr /
to_addr / subject / body; bare bodies are wrapped into RFC 5322 on
load. compose_fragment reads service_cfg["email_seed"] and bind-mounts
the host path read-only at /var/spool/decnet-emails/seed.
This commit is contained in:
anti
2026-05-03 02:47:06 -04:00
parent e0b07651fd
commit b88d67794d
8 changed files with 444 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
tests/service_testing/test_imap_spool.py
View File

@@ -1,12 +1,15 @@
"""Spool-backed email loading for the IMAP template.
"""Seed-backed email loading for the IMAP template.
Verifies that when ``IMAP_EMAIL_SEED`` points at a directory of .eml
files, the IMAP server serves those (replacing the hardcoded
``_BAIT_EMAILS`` fallback). Empty / missing dir falls back gracefully.
Verifies that when ``IMAP_EMAIL_SEED`` points at a directory of .eml /
.json (or a single .json / .eml), the IMAP server CONCATENATES those
entries onto the hardcoded ``_BAIT_EMAILS`` baseline. Empty / missing
input falls back to the baseline alone — the realism-engine output and
operator-curated seeds are additive, never replacing.
"""
from __future__ import annotations
import importlib.util
import json
import sys
from pathlib import Path
from types import ModuleType
@@ -14,6 +17,8 @@ from unittest.mock import MagicMock, patch
import pytest
_HARDCODED = 10 # length of templates/imap/server.py::_BAIT_EMAILS
_EML_TEMPLATE = (
"From: {from_name} <{from_addr}>\r\n"
@@ -78,41 +83,47 @@ def _seed(tmp_path: Path, n: int = 3) -> Path:
def test_falls_back_to_hardcoded_when_seed_unset(tmp_path):
mod = _load_imap({})
emails = mod._get_emails()
# The shipped fallback ships exactly 10 entries.
assert len(emails) == 10
# The shipped baseline is exactly 10 entries.
assert len(emails) == _HARDCODED
assert emails[0]["from_addr"] == "devops@company.internal"
def test_falls_back_when_seed_dir_missing(tmp_path):
mod = _load_imap({"IMAP_EMAIL_SEED": str(tmp_path / "does-not-exist")})
emails = mod._get_emails()
assert len(emails) == 10 # fallback
assert len(emails) == _HARDCODED # baseline only
def test_falls_back_when_seed_dir_empty(tmp_path):
(tmp_path / "spool").mkdir()
mod = _load_imap({"IMAP_EMAIL_SEED": str(tmp_path / "spool")})
assert len(mod._get_emails()) == 10 # fallback (no .eml files)
assert len(mod._get_emails()) == _HARDCODED # baseline only
def test_loads_eml_files_from_spool(tmp_path):
def test_seed_concatenates_with_hardcoded(tmp_path):
spool = _seed(tmp_path, n=3)
mod = _load_imap({"IMAP_EMAIL_SEED": str(spool)})
emails = mod._get_emails()
assert len(emails) == 3
senders = {e["from_addr"] for e in emails}
# Hardcoded 10 + 3 spooled = 13.
assert len(emails) == _HARDCODED + 3
# Hardcoded baseline keeps original UIDs 1..10.
assert emails[0]["uid"] == 1
assert emails[0]["from_addr"] == "devops@company.internal"
assert emails[9]["uid"] == 10
# Seeded entries pick up at UID 11.
assert {e["uid"] for e in emails[10:]} == {11, 12, 13}
senders = {e["from_addr"] for e in emails[10:]}
assert senders == {"sender0@corp.com", "sender1@corp.com", "sender2@corp.com"}
# UIDs are 1-based and unique.
assert {e["uid"] for e in emails} == {1, 2, 3}
def test_loaded_eml_carries_full_rfc822_body(tmp_path):
spool = _seed(tmp_path, n=1)
mod = _load_imap({"IMAP_EMAIL_SEED": str(spool)})
emails = mod._get_emails()
assert "From:" in emails[0]["body"]
assert "Subject: Topic 0" in emails[0]["body"]
assert "Body of message 0." in emails[0]["body"]
seeded = emails[_HARDCODED]
assert "From:" in seeded["body"]
assert "Subject: Topic 0" in seeded["body"]
assert "Body of message 0." in seeded["body"]
def test_corrupt_eml_skipped_not_fatal(tmp_path):
@@ -127,11 +138,56 @@ def test_corrupt_eml_skipped_not_fatal(tmp_path):
(spool / "broken.eml").mkdir()
mod = _load_imap({"IMAP_EMAIL_SEED": str(spool)})
emails = mod._get_emails()
assert len(emails) == 1
assert emails[0]["from_addr"] == "good@corp.com"
assert len(emails) == _HARDCODED + 1
assert emails[-1]["from_addr"] == "good@corp.com"
def test_select_inbox_reflects_spool_count(tmp_path):
def test_json_seed_file_loaded(tmp_path):
seed = tmp_path / "seed.json"
seed.write_text(json.dumps([
{
"from_addr": "ceo@corp.com",
"from_name": "CEO",
"to_addr": "admin@corp.com",
"subject": "Q4 numbers",
"date": "Mon, 27 Apr 2026 09:00:00 +0000",
"body": "Please review attached.",
},
{
# Missing 'subject' — must be skipped, not crash.
"from_addr": "ghost@corp.com",
"to_addr": "admin@corp.com",
"body": "no subject",
},
]))
mod = _load_imap({"IMAP_EMAIL_SEED": str(seed)})
emails = mod._get_emails()
assert len(emails) == _HARDCODED + 1 # one valid, one dropped
seeded = emails[-1]
assert seeded["uid"] == _HARDCODED + 1
assert seeded["from_addr"] == "ceo@corp.com"
# JSON entry without RFC 822 headers gets wrapped into a full message.
assert "From: CEO <ceo@corp.com>" in seeded["body"]
assert "Subject: Q4 numbers" in seeded["body"]
def test_dir_with_eml_and_json_concatenated(tmp_path):
spool = _seed(tmp_path, n=2)
(spool / "extra.json").write_text(json.dumps([
{
"from_addr": "ops@corp.com",
"to_addr": "admin@corp.com",
"subject": "extra",
"body": "hi",
},
]))
mod = _load_imap({"IMAP_EMAIL_SEED": str(spool)})
emails = mod._get_emails()
# Hardcoded + 2 .eml + 1 .json
assert len(emails) == _HARDCODED + 3
def test_select_inbox_reflects_concatenated_count(tmp_path):
spool = _seed(tmp_path, n=4)
mod = _load_imap({"IMAP_EMAIL_SEED": str(spool)})
proto = mod.IMAPProtocol()
@@ -144,5 +200,6 @@ def test_select_inbox_reflects_spool_count(tmp_path):
written.clear()
proto.data_received(b"B0 SELECT INBOX\r\n")
out = b"".join(written)
assert b"* 4 EXISTS" in out
assert b"[UIDNEXT 5]" in out
expected_total = _HARDCODED + 4
assert f"* {expected_total} EXISTS".encode() in out
assert f"[UIDNEXT {expected_total + 1}]".encode() in out

57
tests/service_testing/test_pop3_spool.py
View File

@@ -1,12 +1,19 @@
"""Spool-backed email loading for the POP3 template."""
"""Seed-backed email loading for the POP3 template.
Concat semantics: hardcoded ``_BAIT_EMAILS`` + .eml/.json from the seed
path. Mirrors the IMAP test file.
"""
from __future__ import annotations
import importlib.util
import json
import sys
from pathlib import Path
from types import ModuleType
from unittest.mock import MagicMock, patch
_HARDCODED = 10 # length of templates/pop3/server.py::_BAIT_EMAILS
_EML_TEMPLATE = (
"From: Sender <sender@corp.com>\r\n"
@@ -61,25 +68,54 @@ def _seed(tmp_path: Path, n: int) -> Path:
def test_falls_back_when_seed_unset(tmp_path):
mod = _load_pop3({})
assert len(mod._get_emails()) == 10 # hardcoded fallback
assert len(mod._get_emails()) == _HARDCODED # baseline only
def test_falls_back_when_seed_dir_missing(tmp_path):
mod = _load_pop3({"POP3_EMAIL_SEED": str(tmp_path / "nope")})
assert len(mod._get_emails()) == 10
assert len(mod._get_emails()) == _HARDCODED
def test_loads_emls_from_spool(tmp_path):
def test_seed_concatenates_with_hardcoded(tmp_path):
spool = _seed(tmp_path, n=3)
mod = _load_pop3({"POP3_EMAIL_SEED": str(spool)})
emails = mod._get_emails()
assert len(emails) == 3
# POP3 stores raw RFC 822 strings; verify content round-trips.
assert any("Topic 0" in e for e in emails)
assert all(e.startswith("From:") for e in emails)
# Hardcoded baseline + 3 spooled .eml.
assert len(emails) == _HARDCODED + 3
# Hardcoded entries unchanged at the head.
assert "AWS credentials rotation" in emails[0]
# Seeded entries at the tail.
assert any("Topic 0" in e for e in emails[_HARDCODED:])
assert all(e.startswith("From:") for e in emails[_HARDCODED:])
def test_stat_reflects_spool_size(tmp_path):
def test_json_seed_file_loaded(tmp_path):
seed = tmp_path / "seed.json"
seed.write_text(json.dumps([
{
"from_addr": "ceo@corp.com",
"from_name": "CEO",
"to_addr": "admin@corp.com",
"subject": "Q4 numbers",
"date": "Mon, 27 Apr 2026 09:00:00 +0000",
"body": "Please review attached.",
},
{
# Missing 'subject' — skipped, not fatal.
"from_addr": "ghost@corp.com",
"to_addr": "admin@corp.com",
"body": "no subject",
},
]))
mod = _load_pop3({"POP3_EMAIL_SEED": str(seed)})
emails = mod._get_emails()
assert len(emails) == _HARDCODED + 1
seeded = emails[-1]
assert "Subject: Q4 numbers" in seeded
assert "From: CEO <ceo@corp.com>" in seeded
def test_stat_reflects_concatenated_count(tmp_path):
spool = _seed(tmp_path, n=2)
mod = _load_pop3({"POP3_EMAIL_SEED": str(spool)})
proto = mod.POP3Protocol()
@@ -93,4 +129,5 @@ def test_stat_reflects_spool_size(tmp_path):
written.clear()
proto.data_received(b"STAT\r\n")
out = b"".join(written)
assert out.startswith(b"+OK 2 ")
expected = _HARDCODED + 2
assert out.startswith(f"+OK {expected} ".encode())

36
tests/services/test_services.py
View File

@@ -361,3 +361,39 @@ def test_telnet_no_cowrie_env_vars():
"""Ensure no Cowrie env vars bleed into the real telnet service."""
env = _fragment("telnet").get("environment", {})
assert not any(k.startswith("COWRIE_") for k in env)
# IMAP / POP3 email_seed -----------------------------------------------------
def test_imap_no_email_seed_by_default():
fragment = _fragment("imap")
assert "IMAP_EMAIL_SEED" not in fragment.get("environment", {})
assert "volumes" not in fragment
def test_imap_email_seed_wires_env_and_volume(tmp_path):
seed_dir = tmp_path / "seed"
seed_dir.mkdir()
fragment = _fragment("imap", service_cfg={"email_seed": str(seed_dir)})
assert fragment["environment"]["IMAP_EMAIL_SEED"] == "/var/spool/decnet-emails/seed"
volumes = fragment.get("volumes") or []
assert len(volumes) == 1
assert volumes[0].endswith(":/var/spool/decnet-emails/seed:ro")
assert volumes[0].startswith(str(seed_dir))
def test_pop3_no_email_seed_by_default():
fragment = _fragment("pop3")
assert "POP3_EMAIL_SEED" not in fragment.get("environment", {})
assert "volumes" not in fragment
def test_pop3_email_seed_wires_env_and_volume(tmp_path):
seed_file = tmp_path / "seed.json"
seed_file.write_text("[]")
fragment = _fragment("pop3", service_cfg={"email_seed": str(seed_file)})
assert fragment["environment"]["POP3_EMAIL_SEED"] == "/var/spool/decnet-emails/seed"
volumes = fragment.get("volumes") or []
assert len(volumes) == 1
assert volumes[0].endswith(":/var/spool/decnet-emails/seed:ro")
assert volumes[0].startswith(str(seed_file))