feat: enhance CATMAID authentication with personal API token support

docs/datasource/catmaid/index.rst

@@ -0,0 +1,66 @@
+.. _catmaid-datasource:
+
+CATMAID
+=======
+
+The CATMAID data service driver exposes skeletons from a CATMAID project,
+including spatially indexed skeleton loading and optional editing.
+
+URL syntax
+----------
+
+- :file:`catmaid://https://{host}/{project-id}`
+- :file:`catmaid://http://{host}/{project-id}`
+
+Browser authentication
+----------------------
+
+Neuroglancer first requests CATMAID's anonymous API token so public projects
+continue to load without interaction. If the anonymous account cannot access a
+project, Neuroglancer asks for a personal CATMAID API token.
+
+Open the CATMAID server, use the account menu to obtain an API token, and paste
+it into the Neuroglancer prompt. The token is cached for the current browser tab
+under the CATMAID base URL, allowing other projects on the same server to reuse
+it. A rejected token is removed automatically and requested again. Tokens are
+not included in datasource URLs or serialized viewer state.
+
+Because an API token has the same permissions as its CATMAID account, use HTTPS
+for authenticated deployments and do not share the token.
+
+Python authentication
+---------------------
+
+Python-hosted viewers obtain credentials on the Python server. Configure a
+token before adding CATMAID layers:
+
+.. code-block:: python
+
+   import neuroglancer
+
+   neuroglancer.set_catmaid_token(
+       "https://catmaid.example",
+       "your-api-token",
+   )
+
+Pass ``None`` as the token to remove a configured value. Deployments may
+instead set ``CATMAID_CREDENTIALS`` to a JSON object keyed by CATMAID base
+URL:
+
+.. code-block:: shell
+
+   export CATMAID_CREDENTIALS='{"https://catmaid.example": "your-api-token"}'
+
+Tokens configured with ``set_catmaid_token`` take precedence over the
+environment. If neither is present, the Python provider attempts anonymous
+access.
+
+Server requirements
+-------------------
+
+The CATMAID deployment must allow the Neuroglancer origin to make cross-origin
+``GET`` and ``POST`` requests and must allow the ``Authorization`` and
+``Content-Type`` request headers. Project read and edit access is determined
+by the CATMAID account associated with the token. Separately, Neuroglancer only
+enables editing when the linked stack metadata sets ``read_only`` to
+``false``; see :ref:`skeleton-editing-sources`.

docs/datasource/index.rst

@@ -127,5 +127,6 @@ Data services
    :maxdepth: 1
 
    boss/index
+   catmaid/index
    dvid/index
    render/index

docs/user-guide/skeleton_editing.rst

@@ -15,7 +15,10 @@ CATMAID documentation to set up a CATMAID server. At minimum you will need:
 - CATMAID ``2026.05.06.dev11+g...`` or later by git-describe semantics.
 - A CATMAID project
 - A linked project stack
-- ``AnonymousUser`` permissions to read and edit the data on that project
+- CATMAID read permissions for anonymous access or for the account associated
+  with a personal API token
+- CATMAID edit permissions for that account when editing is enabled
+- Cross-origin access for the Neuroglancer origin and authorization headers
 - Skeletons initialised for that project
 
 The project stack dimensions and resolution are used to inform the bounding box
@@ -56,7 +59,13 @@ If ``spatial`` is absent or empty, Neuroglancer derives a default chunk size
 from the CATMAID project-space bounds and uses ``limit: 0`` for the generated
 spatial level.
 
-After setting this up, enter ``catmaid:<your-catmaid-server-url>/<your-catmaid-project-id>`` as a data source in neuroglancer.
+After setting this up, enter
+``catmaid:<your-catmaid-server-url>/<your-catmaid-project-id>`` as a data
+source in Neuroglancer. Public projects use CATMAID's anonymous API token.
+Private projects prompt for a personal API token, which is retained for the
+current browser tab. Python-hosted viewers can configure the token with
+``neuroglancer.set_catmaid_token`` or ``CATMAID_CREDENTIALS``. See
+:ref:`catmaid-datasource` for authentication and CORS details.
 
 .. _skeleton-editing-subsources:
 

python/neuroglancer/__init__.py

@@ -18,7 +18,10 @@
     server,  # noqa: F401
     skeleton,  # noqa: F401
 )
-from .default_credentials_manager import set_boss_token  # noqa: F401
+from .default_credentials_manager import (  # noqa: F401
+    set_boss_token,
+    set_catmaid_token,
+)
 from .equivalence_map import EquivalenceMap  # noqa: F401
 from .local_volume import LocalVolume  # noqa: F401
 from .screenshot import ScreenshotSaver  # noqa: F401

python/neuroglancer/catmaid_credentials.py

@@ -13,26 +13,115 @@
 # limitations under the License.
 
 import json
+import os
+import threading
 import urllib.request
 
 from . import credentials_provider
 from .futures import run_on_new_thread
 
+_configured_tokens = {}
+_providers = {}
+_providers_lock = threading.Lock()
 
-class CatmaidAnonymousCredentialsProvider(credentials_provider.CredentialsProvider):
+
+def canonicalize_server_url(server_url):
+    if not isinstance(server_url, str) or not server_url:
+        raise ValueError("CATMAID server URL must be a non-empty string")
+    return server_url.rstrip("/")
+
+
+def set_token(server_url, token):
+    """Configure or remove a personal API token for one CATMAID server."""
+    server_url = canonicalize_server_url(server_url)
+    with _providers_lock:
+        if token is None:
+            _configured_tokens.pop(server_url, None)
+            return
+        if not isinstance(token, str) or not token.strip():
+            raise ValueError("CATMAID API token must be a non-empty string or None")
+        _configured_tokens[server_url] = token.strip()
+
+
+def _get_environment_tokens():
+    value = os.environ.get("CATMAID_CREDENTIALS")
+    if not value:
+        return {}
+    try:
+        parsed = json.loads(value)
+    except json.JSONDecodeError as error:
+        raise RuntimeError(
+            "CATMAID_CREDENTIALS must be a JSON object mapping server URLs to tokens"
+        ) from error
+    if not isinstance(parsed, dict) or any(
+        not isinstance(server_url, str)
+        or not isinstance(token, str)
+        or not token.strip()
+        for server_url, token in parsed.items()
+    ):
+        raise RuntimeError(
+            "CATMAID_CREDENTIALS must be a JSON object mapping server URLs to "
+            "non-empty string tokens"
+        )
+    return {
+        canonicalize_server_url(server_url): token.strip()
+        for server_url, token in parsed.items()
+    }
+
+
+def _get_configured_token(server_url):
+    with _providers_lock:
+        token = _configured_tokens.get(server_url)
+    if token is not None:
+        return token
+    return _get_environment_tokens().get(server_url)
+
+
+class CatmaidCredentialsProvider(credentials_provider.CredentialsProvider):
     def __init__(self, parameters):
         super().__init__()
-        self.server_url = (parameters or {}).get("serverUrl", "")
+        self.server_url = canonicalize_server_url(
+            (parameters or {}).get("serverUrl", "")
+        )
+        self._last_personal_token = None
+        self._anonymous_token_was_returned = False
 
     def get_new(self):
         server_url = self.server_url
 
         def func():
+            personal_token = _get_configured_token(server_url)
+            if personal_token is not None:
+                if personal_token == self._last_personal_token:
+                    raise RuntimeError(
+                        f"The configured CATMAID API token for {server_url} was rejected"
+                    )
+                self._last_personal_token = personal_token
+                return {"token": personal_token, "kind": "personal"}
+
+            if self._anonymous_token_was_returned:
+                raise RuntimeError(
+                    f"CATMAID server {server_url} requires a personal API token; "
+                    "call neuroglancer.set_catmaid_token or configure "
+                    "CATMAID_CREDENTIALS"
+                )
+
             token_url = f"{server_url}/accounts/anonymous-api-token"
             with urllib.request.urlopen(token_url) as response:
                 data = json.loads(response.read().decode())
             if not isinstance(data, dict) or not isinstance(data.get("token"), str):
                 raise RuntimeError(f"Unexpected response from {token_url}: {data!r}")
-            return {"token": data["token"]}
+            self._anonymous_token_was_returned = True
+            return {"token": data["token"], "kind": "anonymous"}
 
         return run_on_new_thread(func)
+
+
+def get_credentials_provider(parameters):
+    server_url = canonicalize_server_url((parameters or {}).get("serverUrl", ""))
+    with _providers_lock:
+        provider = _providers.get(server_url)
+        if provider is None:
+            provider = CatmaidCredentialsProvider({"serverUrl": server_url})
+            _providers[server_url] = provider
+        return provider

python/neuroglancer/default_credentials_manager.py

@@ -49,9 +49,7 @@
 
 default_credentials_manager.register(
     "CATMAID",
-    lambda parameters: catmaid_credentials.CatmaidAnonymousCredentialsProvider(
-        parameters
-    ),
+    lambda parameters: catmaid_credentials.get_credentials_provider(parameters),
 )
 
 
@@ -62,3 +60,15 @@ def set_boss_token(token):
       credentials
     """
     boss_credentials_provider.set_token(token)
+
+
+def set_catmaid_token(server_url, token):
+    """Sets or removes a personal API token for a CATMAID server.
+
+    Pass None as the token to remove a previously configured value. Tokens
+    should be configured before adding CATMAID layers.
+
+    Group:
+      credentials
+    """
+    catmaid_credentials.set_token(server_url, token)

python/tests/catmaid_credentials_test.py

@@ -0,0 +1,119 @@
+# @license
+# Copyright 2026 Google Inc.
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import io
+import json
+from unittest import mock
+
+import pytest
+from neuroglancer import catmaid_credentials
+
+
+def get_credentials(provider, invalid_generation=None):
+    return provider.get(invalid_generation).result(timeout=2)
+
+
+def test_runtime_token_precedes_environment(monkeypatch):
+    server_url = "https://runtime-token.catmaid.example"
+    monkeypatch.setenv(
+        "CATMAID_CREDENTIALS", json.dumps({server_url: "environment-token"})
+    )
+    catmaid_credentials.set_token(server_url + "/", " runtime-token ")
+    try:
+        provider = catmaid_credentials.CatmaidCredentialsProvider(
+            {"serverUrl": server_url}
+        )
+        assert get_credentials(provider)["credentials"] == {
+            "token": "runtime-token",
+            "kind": "personal",
+        }
+    finally:
+        catmaid_credentials.set_token(server_url, None)
+
+
+def test_environment_token_is_normalized(monkeypatch):
+    server_url = "https://environment-token.catmaid.example"
+    monkeypatch.setenv(
+        "CATMAID_CREDENTIALS",
+        json.dumps({server_url + "/": " environment-token "}),
+    )
+    provider = catmaid_credentials.CatmaidCredentialsProvider(
+        {"serverUrl": server_url + "/"}
+    )
+
+    assert get_credentials(provider)["credentials"] == {
+        "token": "environment-token",
+        "kind": "personal",
+    }
+
+
+def test_anonymous_token_is_used_without_configuration(monkeypatch):
+    server_url = "https://anonymous.catmaid.example"
+    monkeypatch.delenv("CATMAID_CREDENTIALS", raising=False)
+    provider = catmaid_credentials.CatmaidCredentialsProvider(
+        {"serverUrl": server_url + "/"}
+    )
+    response = io.BytesIO(json.dumps({"token": "anonymous-token"}).encode())
+
+    with mock.patch(
+        "neuroglancer.catmaid_credentials.urllib.request.urlopen",
+        return_value=response,
+    ) as urlopen:
+        credentials = get_credentials(provider)
+
+    assert credentials["credentials"] == {
+        "token": "anonymous-token",
+        "kind": "anonymous",
+    }
+    urlopen.assert_called_once_with(server_url + "/accounts/anonymous-api-token")
+
+    with pytest.raises(RuntimeError, match="requires a personal API token"):
+        get_credentials(provider, credentials["generation"])
+
+
+def test_rejected_personal_token_requires_reconfiguration(monkeypatch):
+    server_url = "https://rejected-token.catmaid.example"
+    monkeypatch.setenv(
+        "CATMAID_CREDENTIALS", json.dumps({server_url: "rejected-token"})
+    )
+    provider = catmaid_credentials.CatmaidCredentialsProvider({"serverUrl": server_url})
+    credentials = get_credentials(provider)
+
+    with pytest.raises(RuntimeError, match="was rejected"):
+        get_credentials(provider, credentials["generation"])
+
+
+def test_malformed_environment_configuration_is_rejected(monkeypatch):
+    monkeypatch.setenv("CATMAID_CREDENTIALS", "[]")
+    provider = catmaid_credentials.CatmaidCredentialsProvider(
+        {"serverUrl": "https://invalid-config.catmaid.example"}
+    )
+
+    with pytest.raises(RuntimeError, match="must be a JSON object"):
+        get_credentials(provider)
+
+
+def test_provider_cache_is_scoped_by_canonical_server_url():
+    first = catmaid_credentials.get_credentials_provider(
+        {"serverUrl": "https://cached.catmaid.example/"}
+    )
+    second = catmaid_credentials.get_credentials_provider(
+        {"serverUrl": "https://cached.catmaid.example"}
+    )
+    other = catmaid_credentials.get_credentials_provider(
+        {"serverUrl": "https://other.catmaid.example"}
+    )
+
+    assert first is second
+    assert first is not other