From 1c60696ea1accbbfc6dd731979a3557e8ea1066c Mon Sep 17 00:00:00 2001 From: catborise Date: Sat, 24 Oct 2026 23:05:27 +0300 Subject: [PATCH] feat(accounts): API tokens for scripts and tools The REST API took only the login session, so a script could not use it without a browser. A user now makes API tokens on the profile page (API Tokens tab): a name, and an expiry of 30, 90 or 365 days or none. The token is shown once, on a page that is not cached, and then kept only as a SHA-512 digest; the list shows its name, first characters, creation and expiry, and a token is deleted there. Tools send it as "Authorization: Token wvc_...", and act as its user, with that user's permissions. Making and deleting a token is logged, without the token. A password alone (HTTP Basic) still does not reach the API, as it would skip the OTP second factor: a token is made in a session that passed the login and its OTP. The token class comes first, so a refused request gets 401 with its challenge instead of 403. django-rest-knox 5.1.0 provides the tokens (accounts.ApiToken adds the name); settings.py takes "knox", KNOX_TOKEN_MODEL and REST_KNOX from the template. --- README.md | 7 +- accounts/forms.py | 10 +- accounts/migrations/0009_api_token.py | 30 ++++++ accounts/models.py | 10 ++ accounts/templates/api_token_created.html | 16 ++++ accounts/templates/profile.html | 63 +++++++++++++ accounts/tests/test_api_tokens.py | 93 +++++++++++++++++++ accounts/urls.py | 2 + accounts/views.py | 41 +++++++- conf/requirements.txt | 1 + webvirtcloud/settings.py.template | 13 ++- webvirtcloud/tests/test_api_schema.py | 4 +- .../tests/test_authorization_matrix.py | 1 + webvirtcloud/tests/test_otp_bypass.py | 12 ++- 14 files changed, 290 insertions(+), 13 deletions(-) create mode 100644 accounts/migrations/0009_api_token.py create mode 100644 accounts/templates/api_token_created.html create mode 100644 accounts/tests/test_api_tokens.py diff --git a/README.md b/README.md index 0f04e782..bb171f27 100644 --- a/README.md +++ b/README.md @@ -629,11 +629,16 @@ See [doc/permissions.md](doc/permissions.md). ### REST API -The REST API (Django REST Framework, documented with `drf-spectacular`) uses the login session. After logging in: +The REST API (Django REST Framework, documented with `drf-spectacular`) takes the login session or an API token. After logging in: * **Swagger UI:** `http:///swagger/` * **ReDoc UI:** `http:///redoc/` * **OpenAPI 3.0 schema:** `http:///api/schema/` (JSON or YAML) +* **API tokens**, for scripts and tools: **Profile → API Tokens**. A token acts as its user and is shown once; send it as a header: + + ```bash + curl -H "Authorization: Token wvc_..." https:///api/v1/instances/ + ``` ### Screenshots diff --git a/accounts/forms.py b/accounts/forms.py index f458a03a..30f4d8cc 100644 --- a/accounts/forms.py +++ b/accounts/forms.py @@ -2,7 +2,7 @@ from appsettings.settings import app_settings from django.contrib.auth import get_user_model from django.contrib.auth.forms import AuthenticationForm from django.core.validators import MaxLengthValidator -from django.forms import EmailField, Form, ModelForm, ValidationError +from django.forms import CharField, EmailField, Form, ModelForm, TypedChoiceField, ValidationError from django.utils.translation import gettext_lazy as _ from .models import UserInstance, UserSSHKey @@ -74,6 +74,14 @@ class UserSSHKeyForm(ModelForm): fields = ("keyname", "keypublic") +class ApiTokenForm(Form): + name = CharField(label=_("Name"), max_length=64, help_text=_("What uses it, such as a script or a host")) + days = TypedChoiceField( + label=_("Expires"), coerce=int, empty_value=None, required=False, initial=90, + choices=[(30, _("in 30 days")), (90, _("in 90 days")), (365, _("in a year")), ("", _("never"))], + ) + + class EmailOTPForm(Form): email = EmailField(label=_("Email")) diff --git a/accounts/migrations/0009_api_token.py b/accounts/migrations/0009_api_token.py new file mode 100644 index 00000000..66960f32 --- /dev/null +++ b/accounts/migrations/0009_api_token.py @@ -0,0 +1,30 @@ +# Generated by Django 5.2.18 on 2026-10-24 19:58 + +import django.db.models.deletion +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('accounts', '0008_apply_show_profile_edit_password_once'), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.CreateModel( + name='ApiToken', + fields=[ + ('digest', models.CharField(max_length=128, primary_key=True, serialize=False)), + ('token_key', models.CharField(db_index=True, max_length=25)), + ('created', models.DateTimeField(auto_now_add=True)), + ('expiry', models.DateTimeField(blank=True, null=True)), + ('name', models.CharField(max_length=64, verbose_name='name')), + ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='auth_token_set', to=settings.AUTH_USER_MODEL)), + ], + options={ + 'ordering': ['-created'], + }, + ), + ] diff --git a/accounts/models.py b/accounts/models.py index a90b7b56..495a3f4d 100644 --- a/accounts/models.py +++ b/accounts/models.py @@ -3,6 +3,7 @@ from django.core.validators import MinValueValidator from django.db import models from django.utils.translation import gettext_lazy as _ from instances.models import Instance +from knox.models import AbstractAuthToken class UserInstanceManager(models.Manager): @@ -81,3 +82,12 @@ class PermissionSet(models.Model): permissions = (("change_password", _("Can change password")),) managed = False + + +class ApiToken(AbstractAuthToken): + """A token for the REST API, kept as a digest (django-rest-knox); the + name says what uses it.""" + name = models.CharField(_("name"), max_length=64) + + class Meta: + ordering = ["-created"] diff --git a/accounts/templates/api_token_created.html b/accounts/templates/api_token_created.html new file mode 100644 index 00000000..73a6e212 --- /dev/null +++ b/accounts/templates/api_token_created.html @@ -0,0 +1,16 @@ +{% extends "base.html" %} +{% load i18n %} + +{% block title %}{% trans "API token created" %}{% endblock %} +{% block page_heading %}{% trans "API token created" %}: {{ name }}{% endblock page_heading %} + +{% block content %} + + + +

{% trans "Use it like this:" %}

+
curl -H "Authorization: Token {{ key }}" {{ request.scheme }}://{{ request.get_host }}{% url 'instance-list' %}
+{% trans "Back to the profile" %} +{% endblock content %} diff --git a/accounts/templates/profile.html b/accounts/templates/profile.html index 6e1cd5bb..9e861fbd 100644 --- a/accounts/templates/profile.html +++ b/accounts/templates/profile.html @@ -20,6 +20,11 @@ {% trans "SSH Keys" %} +
@@ -80,5 +85,63 @@
+
+

{% blocktrans %}A token lets a script or a tool use the REST API as you, without your password or OTP: send it as Authorization: Token <token>. A password change does not delete tokens: delete a token you no longer use, or one that may have leaked.{% endblocktrans %}

+ {% if api_tokens %} +
+ + + + + + + + + + + + {% for token in api_tokens %} + + + + + + + + {% endfor %} + +
{% trans "Name" %}{% trans "Starts with" %}{% trans "Created" %}{% trans "Expires" %}
{{ token.name }}{{ token.token_key }}{{ token.created|date:"Y-m-d H:i" }}{% if token.expiry %}{{ token.expiry|date:"Y-m-d H:i" }}{% else %}{% trans "never" %}{% endif %} + + {% bs_icon 'trash' %} + +
+
+ {% endif %} +
+
+ {% trans "Create an API token" %} +
+
+
+ {% csrf_token %} + {% bootstrap_form api_token_form layout='horizontal' %} +
+ +
+
+
+
+
+{% endblock %} +{% block script %} + {% endblock %} \ No newline at end of file diff --git a/accounts/tests/test_api_tokens.py b/accounts/tests/test_api_tokens.py new file mode 100644 index 00000000..bcf2314e --- /dev/null +++ b/accounts/tests/test_api_tokens.py @@ -0,0 +1,93 @@ +"""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 accounts.models import ApiToken +from django.contrib.auth import get_user_model +from django.test import TestCase +from django.urls import reverse +from django.utils import timezone +from logs.models import Logs + + +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}) + 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"}) + self.assertIn("no-store", response["Cache-Control"]) + + 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}") diff --git a/accounts/urls.py b/accounts/urls.py index 59d04163..fa948731 100644 --- a/accounts/urls.py +++ b/accounts/urls.py @@ -30,6 +30,8 @@ urlpatterns = [ ), path("ssh_key/create/", views.ssh_key_create, name="ssh_key_create"), path("ssh_key//delete/", views.ssh_key_delete, name="ssh_key_delete"), + path("api_token/create/", views.api_token_create, name="api_token_create"), + path("api_token//delete/", views.api_token_delete, name="api_token_delete"), path("email_otp/", views.email_otp, name="email_otp"), path( "admin_email_otp//", diff --git a/accounts/views.py b/accounts/views.py index 5db96b85..af0aa9b6 100644 --- a/accounts/views.py +++ b/accounts/views.py @@ -1,5 +1,6 @@ import logging import os +from datetime import timedelta from admin.decorators import superuser_only from django.conf import settings @@ -15,14 +16,15 @@ from django.core.cache import cache from django.http import Http404 from django.utils.http import url_has_allowed_host_and_scheme from django.utils.translation import gettext_lazy as _ +from django.views.decorators.cache import never_cache from django.views.decorators.http import require_POST from accounts.apps import admin_password_path from instances.models import Instance from logs.audit import audit, audit_vm, changed_fields, failed, snapshot -from accounts.forms import EmailOTPForm, ProfileForm, UserSSHKeyForm -from accounts.models import User, UserAttributes, UserInstance, UserSSHKey +from accounts.forms import ApiTokenForm, EmailOTPForm, ProfileForm, UserSSHKeyForm +from accounts.models import ApiToken, User, UserAttributes, UserInstance, UserSSHKey from . import forms from .utils import get_existing_totp_device, get_user_totp_device, send_email_with_otp @@ -50,6 +52,41 @@ def profile(request): "publickeys": publickeys, "profile_form": profile_form, "ssh_key_form": ssh_key_form, + "api_tokens": ApiToken.objects.filter(user=request.user), + "api_token_form": ApiTokenForm(), + }, + ) + + +@require_POST +@never_cache +def api_token_create(request): + """A REST API token, shown once: only its digest is kept. It is made in + a session that passed the login, and its OTP when OTP is on.""" + form = ApiTokenForm(request.POST) + if not form.is_valid(): + return render(request, "common/form.html", {"form": form, "title": _("Create an API token")}) + name, days = form.cleaned_data["name"], form.cleaned_data["days"] + with audit(request, None, f"API token {name}", "add"): + _token, key = ApiToken.objects.create(request.user, expiry=timedelta(days=days) if days else None, name=name) + return render(request, "api_token_created.html", {"name": name, "key": key}) + + +def api_token_delete(request, token_key): + token = get_object_or_404(ApiToken, token_key=token_key, user=request.user) + if request.method == "POST": + with audit(request, None, f"API token {token.name}", "delete"): + token.delete() + messages.success(request, _("API token deleted")) + return redirect(reverse("accounts:profile") + "#apitokens") + + return render( + request, + "common/confirm_delete.html", + { + "object": token.name, + "title": _("Delete API token"), + "consequence": _("Scripts and tools that use it can no longer reach the API."), }, ) diff --git a/conf/requirements.txt b/conf/requirements.txt index e0ca505d..2c09946b 100644 --- a/conf/requirements.txt +++ b/conf/requirements.txt @@ -9,6 +9,7 @@ requests>=2.33.0 urllib3>=2.8.0 django-axes==8.3.1 djangorestframework==3.18.3 +django-rest-knox==5.1.0 drf-nested-routers==0.95.3 drf-spectacular[sidecar]==0.30.0 gunicorn==26.2.0 diff --git a/webvirtcloud/settings.py.template b/webvirtcloud/settings.py.template index a5f35510..daa82ed4 100644 --- a/webvirtcloud/settings.py.template +++ b/webvirtcloud/settings.py.template @@ -82,6 +82,7 @@ INSTALLED_APPS = [ "logs", "operations", "rest_framework", + "knox", "axes", ] @@ -362,9 +363,12 @@ OTP_ENABLED = False # REST Framework & OpenAPI (drf-spectacular) REST_FRAMEWORK = { "DEFAULT_SCHEMA_CLASS": "drf_spectacular.openapi.AutoSchema", - # Session only: a password alone (HTTP Basic) must not reach the API, as it - # would skip the OTP second factor the login form enforces. + # An API token, or a session: a password alone (HTTP Basic) must not reach + # the API, as it would skip the OTP second factor the login form enforces; + # tokens are made in a session. The token class comes first, so a refused + # request gets 401 with its challenge, not 403. "DEFAULT_AUTHENTICATION_CLASSES": [ + "knox.auth.TokenAuthentication", "rest_framework.authentication.SessionAuthentication", ], # A view that forgets its own permission_classes is not public. @@ -373,6 +377,11 @@ REST_FRAMEWORK = { ], } +# API tokens (django-rest-knox): kept as SHA-512 digests, sent as +# "Authorization: Token wvc_..."; made and deleted on the profile page +KNOX_TOKEN_MODEL = "accounts.ApiToken" +REST_KNOX = {"TOKEN_PREFIX": "wvc_"} + SPECTACULAR_SETTINGS = { "TITLE": "Webvirtcloud REST-API", "DESCRIPTION": "Webvirtcloud REST API", diff --git a/webvirtcloud/tests/test_api_schema.py b/webvirtcloud/tests/test_api_schema.py index 700a8ed0..ff3ca1cb 100644 --- a/webvirtcloud/tests/test_api_schema.py +++ b/webvirtcloud/tests/test_api_schema.py @@ -26,7 +26,7 @@ class SchemaPermissionTestCase(TestCase): def test_schema_requires_login(self): client = APIClient() for url in SCHEMA_URLS: - self.assertEqual(client.get(url).status_code, 403, url) + self.assertEqual(client.get(url).status_code, 401, url) def test_any_user_may_read_the_schema(self): user = get_user_model().objects.create_user("schema_reader", password="pw") @@ -41,4 +41,4 @@ class SchemaPermissionTestCase(TestCase): return Response({}) response = Probe.as_view()(APIRequestFactory().get("/probe/")) - self.assertEqual(response.status_code, 403) + self.assertEqual(response.status_code, 401) diff --git a/webvirtcloud/tests/test_authorization_matrix.py b/webvirtcloud/tests/test_authorization_matrix.py index d180e1ff..f986ba1e 100644 --- a/webvirtcloud/tests/test_authorization_matrix.py +++ b/webvirtcloud/tests/test_authorization_matrix.py @@ -38,6 +38,7 @@ ALLOWED_FOR_ANY_USER = { "accounts:profile": "the user's own profile", "accounts:change_password": "the user's own password; new users get the permission by default", "accounts:ssh_key_create": "the user's own SSH keys", + "accounts:api_token_create": "the user's own API tokens", "set_language": "UI language switch", "console": "renders an error without a valid token; access is checked per VM", "ds_openstack_index": "cloud-init datasource, fetched by VMs", diff --git a/webvirtcloud/tests/test_otp_bypass.py b/webvirtcloud/tests/test_otp_bypass.py index 805c2299..ce05fdbb 100644 --- a/webvirtcloud/tests/test_otp_bypass.py +++ b/webvirtcloud/tests/test_otp_bypass.py @@ -12,11 +12,13 @@ from rest_framework.settings import api_settings class BasicAuthDisabledTestCase(TestCase): - def test_session_is_the_only_default_authentication(self): - # exactly SessionAuthentication: not Basic (a password-only door) and - # not Token/JWT either, which would be another password-free door - names = [c.__name__ for c in api_settings.DEFAULT_AUTHENTICATION_CLASSES] - self.assertEqual(names, ["SessionAuthentication"]) + def test_a_token_or_a_session_are_the_only_default_authentication(self): + # exactly knox's tokens and sessions: not Basic (a password-only door). + # A token is no other door: it is made in a session that passed the + # login and its OTP (accounts.views.api_token_create) + classes = [f"{c.__module__}.{c.__name__}" for c in api_settings.DEFAULT_AUTHENTICATION_CLASSES] + self.assertEqual(classes, ["knox.auth.TokenAuthentication", + "rest_framework.authentication.SessionAuthentication"]) def test_api_rejects_http_basic_credentials(self): User = get_user_model()