diff --git a/docs/exchange-key-rotation.md b/docs/exchange-key-rotation.md new file mode 100644 index 0000000..2f191fe --- /dev/null +++ b/docs/exchange-key-rotation.md @@ -0,0 +1,36 @@ +# Ротация ключа обмена + +`EXCHANGE_SHARED_TOKEN` служит текущим ключом HTTP-доступа к загрузке и материалом +ключа шифрования пакетов. Его нельзя передавать frontend или публиковать в +runtime-config.js. Публичный маркер браузерной загрузки требует отдельной +серверной проверки JWT активного администратора; сам по себе он доступа не даёт. + +При согласованной ротации Мостовик и State Corp получают новый общий ключ и новый +`EXCHANGE_KEY_ID` (у Мостовика — `STATE_CORP_EXCHANGE_KEY_ID`). Чтобы ранее +подготовленные архивы оставались доступны, State Corp может хранить одну прежнюю +пару только на сервере: + +- `EXCHANGE_PREVIOUS_SHARED_TOKEN` — прежний ключ расшифровки; +- `EXCHANGE_PREVIOUS_KEY_ID` — точный `key_id` прежних пакетов. + +Оба предыдущих параметра задаются вместе, предыдущий ID должен отличаться от +текущего. Неполная или неоднозначная конфигурация отклоняет импорт. Старый ключ +выбирается только для точного предыдущего `key_id`; перебора ключей при ошибке +расшифровки нет. При пустых предыдущих параметрах сохраняется прежнее поведение +с единственным текущим ключом, в том числе для контейнеров без `key_id`. + +HTTP-загрузка всегда проверяет только текущий `EXCHANGE_SHARED_TOKEN`. +Предыдущий ключ не предоставляет доступа к API. При загрузке старого архива +с текущими правами действуют прежние правила проверки, атомарного импорта и +распознавания дубликатов; миграции и преобразование предметных записей не нужны. + +Порядок применения: убрать секрет из браузерной конфигурации, дождаться окончания +активных задач отправителя, приватно сохранить текущую конфигурацию, установить +предыдущую пару у получателя и новую текущую пару у обоих участников, пересоздать +соответствующие процессы. Проверить отказ старого HTTP-ключа, загрузку текущим +ключом старого и нового архивов, а также повтор без изменения предметных данных. +Прежний ключ удаляют после согласованного срока хранения старых архивов; после +удаления такие архивы перестанут расшифровываться автоматически. + +Проверки: `uv run pytest tests/apps/exchange/`. Настройки, ключи и старые архивы +не должны попадать в репозиторий, логи или публичные файлы. diff --git a/src/apps/exchange/services.py b/src/apps/exchange/services.py index 34e3e83..e1872aa 100644 --- a/src/apps/exchange/services.py +++ b/src/apps/exchange/services.py @@ -412,6 +412,30 @@ class ExchangePackageImportService: return header, bin_bytes[header_end:] + @classmethod + def _decryption_token(cls, header: dict[str, Any]) -> str: + token = str(getattr(settings, "EXCHANGE_SHARED_TOKEN", "") or "").strip() + if not token: + raise ExchangeImportError("EXCHANGE_SHARED_TOKEN не настроен") + + previous_token = str( + getattr(settings, "EXCHANGE_PREVIOUS_SHARED_TOKEN", "") or "" + ).strip() + previous_id = str( + getattr(settings, "EXCHANGE_PREVIOUS_KEY_ID", "") or "" + ).strip() + current_id = str(getattr(settings, "EXCHANGE_KEY_ID", "") or "").strip() + if previous_token or previous_id: + if not previous_token or not previous_id or previous_id == current_id: + raise ExchangeImportError( + "Некорректно настроен предыдущий ключ расшифровки обмена" + ) + if header.get("key_id") == previous_id: + return previous_token + # Preserve the existing current-key contract, including old containers + # without key_id. Never try the previous secret as a decryption fallback. + return token + @classmethod def _decrypt_payload( cls, @@ -419,10 +443,7 @@ class ExchangePackageImportService: header: dict[str, Any], encrypted_payload: bytes, ) -> dict[str, Any]: - token = str(getattr(settings, "EXCHANGE_SHARED_TOKEN", "") or "").strip() - if not token: - raise ExchangeImportError("EXCHANGE_SHARED_TOKEN не настроен") - + token = cls._decryption_token(header) raw_key = hashlib.sha256(token.encode("utf-8")).digest() nonce = cls._decode_base64_field(header, "nonce") aad = cls._decode_base64_field(header, "aad") diff --git a/src/settings/base.py b/src/settings/base.py index 7344c2b..74c0b3c 100644 --- a/src/settings/base.py +++ b/src/settings/base.py @@ -19,6 +19,9 @@ BACKUP_ENCRYPTION_KEY = os.getenv("BACKUP_ENCRYPTION_KEY", "") BACKUP_KEY_ID = os.getenv("BACKUP_KEY_ID", "default") EXCHANGE_SHARED_TOKEN = os.getenv("EXCHANGE_SHARED_TOKEN", "") EXCHANGE_KEY_ID = os.getenv("EXCHANGE_KEY_ID", "dev-shared-token") +# Read old encrypted archives after rotation; never accepted as HTTP credentials. +EXCHANGE_PREVIOUS_SHARED_TOKEN = os.getenv("EXCHANGE_PREVIOUS_SHARED_TOKEN", "") +EXCHANGE_PREVIOUS_KEY_ID = os.getenv("EXCHANGE_PREVIOUS_KEY_ID", "") warnings.filterwarnings( "ignore", diff --git a/tests/apps/exchange/test_api.py b/tests/apps/exchange/test_api.py index f5fcfd4..4049a6a 100644 --- a/tests/apps/exchange/test_api.py +++ b/tests/apps/exchange/test_api.py @@ -58,6 +58,8 @@ def build_exchange_archive( bin_name: str = "exchange_package_20260407.bin", data: dict[str, list[dict[str, object]]] | None = None, schema_version: int = ExchangePackageImportService.SUPPORTED_SCHEMA_VERSION, + token: str = TEST_TOKEN, + key_id: str = "test-shared-token", ) -> SimpleUploadedFile: """Build encrypted exchange archive compatible with import service.""" provided_data = data or {} @@ -92,13 +94,13 @@ def build_exchange_archive( compressed_payload = zlib.compress(payload_bytes, level=9) nonce = b"sc-exch-0001" aad = ExchangePackageImportService.AAD - raw_key = hashlib.sha256(TEST_TOKEN.encode("utf-8")).digest() + raw_key = hashlib.sha256(token.encode("utf-8")).digest() encrypted_payload = AESGCM(raw_key).encrypt(nonce, compressed_payload, aad) header = { "format": ExchangePackageImportService.BIN_FORMAT, "version": 1, - "key_id": "test-shared-token", + "key_id": key_id, "nonce": _b64url(nonce), "aad": _b64url(aad), "package_id": package_id, diff --git a/tests/apps/exchange/test_key_rotation.py b/tests/apps/exchange/test_key_rotation.py new file mode 100644 index 0000000..d5c8d6e --- /dev/null +++ b/tests/apps/exchange/test_key_rotation.py @@ -0,0 +1,95 @@ +"""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())