Compare commits

..

1 Commits

Author SHA1 Message Date
lebaudantoine 069b3d5b95 🔒️(backend) reject inactive users in resource server backend
The resource server backend returned any user matching the token's
`sub` claim without checking `User.is_active`. The upstream lasuite
backend only validates the token's introspection `active` claim, so a
deactivated Django account kept API access until its token expired.

Raise `SuspiciousOperation` in `get_or_create_user` when the user is
inactive, which the authentication class turns into a 401, consistent
with `BaseJWTAuthentication`. Add unit and end-to-end tests.
2026-09-09 13:47:13 +02:00
12 changed files with 14 additions and 229 deletions
+2 -1
View File
@@ -4,6 +4,7 @@ import copy
from logging import getLogger
from django.conf import settings
from django.contrib.auth.hashers import check_password
from django.core.exceptions import ValidationError
from django.core.validators import validate_email
@@ -73,7 +74,7 @@ class ApplicationViewSet(viewsets.ViewSet):
except models.Application.DoesNotExist as e:
raise drf_exceptions.AuthenticationFailed("Invalid credentials") from e
if not application.check_client_secret(client_secret):
if not check_password(client_secret, application.client_secret):
raise drf_exceptions.AuthenticationFailed("Invalid credentials")
if not application.is_active:
+2 -12
View File
@@ -4,11 +4,9 @@ Core application fields
from logging import getLogger
from django.contrib.auth.hashers import identify_hasher
from django.contrib.auth.hashers import identify_hasher, make_password
from django.db import models
from .hashers import hash_client_secret
logger = getLogger(__name__)
@@ -26,14 +24,6 @@ class SecretField(models.CharField):
secret = getattr(model_instance, self.attname)
if secret.startswith("sha256$"):
logger.debug(
"%s: %s is already hashed with sha256.",
model_instance,
self.attname,
)
return secret
try:
hasher = identify_hasher(secret)
logger.debug(
@@ -46,7 +36,7 @@ class SecretField(models.CharField):
logger.debug(
"%s: %s is not hashed; hashing it now.", model_instance, self.attname
)
hashed_secret = hash_client_secret(secret)
hashed_secret = make_password(secret)
setattr(model_instance, self.attname, hashed_secret)
return hashed_secret
-24
View File
@@ -1,24 +0,0 @@
"""Application secrets only: keep fast hashing out of PASSWORD_HASHERS.
Secrets must be securely randomly generated, not human-chosen.
"""
import hashlib
from django.contrib.auth.hashers import check_password
from django.utils.crypto import constant_time_compare
from django.utils.encoding import force_bytes
def hash_client_secret(raw_secret):
"""Hash a machine-generated application secret without key stretching."""
return f"sha256${hashlib.sha256(force_bytes(raw_secret)).hexdigest()}"
def verify_client_secret(raw_secret, encoded):
"""Verify the application format or a legacy Django password hash."""
if raw_secret is None:
return False
if encoded.startswith("sha256$"):
return constant_time_compare(encoded, hash_client_secret(raw_secret))
return check_password(raw_secret, encoded)
+1 -26
View File
@@ -25,7 +25,7 @@ from django.utils.translation import gettext_lazy as _
from lasuite.tools.email import get_domain_from_email
from timezone_field import TimeZoneField
from . import fields, hashers, utils
from . import fields, utils
from .recording.enums import FileExtension
from .validators import sub_validator
@@ -828,31 +828,6 @@ class Application(BaseModel):
def __str__(self):
return f"{self.name!s}"
def check_client_secret(self, raw_secret):
"""Verify and lazily rehash without overwriting a concurrent rotation."""
original_hash = self.client_secret
if not hashers.verify_client_secret(raw_secret, original_hash):
return False
if original_hash.startswith("sha256$"):
return True
# Fast hashing assumes securely generated, high-entropy secrets.
# APPLICATION_CLIENT_SECRET_LENGTH controls generated length, not randomness.
encoded = hashers.hash_client_secret(raw_secret)
updated = Application.objects.filter(
pk=self.pk, client_secret=original_hash
).update(client_secret=encoded)
if updated:
self.client_secret = encoded
return True
try:
self.refresh_from_db()
except type(self).DoesNotExist:
return False
return hashers.verify_client_secret(raw_secret, self.client_secret)
def can_delegate_email(self, email):
"""Check if this application can delegate the given email."""
@@ -1,146 +0,0 @@
"""Application hashing and migration of existing credentials."""
import hashlib
from unittest import mock
from django.contrib.auth.hashers import check_password, identify_hasher, make_password
from django.db import connection
from django.test.utils import CaptureQueriesContext
from django.utils.crypto import get_random_string
import pytest
from rest_framework.test import APIClient
from core import hashers
from core.factories import ApplicationFactory, UserFactory
from core.models import Application
pytestmark = pytest.mark.django_db
@pytest.mark.parametrize("secret", ["short", "a" * 128, b"byte-secret"])
def test_application_hash(secret):
"""Application hashes verify correctly but are not accepted for user passwords."""
encoded = hashers.hash_client_secret(secret)
raw = secret.encode() if isinstance(secret, str) else secret
assert encoded == f"sha256${hashlib.sha256(raw).hexdigest()}"
assert hashers.verify_client_secret(secret, encoded)
assert not hashers.verify_client_secret("wrong", encoded)
assert not hashers.verify_client_secret(None, encoded)
assert not hashers.verify_client_secret(secret, "sha256$invalid")
assert not check_password(secret, encoded)
with pytest.raises(ValueError):
identify_hasher(encoded)
assert not make_password(raw.decode()).startswith("sha256$")
@pytest.mark.parametrize("algorithm", ["pbkdf2_sha256", "md5"])
def test_token_migrates_legacy_secret_once(algorithm):
"""The same client secret works before and after migration, with no later writes."""
secret = get_random_string(128)
user = UserFactory()
legacy = make_password(secret, hasher=algorithm)
app = ApplicationFactory(client_secret=legacy)
app.refresh_from_db()
assert app.client_secret == legacy
payload = {
"client_id": app.client_id,
"client_secret": secret,
"grant_type": "client_credentials",
"scope": user.email,
}
client = APIClient()
response = client.post(
"/external-api/v1.0/application/token/", payload, format="json"
)
assert response.status_code == 200
app.refresh_from_db()
migrated = app.client_secret
assert migrated == hashers.hash_client_secret(secret)
with CaptureQueriesContext(connection) as queries:
response = client.post(
"/external-api/v1.0/application/token/", payload, format="json"
)
assert response.status_code == 200
assert not any(q["sql"].lstrip().startswith("UPDATE") for q in queries)
app.refresh_from_db()
assert app.client_secret == migrated
def test_wrong_secret_does_not_migrate():
"""Failed authentication leaves a production PBKDF2 hash untouched."""
user = UserFactory()
legacy = make_password(get_random_string(128), hasher="pbkdf2_sha256")
app = ApplicationFactory(client_secret=legacy)
response = APIClient().post(
"/external-api/v1.0/application/token/",
{
"client_id": app.client_id,
"client_secret": "wrong",
"grant_type": "client_credentials",
"scope": user.email,
},
format="json",
)
assert response.status_code == 401
app.refresh_from_db()
assert app.client_secret == legacy
def test_migration_preserves_concurrent_rotation():
"""Migration must not restore a secret rotated after verification."""
secret = get_random_string(128)
app = ApplicationFactory(
client_secret=make_password(secret, hasher="pbkdf2_sha256")
)
replacement = hashers.hash_client_secret(get_random_string(128))
def verify_then_rotate(raw, encoded):
verified = check_password(raw, encoded)
Application.objects.filter(pk=app.pk).update(client_secret=replacement)
return verified
with mock.patch.object(hashers, "check_password", side_effect=verify_then_rotate):
assert app.check_client_secret(secret) is False
app.refresh_from_db()
assert app.client_secret == replacement
def test_migration_preserves_concurrent_migration():
"""Authentication succeeds when another request migrates the same secret."""
secret = get_random_string(128)
app = ApplicationFactory(
client_secret=make_password(secret, hasher="pbkdf2_sha256")
)
migrated = hashers.hash_client_secret(secret)
def verify_then_migrate(raw, encoded):
verified = check_password(raw, encoded)
Application.objects.filter(pk=app.pk).update(client_secret=migrated)
return verified
with mock.patch.object(hashers, "check_password", side_effect=verify_then_migrate):
assert app.check_client_secret(secret) is True
app.refresh_from_db()
assert app.client_secret == migrated
def test_migration_preserves_concurrent_deletion():
"""Authentication fails when the application is deleted after verification."""
secret = get_random_string(128)
app = ApplicationFactory(
client_secret=make_password(secret, hasher="pbkdf2_sha256")
)
def verify_then_delete(raw, encoded):
verified = check_password(raw, encoded)
Application.objects.filter(pk=app.pk).delete()
return verified
with mock.patch.object(hashers, "check_password", side_effect=verify_then_delete):
assert app.check_client_secret(secret) is False
assert not Application.objects.filter(pk=app.pk).exists()
@@ -6,12 +6,12 @@ Unit tests for the Application and ApplicationDomain models
from unittest import mock
from django.contrib.auth.hashers import check_password
from django.core.exceptions import ValidationError
import pytest
from core.factories import ApplicationDomainFactory, ApplicationFactory
from core.hashers import verify_client_secret
from core.models import Application, ApplicationDomain, ApplicationScope
pytestmark = pytest.mark.django_db
@@ -98,8 +98,8 @@ def test_models_application_client_secret_hashed_on_save():
# Secret should be hashed, not plain
assert application.client_secret != plain_secret
# Should verify with the application credential policy
assert verify_client_secret(plain_secret, application.client_secret) is True
# Should verify with check_password
assert check_password(plain_secret, application.client_secret) is True
def test_models_application_client_secret_preserves_existing_hash():
+1 -12
View File
@@ -469,9 +469,6 @@ class Base(Configuration):
# Sentry
SENTRY_DSN = values.Value(None, environ_name="SENTRY_DSN")
SENTRY_TRACES_SAMPLE_RATE = values.FloatValue(
0.0, environ_name="SENTRY_TRACES_SAMPLE_RATE", environ_prefix=None
)
# Easy thumbnails
THUMBNAIL_EXTENSION = "webp"
@@ -1233,14 +1230,7 @@ class Base(Configuration):
dsn=cls.SENTRY_DSN,
environment=cls.__name__.lower(), # build, test, development, production
release=get_release(),
traces_sample_rate=cls.SENTRY_TRACES_SAMPLE_RATE,
integrations=[
DjangoIntegration(
transaction_style="url",
middleware_spans=True,
cache_spans=True,
)
],
integrations=[DjangoIntegration()],
)
sentry_sdk.set_tag("application", "backend")
@@ -1312,7 +1302,6 @@ class Test(Base):
)
PASSWORD_HASHERS = [
"django.contrib.auth.hashers.MD5PasswordHasher",
"django.contrib.auth.hashers.PBKDF2PasswordHasher",
]
USE_SWAGGER = True
EXTERNAL_API_ENABLED = True
+1 -1
View File
@@ -480,7 +480,7 @@
"destination": "Ein neues Dokument wird erstellt auf",
"destinationUnknown": "Ein neues Dokument wird erstellt",
"language": "Meeting-Sprache:",
"recording": "Auch eine Videoaufzeichnung starten"
"recording": "Auch eine Aufzeichnung starten"
},
"button": {
"start": "Meeting-Transkription starten",
+1 -1
View File
@@ -480,7 +480,7 @@
"destination": "A new document will be created on",
"destinationUnknown": "A new document will be created",
"language": "Meeting language:",
"recording": "Also start a video recording"
"recording": "Also start a recording"
},
"button": {
"start": "Start transcribing the meeting",
+1 -1
View File
@@ -479,7 +479,7 @@
"destination": "Se creará un nuevo documento en",
"destinationUnknown": "Se creará un nuevo documento",
"language": "Idioma de la reunión:",
"recording": "Iniciar también una grabación de vídeo"
"recording": "Iniciar también una grabación"
},
"button": {
"start": "Empezar a transcribir la reunión",
+1 -1
View File
@@ -480,7 +480,7 @@
"destination": "Un nouveau document sera créé sur",
"destinationUnknown": "Un nouveau document sera créé",
"language": "Langue de la réunion :",
"recording": "Démarrer aussi un enregistrement vidéo"
"recording": "Démarrer aussi un enregistrement"
},
"button": {
"start": "Commencer à transcrire la réunion",
+1 -1
View File
@@ -480,7 +480,7 @@
"destination": "Er wordt een nieuw document aangemaakt op",
"destinationUnknown": "Een nieuw document wordt aangemaakt",
"language": "Vergadertalen:",
"recording": "Start ook een video-opname"
"recording": "Start ook een opname"
},
"button": {
"start": "Begin met het transcriberen van de vergadering",