Add a Random image endpoint and SPA page (with fastfetch mode)
Backend: GET /api/random/ (aliases /random and /random/) returns a random library image with: - rating=s,q,e filtering (comma separated, default any); - fastfetch mode (?fastfetch=1 or any User-Agent containing "fastfetch") that only considers png/jpg/gif - what terminal viewers can show; - JSON with j_id, filename, extension, rating, size, e621 id plus absolute url/download_url/thumbnail_url. Authenticated callers get signed URLs so fastfetch and image viewers can load them without headers; guests get unsigned URLs and never receive hidden_from_guests items. Tests: apps/library/tests/test_random.py (8 tests) covering the response contract, guest signatures, image-only default, the fastfetch format restriction (flag and User-Agent), rating filters, guest visibility and the short alias. Frontend: /random page with rating pills, R to roll, Open/Download and a library link, plus navigation and command palette entries; needs a backend, hidden in local mode. nginx: /random negotiates on Accept so browsers keep getting the SPA while scripts get the JSON (verified with the proxy and frontend containers). Also fixes a regression from the SSRF change: the guest download proxy still referenced the removed 'parsed' variable on its success path, so every proxied download would have 500'd. Redirect hops are now covered by tests with a mocked requests.get.
This commit is contained in:
@@ -17,6 +17,7 @@ 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
|
||||
@@ -433,6 +434,28 @@ class RemoteUrlTests(SecurityTestCase):
|
||||
"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"):
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
"""Tests for the random image endpoint (`/api/random/`, `/random`).
|
||||
|
||||
Used by the SPA's Random page and by shell greeting scripts (fish_greeting
|
||||
with fastfetch), so the response contract matters:
|
||||
* JSON with an absolute, directly fetchable URL,
|
||||
* signed for authenticated callers (image viewers send no headers),
|
||||
* fastfetch mode restricted to png/jpg/gif,
|
||||
* rating filters and guest visibility applied server-side.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import shutil
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.test import Client, TestCase, override_settings
|
||||
|
||||
from rest_framework.authtoken.models import Token
|
||||
|
||||
from apps.library.models import MediaItem, MediaLocation
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
IMAGE_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp", "apng"}
|
||||
FASTFETCH_EXTENSIONS = {"png", "jpg", "jpeg", "gif"}
|
||||
|
||||
|
||||
class RandomItemTests(TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
super().setUpClass()
|
||||
cls._tmp = tempfile.mkdtemp(prefix="j621-random-")
|
||||
cls._watched = Path(cls._tmp) / "library"
|
||||
cls._watched.mkdir(parents=True, exist_ok=True)
|
||||
cls._settings = override_settings(
|
||||
MEDIA_ROOT=cls._tmp, 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):
|
||||
self.user = User.objects.create_user(
|
||||
username="random-user", password="random-pass-123456"
|
||||
)
|
||||
token = Token.objects.create(user=self.user)
|
||||
self.authed = Client()
|
||||
self.authed.defaults["HTTP_AUTHORIZATION"] = f"Token {token.key}"
|
||||
self.guest = Client()
|
||||
|
||||
def make_item(self, label, extension, rating, *, hidden=False):
|
||||
path = self._watched / f"{label}.{extension}"
|
||||
path.write_bytes(b"random-" + label.encode())
|
||||
item = MediaItem.objects.create(
|
||||
md5=hashlib.md5(label.encode()).hexdigest(),
|
||||
size=path.stat().st_size,
|
||||
rating=rating,
|
||||
uploaded_by=self.user,
|
||||
)
|
||||
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)
|
||||
return item
|
||||
|
||||
def test_returns_image_with_signed_absolute_url(self):
|
||||
item = self.make_item("plain", "png", "s")
|
||||
response = self.authed.get("/api/random/")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
data = response.json()
|
||||
self.assertEqual(data["j_id"], f"J-{item.id}")
|
||||
self.assertEqual(data["extension"], "png")
|
||||
self.assertEqual(data["kind"], "image")
|
||||
self.assertEqual(data["rating"], "s")
|
||||
self.assertTrue(data["url"].startswith("http"))
|
||||
self.assertIn("sig=", data["url"])
|
||||
self.assertIn("download=1", data["download_url"])
|
||||
self.assertFalse(data["fastfetch"])
|
||||
|
||||
def test_guest_url_is_unsigned_and_still_serves(self):
|
||||
self.make_item("guest", "jpg", "s")
|
||||
data = self.guest.get("/api/random/").json()
|
||||
self.assertNotIn("sig=", data["url"])
|
||||
path = data["url"].replace("http://testserver", "")
|
||||
self.assertEqual(self.guest.get(path).status_code, 200)
|
||||
|
||||
def test_default_mode_returns_images_only(self):
|
||||
self.make_item("movie", "mp4", "s")
|
||||
self.make_item("picture", "webp", "s")
|
||||
for _ in range(10):
|
||||
extension = self.authed.get("/api/random/").json()["extension"]
|
||||
self.assertIn(extension, IMAGE_EXTENSIONS)
|
||||
|
||||
def test_fastfetch_mode_flag_and_user_agent_restrict_formats(self):
|
||||
self.make_item("movie", "mp4", "s")
|
||||
self.make_item("modern", "webp", "s")
|
||||
self.make_item("picture", "png", "s")
|
||||
self.make_item("animation", "gif", "s")
|
||||
|
||||
attempts = [("flag", {"fastfetch": "1"}, {}), ("ua", {}, {"HTTP_USER_AGENT": "fastfetch/2.18.1"})]
|
||||
for label, params, headers in attempts:
|
||||
for _ in range(15):
|
||||
response = self.authed.get("/api/random/", params, **headers)
|
||||
self.assertEqual(response.status_code, 200, label)
|
||||
data = response.json()
|
||||
self.assertIn(data["extension"], FASTFETCH_EXTENSIONS, label)
|
||||
self.assertTrue(data["fastfetch"], label)
|
||||
|
||||
def test_rating_filter(self):
|
||||
self.make_item("safe", "png", "s")
|
||||
explicit = self.make_item("explicit", "png", "e")
|
||||
for _ in range(10):
|
||||
data = self.authed.get("/api/random/", {"rating": "e"}).json()
|
||||
self.assertEqual(data["j_id"], f"J-{explicit.id}")
|
||||
self.assertEqual(data["rating"], "e")
|
||||
self.assertEqual(self.authed.get("/api/random/", {"rating": "q"}).status_code, 404)
|
||||
|
||||
def test_guests_never_receive_hidden_items(self):
|
||||
self.make_item("hidden", "png", "s", hidden=True)
|
||||
self.assertEqual(self.guest.get("/api/random/").status_code, 404)
|
||||
self.assertEqual(self.authed.get("/api/random/").status_code, 200)
|
||||
|
||||
def test_no_match_returns_404(self):
|
||||
self.make_item("movie", "mp4", "s") # images only
|
||||
self.assertEqual(self.authed.get("/api/random/").status_code, 404)
|
||||
|
||||
def test_short_top_level_alias(self):
|
||||
self.make_item("alias", "gif", "s")
|
||||
self.assertEqual(self.guest.get("/random/").status_code, 200)
|
||||
self.assertEqual(self.guest.get("/random").status_code, 200)
|
||||
@@ -17,6 +17,7 @@ from .views import (
|
||||
DownloadTaskViewSet,
|
||||
MatchTaskViewSet,
|
||||
MediaItemViewSet,
|
||||
RandomItemView,
|
||||
)
|
||||
|
||||
router = DefaultRouter()
|
||||
@@ -28,6 +29,7 @@ router.register("similarity", SimilarityCheckViewSet, basename="similarity")
|
||||
|
||||
urlpatterns = [
|
||||
path("", include(router.urls)),
|
||||
path("random/", RandomItemView.as_view(), name="random_item"),
|
||||
path("online/file/", ClientDownloadView.as_view(), name="client_download"),
|
||||
path(
|
||||
"duplicates/md5/",
|
||||
|
||||
@@ -520,6 +520,99 @@ class MatchTaskViewSet(
|
||||
return Response({"success": True})
|
||||
|
||||
|
||||
class RandomItemView(APIView):
|
||||
"""A random library image, optionally filtered by rating.
|
||||
|
||||
Two kinds of clients use this:
|
||||
|
||||
* the SPA's Random page, which renders the returned URL, and
|
||||
* shell greeting scripts (fish_greeting) that fetch the URL with
|
||||
fastfetch in a terminal.
|
||||
|
||||
Fastfetch mode — ``?fastfetch=1`` or a User-Agent containing "fastfetch"
|
||||
— only considers png/jpg/gif files, because that is what those terminals
|
||||
display. Responses always carry a signed absolute URL (minted for the
|
||||
requesting user) so image viewers can load it without auth headers;
|
||||
guests get unsigned URLs for guest-visible items only.
|
||||
"""
|
||||
|
||||
permission_classes = [AllowAny]
|
||||
|
||||
FASTFETCH_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif"}
|
||||
|
||||
def get(self, request):
|
||||
fastfetch = request.query_params.get("fastfetch", "").lower() in {
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
} or "fastfetch" in (request.META.get("HTTP_USER_AGENT") or "").lower()
|
||||
|
||||
extensions = (
|
||||
self.FASTFETCH_EXTENSIONS
|
||||
if fastfetch
|
||||
else services.IMAGE_EXTENSIONS
|
||||
)
|
||||
|
||||
queryset = MediaItem.objects.prefetch_related("locations")
|
||||
if not request.user.is_authenticated:
|
||||
queryset = queryset.filter(hidden_from_guests=False)
|
||||
|
||||
ratings = [
|
||||
value
|
||||
for value in request.query_params.get("rating", "").split(",")
|
||||
if value in {"s", "q", "e"}
|
||||
]
|
||||
if ratings:
|
||||
queryset = queryset.filter(rating__in=ratings)
|
||||
|
||||
# Any copy with an allowed extension qualifies. ORDER BY RAND() is
|
||||
# fine for a personal library (same trade-off as the duplicates page).
|
||||
suffixes = "|".join(extension.lstrip(".") for extension in sorted(extensions))
|
||||
item = (
|
||||
queryset.filter(locations__rel_path__iregex=rf"\.({suffixes})$")
|
||||
.distinct()
|
||||
.order_by("?")
|
||||
.first()
|
||||
)
|
||||
if item is None:
|
||||
return Response(
|
||||
{"detail": "No image matches those filters."},
|
||||
status=status.HTTP_404_NOT_FOUND,
|
||||
)
|
||||
|
||||
location = next(
|
||||
(
|
||||
candidate
|
||||
for candidate in item.locations.all()
|
||||
if Path(candidate.rel_path).suffix.lower() in extensions
|
||||
),
|
||||
item.locations.first(),
|
||||
)
|
||||
url = services.signed_media_url(item, request.user, "raw", request=request)
|
||||
return Response(
|
||||
{
|
||||
"j_id": f"J-{item.id}",
|
||||
"md5": item.md5,
|
||||
"filename": Path(location.rel_path).name if location else item.md5,
|
||||
"extension": (
|
||||
Path(location.rel_path).suffix.lower().lstrip(".")
|
||||
if location
|
||||
else ""
|
||||
),
|
||||
"kind": "image",
|
||||
"rating": item.rating or "",
|
||||
"size": item.size,
|
||||
"e621_post_id": item.e621_post_id,
|
||||
"url": url,
|
||||
"download_url": f"{url}{'&' if '?' in url else '?'}download=1",
|
||||
"thumbnail_url": services.signed_media_url(
|
||||
item, request.user, "thumbnail", request=request
|
||||
),
|
||||
"fastfetch": fastfetch,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class ClientDownloadView(APIView):
|
||||
"""Stream an e621 file straight to the browser (no library write)."""
|
||||
|
||||
@@ -562,7 +655,7 @@ class ClientDownloadView(APIView):
|
||||
"Content-Type", "application/octet-stream"
|
||||
)
|
||||
name = get_valid_filename(
|
||||
filename or Path(parsed.path).name or "download"
|
||||
filename or Path(urlparse(url).path).name or "download"
|
||||
)
|
||||
|
||||
def stream():
|
||||
|
||||
Reference in New Issue
Block a user