mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 11:16:47 +00:00
c290cef21a
A double click made two tokens: the second key was shown, the first never, yet it stayed valid until it expired and took a place of the ten. A user's token names are now unique, checked under a lock on the user's row, which also keeps two forms sent at once from passing the limit together. tr and zh too.
216 lines
11 KiB
Python
216 lines
11 KiB
Python
"""API tokens: a user makes them on the profile page, in a session that
|
|
passed the login (and its OTP), and scripts use them on the REST API."""
|
|
|
|
import re
|
|
from datetime import timedelta
|
|
from unittest.mock import patch
|
|
|
|
from accounts.models import ApiToken, UserAttributes
|
|
from django.contrib.auth import get_user_model
|
|
from django.db import connection
|
|
from django.test import TestCase
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
from logs.models import Logs
|
|
from lxml import html
|
|
|
|
|
|
class ApiTokenTests(TestCase):
|
|
def setUp(self):
|
|
self.user = get_user_model().objects.create_user("tok-user", password="pw")
|
|
self.other = get_user_model().objects.create_user("tok-other", password="pw")
|
|
|
|
def make(self, name="ansible", days="90"):
|
|
"""The token the creation page shows once."""
|
|
self.client.force_login(self.user)
|
|
response = self.client.post(reverse("accounts:api_token_create"), {"name": name, "days": days}, follow=True)
|
|
self.assertEqual(response.status_code, 200)
|
|
return re.search(r"wvc_[0-9a-f]{64}", response.content.decode()).group(0)
|
|
|
|
def api(self, token):
|
|
self.client.logout()
|
|
return self.client.get(reverse("instance-list"), HTTP_AUTHORIZATION=f"Token {token}")
|
|
|
|
def test_a_token_opens_the_api_without_a_session(self):
|
|
token = self.make()
|
|
response = self.api(token)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json(), []) # the user's own VMs: none
|
|
|
|
def test_a_token_is_kept_only_as_a_hash_and_shown_once(self):
|
|
token = self.make(name="ci")
|
|
(stored,) = ApiToken.objects.filter(user=self.user)
|
|
self.assertEqual((stored.name, stored.token_key), ("ci", token[:15]))
|
|
self.assertNotIn(token, stored.digest)
|
|
page = self.client.get(reverse("accounts:profile")).content.decode()
|
|
self.assertIn("ci", page)
|
|
self.assertNotIn(token, page)
|
|
|
|
def test_the_page_showing_the_token_is_not_cached(self):
|
|
self.client.force_login(self.user)
|
|
response = self.client.post(reverse("accounts:api_token_create"), {"name": "ci", "days": "30"}, follow=True)
|
|
self.assertIn("no-store", response["Cache-Control"])
|
|
|
|
def test_reloading_the_page_creates_no_second_token(self):
|
|
token = self.make(name="ci")
|
|
response = self.client.get(reverse("accounts:api_token_created"))
|
|
self.assertRedirects(response, reverse("accounts:profile") + "#apitokens", fetch_redirect_response=False)
|
|
self.assertNotIn(token, response.content.decode())
|
|
self.assertEqual(ApiToken.objects.filter(user=self.user).count(), 1)
|
|
|
|
def test_a_token_lasts_the_days_chosen_or_until_deleted(self):
|
|
self.make(name="month", days="30")
|
|
self.make(name="always", days="")
|
|
expiry = dict(ApiToken.objects.values_list("name", "expiry"))
|
|
self.assertAlmostEqual(expiry["month"], timezone.now() + timedelta(days=30), delta=timedelta(minutes=1))
|
|
self.assertIsNone(expiry["always"])
|
|
|
|
def test_an_expired_token_or_a_wrong_one_is_refused(self):
|
|
token = self.make()
|
|
ApiToken.objects.update(expiry=timezone.now() - timedelta(seconds=1))
|
|
self.assertEqual(self.api(token).status_code, 401)
|
|
self.assertEqual(self.api("wvc_" + "0" * 64).status_code, 401)
|
|
|
|
def test_an_inactive_users_token_is_refused(self):
|
|
token = self.make()
|
|
self.user.is_active = False
|
|
self.user.save()
|
|
self.assertEqual(self.api(token).status_code, 401)
|
|
|
|
def test_a_deleted_token_stops_working_and_only_its_owner_deletes_it(self):
|
|
token = self.make()
|
|
stored = ApiToken.objects.get(user=self.user)
|
|
url = reverse("accounts:api_token_delete", args=[stored.token_key])
|
|
self.client.force_login(self.other)
|
|
self.assertEqual(self.client.post(url).status_code, 404)
|
|
self.client.force_login(self.user)
|
|
self.assertRedirects(self.client.post(url), reverse("accounts:profile") + "#apitokens",
|
|
fetch_redirect_response=False)
|
|
self.assertEqual(self.api(token).status_code, 401)
|
|
|
|
def test_making_and_deleting_a_token_is_logged_without_the_token(self):
|
|
token = self.make(name="backup")
|
|
stored = ApiToken.objects.get(user=self.user)
|
|
self.client.post(reverse("accounts:api_token_delete", args=[stored.token_key]))
|
|
rows = list(Logs.objects.order_by("id").values_list("message", "success"))
|
|
self.assertEqual(rows[-2:], [("API token backup: add", True), ("API token backup: delete", True)])
|
|
self.assertFalse(Logs.objects.filter(message__contains=token[4:20]).exists())
|
|
|
|
def test_a_token_needs_a_name(self):
|
|
self.client.force_login(self.user)
|
|
response = self.client.post(reverse("accounts:api_token_create"), {"name": " ", "days": "30"})
|
|
self.assertFalse(ApiToken.objects.exists())
|
|
self.assertNotRegex(response.content.decode(), r"wvc_[0-9a-f]{64}")
|
|
|
|
def test_knox_makes_no_table_of_its_own(self):
|
|
# accounts.ApiToken replaces knox's AuthToken; a knox_authtoken table
|
|
# would reference auth_user unseen, and PostgreSQL could not empty
|
|
# auth_user between tests
|
|
self.assertNotIn("knox_authtoken", connection.introspection.table_names())
|
|
def test_a_token_shows_when_it_was_last_used(self):
|
|
token = self.make()
|
|
self.assertIsNone(ApiToken.objects.get().last_used)
|
|
self.api(token)
|
|
first = ApiToken.objects.get().last_used
|
|
self.assertAlmostEqual(first, timezone.now(), delta=timedelta(seconds=5))
|
|
# within a minute a request writes nothing
|
|
self.api(token)
|
|
self.assertEqual(ApiToken.objects.get().last_used, first)
|
|
later = timezone.now() + timedelta(minutes=2)
|
|
with patch("accounts.auth.timezone.now", return_value=later):
|
|
self.api(token)
|
|
self.assertEqual(ApiToken.objects.get().last_used, later)
|
|
self.client.force_login(self.user)
|
|
self.assertIn("Last used", self.client.get(reverse("accounts:profile")).content.decode())
|
|
|
|
def test_a_token_stops_while_its_user_must_change_the_password(self):
|
|
token = self.make()
|
|
UserAttributes.objects.update_or_create(user=self.user, defaults={"must_change_password": True})
|
|
self.assertEqual(self.api(token).status_code, 401)
|
|
UserAttributes.objects.filter(user=self.user).update(must_change_password=False)
|
|
self.assertEqual(self.api(token).status_code, 200)
|
|
|
|
def test_a_user_has_at_most_ten_tokens(self):
|
|
for number in range(10):
|
|
self.make(name=f"t{number}")
|
|
response = self.client.post(reverse("accounts:api_token_create"), {"name": "one more", "days": "30"})
|
|
self.assertEqual(ApiToken.objects.filter(user=self.user).count(), 10)
|
|
self.assertNotRegex(response.content.decode(), r"wvc_[0-9a-f]{64}")
|
|
self.assertIn("at most 10", response.content.decode())
|
|
|
|
def test_a_form_sent_twice_makes_one_token(self):
|
|
self.make("ci")
|
|
response = self.client.post(reverse("accounts:api_token_create"), {"name": "ci", "days": "90"})
|
|
self.assertContains(response, "You have a token named ci: choose another name.")
|
|
self.assertEqual(ApiToken.objects.filter(user=self.user).count(), 1)
|
|
self.make("ci-2") # another name is fine
|
|
self.assertEqual(ApiToken.objects.filter(user=self.user).count(), 2)
|
|
|
|
def test_expired_tokens_do_not_count_toward_the_limit(self):
|
|
for number in range(10):
|
|
self.make(name=f"t{number}")
|
|
ApiToken.objects.filter(name="t0").update(expiry=timezone.now() - timedelta(days=1))
|
|
self.make(name="new")
|
|
self.assertEqual(sorted(ApiToken.objects.values_list("name", flat=True))[:2], ["new", "t1"])
|
|
|
|
|
|
class AdminApiTokenTests(TestCase):
|
|
"""A superuser sees a user's tokens on the user's page and deletes one
|
|
that may have leaked."""
|
|
|
|
def setUp(self):
|
|
self.user = get_user_model().objects.create_user("tok-user", password="pw")
|
|
self.admin = get_user_model().objects.create_superuser("tok-admin", "admin@example.com", "pw")
|
|
self.client.force_login(self.user)
|
|
response = self.client.post(reverse("accounts:api_token_create"), {"name": "ansible", "days": "90"}, follow=True)
|
|
self.token = re.search(r"wvc_[0-9a-f]{64}", response.content.decode()).group(0)
|
|
self.stored = ApiToken.objects.get(user=self.user)
|
|
|
|
def delete_url(self, user):
|
|
return reverse("accounts:account_api_token_delete", args=[user.id, self.stored.token_key])
|
|
|
|
def test_the_users_page_lists_the_tokens_without_them(self):
|
|
self.client.force_login(self.admin)
|
|
page = self.client.get(reverse("accounts:account", args=[self.user.id])).content.decode()
|
|
self.assertIn("ansible", page)
|
|
self.assertIn(self.stored.token_key, page)
|
|
self.assertIn(self.delete_url(self.user), page)
|
|
self.assertNotIn(self.token, page)
|
|
|
|
def test_a_superuser_deletes_a_users_token(self):
|
|
self.client.force_login(self.admin)
|
|
# the confirmation names whose token it is
|
|
self.assertIn("API token ansible of tok-user", self.client.get(self.delete_url(self.user)).content.decode())
|
|
self.assertRedirects(self.client.post(self.delete_url(self.user)),
|
|
reverse("accounts:account", args=[self.user.id]) + "#apitokens",
|
|
fetch_redirect_response=False)
|
|
self.assertFalse(ApiToken.objects.exists())
|
|
self.assertEqual(Logs.objects.order_by("id").values_list("message", "user").last(),
|
|
("API token ansible of tok-user: delete", "tok-admin"))
|
|
self.client.logout()
|
|
self.assertEqual(self.client.get(reverse("instance-list"), HTTP_AUTHORIZATION=f"Token {self.token}").status_code, 401)
|
|
|
|
def test_only_a_superuser_deletes_another_users_token(self):
|
|
other = get_user_model().objects.create_user("tok-other", password="pw")
|
|
self.client.force_login(other)
|
|
self.assertEqual(self.client.post(self.delete_url(self.user)).status_code, 403)
|
|
# nor through the owner's own address: it names the caller's tokens only
|
|
self.assertEqual(self.client.post(reverse("accounts:api_token_delete", args=[self.stored.token_key])).status_code, 404)
|
|
self.client.force_login(self.admin)
|
|
self.assertEqual(self.client.post(self.delete_url(other)).status_code, 404)
|
|
self.assertTrue(ApiToken.objects.exists())
|
|
|
|
def test_the_user_list_counts_each_users_live_tokens(self):
|
|
self.client.force_login(self.user)
|
|
self.client.post(reverse("accounts:api_token_create"), {"name": "old", "days": "30"})
|
|
ApiToken.objects.filter(name="old").update(expiry=timezone.now() - timedelta(days=1))
|
|
self.client.force_login(self.admin)
|
|
page = html.fromstring(self.client.get(reverse("admin:user_list")).content)
|
|
def tokens(name):
|
|
return page.xpath(f"//div[contains(@class, 'list-group-item')][.//a[normalize-space()='{name}']]"
|
|
"//a[contains(@href, '#apitokens')]")
|
|
# the expired one is not counted; the count opens the user's tokens
|
|
(link,) = tokens("tok-user")
|
|
self.assertEqual(link.text_content().strip(), "1 API token")
|
|
self.assertEqual(link.get("href"), reverse("accounts:account", args=[self.user.id]) + "#apitokens")
|
|
self.assertEqual(tokens("tok-admin"), [])
|