feat: avoid starting login flow unless the tokens are expired (#304)

Author: Ariana-Hlavaty

DEPENDENCIES.md

@@ -14,3 +14,4 @@
 | `pydantic-settings` | `>=2, <3.0` | Settings management using Pydantic | [https://pypi.org/project/pydantic-settings/](https://pypi.org/project/pydantic-settings/) |
 | `quantinuum-schemas` | `>=7.3, <8.0` | Shared data models for Quantinuum. | [https://github.com/CQCL/quantinuum-schemas](https://github.com/CQCL/quantinuum-schemas) |
 | `hugr` | `>=0.14.0, <1.0.0` | Quantinuum's common representation for quantum programs | [https://github.com/CQCL/hugr/tree/main/hugr-py](https://github.com/CQCL/hugr/tree/main/hugr-py) |
+| `pyjwt` | `>=2.10.1,<3.0.0` | JSON Web Token implementation in Python | [https://pypi.org/project/PyJWT/](https://pypi.org/project/PyJWT/) |

integration/test_auth_flows.py

@@ -1,5 +1,6 @@
 """Test basic functionality relating to the auth module."""
 
+from contextlib import redirect_stdout
 from io import StringIO
 from typing import Any
 
@@ -89,3 +90,24 @@ def test_domain_switch() -> None:
     assert original_domain in str(get_nexus_client().base_url)
 
     qnx.users.get_self()
+
+
+def test_login_when_already_logged_in(monkeypatch: Any) -> None:
+    """Test that logging in when already logged in notifies the user appropriately."""
+    username = CONFIG.qa_user_email
+    pwd = CONFIG.qa_user_password
+
+    # Ensure logged out first
+    qnx.logout()
+    # First login
+    monkeypatch.setattr("sys.stdin", StringIO(username + "\n"))
+    monkeypatch.setattr("getpass.getpass", lambda prompt: pwd)
+    qnx.login_with_credentials()
+
+    # Try to login again, should indicate already logged in
+    # Capture output if function prints, or check for raised exception/message
+    output = StringIO()
+    with redirect_stdout(output):
+        qnx.login_with_credentials()
+    out_str = output.getvalue()
+    assert "already logged in" in out_str.lower()

pyproject.toml

@@ -23,6 +23,7 @@ dependencies = [
     "pydantic-settings >=2, <3.0",
     "quantinuum-schemas>=7.3, <8.0",
     "hugr >=0.14.0, <1.0.0",
+    "pyjwt>=2.10.1,<3.0.0",
 ]
 
 [project.optional-dependencies]

qnexus/client/auth.py

@@ -1,11 +1,14 @@
 """Client API for authentication in Nexus."""
 
+import datetime
 import getpass
 import time
+import warnings
 import webbrowser
 from http import HTTPStatus
 
 import httpx
+import jwt
 from colorama import Fore
 from pydantic import EmailStr
 from rich.console import Console
@@ -18,12 +21,52 @@
     _check_version_headers,
     get_nexus_client,
 )
-from qnexus.client.utils import consolidate_error, remove_token, write_token
+from qnexus.client.utils import consolidate_error, read_token, remove_token, write_token
 from qnexus.config import CONFIG
 
 console = Console()
 
 
+def is_logged_in() -> bool:
+    """Check if the user is already logged in by verifying tokens and
+    attempting a lightweight authenticated request."""
+
+    try:
+        refresh_token = read_token("refresh_token")
+        access_token = read_token("access_token")
+
+        # Check that tokens are present
+        if not refresh_token or not access_token:
+            return False
+    except FileNotFoundError:
+        return False
+    # Check expiry of refresh token (assume JWT)
+    try:
+        payload = jwt.decode(refresh_token, options={"verify_signature": False})
+        exp = payload.get("exp")
+        if exp:
+            expiry_dt = datetime.datetime.fromtimestamp(exp)
+            now = datetime.datetime.now()
+            hours_left = (expiry_dt - now).total_seconds() / 3600
+            if hours_left < 24:
+                msg = (
+                    f"Your refresh token expires in less than 24 hours (expires at {expiry_dt}). "
+                    "You will need to login again after this time or use force=True to refresh now."
+                )
+                warnings.warn(msg, category=UserWarning)
+    except jwt.PyJWTError:
+        pass
+    # Try a lightweight authenticated request to check validity
+    try:
+        client = get_nexus_client()
+        resp = client.get("/api/users/v1beta2/me")
+        if resp.status_code == HTTPStatus.OK:
+            return True
+    except (httpx.HTTPError, qnx_exc.AuthenticationError):
+        pass
+    return False
+
+
 def _get_auth_client() -> httpx.Client:
     """Getter function for the Nexus auth client."""
     return httpx.Client(
@@ -33,12 +76,15 @@ def _get_auth_client() -> httpx.Client:
     )
 
 
-def login() -> None:
+def login(force: bool = False) -> None:
     """
     Log in to Quantinuum Nexus using the web browser.
 
     (if web browser can't be launched, displays the link)
     """
+    if not force and is_logged_in():
+        print("Already logged in. Tokens are valid.")
+        return
 
     res = _get_auth_client().post(
         "/device/device_authorization",
@@ -119,8 +165,11 @@ def login() -> None:
     raise qnx_exc.AuthenticationError("Browser login Failed, code has expired.")
 
 
-def login_with_credentials() -> None:
+def login_with_credentials(force: bool = False) -> None:
     """Log in to Nexus using a username and password."""
+    if not force and is_logged_in():
+        print("Already logged in. Tokens are valid.")
+        return
     user_name = input("Enter your Nexus email: ")
     pwd = getpass.getpass(prompt="Enter your Nexus password: ")
 
@@ -129,12 +178,14 @@ def login_with_credentials() -> None:
     print(f"✅ Successfully logged in as {user_name}.")
 
 
-def login_no_interaction(user: EmailStr, pwd: str) -> None:
+def login_no_interaction(user: EmailStr, pwd: str, force: bool = False) -> None:
     """Log in to Nexus using a username and password.
     Please be careful with storing credentials in plain text or source code.
     """
+    if not force and is_logged_in():
+        print("Already logged in. Tokens are valid.")
+        return
     _request_tokens(user=user, pwd=pwd)
-
     print(f"✅ Successfully logged in as {user}.")
 
 

qnexus/client/utils.py

@@ -43,6 +43,7 @@ def normalize_included(included: list[Any]) -> dict[str, dict[str, Any]]:
 def remove_token(token_type: TokenTypes) -> None:
     """Delete a token file."""
     # Don't try to delete refresh token in Jupyterhub
+    # these get mounted in automatically and managed externally
     if is_jupyterhub_environment() and token_type == "refresh_token":
         return
     token_file_path = Path.home() / CONFIG.token_path / token_file_from_type[token_type]
@@ -90,7 +91,8 @@ def read_token(token_type: TokenTypes) -> str:
 def write_token(token_type: TokenTypes, token: str) -> None:
     """Write a token to a file."""
 
-    # don't allow writing of refresh token in Jupyterhub
+    # Don't allow writing of refresh token in Jupyterhub
+    # these get mounted in automatically and managed externally
     if is_jupyterhub_environment() and token_type == "refresh_token":
         return
 

tests/test_warnings.py

@@ -1,3 +1,6 @@
+import base64
+import json
+import time
 import typing
 import warnings
 from importlib.metadata import version
@@ -13,6 +16,7 @@
     VERSION_STATUS_HEADER,
     get_nexus_client,
 )
+from qnexus.client.auth import is_logged_in
 from qnexus.client.utils import write_token
 
 FAKE_LATEST_VERSION = "999.99.999-never-gonna-happen"
@@ -123,6 +127,12 @@ def test_version_check_emits_warning_login(m: mock.Mock) -> None:
     """
     base_url = get_nexus_client().base_url
 
+    # Mock the is_logged_in check (returns 401 to indicate not logged in)
+    respx.get(f"{base_url}/api/users/v1beta2/me").mock(return_value=httpx.Response(401))
+
+    # Mock token refresh attempt (also returns 401 since not logged in)
+    respx.post(f"{base_url}/auth/tokens/refresh").mock(return_value=httpx.Response(401))
+
     # Mock the login endpoints
     respx.post(f"{base_url}/auth/device/device_authorization").mock(
         return_value=httpx.Response(
@@ -156,3 +166,39 @@ def test_version_check_emits_warning_login(m: mock.Mock) -> None:
 
     _check_request_includes_version_data(token_request_route)
     _check_version_warning_emitted(captured)
+
+
+def _base64url_encode(data: bytes) -> str:
+    return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
+
+
+@respx.mock
+def test_refresh_token_expiry_warning_emitted() -> None:
+    """Emits a warning when the refresh token expires in less than 24 hours."""
+    # Create an unsigned JWT with exp in the next hour
+    header = {"alg": "none", "typ": "JWT"}
+    payload = {"exp": int(time.time()) + 3600}
+
+    jwt_token = (
+        _base64url_encode(json.dumps(header).encode())
+        + "."
+        + _base64url_encode(json.dumps(payload).encode())
+        + "."
+    )
+
+    write_token("refresh_token", jwt_token)
+    write_token("access_token", "dummy_id")
+
+    # Mock the lightweight authenticated request
+    me_route = respx.get(f"{get_nexus_client().base_url}/api/users/v1beta2/me").mock(
+        return_value=httpx.Response(200, json={"ok": True})
+    )
+
+    with warnings.catch_warnings(record=True) as w:
+        warnings.simplefilter("always")
+        assert is_logged_in() is True
+
+        messages = [str(item.message) for item in w]
+        assert any("expires in less than 24 hours" in msg for msg in messages)
+
+    assert me_route.called