JakeBreath/J621
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Docker deployment (2 images, 3 composes, Tailscale funnels) + committed security suite

Browse Source 
Test suite (the manual audit harness, now a real test):
- backend/apps/core/tests/test_security.py: 20 transactional tests across
  guest visibility, object ownership, staged-upload/similarity privacy,
  staff role boundaries, deletion rules, encrypted credentials, throttling
  and the download allowlist. Uses temp media folders and clears cache.
  Needs a one-time GRANT on test_j621 (documented in the module + README).

Images (deploy/J621-Frontend, deploy/J621-Backend, repo root as context):
- Frontend: node build -> static nginx with SPA fallback, asset caching and
  an internal health endpoint.
- Backend: gunicorn + whitenoise (admin static collected at build), ffmpeg
  for video thumbnails, migrations applied on start, GIT_HASH build arg so
  the shell's version pill shows the commit.

Composes (distinct project names so they coexist with the dev stack):
- compose.yml (both), compose.frontend.yml, compose.backend.yml.
- A shared nginx proxy service is the only entry point (no host nginx, no
  published host ports): /api,/admin,/static,/health -> backend, everything
  else -> SPA; both upstreams resolve at request time so one config serves
  all variants.
- A Tailscale sidecar per compose shares the nginx network namespace;
  serve.default/frontend/backend.json use funnel ports 443 and 8443 only
  (10000 is the remaining allowance) with ${TS_CERT_DOMAIN} substitution.

Registry: push_frontend.sh / push_backend.sh / push_all.sh build multi-arch
images and push :latest + :<sha> to the Gitea registry, following the
existing Packs-site pattern.

Docs: deploy/README.md + .env.example, ROADMAP section 6 updated,
AGENTS.md deployment and test notes.

Verified: all three composes validate, both nginx configs pass nginx -t,
both images build, the combined stack boots against real MariaDB/Redis
(migrations applied, /health ok, whitenoise serving admin static, env=prod,
git hash baked in), the proxy serves the SPA and routes /api, and
manage.py test apps.core.tests passes 20/20.
This commit is contained in:
JakeBreath
2026-09-18 00:51:29 -05:00
parent f86eccf9a3
commit 8ebda6ab20
23 changed files with 1271 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/core/tests/__init__.py
View File
backend/apps/core/tests/test_security.py
+449
View File
@@ -0,0 +1,449 @@
"""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)