SG-43166 Integrate MEDM authentication into Toolkit bootstrap

python/tank/__init__.py

@@ -155,6 +155,7 @@ def __fix_tank_vendor():
     get_authenticated_user,
 )
 from .api import Sgtk, sgtk_from_path, sgtk_from_entity
+from .authentication.flow_auth import get_flow_client, get_flow_access_token
 from .pipelineconfig_utils import (
     get_python_interpreter_for_config,
     get_core_python_path_for_config,

python/tank/authentication/__init__.py

@@ -33,6 +33,7 @@
     set_shotgun_authenticator_support_web_login,
 )
 from .shotgun_authenticator import ShotgunAuthenticator
+from .flow_auth import get_flow_access_token, FlowAuthenticationHandler, get_flow_client
 from .defaults_manager import DefaultsManager
 from .core_defaults_manager import CoreDefaultsManager
 from .user import (  # noqa

python/tank/authentication/flow_auth/__init__.py

@@ -0,0 +1,41 @@
+# 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.
+
+"""
+Flow / MEDM authentication for Toolkit bootstrap.
+
+Triggered during bootstrap when a project is "AM-ready". Obtains an APS
+access token via PKCE; the token is cached in a local file store for reuse.
+"""
+
+from ._authentication import (
+    init_authentication,
+    get_access_token,
+    get_flow_access_token,
+    check_token_expiry,
+)
+from ._client import FlowAuthenticationHandler, get_flow_client
+from ._constants import AM_READY_PROJECT_FIELD
+from ._settings import FlowAuthSettings, resolve_flow_auth_settings
+from .errors import FlowAuthError, FlowAuthConfigurationError
+
+__all__ = [
+    "init_authentication",
+    "get_access_token",
+    "get_flow_access_token",
+    "check_token_expiry",
+    "FlowAuthenticationHandler",
+    "get_flow_client",
+    "FlowAuthSettings",
+    "resolve_flow_auth_settings",
+    "FlowAuthError",
+    "FlowAuthConfigurationError",
+    "AM_READY_PROJECT_FIELD",
+]

python/tank/authentication/flow_auth/_authentication.py

@@ -0,0 +1,193 @@
+# 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.
+"""
+This module contains utilities for authenticating into Flow.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import time
+
+from tank_vendor.adsk_auth import (
+    AuthConfig,
+    get_access_token as get_access_token_from_adsk_auth,
+)
+
+from ... import LogManager
+from ...util import LocalFileStorageManager
+from ._constants import REQUIRED_SCOPES
+from .errors import FlowAuthConfigurationError
+
+logger = LogManager.get_logger(__name__)
+
+_aps_configuration = None
+
+
+def init_authentication(settings):
+    """Initialize authentication configuration with configured settings.
+
+    Args:
+        settings: FlowAuthSettings (or duck-typed equivalent) containing
+            auth_application_id, auth_base_url, auth_callback_url.
+
+    Raises:
+        FlowAuthConfigurationError: If required authentication settings are missing or invalid.
+    """
+    global _aps_configuration
+
+    auth_application_id = settings.auth_application_id
+    auth_base_url = settings.auth_base_url
+    auth_callback_url = settings.auth_callback_url
+
+    if not auth_application_id:
+        raise FlowAuthConfigurationError(
+            details="Required setting 'auth_application_id' is not configured."
+        )
+    if not auth_base_url:
+        raise FlowAuthConfigurationError(
+            details="Required setting 'auth_base_url' is not configured."
+        )
+    if not auth_callback_url:
+        raise FlowAuthConfigurationError(
+            details="Required setting 'auth_callback_url' is not configured."
+        )
+
+    _aps_configuration = AuthConfig(
+        application_id=auth_application_id,
+        base_url=auth_base_url,
+        callback_url=auth_callback_url,
+        description="PKCE Test App",
+        required_application_scopes=REQUIRED_SCOPES,
+        storage_dir=LocalFileStorageManager.get_global_root(
+            LocalFileStorageManager.CACHE
+        ),
+    )
+
+
+def _get_aps_configuration():
+    """Get the APS configuration.
+
+    Returns:
+        tank_vendor.adsk_auth.AuthConfig: The initialized auth configuration.
+
+    Raises:
+        RuntimeError: If authentication has not been initialized.
+    """
+    if _aps_configuration is None:
+        raise RuntimeError(
+            "Authentication not initialized. Call init_authentication() first."
+        )
+    return _aps_configuration
+
+
+def check_token_expiry(token: str, buffer_seconds: int = 300) -> bool:
+    """
+    Check if the given token is expiring soon (within the buffer period).
+
+    The Flow API can fail if the token is not yet expired but will expire soon,
+    so we proactively refresh when within the buffer.
+
+    Args:
+        token: The access token to check.
+        buffer_seconds: Seconds before expiry to consider the token "expiring soon".
+            Defaults to 300 (5 minutes).
+
+    Returns:
+        True if the token is expiring soon or invalid, False if it has
+        sufficient validity remaining.
+    """
+    try:
+        payload = _decode_token_payload(token)
+    except Exception as e:
+        logger.error("Error decoding token: %s", e)
+        return True
+
+    exp_timestamp = payload.get("exp") if payload else None
+
+    if not exp_timestamp:
+        logger.warning(
+            "Token does not contain 'exp' claim. Treating token as expiring soon."
+        )
+        return True
+
+    current_timestamp = time.time()
+    time_remaining = exp_timestamp - current_timestamp
+    if time_remaining < buffer_seconds:
+        logger.debug(
+            "Token will expire in %.0f seconds, less than buffer of %s seconds.",
+            time_remaining,
+            buffer_seconds,
+        )
+        return True
+
+    return False
+
+
+def _decode_token_payload(token: str):
+    """Decode JWT payload without verification (only used to read exp)."""
+    try:
+        parts = token.split(".")
+        if len(parts) != 3:
+            return None
+        payload_b64 = parts[1]
+        padding = (4 - len(payload_b64) % 4) % 4
+        payload_b64 += "=" * padding
+        payload_bytes = base64.urlsafe_b64decode(payload_b64)
+        return json.loads(payload_bytes)
+    except Exception:
+        return None
+
+
+def _auth_options_from_kwargs(kwargs):
+    """Extract adsk_auth get_access_token options from kwargs (e.g. authentication_options)."""
+    opts = kwargs.get("authentication_options")
+    if opts is None:
+        return {}
+    return {
+        "profile": getattr(opts, "user_profile", None),
+        "force_refresh": getattr(opts, "force_refresh", False),
+        "force_reauthentication": getattr(opts, "force_reauthentication", False),
+        "time_out": getattr(opts, "time_out", 30.0),
+        "browser": getattr(opts, "browser", None),
+    }
+
+
+def get_access_token(*args, **kwargs) -> str:
+    """Get access token from file store or web (PKCE).
+
+    Ensures the returned token has at least 5 minutes of validity remaining,
+    since the Flow API can fail when the token is about to expire.
+    """
+    config = _get_aps_configuration()
+    options = _auth_options_from_kwargs(kwargs)
+    token = get_access_token_from_adsk_auth(config, **options)
+
+    if check_token_expiry(token):
+        logger.info("Access token is expiring within 5 minutes. Forcing a refresh.")
+        options["force_refresh"] = True
+        token = get_access_token_from_adsk_auth(config, **options)
+
+    return token
+
+
+def get_flow_access_token(**kwargs) -> str:
+    """Get a Flow access token, lazy-initialising auth if needed.
+
+    Safe to call without a prior ``init_authentication()`` — if bootstrap has
+    not already initialised the APS configuration, settings are resolved from
+    environment variables and initialisation is performed automatically.
+    """
+    if _aps_configuration is None:
+        from ._settings import resolve_flow_auth_settings
+
+        init_authentication(resolve_flow_auth_settings())
+    return get_access_token(**kwargs)

python/tank/authentication/flow_auth/_client.py

@@ -0,0 +1,48 @@
+# 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.
+
+"""
+Flow GQL SDK client factory with authentication wired in.
+"""
+
+from ._authentication import get_flow_access_token
+
+
+class FlowAuthenticationHandler:
+    """Auth adapter for the Flow GQL SDK client.
+
+    Satisfies the SDK's ``auth_handler`` interface: the client calls
+    ``get_authentication_token()`` on every request so that short-lived tokens
+    are transparently refreshed without recreating the client.
+    """
+
+    def get_authentication_token(self) -> str:
+        return get_flow_access_token()
+
+
+def get_flow_client(endpoint_url=None):
+    """Return a ready-to-use Flow GQL SDK client with authentication wired in.
+
+    Lazy-initialises APS auth if bootstrap has not already done so. Uses the
+    SDK default endpoint (``tank_vendor.flow_data_sdk.config.DEFAULT_ENDPOINT``)
+    when ``endpoint_url`` is not supplied.
+
+    :param endpoint_url: Override the GraphQL endpoint. Defaults to the SDK's
+        production endpoint.
+    :type endpoint_url: str or None
+    :returns: Initialised ``GQLClient`` instance.
+    """
+    from tank_vendor.flow_data_sdk import GQLClient
+    from tank_vendor.flow_data_sdk.config import DEFAULT_ENDPOINT
+
+    return GQLClient(
+        endpoint=endpoint_url or DEFAULT_ENDPOINT,
+        auth_handler=FlowAuthenticationHandler(),
+    )

python/tank/authentication/flow_auth/_constants.py

@@ -0,0 +1,26 @@
+# 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.
+
+"""Constants for Flow / MEDM authentication."""
+
+AM_READY_PROJECT_FIELD = "sg_flow_am_id"
+
+DEFAULT_AUTH_APPLICATION_ID = "8QyoQKXZ7HDuQFmptJGrzsp2GwpATmyV"
+DEFAULT_AUTH_BASE_URL = "https://api.aps.usa.autodesk.com"
+DEFAULT_AUTH_CALLBACK_URL = "http://localhost:4201/auth/callback"
+
+# Previously "openid" was also requested but is not used or required for
+# authentication to Flow, and exceeded Windows Credential Manager's 1280-char
+# limit. Safe to exclude.
+REQUIRED_SCOPES = [
+    "data:read",
+    "data:write",
+    "data:create",
+]

python/tank/authentication/flow_auth/_settings.py

@@ -0,0 +1,51 @@
+# 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.
+
+"""Settings resolver for Flow / MEDM authentication."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+from ._constants import (
+    DEFAULT_AUTH_APPLICATION_ID,
+    DEFAULT_AUTH_BASE_URL,
+    DEFAULT_AUTH_CALLBACK_URL,
+)
+
+
+@dataclass
+class FlowAuthSettings:
+    """APS credentials required to perform PKCE authentication."""
+
+    auth_application_id: str
+    auth_base_url: str
+    auth_callback_url: str
+
+
+def resolve_flow_auth_settings() -> FlowAuthSettings:
+    """
+    Resolve APS auth settings: env-var overrides, falling back to hardcoded defaults.
+
+    Override env vars:
+        TK_FLOW_AUTH_APPLICATION_ID
+        TK_FLOW_AUTH_BASE_URL
+        TK_FLOW_AUTH_CALLBACK_URL
+    """
+    return FlowAuthSettings(
+        auth_application_id=os.environ.get(
+            "TK_FLOW_AUTH_APPLICATION_ID", DEFAULT_AUTH_APPLICATION_ID
+        ),
+        auth_base_url=os.environ.get("TK_FLOW_AUTH_BASE_URL", DEFAULT_AUTH_BASE_URL),
+        auth_callback_url=os.environ.get(
+            "TK_FLOW_AUTH_CALLBACK_URL", DEFAULT_AUTH_CALLBACK_URL
+        ),
+    )