fix: Fix iteration counts breaking personal API keys (#19355)

* Lock the iteration count for password hasher for personal API keys
* Adjust old migration as well
* Make it try both iteration counts

posthog/api/test/test_personal_api_keys.py

@@ -128,6 +128,16 @@ class TestPersonalAPIKeysAPIAuthentication(APIBaseTest):
     def setUp(self):
         self.value = generate_random_token_personal()
         self.key = PersonalAPIKey.objects.create(label="Test", user=self.user, secure_value=hash_key_value(self.value))
+        self.value_390000 = generate_random_token_personal()
+        self.key_390000 = PersonalAPIKey.objects.create(
+            label="Test", user=self.user, secure_value=hash_key_value(self.value_390000, iterations=390000)
+        )
+        self.value_hardcoded = "phx_0a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p"
+        self.key_hardcoded = PersonalAPIKey.objects.create(
+            label="Test",
+            user=self.user,
+            secure_value="pbkdf2_sha256$260000$posthog_personal_api_key$dUOOjl6bYdigHd+QfhYzN6P2vM01ZbFROS8dm9KRK7Y=",
+        )
         return super().setUp()
 
     def test_no_key(self):
@@ -150,6 +160,20 @@ def test_header_resilient(self):
         )
         self.assertEqual(response.status_code, 200)
 
+    def test_header_alternative_iteration_count(self):
+        response = self.client.get(
+            f"/api/projects/{self.team.id}/dashboards/",
+            HTTP_AUTHORIZATION=f"Bearer {self.value_390000}",
+        )
+        self.assertEqual(response.status_code, 200)
+
+    def test_header_hardcoded(self):
+        response = self.client.get(
+            f"/api/projects/{self.team.id}/dashboards/",
+            HTTP_AUTHORIZATION=f"Bearer {self.value_hardcoded}",
+        )
+        self.assertEqual(response.status_code, 200)
+
     def test_query_string(self):
         response = self.client.get(f"/api/projects/{self.team.id}/dashboards/?personal_api_key={self.value}")
         self.assertEqual(response.status_code, 200)

posthog/auth.py

@@ -14,7 +14,10 @@
 
 from posthog.clickhouse.query_tagging import tag_queries
 from posthog.jwt import PosthogJwtAudience, decode_jwt
-from posthog.models.personal_api_key import hash_key_value
+from posthog.models.personal_api_key import (
+    hash_key_value,
+    PERSONAL_API_KEY_ITERATIONS_TO_TRY,
+)
 from posthog.models.sharing_configuration import SharingConfiguration
 from posthog.models.user import User
 from django.contrib.auth.models import AnonymousUser
@@ -65,22 +68,36 @@ def find_key(
         return key_with_source[0] if key_with_source is not None else None
 
     @classmethod
-    def authenticate(cls, request: Union[HttpRequest, Request]) -> Optional[Tuple[Any, None]]:
+    def validate_key(cls, personal_api_key_with_source):
         from posthog.models import PersonalAPIKey
 
+        personal_api_key, source = personal_api_key_with_source
+        personal_api_key_object = None
+
+        for iterations in PERSONAL_API_KEY_ITERATIONS_TO_TRY:
+            secure_value = hash_key_value(personal_api_key, iterations=iterations)
+            try:
+                personal_api_key_object = (
+                    PersonalAPIKey.objects.select_related("user")
+                    .filter(user__is_active=True)
+                    .get(secure_value=secure_value)
+                )
+                break
+            except PersonalAPIKey.DoesNotExist:
+                pass
+
+        if not personal_api_key_object:
+            raise AuthenticationFailed(detail=f"Personal API key found in request {source} is invalid.")
+
+        return personal_api_key_object
+
+    @classmethod
+    def authenticate(cls, request: Union[HttpRequest, Request]) -> Optional[Tuple[Any, None]]:
         personal_api_key_with_source = cls.find_key_with_source(request)
         if not personal_api_key_with_source:
             return None
-        personal_api_key, source = personal_api_key_with_source
-        secure_value = hash_key_value(personal_api_key)
-        try:
-            personal_api_key_object = (
-                PersonalAPIKey.objects.select_related("user")
-                .filter(user__is_active=True)
-                .get(secure_value=secure_value)
-            )
-        except PersonalAPIKey.DoesNotExist:
-            raise AuthenticationFailed(detail=f"Personal API key found in request {source} is invalid.")
+
+        personal_api_key_object = cls.validate_key(personal_api_key_with_source)
 
         now = timezone.now()
         key_last_used_at = personal_api_key_object.last_used_at

posthog/migrations/0260_pak_v2.py

@@ -1,13 +1,8 @@
 # Generated by Django 3.2.15 on 2022-09-20 17:07
 
-from django.contrib.auth.hashers import get_hasher
 from django.db import migrations, models
 
-PERSONAL_API_KEY_SALT = "posthog_personal_api_key"
-
-
-def hash_key_value(value: str) -> str:
-    return get_hasher().encode(value, PERSONAL_API_KEY_SALT)
+from posthog.models.personal_api_key import hash_key_value
 
 
 def hash_all_keys(apps, schema_editor):

posthog/models/personal_api_key.py

@@ -1,16 +1,27 @@
-from django.contrib.auth.hashers import get_hasher
+from django.contrib.auth.hashers import PBKDF2PasswordHasher
 from django.db import models
 from django.utils import timezone
 
 from .utils import generate_random_token
 
+# Fixed iteration count for PBKDF2PasswordHasher hasher.
+# This is the iteration count used by PostHog since the beginning of time.
+# Changing this would break all existing personal API keys.
+PERSONAL_API_KEY_ITERATIONS = 260000
+
+PERSONAL_API_KEY_ITERATIONS_TO_TRY = (
+    PERSONAL_API_KEY_ITERATIONS,
+    390000,  # This is the iteration count used briefly on some API keys.
+)
+
 # A constant salt is not nearly as good as user-specific, but we must be able to look up a personal API key
 # by itself. Some salt is slightly better than none though.
 PERSONAL_API_KEY_SALT = "posthog_personal_api_key"
 
 
-def hash_key_value(value: str) -> str:
-    return get_hasher().encode(value, PERSONAL_API_KEY_SALT)
+def hash_key_value(value: str, iterations: int = PERSONAL_API_KEY_ITERATIONS) -> str:
+    hasher = PBKDF2PasswordHasher()
+    return hasher.encode(value, PERSONAL_API_KEY_SALT, iterations=iterations)
 
 
 class PersonalAPIKey(models.Model):