fuzzyworld

fuzzyworld-tools / files / tests / test_comments_php.py

test_comments_php.py

Blame: the commit that last changed each line · file · history

e9fdc5db 2026-11-01 Pavel Safronov1"""Tests for server/comments/post.php (the off-by-default comment form endpoint).
2
3Skipped unless `php` is installed. Each test starts `php -S 127.0.0.1:<port>` with a temporary
4state directory, sites.json and site root, and talks to it with urllib.
5
6Run: python3 -m unittest tests.test_comments_php (or python3 -m unittest discover tests)
7"""
8
9import hashlib
10import hmac
11import json
12import os
13import shutil
14import socket
15import subprocess
16import sys
17import tempfile
18import time
19import unittest
20import urllib.error
21import urllib.parse
22import urllib.request
23
24ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
25POST_PHP = os.path.join(ROOT, "server", "comments", "post.php")
26COMMENTS_PY = os.path.join(ROOT, "server", "comments.py")
27POST_PATH = "/2024/05/01/slug/"
28ORIGIN = "https://writing.example"
29SECRET = "test-secret-0123456789"
30
31
32class NoRedirect(urllib.request.HTTPRedirectHandler):
33 def redirect_request(self, *args, **kwargs):
34 return None
35
36
37def free_port():
38 with socket.socket() as s:
39 s.bind(("127.0.0.1", 0))
40 return s.getsockname()[1]
41
42
43@unittest.skipUnless(shutil.which("php"), "php is not installed")
44class PostPhpTest(unittest.TestCase):
45 def setUp(self):
46 self.tmp = tempfile.mkdtemp()
47 self.state = os.path.join(self.tmp, "state")
48 self.docroot = os.path.join(self.tmp, "web")
49 self.siteroot = os.path.join(self.tmp, "writing")
50 os.makedirs(self.state)
51 os.makedirs(self.docroot)
52 os.makedirs(self.siteroot + POST_PATH)
53 with open(self.siteroot + POST_PATH + "index.html", "w") as f:
54 f.write("<html></html>")
55 with open(os.path.join(self.state, "secret"), "w") as f:
56 f.write(SECRET + "\n")
57 self.sites = os.path.join(self.tmp, "sites.json")
58 with open(self.sites, "w") as f:
59 json.dump({"writing": {"origin": ORIGIN, "root": self.siteroot},
60 "mailbox": "", "authserv_id": ""}, f)
61 shutil.copy(POST_PHP, self.docroot)
62 self.port = free_port()
63 self.self_origin = f"http://127.0.0.1:{self.port}"
64 self.env = dict(os.environ, FW_COMMENTS_DIR=self.state, FW_COMMENTS_SITES=self.sites,
65 FW_COMMENTS_SELF_ORIGIN=self.self_origin)
66 self.server = subprocess.Popen(
67 ["php", "-S", f"127.0.0.1:{self.port}", "-t", self.docroot],
68 env=self.env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
69 for _ in range(100):
70 try:
71 socket.create_connection(("127.0.0.1", self.port), timeout=0.2).close()
72 break
73 except OSError:
74 time.sleep(0.05)
75 self.opener = urllib.request.build_opener(NoRedirect)
76
77 def tearDown(self):
78 self.server.terminate()
79 self.server.wait()
80 shutil.rmtree(self.tmp, ignore_errors=True)
81
82 # ---- helpers ----
83
84 def request(self, method="POST", fields=None, headers=None):
85 url = f"http://127.0.0.1:{self.port}/post.php"
86 data = urllib.parse.urlencode(fields).encode() if fields is not None else None
87 req = urllib.request.Request(url, data=data, method=method, headers=headers or {})
88 try:
89 r = self.opener.open(req)
90 except urllib.error.HTTPError as e:
91 r = e
92 with r:
93 return r.status, r.headers, r.read().decode("utf-8", "replace")
94
95 def post(self, headers=None, **over):
96 fields = {"site": "writing", "path": POST_PATH, "elapsed": "5000", "website": "",
97 "name": "Ann", "text": "Nice post."}
98 fields.update(over)
99 fields = {k: v for k, v in fields.items() if v is not None}
100 return self.request("POST", fields, headers)
101
102 def pending(self):
103 d = os.path.join(self.state, "pending")
104 return sorted(os.path.join(d, n) for n in os.listdir(d)) if os.path.isdir(d) else []
105
106 def load(self, path):
107 with open(path, encoding="utf-8") as f:
108 return json.load(f)
109
110 def sign(self, ts, site, path, name, text):
111 msg = f"{ts}|{site}|{path}|{hashlib.sha256(name.encode()).hexdigest()}|{hashlib.sha256(text.encode()).hexdigest()}"
112 return hmac.new(SECRET.encode(), msg.encode(), hashlib.sha256).hexdigest()
113
114 def assert_ok(self, resp):
115 status, headers, _ = resp
116 self.assertEqual(status, 303)
117 self.assertEqual(headers["Location"], ORIGIN + POST_PATH + "#comment-pending")
118
119 def assert_refused(self, resp, code=None):
120 status, _, _ = resp
121 self.assertTrue(status >= 400 if code is None else status == code, status)
122 self.assertEqual(self.pending(), [])
123
124 # ---- tests ----
125
126 def test_get_refused(self):
127 status, headers, body = self.request("GET")
128 self.assertEqual(status, 405)
129 self.assertEqual(headers["Allow"], "POST")
130 self.assertEqual(self.pending(), [])
131
132 def test_security_headers(self):
133 _, headers, _ = self.request("GET")
134 self.assertEqual(headers["X-Content-Type-Options"], "nosniff")
135 self.assertEqual(headers["Cache-Control"], "no-store")
136 self.assertIn("default-src 'none'", headers["Content-Security-Policy"])
137 self.assertIn("text/html; charset=utf-8", headers["Content-Type"])
138 self.assertIsNone(headers["Set-Cookie"])
139
140 def test_bad_origin(self):
141 self.assert_refused(self.post(headers={"Origin": "https://evil.example"}), 403)
142 self.assert_refused(self.post(headers={"Origin": "null"}), 403)
143 self.assert_refused(self.post(headers={"Origin": ORIGIN + ".evil.example"}), 403)
144
145 def test_good_origin_and_none(self):
146 self.assert_ok(self.post(headers={"Origin": ORIGIN}))
147 self.assert_ok(self.post())
148 self.assertEqual(len(self.pending()), 2)
149
150 def test_unknown_site(self):
151 self.assert_refused(self.post(site="nope"))
152 self.assert_refused(self.post(site="mailbox"))
153 self.assert_refused(self.post(site="authserv_id"))
154 self.assert_refused(self.post(site=""))
155
156 def test_traversal_paths(self):
157 for bad in ["/../", "/2024/../", "/%2e%2e/x/", "/2024/%2E%2E/", "/%2e/x/", "//", "/2024//05/",
158 "2024/05/01/slug/", "/2024/05/01/slug", "/2024\\05/01/slug/", "/2024/a%2Fb/",
159 "/2024/a%5Cb/", "/2024/a%00b/", "/2024/05/01/slug/\n", "/" + "a/" * 9, "/.hidden/", "",
160 "/etc/passwd/"]:
161 with self.subTest(path=bad):
162 self.assert_refused(self.post(path=bad))
163
164 def test_nonexistent_post(self):
165 self.assert_refused(self.post(path="/2024/05/01/missing/"))
166
167 def test_missing_site_root(self):
168 shutil.rmtree(self.siteroot)
169 self.assert_refused(self.post())
170
171 def test_honeypot_stores_nothing(self):
172 self.assert_ok(self.post(website="http://spam.example"))
173 self.assertEqual(self.pending(), [])
174
175 def test_too_fast_gives_confirm_page(self):
176 for elapsed in ("1000", "0", "", "abc", "2999", None):
177 with self.subTest(elapsed=elapsed):
178 status, _, body = self.post(elapsed=elapsed)
179 self.assertEqual(status, 200)
180 self.assertIn('name="sig"', body)
181 self.assertEqual(self.pending(), [])
182
183 def test_confirm_flow_stores_one_comment(self):
184 ts = str(int(time.time()) - 10)
185 sig = self.sign(ts, "writing", POST_PATH, "Ann", "Nice post.")
186 self.assert_ok(self.request("POST", {"site": "writing", "path": POST_PATH, "name": "Ann",
187 "text": "Nice post.", "ts": ts, "sig": sig}))
188 files = self.pending()
189 self.assertEqual(len(files), 1)
190 c = self.load(files[0])
191 self.assertRegex(c["id"], r"^\d{8}T\d{6}Z-[0-9a-f]{6}$")
192 self.assertEqual(os.path.basename(files[0]), c["id"] + ".json")
193 self.assertEqual({k: v for k, v in c.items() if k not in ("id", "date")},
194 {"site": "writing", "path": POST_PATH, "author": "Ann", "text": "Nice post.",
195 "source": "form", "parent": None, "sender_hash": None})
196 self.assertRegex(c["date"], r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ$")
197 self.assertEqual(os.stat(files[0]).st_mode & 0o777, 0o600)
198 out = subprocess.run([sys.executable, COMMENTS_PY, "list"], env=self.env, capture_output=True, text=True)
199 self.assertEqual(out.returncode, 0, out.stderr)
200 self.assertIn(c["id"], out.stdout)
201 self.assertIn("[form]", out.stdout)
202 self.assertIn("Nice post.", out.stdout)
203 # form comments cannot be trusted, but can be approved
204 trust = subprocess.run([sys.executable, COMMENTS_PY, "approve", c["id"], "--trust"], env=self.env,
205 capture_output=True, text=True)
206 self.assertNotEqual(trust.returncode, 0)
207 ok = subprocess.run([sys.executable, COMMENTS_PY, "approve", c["id"]], env=self.env,
208 capture_output=True, text=True)
209 self.assertEqual(ok.returncode, 0, ok.stderr)
210
211 def test_confirm_page_then_confirm_roundtrip(self):
212 status, _, body = self.post(elapsed="10")
213 self.assertEqual(status, 200)
214 fields = dict(urllib.parse.parse_qsl(
215 "&".join(f"{n}={urllib.parse.quote(v)}" for n, v in
216 __import__("re").findall(r'name="(\w+)" value="([^"]*)"', body))))
217 fields = {k: __import__("html").unescape(v) for k, v in fields.items()}
218 time.sleep(3.2)
219 self.assert_ok(self.request("POST", fields, {"Origin": self.self_origin}))
220 self.assertEqual(len(self.pending()), 1)
221
222 def test_bad_and_expired_sig(self):
223 now = int(time.time())
224 base = {"site": "writing", "path": POST_PATH, "name": "Ann", "text": "Nice post."}
225 good = self.sign(str(now - 10), "writing", POST_PATH, "Ann", "Nice post.")
226 self.assert_refused(self.request("POST", dict(base, ts=str(now - 10), sig="0" * 64)))
227 self.assert_refused(self.request("POST", dict(base, ts=str(now - 10), sig=good, text="Other")))
228 self.assert_refused(self.request("POST", dict(base, ts=str(now - 11), sig=good)))
229 for ts in (now - 4000, now, now + 100):
230 sig = self.sign(str(ts), "writing", POST_PATH, "Ann", "Nice post.")
231 self.assert_refused(self.request("POST", dict(base, ts=str(ts), sig=sig)))
232 self.assert_refused(self.request("POST", dict(base, ts="", sig="x")))
233
234 def test_length_caps(self):
235 self.assert_ok(self.post(name="n" * 60, text="t" * 4000))
236 self.assertEqual(len(self.pending()), 1)
237 for p in self.pending():
238 os.unlink(p)
239 self.assert_refused(self.post(name="n" * 61))
240 self.assert_refused(self.post(text="t" * 4001))
241 self.assert_refused(self.post(name=""))
242 self.assert_refused(self.post(text=""))
243 self.assert_refused(self.post(text=" \n "))
244 self.assert_refused(self.post(name="\u200b\u202e"))
245 self.assert_ok(self.post(name="\u00e9" * 60)) # characters, not bytes
246
247 def test_invalid_utf8(self):
248 body = b"site=writing&path=%2F2024%2F05%2F01%2Fslug%2F&elapsed=5000&website=&name=Ann&text=bad%ff%fe"
249 req = urllib.request.Request(f"http://127.0.0.1:{self.port}/post.php", data=body, method="POST")
250 with self.assertRaises(urllib.error.HTTPError) as cm:
251 self.opener.open(req)
252 self.assertEqual(cm.exception.code, 400)
253 cm.exception.close()
254 self.assertEqual(self.pending(), [])
255
256 def test_array_fields_refused(self):
257 body = b"site=writing&path=%2F2024%2F05%2F01%2Fslug%2F&elapsed=5000&name%5B%5D=Ann&text=hi"
258 req = urllib.request.Request(f"http://127.0.0.1:{self.port}/post.php", data=body, method="POST")
259 with self.assertRaises(urllib.error.HTTPError) as cm:
260 self.opener.open(req)
261 cm.exception.close()
262 self.assertEqual(self.pending(), [])
263
264 def test_at_sign_name(self):
265 self.assert_ok(self.post(name="me@example.com"))
266 self.assertEqual(self.load(self.pending()[0])["author"], "A reader")
267
268 def test_control_and_bidi_stripped(self):
269 self.assert_ok(self.post(name="Bo\u202eb\x00\n x", text=" a\u202eb\u200bc\x00d\r\ne\u2028f\tg\x7f "))
270 c = self.load(self.pending()[0])
271 self.assertEqual(c["author"], "Bob x") # newline becomes a space, then the input's own space
272 self.assertEqual(c["text"], "abcd\nefg")
273
274 def test_hostile_html_stored_raw_never_echoed(self):
275 name = "<script>alert(1)</script>"
276 text = "<img src=x onerror=alert(1)>\"'& </textarea>"
277 status, _, body = self.post(elapsed="1", name=name, text=text)
278 self.assertEqual(status, 200)
279 self.assertNotIn("<script>", body)
280 self.assertNotIn("<img", body)
281 self.assertNotIn("</textarea>", body)
282 self.assertIn("&lt;script&gt;", body)
283 self.assertIn("&lt;img src=x", body)
284 for other in (self.post(path="/nope/", name=name, text=text),
285 self.post(site="<b>x</b>", name=name, text=text)):
286 self.assertNotIn("<b>", other[2])
287 self.assertNotIn("<script>", other[2])
288 self.assert_ok(self.post(name=name, text=text))
289 c = self.load(self.pending()[0])
290 self.assertEqual((c["author"], c["text"]), (name, text))
291
292 def test_rate_limit(self):
293 for _ in range(5):
294 self.assert_ok(self.post())
295 status, _, _ = self.post()
296 self.assertEqual(status, 429)
297 self.assertEqual(len(self.pending()), 5)
298 rate_dir = os.path.join(self.state, "rate")
299 self.assertEqual(len(os.listdir(rate_dir)), 1)
300 self.assertNotIn("127.0.0.1", os.listdir(rate_dir)[0])
301
302 def test_pending_cap(self):
303 d = os.path.join(self.state, "pending")
304 os.makedirs(d)
305 for i in range(200):
306 with open(os.path.join(d, f"20240101T000000Z-{i:06x}.json"), "w") as f:
307 f.write("{}")
308 status, _, _ = self.post()
309 self.assertEqual(status, 503)
310 self.assertEqual(len(os.listdir(d)), 200)
311
312 def make_post(self, rel):
313 os.makedirs(os.path.join(self.siteroot, rel), exist_ok=True)
314 with open(os.path.join(self.siteroot, rel, "index.html"), "w") as f:
315 f.write("<html></html>")
316
317 def test_percent_encoded_slug_uses_decoded_folder(self):
318 self.make_post("2024/caf\u00e9")
319 self.assert_ok_path(self.post(path="/2024/caf%C3%A9/"), "/2024/caf%C3%A9/")
320 self.assertEqual(self.load(self.pending()[0])["path"], "/2024/caf%C3%A9/")
321 for p in self.pending():
322 os.unlink(p)
323 self.assert_refused(self.post(path="/2024/caf%C3%A8/")) # decoded folder missing
324 self.assert_refused(self.post(path="/2024/cafe%CC%81/"))
325
326 def assert_ok_path(self, resp, path):
327 self.assertEqual(resp[0], 303)
328 self.assertEqual(resp[1]["Location"], ORIGIN + path + "#comment-pending")
329
330 def test_path_is_stored_canonical(self):
331 self.make_post("2024/caf\u00e9")
332 for spelled in ("/2024/05/01/sl%75g/", "/2024/05/01/s%6C%75g/"):
333 with self.subTest(path=spelled):
334 self.assert_ok_path(self.post(path=spelled), POST_PATH)
335 for spelled in ("/2024/caf%c3%a9/", "/2024/caf%C3%A9/", "/2024/caf%c3%A9/"):
336 with self.subTest(path=spelled):
337 self.assert_ok_path(self.post(path=spelled), "/2024/caf%C3%A9/")
338 self.assertEqual(sorted(self.load(p)["path"] for p in self.pending()),
339 [POST_PATH] * 2 + ["/2024/caf%C3%A9/"] * 3)
340
341 def test_canonical_path_parity_with_comments_py(self):
342 sys.path.insert(0, os.path.join(ROOT, "server"))
343 try:
344 import comments
345 finally:
346 sys.path.pop(0)
347 tricky = ["/2024/05/01/sl%75g/", "/2024/caf%c3%a9/", "/2024/caf%C3%A9/", "/2024/a+b/", "/2024/a%2bb/",
348 "/~me/", "/_notes/", "/2024/caf%e9/", "/2024/a%20b/", "/2024/a%7eb/", "/2024/a%2Cb/", "/2024/a%25b/",
349 "/2024/a%41/", "/2024/x.y_z-w/", "/2024/%e2%82%ac/"]
350 checked = 0
351 for p in tricky:
352 if not comments.valid_path(p):
353 n = len(self.pending())
354 self.assertGreaterEqual(self.post(path=p)[0], 400) # also invalid here
355 self.assertEqual(len(self.pending()), n)
356 continue
357 expected = comments.canonical_path(p)
358 self.make_post(urllib.parse.unquote(expected.strip("/")))
359 before = set(self.pending())
360 with self.subTest(path=p):
361 self.assert_ok_path(self.post(path=p, name=f"n{checked}"), expected)
362 stored = [self.load(f)["path"] for f in set(self.pending()) - before]
363 self.assertEqual(stored, [expected])
364 checked += 1
365 if checked % 4 == 0: # stay under the rate limit
366 os.unlink(os.path.join(self.state, "rate", os.listdir(os.path.join(self.state, "rate"))[0]))
367 self.assertGreater(checked, 8)
368
369 def test_underscore_and_tilde_segments(self):
370 self.make_post("_x/~u")
371 self.assert_ok_path(self.post(path="/_x/~u/"), "/_x/~u/")
372 self.make_post("2024/a+b")
373 self.assert_ok_path(self.post(path="/2024/a+b/"), "/2024/a+b/")
374
375 def test_decoded_segment_rules(self):
376 for bad in ["/2024/a%2e%2eb/", "/2024/a%EF%BF%BDb/", "/2024/a%E2%80%AEb/", "/2024/%ff/",
377 "/2024/a%0Ab/", "/2024/a%C2%ADb/", "/2024/%2e/"]:
378 with self.subTest(path=bad):
379 self.assert_refused(self.post(path=bad))
380
381 def test_self_origin_allowed_but_not_by_default(self):
382 self.assert_ok(self.post(headers={"Origin": self.self_origin}))
383
384 def test_double_submit_of_confirm_token_stores_two(self):
385 # There is no one-time nonce: a signed token can be replayed within an hour.
386 # Accepted trade-off (the rate limit and moderation bound the damage); this documents it.
387 ts = str(int(time.time()) - 10)
388 fields = {"site": "writing", "path": POST_PATH, "name": "Ann", "text": "Nice post.", "ts": ts,
389 "sig": self.sign(ts, "writing", POST_PATH, "Ann", "Nice post.")}
390 self.assert_ok(self.request("POST", fields))
391 self.assert_ok(self.request("POST", fields))
392 self.assertEqual(len(self.pending()), 2)
393
394 def test_form_cap(self):
395 d = os.path.join(self.state, "pending")
396 os.makedirs(d)
397 for i in range(100):
398 with open(os.path.join(d, f"20240101T000000Z-{i:06x}.json"), "w") as f:
399 json.dump({"source": "email"}, f)
400 self.assert_ok(self.post()) # 100 email comments do not count
401 for i in range(100, 199):
402 with open(os.path.join(d, f"20240101T000000Z-{i:06x}.json"), "w") as f:
403 json.dump({"source": "form"}, f)
404 status, _, _ = self.post() # now 100 form files (99 + the one just stored)
405 self.assertEqual(status, 503)
406
407 def test_json_indent_matches_comments_py(self):
408 self.assert_ok(self.post())
409 with open(self.pending()[0], encoding="utf-8") as f:
410 raw = f.read()
411 self.assertIn('\n "author": "Ann",\n', raw)
412 self.assertTrue(raw.endswith("}\n"))
413
414 def test_missing_secret_fails(self):
415 os.unlink(os.path.join(self.state, "secret"))
416 status, _, body = self.post()
417 self.assertEqual(status, 500)
418 self.assertNotIn("secret", body)
419 self.assertEqual(self.pending(), [])
420
421
422if __name__ == "__main__":
423 unittest.main()