-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[DOP-23149] Replace python-jose with authlib.jose
- Loading branch information
Showing
5 changed files
with
23 additions
and
19 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,29 +1,31 @@ | ||
# SPDX-FileCopyrightText: 2023-2025 MTS PJSC | ||
# SPDX-License-Identifier: Apache-2.0 | ||
|
||
from jose import ExpiredSignatureError, JWTError, jwt | ||
from authlib.jose import JsonWebToken | ||
from authlib.jose.errors import ExpiredTokenError, JoseError | ||
|
||
from horizon.commons.exceptions import AuthorizationError | ||
|
||
|
||
def sign_jwt(payload: dict, secret_key: str, security_algorithm: str) -> str: | ||
jwt = JsonWebToken([security_algorithm]) | ||
return jwt.encode( | ||
payload, | ||
secret_key, | ||
algorithm=security_algorithm, | ||
) | ||
header={"alg": security_algorithm}, | ||
payload=payload, | ||
key=secret_key, | ||
).decode("utf-8") | ||
|
||
|
||
def decode_jwt(token: str, secret_key: str, security_algorithm: str) -> dict: | ||
jwt = JsonWebToken([security_algorithm]) | ||
try: | ||
result = jwt.decode( | ||
token, | ||
secret_key, | ||
algorithms=[security_algorithm], | ||
key=secret_key, | ||
) | ||
if "exp" not in result: | ||
raise ExpiredSignatureError("Missing expiration time in token") | ||
raise ExpiredTokenError("Missing expiration time in token") | ||
|
||
return result | ||
except JWTError as e: | ||
except JoseError as e: | ||
raise AuthorizationError("Invalid token") from e |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -141,6 +141,7 @@ backend = [ | |
"pydantic-settings", | ||
"devtools", | ||
"passlib", | ||
"authlib", | ||
] | ||
postgres = [ | ||
"asyncpg", | ||
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,23 +1,23 @@ | ||
from __future__ import annotations | ||
|
||
import pytest | ||
from jose import ExpiredSignatureError, JWTError | ||
from authlib.jose.errors import ExpiredTokenError, JoseError | ||
|
||
from horizon.client.auth import AccessToken | ||
|
||
pytestmark = [pytest.mark.client_sync, pytest.mark.client] | ||
|
||
|
||
def test_access_token_constructor_expired(access_token_expired: AccessToken): | ||
with pytest.raises(ExpiredSignatureError): | ||
with pytest.raises(ExpiredTokenError): | ||
AccessToken(token=access_token_expired) | ||
|
||
|
||
def test_access_token_constructor_no_expiration_time(access_token_no_expiration_time: AccessToken): | ||
with pytest.raises(ExpiredSignatureError): | ||
with pytest.raises(ExpiredTokenError): | ||
AccessToken(token=access_token_no_expiration_time) | ||
|
||
|
||
def test_access_token_constructor_malformed(access_token_malformed: AccessToken): | ||
with pytest.raises(JWTError): | ||
with pytest.raises(JoseError): | ||
AccessToken(token=access_token_malformed) |