Scoped API tokens for the random endpoint, with a management page
Backend: a GreetingToken model stores only a SHA-256 hash of a j621r_…
key (shown once at creation) plus label, prefix, created/last-used. A
dedicated GreetingTokenAuthentication understands the usual
'Authorization: Token …' header but is registered only on RandomItemView
(alongside the normal token auth), so a greeting token authenticates
/api/random/ and is rejected with 401 everywhere else — exactly the scope
shell greetings need. Endpoints: GET/POST /api/auth/greeting-tokens/ and
DELETE /api/auth/greeting-tokens/{id}/ (own tokens only; the list never
returns keys or hashes).
Frontend: /tokens page (Account → Shell tokens card, command palette entry)
lists tokens with label, prefix, created/last-used and revoke (shared
confirm dialog). Creating one shows the key with Copy and 'Copy for fish'
buttons plus a pointer to extras/fish_greeting.
Tests: apps/accounts/tests/test_greeting_tokens.py — 9 tests covering
create-once semantics and hashing, hidden keys in listings, the scope
guarantee (random 200 with a signed URL; 401 on files, storage, me, tags
cloud, delete and the token list itself), unknown/revoked keys, cross-user
revocation, last-used tracking and label limits.
Verified live: created a token, rolled /random (signed URL), got 401 from
four other endpoints, saw the list omit secrets, revoked it (204) and the
same key then 401'd on /random. Full suite: 39 tests green.
This commit is contained in:
@@ -0,0 +1,42 @@
|
||||
"""Authentication for scope-limited bearer tokens.
|
||||
|
||||
`GreetingTokenAuthentication` understands the same header a normal API token
|
||||
uses (``Authorization: Token <key>``) but only resolves tokens issued for the
|
||||
random-image endpoint. It is registered per-view (currently only
|
||||
`RandomItemView`), so a greeting token is rejected everywhere else by the
|
||||
regular DRF token authentication.
|
||||
"""
|
||||
|
||||
from rest_framework import authentication, exceptions
|
||||
|
||||
from .models import GreetingToken
|
||||
|
||||
|
||||
class GreetingTokenAuthentication(authentication.BaseAuthentication):
|
||||
keyword = b"token"
|
||||
|
||||
def authenticate_header(self, request):
|
||||
# DRF answers 401 (instead of 403) for AuthenticationFailed only when
|
||||
# the first authenticator can name the scheme.
|
||||
return "Token"
|
||||
|
||||
def authenticate(self, request):
|
||||
header = authentication.get_authorization_header(request).split()
|
||||
if not header or header[0].lower() != self.keyword:
|
||||
return None
|
||||
if len(header) != 2:
|
||||
raise exceptions.AuthenticationFailed("Invalid token header.")
|
||||
try:
|
||||
key = header[1].decode()
|
||||
except UnicodeError:
|
||||
raise exceptions.AuthenticationFailed("Invalid token header.")
|
||||
|
||||
# Not one of ours: let the regular token authentication handle it.
|
||||
if not key.startswith(GreetingToken.PREFIX):
|
||||
return None
|
||||
|
||||
token = GreetingToken.resolve(key)
|
||||
if token is None:
|
||||
raise exceptions.AuthenticationFailed("Invalid token.")
|
||||
token.touch()
|
||||
return (token.user, token)
|
||||
@@ -0,0 +1,30 @@
|
||||
# Generated by Django 6.1.1 on 2026-09-18 18:25
|
||||
|
||||
import django.db.models.deletion
|
||||
from django.conf import settings
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('accounts', '0006_encrypt_e621_api_keys'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='GreetingToken',
|
||||
fields=[
|
||||
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('key_hash', models.CharField(max_length=64, unique=True)),
|
||||
('prefix', models.CharField(max_length=16)),
|
||||
('label', models.CharField(blank=True, default='', max_length=100)),
|
||||
('created_at', models.DateTimeField(auto_now_add=True)),
|
||||
('last_used_at', models.DateTimeField(blank=True, null=True)),
|
||||
('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='greeting_tokens', to=settings.AUTH_USER_MODEL)),
|
||||
],
|
||||
options={
|
||||
'ordering': ['-created_at'],
|
||||
},
|
||||
),
|
||||
]
|
||||
@@ -1,5 +1,9 @@
|
||||
import secrets
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib.auth.models import AbstractUser
|
||||
from django.db import models
|
||||
from django.utils import timezone
|
||||
|
||||
|
||||
class User(AbstractUser):
|
||||
@@ -50,3 +54,63 @@ class User(AbstractUser):
|
||||
return bool(
|
||||
self.is_superuser or self.is_staff or self.role == self.ROLE_STAFF
|
||||
)
|
||||
|
||||
|
||||
def hash_bearer_token(value):
|
||||
"""SHA-256 of a high-entropy bearer token (no salt needed)."""
|
||||
import hashlib
|
||||
|
||||
return hashlib.sha256(value.encode()).hexdigest()
|
||||
|
||||
|
||||
class GreetingToken(models.Model):
|
||||
"""Long-lived token that only authenticates the random-image endpoint.
|
||||
|
||||
Meant for shell greetings and similar scripts, so it is safe to keep in a
|
||||
config file: it cannot read the library, upload, or touch an account. Only
|
||||
the SHA-256 hash is stored; the plaintext is returned once at creation.
|
||||
"""
|
||||
|
||||
PREFIX = "j621r_"
|
||||
|
||||
user = models.ForeignKey(
|
||||
settings.AUTH_USER_MODEL,
|
||||
on_delete=models.CASCADE,
|
||||
related_name="greeting_tokens",
|
||||
)
|
||||
key_hash = models.CharField(max_length=64, unique=True)
|
||||
prefix = models.CharField(max_length=16)
|
||||
label = models.CharField(max_length=100, blank=True, default="")
|
||||
created_at = models.DateTimeField(auto_now_add=True)
|
||||
last_used_at = models.DateTimeField(null=True, blank=True)
|
||||
|
||||
class Meta:
|
||||
ordering = ["-created_at"]
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.prefix}… ({self.user})"
|
||||
|
||||
@classmethod
|
||||
def issue(cls, user, label=""):
|
||||
"""Create a token and return ``(token, plaintext_key)``."""
|
||||
key = cls.PREFIX + secrets.token_hex(20)
|
||||
token = cls.objects.create(
|
||||
user=user,
|
||||
key_hash=hash_bearer_token(key),
|
||||
prefix=key[:12],
|
||||
label=label.strip()[:100],
|
||||
)
|
||||
return token, key
|
||||
|
||||
@classmethod
|
||||
def resolve(cls, key):
|
||||
if not key.startswith(cls.PREFIX):
|
||||
return None
|
||||
return (
|
||||
cls.objects.select_related("user")
|
||||
.filter(key_hash=hash_bearer_token(key))
|
||||
.first()
|
||||
)
|
||||
|
||||
def touch(self):
|
||||
GreetingToken.objects.filter(pk=self.pk).update(last_used_at=timezone.now())
|
||||
|
||||
@@ -6,7 +6,7 @@ from rest_framework import serializers
|
||||
from apps.library.services import VIDEO_EXTENSIONS
|
||||
from apps.library.services import signed_media_url as signed_library_url
|
||||
|
||||
from .models import User
|
||||
from .models import User, GreetingToken
|
||||
|
||||
|
||||
def signed_media_url(request, item):
|
||||
@@ -92,6 +92,13 @@ class UserUpdateSerializer(serializers.Serializer):
|
||||
role = serializers.ChoiceField(choices=User.ROLE_CHOICES, required=False)
|
||||
|
||||
|
||||
class GreetingTokenSerializer(serializers.ModelSerializer):
|
||||
class Meta:
|
||||
model = GreetingToken
|
||||
fields = ["id", "prefix", "label", "created_at", "last_used_at"]
|
||||
read_only_fields = fields
|
||||
|
||||
|
||||
class RegisterSerializer(serializers.ModelSerializer):
|
||||
password = serializers.CharField(write_only=True, validators=[validate_password])
|
||||
|
||||
|
||||
@@ -0,0 +1,166 @@
|
||||
"""Scope-limited greeting tokens (`j621r_…`).
|
||||
|
||||
They exist so shell greetings and scripts can hold a credential that only
|
||||
authenticates `/api/random/` — everything else must reject them — and they
|
||||
are stored hashed, shown once.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
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.accounts.models import GreetingToken, hash_bearer_token
|
||||
from apps.library.models import MediaItem, MediaLocation
|
||||
|
||||
User = get_user_model()
|
||||
|
||||
|
||||
def jpost(client, path, body=None):
|
||||
return client.post(path, data=json.dumps(body or {}), content_type="application/json")
|
||||
|
||||
|
||||
class GreetingTokenTests(TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
super().setUpClass()
|
||||
cls._tmp = tempfile.mkdtemp(prefix="j621-tokens-")
|
||||
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="token-user", password="token-pass-123456"
|
||||
)
|
||||
self.other = User.objects.create_user(
|
||||
username="token-other", password="token-pass-123456"
|
||||
)
|
||||
self.client = self.api_client(self.user)
|
||||
self.other_client = self.api_client(self.other)
|
||||
|
||||
# One image so the random endpoint can answer.
|
||||
path = self._watched / "token-test.png"
|
||||
path.write_bytes(b"token-test")
|
||||
self.item = MediaItem.objects.create(
|
||||
md5=hashlib.md5(b"token-test").hexdigest(),
|
||||
size=path.stat().st_size,
|
||||
rating="s",
|
||||
uploaded_by=self.user,
|
||||
)
|
||||
MediaLocation.objects.create(
|
||||
item=self.item, path=str(path), rel_path=path.name, mtime=time.time()
|
||||
)
|
||||
|
||||
def api_client(self, user):
|
||||
client = Client()
|
||||
client.defaults["HTTP_AUTHORIZATION"] = (
|
||||
f"Token {Token.objects.create(user=user).key}"
|
||||
)
|
||||
return client
|
||||
|
||||
def token_client(self, key):
|
||||
client = Client()
|
||||
client.defaults["HTTP_AUTHORIZATION"] = f"Token {key}"
|
||||
return client
|
||||
|
||||
def issue(self, client=None, label=""):
|
||||
response = jpost(client or self.client, "/api/auth/greeting-tokens/", {"label": label})
|
||||
self.assertEqual(response.status_code, 201)
|
||||
return response.json()
|
||||
|
||||
def test_create_returns_the_key_once_and_stores_only_a_hash(self):
|
||||
created = self.issue(self.client, "shell")
|
||||
key = created["key"]
|
||||
self.assertTrue(key.startswith("j621r_"))
|
||||
self.assertEqual(created["label"], "shell")
|
||||
token = GreetingToken.objects.get(pk=created["id"])
|
||||
self.assertEqual(token.key_hash, hash_bearer_token(key))
|
||||
self.assertNotIn(key, token.key_hash)
|
||||
self.assertEqual(token.prefix, key[:12])
|
||||
self.assertIsNone(token.last_used_at)
|
||||
|
||||
def test_list_hides_keys_and_hashes(self):
|
||||
self.issue(self.client, "one")
|
||||
self.issue(self.client, "two")
|
||||
rows = self.client.get("/api/auth/greeting-tokens/").json()
|
||||
self.assertEqual([row["label"] for row in rows], ["two", "one"])
|
||||
for row in rows:
|
||||
self.assertNotIn("key", row)
|
||||
self.assertNotIn("key_hash", row)
|
||||
self.assertTrue(row["prefix"].startswith("j621r_"))
|
||||
|
||||
def test_token_authenticates_random_and_nothing_else(self):
|
||||
key = self.issue(self.other_client, "shell")["key"]
|
||||
client = self.token_client(key)
|
||||
|
||||
response = client.get("/api/random/")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("sig=", response.json()["url"])
|
||||
|
||||
for method, path, body in (
|
||||
("get", "/api/files/", None),
|
||||
("get", "/api/storage/", None),
|
||||
("get", "/api/auth/me/", None),
|
||||
("get", "/api/tags/cloud/", None),
|
||||
("post", "/api/delete/", {"j_ids": []}),
|
||||
("get", "/api/auth/greeting-tokens/", None),
|
||||
):
|
||||
call = getattr(client, method)
|
||||
if body is None:
|
||||
self.assertEqual(call(path).status_code, 401, path)
|
||||
else:
|
||||
self.assertEqual(jpost(client, path, body).status_code, 401, path)
|
||||
|
||||
def test_normal_api_token_still_authenticates_random(self):
|
||||
response = self.client.get("/api/random/")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertIn("sig=", response.json()["url"])
|
||||
|
||||
def test_unknown_greeting_key_is_rejected(self):
|
||||
client = self.token_client("j621r_" + "0" * 40)
|
||||
self.assertEqual(client.get("/api/random/").status_code, 401)
|
||||
|
||||
def test_revoked_token_stops_working(self):
|
||||
created = self.issue(self.client, "temporary")
|
||||
client = self.token_client(created["key"])
|
||||
self.assertEqual(client.get("/api/random/").status_code, 200)
|
||||
|
||||
response = self.client.delete(f"/api/auth/greeting-tokens/{created['id']}/")
|
||||
self.assertEqual(response.status_code, 204)
|
||||
self.assertEqual(client.get("/api/random/").status_code, 401)
|
||||
self.assertFalse(GreetingToken.objects.filter(pk=created["id"]).exists())
|
||||
|
||||
def test_cannot_revoke_someone_elses_token(self):
|
||||
created = self.issue(self.other_client, "theirs")
|
||||
response = self.client.delete(f"/api/auth/greeting-tokens/{created['id']}/")
|
||||
self.assertEqual(response.status_code, 404)
|
||||
self.assertEqual(self.token_client(created["key"]).get("/api/random/").status_code, 200)
|
||||
|
||||
def test_last_used_is_recorded(self):
|
||||
created = self.issue(self.client, "used")
|
||||
self.token_client(created["key"]).get("/api/random/")
|
||||
token = GreetingToken.objects.get(pk=created["id"])
|
||||
self.assertIsNotNone(token.last_used_at)
|
||||
|
||||
def test_long_labels_are_rejected(self):
|
||||
response = jpost(
|
||||
self.client, "/api/auth/greeting-tokens/", {"label": "x" * 101}
|
||||
)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
@@ -3,6 +3,8 @@ from django.urls import path
|
||||
from .views import (
|
||||
AvatarView,
|
||||
E621CredentialsView,
|
||||
GreetingTokenDetailView,
|
||||
GreetingTokenListView,
|
||||
LoginView,
|
||||
LogoutView,
|
||||
MeView,
|
||||
@@ -18,4 +20,14 @@ urlpatterns = [
|
||||
path("avatar/", AvatarView.as_view(), name="avatar"),
|
||||
path("preferences/", PreferencesView.as_view(), name="preferences"),
|
||||
path("e621/", E621CredentialsView.as_view(), name="e621_credentials"),
|
||||
path(
|
||||
"greeting-tokens/",
|
||||
GreetingTokenListView.as_view(),
|
||||
name="greeting_tokens",
|
||||
),
|
||||
path(
|
||||
"greeting-tokens/<int:pk>/",
|
||||
GreetingTokenDetailView.as_view(),
|
||||
name="greeting_token",
|
||||
),
|
||||
]
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import logging
|
||||
|
||||
from django.http import Http404
|
||||
from rest_framework import mixins, status, viewsets
|
||||
from rest_framework.authtoken.models import Token
|
||||
from rest_framework.authtoken.views import ObtainAuthToken
|
||||
@@ -13,9 +14,10 @@ from apps.core.permissions import IsAppStaff
|
||||
from apps.library.models import MediaItem
|
||||
|
||||
from .crypto import encrypt_secret
|
||||
from .models import User
|
||||
from .models import GreetingToken, User
|
||||
from .serializers import (
|
||||
E621CredentialsSerializer,
|
||||
GreetingTokenSerializer,
|
||||
PreferencesSerializer,
|
||||
RegisterSerializer,
|
||||
UserListSerializer,
|
||||
@@ -157,6 +159,52 @@ class PreferencesView(APIView):
|
||||
return Response(preferences)
|
||||
|
||||
|
||||
class GreetingTokenListView(APIView):
|
||||
"""List and create the caller's random-endpoint tokens.
|
||||
|
||||
The plaintext key is returned once on creation; only its hash is stored,
|
||||
and it only authenticates `/api/random/` (see GreetingToken).
|
||||
"""
|
||||
|
||||
permission_classes = [IsAuthenticated]
|
||||
|
||||
def get(self, request):
|
||||
tokens = GreetingToken.objects.filter(user=request.user)
|
||||
return Response(GreetingTokenSerializer(tokens, many=True).data)
|
||||
|
||||
def post(self, request):
|
||||
label = str(request.data.get("label") or "").strip()
|
||||
if len(label) > 100:
|
||||
return Response(
|
||||
{"detail": "Label is too long (100 characters max)."},
|
||||
status=status.HTTP_400_BAD_REQUEST,
|
||||
)
|
||||
token, key = GreetingToken.issue(request.user, label)
|
||||
logger.info(
|
||||
"Greeting token %s created by %s", token.prefix, request.user.username
|
||||
)
|
||||
return Response(
|
||||
{**GreetingTokenSerializer(token).data, "key": key},
|
||||
status=status.HTTP_201_CREATED,
|
||||
)
|
||||
|
||||
|
||||
class GreetingTokenDetailView(APIView):
|
||||
"""Revoke one of the caller's tokens."""
|
||||
|
||||
permission_classes = [IsAuthenticated]
|
||||
|
||||
def delete(self, request, pk):
|
||||
token = GreetingToken.objects.filter(pk=pk, user=request.user).first()
|
||||
if token is None:
|
||||
raise Http404
|
||||
logger.info(
|
||||
"Greeting token %s revoked by %s", token.prefix, request.user.username
|
||||
)
|
||||
token.delete()
|
||||
return Response(status=status.HTTP_204_NO_CONTENT)
|
||||
|
||||
|
||||
class UserViewSet(
|
||||
mixins.ListModelMixin,
|
||||
mixins.RetrieveModelMixin,
|
||||
|
||||
Reference in New Issue
Block a user