diff --git a/gns3server/services/authentication.py b/gns3server/services/authentication.py index 574c96c69..6103ee6f9 100644 --- a/gns3server/services/authentication.py +++ b/gns3server/services/authentication.py @@ -16,7 +16,9 @@ from joserfc import jwt from joserfc.jwk import OctKey -from joserfc.errors import JoseError +from joserfc.errors import JoseError, BadSignatureError +import base64 +import json import time from datetime import datetime, timedelta, timezone import bcrypt @@ -34,6 +36,17 @@ log = logging.getLogger(__name__) DEFAULT_JWT_SECRET_KEY = "efd08eccec3bd0a1be2e086670e5efa90969c68d07e072d7354a76cea5e33d4e" +def _extract_alg(token: str) -> str: + """Best-effort extraction of the unverified JWT header "alg" value — for logging only.""" + + try: + header_segment = token.split(".", 1)[0] + header = json.loads(base64.urlsafe_b64decode(header_segment + "=" * (-len(header_segment) % 4))) + return str(header.get("alg", "")) + except Exception: + return "" + + class AuthService: def hash_password(self, password: str) -> str: @@ -75,32 +88,38 @@ class AuthService: def get_token_data(self, token: str, secret_key: str = None) -> TokenData: - credentials_exception = HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Could not validate credentials", - headers={"WWW-Authenticate": "Bearer"}, - ) + def auth_error(detail: str) -> HTTPException: + return HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail=detail, + headers={"WWW-Authenticate": "Bearer"}, + ) + + if secret_key is None: + secret_key = Config.instance().settings.Controller.jwt_secret_key + if secret_key is None: + secret_key = DEFAULT_JWT_SECRET_KEY + log.error("A JWT secret key must be configured to secure the server, using an unsecured default key!") + algorithm = Config.instance().settings.Controller.jwt_algorithm + key = OctKey.import_key(secret_key) try: - if secret_key is None: - secret_key = Config.instance().settings.Controller.jwt_secret_key - if secret_key is None: - secret_key = DEFAULT_JWT_SECRET_KEY - log.error("A JWT secret key must be configured to secure the server, using an unsecured default key!") - algorithm = Config.instance().settings.Controller.jwt_algorithm - key = OctKey.import_key(secret_key) payload = jwt.decode(token, key, algorithms=[algorithm]) username: str = payload.claims.get("sub") if username is None: - raise credentials_exception + raise auth_error("Invalid token: missing subject claim") # Validate the exp claim — joserfc does not validate time-based claims by default token_exp: int = payload.claims.get("exp", 0) if token_exp and time.time() > token_exp: - raise credentials_exception + raise auth_error("Token has expired") token_version: int = payload.claims.get("ver", 0) token_use: str = payload.claims.get("type", "access") token_data = TokenData(username=username, token_version=token_version, token_use=token_use) - except (JoseError, ValidationError, ValueError): - raise credentials_exception + except BadSignatureError as e: + log.error("JWT rejected: bad signature (header alg: '%s', error: %s)", _extract_alg(token), e) + raise auth_error("Invalid token signature") + except (JoseError, ValidationError, ValueError) as e: + log.error("JWT rejected: %s: %s (header alg: '%s')", type(e).__name__, e, _extract_alg(token)) + raise auth_error(f"Invalid token ({type(e).__name__})") return token_data def get_username_from_token(self, token: str, secret_key: str = None) -> Optional[str]: diff --git a/tests/api/routes/test_routes.py b/tests/api/routes/test_routes.py index 5c8a59166..0854581f7 100644 --- a/tests/api/routes/test_routes.py +++ b/tests/api/routes/test_routes.py @@ -106,7 +106,7 @@ class TestRoutes: async with aconnect_ws(path, client, params=params) as ws: json_notification = await ws.receive_json() assert json_notification['event'] == { - 'message': 'Could not authenticate while connecting to controller WebSocket: Could not validate credentials' + 'message': 'Could not authenticate while connecting to controller WebSocket: Invalid token (DecodeError)' }