mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-08-27 12:30:13 +03:00
auth: split JWT validation errors by failure cause
get_token_data used to raise the same "Could not validate credentials" for every JWT-level failure (bad signature, expired, malformed), which made console WebSocket auth failures impossible to tell apart. Return a distinct detail per cause and log the underlying exception plus the unverified header alg value on rejection.
This commit is contained in:
parent
fd7594f62e
commit
704b5d80c2
@ -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", "<missing>"))
|
||||
except Exception:
|
||||
return "<undecodable>"
|
||||
|
||||
|
||||
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]:
|
||||
|
||||
@ -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)'
|
||||
}
|
||||
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user