Isolate tunnel publisher credentials on the runtime origin

dataplicity_cli/api.py

@@ -9,6 +9,7 @@
 from typing import Any, Callable, Dict, Optional
 
 import requests
+from requests.structures import CaseInsensitiveDict
 
 from .config import Config
 from .client_identity import identity_headers
@@ -140,11 +141,13 @@ def _attempt_refresh(self) -> str:
             return "failed"
         url = self._build_url("/api/token/refresh/")
         try:
-            resp = self.session.post(url, json={"refresh": self.config.refresh_token}, timeout=10)
+            resp = self.session.post(url, json={"refresh": self.config.refresh_token}, timeout=10, allow_redirects=False)
         except Exception:
             return "failed"
         if resp.status_code >= 400:
             return "invalid" if self._response_indicates_invalid_jwt(resp) else "failed"
+        if not 200 <= resp.status_code < 300:
+            return "failed"
         try:
             payload = resp.json()
         except Exception:
@@ -228,17 +231,18 @@ def request(
         timeout: int = 20,
         allow_refresh: bool = True,
     ) -> ApiResponse:
-        if allow_refresh and self.config.auth_method == "jwt" and self.config.refresh_token:
+        supplied_headers = CaseInsensitiveDict(headers or {})
+        owns_session_auth = "Authorization" not in supplied_headers
+        if allow_refresh and owns_session_auth and self.config.auth_method == "jwt" and self.config.refresh_token:
             self.ensure_fresh_access_token()
 
         url = self._build_url(path)
-        req_headers = {
+        req_headers = CaseInsensitiveDict({
             **identity_headers(self.config.install_id),
             "Accept": "application/json",
             **self._auth_headers(),
-        }
-        if headers:
-            req_headers.update(headers)
+        })
+        req_headers.update(supplied_headers)
         try:
             resp = self.session.request(
                 method.upper(),
@@ -248,12 +252,14 @@ def request(
                 data=data,
                 headers=req_headers,
                 timeout=timeout,
+                allow_redirects=False,
             )
         except requests.RequestException as exc:
             return ApiResponse(False, 0, None, str(exc))
 
         should_retry_with_refresh = (
             allow_refresh
+            and owns_session_auth
             and self.config.auth_method == "jwt"
             and (resp.status_code == 401 or self._response_indicates_invalid_jwt(resp))
         )
@@ -269,7 +275,9 @@ def request(
                     timeout=timeout,
                     allow_refresh=False,
                 )
-        if self.config.auth_method == "jwt" and self._response_indicates_invalid_jwt(resp):
+        if (owns_session_auth and self.config.auth_method == "jwt"
+                and req_headers.get("Authorization") == f"Bearer {self.config.access_token}"
+                and self._response_indicates_invalid_jwt(resp)):
             self._invalidate_jwt_session()
 
         text = resp.text or ""

dataplicity_cli/tunnels.py

@@ -16,6 +16,25 @@
 from .tunnel_dashboard import TunnelStats
 
 
+def publisher_origin(base_url: str) -> str:
+    """Move only known service aliases; explicit private deployments keep theirs."""
+    value = base_url.rstrip('/')
+    try:
+        url = urlsplit(value)
+        if (url.scheme != "https" or url.port not in (None, 443)
+                or url.username is not None or url.password is not None
+                or url.path or url.query or url.fragment):
+            return value
+    except ValueError:
+        return value
+    return {
+        **{f"{host}.dataplicity.com": "https://runtime.dataplicity.com"
+           for host in ("api", "app-api", "gateway")},
+        **{f"{host}.staging.dpenv.com": "https://runtime.staging.dpenv.com"
+           for host in ("api", "app-api", "gateway")},
+    }.get(url.hostname, value)
+
+
 class TunnelError(RuntimeError):
     def __init__(self, message: str, code: str = "tunnel_error") -> None:
         super().__init__(message)
@@ -58,6 +77,8 @@ def __init__(self, api: ApiClient, organisation: str, token: Optional[str] = Non
         self.api = api
         self.organisation = organisation
         self.base = f"/api/organisations/{organisation}/development-tunnels/"
+        if token:
+            self.base = publisher_origin(api.config.base_url) + self.base
         self.headers = {"Authorization": f"TunnelPublisher {token}"} if token else None
         self._request_lock = threading.Lock()
         self._call_lock: Optional[asyncio.Lock] = None

tests/test_api_credentials.py

@@ -0,0 +1,69 @@
+"""Machine publisher calls must not borrow or invalidate a human session."""
+import unittest
+from unittest.mock import Mock
+
+from dataplicity_cli.api import ApiClient
+from dataplicity_cli.config import Config
+from dataplicity_cli.tunnels import TunnelAPI, TunnelError
+
+
+class CredentialBoundaryTests(unittest.TestCase):
+    def client(self):
+        api = ApiClient(Config(auth_method="jwt", access_token="human-access", refresh_token="human-refresh"))
+        api.session.request = Mock(return_value=Mock(status_code=200, text="{}", json=Mock(return_value={})))
+        api.ensure_fresh_access_token = Mock()
+        return api
+
+    def test_publisher_uses_runtime_without_human_tokens_or_refresh(self):
+        api = self.client()
+        TunnelAPI(api, "org", token="machine-secret").request("GET", "bootstrap/")
+        call = api.session.request.call_args
+        self.assertEqual(call.args[1], "https://runtime.dataplicity.com/api/organisations/org/development-tunnels/bootstrap/")
+        self.assertEqual(call.kwargs["headers"]["Authorization"], "TunnelPublisher machine-secret")
+        self.assertFalse(call.kwargs["allow_redirects"])
+        api.ensure_fresh_access_token.assert_not_called()
+
+    def test_human_calls_keep_gateway_and_their_refresh_lifecycle(self):
+        api = self.client()
+        TunnelAPI(api, "org").request("GET", "bootstrap/")
+        call = api.session.request.call_args
+        self.assertEqual(call.args[1], "https://gateway.dataplicity.com/api/organisations/org/development-tunnels/bootstrap/")
+        self.assertEqual(call.kwargs["headers"]["Authorization"], "Bearer human-access")
+        api.ensure_fresh_access_token.assert_called_once()
+
+    def test_rejected_publisher_does_not_erase_saved_human_tokens(self):
+        api = self.client()
+        api.session.request.return_value = Mock(status_code=401, text="invalid token", json=Mock(return_value={"detail": "invalid token"}))
+        with self.assertRaises(TunnelError):
+            TunnelAPI(api, "org", token="revoked-machine").request("GET", "bootstrap/")
+        self.assertEqual(api.config.access_token, "human-access")
+        self.assertEqual(api.config.refresh_token, "human-refresh")
+        api.ensure_fresh_access_token.assert_not_called()
+
+    def test_refresh_never_follows_a_redirect_with_the_human_secret(self):
+        api = self.client()
+        api.session.post = Mock(return_value=Mock(status_code=302))
+        self.assertEqual(api._attempt_refresh(), "failed")
+        self.assertFalse(api.session.post.call_args.kwargs["allow_redirects"])
+        self.assertEqual(api.config.refresh_token, "human-refresh")
+
+
+    def test_explicit_lowercase_machine_authority_has_no_human_session_lifecycle(self):
+        api = self.client()
+        api.session.request.return_value = Mock(status_code=401, text="invalid token", json=Mock(return_value={"detail": "invalid token"}))
+        api._refresh_access_token = Mock()
+        response = api.request('GET', '/api/runtime/', headers={'authorization': 'TunnelPublisher revoked'})
+        self.assertFalse(response.ok)
+        self.assertEqual(api.session.request.call_args.kwargs['headers']['Authorization'], 'TunnelPublisher revoked')
+        self.assertEqual(sum(key.lower() == 'authorization' for key in api.session.request.call_args.kwargs['headers']), 1)
+        api.ensure_fresh_access_token.assert_not_called()
+        api._refresh_access_token.assert_not_called()
+        self.assertEqual(api.config.access_token, 'human-access')
+
+    def test_publisher_standard_https_aliases_and_custom_authority(self):
+        from dataplicity_cli.tunnels import publisher_origin
+        for suffix in ('dataplicity.com', 'staging.dpenv.com'):
+            for alias in ('api', 'app-api', 'gateway'):
+                self.assertEqual(publisher_origin(f'https://{alias}.{suffix}:443/'), f'https://runtime.{suffix}')
+        for origin in ('https://gateway.dataplicity.com:8443', 'https://custom.example/runtime', 'https://gateway.dataplicity.com.attacker.test', 'https://gateway.dataplicity.com:invalid', 'https://[invalid'):
+            self.assertEqual(publisher_origin(origin), origin)

tests/test_tunnels.py

@@ -33,7 +33,7 @@ def test_names_and_secure_owner_urls(self):
                 websocket_url(url)
 
     def test_auth_separation_and_errors_never_reflect_secrets(self):
-        api = Mock(config=SimpleNamespace(auth_method="jwt", access_token="jwt"))
+        api = Mock(config=SimpleNamespace(auth_method="jwt", access_token="jwt", base_url="https://gateway.dataplicity.com"))
         control = TunnelAPI(api, "org-1")
         api.request.return_value = ApiResponse(True, 200, {"results": []}, "")
         self.assertEqual(control.request("GET", ""), {"results": []})
@@ -57,6 +57,7 @@ def test_auth_separation_and_errors_never_reflect_secrets(self):
         machine.request("POST", "publish/", payload={"name": "api"})
         self.assertEqual(api.request.call_args.kwargs["headers"], {"Authorization": "TunnelPublisher machine"})
         self.assertFalse(api.request.call_args.kwargs["allow_refresh"])
+        self.assertEqual(api.request.call_args.args[1], "https://runtime.dataplicity.com/api/organisations/org-1/development-tunnels/publish/")
 
 
 class LifecycleTests(unittest.IsolatedAsyncioTestCase):