fd16d033fc
The custom API key authentication only verified that the APIToken row was active and unexpired; it never checked the owning user's is_active flag. DRF's IsAuthenticated only checks user.is_authenticated (always True for a real User), so a user whose account was deactivated could keep using a previously issued API key indefinitely. Add user__is_active=True to the validate_api_token() lookup so a token tied to a disabled account is treated as invalid (a generic AuthenticationFailed, avoiding account-state disclosure). Applied to both the external API middleware (plane/api) and the identical, currently unused copy in plane/app to prevent the gap from being reintroduced. Adds unit coverage on validate_api_token and an end-to-end contract test proving GET /api/v1/users/me/ is denied once the account is deactivated.
52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
# Copyright (c) 2023-present Plane Software, Inc. and contributors
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
# See the LICENSE file for details.
|
|
|
|
# Django imports
|
|
from django.utils import timezone
|
|
from django.db.models import Q
|
|
|
|
# Third party imports
|
|
from rest_framework import authentication
|
|
from rest_framework.exceptions import AuthenticationFailed
|
|
|
|
# Module imports
|
|
from plane.db.models import APIToken
|
|
|
|
|
|
class APIKeyAuthentication(authentication.BaseAuthentication):
|
|
"""
|
|
Authentication with an API Key
|
|
"""
|
|
|
|
www_authenticate_realm = "api"
|
|
media_type = "application/json"
|
|
auth_header_name = "X-Api-Key"
|
|
|
|
def get_api_token(self, request):
|
|
return request.headers.get(self.auth_header_name)
|
|
|
|
def validate_api_token(self, token):
|
|
try:
|
|
api_token = APIToken.objects.get(
|
|
Q(Q(expired_at__gt=timezone.now()) | Q(expired_at__isnull=True)),
|
|
token=token,
|
|
is_active=True,
|
|
user__is_active=True,
|
|
)
|
|
except APIToken.DoesNotExist:
|
|
raise AuthenticationFailed("Given API token is not valid")
|
|
|
|
# save api token last used
|
|
api_token.last_used = timezone.now()
|
|
api_token.save(update_fields=["last_used"])
|
|
return (api_token.user, api_token.token)
|
|
|
|
def authenticate(self, request):
|
|
token = self.get_api_token(request=request)
|
|
if not token:
|
|
return None
|
|
|
|
# Validate the API token
|
|
user, token = self.validate_api_token(token)
|
|
return user, token
|