fuzzyworld

fuzzyworld-tools / files / tests / test_comments.py

test_comments.py

45.3 KB · history · blame

1"""Tests for server/comments.py (blog comments by email, moderation CLI, publishing).
2
3Run with: python3 -m unittest discover tests
4"""
5
6import contextlib
7import importlib.util
8import io
9import itertools
10import json
11import mailbox
12import os
13import shutil
14import tempfile
15import unittest
16
17ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
18spec = importlib.util.spec_from_file_location("comments", os.path.join(ROOT, "server", "comments.py"))
19comments = importlib.util.module_from_spec(spec)
20spec.loader.exec_module(comments)
21
22POST = "/2024/05/01/hello/"
23
24
25_counter = itertools.count()
26
27
28def mail(subject="Comment [writing:/2024/05/01/hello/]", body="Nice post.\n", sender="Ann <ann@example.org>",
29 auth=("mx; dkim=pass header.d=example.org; dmarc=fail",), extra_auth_below=(), extra_headers=()):
30 lines = [f"Message-ID: <{next(_counter)}@test>", *extra_headers]
31 for a in auth:
32 lines.append(f"Authentication-Results: {a}")
33 lines += [f"From: {sender}", "To: comments@fuzzyworld.net", f"Subject: {subject}"]
34 for a in extra_auth_below:
35 lines.append(f"Authentication-Results: {a}")
36 lines += ["Content-Type: text/plain; charset=utf-8", "", body]
37 return "\n".join(lines).encode()
38
39
40class Base(unittest.TestCase):
41 def setUp(self):
42 self.tmp = tempfile.mkdtemp()
43 self.addCleanup(shutil.rmtree, self.tmp)
44 j = lambda *p: os.path.join(self.tmp, *p)
45 self.root = j("writing")
46 os.makedirs(j("writing", POST.strip("/")))
47 open(j("writing", POST.strip("/"), "index.html"), "w").write("x")
48 self.til = j("til") # does not exist: existence check skipped
49 json.dump({"writing": {"origin": "https://w", "root": self.root}, "til": {"origin": "https://t", "root": self.til},
50 "mailbox": "", "authserv_id": "mx"}, open(j("sites.json"), "w"))
51 self.maildir = j("Maildir", ".comments")
52 env = {"FW_COMMENTS_DIR": j("state"), "FW_COMMENTS_SITES": j("sites.json"), "FW_COMMENTS_MAILDIR": self.maildir}
53 old = {k: os.environ.get(k) for k in env}
54 os.environ.update(env)
55 self.addCleanup(lambda: [os.environ.pop(k) if v is None else os.environ.__setitem__(k, v) for k, v in old.items()])
56 self.state = j("state")
57
58 def run_cli(self, *argv):
59 out, err = io.StringIO(), io.StringIO()
60 with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
61 code = comments.main(list(argv))
62 return code, out.getvalue(), err.getvalue()
63
64 def deliver(self, raw):
65 os.makedirs(os.path.dirname(self.maildir), exist_ok=True)
66 mailbox.Maildir(self.maildir, create=True).add(raw)
67
68 def count(self, suffix):
69 d = self.maildir + suffix
70 return len(mailbox.Maildir(d, create=False).keys()) if os.path.isdir(d) else 0
71
72 def pending(self):
73 return comments.pending_ids()
74
75 def pub(self, path="2024/05/01/hello.json"):
76 return os.path.join(self.root, "_comments", path)
77
78 def add_pending(self, **kw):
79 c = {"id": comments.new_id(), "site": "writing", "path": POST, "author": "Ann", "text": "hi",
80 "date": comments.now_iso(), "source": "form", "parent": None}
81 c.update(kw)
82 comments.write_json(comments.pending_path(c["id"]), c)
83 return c["id"]
84
85
86class ListJson(Base):
87 def js(self):
88 code, out, err = self.run_cli("list", "--json")
89 self.assertEqual((code, err), (0, ""))
90 for word in ("sender_hash", "msg_key", "mkey"):
91 self.assertNotIn(word, out)
92 return json.loads(out)
93
94 def test_pending_fields_post_url_and_trusted(self):
95 a = self.add_pending(source="email", sender_hash="h1", text="full <b>text</b>\nline two " * 30)
96 b = self.add_pending(source="form")
97 comments.write_json(comments.trusted_path(), ["h1"])
98 pending = {c["id"]: c for c in self.js()["pending"]}
99 self.assertEqual(set(pending[a]), {"id", "site", "path", "post_url", "author", "date", "source", "parent", "text", "trusted"})
100 self.assertEqual(pending[a]["post_url"], "https://w" + POST)
101 self.assertEqual(pending[a]["text"], "full <b>text</b>\nline two " * 30) # in full, not cut to 160
102 self.assertEqual((pending[a]["trusted"], pending[b]["trusted"]), (True, False))
103
104 def test_approved_are_the_newest_twenty_and_removable(self):
105 ids = []
106 for i in range(23):
107 ids.append(self.add_pending(date=f"2024-05-01T10:00:{i:02d}Z", text=f"c{i}", sender_hash=f"s{i}"))
108 self.assertEqual(self.run_cli("approve", *ids)[0], 0)
109 data = self.js()
110 self.assertEqual(data["pending"], [])
111 self.assertEqual(len(data["approved"]), 20)
112 self.assertEqual(data["approved"][0]["text"], "c22")
113 self.assertEqual(data["approved"][-1]["text"], "c3")
114 self.assertEqual(set(data["approved"][0]), {"id", "site", "path", "post_url", "author", "date", "source", "parent", "text", "trusted"})
115 self.assertFalse(data["approved"][0]["trusted"])
116 self.run_cli("remove", data["approved"][0]["id"])
117 self.assertEqual(len(self.js()["approved"]), 20)
118 self.assertNotIn("c22", [c["text"] for c in self.js()["approved"]])
119
120 def test_empty_and_bad_flags(self):
121 self.assertEqual(self.js(), {"pending": [], "approved": []})
122 self.assertEqual(self.run_cli("list", "--bogus")[0], 2)
123
124
125class Moderation(Base):
126 def test_approve_publishes_without_private_fields(self):
127 cid = self.add_pending(source="email", sender_hash="abc")
128 self.assertEqual(self.run_cli("approve", cid)[0], 0)
129 self.assertEqual(self.pending(), [])
130 pub = json.load(open(self.pub()))
131 self.assertEqual([c["id"] for c in pub], [cid])
132 self.assertEqual(set(pub[0]), {"author", "date", "id", "parent", "text"})
133 self.assertNotIn("abc", open(os.path.join(self.state, "approved", "writing.json")).read())
134 self.assertEqual(json.load(open(os.path.join(self.state, "approved-meta.json"))), {cid: "abc"})
135
136 def test_reject_logs(self):
137 cid = self.add_pending()
138 self.run_cli("reject", cid)
139 self.assertEqual(self.pending(), [])
140 self.assertIn(cid, open(os.path.join(self.state, "rejected.log")).read())
141 self.assertFalse(os.path.exists(self.pub()))
142
143 def test_remove_republishes_and_empty_removes_file(self):
144 a, b = self.add_pending(), self.add_pending()
145 self.run_cli("approve", a, b)
146 self.run_cli("remove", a)
147 self.assertEqual([c["id"] for c in json.load(open(self.pub()))], [b])
148 self.run_cli("remove", b)
149 self.assertFalse(os.path.exists(self.pub()))
150 self.assertEqual(self.run_cli("remove", b)[0], 1)
151
152 def test_trust_and_untrust(self):
153 cid = self.add_pending(source="email", sender_hash="h1")
154 self.run_cli("approve", cid, "--trust")
155 self.assertEqual(json.load(open(os.path.join(self.state, "trusted.json"))), ["h1"])
156 self.run_cli("untrust", cid)
157 self.assertEqual(json.load(open(os.path.join(self.state, "trusted.json"))), [])
158
159 def test_trust_refused_for_form_comment(self):
160 cid = self.add_pending(source="form")
161 self.assertEqual(self.run_cli("approve", cid, "--trust")[0], 1)
162 self.assertEqual(self.pending(), [cid])
163
164 def test_publish_rewrites_files(self):
165 cid = self.add_pending()
166 self.run_cli("approve", cid)
167 os.unlink(self.pub())
168 self.run_cli("publish", "writing")
169 self.assertTrue(os.path.exists(self.pub()))
170 self.assertIn(cid, self.run_cli("export", "writing")[1])
171
172 def test_list_and_show(self):
173 cid = self.add_pending(text="x" * 300)
174 out = self.run_cli("list")[1]
175 self.assertIn(cid, out)
176 self.assertNotIn("x" * 161, out)
177 self.assertIn(cid, self.run_cli("show", cid)[1])
178
179 def test_traversal_paths_refused(self):
180 for bad in ("/../../etc/", "/a/../b/", "/./", "/..%2f/", "/x", "/", "//", "/a//b/", "/%2e%2e/", "/a/%2E/", "/a%2fb/", "/a%00b/", "/caf%e9/", "/%2e/", "/" + "a/" * 9):
181 self.assertFalse(comments.valid_path(bad), bad)
182 cid = self.add_pending(path="/../../evil/")
183 self.assertEqual(self.run_cli("approve", cid)[0], 1)
184 self.assertFalse(os.path.exists(os.path.join(self.tmp, "evil.json")))
185 self.assertEqual(self.run_cli("show", "../x")[0], 1)
186
187 def test_missing_site_root_is_not_created(self):
188 shutil.rmtree(self.root)
189 cid = self.add_pending()
190 self.assertEqual(self.run_cli("approve", cid)[0], 0)
191 self.assertFalse(os.path.exists(self.root))
192 self.assertEqual(len(json.load(open(os.path.join(self.state, "approved", "writing.json")))[POST]), 1)
193 os.makedirs(self.root)
194 self.run_cli("publish")
195 self.assertTrue(os.path.exists(self.pub()))
196
197 def test_valid_paths(self):
198 for ok in ("/hello/", POST, "/about.me/", "/a%20b/", "/a/b/c/", "/2024/05/01/", "/blog/hello/", "/caf%c3%a9/", "/_posts/~me/x+y/"):
199 self.assertTrue(comments.valid_path(ok), ok)
200
201
202class Ingest(Base):
203 def ingest(self):
204 return self.run_cli("ingest")
205
206 def test_aligned_dkim_pass(self):
207 self.deliver(mail())
208 self.ingest()
209 self.assertEqual(len(self.pending()), 1)
210 c = json.load(open(comments.pending_path(self.pending()[0])))
211 self.assertEqual((c["author"], c["site"], c["path"], c["source"]), ("Ann", "writing", POST, "email"))
212 self.assertEqual(c["sender_hash"], comments.sender_hash("ANN@example.org"))
213 self.assertNotIn("ann@example.org", json.dumps(c))
214 self.assertEqual((self.count(""), self.count(".done"), self.count(".bad")), (0, 1, 0))
215
216 def test_dkim_subdomain_of_from_domain_ok_but_not_parent(self):
217 self.deliver(mail(auth=("mx; dkim=pass header.d=mail.example.org",)))
218 self.deliver(mail(sender="Bob <bob@mail.example.org>", auth=("mx; dkim=pass header.d=example.org",)))
219 self.ingest()
220 self.assertEqual((len(self.pending()), self.count(".bad")), (1, 1))
221
222 def test_dmarc_pass(self):
223 self.deliver(mail(auth=("mx; dmarc=pass header.from=example.org",)))
224 self.ingest()
225 self.assertEqual(len(self.pending()), 1)
226
227 def test_dmarc_fail(self):
228 self.deliver(mail(auth=("mx; dkim=fail; dmarc=fail",)))
229 self.ingest()
230 self.assertEqual((len(self.pending()), self.count(".bad")), (0, 1))
231 self.assertIn("not authenticated", open(os.path.join(self.state, "rejected.log")).read())
232
233 def test_unaligned_dkim_rejected(self):
234 self.deliver(mail(auth=("mx; dkim=pass header.d=evil.test; dmarc=fail",)))
235 self.ingest()
236 self.assertEqual(self.count(".bad"), 1)
237
238 def test_forged_lower_header_ignored(self):
239 self.deliver(mail(auth=("mx; dmarc=fail",), extra_auth_below=("mx; dmarc=pass",)))
240 self.ingest()
241 self.assertEqual((len(self.pending()), self.count(".bad")), (0, 1))
242
243 def set_cfg(self, **kw):
244 cfg = json.load(open(os.environ["FW_COMMENTS_SITES"]))
245 cfg.update(kw)
246 json.dump(cfg, open(os.environ["FW_COMMENTS_SITES"], "w"))
247
248 def test_empty_authserv_id_refuses(self):
249 self.set_cfg(authserv_id="")
250 self.deliver(mail())
251 code, out, err = self.ingest()
252 self.assertEqual(code, 1)
253 self.assertIn("authserv_id", err)
254 self.assertEqual((self.count(""), self.count(".done"), self.count(".bad"), self.pending()), (1, 0, 0, []))
255
256 def test_only_topmost_header_counts(self):
257 # topmost is someone else's: rejected even though a lower header has our id and passes
258 self.deliver(mail(auth=("evil.test; dmarc=pass header.from=example.org",), extra_auth_below=("mx; dmarc=pass header.from=example.org",)))
259 # topmost absent, only a forged lower header with our id (it is then the topmost, but a sender-made one
260 # cannot be told apart, so the real protection is the receiving server placing its own on top)
261 self.deliver(mail(auth=("evil.test; dmarc=pass header.from=example.org", "mx; dmarc=pass header.from=example.org")))
262 self.deliver(mail(auth=("mx; dmarc=pass header.from=example.org", "evil.test; dmarc=fail")))
263 self.ingest()
264 self.assertEqual((len(self.pending()), self.count(".bad")), (1, 2))
265
266 def test_no_auth_header(self):
267 self.deliver(mail(auth=()))
268 self.ingest()
269 self.assertEqual(self.count(".bad"), 1)
270
271 def test_subject_without_tag(self):
272 self.deliver(mail(subject="hello there"))
273 self.ingest()
274 self.assertEqual((len(self.pending()), self.count(".bad")), (0, 1))
275
276 def test_unknown_post_when_root_exists(self):
277 self.deliver(mail(subject="[writing:/2024/05/01/missing/]"))
278 self.deliver(mail(subject="[writing:/../../etc/]"))
279 self.ingest()
280 self.assertEqual((len(self.pending()), self.count(".bad")), (0, 2))
281
282 def test_til_site_root_missing_skips_existence_check(self):
283 self.deliver(mail(subject="[til:/2024/05/01/x/]"))
284 self.ingest()
285 self.assertEqual(len(self.pending()), 1)
286
287 def test_re_prefix(self):
288 self.deliver(mail(subject="Re: Fwd: Thoughts [writing:/2024/05/01/hello/] more"))
289 self.ingest()
290 self.assertEqual(len(self.pending()), 1)
291
292 def test_quotes_and_signature_cut(self):
293 self.deliver(mail(body="> old text\nMy reply\n> more old\nsecond line\n\n-- \nAnn\nsecret sig\n"))
294 self.ingest()
295 c = json.load(open(comments.pending_path(self.pending()[0])))
296 self.assertEqual(c["text"], "My reply\nsecond line")
297
298 def test_oversized_body_and_name_capped_and_controls_dropped(self):
299 self.deliver(mail(sender='"' + "N" * 100 + '" <ann@example.org>', body="a\x01b" + "x" * 9000 + "\n"))
300 self.ingest()
301 c = json.load(open(comments.pending_path(self.pending()[0])))
302 self.assertEqual(len(c["text"]), 4000)
303 self.assertNotIn("\x01", c["text"])
304 self.assertEqual(len(c["author"]), 60)
305
306 def test_missing_name_gives_a_reader(self):
307 self.deliver(mail(sender="ann@example.org"))
308 self.ingest()
309 self.assertEqual(json.load(open(comments.pending_path(self.pending()[0])))["author"], "A reader")
310
311 def test_trusted_sender_auto_publishes(self):
312 comments.write_json(os.path.join(self.state, "trusted.json"), [comments.sender_hash("ann@example.org")])
313 self.deliver(mail())
314 self.ingest()
315 self.assertEqual(self.pending(), [])
316 self.assertEqual(json.load(open(self.pub()))[0]["text"], "Nice post.")
317
318 def test_second_run_is_noop(self):
319 self.deliver(mail())
320 self.deliver(mail(subject="nope"))
321 self.ingest()
322 before = (self.pending(), open(os.path.join(self.state, "rejected.log")).read())
323 code, out, _ = self.ingest()
324 self.assertEqual((code, out), (0, ""))
325 self.assertEqual((self.pending(), open(os.path.join(self.state, "rejected.log")).read()), before)
326
327 def test_pending_cap(self):
328 for _ in range(comments.PENDING_CAP):
329 self.add_pending()
330 self.deliver(mail())
331 self.ingest()
332 self.assertEqual((len(self.pending()), self.count(".bad")), (comments.PENDING_CAP, 1))
333 self.assertIn("queue full", open(os.path.join(self.state, "rejected.log")).read())
334
335 def test_multiple_from_headers_rejected(self):
336 self.deliver(mail(extra_headers=("From: Bob <bob@example.org>",)))
337 self.deliver(mail(sender="Ann <ann@example.org>, Bob <bob@example.org>"))
338 self.ingest()
339 self.assertEqual((len(self.pending()), self.count(".bad")), (0, 2))
340
341 def test_dmarc_pass_needs_matching_header_from(self):
342 self.deliver(mail(auth=("mx; dmarc=pass header.from=evil.test",)))
343 self.deliver(mail(auth=("mx; dmarc=pass",)))
344 self.deliver(mail(auth=('mx; dmarc=pass header.from="Example.org"',)))
345 self.ingest()
346 self.assertEqual((len(self.pending()), self.count(".bad")), (1, 2))
347
348 def test_quoted_semicolon_does_not_fake_a_result(self):
349 self.deliver(mail(auth=('mx; dmarc=fail; spf=pass smtp.mailfrom="x;dmarc=pass header.from=example.org@evil.test"',)))
350 self.deliver(mail(auth=('mx (a;dmarc=pass header.from=example.org); dmarc=fail',)))
351 self.ingest()
352 self.assertEqual((len(self.pending()), self.count(".bad")), (0, 2))
353
354 def test_worst_result_of_a_method_wins(self):
355 self.deliver(mail(auth=("mx; dmarc=pass header.from=example.org; dmarc=fail header.from=example.org",)))
356 self.deliver(mail(auth=("mx; dkim=pass header.d=example.org; dkim=fail header.d=other.test",)))
357 self.deliver(mail(auth=("mx; dkim=pass header.d=example.org; dkim=pass header.d=other.test",)))
358 self.ingest()
359 self.assertEqual((len(self.pending()), self.count(".bad")), (1, 2))
360
361 def test_crash_between_store_and_move_does_not_duplicate(self):
362 self.deliver(mail())
363 real = comments.MaildirSource.finish
364 def crash(*a, **k):
365 raise RuntimeError("crash")
366 comments.MaildirSource.finish = crash
367 try:
368 with self.assertRaises(RuntimeError):
369 self.ingest()
370 finally:
371 comments.MaildirSource.finish = real
372 self.assertEqual((len(self.pending()), self.count("")), (1, 1))
373 self.ingest()
374 self.assertEqual((len(self.pending()), self.count(""), self.count(".done")), (1, 0, 1))
375 self.assertIn("duplicate", open(os.path.join(self.state, "rejected.log")).read())
376
377 def test_crash_after_approval_does_not_duplicate_approved(self):
378 comments.write_json(os.path.join(self.state, "trusted.json"), [comments.sender_hash("ann@example.org")])
379 self.deliver(mail())
380 real = comments.MaildirSource.finish
381 comments.MaildirSource.finish = lambda *a: (_ for _ in ()).throw(RuntimeError("crash"))
382 try:
383 with self.assertRaises(RuntimeError):
384 self.ingest()
385 finally:
386 comments.MaildirSource.finish = real
387 self.ingest()
388 self.assertEqual(len(json.load(open(self.pub()))), 1)
389 self.assertEqual(self.count(""), 0)
390
391 def test_overlapping_run_is_a_noop(self):
392 self.deliver(mail())
393 self.deliver(mail())
394 real = comments.MaildirSource.finish
395 comments.MaildirSource.finish = lambda *a: None # messages stay: as if another run held them
396 try:
397 self.ingest()
398 finally:
399 comments.MaildirSource.finish = real
400 self.assertEqual(len(self.pending()), 2)
401 self.ingest()
402 self.assertEqual((len(self.pending()), self.count("")), (2, 0))
403
404 def test_rejected_mail_moved_back_is_retried(self):
405 self.set_cfg(authserv_id="other")
406 self.deliver(mail())
407 self.ingest()
408 self.assertEqual((len(self.pending()), self.count(".bad")), (0, 1))
409 self.set_cfg(authserv_id="mx")
410 bad = mailbox.Maildir(self.maildir + ".bad")
411 for k in bad.keys():
412 self.deliver(bad.get_bytes(k))
413 bad.remove(k)
414 self.ingest()
415 self.assertEqual((len(self.pending()), self.count("")), (1, 0))
416
417 def test_sender_chosen_message_id_cannot_block_other_mail(self):
418 self.deliver(mail(body="one\n", extra_headers=("X-Test: 1",)))
419 self.ingest()
420 self.deliver(mail(body="two\n"))
421 self.ingest()
422 self.assertEqual(len(self.pending()), 2)
423
424 def test_keys_pruned_when_comment_removed(self):
425 comments.write_json(os.path.join(self.state, "trusted.json"), [comments.sender_hash("ann@example.org")])
426 self.deliver(mail())
427 self.ingest()
428 cid = json.load(open(self.pub()))[0]["id"]
429 self.assertEqual(len(json.load(open(os.path.join(self.state, "approved-keys.json")))), 1)
430 self.run_cli("remove", cid)
431 self.deliver(mail(body="x\n"))
432 self.ingest()
433 self.assertEqual(len(json.load(open(os.path.join(self.state, "approved-keys.json")))), 1)
434
435 def test_publish_failure_for_trusted_sender_stores_once(self):
436 comments.write_json(os.path.join(self.state, "trusted.json"), [comments.sender_hash("ann@example.org")])
437 self.deliver(mail())
438 real = comments.publish_path
439 def boom(*a):
440 raise OSError("disk full")
441 comments.publish_path = boom
442 try:
443 for _ in range(3):
444 self.assertEqual(self.ingest()[0], 0)
445 finally:
446 comments.publish_path = real
447 self.assertEqual(len(json.load(open(os.path.join(self.state, "approved", "writing.json")))[POST]), 1)
448 self.assertEqual(self.count(""), 0)
449 self.run_cli("publish")
450 self.assertEqual(len(json.load(open(self.pub()))), 1)
451
452 def test_state_error_leaves_mail_in_inbox(self):
453 os.makedirs(os.path.join(self.state, "secret")) # unreadable as a file
454 self.deliver(mail())
455 code, _, err = self.ingest()
456 self.assertEqual(code, 1)
457 self.assertEqual((self.count(""), self.count(".bad"), self.count(".done")), (1, 0, 0))
458
459 def test_quiet_when_no_maildir_even_without_authserv_id(self):
460 self.set_cfg(authserv_id="")
461 self.assertEqual(self.ingest(), (0, "", ""))
462
463 def test_rejected_log_rotates(self):
464 os.makedirs(self.state)
465 log = os.path.join(self.state, "rejected.log")
466 open(log, "w").write("x" * 1_000_001)
467 comments.log_rejection("id", "why")
468 self.assertTrue(os.path.exists(log + ".1"))
469 self.assertEqual(len(open(log).read().splitlines()), 1)
470
471 def test_percent_encoded_slug_maps_to_decoded_folder(self):
472 folder = os.path.join(self.root, "2024", "05", "01", "café")
473 os.makedirs(folder)
474 open(os.path.join(folder, "index.html"), "w").close()
475 self.deliver(mail(subject="[writing:/2024/05/01/caf%c3%a9/]"))
476 self.deliver(mail(subject="[writing:/2024/05/01/caf%e9/]"))
477 self.ingest()
478 self.assertEqual((len(self.pending()), self.count(".bad")), (1, 1))
479 cid = self.pending()[0]
480 self.assertEqual(json.load(open(comments.pending_path(cid)))["path"], "/2024/05/01/caf%C3%A9/")
481 self.assertIn("invalid post path", open(os.path.join(self.state, "rejected.log")).read())
482 self.run_cli("approve", cid)
483 self.assertEqual(len(json.load(open(self.pub("2024/05/01/café.json")))), 1)
484 self.assertEqual(os.listdir(os.path.join(self.root, "_comments", "2024", "05", "01")), ["café.json"])
485
486 def test_encoded_traversal_rejected(self):
487 self.deliver(mail(subject="[writing:/%2e%2e/%2E%2e/etc/]"))
488 self.deliver(mail(subject="[writing:/a%2f..%2fb/]"))
489 self.deliver(mail(subject="[writing:/a%5cb/]"))
490 self.ingest()
491 self.assertEqual((len(self.pending()), self.count(".bad")), (0, 3))
492
493 def test_sender_limit(self):
494 for i in range(comments.SENDER_LIMIT + 2):
495 self.deliver(mail(body=f"c{i}\n"))
496 self.ingest()
497 self.assertEqual((len(self.pending()), self.count(".bad")), (comments.SENDER_LIMIT, 2))
498 self.assertIn("sender limit", open(os.path.join(self.state, "rejected.log")).read())
499
500 def test_sender_limit_counts_trusted_and_approved(self):
501 comments.write_json(os.path.join(self.state, "trusted.json"), [comments.sender_hash("ann@example.org")])
502 for i in range(comments.SENDER_LIMIT + 1):
503 self.deliver(mail(body=f"c{i}\n"))
504 self.ingest()
505 self.assertEqual(len(json.load(open(self.pub()))), comments.SENDER_LIMIT)
506 self.assertEqual(self.count(".bad"), 1)
507
508 def test_format_characters_stripped(self):
509 self.deliver(mail(sender="Ann\u202e <ann@example.org>", body="a\u200bb\u2028c\u2066d\n"))
510 self.ingest()
511 c = json.load(open(comments.pending_path(self.pending()[0])))
512 self.assertEqual((c["text"], c["author"]), ("abcd", "Ann"))
513
514 def test_sender_hash_is_salted(self):
515 h = comments.sender_hash("ann@example.org")
516 import hashlib
517 self.assertNotEqual(h, hashlib.sha256(b"ann@example.org").hexdigest())
518 self.assertEqual(h, comments.sender_hash(" ANN@example.org"))
519 self.assertEqual(os.stat(os.path.join(self.state, "secret")).st_mode & 0o777, 0o600)
520
521 def test_unreadable_state_file_reported(self):
522 os.makedirs(self.state)
523 open(os.path.join(self.state, "trusted.json"), "w").close()
524 self.deliver(mail())
525 code, _, err = self.ingest()
526 self.assertEqual(code, 1)
527 self.assertIn("trusted.json", err)
528
529 def test_missing_maildir_is_fine(self):
530 self.assertEqual(self.ingest()[0], 0)
531
532 def test_approve_after_ingest_then_trust_flow(self):
533 self.deliver(mail())
534 self.ingest()
535 cid = self.pending()[0]
536 self.run_cli("approve", cid, "--trust")
537 self.deliver(mail(body="Second\n"))
538 self.ingest()
539 self.assertEqual(self.pending(), [])
540 self.assertEqual(len(json.load(open(self.pub()))), 2)
541 self.run_cli("untrust", cid)
542 self.deliver(mail(body="Third\n"))
543 self.ingest()
544 self.assertEqual(len(self.pending()), 1)
545
546
547PASSWORD = "s3cret-pw-xyz"
548
549
550class FakeServer:
551 """One IMAP account shared by all connections: folder name -> list of [uid, raw, deleted]."""
552
553 def __init__(self):
554 self.folders = {"INBOX": []}
555 self.next_uid = 1
556 self.instances = []
557 self.body_fetches = 0
558 self.ssl_context = None
559 self.fail_login = False
560 self.prefix = "" # namespace prefix the server demands for new folders, e.g. "INBOX."
561 self.namespace = True # answer NAMESPACE (else only LIST shows the delimiter)
562 self.uidplus = False
563 self.fail = {} # operation -> number of times it still fails: copy, store, expunge, create
564 self.refuse_fetch = set() # UIDs whose body fetch answers NO
565 self.expunged_by_uid = 0
566 self.literal_folder = False
567
568 def add(self, raw, folder="INBOX"):
569 self.folders.setdefault(folder, []).append([self.next_uid, raw, False])
570 self.next_uid += 1
571
572 def raws(self, folder):
573 return [m[1] for m in self.folders.get(folder, [])]
574
575
576class FakeIMAP:
577 def __init__(self, server, host, port, ssl_context=None, timeout=None):
578 self.s, self.host, self.port, self.timeout = server, host, port, timeout
579 server.ssl_context = ssl_context
580 self.logged_out = False
581 self.folder = None
582 server.instances.append(self)
583
584 @property
585 def capabilities(self):
586 return ("IMAP4REV1",) # as imaplib: fixed at connect time, never refreshed after login
587
588 def capability(self):
589 caps = b"IMAP4REV1 UIDPLUS" if self.s.uidplus else b"IMAP4REV1"
590 return "OK", [caps]
591
592 def _fails(self, op):
593 if self.s.fail.get(op, 0) > 0:
594 self.s.fail[op] -= 1
595 return True
596 return False
597
598 def list(self, directory='""', pattern="*"):
599 out = [f'(\\HasNoChildren) "." "{name}"'.encode() for name in self.s.folders]
600 if self.s.literal_folder:
601 out.append((b'(\\HasNoChildren) "." {11}', b"INBOX.Weird\xc3\xa9"))
602 return "OK", out
603
604 def namespace(self):
605 if not self.s.namespace:
606 return "NO", [b"not supported"]
607 return "OK", [f'(("{self.s.prefix}" ".")) NIL NIL'.encode()]
608
609 def login(self, user, password):
610 if self.s.fail_login:
611 raise comments.imaplib.IMAP4.error(f"login failed for {user} {password}")
612 self.user = user
613
614 def select(self, folder, readonly=False):
615 self.folder = folder
616 return "OK", [b"1"]
617
618 def response(self, name):
619 return name, [b"777"]
620
621 def create(self, name):
622 if self._fails("create") or (self.s.prefix and not name.startswith(self.s.prefix)):
623 return "NO", [b"[CANNOT] namespace prefix required"]
624 self.s.folders.setdefault(name, [])
625 return "OK", [b""]
626
627 def logout(self):
628 self.logged_out = True
629 return "BYE", [b""]
630
631 def expunge(self):
632 if self._fails("expunge"):
633 return "NO", [b"expunge failed"]
634 box = self.s.folders[self.folder]
635 box[:] = [m for m in box if not m[2]]
636 return "OK", [b""]
637
638 def uid(self, cmd, *args):
639 box = self.s.folders[self.folder]
640 if cmd == "SEARCH":
641 assert args[1] == "UNDELETED"
642 return "OK", [" ".join(str(m[0]) for m in box if not m[2]).encode()]
643 if cmd == "FETCH":
644 wanted = {int(u) for u in args[0].split(",")}
645 found = [m for m in box if m[0] in wanted]
646 if args[1] == "(RFC822.SIZE)":
647 return "OK", [f"{i} (UID {m[0]} RFC822.SIZE {len(m[1])})".encode() for i, m in enumerate(found, 1)]
648 if not found:
649 return "OK", [None]
650 m = found[0]
651 if m[0] in self.s.refuse_fetch:
652 return "NO", [b"[SERVERBUG] cannot fetch"]
653 if args[1] == "(BODY.PEEK[HEADER])":
654 return "OK", [(b"1 (UID 1 BODY[HEADER] {1}", m[1].split(b"\n\n", 1)[0] + b"\n\n"), b")"]
655 self.s.body_fetches += 1
656 return "OK", [(b"1 (UID 1 BODY[] {1}", m[1]), b")"]
657 if cmd == "COPY":
658 if self._fails("copy"):
659 return "NO", [b"copy failed"]
660 if args[1] not in self.s.folders:
661 return "NO", [b"[TRYCREATE]"]
662 m = next(m for m in box if m[0] == int(args[0]))
663 self.s.add(m[1], args[1])
664 return "OK", [b""]
665 if cmd == "EXPUNGE":
666 assert self.s.uidplus
667 if self._fails("expunge"):
668 return "NO", [b"expunge failed"]
669 self.s.expunged_by_uid += 1
670 box[:] = [m for m in box if not (m[2] and m[0] == int(args[0]))]
671 return "OK", [b""]
672 if cmd == "STORE":
673 if self._fails("store"):
674 return "NO", [b"store failed"]
675 for m in box:
676 if m[0] == int(args[0]):
677 m[2] = True
678 return "OK", [b""]
679 raise AssertionError(cmd)
680
681
682class IMAPBase(Base):
683 def setUp(self):
684 super().setUp()
685 self.server = FakeServer()
686 self.cfgpath = os.path.join(self.tmp, "imap.json")
687 os.environ["FW_COMMENTS_IMAP"] = self.cfgpath
688 self.addCleanup(os.environ.pop, "FW_COMMENTS_IMAP", None)
689 self.write_cfg()
690 real = comments.imaplib.IMAP4_SSL
691 comments.imaplib.IMAP4_SSL = lambda host, port=993, **kw: FakeIMAP(self.server, host, port, **kw)
692 self.addCleanup(setattr, comments.imaplib, "IMAP4_SSL", real)
693
694 def write_cfg(self, mode=0o600, **override):
695 cfg = {"host": "imap.example.net", "user": "c@example.net", "password": PASSWORD, "folder": "INBOX"}
696 cfg.update(override)
697 cfg = {k: v for k, v in cfg.items() if v is not None}
698 with open(self.cfgpath, "w") as f:
699 json.dump(cfg, f)
700 os.chmod(self.cfgpath, mode)
701
702
703class IMAPIngest(IMAPBase):
704 def test_end_to_end(self):
705 self.server.add(mail())
706 self.server.add(mail(subject="no tag"))
707 code, out, err = self.run_cli("ingest")
708 self.assertEqual(code, 0, err)
709 self.assertEqual(len(self.pending()), 1)
710 self.assertEqual((len(self.server.raws("INBOX")), len(self.server.raws("comments-done")), len(self.server.raws("comments-bad"))), (0, 1, 1))
711 self.assertTrue(all(i.logged_out for i in self.server.instances))
712 self.assertEqual((self.server.instances[0].host, self.server.instances[0].port, self.server.instances[0].timeout),
713 ("imap.example.net", 993, 30))
714 self.assertEqual(self.server.ssl_context.verify_mode, comments.ssl.CERT_REQUIRED)
715 self.assertEqual(self.run_cli("ingest")[1], "") # second run: nothing to do
716
717 def test_insecure_file_mode_refused(self):
718 self.write_cfg(mode=0o644)
719 self.server.add(mail())
720 code, out, err = self.run_cli("ingest")
721 self.assertEqual(code, 1)
722 self.assertIn("readable", err)
723 self.assertNotIn(PASSWORD, out + err)
724 self.assertEqual(len(self.server.raws("INBOX")), 1)
725 self.assertEqual(self.server.instances, [])
726
727 def test_missing_keys_and_empty_password_refused(self):
728 for bad in ({"host": None}, {"user": None}, {"password": None}, {"password": ""}):
729 self.write_cfg(**bad)
730 code, out, err = self.run_cli("ingest")
731 self.assertEqual(code, 1, bad)
732 self.assertNotIn(PASSWORD, out + err)
733
734 def test_password_never_in_output_on_login_failure(self):
735 self.server.fail_login = True
736 self.server.add(mail())
737 code, out, err = self.run_cli("ingest")
738 self.assertEqual(code, 1)
739 self.assertNotIn(PASSWORD, out + err)
740 self.assertTrue(all(i.logged_out for i in self.server.instances))
741
742 def test_finish_copies_creates_folder_and_expunges(self):
743 self.server.add(mail())
744 src = comments.IMAPSource()
745 try:
746 key = src.keys()[0]
747 raw = src.read(key)
748 src.finish(key, raw, "done")
749 finally:
750 src.close()
751 self.assertEqual((self.server.raws("INBOX"), self.server.raws("comments-done")), ([], [raw]))
752 self.assertEqual(self.server.body_fetches, 1)
753
754 def test_read_of_vanished_message_is_keyerror(self):
755 self.server.add(mail())
756 src = comments.IMAPSource()
757 try:
758 key = src.keys()[0]
759 self.server.folders["INBOX"].clear()
760 with self.assertRaises(KeyError):
761 src.read(key)
762 with self.assertRaises(KeyError):
763 src.read("1:1") # other UIDVALIDITY
764 finally:
765 src.close()
766
767 def test_oversize_skipped_without_download(self):
768 self.server.add(mail(body="x" * (300 * 1024)))
769 self.server.add(mail())
770 self.run_cli("ingest")
771 self.assertEqual(self.server.body_fetches, 1)
772 self.assertEqual((len(self.pending()), len(self.server.raws("comments-bad")), len(self.server.raws("INBOX"))), (1, 1, 0))
773 self.assertIn("over 256 KB", open(os.path.join(self.state, "rejected.log")).read())
774
775 def test_missing_config_is_quiet(self):
776 os.unlink(self.cfgpath)
777 self.assertEqual(self.run_cli("ingest"), (0, "", ""))
778 self.assertEqual(self.server.instances, [])
779
780 def test_imap_check_prints_only_names_and_tokens(self):
781 self.server.add(mail(sender="Ann Secret <ann@example.org>", subject="Private subject [writing:/x/]",
782 auth=("mx (comment); dkim=pass header.d=example.org; dmarc=pass header.from=example.org",)))
783 code, out, err = self.run_cli("imap-check")
784 self.assertEqual(code, 0, err)
785 self.assertIn("messages in INBOX: 1", out)
786 self.assertIn("Authentication-Results: authserv-id mx; dkim=pass dmarc=pass", out)
787 for secret in ("Secret", "ann@", "Private", "example.org", PASSWORD, "Nice post"):
788 self.assertNotIn(secret, out + err)
789 self.assertTrue(self.server.instances[0].logged_out)
790
791
792class IMAPRobustness(IMAPBase):
793 def run_twice(self):
794 self.run_cli("ingest")
795 return self.run_cli("ingest")
796
797 def test_partial_failure_leaves_mail_and_next_run_finishes(self):
798 for op in ("copy", "store", "expunge"):
799 with self.subTest(op=op):
800 self.setUp()
801 self.server.add(mail())
802 self.server.fail[op] = 5 # the first move attempts fail
803 code, out, err = self.run_cli("ingest")
804 self.assertEqual(code, 1, op)
805 self.assertNotIn(PASSWORD, out + err)
806 self.assertEqual(len(self.pending()), 1) # stored once
807 self.server.fail[op] = 0
808 self.run_cli("ingest")
809 self.assertEqual(len(self.pending()), 1)
810 inbox_live = [m for m in self.server.folders["INBOX"] if not m[2]]
811 self.assertEqual(inbox_live, [])
812 self.assertEqual(len(self.server.raws("comments-done")) >= 1, True)
813
814 def test_namespace_prefix_used_for_folders(self):
815 self.server.prefix = "INBOX."
816 self.server.folders = {"INBOX": [], "INBOX.Sent": []}
817 self.server.add(mail())
818 self.server.add(mail(subject="no tag"))
819 code, out, err = self.run_cli("ingest")
820 self.assertEqual(code, 0, err)
821 self.assertEqual((len(self.server.raws("INBOX.comments-done")), len(self.server.raws("INBOX.comments-bad"))), (1, 1))
822 self.assertNotIn("comments-done", self.server.folders)
823
824 def test_namespace_from_list_when_namespace_command_missing(self):
825 self.server.prefix = "INBOX."
826 self.server.namespace = False
827 self.server.folders = {"INBOX": [], "INBOX.Sent": []}
828 self.server.add(mail())
829 self.assertEqual(self.run_cli("ingest")[0], 0)
830 self.assertEqual(len(self.server.raws("INBOX.comments-done")), 1)
831
832 def test_create_failure_reported_and_mail_kept(self):
833 self.server.prefix = "INBOX."
834 self.server.namespace = False
835 self.server.folders = {"INBOX": []} # no hint of the prefix anywhere: the wrong guess fails
836 self.server.add(mail())
837 code, out, err = self.run_cli("ingest")
838 self.assertEqual(code, 1)
839 self.assertIn("create folder comments-done", err)
840 self.assertIn("namespace prefix required", err)
841 self.assertNotIn(PASSWORD, out + err)
842 self.assertEqual(len(self.server.raws("INBOX")), 1)
843
844 def test_folder_overrides(self):
845 self.write_cfg(done_folder="INBOX.mine-done", bad_folder="INBOX.mine-bad")
846 self.server.add(mail())
847 self.server.folders["INBOX.mine-done"] = []
848 self.run_cli("ingest")
849 self.assertEqual(len(self.server.raws("INBOX.mine-done")), 1)
850 self.write_cfg(done_folder="bad folder")
851 self.assertEqual(self.run_cli("ingest")[0], 1)
852
853 def test_deleted_flagged_message_skipped(self):
854 self.server.add(mail())
855 self.server.folders["INBOX"][0][2] = True
856 self.assertEqual(self.run_cli("ingest")[0], 0)
857 self.assertEqual(self.pending(), [])
858
859 def test_uid_expunge_with_uidplus(self):
860 self.server.uidplus = True
861 self.server.add(mail())
862 self.run_cli("ingest")
863 self.assertEqual((self.server.expunged_by_uid, self.server.raws("INBOX")), (1, []))
864
865 def test_refused_fetch_does_not_block_others(self):
866 self.server.add(mail(body="first\n"))
867 self.server.add(mail(body="second\n"))
868 self.server.refuse_fetch = {1}
869 code, out, err = self.run_cli("ingest")
870 self.assertEqual(code, 0, err)
871 self.assertEqual(len(self.pending()), 1)
872 self.assertEqual(len(self.server.raws("INBOX")), 1) # the refused one stays
873 self.assertIn("skipped", open(os.path.join(self.state, "rejected.log")).read())
874
875 def test_non_ascii_credentials_refused_without_echo(self):
876 self.write_cfg(password="pässword-ünique")
877 code, out, err = self.run_cli("ingest")
878 self.assertEqual(code, 1)
879 self.assertIn("ASCII", err)
880 self.assertNotIn("pässword", out + err)
881
882 def test_imap_check_lists_folders_and_namespace(self):
883 self.server.prefix = "INBOX."
884 self.server.folders = {"INBOX": [], "INBOX.Sent": []}
885 self.server.add(mail())
886 code, out, err = self.run_cli("imap-check")
887 self.assertIn("folders: INBOX, INBOX.Sent", out)
888 self.assertIn('namespace: (("INBOX." "."))', out)
889
890 def test_list_with_literal_folder_name(self):
891 self.server.literal_folder = True
892 self.server.namespace = False
893 self.server.prefix = "INBOX."
894 self.server.folders = {"INBOX": [], "INBOX.Sent": []}
895 self.server.add(mail())
896 code, out, err = self.run_cli("imap-check")
897 self.assertEqual(code, 0, err)
898 self.assertIn("INBOX.Weird", out)
899 self.run_cli("ingest")
900 self.assertEqual(len(self.server.raws("INBOX.comments-done")), 1)
901
902 def test_unparseable_namespace_falls_back_to_empty_prefix(self):
903 self.server.add(mail())
904 real = FakeIMAP.namespace
905 FakeIMAP.namespace = lambda self: ("OK", [(b"weird", b"tuple")])
906 self.addCleanup(setattr, FakeIMAP, "namespace", real)
907 self.assertEqual(self.run_cli("ingest")[0], 0)
908 self.assertEqual(len(self.server.raws("comments-done")), 1)
909
910 def test_stale_snapshot_cannot_double_store(self):
911 import fcntl
912 import threading
913 raw = mail()
914 self.server.add(raw)
915 os.makedirs(self.state, exist_ok=True)
916 lock = open(os.path.join(self.state, "ingest.lock"), "w")
917 fcntl.flock(lock, fcntl.LOCK_EX) # "another run" is in progress
918 t = threading.Thread(target=self.run_cli, args=("ingest",))
919 t.start()
920 t.join(0.5)
921 self.assertTrue(t.is_alive()) # the second run waits for the job lock
922 # the other run finishes: stores the comment (its record carries the message key) but leaves the mail
923 import hashlib
924 self.add_pending(msg_key=hashlib.sha256(raw).hexdigest())
925 fcntl.flock(lock, fcntl.LOCK_UN)
926 t.join(5)
927 self.assertFalse(t.is_alive())
928 self.assertEqual(len(self.pending()), 1) # not stored twice
929 self.assertEqual(len(self.server.raws("comments-done")), 1)
930
931
932class MalformedSubjects(Base):
933 def test_bad_subjects_go_to_bad_and_do_not_abort_the_run(self):
934 subjects = ["[writing:/a" + "%" * 290 + "/]", "[writing:/" + "a/" * 200 + "]", "[writing:/%/]", "[writing:/%zz/]",
935 "[writing:/%ff%fe/]", "[writing:/a%00b/]", "[writing:/\u00e9/]", "[nosuchsite:/x/]", "[writing:]",
936 "[writing:/" + "%25" * 120 + "/]"]
937 for sub in subjects:
938 self.deliver(mail(subject=sub))
939 self.deliver(mail(subject="Good [writing:/2024/05/01/hello/]"))
940 code, out, err = self.run_cli("ingest")
941 self.assertEqual(code, 0, err)
942 self.assertEqual((len(self.pending()), self.count(".bad"), self.count("")), (1, len(subjects), 0))
943
944
945class Canonical(Base):
946 def test_equivalent_spellings_share_one_key(self):
947 cases = [("/sl%75g/", "/slug/"), ("/caf%c3%a9/", "/caf%C3%A9/"), ("/a%2db/", "/a-b/"), ("/x%7e/", "/x~/")]
948 for a, b in cases:
949 self.assertEqual(comments.canonical_path(a), comments.canonical_path(b), (a, b))
950 self.assertEqual(comments.canonical_path("/caf%c3%a9/"), "/caf%C3%A9/")
951
952 def test_ingest_collapses_spellings_and_publishes_one_file(self):
953 for d in ("slug", "café"):
954 os.makedirs(os.path.join(self.root, d))
955 open(os.path.join(self.root, d, "index.html"), "w").close()
956 for subject in ("[writing:/sl%75g/]", "[writing:/slug/]", "[writing:/caf%c3%a9/]", "[writing:/caf%C3%A9/]"):
957 self.deliver(mail(subject=subject, body=subject + "\n"))
958 comments.write_json(os.path.join(self.state, "trusted.json"), [comments.sender_hash("ann@example.org")])
959 self.run_cli("ingest")
960 approved = json.load(open(os.path.join(self.state, "approved", "writing.json")))
961 self.assertEqual(sorted(approved), ["/caf%C3%A9/", "/slug/"])
962 self.assertEqual(len(approved["/slug/"]), 2)
963 self.assertEqual(len(json.load(open(self.pub("slug.json")))), 2)
964 self.assertEqual(len(json.load(open(self.pub("café.json")))), 2)
965
966 def test_existing_duplicate_spellings_are_merged_on_publish_and_remove(self):
967 mk = lambda cid, path, date: {"id": cid, "date": date, "author": "A", "parent": None, "text": cid}
968 approved = {"/slug/": [mk("20240101T000000Z-aaaaaa", "", "2024-01-01T00:00:00Z")],
969 "/sl%75g/": [mk("20240102T000000Z-bbbbbb", "", "2024-01-02T00:00:00Z")]}
970 comments.write_json(os.path.join(self.state, "approved", "writing.json"), approved)
971 self.run_cli("publish", "writing")
972 self.assertEqual([c["id"][-6:] for c in json.load(open(self.pub("slug.json")))], ["aaaaaa", "bbbbbb"])
973 self.run_cli("remove", "20240101T000000Z-aaaaaa")
974 self.assertEqual([c["id"][-6:] for c in json.load(open(self.pub("slug.json")))], ["bbbbbb"])
975 self.run_cli("remove", "20240102T000000Z-bbbbbb")
976 self.assertFalse(os.path.exists(self.pub("slug.json")))
977
978
979class MailChannelsHeaders(Base):
980 def ingest(self):
981 return self.run_cli("ingest")
982
983 def set_cfg(self, **kw):
984 cfg = json.load(open(os.environ["FW_COMMENTS_SITES"]))
985 cfg.update(kw)
986 json.dump(cfg, open(os.environ["FW_COMMENTS_SITES"], "w"))
987
988 REAL = "inbound.mailchannels.net; spf=pass dkim=pass header.d=example.org; dmarc=pass header.from=example.org"
989
990 def setUp(self):
991 super().setUp()
992 self.set_cfg(authserv_id="inbound.mailchannels.net")
993
994 def test_only_the_real_authentication_results_counts(self):
995 arc_above = ("ARC-Seal: i=1; a=rsa-sha256; d=mailchannels.net; s=arc; cv=none; b=abc",
996 "ARC-Authentication-Results: i=1; inbound.mailchannels.net; dmarc=pass header.from=example.org")
997 # real header on top, ARC headers around it: accepted
998 self.deliver(mail(auth=(self.REAL,), extra_headers=arc_above))
999 # a forged "ARC-Authentication-Results: <our id>; dmarc=pass" with a failing real header: rejected
1000 self.deliver(mail(auth=("inbound.mailchannels.net; dkim=fail; dmarc=fail",), extra_headers=arc_above))
1001 # no real header at all, only the ARC copy carrying our id: rejected
1002 raw = mail(auth=()).replace(b"From:", b"ARC-Authentication-Results: i=1; inbound.mailchannels.net; dmarc=pass header.from=example.org\nFrom:", 1)
1003 self.deliver(raw)
1004 # real header below a forged foreign one: rejected (topmost is not ours)
1005 self.deliver(mail(auth=("evil.test; dmarc=pass header.from=example.org", self.REAL)))
1006 self.ingest()
1007 self.assertEqual((len(self.pending()), self.count(".bad")), (1, 3))
1008
1009
1010if __name__ == "__main__":
1011 unittest.main()