(cls, request: Union[HttpRequest, Request])
| 63 | |
| 64 | @classmethod |
| 65 | def authenticate(cls, request: Union[HttpRequest, Request]) -> Optional[Tuple[Any, None]]: |
| 66 | from posthog.models import PersonalAPIKey |
| 67 | |
| 68 | personal_api_key_with_source = cls.find_key_with_source(request) |
| 69 | if not personal_api_key_with_source: |
| 70 | return None |
| 71 | personal_api_key, source = personal_api_key_with_source |
| 72 | secure_value = hash_key_value(personal_api_key) |
| 73 | try: |
| 74 | personal_api_key_object = ( |
| 75 | PersonalAPIKey.objects.select_related("user") |
| 76 | .filter(user__is_active=True) |
| 77 | .get(secure_value=secure_value) |
| 78 | ) |
| 79 | except PersonalAPIKey.DoesNotExist: |
| 80 | raise AuthenticationFailed(detail=f"Personal API key found in request {source} is invalid.") |
| 81 | |
| 82 | now = timezone.now() |
| 83 | key_last_used_at = personal_api_key_object.last_used_at |
| 84 | # Only updating last_used_at if the hour's changed |
| 85 | # This is to avooid excessive UPDATE queries, while still presenting accurate (down to the hour) info in the UI |
| 86 | if key_last_used_at is None or (now.year, now.month, now.day, now.hour) > ( |
| 87 | key_last_used_at.year, |
| 88 | key_last_used_at.month, |
| 89 | key_last_used_at.day, |
| 90 | key_last_used_at.hour, |
| 91 | ): |
| 92 | key_last_used_at = now |
| 93 | personal_api_key_object.save() |
| 94 | assert personal_api_key_object.user is not None |
| 95 | |
| 96 | # :KLUDGE: CHMiddleware does not receive the correct user when authenticating by api key. |
| 97 | tag_queries( |
| 98 | user_id=personal_api_key_object.user.pk, |
| 99 | team_id=personal_api_key_object.user.current_team_id, |
| 100 | access_method="personal_api_key", |
| 101 | ) |
| 102 | request.using_personal_api_key = True # type: ignore |
| 103 | return personal_api_key_object.user, None |
| 104 | |
| 105 | @classmethod |
| 106 | def authenticate_header(cls, request) -> str: |
nothing calls this directly
no test coverage detected