diff --git a/CHANGELOG.md b/CHANGELOG.md index ea36375fe..88ac5a5cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,10 @@ and this project adheres to ## [Unreleased] +### Changed + +- ⚡️(backend) hash application secrets with SHA-256 + ## [1.34.0] - 2026-10-07 ### Added diff --git a/src/backend/core/external_api/viewsets.py b/src/backend/core/external_api/viewsets.py index d61d095fa..545ebec5a 100644 --- a/src/backend/core/external_api/viewsets.py +++ b/src/backend/core/external_api/viewsets.py @@ -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: diff --git a/src/backend/core/fields.py b/src/backend/core/fields.py index 2b3d6ddc5..4c165397f 100644 --- a/src/backend/core/fields.py +++ b/src/backend/core/fields.py @@ -7,6 +7,8 @@ from logging import getLogger from django.contrib.auth.hashers import identify_hasher, make_password from django.db import models +from .hashers import CLIENT_SECRET_HASH_PATTERN + logger = getLogger(__name__) @@ -24,6 +26,14 @@ class SecretField(models.CharField): secret = getattr(model_instance, self.attname) + if CLIENT_SECRET_HASH_PATTERN.fullmatch(secret): + logger.debug( + "%s: %s is already hashed with sha256.", + model_instance, + self.attname, + ) + return secret + try: hasher = identify_hasher(secret) logger.debug( diff --git a/src/backend/core/hashers.py b/src/backend/core/hashers.py new file mode 100644 index 000000000..cd42dff70 --- /dev/null +++ b/src/backend/core/hashers.py @@ -0,0 +1,46 @@ +"""Application secrets only: keep fast hashing out of PASSWORD_HASHERS. + +Secrets must be securely randomly generated, not human-chosen. +""" + +import hashlib +import re + +from django.contrib.auth.hashers import check_password +from django.utils.crypto import constant_time_compare +from django.utils.encoding import force_bytes + +CLIENT_SECRET_HASH_ALGORITHM = "sha256" # noqa: S105 +CLIENT_SECRET_HASH_VERSION = "v0" # noqa: S105 +CLIENT_SECRET_HASH_PREFIX = ( + f"{CLIENT_SECRET_HASH_ALGORITHM}${CLIENT_SECRET_HASH_VERSION}$" +) + +# Accept only the versioned format: sha256$v0$. +CLIENT_SECRET_HASH_PATTERN = re.compile( + rf"{re.escape(CLIENT_SECRET_HASH_PREFIX)}(?P[0-9a-f]{{64}})" +) + + +def _digest(raw_secret): + """Return the hex SHA-256 digest of a raw secret.""" + return hashlib.sha256(force_bytes(raw_secret)).hexdigest() + + +def hash_client_secret(raw_secret): + """Hash a machine-generated application secret without key stretching.""" + return f"{CLIENT_SECRET_HASH_PREFIX}{_digest(raw_secret)}" + + +def verify_client_secret(raw_secret, encoded): + """Verify the versioned application format or a legacy Django password hash.""" + if raw_secret is None: + return False + + match = CLIENT_SECRET_HASH_PATTERN.fullmatch(encoded) + + # Legacy path + if not match: + return check_password(raw_secret, encoded) + + return constant_time_compare(match["digest"], _digest(raw_secret)) diff --git a/src/backend/core/migrations/0025_application_client_secret_sha256.py b/src/backend/core/migrations/0025_application_client_secret_sha256.py new file mode 100644 index 000000000..bf13b1165 --- /dev/null +++ b/src/backend/core/migrations/0025_application_client_secret_sha256.py @@ -0,0 +1,19 @@ +"""Add a separate fast hash while preserving legacy credentials for rollback.""" + +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("core", "0024_room_last_started_at"), + ] + + operations = [ + migrations.AddField( + model_name="application", + name="client_secret_sha256", + field=models.CharField( + max_length=255, null=True, blank=True + ), + ), + ] diff --git a/src/backend/core/models.py b/src/backend/core/models.py index 8fe519400..8d4d13a3e 100644 --- a/src/backend/core/models.py +++ b/src/backend/core/models.py @@ -14,6 +14,7 @@ from typing import List, Optional from django.conf import settings from django.contrib.auth import models as auth_models from django.contrib.auth.base_user import AbstractBaseUser +from django.contrib.auth.hashers import identify_hasher from django.contrib.postgres.fields import ArrayField from django.core import mail, validators from django.core.exceptions import PermissionDenied, ValidationError @@ -25,7 +26,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 @@ -824,6 +825,9 @@ class Application(BaseModel): default=utils.generate_client_secret, help_text=_("Hashed on Save. Copy it now if this is a new secret."), ) + client_secret_sha256 = models.CharField( + max_length=255, null=True, blank=True, editable=False + ) scopes = ArrayField( models.CharField(max_length=50, choices=ApplicationScope.choices), default=list, @@ -839,6 +843,63 @@ class Application(BaseModel): def __str__(self): return f"{self.name!s}" + def save(self, *args, **kwargs): + """Populate the fast hash on creation when the raw secret is available.""" + if self._state.adding: + # Prevent hashing an existing hash instead of the original secret + try: + if not hashers.CLIENT_SECRET_HASH_PATTERN.fullmatch(self.client_secret): + identify_hasher(self.client_secret) + except ValueError: + # SecretField.pre_save hashes the legacy field after this method + self.client_secret_sha256 = hashers.hash_client_secret( + self.client_secret + ) + + return super().save(*args, **kwargs) + + def rotate_client_secret(self): + """Persist a new generated secret and return its raw value to the caller. + + This is the only supported rotation path while both credential fields coexist. + Direct writes may leave a stale fast hash that still accepts the revoked secret, + while saving a stale instance may restore previous credentials. + + This transitional risk is accepted until the legacy field is removed + and rotation writes only the fast hash. + """ + secret = utils.generate_client_secret() + self.client_secret = secret + self.client_secret_sha256 = hashers.hash_client_secret(secret) + self.save(update_fields=["client_secret", "client_secret_sha256"]) + return secret + + def check_client_secret(self, raw_secret): + """Verify the secret and lazily populate its fast hash for future logins.""" + if self.client_secret_sha256 is not None: + return hashers.verify_client_secret(raw_secret, self.client_secret_sha256) + + original_hash = self.client_secret + if not hashers.verify_client_secret(raw_secret, original_hash): + return False + + encoded = hashers.hash_client_secret(raw_secret) + updated = Application.objects.filter( + pk=self.pk, client_secret=original_hash, client_secret_sha256__isnull=True + ).update(client_secret_sha256=encoded) + + if updated: + self.client_secret_sha256 = encoded + return True + + try: + self.refresh_from_db() + except Application.DoesNotExist: + return False + + current_hash = self.client_secret_sha256 or self.client_secret + return hashers.verify_client_secret(raw_secret, current_hash) + def can_delegate_email(self, email): """Check if this application can delegate the given email.""" diff --git a/src/backend/core/tests/test_application_secret_hashers.py b/src/backend/core/tests/test_application_secret_hashers.py new file mode 100644 index 000000000..5bc26b49e --- /dev/null +++ b/src/backend/core/tests/test_application_secret_hashers.py @@ -0,0 +1,350 @@ +"""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 + algorithm, version, digest = encoded.split("$") + assert algorithm == "sha256" + assert version == "v0" + assert digest == hashlib.sha256(raw).hexdigest() + assert hashers.hash_client_secret(secret) == encoded + 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 hashers.verify_client_secret(secret, "sha256$v1$" + digest) + 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 + assert app.client_secret_sha256 is None + 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_sha256 + assert check_password(secret, app.client_secret) + assert hashers.CLIENT_SECRET_HASH_PATTERN.fullmatch(migrated)["digest"] + assert hashers.verify_client_secret(secret, migrated) + 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_sha256 == migrated + assert app.client_secret == legacy + + +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 + assert app.client_secret_sha256 is None + + +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 = make_password(get_random_string(128), hasher="pbkdf2_sha256") + + 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 + assert app.client_secret_sha256 is None + + +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_sha256=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_sha256 == 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() + + +@pytest.mark.parametrize( + "secret", + [ + "sha256$my-secret", + "sha256$" + "a" * 63, + "sha256$" + "a" * 64, + "sha256$" + "g" * 64, + "sha256$" + "a" * 64 + "\n", + "sha256$$" + "a" * 64, + "sha256$short$" + "a" * 64, + "sha256$" + "b" * 22 + "$" + "g" * 64, + "sha256$" + "b" * 22 + "$" + "a" * 64, + "sha256$v0$" + "g" * 64, + "sha256$v0$" + "a" * 63, + "sha256$v1$" + "a" * 64, + ], +) +def test_prefixed_plaintext_is_hashed(secret): + """A prefix alone must not cause a raw secret to bypass hashing.""" + assert not hashers.CLIENT_SECRET_HASH_PATTERN.fullmatch(secret) + app = ApplicationFactory(client_secret=secret) + app.refresh_from_db() + encoded = app.client_secret_sha256 + assert encoded != secret + assert hashers.CLIENT_SECRET_HASH_PATTERN.fullmatch(encoded)["digest"] + assert app.check_client_secret(secret) + app.name = "Updated application" + app.save() + app.refresh_from_db() + assert app.client_secret_sha256 == encoded + + +def test_unsalted_secret_is_rejected(): + """Only salted SHA-256 hashes are accepted.""" + secret = get_random_string(128) + encoded = f"sha256${hashlib.sha256(secret.encode()).hexdigest()}" + + assert hashers.CLIENT_SECRET_HASH_PATTERN.fullmatch(encoded) is None + assert not hashers.verify_client_secret(secret, encoded) + + +def test_new_application_supports_legacy_verification(settings): + """A rollback can authenticate applications created by the new release.""" + settings.PASSWORD_HASHERS = [ + "django.contrib.auth.hashers.PBKDF2PasswordHasher", + ] + secret = get_random_string(128) + app = ApplicationFactory(client_secret=secret) + app.refresh_from_db() + assert app.client_secret.startswith("pbkdf2_sha256$") + assert check_password(secret, app.client_secret) + assert hashers.verify_client_secret(secret, app.client_secret_sha256) + with mock.patch.object(hashers, "check_password", side_effect=AssertionError): + assert app.check_client_secret(secret) + assert not app.check_client_secret("wrong") + + +def test_unrelated_save_preserves_both_hashes(): + """Saving an application's metadata does not change either credential hash.""" + app = ApplicationFactory() + original = (app.client_secret, app.client_secret_sha256) + app.name = "Renamed" + app.save() + app.refresh_from_db() + assert (app.client_secret, app.client_secret_sha256) == original + + +def test_creation_with_legacy_hash_defers_fast_hash_until_login(): + """An imported Django hash is preserved, never treated as the raw secret.""" + secret = get_random_string(128) + legacy = make_password(secret, hasher="pbkdf2_sha256") + app = ApplicationFactory(client_secret=legacy) + app.refresh_from_db() + assert app.client_secret == legacy + assert app.client_secret_sha256 is None + assert not app.check_client_secret(legacy) + assert app.check_client_secret(secret) + app.refresh_from_db() + assert app.client_secret == legacy + assert hashers.verify_client_secret(secret, app.client_secret_sha256) + + +def test_metadata_only_save_does_not_rotate_secret(): + """A secret excluded from update_fields must not change either stored hash.""" + app = ApplicationFactory() + original = (app.client_secret, app.client_secret_sha256) + app.client_secret = get_random_string(128) + app.name = "Renamed" + app.save(update_fields=["name"]) + app.refresh_from_db() + assert (app.client_secret, app.client_secret_sha256) == original + + +def test_empty_update_fields_does_not_rotate_secret(): + """Django's explicit no-op save must not update either credential field.""" + app = ApplicationFactory() + original = (app.client_secret, app.client_secret_sha256) + app.client_secret = get_random_string(128) + with CaptureQueriesContext(connection) as queries: + app.save(update_fields=[]) + assert not any(q["sql"].lstrip().startswith("UPDATE") for q in queries) + app.refresh_from_db() + assert (app.client_secret, app.client_secret_sha256) == original + + +def test_creation_with_salted_hash_skips_fast_hash(): + """An existing salted hash must not be hashed again as plaintext.""" + encoded = hashers.hash_client_secret(get_random_string(128)) + app = ApplicationFactory(client_secret=encoded) + app.refresh_from_db() + assert app.client_secret == encoded + assert app.client_secret_sha256 is None + + +@pytest.mark.parametrize("legacy_only", [False, True]) +def test_rotate_client_secret_updates_both_hashes(legacy_only, settings): + """Rotation revokes the old secret for both current and rollback releases.""" + settings.PASSWORD_HASHERS = [ + "django.contrib.auth.hashers.PBKDF2PasswordHasher", + ] + secret = get_random_string(128) + app = ApplicationFactory( + client_secret=make_password(secret) if legacy_only else secret + ) + + replacement = app.rotate_client_secret() + + assert replacement != secret + assert len(replacement) == settings.APPLICATION_CLIENT_SECRET_LENGTH + assert app.check_client_secret(replacement) + assert not app.check_client_secret(secret) + app.refresh_from_db() + assert app.client_secret.startswith("pbkdf2_sha256$") + assert check_password(replacement, app.client_secret) + assert not check_password(secret, app.client_secret) + assert hashers.verify_client_secret(replacement, app.client_secret_sha256) + assert not app.check_client_secret(secret) + assert app.client_secret != replacement + assert app.client_secret_sha256 != replacement + + +def test_rotate_client_secret_preserves_metadata(): + """Rotation persists only the credential fields, not other pending changes.""" + app = ApplicationFactory() + original_name = app.name + original_client_id = app.client_id + app.name = "Unsaved metadata" + + app.rotate_client_secret() + + app.refresh_from_db() + assert app.name == original_name + assert app.client_id == original_client_id + + +def test_rotate_client_secret_repeatedly_revokes_previous_secrets(): + """Only the latest generated secret remains valid after successive rotations.""" + original = get_random_string(128) + app = ApplicationFactory(client_secret=original) + first = app.rotate_client_secret() + second = app.rotate_client_secret() + + app.refresh_from_db() + assert len({original, first, second}) == 3 + assert app.check_client_secret(second) + assert check_password(second, app.client_secret) + for revoked in (original, first): + assert not app.check_client_secret(revoked) + assert not check_password(revoked, app.client_secret) + + +def test_token_endpoint_rejects_rotated_secret(): + """New token requests reject the revoked secret and accept its replacement.""" + secret = get_random_string(128) + app = ApplicationFactory(client_secret=secret) + user = UserFactory() + client = APIClient() + payload = { + "client_id": app.client_id, + "client_secret": secret, + "grant_type": "client_credentials", + "scope": user.email, + } + endpoint = "/external-api/v1.0/application/token/" + assert client.post(endpoint, payload, format="json").status_code == 200 + + replacement = app.rotate_client_secret() + + assert client.post(endpoint, payload, format="json").status_code == 401 + payload["client_secret"] = replacement + assert client.post(endpoint, payload, format="json").status_code == 200 diff --git a/src/backend/core/tests/test_external_api_token.py b/src/backend/core/tests/test_external_api_token.py index d0505c53c..a4e151227 100644 --- a/src/backend/core/tests/test_external_api_token.py +++ b/src/backend/core/tests/test_external_api_token.py @@ -7,17 +7,20 @@ Tests for external API /token endpoint from unittest import mock from urllib.parse import urlencode +from django.contrib.auth.hashers import check_password + import jwt import pytest from freezegun import freeze_time from rest_framework.test import APIClient +from core import hashers from core.factories import ( ApplicationDomainFactory, ApplicationFactory, UserFactory, ) -from core.models import ApplicationScope, User +from core.models import Application, ApplicationScope, User from core.services import provisional_user_service pytestmark = pytest.mark.django_db @@ -28,15 +31,13 @@ def test_api_applications_generate_token_application_disabled(settings): settings.APPLICATION_ENABLED = False user = UserFactory(email="user@example.com") + plain_secret = "test-secret-123" application = ApplicationFactory( + client_secret=plain_secret, is_active=True, scopes=[ApplicationScope.ROOMS_LIST], ) - plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() - client = APIClient() response = client.post( "/external-api/v1.0/application/token/", @@ -55,16 +56,13 @@ def test_api_applications_generate_token_application_disabled(settings): def test_api_applications_generate_token_success(settings): """Valid credentials should return a JWT token.""" UserFactory(email="User.Family@example.com") + plain_secret = "test-secret-123" application = ApplicationFactory( + client_secret=plain_secret, is_active=True, scopes=[ApplicationScope.ROOMS_LIST, ApplicationScope.ROOMS_CREATE], ) - # Store plain secret before it's hashed - plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() - client = APIClient() response = client.post( "/external-api/v1.0/application/token/", @@ -95,15 +93,13 @@ def test_api_applications_generate_token_form_urlencoded(settings): token endpoints, so that standard OAuth 2.0 client libraries work out of the box.""" UserFactory(email="user@example.com") + plain_secret = "test-secret-123" application = ApplicationFactory( + client_secret=plain_secret, is_active=True, scopes=[ApplicationScope.ROOMS_LIST, ApplicationScope.ROOMS_CREATE], ) - plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() - client = APIClient() response = client.post( "/external-api/v1.0/application/token/", @@ -170,11 +166,8 @@ def test_api_applications_generate_token_form_urlencoded_missing_fields(): def test_api_applications_generate_token_form_urlencoded_invalid_grant_type(): """An unsupported grant_type sent as form-urlencoded should return 400.""" user = UserFactory(email="user@example.com") - application = ApplicationFactory(is_active=True) - plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() + application = ApplicationFactory(client_secret=plain_secret, is_active=True) client = APIClient() response = client.post( @@ -198,15 +191,13 @@ def test_api_applications_generate_token_form_urlencoded_special_characters(): """Percent-encoded reserved characters ("&", "=", "+", "%") in the client_secret should survive form-urlencoded decoding.""" UserFactory(email="user@example.com") + plain_secret = "s3cr3t&with=special+chars%42" application = ApplicationFactory( + client_secret=plain_secret, is_active=True, scopes=[ApplicationScope.ROOMS_LIST], ) - plain_secret = "s3cr3t&with=special+chars%42" - application.client_secret = plain_secret - application.save() - client = APIClient() response = client.post( "/external-api/v1.0/application/token/", @@ -279,14 +270,54 @@ def test_api_applications_generate_token_invalid_client_secret(): assert "Invalid credentials" in str(response.data) +def test_token_unknown_client_id_with_valid_secret(): + """A valid secret cannot authenticate an unknown client ID.""" + secret = "application-a-secret" + ApplicationFactory(client_secret=secret) + user = UserFactory() + + response = APIClient().post( + "/external-api/v1.0/application/token/", + { + "client_id": "unknown-client-id", + "client_secret": secret, + "grant_type": "client_credentials", + "scope": user.email, + }, + format="json", + ) + + assert response.status_code == 401 + assert "Invalid credentials" in str(response.data) + + +def test_token_rejects_secret_owned_by_another_application(): + """Application A's secret cannot authenticate application B.""" + secret_a = "application-a-secret" + ApplicationFactory(client_secret=secret_a) + application_b = ApplicationFactory(client_secret="application-b-secret") + user = UserFactory() + + response = APIClient().post( + "/external-api/v1.0/application/token/", + { + "client_id": application_b.client_id, + "client_secret": secret_a, + "grant_type": "client_credentials", + "scope": user.email, + }, + format="json", + ) + + assert response.status_code == 401 + assert "Invalid credentials" in str(response.data) + + def test_api_applications_generate_token_inactive_application(): """Inactive application should return 401.""" user = UserFactory(email="user@example.com") - application = ApplicationFactory(is_active=False) - plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() + application = ApplicationFactory(client_secret=plain_secret, is_active=False) client = APIClient() response = client.post( @@ -328,11 +359,8 @@ def test_api_applications_generate_token_inactive_application_wrong_secret(): def test_api_applications_generate_token_invalid_email_format(): """Invalid email format should return 400.""" - application = ApplicationFactory(is_active=True) - plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() + application = ApplicationFactory(client_secret=plain_secret, is_active=True) client = APIClient() response = client.post( @@ -353,12 +381,9 @@ def test_api_applications_generate_token_invalid_email_format(): def test_api_applications_generate_token_domain_not_authorized(): """Application without domain authorization should return 403.""" user = UserFactory(email="user@denied.com") - application = ApplicationFactory(is_active=True) - ApplicationDomainFactory(application=application, domain="allowed.com") - plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() + application = ApplicationFactory(client_secret=plain_secret, is_active=True) + ApplicationDomainFactory(application=application, domain="allowed.com") client = APIClient() response = client.post( @@ -379,16 +404,14 @@ def test_api_applications_generate_token_domain_not_authorized(): def test_api_applications_generate_token_domain_authorized(): """Application with domain authorization should succeed.""" user = UserFactory(email="user@allowed.com") + plain_secret = "test-secret-123" application = ApplicationFactory( + client_secret=plain_secret, is_active=True, scopes=[ApplicationScope.ROOMS_LIST], ) ApplicationDomainFactory(application=application, domain="allowed.com") - plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() - client = APIClient() response = client.post( "/external-api/v1.0/application/token/", @@ -407,11 +430,8 @@ def test_api_applications_generate_token_domain_authorized(): def test_api_applications_generate_token_user_not_found(): """Non-existent user should return 404.""" - application = ApplicationFactory(is_active=True) - plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() + application = ApplicationFactory(client_secret=plain_secret, is_active=True) client = APIClient() response = client.post( @@ -434,15 +454,13 @@ def test_api_applications_token_payload_structure(settings): """Generated token should have correct payload structure.""" user = UserFactory(email="user@example.com") + plain_secret = "test-secret-123" application = ApplicationFactory( + client_secret=plain_secret, is_active=True, scopes=[ApplicationScope.ROOMS_LIST, ApplicationScope.ROOMS_CREATE], ) - plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() - client = APIClient() response = client.post( "/external-api/v1.0/application/token/", @@ -487,15 +505,13 @@ def test_api_applications_token_new_user(settings): assert len(User.objects.all()) == 0 + plain_secret = "test-secret-123" application = ApplicationFactory( + client_secret=plain_secret, is_active=True, scopes=[ApplicationScope.ROOMS_LIST, ApplicationScope.ROOMS_CREATE], ) - plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() - client = APIClient() response = client.post( "/external-api/v1.0/application/token/", @@ -545,15 +561,13 @@ def test_api_applications_token_existing_user(settings): assert len(User.objects.all()) == 1 + plain_secret = "test-secret-123" application = ApplicationFactory( + client_secret=plain_secret, is_active=True, scopes=[ApplicationScope.ROOMS_LIST, ApplicationScope.ROOMS_CREATE], ) - plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() - client = APIClient() response = client.post( "/external-api/v1.0/application/token/", @@ -598,12 +612,10 @@ def test_api_applications_token_new_user_race_condition(mock_get_by_email, setti settings.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION = True settings.OIDC_USER_SUB_FIELD_IMMUTABLE = False - application = ApplicationFactory( - is_active=True, scopes=[ApplicationScope.ROOMS_LIST] - ) plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() + application = ApplicationFactory( + client_secret=plain_secret, is_active=True, scopes=[ApplicationScope.ROOMS_LIST] + ) email = "john.doe@example.com" @@ -653,12 +665,10 @@ def test_api_applications_token_new_user_race_condition_unrecoverable( settings.OIDC_FALLBACK_TO_EMAIL_FOR_IDENTIFICATION = True settings.OIDC_USER_SUB_FIELD_IMMUTABLE = False - application = ApplicationFactory( - is_active=True, scopes=[ApplicationScope.ROOMS_LIST] - ) plain_secret = "test-secret-123" - application.client_secret = plain_secret - application.save() + application = ApplicationFactory( + client_secret=plain_secret, is_active=True, scopes=[ApplicationScope.ROOMS_LIST] + ) client = APIClient() response = client.post( @@ -674,3 +684,183 @@ def test_api_applications_token_new_user_race_condition_unrecoverable( assert response.status_code == 409 assert mock_get_or_create.call_count == 1 + + +def test_token_populates_fast_hash_and_stops_using_legacy_hash(): + """First login migrates; subsequent logins use only the fast hash.""" + secret = "application-secret" + + application = ApplicationFactory(client_secret=secret) + Application.objects.filter(pk=application.pk).update(client_secret_sha256=None) + application.refresh_from_db() + + original_hash = application.client_secret + + user = UserFactory() + payload = { + "client_id": application.client_id, + "client_secret": secret, + "grant_type": "client_credentials", + "scope": user.email, + } + client = APIClient() + + with mock.patch.object( + hashers, "check_password", wraps=hashers.check_password + ) as legacy_verifier: + response = client.post( + "/external-api/v1.0/application/token/", payload, format="json" + ) + + assert response.status_code == 200 + legacy_verifier.assert_called_once_with(secret, original_hash) + + application.refresh_from_db() + + migrated_hash = application.client_secret_sha256 + assert hashers.CLIENT_SECRET_HASH_PATTERN.fullmatch(migrated_hash) + assert hashers.verify_client_secret(secret, migrated_hash) + assert application.client_secret == original_hash + + # Fail immediately if a subsequent login tries the legacy verifier. + with mock.patch.object( + hashers, + "check_password", + side_effect=AssertionError("Legacy hash must no longer be used"), + ): + response = client.post( + "/external-api/v1.0/application/token/", payload, format="json" + ) + + assert response.status_code == 200 + application.refresh_from_db() + assert application.client_secret_sha256 == migrated_hash + assert application.client_secret == original_hash + + +def test_token_failed_login_leaves_legacy_credentials_untouched(): + """An incorrect secret neither migrates nor changes the legacy hash.""" + + application = ApplicationFactory(client_secret="application-secret") + Application.objects.filter(pk=application.pk).update(client_secret_sha256=None) + application.refresh_from_db() + + original_hash = application.client_secret + user = UserFactory() + + response = APIClient().post( + "/external-api/v1.0/application/token/", + { + "client_id": application.client_id, + "client_secret": "wrong-secret", + "grant_type": "client_credentials", + "scope": user.email, + }, + format="json", + ) + + assert response.status_code == 401 + application.refresh_from_db() + assert application.client_secret == original_hash + assert application.client_secret_sha256 is None + + +def test_token_concurrent_successful_logins_preserve_first_migration(): + """Both logins succeed; the later migration preserves the first hash.""" + secret = "application-secret" + application = ApplicationFactory(client_secret=secret) + Application.objects.filter(pk=application.pk).update(client_secret_sha256=None) + application.refresh_from_db() + + original_hash = application.client_secret + user = UserFactory() + payload = { + "client_id": application.client_id, + "client_secret": secret, + "grant_type": "client_credentials", + "scope": user.email, + } + legacy_verifier = hashers.check_password + winning_hashes = [] + + def verify_then_complete_other_login(raw_secret, encoded): + verified = legacy_verifier(raw_secret, encoded) + + # Complete another login before this request writes its migration. + # Restore the real verifier to avoid recursively invoking this callback. + with mock.patch.object(hashers, "check_password", new=legacy_verifier): + other_response = APIClient().post( + "/external-api/v1.0/application/token/", payload, format="json" + ) + + assert other_response.status_code == 200 + application.refresh_from_db() + winning_hashes.append(application.client_secret_sha256) + return verified + + with mock.patch.object( + hashers, "check_password", side_effect=verify_then_complete_other_login + ) as verifier: + response = APIClient().post( + "/external-api/v1.0/application/token/", payload, format="json" + ) + + assert response.status_code == 200 + verifier.assert_called_once_with(secret, original_hash) + application.refresh_from_db() + assert application.client_secret_sha256 == winning_hashes[0] + assert hashers.verify_client_secret(secret, application.client_secret_sha256) + assert application.client_secret == original_hash + + +def test_token_authenticates_after_rollback(): + """Legacy authentication still works after the fast hash is discarded.""" + secret = "application-secret" + application = ApplicationFactory(client_secret=secret) + Application.objects.filter(pk=application.pk).update(client_secret_sha256=None) + application.refresh_from_db() + + original_hash = application.client_secret + + user = UserFactory() + payload = { + "client_id": application.client_id, + "client_secret": secret, + "grant_type": "client_credentials", + "scope": user.email, + } + client = APIClient() + + # Authenticate with the new implementation and migrate the hash. + response = client.post( + "/external-api/v1.0/application/token/", payload, format="json" + ) + + assert response.status_code == 200 + + application.refresh_from_db() + + assert hashers.verify_client_secret(secret, application.client_secret_sha256) + assert application.client_secret == original_hash + + Application.objects.filter(pk=application.pk).update(client_secret_sha256=None) + + def legacy_check(instance, raw_secret): + return check_password(raw_secret, instance.client_secret) + + # Simulate the old release's verification using only the legacy field. + with mock.patch.object( + Application, + "check_client_secret", + autospec=True, + side_effect=legacy_check, + ) as verifier: + response = client.post( + "/external-api/v1.0/application/token/", payload, format="json" + ) + + assert response.status_code == 200 + verifier.assert_called_once() + application.refresh_from_db() + assert application.client_secret == original_hash + assert application.client_secret_sha256 is None diff --git a/src/backend/core/tests/test_models_applications.py b/src/backend/core/tests/test_models_applications.py index dd6cf9ed8..8363852df 100644 --- a/src/backend/core/tests/test_models_applications.py +++ b/src/backend/core/tests/test_models_applications.py @@ -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(): diff --git a/src/backend/meet/settings.py b/src/backend/meet/settings.py index f39a633bb..128b62029 100755 --- a/src/backend/meet/settings.py +++ b/src/backend/meet/settings.py @@ -1070,7 +1070,7 @@ class Base(Configuration): environ_prefix=None, ) APPLICATION_CLIENT_SECRET_LENGTH = values.PositiveIntegerValue( - 128, + 50, environ_name="APPLICATION_CLIENT_SECRET_LENGTH", environ_prefix=None, ) @@ -1432,6 +1432,20 @@ class Base(Configuration): stacklevel=2, ) + # Secrets use a 62-character alphanumeric charset (~5.95 bits/char). + # 43 characters provide at least 256 bits of entropy; 42 provide ~250 bits. + if cls.APPLICATION_CLIENT_SECRET_LENGTH < 43: + warnings.warn( + f"APPLICATION_CLIENT_SECRET_LENGTH={cls.APPLICATION_CLIENT_SECRET_LENGTH} " + "is below the recommended 43 characters (256 bits of entropy). " + "Application secrets use a fast hash and rely on high entropy to " + "resist offline guessing if the database leaks. " + "Please set APPLICATION_CLIENT_SECRET_LENGTH to at least 43.", + # We use UserWarning to make sure it shows up in production deployment + UserWarning, + stacklevel=2, + ) + # The SENTRY_DSN setting should be available to activate sentry for an environment if cls.SENTRY_DSN is not None: sentry_sdk.init( @@ -1517,6 +1531,7 @@ class Test(Base): ) PASSWORD_HASHERS = [ "django.contrib.auth.hashers.MD5PasswordHasher", + "django.contrib.auth.hashers.PBKDF2PasswordHasher", ] USE_SWAGGER = True EXTERNAL_API_ENABLED = True