#!/usr/bin/env python3 """Cross-platform token provider for coding assistants using a persistent MSAL cache.""" import argparse import importlib import os from pathlib import Path import sys import tempfile from typing import Any DEFAULT_TENANT_ID = "021af73b-7bf8-4014-aa49-bcf05c8e29b1" DEFAULT_CLIENT_ID = "7507e4d7-e302-4814-8b91-adb7174320d6" DEFAULT_SCOPE = "api://f1d85b2e-d001-4cc6-9269-1b3b0b151d5d/access_api" CACHE_SERVICE_NAME = "org.childrensnational.coding-assistants" CACHE_ACCOUNT_NAME = "cnh-token" CACHE_FILE_NAME = "cnh-token-cache.bin" CACHE_DIRECTORY_NAME = "CNH" def write_stderr(message: str) -> None: print(message, file=sys.stderr) def get_cache_file_path() -> Path: if sys.platform == "darwin": cache_root = Path.home() / "Library" / "Application Support" elif sys.platform == "win32": cache_root = Path( os.environ.get("LOCALAPPDATA") or Path.home() / "AppData" / "Local" ) else: cache_root = Path( os.environ.get("XDG_CACHE_HOME") or Path.home() / ".cache" ).expanduser() cache_directory = cache_root / CACHE_DIRECTORY_NAME cache_directory.mkdir(parents=True, exist_ok=True) cache_file = cache_directory / CACHE_FILE_NAME try: if cache_file.exists(): if not cache_file.is_file(): raise OSError(f"Token cache path is not a file: {cache_file}") with cache_file.open("r+b"): pass else: probe_path: str | None = None try: descriptor, probe_path = tempfile.mkstemp( prefix=f".{CACHE_FILE_NAME}.", dir=cache_directory, ) os.close(descriptor) finally: if probe_path: Path(probe_path).unlink(missing_ok=True) except OSError as error: raise OSError(f"Token cache is not writable at {cache_file}: {error}") from error return cache_file def parse_arguments() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Acquire a CNH coding-assistant access token." ) parser.add_argument("--tenant-id", default=DEFAULT_TENANT_ID) parser.add_argument("--client-id", default=DEFAULT_CLIENT_ID) parser.add_argument("--scope", dest="scopes", action="append") parser.add_argument("--login-hint", default="") parser.add_argument("--expiration-skew-minutes", type=int, default=5) parser.add_argument( "--no-interactive-fallback", dest="allow_interactive_fallback", action="store_false", help="Fail when no usable cached token is available.", ) parser.set_defaults(allow_interactive_fallback=True) return parser.parse_args() def token_is_usable(token_result: dict[str, Any] | None, minimum_remaining_minutes: int) -> bool: if not token_result or not token_result.get("access_token"): return False try: expires_in = int(token_result["expires_in"]) except (KeyError, TypeError, ValueError): return False return expires_in > minimum_remaining_minutes * 60 def write_token_only(access_token: str) -> None: if not access_token or access_token.isspace(): raise ValueError("Access token was empty.") if len(access_token.split(".")) != 3: raise ValueError("Access token is not a well-formed JWT.") print(access_token) def acquire_silent_token( application: Any, scopes: list[str], login_hint: str, expiration_skew_minutes: int, ) -> dict[str, Any] | None: accounts = application.get_accounts(username=login_hint) if login_hint else application.get_accounts() for account in accounts: token_result = application.acquire_token_silent(scopes=scopes, account=account) if token_is_usable(token_result, expiration_skew_minutes): return token_result return None def create_token_cache(msal_extensions: Any) -> Any: # Native keychains store the secret; msal-extensions still needs this # writable path for cache synchronization and locking. cache_file = get_cache_file_path() if sys.platform == "darwin": persistence = msal_extensions.KeychainPersistence( str(cache_file), service_name=CACHE_SERVICE_NAME, account_name=CACHE_ACCOUNT_NAME, ) elif sys.platform == "win32": persistence = msal_extensions.FilePersistenceWithDataProtection( str(cache_file) ) else: persistence = msal_extensions.LibsecretPersistence( str(cache_file), schema_name=CACHE_SERVICE_NAME, attributes={"account_name": CACHE_ACCOUNT_NAME}, ) return msal_extensions.PersistedTokenCache(persistence) def main() -> int: arguments = parse_arguments() if arguments.expiration_skew_minutes < 0: write_stderr("Expiration skew minutes cannot be negative.") return 2 try: msal = importlib.import_module("msal") msal_extensions = importlib.import_module("msal_extensions") except ImportError as error: write_stderr( "The required Python packages could not be loaded. Install them with: " "python3 -m pip install --user msal msal-extensions. " f"Error: {error}" ) return 3 try: cache = create_token_cache(msal_extensions) application = msal.PublicClientApplication( client_id=arguments.client_id, authority=f"https://login.microsoftonline.com/{arguments.tenant_id}", token_cache=cache, ) scopes = arguments.scopes or [DEFAULT_SCOPE] token_result = acquire_silent_token( application, scopes, arguments.login_hint, arguments.expiration_skew_minutes, ) if token_result: write_token_only(token_result["access_token"]) return 0 if not arguments.allow_interactive_fallback: write_stderr( "No usable cached token was found, and interactive " "authentication is disabled." ) return 4 interactive_parameters: dict[str, Any] = { "scopes": scopes, } if arguments.login_hint: interactive_parameters["login_hint"] = arguments.login_hint token_result = application.acquire_token_interactive(**interactive_parameters) if not token_is_usable(token_result, arguments.expiration_skew_minutes): error_message = ( token_result.get("error_description", "No usable token was returned.") if token_result else "No usable token was returned." ) raise RuntimeError(f"Interactive authentication failed: {error_message}") write_token_only(token_result["access_token"]) return 0 except Exception as error: write_stderr(f"CNH token acquisition failed: {error}") return 1 if __name__ == "__main__": sys.exit(main())