vynte-infra/plane
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e84c7f12f36e8ef759700877a6a28cc0ae21035a
plane/apps/api/plane/tests/contract/app/test_api_token.py
T
sriram veeraghantaandGitHub edf2475413 refactor: logging with retention + API token hardening (#9148) 
* fix: harden API token handling against rate-limit tampering and plaintext logging

- Make `allowed_rate_limit` read-only on APITokenSerializer so users can no
  longer raise their own API token rate limit via PATCH (GHSA-xfgr-2x3f-g2cf).
- Stop persisting API keys in plaintext in APITokenLogMiddleware: store a
  SHA-256 hash as the token identifier and redact sensitive request headers
  (X-Api-Key, Authorization, Cookie) before logging (GHSA-r5p8-cj3q-38cc).

* refactor: remove MongoDB log sink and add per-log-type retention

Logs are now written to and cleared from PostgreSQL only; MongoDB is no
longer used as a log sink or archive.

- Drop the MongoDB write/archival paths from the API request logger, the
  webhook log writer, and the cleanup tasks; Postgres is the sole sink.
- Cleanup tasks now hard-delete expired rows in batches via `all_objects`
  (rows are removed immediately, not soft-deleted).
- Add env-backed, per-log-type retention settings: API activity logs
  (API_ACTIVITY_LOG_RETENTION_DAYS, default 14), webhook logs
  (WEBHOOK_LOG_RETENTION_DAYS, default 14), email logs
  (EMAIL_LOG_RETENTION_DAYS, default 7). HARD_DELETE_AFTER_DAYS no longer
  drives any log cleanup.
- Delete settings/mongo.py, remove MONGO_DB_* settings and the plane.mongo
  loggers, and drop the pymongo dependency.

* chore: gitignore local advisories.md notes file

* fix: use keyed HMAC-SHA256 for API token log identifier

Address CodeQL "weak hashing of sensitive data" by hashing the API key with
a SECRET_KEY-keyed HMAC instead of a bare SHA-256. The identifier is a
non-reversible tokenization of a high-entropy key (not password storage);
keying it also prevents precomputing the digest from a known key value.

* chore: address review feedback on log cleanup and request logging

- process_logs accepts extra kwargs so jobs enqueued by an older release
  (with a mongo_log arg) don't fail during a rolling deploy.
- Log-cleanup batch delete failures are logged and skipped rather than
  aborting the run, so a single bad batch can't block the rest.
- Extend logger middleware test to assert Authorization and Cookie headers
  are redacted; add a test that a failing cleanup batch is swallowed.

* fix: fall back to default when a log retention env value is invalid

Negative (or unparseable) retention values would compute a future cutoff and
delete every log row. The retention settings now fall back to their defaults
in that case via a shared `_retention_days` helper.
2026-05-27 16:00:05 +05:30

420 lines
17 KiB
Python
Raw Blame History

# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.
import pytest
from datetime import timedelta
from uuid import uuid4
from django.urls import reverse
from django.utils import timezone
from rest_framework import status
from plane.db.models import APIToken, User
@pytest.mark.contract
class TestApiTokenEndpoint:
"""Test cases for ApiTokenEndpoint"""
# POST /user/api-tokens/ tests
@pytest.mark.django_db
def test_create_api_token_success(self, session_client, create_user, api_token_data):
"""Test successful API token creation"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens")
# Act
response = session_client.post(url, api_token_data, format="json")
# Assert
assert response.status_code == status.HTTP_201_CREATED
assert "token" in response.data
assert response.data["label"] == api_token_data["label"]
assert response.data["description"] == api_token_data["description"]
assert response.data["user_type"] == 0 # Human user
# Verify token was created in database
token = APIToken.objects.get(pk=response.data["id"])
assert token.user == create_user
assert token.label == api_token_data["label"]
@pytest.mark.django_db
def test_create_api_token_for_bot_user(self, session_client, create_bot_user, api_token_data):
"""Test API token creation for bot user"""
# Arrange
session_client.force_authenticate(user=create_bot_user)
url = reverse("api-tokens")
# Act
response = session_client.post(url, api_token_data, format="json")
# Assert
assert response.status_code == status.HTTP_201_CREATED
assert response.data["user_type"] == 1 # Bot user
@pytest.mark.django_db
def test_create_api_token_minimal_data(self, session_client, create_user):
"""Test API token creation with minimal data"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens")
# Act
response = session_client.post(url, {}, format="json")
# Assert
assert response.status_code == status.HTTP_201_CREATED
assert "token" in response.data
assert len(response.data["label"]) == 32 # UUID hex length
assert response.data["description"] == ""
@pytest.mark.django_db
def test_create_api_token_with_expiry(self, session_client, create_user):
"""Test API token creation with expiry date"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens")
future_date = timezone.now() + timedelta(days=30)
data = {"label": "Expiring Token", "expired_at": future_date.isoformat()}
# Act
response = session_client.post(url, data, format="json")
# Assert
assert response.status_code == status.HTTP_201_CREATED
# Verify expiry date was set
token = APIToken.objects.get(pk=response.data["id"])
assert token.expired_at is not None
@pytest.mark.django_db
def test_create_api_token_unauthenticated(self, api_client, api_token_data):
"""Test API token creation without authentication"""
# Arrange
url = reverse("api-tokens")
# Act
response = api_client.post(url, api_token_data, format="json")
# Assert
assert response.status_code == status.HTTP_401_UNAUTHORIZED
# GET /user/api-tokens/ tests
@pytest.mark.django_db
def test_get_all_api_tokens(self, session_client, create_user):
"""Test retrieving all API tokens for user"""
# Arrange
session_client.force_authenticate(user=create_user)
# Create multiple tokens
APIToken.objects.create(label="Token 1", user=create_user, user_type=0)
APIToken.objects.create(label="Token 2", user=create_user, user_type=0)
# Create a service token (should be excluded)
APIToken.objects.create(label="Service Token", user=create_user, user_type=0, is_service=True)
url = reverse("api-tokens")
# Act
response = session_client.get(url)
# Assert
assert response.status_code == status.HTTP_200_OK
assert len(response.data) == 2 # Only non-service tokens
assert all(token["is_service"] is False for token in response.data)
@pytest.mark.django_db
def test_get_empty_api_tokens_list(self, session_client, create_user):
"""Test retrieving API tokens when none exist"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens")
# Act
response = session_client.get(url)
# Assert
assert response.status_code == status.HTTP_200_OK
assert response.data == []
# GET /user/api-tokens/<pk>/ tests
@pytest.mark.django_db
def test_get_specific_api_token(self, session_client, create_user, create_api_token_for_user):
"""Test retrieving a specific API token"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": create_api_token_for_user.pk})
# Act
response = session_client.get(url)
# Assert
assert response.status_code == status.HTTP_200_OK
assert str(response.data["id"]) == str(create_api_token_for_user.pk)
assert response.data["label"] == create_api_token_for_user.label
assert "token" not in response.data # Token should not be visible in read serializer
@pytest.mark.django_db
def test_get_nonexistent_api_token(self, session_client, create_user):
"""Test retrieving a non-existent API token"""
# Arrange
session_client.force_authenticate(user=create_user)
fake_pk = uuid4()
url = reverse("api-tokens-details", kwargs={"pk": fake_pk})
# Act
response = session_client.get(url)
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.django_db
def test_get_other_users_api_token(self, session_client, create_user, db):
"""Test retrieving another user's API token (should fail)"""
# Arrange
# Create another user and their token with unique email and username
unique_id = uuid4().hex[:8]
unique_email = f"other-{unique_id}@plane.so"
unique_username = f"other_user_{unique_id}"
other_user = User.objects.create(email=unique_email, username=unique_username)
other_token = APIToken.objects.create(label="Other Token", user=other_user, user_type=0)
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": other_token.pk})
# Act
response = session_client.get(url)
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
# DELETE /user/api-tokens/<pk>/ tests
@pytest.mark.django_db
def test_delete_api_token_success(self, session_client, create_user, create_api_token_for_user):
"""Test successful API token deletion"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": create_api_token_for_user.pk})
# Act
response = session_client.delete(url)
# Assert
assert response.status_code == status.HTTP_204_NO_CONTENT
assert not APIToken.objects.filter(pk=create_api_token_for_user.pk).exists()
@pytest.mark.django_db
def test_delete_nonexistent_api_token(self, session_client, create_user):
"""Test deleting a non-existent API token"""
# Arrange
session_client.force_authenticate(user=create_user)
fake_pk = uuid4()
url = reverse("api-tokens-details", kwargs={"pk": fake_pk})
# Act
response = session_client.delete(url)
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.django_db
def test_delete_other_users_api_token(self, session_client, create_user, db):
"""Test deleting another user's API token (should fail)"""
# Arrange
# Create another user and their token with unique email and username
unique_id = uuid4().hex[:8]
unique_email = f"delete-other-{unique_id}@plane.so"
unique_username = f"delete_other_user_{unique_id}"
other_user = User.objects.create(email=unique_email, username=unique_username)
other_token = APIToken.objects.create(label="Other Token", user=other_user, user_type=0)
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": other_token.pk})
# Act
response = session_client.delete(url)
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
# Verify token still exists
assert APIToken.objects.filter(pk=other_token.pk).exists()
@pytest.mark.django_db
def test_delete_service_api_token_forbidden(self, session_client, create_user):
"""Test deleting a service API token (should fail)"""
# Arrange
service_token = APIToken.objects.create(label="Service Token", user=create_user, user_type=0, is_service=True)
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": service_token.pk})
# Act
response = session_client.delete(url)
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
# Verify token still exists
assert APIToken.objects.filter(pk=service_token.pk).exists()
# PATCH /user/api-tokens/<pk>/ tests
@pytest.mark.django_db
def test_patch_api_token_success(self, session_client, create_user, create_api_token_for_user):
"""Test successful API token update"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": create_api_token_for_user.pk})
update_data = {
"label": "Updated Token Label",
"description": "Updated description",
}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_200_OK
assert response.data["label"] == update_data["label"]
assert response.data["description"] == update_data["description"]
# Verify database was updated
create_api_token_for_user.refresh_from_db()
assert create_api_token_for_user.label == update_data["label"]
assert create_api_token_for_user.description == update_data["description"]
@pytest.mark.django_db
def test_patch_api_token_partial_update(self, session_client, create_user, create_api_token_for_user):
"""Test partial API token update"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": create_api_token_for_user.pk})
original_description = create_api_token_for_user.description
update_data = {"label": "Only Label Updated"}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_200_OK
assert response.data["label"] == update_data["label"]
assert response.data["description"] == original_description
@pytest.mark.django_db
def test_patch_nonexistent_api_token(self, session_client, create_user):
"""Test updating a non-existent API token"""
# Arrange
session_client.force_authenticate(user=create_user)
fake_pk = uuid4()
url = reverse("api-tokens-details", kwargs={"pk": fake_pk})
update_data = {"label": "New Label"}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.django_db
def test_patch_other_users_api_token(self, session_client, create_user, db):
"""Test updating another user's API token (should fail)"""
# Arrange
# Create another user and their token with unique email and username
unique_id = uuid4().hex[:8]
unique_email = f"patch-other-{unique_id}@plane.so"
unique_username = f"patch_other_user_{unique_id}"
other_user = User.objects.create(email=unique_email, username=unique_username)
other_token = APIToken.objects.create(label="Other Token", user=other_user, user_type=0)
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": other_token.pk})
update_data = {"label": "Hacked Label"}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
# Verify token was not updated
other_token.refresh_from_db()
assert other_token.label == "Other Token"
@pytest.mark.django_db
def test_patch_cannot_modify_token(self, session_client, create_user, create_api_token_for_user):
"""Test that token value cannot be modified via PATCH"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": create_api_token_for_user.pk})
original_token = create_api_token_for_user.token
update_data = {"token": "plane_api_malicious_token_value"}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_200_OK
create_api_token_for_user.refresh_from_db()
assert create_api_token_for_user.token == original_token
@pytest.mark.django_db
def test_patch_cannot_modify_user_type(self, session_client, create_user, create_api_token_for_user):
"""Test that user_type cannot be modified via PATCH"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": create_api_token_for_user.pk})
update_data = {"user_type": 1}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_200_OK
create_api_token_for_user.refresh_from_db()
assert create_api_token_for_user.user_type == 0
@pytest.mark.django_db
def test_patch_cannot_modify_allowed_rate_limit(self, session_client, create_user, create_api_token_for_user):
"""Test that allowed_rate_limit cannot be modified via PATCH"""
# Arrange
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": create_api_token_for_user.pk})
original_rate_limit = create_api_token_for_user.allowed_rate_limit
update_data = {"allowed_rate_limit": "100000/min"}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_200_OK
create_api_token_for_user.refresh_from_db()
assert create_api_token_for_user.allowed_rate_limit == original_rate_limit
@pytest.mark.django_db
def test_patch_cannot_modify_service_token(self, session_client, create_user):
"""Test that service tokens cannot be modified through user token endpoint"""
# Arrange
service_token = APIToken.objects.create(label="Service Token", user=create_user, user_type=0, is_service=True)
session_client.force_authenticate(user=create_user)
url = reverse("api-tokens-details", kwargs={"pk": service_token.pk})
update_data = {"label": "Hacked Service Token"}
# Act
response = session_client.patch(url, update_data, format="json")
# Assert
assert response.status_code == status.HTTP_404_NOT_FOUND
service_token.refresh_from_db()
assert service_token.label == "Service Token"
# Authentication tests
@pytest.mark.django_db
def test_all_endpoints_require_authentication(self, api_client):
"""Test that all endpoints require authentication"""
# Arrange
endpoints = [
(reverse("api-tokens"), "get"),
(reverse("api-tokens"), "post"),
(reverse("api-tokens-details", kwargs={"pk": uuid4()}), "get"),
(reverse("api-tokens-details", kwargs={"pk": uuid4()}), "patch"),
(reverse("api-tokens-details", kwargs={"pk": uuid4()}), "delete"),
]
# Act & Assert
for url, method in endpoints:
response = getattr(api_client, method)(url)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
Reference in New Issue View Git Blame Copy Permalink