Fix token required check for api.tilebox.com (#1)

* Potential fix for code scanning alert no. 1: Incomplete URL substring sanitization

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* Fix token required check for api.tilebox.com

* Restrict permissions of github workflow

* Fix aio channel

---------

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>
Co-authored-by: Lukas Bindreiter <lukas.bindreiter@tilebox.com>

Author: 3 people

.github/workflows/main.yml

@@ -1,5 +1,7 @@
 name: CI
 
+permissions: {}
+
 on:
   push:
     branches: [main]

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- `tilebox-grpc`: More robust parsing of GRPC channel URLs.
+
 ## [0.37.0] - 2025-06-06
 
 ### Changed

README.md

@@ -71,7 +71,7 @@ area_of_interest = shape({
 )
 s2a_l1c = sentinel2_msi.collection("S2A_S2MSI1C")
 results = s2a_l1c.query(
-  temporal_extent=("2022-07-13", "2022-07-13T02:00"),
+  temporal_extent=("2025-03-01", "2025-06-01"),
   spatial_extent=area_of_interest,
   show_progress=True
 )

tilebox-datasets/README.md

@@ -66,7 +66,7 @@ Query data:
 ```python
 s2a_l1c = sentinel2_msi.collection("S2A_S2MSI1C")
 results = s2a_l1c.query(
-  temporal_extent=("2017-01-01", "2023-01-01"),
+  temporal_extent=("2025-03-01", "2025-06-01"),
   show_progress=True
 )
 print(f"Found {results.sizes['time']} datapoints")  # Found 220542 datapoints
@@ -83,7 +83,7 @@ area_of_interest = shape({
 )
 s2a_l1c = sentinel2_msi.collection("S2A_S2MSI1C")
 results = s2a_l1c.query(
-  temporal_extent=("2022-07-13", "2022-07-13T02:00"),
+  temporal_extent=("2025-03-01", "2025-06-01"),
   spatial_extent=area_of_interest,
   show_progress=True
 )

tilebox-datasets/tilebox/datasets/client.py

@@ -6,6 +6,7 @@
 from loguru import logger
 from promise import Promise
 
+from _tilebox.grpc.channel import parse_channel_info
 from tilebox.datasets.data.datasets import Dataset, DatasetGroup, ListDatasetsResponse
 from tilebox.datasets.data.uuid import as_uuid
 from tilebox.datasets.group import Group
@@ -66,11 +67,13 @@ def _dataset_by_id(self, dataset_id: str | UUID, dataset_type: type[T]) -> Promi
 def token_from_env(url: str, token: str | None) -> str | None:
     if token is None:  # if no token is provided, try to get it from the environment
         token = os.environ.get("TILEBOX_API_KEY", None)
-    if "api.tilebox.com" in url and token is None:
+
+    if token is None and parse_channel_info(url).address == "api.tilebox.com":
         raise ValueError(
             "No API key provided and no TILEBOX_API_KEY environment variable set. Please specify an API key using "
             "the token argument. For example: `Client(token='YOUR_TILEBOX_API_KEY')`"
         )
+
     return token
 
 

tilebox-grpc/_tilebox/grpc/aio/channel.py

@@ -1,8 +1,8 @@
 from collections.abc import Callable, Sequence
 from typing import TypeVar
 
-from _tilebox.grpc.channel import CHANNEL_OPTIONS, ChannelInfo, add_metadata, parse_channel_info
-from grpc import ssl_channel_credentials
+from _tilebox.grpc.channel import CHANNEL_OPTIONS, ChannelInfo, ChannelProtocol, add_metadata, parse_channel_info
+from grpc import Compression, ssl_channel_credentials
 from grpc.aio import (
     Channel,
     ClientCallDetails,
@@ -34,11 +34,28 @@ def open_channel(url: str, auth_token: str | None = None) -> Channel:
 
 
 def _open_channel(channel_info: ChannelInfo, interceptors: Sequence[ClientInterceptor]) -> Channel:
-    if channel_info.use_ssl:
-        return secure_channel(
-            channel_info.url_without_protocol, ssl_channel_credentials(), CHANNEL_OPTIONS, interceptors=interceptors
-        )
-    return insecure_channel(channel_info.url_without_protocol, CHANNEL_OPTIONS, interceptors=interceptors)
+    match channel_info.protocol:
+        case ChannelProtocol.HTTPS:
+            return secure_channel(
+                f"{channel_info.address}:{channel_info.port}",
+                ssl_channel_credentials(),
+                CHANNEL_OPTIONS,
+                compression=Compression.Gzip,
+                interceptors=interceptors,
+            )
+        case ChannelProtocol.HTTP:
+            return insecure_channel(
+                f"{channel_info.address}:{channel_info.port}",
+                CHANNEL_OPTIONS,
+                compression=Compression.NoCompression,
+                interceptors=interceptors,
+            )
+        case ChannelProtocol.UNIX:
+            return insecure_channel(
+                channel_info.address, CHANNEL_OPTIONS, compression=Compression.NoCompression, interceptors=interceptors
+            )
+        case _:
+            raise ValueError(f"Unsupported channel protocol: {channel_info.protocol}")
 
 
 RequestType = TypeVar("RequestType")

tilebox-grpc/_tilebox/grpc/channel.py

@@ -2,6 +2,7 @@
 import re
 from collections.abc import Callable
 from dataclasses import dataclass
+from enum import Enum
 from typing import TypeVar
 
 from grpc import (
@@ -43,10 +44,21 @@
 ]
 
 
+class ChannelProtocol(Enum):
+    HTTPS = 1
+    HTTP = 2
+    UNIX = 3
+
+
 @dataclass
 class ChannelInfo:
-    url_without_protocol: str
-    use_ssl: bool
+    address: str
+    """GRPC target address. For http(s) connections this is the host[:port] format without a scheme. For unix sockets
+    this is the unix socket path including the `unix://` prefix for absolute paths or `unix:` for relative paths."""
+    port: int
+    """Port number for http(s) connections. For unix sockets this is always 0."""
+    protocol: ChannelProtocol
+    """The protocol to use for the channel."""
 
 
 def open_channel(url: str, auth_token: str | None = None) -> Channel:
@@ -69,14 +81,22 @@ def open_channel(url: str, auth_token: str | None = None) -> Channel:
 
 
 def _open_channel(channel_info: ChannelInfo) -> Channel:
-    if channel_info.use_ssl:
-        return secure_channel(
-            channel_info.url_without_protocol,
-            ssl_channel_credentials(),
-            CHANNEL_OPTIONS,
-            compression=Compression.Gzip,
-        )
-    return insecure_channel(channel_info.url_without_protocol, CHANNEL_OPTIONS, compression=Compression.NoCompression)
+    match channel_info.protocol:
+        case ChannelProtocol.HTTPS:
+            return secure_channel(
+                f"{channel_info.address}:{channel_info.port}",
+                ssl_channel_credentials(),
+                CHANNEL_OPTIONS,
+                compression=Compression.Gzip,
+            )
+        case ChannelProtocol.HTTP:
+            return insecure_channel(
+                f"{channel_info.address}:{channel_info.port}", CHANNEL_OPTIONS, compression=Compression.NoCompression
+            )
+        case ChannelProtocol.UNIX:
+            return insecure_channel(channel_info.address, CHANNEL_OPTIONS, compression=Compression.NoCompression)
+        case _:
+            raise ValueError(f"Unsupported channel protocol: {channel_info.protocol}")
 
 
 _URL_SCHEME = re.compile(r"^(https?://)?([^: ]+)(:\d+)?/?$")
@@ -98,27 +118,28 @@ def parse_channel_info(url: str) -> ChannelInfo:
         A ChannelInfo object that can be used to create a gRPC channel.
     """
     # See https://github.com/grpc/grpc/blob/master/doc/naming.md
-    if url.startswith("unix:"):
-        return ChannelInfo(url, False)
+    if url.startswith("unix:"):  ## unix:///absolute/path or unix://path
+        return ChannelInfo(url, 0, ChannelProtocol.UNIX)
 
     # `urllib.parse.urlparse` behaves a bit weird with URLs that don't have a scheme but a port number, so regex it is
     if (match := _URL_SCHEME.match(url)) is None:
         raise ValueError(f"Invalid URL: {url}")
     scheme, netloc, port = match.groups()
     netloc = netloc.rstrip("/")
-    use_ssl = True
+    protocol = ChannelProtocol.HTTPS
 
     if scheme == "http://":  # explicitly set http -> require a port
         if port is None:
             raise ValueError("Explicit port required for insecure HTTP channel")
-        use_ssl = False
+        protocol = ChannelProtocol.HTTP
+
+    # no scheme, but a port that looks like a dev port -> insecure
+    if scheme is None and port is not None and port != ":443":
+        protocol = ChannelProtocol.HTTP
 
-    if scheme is None and port is not None:  # no scheme, but a port that looks like a dev port -> insecure
-        use_ssl = port == ":443"
+    port_number = 443 if port is None else int(port.removeprefix(":"))
 
-    if use_ssl:
-        return ChannelInfo(netloc + (port or ":443"), True)
-    return ChannelInfo(netloc + port, False)
+    return ChannelInfo(netloc, port_number, protocol)
 
 
 RequestType = TypeVar("RequestType")

tilebox-grpc/tests/test_channel.py

@@ -4,6 +4,7 @@
 
 from _tilebox.grpc.channel import (
     CHANNEL_OPTIONS,
+    ChannelProtocol,
     open_channel,
     parse_channel_info,
 )
@@ -50,24 +51,26 @@ def test_open_authenticated_channel(open_func: MagicMock, intercept_func: MagicM
 )
 def test_parse_channel_info_secure(url: str) -> None:
     channel_info = parse_channel_info(url)
-    assert channel_info.url_without_protocol == "api.tilebox.com:443"
-    assert channel_info.use_ssl
+    assert channel_info.address == "api.tilebox.com"
+    assert channel_info.port == 443
+    assert channel_info.protocol == ChannelProtocol.HTTPS
 
 
 @pytest.mark.parametrize(
     ("url", "expected_url_without_protocol"),
     [
-        ("0.0.0.0:8083", "0.0.0.0:8083"),
-        ("http://0.0.0.0:8083", "0.0.0.0:8083"),
-        ("http://localhost:8083", "localhost:8083"),
-        ("localhost:8083", "localhost:8083"),
-        ("http://some.insecure.url:1234", "some.insecure.url:1234"),
+        ("0.0.0.0:8083", "0.0.0.0"),  # noqa: S104
+        ("http://0.0.0.0:8083", "0.0.0.0"),  # noqa: S104
+        ("http://localhost:8083", "localhost"),
+        ("localhost:8083", "localhost"),
+        ("http://some.insecure.url:8083", "some.insecure.url"),
     ],
 )
 def test_parse_channel_info_insecure(url: str, expected_url_without_protocol: str) -> None:
     channel_info = parse_channel_info(url)
-    assert channel_info.url_without_protocol == expected_url_without_protocol
-    assert not channel_info.use_ssl
+    assert channel_info.address == expected_url_without_protocol
+    assert channel_info.port == 8083
+    assert channel_info.protocol == ChannelProtocol.HTTP
 
 
 @pytest.mark.parametrize(
@@ -79,8 +82,9 @@ def test_parse_channel_info_insecure(url: str, expected_url_without_protocol: st
 )
 def test_parse_channel_info_unix(url: str) -> None:
     channel_info = parse_channel_info(url)
-    assert channel_info.url_without_protocol == url
-    assert not channel_info.use_ssl
+    assert channel_info.address == url
+    assert channel_info.port == 0
+    assert channel_info.protocol == ChannelProtocol.UNIX
 
 
 def test_parse_channel_invalid() -> None:

tilebox-workflows/tilebox/workflows/client.py

@@ -1,7 +1,7 @@
 import logging
 import os
 
-from _tilebox.grpc.channel import open_channel
+from _tilebox.grpc.channel import open_channel, parse_channel_info
 from tilebox.datasets.sync.client import Client as DatasetsClient
 from tilebox.workflows.automations.client import AutomationClient, AutomationService
 from tilebox.workflows.cache import JobCache, NoCache
@@ -28,13 +28,7 @@ def __init__(self, *, url: str = "https://api.tilebox.com", token: str | None =
             url: Tilebox API Url. Defaults to "https://api.tilebox.com".
             token: The API Key to authenticate with. If not set the `TILEBOX_API_KEY` environment variable will be used.
         """
-        if token is None:  # if no token is provided, try to get it from the environment
-            token = os.environ.get("TILEBOX_API_KEY", None)
-        if url == "https://api.tilebox.com" and token is None:
-            raise ValueError(
-                "No API key provided and no TILEBOX_API_KEY environment variable set. Please specify an API key using "
-                "the token argument. For example: `Client(token='YOUR_TILEBOX_API_KEY')`"
-            )
+        token = _token_from_env(url, token)
         self._auth = {"token": token, "url": url}
         self._channel = open_channel(url, token)
 
@@ -147,3 +141,16 @@ def automations(self) -> AutomationClient:
             A client for the automations service.
         """
         return AutomationClient(AutomationService(self._channel))
+
+
+def _token_from_env(url: str, token: str | None) -> str | None:
+    if token is None:  # if no token is provided, try to get it from the environment
+        token = os.environ.get("TILEBOX_API_KEY", None)
+
+    if token is None and parse_channel_info(url).address == "api.tilebox.com":
+        raise ValueError(
+            "No API key provided and no TILEBOX_API_KEY environment variable set. Please specify an API key using "
+            "the token argument. For example: `Client(token='YOUR_TILEBOX_API_KEY')`"
+        )
+
+    return token