78412f9e3b
All checks were successful
State Corp Backend CI/CD / Quality gate (push) Successful in 3m19s
State Corp Backend CI/CD / Build linux/amd64 images once (push) Successful in 2m54s
State Corp Backend CI/CD / Refresh and release internal main (push) Has been skipped
State Corp Backend CI/CD / Release customer main (push) Has been skipped
State Corp Backend CI/CD / Release dev (push) Successful in 54s
96 lines
3.8 KiB
Python
96 lines
3.8 KiB
Python
"""Old encrypted archives survive rotation without retaining old HTTP access."""
|
|
|
|
from apps.exchange.models import ExchangePackageImport
|
|
from apps.organization.models import Organization
|
|
from django.test import override_settings
|
|
from django.urls import reverse
|
|
from django.utils.crypto import get_random_string
|
|
from rest_framework.test import APITestCase
|
|
|
|
from tests.apps.exchange.test_api import (
|
|
TEST_TOKEN,
|
|
build_exchange_archive,
|
|
build_exchange_payload,
|
|
)
|
|
|
|
CURRENT_TOKEN = get_random_string(32)
|
|
CURRENT_KEY_ID = "current-test-key"
|
|
PREVIOUS_KEY_ID = "test-shared-token"
|
|
|
|
|
|
@override_settings(
|
|
EXCHANGE_SHARED_TOKEN=CURRENT_TOKEN,
|
|
EXCHANGE_KEY_ID=CURRENT_KEY_ID,
|
|
EXCHANGE_PREVIOUS_SHARED_TOKEN=TEST_TOKEN,
|
|
EXCHANGE_PREVIOUS_KEY_ID=PREVIOUS_KEY_ID,
|
|
)
|
|
class ExchangeKeyRotationTest(APITestCase):
|
|
def upload(self, *, token=TEST_TOKEN, key_id=PREVIOUS_KEY_ID, header=CURRENT_TOKEN):
|
|
return self.client.post(
|
|
reverse("api_v1:exchange:package-upload"),
|
|
{
|
|
"file": build_exchange_archive(
|
|
data=build_exchange_payload(), token=token, key_id=key_id
|
|
)
|
|
},
|
|
format="multipart",
|
|
HTTP_X_EXCHANGE_TOKEN=header,
|
|
)
|
|
|
|
def test_previous_archive_with_current_auth_imports_and_deduplicates(self):
|
|
first = self.upload()
|
|
self.assertEqual(first.status_code, 201)
|
|
self.assertFalse(first.data["result"]["duplicate"])
|
|
count = Organization.objects.count()
|
|
repeated = self.upload()
|
|
self.assertEqual(repeated.status_code, 201)
|
|
self.assertTrue(repeated.data["result"]["duplicate"])
|
|
self.assertEqual(
|
|
repeated.data["result"]["duplicate_of"],
|
|
first.data["result"]["import_id"],
|
|
)
|
|
self.assertEqual(Organization.objects.count(), count)
|
|
|
|
def test_previous_token_never_authorizes_http(self):
|
|
response = self.upload(header=TEST_TOKEN)
|
|
self.assertEqual(response.status_code, 401)
|
|
self.assertFalse(ExchangePackageImport.objects.exists())
|
|
self.assertFalse(Organization.objects.exists())
|
|
|
|
def test_current_archive_uses_current_key(self):
|
|
self.assertEqual(
|
|
self.upload(token=CURRENT_TOKEN, key_id=CURRENT_KEY_ID).status_code, 201
|
|
)
|
|
|
|
def test_current_key_retains_legacy_header_id_compatibility(self):
|
|
for key_id in ("", "legacy-client-id"):
|
|
with self.subTest(key_id=key_id):
|
|
self.assertEqual(
|
|
self.upload(token=CURRENT_TOKEN, key_id=key_id).status_code, 201
|
|
)
|
|
|
|
def test_previous_key_requires_its_exact_header_id(self):
|
|
self.assertEqual(self.upload(key_id="unknown-key").status_code, 400)
|
|
self.assertFalse(ExchangePackageImport.objects.exists())
|
|
|
|
def test_previous_key_id_never_falls_back_to_current_key(self):
|
|
self.assertEqual(self.upload(token=CURRENT_TOKEN).status_code, 400)
|
|
self.assertFalse(ExchangePackageImport.objects.exists())
|
|
|
|
@override_settings(EXCHANGE_PREVIOUS_SHARED_TOKEN="", EXCHANGE_PREVIOUS_KEY_ID="")
|
|
def test_previous_archive_requires_explicit_server_configuration(self):
|
|
self.assertEqual(self.upload().status_code, 400)
|
|
self.assertFalse(ExchangePackageImport.objects.exists())
|
|
|
|
def test_invalid_previous_configuration_fails_closed(self):
|
|
for configuration in (
|
|
{"EXCHANGE_PREVIOUS_SHARED_TOKEN": ""},
|
|
{"EXCHANGE_PREVIOUS_KEY_ID": ""},
|
|
{"EXCHANGE_PREVIOUS_KEY_ID": CURRENT_KEY_ID},
|
|
):
|
|
with self.subTest(configuration=tuple(configuration)):
|
|
with override_settings(**configuration):
|
|
response = self.upload(token=CURRENT_TOKEN, key_id=CURRENT_KEY_ID)
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertFalse(ExchangePackageImport.objects.exists())
|