J621/backend/apps/accounts/tests/test_greeting_tokens.py

"""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)