mirror of
https://github.com/suitenumerique/docs.git
synced 2026-09-11 12:17:55 +02:00
🛂(django) use jwt token for converter services
The Y_PROVIDER_API_KEY shared secret is replaced by a signed admin JWT when Django calls the y-provider conversion endpoint.
This commit is contained in:
@@ -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."""
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user