mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-08-27 12:30:13 +03:00
0e8d0cb87b
- New config: Controller.jwt_refresh_token_expire_minutes (default 30 days) - New endpoint: POST /v3/access/users/refresh (public, unauthenticated) - Login/authenticate responses now include refresh_token - AuthService: _create_token helper, create_refresh_token, get_token_data now parses type claim (token_use) for token classification - Security: refresh tokens rejected on HTTP + WebSocket access paths; /refresh strictly requires type=='refresh' - Logout works for free via existing token_version mechanism - Tests: 9 new TestRefreshToken cases, all passing; 34 existing tests still pass (no regressions)
108 lines
4.8 KiB
Python
108 lines
4.8 KiB
Python
#
|
|
# Copyright (C) 2020 GNS3 Technologies Inc.
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
from joserfc import jwt
|
|
from joserfc.jwk import OctKey
|
|
from joserfc.errors import JoseError
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
import bcrypt
|
|
|
|
from typing import Optional
|
|
from fastapi import HTTPException, status
|
|
from gns3server.schemas.controller.tokens import TokenData
|
|
from gns3server.config import Config
|
|
from pydantic import ValidationError
|
|
|
|
import logging
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
DEFAULT_JWT_SECRET_KEY = "efd08eccec3bd0a1be2e086670e5efa90969c68d07e072d7354a76cea5e33d4e"
|
|
|
|
|
|
class AuthService:
|
|
|
|
def hash_password(self, password: str) -> str:
|
|
|
|
salt = bcrypt.gensalt()
|
|
hashed_password = bcrypt.hashpw(password=password.encode('utf-8'), salt=salt)
|
|
return hashed_password.decode('utf-8')
|
|
|
|
def verify_password(self, password, hashed_password) -> bool:
|
|
|
|
return bcrypt.checkpw(password=password.encode('utf-8'), hashed_password=hashed_password.encode('utf-8'))
|
|
|
|
def _create_token(self, username, token_version, token_type, expires_in, secret_key=None) -> str:
|
|
"""Shared helper to create any kind of signed JWT token."""
|
|
|
|
expire = datetime.now(timezone.utc) + timedelta(minutes=expires_in)
|
|
to_encode = {"sub": username, "exp": expire, "ver": token_version, "type": token_type}
|
|
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)
|
|
encoded_jwt = jwt.encode({"alg": algorithm}, to_encode, key)
|
|
return encoded_jwt
|
|
|
|
def create_access_token(self, username, token_version: int = 0, secret_key: str = None, expires_in: int = 0) -> str:
|
|
|
|
if not expires_in:
|
|
expires_in = Config.instance().settings.Controller.jwt_access_token_expire_minutes
|
|
return self._create_token(username, token_version, "access", expires_in, secret_key)
|
|
|
|
def create_refresh_token(self, username, token_version: int = 0, secret_key: str = None, expires_in: int = 0) -> str:
|
|
|
|
if not expires_in:
|
|
expires_in = Config.instance().settings.Controller.jwt_refresh_token_expire_minutes
|
|
return self._create_token(username, token_version, "refresh", expires_in, secret_key)
|
|
|
|
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"},
|
|
)
|
|
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
|
|
# 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
|
|
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
|
|
return token_data
|
|
|
|
def get_username_from_token(self, token: str, secret_key: str = None) -> Optional[str]:
|
|
return self.get_token_data(token, secret_key).username
|