479 lines
18 KiB
Python
479 lines
18 KiB
Python
"""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 unittest import mock
|
|
|
|
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(),
|
|
)
|
|
# The board is per-user: staff only see their own staged uploads.
|
|
self.assertNotIn(
|
|
str(temp.id),
|
|
self.client_for("sec-staff").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",
|
|
)
|
|
|
|
@mock.patch("requests.get")
|
|
def test_redirects_off_the_allowlist_are_refused(self, mocked_get):
|
|
redirect = mock.Mock(is_redirect=True, is_permanent_redirect=False)
|
|
redirect.headers = {"Location": "http://127.0.0.1:8000/health"}
|
|
mocked_get.return_value = redirect
|
|
|
|
with self.assertRaises(services.RemoteUrlError):
|
|
services.open_remote("https://static1.e621.net/x.png")
|
|
# The internal address was never requested: only the first hop was.
|
|
self.assertEqual(mocked_get.call_count, 1)
|
|
redirect.close.assert_called()
|
|
|
|
@mock.patch("requests.get")
|
|
def test_redirects_within_the_allowlist_are_followed(self, mocked_get):
|
|
redirect = mock.Mock(is_redirect=True, is_permanent_redirect=False)
|
|
redirect.headers = {"Location": "https://static2.e621.net/x.png"}
|
|
final = mock.Mock(is_redirect=False, is_permanent_redirect=False)
|
|
mocked_get.side_effect = [redirect, final]
|
|
|
|
self.assertIs(services.open_remote("https://static1.e621.net/x.png"), final)
|
|
self.assertEqual(mocked_get.call_count, 2)
|
|
|
|
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)
|