"""Security regression suite. Covers the checks that must never live on the SPA side alone: guest visibility, object ownership, staged-upload/similarity privacy, staff role boundaries, account deletion rules, encrypted credentials, throttling and the SSRF allowlist for server-side downloads. Run: backend/venv/bin/python manage.py test apps.core.tests The database user needs rights to create the test database once, e.g.: GRANT ALL ON `test_j621`.* TO 'j621'@'%'; """ import hashlib import json import shutil import tempfile import time from pathlib import Path from django.contrib.auth import get_user_model from django.core import signing from django.core.cache import cache from django.core.files.uploadedfile import SimpleUploadedFile from django.test import Client, TestCase, override_settings from rest_framework.authtoken.models import Token from apps.library import services from apps.library.models import ( DownloadTask, MediaItem, MediaLocation, SimilarityCheck, TempUpload, ) from apps.library.services import MEDIA_FILE_SALT User = get_user_model() def jpost(client, path, body=None): return client.post(path, data=json.dumps(body or {}), content_type="application/json") def jpatch(client, path, body=None): return client.patch(path, data=json.dumps(body or {}), content_type="application/json") class SecurityTestCase(TestCase): """Shared fixtures: throwaway accounts for every role and temp folders.""" @classmethod def setUpClass(cls): super().setUpClass() cls._tmp = tempfile.mkdtemp(prefix="j621-security-") cls._media = Path(cls._tmp) / "media" cls._watched = cls._media / "library" cls._watched.mkdir(parents=True, exist_ok=True) cls._settings = override_settings( MEDIA_ROOT=str(cls._media), WATCHED_FOLDER=str(cls._watched), ) cls._settings.enable() @classmethod def tearDownClass(cls): cls._settings.disable() shutil.rmtree(cls._tmp, ignore_errors=True) super().tearDownClass() def setUp(self): cache.clear() # throttle counters and cached payloads self.users = {} for name, role in ( ("sec-user", "user"), ("sec-uploader", "uploader"), ("sec-uploader2", "uploader"), ("sec-staff", "staff"), ("sec-staff2", "staff"), ): user = User.objects.create_user( username=name, password="audit-pass-123456" ) user.role = role user.save(update_fields=["role"]) self.users[name] = user self.admin = User.objects.create_superuser( username="sec-admin", password="audit-pass-123456" ) self.tokens = { name: Token.objects.create(user=user).key for name, user in self.users.items() } self.tokens["sec-admin"] = Token.objects.create(user=self.admin).key def client_for(self, name): client = Client() client.defaults["HTTP_AUTHORIZATION"] = f"Token {self.tokens[name]}" return client @property def guest(self): return Client() def make_item(self, label, *, hidden=False, owner=None, tags=None): path = self._watched / f"{label}.bin" path.write_bytes(b"audit-" + label.encode()) item = MediaItem.objects.create( md5=hashlib.md5(label.encode()).hexdigest(), size=path.stat().st_size, uploaded_by=owner, tags=tags or [], ) MediaLocation.objects.create( item=item, path=str(path), rel_path=path.name, mtime=time.time() ) if hidden: MediaItem.objects.filter(pk=item.pk).update(hidden_from_guests=True) item.refresh_from_db() return item def old_signature(self, item, action="raw", age=3 * 86400): """A valid signature minted `age` seconds ago.""" real_time = signing.time class Backdated: def time(self): return real_time.time() - age try: signing.time = Backdated() return signing.dumps( {"item": item.id, "user": self.users["sec-uploader"].id, "action": action}, salt=MEDIA_FILE_SALT, ) finally: signing.time = real_time class GuestVisibilityTests(SecurityTestCase): def setUp(self): super().setUp() self.visible = self.make_item("visible", owner=self.users["sec-uploader"]) self.hidden = self.make_item( "hidden", hidden=True, owner=self.users["sec-uploader"], tags=["sec_audit_hidden_tag"], ) def test_guest_list_and_retrieve_hide_protected_items(self): response = self.guest.get( f"/api/files/?j_ids=J-{self.visible.id},J-{self.hidden.id}" ) self.assertEqual( {row["j_id"] for row in response.json()["results"]}, {f"J-{self.visible.id}"}, ) self.assertEqual(self.guest.get(f"/api/files/J-{self.hidden.id}/").status_code, 404) self.assertEqual( self.guest.get(f"/api/files/J-{self.hidden.id}/raw/").status_code, 404 ) self.assertEqual( self.guest.get(f"/api/files/J-{self.hidden.id}/thumbnail/").status_code, 404 ) self.assertEqual(self.guest.get(f"/api/files/{self.hidden.md5}/").status_code, 404) self.assertEqual( self.guest.get(f"/api/files/J-{self.hidden.id}/neighbors/").status_code, 404 ) def test_guest_lookup_and_tag_cloud_hide_protected_items(self): response = jpost( self.guest, "/api/files/lookup/", {"md5s": [self.visible.md5, self.hidden.md5]}, ) self.assertEqual(response.json()["found"], [self.visible.md5]) cloud = self.guest.get("/api/tags/cloud/").json()["tags"] self.assertNotIn("sec_audit_hidden_tag", [entry["tag"] for entry in cloud]) def test_authenticated_users_and_signed_urls_see_protected_items(self): uploader = self.client_for("sec-uploader") self.assertEqual(uploader.get(f"/api/files/J-{self.hidden.id}/").status_code, 200) signed = signing.dumps( {"item": self.hidden.id, "user": self.users["sec-uploader"].id, "action": "raw"}, salt=MEDIA_FILE_SALT, ) self.assertEqual( self.guest.get(f"/api/files/J-{self.hidden.id}/raw/?sig={signed}").status_code, 200, ) def test_signature_integrity(self): signed = signing.dumps( {"item": self.hidden.id, "user": self.users["sec-uploader"].id, "action": "raw"}, salt=MEDIA_FILE_SALT, ) raw = f"/api/files/J-{self.hidden.id}/raw/" thumbnail = f"/api/files/J-{self.hidden.id}/thumbnail/" # Tampered signature (same action). self.assertEqual(self.guest.get(f"{raw}?sig={signed[:-4]}AAAA").status_code, 404) # Valid signature, wrong action. self.assertEqual(self.guest.get(f"{thumbnail}?sig={signed}").status_code, 404) # Expired signature (minted three days ago). expired = self.old_signature(self.hidden) self.assertEqual(self.guest.get(f"{raw}?sig={expired}").status_code, 404) class RoleBoundaryTests(SecurityTestCase): def test_non_uploader_is_read_only(self): user = self.client_for("sec-user") self.assertEqual(user.get("/api/storage/").status_code, 403) self.assertEqual(user.get("/api/duplicates/md5/").status_code, 403) self.assertEqual( jpost(user, "/api/duplicates/visual/", {"j_id": "J-1"}).status_code, 403 ) self.assertEqual(user.get("/api/uploads/").status_code, 403) self.assertEqual(user.get("/api/online/downloads/").status_code, 403) self.assertEqual(user.get("/api/matches/").status_code, 403) self.assertEqual(user.get("/api/stats/").status_code, 403) self.assertEqual(user.get("/api/users/").status_code, 403) self.assertEqual(jpost(user, "/api/delete/", {"j_ids": []}).status_code, 403) self.assertEqual(jpost(user, "/api/temp/clear/").status_code, 403) def test_uploader_and_staff_reach_their_tools(self): uploader = self.client_for("sec-uploader") staff = self.client_for("sec-staff") self.assertEqual(uploader.get("/api/storage/").status_code, 200) self.assertEqual(jpost(uploader, "/api/delete/", {"j_ids": []}).status_code, 200) self.assertEqual(uploader.get("/api/duplicates/md5/").status_code, 200) self.assertEqual(staff.get("/api/stats/").status_code, 200) self.assertEqual(staff.get("/api/users/").status_code, 200) def test_only_admins_change_staff_roles(self): staff = self.client_for("sec-staff") user_id = self.users["sec-user"].id self.assertEqual( jpatch(staff, f"/api/users/{user_id}/", {"role": "staff"}).status_code, 403, ) self.assertEqual( jpatch( staff, f"/api/users/{self.users['sec-staff2'].id}/", {"role": "user"}, ).status_code, 403, ) admin = self.client_for("sec-admin") self.assertEqual( jpatch(admin, f"/api/users/{user_id}/", {"role": "uploader"}).status_code, 200, ) def test_deletion_rules(self): staff = self.client_for("sec-staff") admin = self.client_for("sec-admin") self.assertEqual( staff.delete(f"/api/users/{self.users['sec-user'].id}/").status_code, 204 ) self.assertEqual( staff.delete(f"/api/users/{self.users['sec-staff2'].id}/").status_code, 403 ) self.assertEqual( staff.delete(f"/api/users/{self.users['sec-staff'].id}/").status_code, 400 ) self.assertEqual( admin.delete(f"/api/users/{self.users['sec-staff2'].id}/").status_code, 204 ) def test_deleting_a_user_keeps_library_items(self): owner = self.users["sec-uploader2"] item = self.make_item("orphan", owner=owner) self.client_for("sec-admin").delete(f"/api/users/{owner.id}/") item.refresh_from_db() self.assertIsNone(item.uploaded_by_id) class ItemOwnershipTests(SecurityTestCase): def setUp(self): super().setUp() self.item = self.make_item("owned", owner=self.users["sec-uploader"]) def test_other_uploaders_cannot_touch_items(self): other = self.client_for("sec-uploader2") self.assertEqual( jpost(other, f"/api/files/J-{self.item.id}/match/", {"post_id": "x"}).status_code, 403, ) self.assertEqual( jpost(other, f"/api/files/J-{self.item.id}/unlink/").status_code, 403 ) self.assertEqual( other.post(f"/api/files/J-{self.item.id}/optimize/").status_code, 403 ) self.assertEqual( jpatch(other, f"/api/files/J-{self.item.id}/", {"rating": "s"}).status_code, 403, ) def test_owner_and_staff_pass_the_permission_check(self): for name in ("sec-uploader", "sec-staff"): client = self.client_for(name) # An invalid post id proves authorization happened: validation answers. self.assertEqual( jpost(client, f"/api/files/J-{self.item.id}/match/", {"post_id": "x"}).status_code, 400, ) self.assertEqual( client.post(f"/api/files/J-{self.item.id}/optimize/").status_code, 400 ) owner = self.client_for("sec-uploader") self.assertEqual( jpatch( owner, f"/api/files/J-{self.item.id}/", {"rating": "", "tags": [], "notes": ""}, ).status_code, 200, ) class PrivacyTests(SecurityTestCase): def test_staged_uploads_are_private(self): temp = TempUpload.objects.create( user=self.users["sec-uploader2"], file=SimpleUploadedFile("staged.bin", b"staged"), original_filename="staged.bin", md5=hashlib.md5(b"staged").hexdigest(), size=6, ) self.assertEqual( self.client_for("sec-uploader2").get(f"/api/uploads/{temp.id}/file/").status_code, 200, ) self.assertEqual( self.client_for("sec-uploader").get(f"/api/uploads/{temp.id}/file/").status_code, 404, ) self.assertEqual( self.client_for("sec-staff").get(f"/api/uploads/{temp.id}/file/").status_code, 200, ) self.assertEqual(self.guest.get(f"/api/uploads/{temp.id}/file/").status_code, 401) self.assertNotIn( str(temp.id), self.client_for("sec-uploader").get("/api/uploads/").content.decode() ) def test_similarity_checks_are_private(self): check = SimilarityCheck.objects.create( user=self.users["sec-uploader2"], file=SimpleUploadedFile("similar.bin", b"similar"), original_filename="similar.bin", md5=hashlib.md5(b"similar").hexdigest(), size=7, ) self.assertEqual( self.client_for("sec-uploader2").get(f"/api/similarity/{check.id}/").status_code, 200, ) self.assertEqual( self.client_for("sec-uploader").get(f"/api/similarity/{check.id}/").status_code, 404, ) self.assertEqual( self.client_for("sec-uploader").get(f"/api/similarity/{check.id}/file/").status_code, 404, ) self.assertEqual( self.client_for("sec-staff").get(f"/api/similarity/{check.id}/").status_code, 200, ) class CredentialEncryptionTests(SecurityTestCase): def test_keys_are_encrypted_at_rest_and_only_owners_see_them(self): client = self.client_for("sec-user") response = client.put( "/api/auth/e621/", data=json.dumps( {"username": "tester", "api_key": "abc123def456", "base_url": "https://e621.net"} ), content_type="application/json", ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["api_key"], "abc123def456") user = User.objects.get(pk=self.users["sec-user"].pk) self.assertTrue(user.e621_api_key.startswith("enc:")) self.assertNotIn("abc123def456", user.e621_api_key) self.assertTrue(user.e621_configured) # Other users never see it. other = self.client_for("sec-staff") self.assertNotIn("abc123def456", other.get("/api/auth/me/").content.decode()) self.assertNotIn("abc123def456", other.get("/api/auth/e621/").content.decode()) def test_undecryptable_values_read_as_not_configured(self): user = User.objects.get(pk=self.users["sec-user"].pk) user.e621_username = "tester" user.e621_api_key = "enc:not-a-valid-token" user.save(update_fields=["e621_username", "e621_api_key"]) user.refresh_from_db() self.assertEqual(user.e621_api_key_plain, "") self.assertFalse(user.e621_configured) class ThrottleTests(SecurityTestCase): def test_login_is_rate_limited(self): codes = [ self.guest.post( "/api/auth/token/", data=json.dumps({"username": "nobody", "password": "wrong"}), content_type="application/json", ).status_code for _ in range(6) ] self.assertEqual(codes[:5], [400] * 5) self.assertEqual(codes[5], 429) def test_anonymous_polling_is_not_throttled(self): self.assertEqual( {self.guest.get("/api/status/").status_code for _ in range(12)}, {200} ) class RemoteUrlTests(SecurityTestCase): def test_allowlist(self): for url in ("file:///etc/passwd", "http://127.0.0.1:1/", "https://evil.example/"): with self.assertRaises(services.RemoteUrlError): services.validate_remote_url(url) self.assertEqual( services.validate_remote_url("https://static1.e621.net/data/x.png"), "https://static1.e621.net/data/x.png", ) def test_download_creation_rejects_internal_urls(self): uploader = self.client_for("sec-uploader") for url in ("http://127.0.0.1:1/", "http://192.168.1.1/", "file:///etc/passwd"): response = jpost( uploader, "/api/online/downloads/", {"url": url, "filename": "x.bin"} ) self.assertEqual(response.status_code, 400, url) self.assertEqual(DownloadTask.objects.count(), 0) def test_guest_proxy_rejects_internal_urls(self): response = self.guest.get( "/api/online/file/?url=http%3A%2F%2F127.0.0.1%3A1%2F" ) self.assertEqual(response.status_code, 400)