diff --git a/tests/test_web_features.py b/tests/test_web_features.py new file mode 100644 index 0000000..3e15ccf --- /dev/null +++ b/tests/test_web_features.py @@ -0,0 +1,188 @@ +"""Creator and administrator tools enforce the same visibility on every route.""" +from __future__ import annotations + +from contextlib import ExitStack +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest.mock import patch +import io +import unittest + +import database +from database import DatabaseBackend +import server_config +import web_features +import webapp + + +TOKEN = "a" * 64 + + +class Request: + def __init__(self, path, *, ip="192.0.2.9", fields=None, query=None, csrf=TOKEN): + self.path = path.encode("utf-8") + self._ip = ip + self.args = {key.encode(): [str(value).encode()] for key, value in (query or {}).items()} + if fields is not None: + self.args.update({key.encode(): [str(value).encode()] for key, value in fields.items()}) + self.args[b"csrf"] = [csrf.encode()] + self.content = io.BytesIO() + self.headers = {b"host": b"localhost:8080", b"cookie": ("ose_csrf=" + TOKEN).encode()} + self.response_headers = {} + self.response_code = 200 + + def getClientIP(self): + return self._ip + + def getAllHeaders(self): + return self.headers + + def setHeader(self, key, value): + self.response_headers[key.lower()] = value + + def setResponseCode(self, code): + self.response_code = code + + def isSecure(self): + return False + + +class WebFeatureTests(unittest.TestCase): + def setUp(self): + self.stack = ExitStack() + self.addCleanup(self.stack.close) + self.root = Path(self.stack.enter_context(TemporaryDirectory())) + self.stack.enter_context(patch.dict("os.environ", {"REXIMEMO_CONFIG": str(self.root / "settings.json")})) + self.creators = self.root / "Creators" + self.comments = self.root / "Comments" + self.creators.mkdir() + self.comments.mkdir() + self.stack.enter_context(patch.object(database, "CREATORS_DIR", self.creators)) + self.stack.enter_context(patch.object(database, "COMMENTS_DIR", self.comments)) + self.db = DatabaseBackend(self.root / "test.sqlite3") + self.stack.callback(self.db.close) + self.stack.enter_context(patch.object(webapp, "Database", self.db)) + self.stack.enter_context(patch.object(web_features, "Database", self.db)) + self.router = webapp.WebRoot() + self.alice = self.db.create_account("Alice", "test password") + self.bob = self.db.create_account("Bob", "test password") + self.admin = self.db.get_account_by_username("reximemo")["id"] + for account, address in ((self.alice, "192.0.2.1"), (self.bob, "192.0.2.2"), (self.admin, "192.0.2.3")): + self.db.login_ip(address, account) + room = self.db.get_or_create_creator("192.0.2.1", self.alice, fsid="A000000000000001") + with self.db.conn: + cur = self.db.conn.execute( + "INSERT INTO flipnotes(public_id,creator_room_id,creator_fsid,filename,title,channel_id,posted_at) VALUES(?,?,?,?,?,?,?)", + ("f_test", room["id"], room["fsid"], "sample", "Original title", 1, 100), + ) + self.note_id = cur.lastrowid + path = self.db.FlipnotePath(room["fsid"], "sample") + path.parent.mkdir(parents=True) + path.write_bytes(b"Original PPM bytes") + + def get(self, path, *, ip="192.0.2.9", query=None): + request = Request(path, ip=ip, query=query) + body = self.router.render_GET(request) + return request, body + + def post(self, path, fields=None, *, ip="192.0.2.3", csrf=TOKEN): + request = Request(path, ip=ip, fields=fields or {}, csrf=csrf) + body = self.router.render_POST(request) + return request, body + + def test_creator_can_edit_but_another_account_cannot(self): + fields = {"title": "Edited title", "description": "A description", "channel_id": 1} + denied, _ = self.post("/watch/f_test/manage", fields, ip="192.0.2.2") + self.assertEqual(denied.response_code, 403) + self.assertEqual(self.db.get_flipnote_by_id(self.note_id)["title"], "Original title") + allowed, _ = self.post("/watch/f_test/manage", fields, ip="192.0.2.1") + self.assertEqual(allowed.response_code, 303) + self.assertEqual(self.db.get_flipnote_by_id(self.note_id)["title"], "Edited title") + self.assertEqual(self.db.get_flipnote_by_id(self.note_id)["description"], "A description") + self.assertEqual(self.get("/watch/f_test/manage", ip="192.0.2.2")[0].response_code, 403) + self.assertEqual(self.get("/watch/f_test/manage", ip="192.0.2.1")[0].response_code, 200) + + def test_only_commenter_or_administrator_can_remove_comment(self): + comment = self.db.add_text_comment(self.note_id, "Bob's comment", account_id=self.bob) + denied, _ = self.post(f"/comments/{comment}/delete", ip="192.0.2.1") + self.assertEqual(denied.response_code, 403) + self.assertIsNotNone(self.db.get_comment(comment)) + allowed, _ = self.post(f"/comments/{comment}/delete", ip="192.0.2.2") + self.assertEqual(allowed.response_code, 303) + self.assertIsNone(self.db.get_comment(comment)) + + def test_moderation_hides_public_pages_and_media_but_staff_can_review(self): + note = self.db.get_flipnote_by_id(self.note_id) + request, _ = self.post("/admin/flipnote-visibility", {"id": self.note_id, "hidden": 1, "reason": "Review"}) + self.assertEqual(request.response_code, 303) + with patch.object(webapp, "ppm_thumbnail_png", return_value=b"PNG preview"): + for path in ("/watch/f_test", "/media/f_test.ppm", "/flipnote/f_test.ppm", "/thumb/f_test.png"): + with self.subTest(path=path): + public, _ = self.get(path) + owner, _ = self.get(path, ip="192.0.2.1") + reviewer, body = self.get(path, ip="192.0.2.3") + self.assertEqual(public.response_code, 404) + self.assertEqual(owner.response_code, 404) + self.assertEqual(reviewer.response_code, 200) + self.assertIn(b"no-store", reviewer.response_headers[b"cache-control"]) + self.assertTrue(body) + public, body = self.get("/browse") + self.assertNotIn(note["public_id"].encode(), body) + admin, body = self.get("/browse", ip="192.0.2.3") + self.assertIn(b"hidden-content", body) + self.assertIn(b"f_test", body) + request, _ = self.post("/admin/flipnote-visibility", {"id": self.note_id, "hidden": 0}) + self.assertEqual(request.response_code, 303) + self.assertEqual(self.get("/media/f_test.ppm")[1], b"Original PPM bytes") + + def test_administrator_tabs_render_and_regular_accounts_are_denied(self): + self.db.add_text_comment(self.note_id, "Example comment", account_id=self.bob) + for tab, heading in (("flipnotes", "Flipnotes"), ("comments", "Example comment"), + ("channels", "New channel"), ("server", "Server appearance"), ("bans", "Block an IP")): + with self.subTest(tab=tab): + admin, body = self.get("/admin", ip="192.0.2.3", query={"tab": tab}) + self.assertEqual(admin.response_code, 200) + self.assertIn(heading.encode(), body) + self.assertIn(b'name="csrf"', body) + regular, _ = self.get("/admin", ip="192.0.2.1", query={"tab": tab}) + self.assertEqual(regular.response_code, 403) + + def test_channel_removal_moves_notes_and_preserves_default_channel(self): + channel = self.db.create_channel("Animations", "Drawn stories") + self.db.update_flipnote(self.note_id, "Film", channel_id=channel) + response, _ = self.post("/admin/channel-delete", {"id": channel, "target_channel_id": 1}) + self.assertEqual(response.response_code, 303) + self.assertIsNone(self.db.get_channel(channel)) + self.assertEqual(self.db.get_flipnote_by_id(self.note_id)["channel_id"], 1) + protected, _ = self.post("/admin/channel-delete", {"id": 1, "target_channel_id": 1}) + self.assertEqual(protected.response_code, 400) + self.assertIsNotNone(self.db.get_channel(1)) + + def test_server_settings_save_and_guest_posting_setting_is_enforced(self): + fields = {"site_name": "My Flipnote Server", "tagline": "Our drawings", "welcome_message": "Welcome!", + "accent_colour": "#123456", "max_upload_kib": 2048, "registration_enabled": 1} + request, _ = self.post("/admin/server", fields) + self.assertEqual(request.response_code, 303) + config = server_config.load_config() + self.assertEqual(config["site_name"], "My Flipnote Server") + self.assertFalse(config["guest_posting"]) + public, body = self.get("/") + self.assertIn(b"My Flipnote Server", body) + rejected, _ = self.post("/watch/f_test/comment", {"text": "Guest comment"}, ip="192.0.2.9") + self.assertEqual(rejected.response_code, 403) + allowed, _ = self.post("/watch/f_test/comment", {"text": "Signed in comment"}, ip="192.0.2.1") + self.assertEqual(allowed.response_code, 303) + self.assertEqual(self.db.count_comments(self.note_id), 1) + + def test_invalid_csrf_rejects_creator_and_staff_mutations(self): + fields = {"title": "Bad change", "channel_id": 1} + rejected, _ = self.post("/watch/f_test/manage", fields, ip="192.0.2.1", csrf="b" * 64) + self.assertEqual(rejected.response_code, 403) + rejected, _ = self.post("/admin/flipnote-visibility", {"id": self.note_id, "hidden": 1}, csrf="") + self.assertEqual(rejected.response_code, 403) + self.assertFalse(self.db.get_flipnote_by_id(self.note_id)["hidden"]) + self.assertEqual(self.db.get_flipnote_by_id(self.note_id)["title"], "Original title") + + +if __name__ == "__main__": + unittest.main()