ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 13:01:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(api-key): use admin connector to validate authentication (#8883)

Browse Source
This commit is contained in:
Víctor Fernández Poyatos
2025-10-09 11:26:21 +02:00
committed by GitHub
parent d6685eec1f
commit b630234cdf
3 changed files with 903 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/backend/api/authentication.py
+21 -2
View File
@@ -10,6 +10,7 @@ from rest_framework.exceptions import AuthenticationFailed
from rest_framework.request import Request
from rest_framework_simplejwt.authentication import JWTAuthentication
from api.db_router import MainRouter
from api.models import TenantAPIKey, TenantAPIKeyManager
@@ -19,6 +20,22 @@ class TenantAPIKeyAuthentication(BaseAPIKeyAuth):
def __init__(self):
self.key_crypto = get_crypto()
def _authenticate_credentials(self, request, key):
"""
Override to use admin connection, bypassing RLS during authentication.
Delegates to parent after temporarily routing model queries to admin DB.
"""
# Temporarily point the model's manager to admin database
original_objects = self.model.objects
self.model.objects = self.model.objects.using(MainRouter.admin_db)
try:
# Call parent method which will now use admin database
return super()._authenticate_credentials(request, key)
finally:
# Restore original manager
self.model.objects = original_objects
def authenticate(self, request: Request):
prefixed_key = self.get_key(request)
@@ -43,13 +60,15 @@ class TenantAPIKeyAuthentication(BaseAPIKeyAuth):
api_key_pk = UUID(api_key_pk)
try:
api_key_instance = TenantAPIKey.objects.get(id=api_key_pk, prefix=prefix)
api_key_instance = TenantAPIKey.objects.using(MainRouter.admin_db).get(
id=api_key_pk, prefix=prefix
)
except TenantAPIKey.DoesNotExist:
raise AuthenticationFailed("Invalid API Key.")
# Update last_used_at
api_key_instance.last_used_at = timezone.now()
api_key_instance.save(update_fields=["last_used_at"])
api_key_instance.save(update_fields=["last_used_at"], using=MainRouter.admin_db)
return entity, {
"tenant_id": str(api_key_instance.tenant_id),
api/src/backend/api/tests/integration/test_authentication.py
+498
View File
@@ -1003,6 +1003,504 @@ class TestCombinedAuthentication:
assert api_key.last_used_at is not None
@pytest.mark.django_db
class TestAPIKeyRLSBypass:
"""Test RLS bypass fix for API key authentication.
These tests verify that API key authentication works correctly even when
RLS context is not set, which is critical since we don't know the tenant_id
until we look up the API key (which itself is protected by RLS).
The fix ensures all database operations during authentication use the admin
database, bypassing RLS constraints.
"""
def test_api_key_authentication_without_rls_context(
self, create_test_user, tenants_fixture, api_keys_fixture
):
"""Verify API key authentication works without pre-existing RLS context.
This is the core fix: authentication must succeed even when prowler.tenant_id
is not set, since we need to look up the API key to discover the tenant.
"""
client = APIClient()
api_key = api_keys_fixture[0]
api_key_headers = get_api_key_header(api_key._raw_key)
response = client.get(reverse("provider-list"), headers=api_key_headers)
assert response.status_code == 200
assert "data" in response.json()
def test_api_key_lookup_uses_admin_database(
self, create_test_user, tenants_fixture
):
"""Verify API key lookup uses admin database during authentication.
The TenantAPIKey model is RLS-protected, so queries against it would
normally fail without prowler.tenant_id set. The fix routes lookups
to the admin database which bypasses RLS.
"""
client = APIClient()
tenant = tenants_fixture[0]
role = Role.objects.create(
tenant_id=tenant.id,
name="Admin DB Test Role",
unlimited_visibility=True,
manage_account=True,
)
UserRoleRelationship.objects.create(
user=create_test_user,
role=role,
tenant_id=tenant.id,
)
api_key, raw_key = TenantAPIKey.objects.create_api_key(
name="Admin DB Test Key",
tenant_id=tenant.id,
entity=create_test_user,
)
api_key_headers = get_api_key_header(raw_key)
response = client.get(reverse("provider-list"), headers=api_key_headers)
assert response.status_code == 200
api_key.refresh_from_db()
assert api_key.last_used_at is not None
def test_tenant_context_established_after_authentication(
self, create_test_user, tenants_fixture, api_keys_fixture
):
"""Verify correct tenant context is established after API key auth.
After authentication, the tenant_id from the API key should be used
to set up the proper RLS context for subsequent queries.
"""
client = APIClient()
api_key = api_keys_fixture[0]
api_key_headers = get_api_key_header(api_key._raw_key)
# Use tenant-list endpoint to get actual tenant IDs
tenant_response = client.get(reverse("tenant-list"), headers=api_key_headers)
assert tenant_response.status_code == 200
tenant_data = tenant_response.json()["data"]
tenant_ids = [t["id"] for t in tenant_data]
# Verify the API key's tenant is in the list of accessible tenants
assert str(api_key.tenant_id) in tenant_ids
def test_concurrent_authentication_different_tenants(self, tenants_fixture):
"""Verify multiple API keys from different tenants can authenticate simultaneously.
This tests that the admin database routing works correctly in concurrent
scenarios and doesn't cause tenant isolation issues.
"""
client = APIClient()
user1 = User.objects.create_user(
name="concurrent_user1",
email="concurrent1@test.com",
password=TEST_PASSWORD,
)
user2 = User.objects.create_user(
name="concurrent_user2",
email="concurrent2@test.com",
password=TEST_PASSWORD,
)
tenant1 = tenants_fixture[0]
tenant2 = tenants_fixture[1]
Membership.objects.create(user=user1, tenant=tenant1)
Membership.objects.create(user=user2, tenant=tenant2)
role1 = Role.objects.create(
tenant_id=tenant1.id,
name="Concurrent Role 1",
unlimited_visibility=True,
manage_account=True,
)
role2 = Role.objects.create(
tenant_id=tenant2.id,
name="Concurrent Role 2",
unlimited_visibility=True,
manage_account=True,
)
UserRoleRelationship.objects.create(
user=user1,
role=role1,
tenant_id=tenant1.id,
)
UserRoleRelationship.objects.create(
user=user2,
role=role2,
tenant_id=tenant2.id,
)
api_key1, raw_key1 = TenantAPIKey.objects.create_api_key(
name="Concurrent Key 1",
tenant_id=tenant1.id,
entity=user1,
)
api_key2, raw_key2 = TenantAPIKey.objects.create_api_key(
name="Concurrent Key 2",
tenant_id=tenant2.id,
entity=user2,
)
headers1 = get_api_key_header(raw_key1)
headers2 = get_api_key_header(raw_key2)
response1 = client.get(reverse("provider-list"), headers=headers1)
response2 = client.get(reverse("provider-list"), headers=headers2)
assert response1.status_code == 200
assert response2.status_code == 200
api_key1.refresh_from_db()
api_key2.refresh_from_db()
assert api_key1.last_used_at is not None
assert api_key2.last_used_at is not None
assert api_key1.tenant_id == tenant1.id
assert api_key2.tenant_id == tenant2.id
def test_api_key_update_last_used_uses_admin_db(
self, create_test_user, tenants_fixture, api_keys_fixture
):
"""Verify last_used_at update uses admin database.
The update to last_used_at during authentication must also use the
admin database since it occurs before RLS context is established.
"""
client = APIClient()
api_key = api_keys_fixture[0]
assert api_key.last_used_at is None
api_key_headers = get_api_key_header(api_key._raw_key)
first_response = client.get(reverse("provider-list"), headers=api_key_headers)
assert first_response.status_code == 200
api_key.refresh_from_db()
first_timestamp = api_key.last_used_at
assert first_timestamp is not None
time.sleep(0.1)
second_response = client.get(reverse("provider-list"), headers=api_key_headers)
assert second_response.status_code == 200
api_key.refresh_from_db()
second_timestamp = api_key.last_used_at
assert second_timestamp > first_timestamp
def test_api_key_prefix_lookup_bypasses_rls(
self, create_test_user, tenants_fixture
):
"""Verify prefix-based API key lookup works without RLS context.
The authentication process splits the key into prefix and encrypted parts,
then looks up by prefix. This lookup must work via admin database.
"""
client = APIClient()
tenant = tenants_fixture[0]
role = Role.objects.create(
tenant_id=tenant.id,
name="Prefix Test Role",
unlimited_visibility=True,
manage_account=True,
)
UserRoleRelationship.objects.create(
user=create_test_user,
role=role,
tenant_id=tenant.id,
)
api_key, raw_key = TenantAPIKey.objects.create_api_key(
name="Prefix Test Key",
tenant_id=tenant.id,
entity=create_test_user,
)
prefix = raw_key.split(".")[0]
assert prefix == api_key.prefix
api_key_headers = get_api_key_header(raw_key)
response = client.get(reverse("provider-list"), headers=api_key_headers)
assert response.status_code == 200
def test_expired_api_key_check_uses_admin_db(
self, create_test_user, tenants_fixture
):
"""Verify expired API key validation works via admin database.
Checking if a key is expired requires reading from TenantAPIKey,
which must use admin database during authentication.
"""
client = APIClient()
tenant = tenants_fixture[0]
expired_key, raw_key = TenantAPIKey.objects.create_api_key(
name="Expired Test Key",
tenant_id=tenant.id,
entity=create_test_user,
expiry_date=datetime.now(timezone.utc) - timedelta(days=1),
)
api_key_headers = get_api_key_header(raw_key)
response = client.get(reverse("provider-list"), headers=api_key_headers)
assert response.status_code == 401
assert "expired" in response.json()["errors"][0]["detail"].lower()
def test_revoked_api_key_check_uses_admin_db(
self, create_test_user, tenants_fixture
):
"""Verify revoked API key validation works via admin database.
Checking if a key is revoked requires reading from TenantAPIKey,
which must use admin database during authentication.
"""
client = APIClient()
tenant = tenants_fixture[0]
role = Role.objects.create(
tenant_id=tenant.id,
name="Revoked Test Role",
unlimited_visibility=True,
manage_account=True,
)
UserRoleRelationship.objects.create(
user=create_test_user,
role=role,
tenant_id=tenant.id,
)
api_key, raw_key = TenantAPIKey.objects.create_api_key(
name="Revoked Test Key",
tenant_id=tenant.id,
entity=create_test_user,
)
api_key.revoked = True
api_key.save()
api_key_headers = get_api_key_header(raw_key)
response = client.get(reverse("provider-list"), headers=api_key_headers)
assert response.status_code == 401
assert "revoked" in response.json()["errors"][0]["detail"].lower()
@pytest.mark.django_db
class TestAPIKeyMultiTenantWorkflows:
"""Test complete multi-tenant workflows using API keys.
These integration tests verify end-to-end scenarios where API keys
are used across different tenants and ensure proper isolation.
"""
def test_user_with_multiple_tenant_memberships_api_keys(self, tenants_fixture):
"""User with memberships in multiple tenants can use different API keys.
Tests that a user can have separate API keys for different tenants
and each key only accesses resources in its tenant.
"""
client = APIClient()
user = User.objects.create_user(
name="multi_tenant_user",
email="multitenant@test.com",
password=TEST_PASSWORD,
)
tenant1 = tenants_fixture[0]
tenant2 = tenants_fixture[1]
Membership.objects.create(user=user, tenant=tenant1)
Membership.objects.create(user=user, tenant=tenant2)
role1 = Role.objects.create(
tenant_id=tenant1.id,
name="Multi Tenant Role 1",
unlimited_visibility=True,
manage_account=True,
)
role2 = Role.objects.create(
tenant_id=tenant2.id,
name="Multi Tenant Role 2",
unlimited_visibility=True,
manage_account=True,
)
UserRoleRelationship.objects.create(
user=user,
role=role1,
tenant_id=tenant1.id,
)
UserRoleRelationship.objects.create(
user=user,
role=role2,
tenant_id=tenant2.id,
)
key1, raw_key1 = TenantAPIKey.objects.create_api_key(
name="Tenant 1 Key",
tenant_id=tenant1.id,
entity=user,
)
key2, raw_key2 = TenantAPIKey.objects.create_api_key(
name="Tenant 2 Key",
tenant_id=tenant2.id,
entity=user,
)
headers1 = get_api_key_header(raw_key1)
headers2 = get_api_key_header(raw_key2)
response1 = client.get(reverse("provider-list"), headers=headers1)
response2 = client.get(reverse("provider-list"), headers=headers2)
assert response1.status_code == 200
assert response2.status_code == 200
me_response1 = client.get(reverse("user-me"), headers=headers1)
me_response2 = client.get(reverse("user-me"), headers=headers2)
assert me_response1.status_code == 200
assert me_response2.status_code == 200
assert me_response1.json()["data"]["id"] == str(user.id)
assert me_response2.json()["data"]["id"] == str(user.id)
def test_api_key_cannot_access_different_tenant_resources(
self, tenants_fixture, providers_fixture
):
"""API key from one tenant cannot access resources from another tenant.
Verifies RLS enforcement after authentication ensures tenant isolation.
"""
client = APIClient()
user1 = User.objects.create_user(
name="tenant1_user",
email="tenant1user@test.com",
password=TEST_PASSWORD,
)
user2 = User.objects.create_user(
name="tenant2_user",
email="tenant2user@test.com",
password=TEST_PASSWORD,
)
tenant1 = tenants_fixture[0]
tenant2 = tenants_fixture[1]
Membership.objects.create(user=user1, tenant=tenant1)
Membership.objects.create(user=user2, tenant=tenant2)
role1 = Role.objects.create(
tenant_id=tenant1.id,
name="Isolation Test Role 1",
unlimited_visibility=True,
manage_account=True,
)
role2 = Role.objects.create(
tenant_id=tenant2.id,
name="Isolation Test Role 2",
unlimited_visibility=True,
manage_account=True,
)
UserRoleRelationship.objects.create(
user=user1,
role=role1,
tenant_id=tenant1.id,
)
UserRoleRelationship.objects.create(
user=user2,
role=role2,
tenant_id=tenant2.id,
)
key1, raw_key1 = TenantAPIKey.objects.create_api_key(
name="Isolation Key 1",
tenant_id=tenant1.id,
entity=user1,
)
headers1 = get_api_key_header(raw_key1)
provider_response = client.get(reverse("provider-list"), headers=headers1)
assert provider_response.status_code == 200
providers_data = provider_response.json()["data"]
if providers_data:
for provider in providers_data:
provider_tenant_id = str(tenants_fixture[0].id)
assert str(tenant2.id) != provider_tenant_id
def test_api_key_workflow_create_authenticate_revoke(
self, create_test_user_rbac, tenants_fixture
):
"""Complete workflow: create API key via JWT, use it, then revoke via JWT.
Tests the full lifecycle using both JWT and API key authentication.
"""
client = APIClient()
tenants_fixture[0]
jwt_access_token, _ = get_api_tokens(
client, create_test_user_rbac.email, TEST_PASSWORD
)
jwt_headers = get_authorization_header(jwt_access_token)
create_response = client.post(
reverse("api-key-list"),
data={
"data": {
"type": "api-keys",
"attributes": {
"name": "Workflow Test Key",
},
}
},
format="vnd.api+json",
headers=jwt_headers,
)
assert create_response.status_code == 201
api_key_data = create_response.json()["data"]
api_key_id = api_key_data["id"]
raw_api_key = api_key_data["attributes"]["api_key"]
api_key_headers = get_api_key_header(raw_api_key)
auth_response = client.get(reverse("provider-list"), headers=api_key_headers)
assert auth_response.status_code == 200
revoke_response = client.delete(
reverse("api-key-revoke", kwargs={"pk": api_key_id}),
headers=jwt_headers,
)
assert revoke_response.status_code == 200
revoked_auth_response = client.get(
reverse("provider-list"), headers=api_key_headers
)
assert revoked_auth_response.status_code == 401
assert "revoked" in revoked_auth_response.json()["errors"][0]["detail"].lower()
def get_api_key_header(api_key: str) -> dict:
"""Helper to create API key authorization header."""
return {"Authorization": f"Api-Key {api_key}"}
api/src/backend/api/tests/test_authentication.py
+384
View File
@@ -0,0 +1,384 @@
import time
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
from uuid import uuid4
import pytest
from django.test import RequestFactory
from rest_framework.exceptions import AuthenticationFailed
from api.authentication import TenantAPIKeyAuthentication
from api.db_router import MainRouter
from api.models import TenantAPIKey
@pytest.mark.django_db
class TestTenantAPIKeyAuthentication:
@pytest.fixture
def auth_backend(self):
"""Create an instance of TenantAPIKeyAuthentication."""
return TenantAPIKeyAuthentication()
@pytest.fixture
def request_factory(self):
"""Create a Django request factory."""
return RequestFactory()
def test_authenticate_credentials_uses_admin_database(
self, auth_backend, api_keys_fixture, request_factory
):
"""Test that _authenticate_credentials routes queries to admin database."""
api_key = api_keys_fixture[0]
raw_key = api_key._raw_key
# Extract the encrypted key part (after the prefix and separator)
_, encrypted_key = raw_key.split(TenantAPIKey.objects.separator, 1)
# Create a mock request
request = request_factory.get("/")
# Call the method
entity, auth_dict = auth_backend._authenticate_credentials(
request, encrypted_key
)
# Verify that the entity is the user associated with the API key
assert entity == api_key.entity
assert entity.id == api_key.entity.id
def test_authenticate_credentials_restores_manager_on_success(
self, auth_backend, api_keys_fixture, request_factory
):
"""Test that the manager is restored after successful authentication."""
api_key = api_keys_fixture[0]
raw_key = api_key._raw_key
_, encrypted_key = raw_key.split(TenantAPIKey.objects.separator, 1)
# Store the original manager
original_manager = TenantAPIKey.objects
request = request_factory.get("/")
# Call the method
auth_backend._authenticate_credentials(request, encrypted_key)
# Verify the manager was restored
assert TenantAPIKey.objects == original_manager
def test_authenticate_credentials_restores_manager_on_exception(
self, auth_backend, request_factory
):
"""Test that the manager is restored even when an exception occurs."""
# Store the original manager
original_manager = TenantAPIKey.objects
request = request_factory.get("/")
# Try to authenticate with an invalid key that will raise an exception
with pytest.raises(Exception):
auth_backend._authenticate_credentials(request, "invalid_encrypted_key")
# Verify the manager was restored despite the exception
assert TenantAPIKey.objects == original_manager
def test_authenticate_valid_api_key(
self, auth_backend, api_keys_fixture, request_factory
):
"""Test successful authentication with a valid API key."""
api_key = api_keys_fixture[0]
raw_key = api_key._raw_key
# Create a request with the API key in the Authorization header
request = request_factory.get("/")
request.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key}"
# Authenticate
entity, auth_dict = auth_backend.authenticate(request)
# Verify the entity and auth dict
assert entity == api_key.entity
assert auth_dict["tenant_id"] == str(api_key.tenant_id)
assert auth_dict["sub"] == str(api_key.entity.id)
assert auth_dict["api_key_prefix"] == api_key.prefix
# Verify that last_used_at was updated
api_key.refresh_from_db()
assert api_key.last_used_at is not None
assert (datetime.now(timezone.utc) - api_key.last_used_at).seconds < 5
def test_authenticate_valid_api_key_uses_admin_database(
self, auth_backend, api_keys_fixture, request_factory
):
"""Test that authenticate uses admin database for API key lookup."""
api_key = api_keys_fixture[0]
raw_key = api_key._raw_key
request = request_factory.get("/")
request.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key}"
# Mock the manager's using method to verify it's called with admin_db
with patch.object(
TenantAPIKey.objects, "using", wraps=TenantAPIKey.objects.using
) as mock_using:
auth_backend.authenticate(request)
# Verify that .using('admin') was called
mock_using.assert_called_with(MainRouter.admin_db)
def test_authenticate_invalid_key_format_missing_separator(
self, auth_backend, request_factory
):
"""Test authentication fails with invalid API key format (no separator)."""
request = request_factory.get("/")
request.META["HTTP_AUTHORIZATION"] = "Api-Key invalid_key_no_separator"
with pytest.raises(AuthenticationFailed) as exc_info:
auth_backend.authenticate(request)
assert str(exc_info.value.detail) == "Invalid API Key."
def test_authenticate_invalid_key_format_empty_prefix(
self, auth_backend, request_factory
):
"""Test authentication fails with empty prefix."""
request = request_factory.get("/")
request.META["HTTP_AUTHORIZATION"] = "Api-Key .encrypted_part"
with pytest.raises(AuthenticationFailed) as exc_info:
auth_backend.authenticate(request)
assert str(exc_info.value.detail) == "Invalid API Key."
def test_authenticate_invalid_encrypted_key(
self, auth_backend, api_keys_fixture, request_factory
):
"""Test authentication fails with invalid encrypted key."""
api_key = api_keys_fixture[0]
# Create an invalid key with valid prefix but invalid encryption
invalid_key = f"{api_key.prefix}.invalid_encrypted_data"
request = request_factory.get("/")
request.META["HTTP_AUTHORIZATION"] = f"Api-Key {invalid_key}"
with pytest.raises(AuthenticationFailed) as exc_info:
auth_backend.authenticate(request)
assert str(exc_info.value.detail) == "Invalid API Key."
def test_authenticate_revoked_api_key(
self, auth_backend, api_keys_fixture, request_factory
):
"""Test authentication fails with a revoked API key."""
# Use the revoked API key (index 2 from fixture)
api_key = api_keys_fixture[2]
raw_key = api_key._raw_key
request = request_factory.get("/")
request.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key}"
# The revoked key should fail during credential validation
with pytest.raises(AuthenticationFailed) as exc_info:
auth_backend.authenticate(request)
assert str(exc_info.value.detail) == "This API Key has been revoked."
def test_authenticate_expired_api_key(
self, auth_backend, create_test_user, tenants_fixture, request_factory
):
"""Test authentication fails with an expired API key."""
tenant = tenants_fixture[0]
user = create_test_user
# Create an expired API key
api_key, raw_key = TenantAPIKey.objects.create_api_key(
name="Expired API Key",
tenant_id=tenant.id,
entity=user,
expiry_date=datetime.now(timezone.utc) - timedelta(days=1),
)
request = request_factory.get("/")
request.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key}"
with pytest.raises(AuthenticationFailed) as exc_info:
auth_backend.authenticate(request)
assert str(exc_info.value.detail) == "API Key has already expired."
def test_authenticate_nonexistent_api_key(
self, auth_backend, api_keys_fixture, request_factory
):
"""Test authentication fails when API key doesn't exist in database."""
# Create a valid-looking encrypted key with a non-existent UUID
api_key = api_keys_fixture[0]
non_existent_uuid = str(uuid4())
# Manually create an encrypted key with a non-existent ID
payload = {
"_pk": non_existent_uuid,
"_exp": (datetime.now(timezone.utc) + timedelta(days=30)).timestamp(),
}
encrypted_key = auth_backend.key_crypto.generate(payload)
fake_key = f"{api_key.prefix}.{encrypted_key}"
request = request_factory.get("/")
request.META["HTTP_AUTHORIZATION"] = f"Api-Key {fake_key}"
with pytest.raises(AuthenticationFailed) as exc_info:
auth_backend.authenticate(request)
assert str(exc_info.value.detail) == "No entity matching this api key."
def test_authenticate_updates_last_used_at(
self, auth_backend, api_keys_fixture, request_factory
):
"""Test that last_used_at is updated on successful authentication."""
api_key = api_keys_fixture[0]
raw_key = api_key._raw_key
# Store the original last_used_at
original_last_used = api_key.last_used_at
request = request_factory.get("/")
request.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key}"
# Authenticate
auth_backend.authenticate(request)
# Refresh from database
api_key.refresh_from_db()
# Verify last_used_at was updated
assert api_key.last_used_at is not None
if original_last_used:
assert api_key.last_used_at > original_last_used
def test_authenticate_saves_to_admin_database(
self, auth_backend, api_keys_fixture, request_factory
):
"""Test that the API key save operation uses admin database."""
api_key = api_keys_fixture[0]
raw_key = api_key._raw_key
request = request_factory.get("/")
request.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key}"
# Mock the save method to verify it's called with using='admin'
with patch.object(TenantAPIKey, "save") as mock_save:
auth_backend.authenticate(request)
# Verify save was called with using=admin_db
mock_save.assert_called_once_with(
update_fields=["last_used_at"], using=MainRouter.admin_db
)
def test_authenticate_returns_correct_auth_dict(
self, auth_backend, api_keys_fixture, request_factory
):
"""Test that the auth dict contains all required fields."""
api_key = api_keys_fixture[0]
raw_key = api_key._raw_key
request = request_factory.get("/")
request.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key}"
entity, auth_dict = auth_backend.authenticate(request)
# Verify all required fields are present
assert "tenant_id" in auth_dict
assert "sub" in auth_dict
assert "api_key_prefix" in auth_dict
# Verify values are correct
assert auth_dict["tenant_id"] == str(api_key.tenant_id)
assert auth_dict["sub"] == str(api_key.entity.id)
assert auth_dict["api_key_prefix"] == api_key.prefix
def test_authenticate_with_multiple_api_keys_same_tenant(
self, auth_backend, api_keys_fixture, request_factory
):
"""Test that authentication works correctly with multiple API keys for the same tenant."""
# Test with first API key
api_key1 = api_keys_fixture[0]
raw_key1 = api_key1._raw_key
request1 = request_factory.get("/")
request1.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key1}"
entity1, auth_dict1 = auth_backend.authenticate(request1)
assert entity1 == api_key1.entity
assert auth_dict1["api_key_prefix"] == api_key1.prefix
# Test with second API key
api_key2 = api_keys_fixture[1]
raw_key2 = api_key2._raw_key
request2 = request_factory.get("/")
request2.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key2}"
entity2, auth_dict2 = auth_backend.authenticate(request2)
assert entity2 == api_key2.entity
assert auth_dict2["api_key_prefix"] == api_key2.prefix
# Verify they're different keys but same tenant
assert auth_dict1["api_key_prefix"] != auth_dict2["api_key_prefix"]
assert auth_dict1["tenant_id"] == auth_dict2["tenant_id"]
def test_authenticate_with_wrong_prefix_in_db(
self, auth_backend, api_keys_fixture, request_factory
):
"""Test authentication fails when prefix doesn't match database."""
api_key = api_keys_fixture[0]
raw_key = api_key._raw_key
# Extract the encrypted part and combine with wrong prefix
_, encrypted_part = raw_key.split(TenantAPIKey.objects.separator, 1)
wrong_key = f"pk_wrong123.{encrypted_part}"
request = request_factory.get("/")
request.META["HTTP_AUTHORIZATION"] = f"Api-Key {wrong_key}"
with pytest.raises(AuthenticationFailed) as exc_info:
auth_backend.authenticate(request)
assert str(exc_info.value.detail) == "Invalid API Key."
def test_authenticate_credentials_exception_handling(
self, auth_backend, request_factory
):
"""Test that exceptions in _authenticate_credentials are properly handled."""
request = request_factory.get("/")
# Test with completely invalid data that will cause InvalidToken
with pytest.raises(Exception):
auth_backend._authenticate_credentials(request, "completely_invalid")
def test_authenticate_with_expired_timestamp(
self, auth_backend, create_test_user, tenants_fixture, request_factory
):
"""Test that expired timestamp in encrypted key causes authentication failure."""
tenant = tenants_fixture[0]
user = create_test_user
# Create an API key with a very short expiry
api_key, raw_key = TenantAPIKey.objects.create_api_key(
name="Short-lived API Key",
tenant_id=tenant.id,
entity=user,
expiry_date=datetime.now(timezone.utc) + timedelta(seconds=1),
)
# Wait for the key to expire
time.sleep(2)
request = request_factory.get("/")
request.META["HTTP_AUTHORIZATION"] = f"Api-Key {raw_key}"
# Should fail with expired key
with pytest.raises(AuthenticationFailed) as exc_info:
auth_backend.authenticate(request)
assert str(exc_info.value.detail) == "API Key has already expired."