feat(email): add attachment extraction support

Save inbound email attachments to the media directory with configurable MIME type filtering (glob patterns like "image/*"), per-attachment size limits, and max attachment count. Filenames are sanitized to prevent path traversal. Controlled by allowed_attachment_types — empty (default) means disabled, non-empty enables extraction for matching types.
2026-04-06 15:09:44 +08:00
parent 9174a85b4e
commit d0527a8cf4
2 changed files with 303 additions and 0 deletions
@@ -1,5 +1,6 @@
 from email.message import EmailMessage
 from datetime import date
+from pathlib import Path
 import imaplib

 import pytest
@@ -650,3 +651,224 @@ def test_check_authentication_results_method() -> None:
    spf, dkim = EmailChannel._check_authentication_results(parsed)
    assert spf is False
    assert dkim is True
+
+
+# ---------------------------------------------------------------------------
+# Attachment extraction tests
+# ---------------------------------------------------------------------------
+
+
+def _make_raw_email_with_attachment(
+    from_addr: str = "alice@example.com",
+    subject: str = "With attachment",
+    body: str = "See attached.",
+    attachment_name: str = "doc.pdf",
+    attachment_content: bytes = b"%PDF-1.4 fake pdf content",
+    attachment_mime: str = "application/pdf",
+    auth_results: str | None = None,
+) -> bytes:
+    msg = EmailMessage()
+    msg["From"] = from_addr
+    msg["To"] = "bot@example.com"
+    msg["Subject"] = subject
+    msg["Message-ID"] = "<m1@example.com>"
+    if auth_results:
+        msg["Authentication-Results"] = auth_results
+    msg.set_content(body)
+    maintype, subtype = attachment_mime.split("/", 1)
+    msg.add_attachment(
+        attachment_content,
+        maintype=maintype,
+        subtype=subtype,
+        filename=attachment_name,
+    )
+    return msg.as_bytes()
+
+
+def test_extract_attachments_saves_pdf(tmp_path, monkeypatch) -> None:
+    """PDF attachment is saved to media dir and path returned in media list."""
+    monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
+
+    raw = _make_raw_email_with_attachment()
+    fake = _make_fake_imap(raw)
+    monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
+
+    cfg = _make_config(allowed_attachment_types=["application/pdf"], verify_dkim=False, verify_spf=False)
+    channel = EmailChannel(cfg, MessageBus())
+    items = channel._fetch_new_messages()
+
+    assert len(items) == 1
+    assert len(items[0]["media"]) == 1
+    saved_path = Path(items[0]["media"][0])
+    assert saved_path.exists()
+    assert saved_path.read_bytes() == b"%PDF-1.4 fake pdf content"
+    assert "500_doc.pdf" in saved_path.name
+    assert "[attachment:" in items[0]["content"]
+
+
+def test_extract_attachments_disabled_by_default(monkeypatch) -> None:
+    """With no allowed_attachment_types (default), no attachments are extracted."""
+    raw = _make_raw_email_with_attachment()
+    fake = _make_fake_imap(raw)
+    monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
+
+    cfg = _make_config(verify_dkim=False, verify_spf=False)
+    assert cfg.allowed_attachment_types == []
+    channel = EmailChannel(cfg, MessageBus())
+    items = channel._fetch_new_messages()
+
+    assert len(items) == 1
+    assert items[0]["media"] == []
+    assert "[attachment:" not in items[0]["content"]
+
+
+def test_extract_attachments_mime_type_filter(tmp_path, monkeypatch) -> None:
+    """Non-allowed MIME types are skipped."""
+    monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
+
+    raw = _make_raw_email_with_attachment(
+        attachment_name="image.png",
+        attachment_content=b"\x89PNG fake",
+        attachment_mime="image/png",
+    )
+    fake = _make_fake_imap(raw)
+    monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
+
+    cfg = _make_config(
+        allowed_attachment_types=["application/pdf"],
+        verify_dkim=False,
+        verify_spf=False,
+    )
+    channel = EmailChannel(cfg, MessageBus())
+    items = channel._fetch_new_messages()
+
+    assert len(items) == 1
+    assert items[0]["media"] == []
+
+
+def test_extract_attachments_empty_allowed_types_rejects_all(tmp_path, monkeypatch) -> None:
+    """Empty allowed_attachment_types means no types are accepted."""
+    monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
+
+    raw = _make_raw_email_with_attachment(
+        attachment_name="image.png",
+        attachment_content=b"\x89PNG fake",
+        attachment_mime="image/png",
+    )
+    fake = _make_fake_imap(raw)
+    monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
+
+    cfg = _make_config(
+        allowed_attachment_types=[],
+        verify_dkim=False,
+        verify_spf=False,
+    )
+    channel = EmailChannel(cfg, MessageBus())
+    items = channel._fetch_new_messages()
+
+    assert len(items) == 1
+    assert items[0]["media"] == []
+
+
+def test_extract_attachments_wildcard_pattern(tmp_path, monkeypatch) -> None:
+    """Glob patterns like 'image/*' match attachment MIME types."""
+    monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
+
+    raw = _make_raw_email_with_attachment(
+        attachment_name="photo.jpg",
+        attachment_content=b"\xff\xd8\xff fake jpeg",
+        attachment_mime="image/jpeg",
+    )
+    fake = _make_fake_imap(raw)
+    monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
+
+    cfg = _make_config(
+        allowed_attachment_types=["image/*"],
+        verify_dkim=False,
+        verify_spf=False,
+    )
+    channel = EmailChannel(cfg, MessageBus())
+    items = channel._fetch_new_messages()
+
+    assert len(items) == 1
+    assert len(items[0]["media"]) == 1
+
+
+def test_extract_attachments_size_limit(tmp_path, monkeypatch) -> None:
+    """Attachments exceeding max_attachment_size are skipped."""
+    monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
+
+    raw = _make_raw_email_with_attachment(
+        attachment_content=b"x" * 1000,
+    )
+    fake = _make_fake_imap(raw)
+    monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
+
+    cfg = _make_config(
+        allowed_attachment_types=["*"],
+        max_attachment_size=500,
+        verify_dkim=False,
+        verify_spf=False,
+    )
+    channel = EmailChannel(cfg, MessageBus())
+    items = channel._fetch_new_messages()
+
+    assert len(items) == 1
+    assert items[0]["media"] == []
+
+
+def test_extract_attachments_max_count(tmp_path, monkeypatch) -> None:
+    """Only max_attachments_per_email are saved."""
+    monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
+
+    # Build email with 3 attachments
+    msg = EmailMessage()
+    msg["From"] = "alice@example.com"
+    msg["To"] = "bot@example.com"
+    msg["Subject"] = "Many attachments"
+    msg["Message-ID"] = "<m1@example.com>"
+    msg.set_content("See attached.")
+    for i in range(3):
+        msg.add_attachment(
+            f"content {i}".encode(),
+            maintype="application",
+            subtype="pdf",
+            filename=f"doc{i}.pdf",
+        )
+    raw = msg.as_bytes()
+
+    fake = _make_fake_imap(raw)
+    monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
+
+    cfg = _make_config(
+        allowed_attachment_types=["*"],
+        max_attachments_per_email=2,
+        verify_dkim=False,
+        verify_spf=False,
+    )
+    channel = EmailChannel(cfg, MessageBus())
+    items = channel._fetch_new_messages()
+
+    assert len(items) == 1
+    assert len(items[0]["media"]) == 2
+
+
+def test_extract_attachments_sanitizes_filename(tmp_path, monkeypatch) -> None:
+    """Path traversal in filenames is neutralized."""
+    monkeypatch.setattr("nanobot.channels.email.get_media_dir", lambda ch: tmp_path)
+
+    raw = _make_raw_email_with_attachment(
+        attachment_name="../../../etc/passwd",
+    )
+    fake = _make_fake_imap(raw)
+    monkeypatch.setattr("nanobot.channels.email.imaplib.IMAP4_SSL", lambda _h, _p: fake)
+
+    cfg = _make_config(allowed_attachment_types=["*"], verify_dkim=False, verify_spf=False)
+    channel = EmailChannel(cfg, MessageBus())
+    items = channel._fetch_new_messages()
+
+    assert len(items) == 1
+    assert len(items[0]["media"]) == 1
+    saved_path = Path(items[0]["media"][0])
+    # File must be inside the media dir, not escaped via path traversal
+    assert saved_path.parent == tmp_path