feat(api): return token lifetimes and document API keys as bearer tokens

gns3server/api/routes/controller/dependencies/authentication.py

@@ -36,7 +36,12 @@ from gns3server.services.access_tickets import TICKET_PREFIX
 from .database import get_repository
 
 log = logging.getLogger(__name__)
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/v3/access/users/login", auto_error=False)
+oauth2_scheme = OAuth2PasswordBearer(
+    tokenUrl="/v3/access/users/login",
+    auto_error=False,
+    description="Send the access token as a bearer token. "
+    "An API key (gns3_<api_key_id>_<secret>) is also accepted as the bearer token.",
+)
 
 
 def _reject_refresh_token(token_data) -> None:

gns3server/api/routes/controller/users.py

@@ -65,12 +65,7 @@ async def login(
             headers={"WWW-Authenticate": "Bearer"},
         )
 
-    token = schemas.Token(
+    return auth_service.create_token_response(user)
-        access_token=auth_service.create_access_token(user.username, token_version=user.token_version),
-        token_type="bearer",
-        refresh_token=auth_service.create_refresh_token(user.username, token_version=user.token_version),
-    )
-    return token
 
 
 @router.post("/authenticate", response_model=schemas.Token)
@@ -91,12 +86,7 @@ async def authenticate(
             headers={"WWW-Authenticate": "Bearer"},
         )
 
-    token = schemas.Token(
+    return auth_service.create_token_response(user)
-        access_token=auth_service.create_access_token(user.username, token_version=user.token_version),
-        token_type="bearer",
-        refresh_token=auth_service.create_refresh_token(user.username, token_version=user.token_version),
-    )
-    return token
 
 
 @router.post("/refresh", response_model=schemas.Token)
@@ -135,11 +125,7 @@ async def refresh_access_token(
             headers={"WWW-Authenticate": "Bearer"},
         )
 
-    return schemas.Token(
+    return auth_service.create_token_response(user)
-        access_token=auth_service.create_access_token(user.username, token_version=user.token_version),
-        token_type="bearer",
-        refresh_token=auth_service.create_refresh_token(user.username, token_version=user.token_version),
-    )
 
 
 @router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)

gns3server/schemas/controller/tokens.py

@@ -22,7 +22,9 @@ from pydantic import BaseModel
 class Token(BaseModel):
     access_token: str
     token_type: str
+    expires_in: int
     refresh_token: Optional[str] = None
+    refresh_expires_in: Optional[int] = None
 
 
 class TokenData(BaseModel):

gns3server/services/authentication.py

@@ -29,7 +29,7 @@ from joserfc.jwk import OctKey
 from pydantic import ValidationError
 
 from gns3server.config import Config
-from gns3server.schemas.controller.tokens import TokenData
+from gns3server.schemas.controller.tokens import Token, TokenData
 
 log = logging.getLogger(__name__)
 
@@ -47,6 +47,9 @@ def _extract_alg(token: str) -> str:
         return "<undecodable>"
 
 
+BEARER_TOKEN_TYPE = "bearer"
+
+
 class AuthService:
     def hash_password(self, password: str) -> str:
 
@@ -61,8 +64,9 @@ class AuthService:
     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)
+        issued_at = datetime.now(timezone.utc)
-        to_encode = {"sub": username, "exp": expire, "ver": token_version, "type": token_type}
+        expire = issued_at + timedelta(minutes=expires_in)
+        to_encode = {"sub": username, "iat": issued_at, "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:
@@ -89,6 +93,17 @@ class AuthService:
             expires_in = Config.instance().settings.Controller.jwt_refresh_token_expire_minutes
         return self._create_token(username, token_version, "refresh", expires_in, secret_key)
 
+    def create_token_response(self, user) -> Token:
+
+        settings = Config.instance().settings.Controller
+        return Token(
+            access_token=self.create_access_token(user.username, token_version=user.token_version),
+            token_type=BEARER_TOKEN_TYPE,
+            expires_in=settings.jwt_access_token_expire_minutes * 60,
+            refresh_token=self.create_refresh_token(user.username, token_version=user.token_version),
+            refresh_expires_in=settings.jwt_refresh_token_expire_minutes * 60,
+        )
+
     def get_token_data(self, token: str, secret_key: Optional[str] = None) -> TokenData:
 
         def auth_error(detail: str) -> HTTPException:

tests/api/routes/controller/test_users.py

@@ -446,6 +446,74 @@ class TestLogout:
         assert response.status_code == status.HTTP_401_UNAUTHORIZED
 
 
+class TestTokenLifetime:
+    async def _issue(self, app, client, user, endpoint):
+        credentials = {"username": user.username, "password": "user1_password"}
+        if endpoint == "login":
+            return await client.post(
+                app.url_path_for("login"),
+                data=credentials,
+                headers={"content-type": "application/x-www-form-urlencoded"},
+            )
+        auth = await client.post(app.url_path_for("authenticate"), json=credentials)
+        if endpoint == "authenticate":
+            return auth
+        return await client.post(
+            app.url_path_for("refresh_access_token"), json={"refresh_token": auth.json()["refresh_token"]}
+        )
+
+    @pytest.mark.parametrize("endpoint", ("login", "authenticate", "refresh_access_token"))
+    async def test_token_response_includes_lifetimes(
+        self,
+        app: FastAPI,
+        unauthorized_client: AsyncClient,
+        test_user: User,
+        config: Config,
+        endpoint: str,
+    ) -> None:
+
+        response = await self._issue(app, unauthorized_client, test_user, endpoint)
+        assert response.status_code == status.HTTP_200_OK
+        body = response.json()
+        controller = config.settings.Controller
+        assert body["expires_in"] == controller.jwt_access_token_expire_minutes * 60
+        assert body["refresh_expires_in"] == controller.jwt_refresh_token_expire_minutes * 60
+        assert body["access_token"]
+        assert body["token_type"] == "bearer"
+        assert body["refresh_token"]
+
+    @pytest.mark.parametrize("endpoint", ("login", "authenticate", "refresh_access_token"))
+    async def test_expires_in_matches_jwt_claims(
+        self,
+        app: FastAPI,
+        unauthorized_client: AsyncClient,
+        test_user: User,
+        config: Config,
+        endpoint: str,
+    ) -> None:
+
+        response = await self._issue(app, unauthorized_client, test_user, endpoint)
+        body = response.json()
+        key = OctKey.import_key(config.settings.Controller.jwt_secret_key)
+        access = jwt.decode(body["access_token"], key, algorithms=["HS256"]).claims
+        refresh = jwt.decode(body["refresh_token"], key, algorithms=["HS256"]).claims
+        assert access["exp"] - access["iat"] == body["expires_in"]
+        assert refresh["exp"] - refresh["iat"] == body["refresh_expires_in"]
+
+    async def test_openapi_declares_token_lifetimes_as_integers(self, app: FastAPI) -> None:
+
+        schema = app.openapi()["components"]["schemas"]["Token"]
+        assert schema["properties"]["expires_in"]["type"] == "integer"
+        refresh_expires_in = schema["properties"]["refresh_expires_in"]
+        assert "integer" in str(refresh_expires_in)
+        assert "expires_in" in schema["required"]
+
+    async def test_openapi_security_scheme_mentions_api_keys(self, app: FastAPI) -> None:
+
+        schemes = app.openapi()["components"]["securitySchemes"]
+        assert any("API key" in scheme.get("description", "") for scheme in schemes.values())
+
+
 class TestRefreshToken:
     async def test_login_returns_refresh_token(
         self,