Authentication successful. You can close this window.
+ +""" + + +class _CallbackHandler(BaseHTTPRequestHandler): + """Capture OAuth callback ?code=...&state=... from the local HTTP server.""" + + session_store: Dict[str, Any] = {} + + def do_GET(self) -> None: + q = parse_qs(urlparse(self.path).query) + if "error" in q: + msg = q.get("error_description", q.get("error", [b"Unknown error"])) + _logger.error("OAuth callback error: %s", msg[0] if msg else "unknown") + elif "code" in q and "state" in q: + self.session_store[q["state"][0]] = q["code"][0] + _logger.info("Received OAuth callback with code and state") + self.send_response(200) + self.end_headers() + self.wfile.write(_CALLBACK_HTML) + # Do NOT call server.shutdown() here: with ThreadingMixIn the handler runs + # in a worker thread; the main server thread is blocked in accept(). Rely + # on the main thread polling session_store and timing out instead. + + def log_message(self, format: str, *args: Any) -> None: + _logger.debug("%s", args) + + +class _ThreadingCallbackServerDualStack(ThreadingMixIn, HTTPServer): + """Threaded HTTP server binding to :: with IPV6_V6ONLY=0 (dual-stack).""" + + address_family = socket.AF_INET6 + + def server_bind(self) -> None: + self.socket = socket.socket(self.address_family, socket.SOCK_STREAM) + self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + if hasattr(socket, "IPV6_V6ONLY"): + self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) + self.socket.bind(("::", self.server_address[1])) + self.server_address = self.socket.getsockname() + + +class _ThreadingCallbackServerIPv4(ThreadingMixIn, HTTPServer): + """Threaded HTTP server binding to 0.0.0.0 (IPv4 only). Fallback when IPv6 is unavailable.""" + + def server_bind(self) -> None: + self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self.socket.bind(self.server_address) + self.server_address = self.socket.getsockname() + + +def run_callback_server( + session_store: Dict[str, Any], + port: int, + ready_event: threading.Event | None = None, + init_error: list[BaseException] | None = None, +) -> None: + """Run HTTP server in current thread to capture ?code=...&state=... .""" + _CallbackHandler.session_store = session_store + port = int(port) + try: + server = _ThreadingCallbackServerDualStack(("::", port), _CallbackHandler) + except OSError as e: + err = getattr(e, "errno", None) + if err in (errno.EADDRINUSE, errno.EACCES): + if init_error is not None: + init_error.append(e) + if ready_event is not None: + ready_event.set() + return + if err in _ERRNO_AF_NOT_SUPPORTED: + _logger.debug("IPv6 not available, using IPv4 callback server") + server = _ThreadingCallbackServerIPv4(("0.0.0.0", port), _CallbackHandler) + else: + raise + # Request handler threads must be daemon so the process exits after we have the code. + server.daemon_threads = True + if ready_event is not None: + ready_event.set() + server.serve_forever() + + +def web_authenticate( + config: AuthConfig, + *, + time_out: float = 30.0, + browser: Any = None, +) -> Dict[str, Any]: + """Run PKCE in browser; return token dict (access_token, refresh_token, ...).""" + import time + + code_challenge, code_verifier = create_code_pair() + auth_url, state = build_authorize_url(config, code_challenge) + session_store: Dict[str, Any] = {state: None} + port = _callback_server_port(config.callback_url) + if _is_port_in_use(port): + raise RuntimeError( + f"Port {port} is already in use (callback URL {config.callback_url}). " + "Another auth process may be running. Exit it or use a different callback URL." + ) + ready_event = threading.Event() + init_error: list[BaseException] = [] + + server_thread = threading.Thread( + target=run_callback_server, + args=(session_store, port), + kwargs={"ready_event": ready_event, "init_error": init_error}, + ) + server_thread.daemon = True + server_thread.start() + # Ensure server is bound and listening before opening browser (avoids race on Windows) + if not ready_event.wait(timeout=5): + _logger.warning("Callback server may not be ready yet") + if init_error: + e = init_error[0] + raise RuntimeError( + f"Port {port} is already in use (callback URL {config.callback_url}). " + "Another auth process may be running. Exit it or use a different callback URL." + ) from e + time.sleep(0.5) + try: + b = webbrowser.get(using=browser) + b.open_new(auth_url) + # With ThreadingMixIn the server thread never exits; poll with short joins + # so we return as soon as the callback handler has set session_store[state]. + deadline = time.monotonic() + time_out + while time.monotonic() < deadline: + if session_store.get(state) is not None: + break + server_thread.join(timeout=0.25) + except webbrowser.Error as e: + _logger.error("Browser error: %s", e) + server_thread.join(timeout=0.5) + + if session_store.get(state) is None: + raise RuntimeError("Failed to obtain authorization code from browser") + code = session_store[state] + token = exchange_code(config, code, code_verifier) + if not token or "access_token" not in token: + raise RuntimeError("Token exchange did not return an access_token") + return token diff --git a/python/tank_vendor/adsk_auth/token.py b/python/tank_vendor/adsk_auth/token.py new file mode 100644 index 000000000..2eee0984b --- /dev/null +++ b/python/tank_vendor/adsk_auth/token.py @@ -0,0 +1,107 @@ +# Copyright (c) 2025 Shotgun Software Inc. +# CONFIDENTIAL AND PROPRIETARY + +"""Get access token: file store -> refresh -> browser PKCE.""" + +from __future__ import annotations + +import logging +from typing import Any, Optional + +import jwt +from urllib.error import HTTPError + +from .config import AuthConfig +from .file_store import ( + delete_tokens, + get_access_token as get_access_token_from_store, + get_refresh_token, + get_user_profile, + persist_tokens, +) +from .pkce import exchange_refresh_token, web_authenticate + +_logger = logging.getLogger(__name__) + +# In-memory cache: profile -> access_token (avoids file read every call) +_access_token_cache: dict[str, str] = {} + + +def get_access_token( + config: AuthConfig, + *, + profile: Optional[str] = None, + force_refresh: bool = False, + force_reauthentication: bool = False, + time_out: float = 30.0, + browser: Any = None, +) -> str: + """ + Return a valid access token: use cache, then file store, then refresh, then browser PKCE. + + Raises: + RuntimeError: If a token could not be obtained. + """ + global _access_token_cache + user_profile = get_user_profile(profile) + + if force_reauthentication or force_refresh: + _access_token_cache.pop(user_profile, None) + + # 1. Valid token in cache? + cached = _access_token_cache.get(user_profile) + if cached and not (force_refresh or force_reauthentication): + try: + jwt.decode(cached, options={"verify_signature": False, "verify_exp": True}) + return cached + except (jwt.ExpiredSignatureError, jwt.DecodeError): + pass + + if force_reauthentication: + delete_tokens(config.storage_dir, config.application_id, user_profile) + + # 2. Valid token in file store? + if not (force_reauthentication or force_refresh): + try: + access_token = get_access_token_from_store( + config.storage_dir, config.application_id, user_profile + ) + if access_token: + jwt.decode( + access_token, + options={"verify_signature": False, "verify_exp": True}, + ) + _access_token_cache[user_profile] = access_token + return access_token + except (jwt.ExpiredSignatureError, jwt.DecodeError): + pass + + # 3. Refresh token? + try: + refresh_token = get_refresh_token( + config.storage_dir, config.application_id, user_profile + ) + if refresh_token: + _logger.debug("Using refresh token") + token_dict = exchange_refresh_token(config, refresh_token) + persist_tokens(config.storage_dir, config.application_id, user_profile, token_dict) + access_token = token_dict["access_token"] + _access_token_cache[user_profile] = access_token + return access_token + except (RuntimeError, HTTPError) as e: + _logger.debug("Refresh failed: %s", e) + + # 4. Browser PKCE + _logger.warning("Opening browser to authenticate (timeout %.1fs)", time_out) + token_dict = web_authenticate(config, time_out=time_out, browser=browser) + persist_tokens(config.storage_dir, config.application_id, user_profile, token_dict) + access_token = token_dict["access_token"] + _access_token_cache[user_profile] = access_token + return access_token + + +def clear_stored_tokens(config: AuthConfig, profile: Optional[str] = None) -> None: + """Remove tokens from file store and in-memory cache for this app/profile.""" + user_profile = get_user_profile(profile) + _access_token_cache.pop(user_profile, None) + delete_tokens(config.storage_dir, config.application_id, user_profile) diff --git a/requirements/3.10/frozen_requirements.txt b/requirements/3.10/frozen_requirements.txt index 589c0f0ad..8c5a85cc4 100644 --- a/requirements/3.10/frozen_requirements.txt +++ b/requirements/3.10/frozen_requirements.txt @@ -1,5 +1,6 @@ distro==1.4.0 packaging==25.0 +PyJWT==2.10.1 PyYAML==6.0.1 ruamel.yaml==0.18.14 ruamel.yaml.clib==0.2.15 diff --git a/requirements/3.10/pkgs.zip b/requirements/3.10/pkgs.zip index 4811a07cf..27cccb4d7 100644 Binary files a/requirements/3.10/pkgs.zip and b/requirements/3.10/pkgs.zip differ diff --git a/requirements/3.10/requirements.txt b/requirements/3.10/requirements.txt index eed0486f1..a6040e76c 100644 --- a/requirements/3.10/requirements.txt +++ b/requirements/3.10/requirements.txt @@ -1,5 +1,6 @@ distro==1.4.0 packaging==25.0 +pyjwt==2.10.1 pyyaml==6.0.1 ruamel.yaml==0.18.14 shotgun_api3==3.10.0 diff --git a/requirements/3.11/frozen_requirements.txt b/requirements/3.11/frozen_requirements.txt index 284e721cd..8f24cb5b0 100644 --- a/requirements/3.11/frozen_requirements.txt +++ b/requirements/3.11/frozen_requirements.txt @@ -1,5 +1,6 @@ distro==1.9.0 packaging==25.0 +PyJWT==2.10.1 PyYAML==6.0.2 ruamel.yaml==0.18.14 ruamel.yaml.clib==0.2.15 diff --git a/requirements/3.11/pkgs.zip b/requirements/3.11/pkgs.zip index 4d502a0e3..cc5cd627f 100644 Binary files a/requirements/3.11/pkgs.zip and b/requirements/3.11/pkgs.zip differ diff --git a/requirements/3.11/requirements.txt b/requirements/3.11/requirements.txt index 4c705e437..dee6aac79 100644 --- a/requirements/3.11/requirements.txt +++ b/requirements/3.11/requirements.txt @@ -1,5 +1,6 @@ distro==1.9.0 packaging==25.0 +pyjwt==2.10.1 pyyaml==6.0.2 ruamel.yaml==0.18.14 shotgun_api3==3.10.0 diff --git a/requirements/3.13/frozen_requirements.txt b/requirements/3.13/frozen_requirements.txt index 284e721cd..8f24cb5b0 100644 --- a/requirements/3.13/frozen_requirements.txt +++ b/requirements/3.13/frozen_requirements.txt @@ -1,5 +1,6 @@ distro==1.9.0 packaging==25.0 +PyJWT==2.10.1 PyYAML==6.0.2 ruamel.yaml==0.18.14 ruamel.yaml.clib==0.2.15 diff --git a/requirements/3.13/pkgs.zip b/requirements/3.13/pkgs.zip index 78a1b8f1a..e77ab94fb 100644 Binary files a/requirements/3.13/pkgs.zip and b/requirements/3.13/pkgs.zip differ diff --git a/requirements/3.13/requirements.txt b/requirements/3.13/requirements.txt index 4c705e437..dee6aac79 100644 --- a/requirements/3.13/requirements.txt +++ b/requirements/3.13/requirements.txt @@ -1,5 +1,6 @@ distro==1.9.0 packaging==25.0 +pyjwt==2.10.1 pyyaml==6.0.2 ruamel.yaml==0.18.14 shotgun_api3==3.10.0 diff --git a/requirements/3.7/frozen_requirements.txt b/requirements/3.7/frozen_requirements.txt index fa55883f0..4297e8a40 100644 --- a/requirements/3.7/frozen_requirements.txt +++ b/requirements/3.7/frozen_requirements.txt @@ -1,6 +1,8 @@ distro==1.4.0 packaging==24.0 +PyJWT==2.8.0 PyYAML==5.4.1 ruamel.yaml==0.18.13 ruamel.yaml.clib==0.2.8 shotgun-api3==3.9.0 +typing-extensions==4.7.1 diff --git a/requirements/3.7/pkgs.zip b/requirements/3.7/pkgs.zip index 42f309292..d4bb313f8 100644 Binary files a/requirements/3.7/pkgs.zip and b/requirements/3.7/pkgs.zip differ diff --git a/requirements/3.7/requirements.txt b/requirements/3.7/requirements.txt index e4f56efec..360720fde 100644 --- a/requirements/3.7/requirements.txt +++ b/requirements/3.7/requirements.txt @@ -1,5 +1,6 @@ distro==1.4.0 packaging==24.0 +pyjwt==2.8.0 pyyaml==5.4.1 ruamel.yaml==0.18.13 shotgun_api3==3.9.0 diff --git a/requirements/3.9/frozen_requirements.txt b/requirements/3.9/frozen_requirements.txt index 212246dbc..ac9a68fc1 100644 --- a/requirements/3.9/frozen_requirements.txt +++ b/requirements/3.9/frozen_requirements.txt @@ -1,5 +1,6 @@ distro==1.4.0 packaging==25.0 +PyJWT==2.10.1 PyYAML==5.4.1 ruamel.yaml==0.18.14 ruamel.yaml.clib==0.2.15 diff --git a/requirements/3.9/pkgs.zip b/requirements/3.9/pkgs.zip index c0043ab13..1ada08fbd 100644 Binary files a/requirements/3.9/pkgs.zip and b/requirements/3.9/pkgs.zip differ diff --git a/requirements/3.9/requirements.txt b/requirements/3.9/requirements.txt index 4db18269d..b4f9ec9e3 100644 --- a/requirements/3.9/requirements.txt +++ b/requirements/3.9/requirements.txt @@ -1,5 +1,6 @@ distro==1.4.0 packaging==25.0 +pyjwt==2.10.1 pyyaml==5.4.1 ruamel.yaml==0.18.14 shotgun_api3==3.10.0 diff --git a/requirements/update_python_packages.py b/requirements/update_python_packages.py index 77c5e684d..abfabba5e 100644 --- a/requirements/update_python_packages.py +++ b/requirements/update_python_packages.py @@ -88,9 +88,14 @@ def main(): ] # Make sure we found as many Python packages as there - # are packages listed inside frozen_requirements.txt - # assert len(package_names) == nb_dependencies - assert len(package_names) >= nb_dependencies + # are packages listed inside frozen_requirements.txt. + # Count .dist-info directories rather than package directories because + # namespace packages (e.g. ruamel.yaml + ruamel.yaml.clib, jaraco.*) + # share a single parent directory, making package_names < nb_dependencies. + dist_info_count = len( + [name for name in os.listdir(temp_dir) if name.endswith(".dist-info")] + ) + assert dist_info_count >= nb_dependencies # Write out the zip file for python packages. Compress the zip file with ZIP_DEFLATED. Note # that this requires zlib to decompress when importing. Compression also causes import to diff --git a/tests/authentication_tests/test_adsk_auth_file_store.py b/tests/authentication_tests/test_adsk_auth_file_store.py new file mode 100644 index 000000000..9acea93dd --- /dev/null +++ b/tests/authentication_tests/test_adsk_auth_file_store.py @@ -0,0 +1,141 @@ +# Copyright (c) 2025 Shotgun Software Inc. +# +# CONFIDENTIAL AND PROPRIETARY +# +# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit +# Source Code License included in this distribution package. See LICENSE. +# By accessing, using, copying or modifying this work you indicate your +# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights +# not expressly granted therein are reserved by Shotgun Software Inc. + +import json +import os +import sys +import tempfile +import unittest + +from tank_test.tank_test_base import setUpModule # noqa + +from tank_vendor.adsk_auth import file_store + + +class FileStoreRoundTripTests(unittest.TestCase): + def setUp(self): + self._tmpdir = tempfile.TemporaryDirectory() + self.storage_dir = self._tmpdir.name + + def tearDown(self): + self._tmpdir.cleanup() + + def test_persist_and_get_access_token(self): + file_store.persist_tokens( + self.storage_dir, "myapp", "alice", {"access_token": "acc123"} + ) + result = file_store.get_access_token(self.storage_dir, "myapp", "alice") + self.assertEqual(result, "acc123") + + def test_persist_and_get_refresh_token(self): + file_store.persist_tokens( + self.storage_dir, + "myapp", + "alice", + {"access_token": "acc123", "refresh_token": "ref456"}, + ) + result = file_store.get_refresh_token(self.storage_dir, "myapp", "alice") + self.assertEqual(result, "ref456") + + def test_missing_file_returns_none(self): + self.assertIsNone( + file_store.get_access_token(self.storage_dir, "myapp", "nobody") + ) + self.assertIsNone( + file_store.get_refresh_token(self.storage_dir, "myapp", "nobody") + ) + + def test_missing_profile_returns_none(self): + file_store.persist_tokens( + self.storage_dir, "myapp", "alice", {"access_token": "acc123"} + ) + self.assertIsNone(file_store.get_access_token(self.storage_dir, "myapp", "bob")) + + def test_delete_tokens_clears_entry(self): + file_store.persist_tokens( + self.storage_dir, + "myapp", + "alice", + {"access_token": "acc123", "refresh_token": "ref456"}, + ) + file_store.delete_tokens(self.storage_dir, "myapp", "alice") + self.assertIsNone( + file_store.get_access_token(self.storage_dir, "myapp", "alice") + ) + self.assertIsNone( + file_store.get_refresh_token(self.storage_dir, "myapp", "alice") + ) + + def test_delete_tokens_on_missing_file_is_silent(self): + # Should not raise when no tokens exist yet. + file_store.delete_tokens(self.storage_dir, "myapp", "alice") + + def test_delete_tokens_only_removes_target_profile(self): + file_store.persist_tokens( + self.storage_dir, "myapp", "alice", {"access_token": "acc_a"} + ) + file_store.persist_tokens( + self.storage_dir, "myapp", "bob", {"access_token": "acc_b"} + ) + file_store.delete_tokens(self.storage_dir, "myapp", "alice") + self.assertIsNone( + file_store.get_access_token(self.storage_dir, "myapp", "alice") + ) + self.assertEqual( + file_store.get_access_token(self.storage_dir, "myapp", "bob"), "acc_b" + ) + + def test_persist_overwrites_existing_token(self): + file_store.persist_tokens( + self.storage_dir, "myapp", "alice", {"access_token": "old"} + ) + file_store.persist_tokens( + self.storage_dir, "myapp", "alice", {"access_token": "new"} + ) + self.assertEqual( + file_store.get_access_token(self.storage_dir, "myapp", "alice"), "new" + ) + + def test_storage_dir_created_on_first_write(self): + nested = os.path.join(self.storage_dir, "a", "b", "c") + file_store.persist_tokens(nested, "myapp", "alice", {"access_token": "t"}) + self.assertTrue(os.path.isdir(nested)) + + @unittest.skipIf(sys.platform == "win32", "POSIX permissions only") + def test_file_permissions_are_restrictive(self): + file_store.persist_tokens( + self.storage_dir, "myapp", "alice", {"access_token": "t"} + ) + path = os.path.join(self.storage_dir, "adsk_flow_tokens.json") + mode = oct(os.stat(path).st_mode & 0o777) + self.assertEqual(mode, oct(0o600)) + + def test_token_file_is_valid_json(self): + file_store.persist_tokens( + self.storage_dir, + "myapp", + "alice", + {"access_token": "acc", "refresh_token": "ref"}, + ) + path = os.path.join(self.storage_dir, "adsk_flow_tokens.json") + with open(path) as fh: + data = json.load(fh) + self.assertIn("adsk.flow.myapp.access_token", data) + self.assertIn("adsk.flow.myapp.refresh_token", data) + + +class GetUserProfileTests(unittest.TestCase): + def test_returns_provided_profile(self): + self.assertEqual(file_store.get_user_profile("alice"), "alice") + + def test_defaults_to_os_user(self): + import getpass + + self.assertEqual(file_store.get_user_profile(None), getpass.getuser()) diff --git a/tests/authentication_tests/test_flow_auth.py b/tests/authentication_tests/test_flow_auth.py new file mode 100644 index 000000000..ef7643f4f --- /dev/null +++ b/tests/authentication_tests/test_flow_auth.py @@ -0,0 +1,278 @@ +# Copyright (c) 2026 Shotgun Software Inc. +# +# CONFIDENTIAL AND PROPRIETARY +# +# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit +# Source Code License included in this distribution package. See LICENSE. +# By accessing, using, copying or modifying this work you indicate your +# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights +# not expressly granted therein are reserved by Shotgun Software Inc. + +import base64 +import json +import time + +from tank_test.tank_test_base import setUpModule # noqa +from tank_test.tank_test_base import mock, ShotgunTestBase + +from tank.authentication import flow_auth +from tank.authentication.flow_auth import _authentication as flow_auth_impl +from tank.authentication.flow_auth.errors import FlowAuthConfigurationError + + +def _make_jwt(payload): + """Build a minimal unsigned JWT-shaped string. Header/signature are ignored + by the unverified decode used in flow_auth.""" + header_b64 = ( + base64.urlsafe_b64encode(b'{"alg":"none"}').rstrip(b"=").decode("ascii") + ) + payload_b64 = ( + base64.urlsafe_b64encode(json.dumps(payload).encode("utf-8")) + .rstrip(b"=") + .decode("ascii") + ) + return f"{header_b64}.{payload_b64}.sig" + + +class _Settings: + """Duck-typed FlowAuthSettings stand-in.""" + + def __init__( + self, + app_id="app", + base_url="https://aps.example.com", + callback="http://localhost:8080/cb", + ): + self.auth_application_id = app_id + self.auth_base_url = base_url + self.auth_callback_url = callback + + +class InitAuthenticationTests(ShotgunTestBase): + def setUp(self): + super().setUp() + flow_auth_impl._aps_configuration = None + + def tearDown(self): + flow_auth_impl._aps_configuration = None + super().tearDown() + + def test_missing_application_id_raises(self): + with self.assertRaises(FlowAuthConfigurationError): + flow_auth.init_authentication(_Settings(app_id="")) + + def test_missing_base_url_raises(self): + with self.assertRaises(FlowAuthConfigurationError): + flow_auth.init_authentication(_Settings(base_url="")) + + def test_missing_callback_url_raises(self): + with self.assertRaises(FlowAuthConfigurationError): + flow_auth.init_authentication(_Settings(callback="")) + + def test_valid_settings_initializes_config(self): + flow_auth.init_authentication(_Settings()) + self.assertIsNotNone(flow_auth_impl._aps_configuration) + self.assertEqual(flow_auth_impl._aps_configuration.application_id, "app") + + +class CheckTokenExpiryTests(ShotgunTestBase): + def test_fresh_token_returns_false(self): + token = _make_jwt({"exp": int(time.time()) + 3600}) # 1h in future + self.assertFalse(flow_auth.check_token_expiry(token)) + + def test_expiring_within_buffer_returns_true(self): + token = _make_jwt({"exp": int(time.time()) + 60}) # well under 300s buffer + self.assertTrue(flow_auth.check_token_expiry(token)) + + def test_expired_token_returns_true(self): + token = _make_jwt({"exp": int(time.time()) - 60}) + self.assertTrue(flow_auth.check_token_expiry(token)) + + def test_token_without_exp_claim_returns_true(self): + token = _make_jwt({"sub": "user"}) + self.assertTrue(flow_auth.check_token_expiry(token)) + + def test_custom_buffer(self): + token = _make_jwt({"exp": int(time.time()) + 400}) + self.assertFalse(flow_auth.check_token_expiry(token, buffer_seconds=300)) + self.assertTrue(flow_auth.check_token_expiry(token, buffer_seconds=600)) + + +class DecodeTokenPayloadTests(ShotgunTestBase): + def test_valid_jwt_returns_payload(self): + payload = {"sub": "alice", "exp": 1234567890} + token = _make_jwt(payload) + self.assertEqual(flow_auth_impl._decode_token_payload(token), payload) + + def test_malformed_jwt_returns_none(self): + self.assertIsNone( + flow_auth_impl._decode_token_payload("not.a.jwt.too.many.dots") + ) + + def test_non_jwt_string_returns_none(self): + self.assertIsNone(flow_auth_impl._decode_token_payload("plain-string")) + + def test_invalid_base64_returns_none(self): + self.assertIsNone(flow_auth_impl._decode_token_payload("a.@@@.c")) + + +class GetFlowAccessTokenTests(ShotgunTestBase): + def setUp(self): + super().setUp() + flow_auth_impl._aps_configuration = None + + def tearDown(self): + flow_auth_impl._aps_configuration = None + super().tearDown() + + @mock.patch( + "tank.authentication.flow_auth._authentication.get_access_token_from_adsk_auth" + ) + def test_delegates_when_already_initialised(self, mock_adsk): + flow_auth.init_authentication(_Settings()) + fresh = _make_jwt({"exp": int(time.time()) + 3600}) + mock_adsk.return_value = fresh + + result = flow_auth.get_flow_access_token() + + self.assertEqual(result, fresh) + self.assertEqual(mock_adsk.call_count, 1) + + @mock.patch( + "tank.authentication.flow_auth._authentication.get_access_token_from_adsk_auth" + ) + @mock.patch( + "tank.authentication.flow_auth._settings.resolve_flow_auth_settings", + return_value=_Settings(), + ) + def test_lazy_init_when_not_initialised(self, mock_resolve, mock_adsk): + fresh = _make_jwt({"exp": int(time.time()) + 3600}) + mock_adsk.return_value = fresh + + result = flow_auth.get_flow_access_token() + + # resolve_flow_auth_settings() was called to build settings, and + # _aps_configuration is now populated (real init_authentication ran). + mock_resolve.assert_called_once() + self.assertIsNotNone(flow_auth_impl._aps_configuration) + self.assertEqual(result, fresh) + + +class FlowAuthenticationHandlerTests(ShotgunTestBase): + def setUp(self): + super().setUp() + flow_auth_impl._aps_configuration = None + + def tearDown(self): + flow_auth_impl._aps_configuration = None + super().tearDown() + + @mock.patch( + "tank.authentication.flow_auth._authentication.get_access_token_from_adsk_auth" + ) + def test_get_authentication_token_returns_token(self, mock_adsk): + flow_auth.init_authentication(_Settings()) + fresh = _make_jwt({"exp": int(time.time()) + 3600}) + mock_adsk.return_value = fresh + + from tank.authentication.flow_auth import FlowAuthenticationHandler + + handler = FlowAuthenticationHandler() + self.assertEqual(handler.get_authentication_token(), fresh) + + +class GetFlowClientTests(ShotgunTestBase): + _DEFAULT = "https://default.example.com/graphql" + + def setUp(self): + super().setUp() + flow_auth_impl._aps_configuration = None + # Stub the GQL SDK vendor modules so tests run without the zip installed. + self._mock_gql_cls = mock.MagicMock() + config_mod = mock.MagicMock(DEFAULT_ENDPOINT=self._DEFAULT) + data_mod = mock.MagicMock(GQLClient=self._mock_gql_cls) + self._sys_modules_patch = mock.patch.dict( + "sys.modules", + { + "tank_vendor.flow_data_sdk": data_mod, + "tank_vendor.flow_data_sdk.config": config_mod, + }, + ) + self._sys_modules_patch.start() + + def tearDown(self): + self._sys_modules_patch.stop() + flow_auth_impl._aps_configuration = None + super().tearDown() + + @mock.patch( + "tank.authentication.flow_auth._authentication.get_access_token_from_adsk_auth" + ) + def test_returns_client_with_default_endpoint(self, mock_adsk): + flow_auth.init_authentication(_Settings()) + mock_adsk.return_value = _make_jwt({"exp": int(time.time()) + 3600}) + + flow_auth.get_flow_client() + + self._mock_gql_cls.assert_called_once() + _, kwargs = self._mock_gql_cls.call_args + self.assertEqual(kwargs["endpoint"], self._DEFAULT) + self.assertIn("auth_handler", kwargs) + + @mock.patch( + "tank.authentication.flow_auth._authentication.get_access_token_from_adsk_auth" + ) + def test_passes_custom_endpoint(self, mock_adsk): + flow_auth.init_authentication(_Settings()) + mock_adsk.return_value = _make_jwt({"exp": int(time.time()) + 3600}) + custom_url = "https://staging.example.com/graphql" + + flow_auth.get_flow_client(custom_url) + + _, kwargs = self._mock_gql_cls.call_args + self.assertEqual(kwargs["endpoint"], custom_url) + + +class GetAccessTokenTests(ShotgunTestBase): + def setUp(self): + super().setUp() + flow_auth.init_authentication(_Settings()) + + def tearDown(self): + flow_auth_impl._aps_configuration = None + super().tearDown() + + @mock.patch( + "tank.authentication.flow_auth._authentication.get_access_token_from_adsk_auth" + ) + def test_returns_fresh_token_without_refresh(self, mock_adsk): + fresh = _make_jwt({"exp": int(time.time()) + 3600}) + mock_adsk.return_value = fresh + + result = flow_auth.get_access_token() + + self.assertEqual(result, fresh) + self.assertEqual(mock_adsk.call_count, 1) + + @mock.patch( + "tank.authentication.flow_auth._authentication.get_access_token_from_adsk_auth" + ) + def test_forces_refresh_when_expiring_soon(self, mock_adsk): + # First call returns a token expiring within the buffer; second call + # returns a fresh one after force_refresh is set. + expiring = _make_jwt({"exp": int(time.time()) + 60}) + fresh = _make_jwt({"exp": int(time.time()) + 3600}) + mock_adsk.side_effect = [expiring, fresh] + + result = flow_auth.get_access_token() + + self.assertEqual(result, fresh) + self.assertEqual(mock_adsk.call_count, 2) + # The second invocation should have force_refresh=True. + _, second_call_kwargs = mock_adsk.call_args_list[1] + self.assertTrue(second_call_kwargs.get("force_refresh")) + + def test_get_access_token_without_init_raises(self): + flow_auth_impl._aps_configuration = None + with self.assertRaises(RuntimeError): + flow_auth.get_access_token() diff --git a/tests/bootstrap_tests/test_manager_flow_auth.py b/tests/bootstrap_tests/test_manager_flow_auth.py new file mode 100644 index 000000000..441482514 --- /dev/null +++ b/tests/bootstrap_tests/test_manager_flow_auth.py @@ -0,0 +1,191 @@ +# Copyright (c) 2026 Shotgun Software Inc. +# +# CONFIDENTIAL AND PROPRIETARY +# +# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit +# Source Code License included in this distribution package. See LICENSE. +# By accessing, using, copying or modifying this work you indicate your +# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights +# not expressly granted therein are reserved by Shotgun Software Inc. + +from sgtk.bootstrap import ToolkitManager +from sgtk.bootstrap.errors import TankBootstrapError + +from tank.authentication import flow_auth + +from tank_test.tank_test_base import setUpModule # noqa +from tank_test.tank_test_base import ( + mock, + ShotgunTestBase, + temp_env_var, +) + + +@mock.patch( + "tank.authentication.ShotgunAuthenticator.get_user", + return_value=mock.Mock(), +) +class FlowAuthHookTests(ShotgunTestBase): + """Coverage for ToolkitManager._check_and_trigger_am_auth.""" + + PROJECT_ID = 42 + + def _build_manager_with_sg(self, sg_project_payload): + """Create a ToolkitManager whose _sg_connection.find_one returns + ``sg_project_payload`` for a Project query.""" + mgr = ToolkitManager() + mgr._sg_connection = mock.Mock() + mgr._sg_connection.find_one.return_value = sg_project_payload + return mgr + + @mock.patch("tank.authentication.flow_auth.get_access_token") + @mock.patch("tank.authentication.flow_auth.init_authentication") + @mock.patch("tank.authentication.flow_auth.resolve_flow_auth_settings") + def test_am_ready_project_triggers_auth(self, mock_resolve, mock_init, mock_get, _): + mock_resolve.return_value = mock.Mock() + mgr = self._build_manager_with_sg({flow_auth.AM_READY_PROJECT_FIELD: "abc-123"}) + + mgr._check_and_trigger_am_auth( + {"type": "Project", "id": self.PROJECT_ID}, progress_callback=mock.Mock() + ) + + mock_resolve.assert_called_once() + mock_init.assert_called_once_with(mock_resolve.return_value) + mock_get.assert_called_once() + + @mock.patch("tank.authentication.flow_auth.get_access_token") + @mock.patch("tank.authentication.flow_auth.init_authentication") + def test_non_am_ready_project_skips_auth(self, mock_init, mock_get, _): + mgr = self._build_manager_with_sg({flow_auth.AM_READY_PROJECT_FIELD: None}) + + mgr._check_and_trigger_am_auth( + {"type": "Project", "id": self.PROJECT_ID}, progress_callback=None + ) + + mock_init.assert_not_called() + mock_get.assert_not_called() + + @mock.patch("tank.authentication.flow_auth.get_access_token") + @mock.patch("tank.authentication.flow_auth.init_authentication") + def test_missing_project_entity_skips_auth(self, mock_init, mock_get, _): + mgr = self._build_manager_with_sg(None) + + mgr._check_and_trigger_am_auth( + {"type": "Project", "id": self.PROJECT_ID}, progress_callback=None + ) + + mock_init.assert_not_called() + mock_get.assert_not_called() + + @mock.patch("tank.authentication.flow_auth.get_access_token") + @mock.patch("tank.authentication.flow_auth.init_authentication") + def test_empty_project_entity_skips_auth(self, mock_init, mock_get, _): + mgr = self._build_manager_with_sg({}) + + mgr._check_and_trigger_am_auth( + {"type": "Project", "id": self.PROJECT_ID}, progress_callback=None + ) + + mock_init.assert_not_called() + mock_get.assert_not_called() + + @mock.patch("tank.authentication.flow_auth.get_access_token") + @mock.patch("tank.authentication.flow_auth.init_authentication") + def test_none_entity_skips_auth(self, mock_init, mock_get, _): + mgr = self._build_manager_with_sg({flow_auth.AM_READY_PROJECT_FIELD: "abc-123"}) + + mgr._check_and_trigger_am_auth(None, progress_callback=None) + + # find_one should not even be called when entity is None + mgr._sg_connection.find_one.assert_not_called() + mock_init.assert_not_called() + mock_get.assert_not_called() + + @mock.patch("tank.authentication.flow_auth.init_authentication") + @mock.patch("tank.authentication.flow_auth.resolve_flow_auth_settings") + def test_configuration_error_raises_TankBootstrapError( + self, mock_resolve, mock_init, _ + ): + mock_resolve.return_value = mock.Mock() + mock_init.side_effect = flow_auth.FlowAuthConfigurationError("missing app id") + mgr = self._build_manager_with_sg({flow_auth.AM_READY_PROJECT_FIELD: "abc-123"}) + + with self.assertRaises(TankBootstrapError): + mgr._check_and_trigger_am_auth( + {"type": "Project", "id": self.PROJECT_ID}, + progress_callback=mock.Mock(), + ) + + @mock.patch("tank.authentication.flow_auth.get_access_token") + @mock.patch("tank.authentication.flow_auth.init_authentication") + @mock.patch("tank.authentication.flow_auth.resolve_flow_auth_settings") + def test_runtime_error_soft_fails_by_default( + self, mock_resolve, mock_init, mock_get, _ + ): + mock_resolve.return_value = mock.Mock() + mock_get.side_effect = RuntimeError("network down") + mgr = self._build_manager_with_sg({flow_auth.AM_READY_PROJECT_FIELD: "abc-123"}) + + # Should not raise. + mgr._check_and_trigger_am_auth( + {"type": "Project", "id": self.PROJECT_ID}, progress_callback=mock.Mock() + ) + + @mock.patch("tank.authentication.flow_auth.get_access_token") + @mock.patch("tank.authentication.flow_auth.init_authentication") + @mock.patch("tank.authentication.flow_auth.resolve_flow_auth_settings") + def test_runtime_error_hard_fails_with_env_var( + self, mock_resolve, mock_init, mock_get, _ + ): + mock_resolve.return_value = mock.Mock() + mock_get.side_effect = RuntimeError("network down") + mgr = self._build_manager_with_sg({flow_auth.AM_READY_PROJECT_FIELD: "abc-123"}) + + with temp_env_var(TK_FLOW_AUTH_REQUIRED="1"): + with self.assertRaises(TankBootstrapError): + mgr._check_and_trigger_am_auth( + {"type": "Project", "id": self.PROJECT_ID}, + progress_callback=mock.Mock(), + ) + + +@mock.patch( + "tank.authentication.ShotgunAuthenticator.get_user", + return_value=mock.Mock(), +) +class ResolveProjectIdTests(ShotgunTestBase): + """Coverage for ToolkitManager._resolve_project_id (refactored helper).""" + + def test_none_entity_returns_none(self, _): + mgr = ToolkitManager() + self.assertIsNone(mgr._resolve_project_id(None)) + + def test_project_entity_returns_id(self, _): + mgr = ToolkitManager() + self.assertEqual(mgr._resolve_project_id({"type": "Project", "id": 99}), 99) + + def test_entity_with_project_link_returns_id(self, _): + mgr = ToolkitManager() + self.assertEqual( + mgr._resolve_project_id( + {"type": "Shot", "id": 1, "project": {"type": "Project", "id": 77}} + ), + 77, + ) + + def test_entity_without_project_link_queries_sg(self, _): + mgr = ToolkitManager() + mgr._sg_connection = mock.Mock() + mgr._sg_connection.find_one.return_value = { + "project": {"type": "Project", "id": 55} + } + + self.assertEqual(mgr._resolve_project_id({"type": "Shot", "id": 1}), 55) + + def test_entity_with_no_project_raises(self, _): + mgr = ToolkitManager() + mgr._sg_connection = mock.Mock() + mgr._sg_connection.find_one.return_value = None + + with self.assertRaises(TankBootstrapError): + mgr._resolve_project_id({"type": "Shot", "id": 1})