Never revoke the configured bootstrap API token

src/groundlight/client.py

@@ -125,8 +125,6 @@ class Groundlight:  # pylint: disable=too-many-instance-attributes,too-many-publ
             Warning: Only disable verification when connecting to a Groundlight Edge Endpoint using
             self-signed certificates. For security, always keep verification enabled when using the
             Groundlight cloud service.
-    :param enable_token_rotation: If True (default), automatically rotate tokens whose identity has a
-            non-null Token TTL.
 
     :return: Groundlight client instance
     """
@@ -137,13 +135,12 @@ class Groundlight:  # pylint: disable=too-many-instance-attributes,too-many-publ
     POLLING_EXPONENTIAL_BACKOFF = 1.3  # This still has the nice backoff property that the max number of requests
     # is O(log(time)), but with 1.3 the guarantee is that the call will return no more than 30% late
 
-    def __init__(  # noqa: PLR0913  # pylint: disable=too-many-arguments
+    def __init__(
         self,
         endpoint: Optional[str] = None,
         api_token: Optional[str] = None,
         disable_tls_verification: Optional[bool] = None,
         http_transport_retries: Optional[Union[int, Retry]] = None,
-        enable_token_rotation: bool = True,
     ):
         """
         Initialize a new Groundlight client instance.
@@ -159,8 +156,6 @@ def __init__(  # noqa: PLR0913  # pylint: disable=too-many-arguments
             certificates. For security, always keep verification enabled when using the Groundlight cloud service.
         :param http_transport_retries: Overrides urllib3 `PoolManager` retry policy for HTTP/HTTPS (forwarded to
             `Configuration.retries`). Not the same as SDK 5xx retries handled by `RequestsRetryDecorator`.
-        :param enable_token_rotation: If True (default), automatically rotate tokens whose identity has a
-            non-null Token TTL.
 
         :return: Groundlight client
         """
@@ -199,16 +194,14 @@ def __init__(  # noqa: PLR0913  # pylint: disable=too-many-arguments
         self.configuration.api_key["ApiToken"] = api_token
 
         self.api_client = GroundlightApiClient(self.configuration)
-        self._token_manager: Optional[TokenManager] = None
-        if enable_token_rotation:
-            try:
-                self._token_manager = TokenManager(
-                    configured_token=api_token,
-                    configuration=self.configuration,
-                    request_timeout=DEFAULT_REQUEST_TIMEOUT,
-                )
-            except TokenManagerError as exc:
-                raise ApiTokenError(str(exc)) from exc
+        try:
+            self._token_manager = TokenManager(
+                configured_token=api_token,
+                configuration=self.configuration,
+                request_timeout=DEFAULT_REQUEST_TIMEOUT,
+            )
+        except TokenManagerError as exc:
+            raise ApiTokenError(str(exc)) from exc
         self.detectors_api = DetectorsApi(self.api_client)
         self.detector_group_api = DetectorGroupsApi(self.api_client)
         self.images_api = ImageQueriesApi(self.api_client)
@@ -219,8 +212,7 @@ def __init__(  # noqa: PLR0913  # pylint: disable=too-many-arguments
         self.logged_in_user = "(not-logged-in)"
         self._verify_connectivity()
         # No-op when the working token has no identity Token TTL.
-        if self._token_manager is not None:
-            self._token_manager.start()
+        self._token_manager.start()
 
     def __repr__(self) -> str:
         # Don't call the API here because that can get us stuck in a loop rendering exception strings
@@ -236,8 +228,7 @@ def __exit__(self, exc_type, exc_value, traceback) -> None:
 
     def close(self) -> None:
         """Stop the token refresh thread and close the HTTP client."""
-        if self._token_manager is not None:
-            self._token_manager.close()
+        self._token_manager.close()
         self.api_client.close()
 
     def _verify_connectivity(self) -> None:

src/groundlight/experimental_api.py

@@ -52,7 +52,6 @@ def __init__(
         endpoint: Union[str, None] = None,
         api_token: Union[str, None] = None,
         disable_tls_verification: Optional[bool] = None,
-        enable_token_rotation: bool = True,
     ):
         """
         Constructs an experimental Groundlight client.
@@ -84,15 +83,8 @@ def __init__(
                 Warning: Only disable verification when connecting to a Groundlight Edge Endpoint using
                 self-signed certificates. For security, always keep verification enabled when using the
                 Groundlight cloud service.
-        :param enable_token_rotation: If True (default), automatically rotate tokens whose identity has a
-                non-null Token TTL.
-        """
-        super().__init__(
-            endpoint=endpoint,
-            api_token=api_token,
-            disable_tls_verification=disable_tls_verification,
-            enable_token_rotation=enable_token_rotation,
-        )
+        """
+        super().__init__(endpoint=endpoint, api_token=api_token, disable_tls_verification=disable_tls_verification)
         self.notes_api = NotesApi(self.api_client)
         self.detector_group_api = DetectorGroupsApi(self.api_client)
         self.detector_reset_api = DetectorResetApi(self.api_client)

src/groundlight/token_manager.py

@@ -217,7 +217,11 @@ def _ensure_token_dir(self) -> None:
             raise TokenManagerError(f"Groundlight token directory '{self._token_dir}' is not writable")
 
     def _initialize_token(self) -> None:
-        """Load a valid cached token, use a never-expire configured token, or mint a child."""
+        """Load a valid cached token, use a never-expire configured token, or mint a child.
+
+        When minting the first working token, the configured bootstrap token is left out of the
+        cache slot (previous=None) and is never revoked by later refreshes.
+        """
         try:
             with self._lock:
                 slot = self._load_slot()
@@ -241,11 +245,13 @@ def _initialize_token(self) -> None:
                     # No identity Token TTL: behave like pre-rotation Groundlight.
                     return
 
+                # Mint a working child without recording the configured (bootstrap) token as
+                # previous. Bootstrap tokens are never written into the cache slot and never
+                # revoked by this manager, so other processes that share the same configured
+                # token can keep using it (or mint their own independent child chains) until
+                # the bootstrap token expires on its own.
                 base_name = TOKEN_NAME_SUFFIX_PATTERN.sub("", configured_meta.name)
-                self._mint_replacement(
-                    base_name=base_name,
-                    previous=PreviousToken(name=configured_meta.name, minted_at=_utc_now()),
-                )
+                self._mint_replacement(base_name=base_name, previous=None)
         except FileLockTimeout as exc:
             raise TokenManagerError(f"Timed out waiting for token cache lock '{self._lock_path}'") from exc
         except NotFoundException:

test/unit/test_token_manager.py

@@ -93,22 +93,34 @@ def test_initialization_mints_and_privately_caches_token(mocker, tmp_path):
     assert cached["base_name"] == "Device token"
     assert cached["current"]["raw_key"] == "api_working_token_one"
     assert cached["current"]["token_ttl_seconds"] == TOKEN_TTL.total_seconds()
-    assert cached["previous"]["name"] == "Device token"
+    assert cached["previous"] is None
     api.delete_api_token.assert_not_called()
 
 
-def test_initialization_parks_configured_token_as_previous(mocker, tmp_path):
-    """After minting the first working token, the configured token is parked as previous."""
+def test_initialization_does_not_park_or_revoke_bootstrap_token(mocker, tmp_path):
+    """The configured bootstrap token is omitted from the slot and never deleted on first refresh."""
     api = Mock()
     api.get_api_token_by_snippet.return_value = _expiring_metadata("Device token", CONFIGURED_TOKEN)
     api.create_api_token.return_value = _created_token("Device token abc123", "api_working_token_one", NOW)
 
     manager = _manager(mocker, tmp_path, api)
 
+    cached = json.loads(manager._slot_path.read_text())
+    assert cached["previous"] is None
+    api.delete_api_token.assert_not_called()
+
+    later = NOW + REFRESH_INTERVAL + timedelta(seconds=1)
+    mocker.patch.object(token_manager, "_utc_now", return_value=later)
+    api.reset_mock()
+    api.create_api_token.return_value = _created_token("Device token def456", "api_working_token_two", later)
+
+    manager.refresh()
+
+    # Only manager-minted tokens enter the revoke cycle; the bootstrap name is never deleted.
     api.delete_api_token.assert_not_called()
     cached = json.loads(manager._slot_path.read_text())
-    assert cached["previous"]["name"] == "Device token"
-    assert cached["previous"]["minted_at"] == NOW.isoformat().replace("+00:00", "Z")
+    assert cached["current"]["raw_key"] == "api_working_token_two"
+    assert cached["previous"]["name"] == "Device token abc123"
 
 
 def test_initialization_uses_never_expire_configured_token_as_is(mocker, tmp_path):

test/unit/test_token_refresh_client.py

@@ -18,18 +18,3 @@ def test_groundlight_starts_and_closes_token_manager(mocker):
 
     manager.close.assert_called_once()
     api_client_close.assert_called_once()
-
-
-def test_groundlight_skips_token_manager_when_rotation_disabled(mocker):
-    """Disabling rotation leaves the configured token in place with no TokenManager."""
-    token_manager_class = mocker.patch("groundlight.client.TokenManager")
-    mocker.patch.object(Groundlight, "_verify_connectivity")
-
-    client = Groundlight(api_token="api_bootstrap_token_value_long_enough", enable_token_rotation=False)
-    api_client_close = mocker.patch.object(client.api_client, "close")
-    client.close()
-
-    token_manager_class.assert_not_called()
-    assert client._token_manager is None
-    assert client.configuration.api_key["ApiToken"] == "api_bootstrap_token_value_long_enough"
-    api_client_close.assert_called_once()