mirror of
https://github.com/suitenumerique/meet.git
synced 2026-09-11 13:39:03 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 016221638e |
@@ -4,7 +4,6 @@ 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
|
||||
|
||||
@@ -74,7 +73,7 @@ class ApplicationViewSet(viewsets.ViewSet):
|
||||
except models.Application.DoesNotExist as e:
|
||||
raise drf_exceptions.AuthenticationFailed("Invalid credentials") from e
|
||||
|
||||
if not check_password(client_secret, application.client_secret):
|
||||
if not application.check_client_secret(client_secret):
|
||||
raise drf_exceptions.AuthenticationFailed("Invalid credentials")
|
||||
|
||||
if not application.is_active:
|
||||
|
||||
@@ -4,9 +4,11 @@ Core application fields
|
||||
|
||||
from logging import getLogger
|
||||
|
||||
from django.contrib.auth.hashers import identify_hasher, make_password
|
||||
from django.contrib.auth.hashers import identify_hasher
|
||||
from django.db import models
|
||||
|
||||
from .hashers import hash_client_secret
|
||||
|
||||
logger = getLogger(__name__)
|
||||
|
||||
|
||||
@@ -24,6 +26,14 @@ 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(
|
||||
@@ -36,7 +46,7 @@ class SecretField(models.CharField):
|
||||
logger.debug(
|
||||
"%s: %s is not hashed; hashing it now.", model_instance, self.attname
|
||||
)
|
||||
hashed_secret = make_password(secret)
|
||||
hashed_secret = hash_client_secret(secret)
|
||||
setattr(model_instance, self.attname, hashed_secret)
|
||||
return hashed_secret
|
||||
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
"""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)
|
||||
@@ -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, utils
|
||||
from . import fields, hashers, utils
|
||||
from .recording.enums import FileExtension
|
||||
from .validators import sub_validator
|
||||
|
||||
@@ -828,6 +828,31 @@ 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."""
|
||||
|
||||
|
||||
@@ -0,0 +1,146 @@
|
||||
"""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 check_password
|
||||
assert check_password(plain_secret, application.client_secret) is True
|
||||
# Should verify with the application credential policy
|
||||
assert verify_client_secret(plain_secret, application.client_secret) is True
|
||||
|
||||
|
||||
def test_models_application_client_secret_preserves_existing_hash():
|
||||
|
||||
@@ -1312,6 +1312,7 @@ class Test(Base):
|
||||
)
|
||||
PASSWORD_HASHERS = [
|
||||
"django.contrib.auth.hashers.MD5PasswordHasher",
|
||||
"django.contrib.auth.hashers.PBKDF2PasswordHasher",
|
||||
]
|
||||
USE_SWAGGER = True
|
||||
EXTERNAL_API_ENABLED = True
|
||||
|
||||
Reference in New Issue
Block a user