diff --git a/src/backend/core/services/converter_services.py b/src/backend/core/services/converter_services.py index 3cd4498da..ee542b1c7 100644 --- a/src/backend/core/services/converter_services.py +++ b/src/backend/core/services/converter_services.py @@ -9,6 +9,7 @@ from django.conf import settings import requests from core.services import mime_types +from core.services.jwt_services import JWTService logger = logging.getLogger(__name__) @@ -109,8 +110,7 @@ class YdocConverter: @property def auth_header(self): """Build microservice authentication header.""" - # Note: Yprovider microservice accepts only raw token, which is not recommended - return f"Bearer {settings.Y_PROVIDER_API_KEY}" + return f"Bearer {JWTService().get_admin_token()}" def _request(self, url, data, content_type, accept): """Make a request to the Y-Provider API.""" diff --git a/src/backend/core/tests/test_api_jwks.py b/src/backend/core/tests/test_api_jwks.py index 3c76bd1fb..348d55a6c 100644 --- a/src/backend/core/tests/test_api_jwks.py +++ b/src/backend/core/tests/test_api_jwks.py @@ -6,11 +6,10 @@ from django.urls import resolve import jwt import pytest -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric import rsa from rest_framework.test import APIClient from core.services.jwt_services import JWTService +from core.tests.utils.jwt import generate_key_pair from core.tests.utils.urls import reload_urls pytestmark = pytest.mark.django_db @@ -18,23 +17,9 @@ pytestmark = pytest.mark.django_db # Private members of a RSA JWK, none of them may ever leak in the JWKS PRIVATE_JWK_MEMBERS = {"d", "p", "q", "dp", "dq", "qi", "oth"} - -def generate_private_key(): - """Generate a PEM encoded RSA private key.""" - return ( - rsa.generate_private_key(public_exponent=65537, key_size=2048) - .private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ) - .decode("utf-8") - ) - - # Generating RSA keys is expensive, do it once for the whole module -PRIVATE_KEY = generate_private_key() -OTHER_PRIVATE_KEY = generate_private_key() +PRIVATE_KEY, _ = generate_key_pair() +OTHER_PRIVATE_KEY, _ = generate_key_pair() @pytest.fixture(name="jwt_settings") diff --git a/src/backend/core/tests/test_services_converter_services.py b/src/backend/core/tests/test_services_converter_services.py index 760504cec..345e4fb24 100644 --- a/src/backend/core/tests/test_services_converter_services.py +++ b/src/backend/core/tests/test_services_converter_services.py @@ -3,6 +3,7 @@ from base64 import b64decode from unittest.mock import MagicMock, patch +import jwt import pytest import requests @@ -12,13 +13,28 @@ from core.services.converter_services import ( ValidationError, YdocConverter, ) +from core.tests.utils.jwt import generate_key_pair + +# Generating an RSA key is expensive, do it once for the whole module +PRIVATE_KEY, PUBLIC_KEY = generate_key_pair() -def test_auth_header(settings): - """Test authentication header generation.""" - settings.Y_PROVIDER_API_KEY = "test-key" +@pytest.fixture(autouse=True) +def jwt_settings(settings): + """Setup valid settings for the JWT service used to sign the auth header.""" + settings.JWT_PRIVATE_KEY = PRIVATE_KEY + settings.JWT_TOKEN_LIFETIME = 3600 + + +def test_auth_header(): + """The auth header carries an admin JWT signed with the configured key.""" converter = YdocConverter() - assert converter.auth_header == "Bearer test-key" + + scheme, token = converter.auth_header.split(" ") + + assert scheme == "Bearer" + payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"]) + assert payload["admin"] is True def test_convert_empty_text(): @@ -63,12 +79,12 @@ def test_convert_full_integration(mock_post, settings): """Test full integration with all settings.""" settings.Y_PROVIDER_API_BASE_URL = "http://test.com/" - settings.Y_PROVIDER_API_KEY = "test-key" settings.CONVERSION_API_ENDPOINT = "conversion-endpoint" settings.CONVERSION_API_TIMEOUT = 5 settings.CONVERSION_API_CONTENT_FIELD = "content" converter = YdocConverter() + auth_header = converter.auth_header expected_content = b"converted content" mock_response = MagicMock() @@ -83,7 +99,7 @@ def test_convert_full_integration(mock_post, settings): "http://test.com/conversion-endpoint/", data="test markdown", headers={ - "Authorization": "Bearer test-key", + "Authorization": auth_header, "Content-Type": mime_types.MARKDOWN, "Accept": mime_types.YJS, }, @@ -96,12 +112,12 @@ def test_convert_full_integration(mock_post, settings): def test_convert_full_integration_with_specific_headers(mock_post, settings): """Test successful conversion with specific content type and accept headers.""" settings.Y_PROVIDER_API_BASE_URL = "http://test.com/" - settings.Y_PROVIDER_API_KEY = "test-key" settings.CONVERSION_API_ENDPOINT = "conversion-endpoint" settings.CONVERSION_API_TIMEOUT = 5 settings.CONVERSION_API_SECURE = False converter = YdocConverter() + auth_header = converter.auth_header expected_response = "# Test Document\n\nThis is test content." mock_response = MagicMock() @@ -116,7 +132,7 @@ def test_convert_full_integration_with_specific_headers(mock_post, settings): "http://test.com/conversion-endpoint/", data=b"test_content", headers={ - "Authorization": "Bearer test-key", + "Authorization": auth_header, "Content-Type": mime_types.YJS, "Accept": mime_types.MARKDOWN, }, diff --git a/src/backend/core/tests/test_services_jwt_services.py b/src/backend/core/tests/test_services_jwt_services.py index be8b58418..f2bc8f2b2 100644 --- a/src/backend/core/tests/test_services_jwt_services.py +++ b/src/backend/core/tests/test_services_jwt_services.py @@ -10,8 +10,6 @@ from django.core.cache import cache import jwt import pytest -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric import rsa from freezegun import freeze_time from core.services.jwt_services import ( @@ -19,26 +17,7 @@ from core.services.jwt_services import ( JWTService, TokenGenerationError, ) - - -def generate_key_pair(): - """Generate a PEM encoded RSA key pair to sign and verify test tokens.""" - private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) - private_pem = private_key.private_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PrivateFormat.PKCS8, - encryption_algorithm=serialization.NoEncryption(), - ).decode("utf-8") - public_pem = ( - private_key.public_key() - .public_bytes( - encoding=serialization.Encoding.PEM, - format=serialization.PublicFormat.SubjectPublicKeyInfo, - ) - .decode("utf-8") - ) - return private_pem, public_pem - +from core.tests.utils.jwt import generate_key_pair # Generating RSA keys is expensive, do it once for the whole module PRIVATE_KEY, PUBLIC_KEY = generate_key_pair() diff --git a/src/backend/core/tests/utils/jwt.py b/src/backend/core/tests/utils/jwt.py new file mode 100644 index 000000000..f97bb18ea --- /dev/null +++ b/src/backend/core/tests/utils/jwt.py @@ -0,0 +1,23 @@ +"""Utils for testing JWT-signed tokens.""" + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa + + +def generate_key_pair(): + """Generate a PEM encoded RSA key pair to sign and verify test tokens.""" + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + private_pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode("utf-8") + public_pem = ( + private_key.public_key() + .public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ) + .decode("utf-8") + ) + return private_pem, public_pem