RoyirinRosmontis/shirobana-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial import of Shirobana bot backend

Browse Source 
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
This commit is contained in:
RRRosmontis
2026-11-11 00:25:30 +08:00
commit 34089d0042
106 changed files with 28028 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_mail_attachments.py
+369
View File
@@ -0,0 +1,369 @@
# -*- coding: utf-8 -*-
"""附件邮件协议、校验与 MIME 测试；全部外部邮件/COM 操作均 mock。"""
import asyncio
import base64
import email
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
from flask import Flask
BASE = Path(__file__).resolve().parents[1]
sys.path[:0] = [str(BASE / "bot-utility"), str(BASE / "mail-bridge"), str(BASE)]
import mail # noqa: E402
# com_backend 在 import 时读取配置；这里只提供无真实凭据的测试配置。
_CONFIG_TMP = tempfile.TemporaryDirectory(prefix="mail-attachment-test-config-")
_CONFIG_PATH = Path(_CONFIG_TMP.name) / "com-private.json"
_CONFIG_PATH.write_text(json.dumps({
"hmail_admin_password": "not-used-by-mock-tests",
"token": "t" * 128,
"allowed_domains": ["rrrosmontis.icu", "aeolianchi.me"],
}), encoding="utf-8")
with patch.dict(os.environ, {"MAIL_COM_CONFIG": str(_CONFIG_PATH)}):
import com_backend as backend # noqa: E402
import admin_com # noqa: E402
SENDER = "sender@rrrosmontis.icu"
RECIPIENT = "receiver@example.net"
AUTH = {"Authorization": "Bearer " + "t" * 128}
ENDPOINT = "/v1/mail/api/admin/com/send_with_attachments"
SEND_ENDPOINT = "/v1/mail/api/admin/com/send"
def outgoing_payload(**overrides):
data = {
"from_address": SENDER,
"to_address": RECIPIENT,
"subject": "附件测试",
"body_text": "正文",
"attachments": [],
}
data.update(overrides)
return data
def wire_attachment(content=b"hello", filename="测试.txt",
content_type="text/plain"):
return {
"filename": filename,
"content_type": content_type,
"data_base64": base64.b64encode(content).decode("ascii"),
}
class ClientAttachmentTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix="mail-attachment-client-")
self.mail_dir = Path(self.tmp.name) / "mail"
self.mail_dir.mkdir()
self.dir_patch = patch.object(mail, "MAIL_DIR", str(self.mail_dir))
self.dir_patch.start()
mail._save_user(42, {
"addresses": [SENDER],
"default": SENDER,
"push": {"enabled": False, "interval_minutes": 5, "last_seen": {}},
})
self.files = Path(self.tmp.name) / "files"
self.files.mkdir()
def tearDown(self):
self.dir_patch.stop()
self.tmp.cleanup()
def send(self, paths, **kwargs):
args = dict(user_id=42, from_address=SENDER, to_address=RECIPIENT,
subject="主题", body_text="正文", attachment_paths=paths)
args.update(kwargs)
return asyncio.run(mail.send_with_attachments(**args))
def test_success_uses_base64_protocol_and_chinese_filename(self):
path = self.files / "中文 报告.txt"
content = "附件内容".encode("utf-8")
path.write_bytes(content)
app = MagicMock()
app.post.return_value = {"ok": True, "accepted": True, "sent_saved": True}
with patch.object(mail, "_ComSession") as session:
session.return_value.__enter__.return_value = app
result = self.send([path])
self.assertTrue(result["ok"])
self.assertEqual(result["from"], SENDER)
self.assertEqual(result["to"], RECIPIENT)
self.assertEqual(result["attachment_names"], ["中文 报告.txt"])
self.assertEqual(result["attachment_count"], 1)
self.assertEqual(result["size"], len(content))
self.assertTrue(result["sent_saved"])
endpoint, payload = app.post.call_args.args
self.assertEqual(endpoint, "/send_with_attachments")
self.assertNotIn("path", payload["attachments"][0])
self.assertEqual(payload["attachments"][0]["filename"], "中文 报告.txt")
self.assertEqual(base64.b64decode(
payload["attachments"][0]["data_base64"], validate=True), content)
self.assertEqual(payload["attachments"][0]["content_type"], "text/plain")
def test_from_address_must_be_bound(self):
with patch.object(mail, "_ComSession") as session:
result = self.send([], from_address="other@rrrosmontis.icu")
self.assertFalse(result["ok"])
self.assertEqual(result["error_code"], "from_not_bound")
session.assert_not_called()
def test_recipient_subject_and_body_validation(self):
cases = [
({"to_address": "not-an-address"}, "invalid_recipient"),
({"subject": "bad\nsubject"}, "invalid_subject"),
({"subject": " "}, "invalid_subject"),
({"body_text": None}, "invalid_body"),
]
for overrides, code in cases:
with self.subTest(code=code):
with patch.object(mail, "_ComSession") as session:
result = self.send([], **overrides)
self.assertFalse(result["ok"])
self.assertEqual(result["error_code"], code)
session.assert_not_called()
def test_single_file_total_and_count_limits(self):
one = self.files / "one.bin"
one.write_bytes(b"12345")
with patch.object(mail, "MAX_ATTACHMENT_BYTES", 4):
result = self.send([one])
self.assertEqual(result["error_code"], "file_too_large")
two = self.files / "two.bin"
one.write_bytes(b"123")
two.write_bytes(b"456")
with patch.object(mail, "MAX_ATTACHMENT_BYTES", 4), \
patch.object(mail, "MAX_ATTACHMENT_TOTAL_BYTES", 5):
result = self.send([one, two])
self.assertEqual(result["error_code"], "total_too_large")
with patch.object(mail, "MAX_ATTACHMENT_COUNT", 1):
result = self.send([one, two])
self.assertEqual(result["error_code"], "too_many_attachments")
def test_directory_and_special_file_are_rejected(self):
result = self.send([self.files])
self.assertFalse(result["ok"])
self.assertEqual(result["error_code"], "not_regular_file")
if Path("/dev/null").exists():
result = self.send(["/dev/null"])
self.assertEqual(result["error_code"], "not_regular_file")
def test_transport_failure_is_safe_and_structured(self):
secret_path = self.files / "private-name.txt"
secret_path.write_bytes(b"x")
with patch.object(mail, "_ComSession", side_effect=RuntimeError(
"token and /private/path must not leak")):
result = self.send([secret_path])
self.assertFalse(result["ok"])
self.assertEqual(result["error_code"], "send_unknown")
self.assertNotIn("private", result["error"])
self.assertNotIn("token", result["error"])
def test_legacy_client_send_payload_is_unchanged(self):
app = MagicMock()
mail._send_mail(app, SENDER, RECIPIENT, "旧主题", "旧正文")
app.post.assert_called_once_with("/send", {
"from_address": SENDER,
"to_address": RECIPIENT,
"subject": "旧主题",
"body_text": "旧正文",
})
class BridgeAttachmentTests(unittest.TestCase):
def setUp(self):
self.flask = Flask(__name__)
self.flask.config.update(TESTING=True)
admin_com.install(self.flask)
self.client = self.flask.test_client()
self.app = MagicMock()
self.session_patch = patch.object(backend, "_ComSession")
session = self.session_patch.start()
session.return_value.__enter__.return_value = self.app
self.account_patch = patch.object(backend, "_get_account")
self.account_patch.start()
def tearDown(self):
self.account_patch.stop()
self.session_patch.stop()
def post(self, payload, endpoint=ENDPOINT):
return self.client.post(endpoint, json=payload, headers=AUTH)
def test_new_endpoint_decodes_attachments_and_preserves_chinese_name(self):
content = "中文附件".encode("utf-8")
with patch.object(backend, "_send_mail_with_attachments", return_value=True) as send:
response = self.post(outgoing_payload(
attachments=[wire_attachment(content=content)]))
self.assertEqual(response.status_code, 200)
self.assertEqual(response.get_json(), {
"ok": True,
"accepted": True,
"sent_saved": True,
"attachment_names": ["测试.txt"],
"size": len(content),
})
args = send.call_args.args
self.assertEqual(args[1:5], (SENDER, RECIPIENT, "附件测试", "正文"))
self.assertEqual(args[5][0]["filename"], "测试.txt")
self.assertEqual(args[5][0]["content"], content)
def test_new_endpoint_rejects_paths_bad_mime_and_bad_base64(self):
bad_attachments = [
dict(wire_attachment(), filename="../secret.txt"),
dict(wire_attachment(), filename="C:\\secret.txt"),
dict(wire_attachment(), content_type="text/plain\r\nX-Evil: yes"),
dict(wire_attachment(), content_type="text/plain; charset=utf-8"),
dict(wire_attachment(), data_base64="not base64!"),
dict(wire_attachment(), path="C:/secret.txt"),
]
with patch.object(backend, "_send_mail_with_attachments") as send:
for attachment in bad_attachments:
with self.subTest(attachment=attachment):
response = self.post(outgoing_payload(attachments=[attachment]))
self.assertEqual(response.status_code, 400)
self.assertEqual(response.get_json(), {"error": "invalid request"})
send.assert_not_called()
def test_new_endpoint_rejects_empty_attachment_list_without_sending(self):
with patch.object(backend, "_send_mail_with_attachments") as send:
response = self.post(outgoing_payload(attachments=[]))
self.assertEqual(response.status_code, 400)
send.assert_not_called()
def test_new_endpoint_revalidates_count_file_and_total_size(self):
with patch.object(admin_com, "MAX_ATTACHMENT_COUNT", 1), \
patch.object(backend, "_send_mail_with_attachments") as send:
response = self.post(outgoing_payload(attachments=[
wire_attachment(filename="a.txt"), wire_attachment(filename="b.txt")]))
self.assertEqual(response.status_code, 400)
send.assert_not_called()
with patch.object(admin_com, "MAX_ATTACHMENT_BYTES", 4), \
patch.object(backend, "_send_mail_with_attachments") as send:
response = self.post(outgoing_payload(
attachments=[wire_attachment(content=b"12345")]))
self.assertEqual(response.status_code, 400)
send.assert_not_called()
with patch.object(admin_com, "MAX_ATTACHMENT_BYTES", 4), \
patch.object(admin_com, "MAX_ATTACHMENT_TOTAL_BYTES", 5), \
patch.object(backend, "_send_mail_with_attachments") as send:
response = self.post(outgoing_payload(attachments=[
wire_attachment(content=b"123", filename="a.txt"),
wire_attachment(content=b"456", filename="b.txt")]))
self.assertEqual(response.status_code, 400)
send.assert_not_called()
def test_request_body_limits_are_endpoint_specific(self):
# 大于旧端点 1 MiB、小于附件端点默认 40 MiB 的合法请求可通过入口限制。
large_content = b"x" * (800 * 1024)
with patch.object(backend, "_send_mail_with_attachments", return_value=True):
response = self.post(outgoing_payload(
attachments=[wire_attachment(content=large_content)]))
self.assertEqual(response.status_code, 200)
response = self.client.post(
SEND_ENDPOINT, data=b"x" * (1024 * 1024 + 1),
content_type="application/json", headers=AUTH)
self.assertEqual(response.status_code, 413)
# 用小上限 mock 大请求，验证附件端点自身也有独立硬限制。
with patch.object(admin_com, "ATTACHMENT_REQUEST_MAX_BYTES", 1024):
response = self.client.post(
ENDPOINT, data=b"x" * 1025,
content_type="application/json", headers=AUTH)
self.assertEqual(response.status_code, 413)
def test_legacy_send_endpoint_contract_is_unchanged(self):
payload = outgoing_payload()
payload.pop("attachments")
with patch.object(backend, "_send_mail") as old_send, \
patch.object(backend, "_send_mail_with_attachments") as new_send:
response = self.post(payload, endpoint=SEND_ENDPOINT)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.get_json(), {"accepted": True})
old_send.assert_called_once_with(
self.app, SENDER, RECIPIENT, "附件测试", "正文")
new_send.assert_not_called()
class BackendMimeTests(unittest.TestCase):
def test_mime_contains_utf8_body_and_rfc2231_chinese_filename(self):
content = "二进制\x00内容".encode("utf-8")
msg = backend._build_attachment_mime(
SENDER, RECIPIENT, "中文主题", "纯文本正文",
[{"filename": "中文 报告.txt", "content_type": "text/plain",
"content": content}])
parsed = email.message_from_bytes(msg.as_bytes())
self.assertEqual(parsed.get_content_type(), "multipart/mixed")
self.assertEqual(parsed["To"], RECIPIENT)
attachment = None
plain = None
for part in parsed.walk():
if part.get_content_disposition() == "attachment":
attachment = part
elif part.get_content_type() == "text/plain":
plain = part
self.assertIsNotNone(attachment)
self.assertEqual(attachment.get_filename(), "中文 报告.txt")
self.assertEqual(attachment.get_payload(decode=True), content)
self.assertEqual(plain.get_payload(decode=True).decode("utf-8"), "纯文本正文")
def test_smtp_delivery_is_mocked_and_uses_raw_mime(self):
smtp = MagicMock()
smtp.sendmail.return_value = {}
manager = MagicMock()
manager.__enter__.return_value = smtp
with patch.object(backend.smtplib, "SMTP", return_value=manager) as factory, \
patch.dict(backend.CONFIG, {"smtp_host": "127.0.0.1", "smtp_port": 25,
"smtp_ssl": False, "smtp_starttls": False,
"smtp_username": ""}, clear=False):
backend._deliver_mime(SENDER, RECIPIENT, b"MIME-BYTES")
factory.assert_called_once_with("127.0.0.1", 25, timeout=30.0)
smtp.sendmail.assert_called_once_with(SENDER, [RECIPIENT], b"MIME-BYTES")
def test_no_attachments_reuses_legacy_send_and_sent(self):
app = MagicMock()
with patch.object(backend, "_send_mail") as old_send, \
patch.object(backend, "_save_to_sent") as old_sent, \
patch.object(backend, "_deliver_mime") as smtp:
result = backend._send_mail_with_attachments(
app, SENDER, RECIPIENT, "主题", "正文", [])
self.assertIsNone(result)
old_send.assert_called_once_with(app, SENDER, RECIPIENT, "主题", "正文")
old_sent.assert_called_once_with(app, SENDER, RECIPIENT, "主题", "正文")
smtp.assert_not_called()
def test_sent_copy_uses_documented_com_attachment_add(self):
message = MagicMock()
added = []
def add_attachment(path):
added.append((os.path.basename(path), Path(path).read_bytes()))
message.Attachments.Add.side_effect = add_attachment
account = MagicMock()
account.IMAPFolders.ItemByName.return_value.Messages.Add.return_value = message
with patch.object(backend, "_get_account", return_value=account):
saved = backend._save_attachment_copy_to_sent(
MagicMock(), SENDER, RECIPIENT, "主题", "正文",
[{"filename": "中文.txt", "content_type": "text/plain",
"content": b"content"}])
self.assertTrue(saved)
self.assertEqual(added, [("中文.txt", b"content")])
message.AddRecipient.assert_called_once_with("", RECIPIENT)
message.Save.assert_called_once_with()
if __name__ == "__main__":
unittest.main()