feat(auth): Add SCM authentication support for repository access

Author: akosyakov

examples/anthropic_tool_use.py

@@ -10,6 +10,7 @@
 import gitpod.lib as util
 from gitpod import AsyncGitpod
 from gitpod.types.environment_initializer_param import Spec
+from gitpod.lib.cli_scm_auth import verify_context_url
 
 gpclient = AsyncGitpod()
 llmclient = Anthropic()
@@ -43,6 +44,9 @@ async def create_environment(args: dict[str, str], cleanup: util.Disposables) ->
         raise Exception("No environment class found. Please create one first.")
     env_class_id = env_class.id
     assert env_class_id is not None
+    
+    assert env_class.runner_id is not None
+    await verify_context_url(gpclient, args["context_url"], env_class.runner_id)
 
     environment = (await gpclient.environments.create(
         spec={
@@ -63,7 +67,7 @@ async def create_environment(args: dict[str, str], cleanup: util.Disposables) ->
     cleanup.add(lambda: asyncio.run(gpclient.environments.delete(environment_id=environment_id)))    
 
     print(f"\nCreated environment: {environment_id} - waiting for it to be ready...")
-    await util.wait_for_environment_ready(gpclient, environment_id)
+    await util.wait_for_environment_running(gpclient, environment_id)
     print(f"\nEnvironment is ready: {environment_id}")
     return environment_id
 

examples/fs_access.py

@@ -10,7 +10,7 @@
 from gitpod import AsyncGitpod
 from gitpod.types.environment_spec_param import EnvironmentSpecParam
 from gitpod.types.environment_initializer_param import Spec
-
+from gitpod.lib.cli_scm_auth import verify_context_url
 
 # Examples:
 # - ./examples/fs_access.py
@@ -46,6 +46,8 @@ async def main(cleanup: util.Disposables) -> None:
         }]
     }
     if context_url:
+        assert env_class.runner_id is not None
+        await verify_context_url(client, context_url, env_class.runner_id)
         spec["content"] = {
             "initializer": {"specs": [Spec(
                 context_url={

examples/run_command.py

@@ -3,12 +3,12 @@
 import sys
 import asyncio
 
+from gitpod.lib.cli_scm_auth import verify_context_url
 import gitpod.lib as util
 from gitpod import AsyncGitpod
 from gitpod.types.environment_spec_param import EnvironmentSpecParam
 from gitpod.types.environment_initializer_param import Spec
 
-
 # Examples:
 # - ./examples/run_command.py 'echo "Hello World!"'
 # - ./examples/run_command.py 'echo "Hello World!"' https://github.com/gitpod-io/empty
@@ -35,6 +35,8 @@ async def main(cleanup: util.Disposables) -> None:
         "machine": {"class": env_class_id},
     }
     if context_url:
+        assert env_class.runner_id is not None
+        await verify_context_url(client, context_url, env_class.runner_id)
         spec["content"] = {
             "initializer": {"specs": [Spec(
                 context_url={
@@ -51,7 +53,7 @@ async def main(cleanup: util.Disposables) -> None:
     cleanup.add(lambda: asyncio.run(client.environments.delete(environment_id=environment_id)))
 
     print("Waiting for environment to be ready")
-    await util.wait_for_environment_ready(client, environment_id)
+    await util.wait_for_environment_running(client, environment_id)
 
     print("Running command")
     lines = await util.run_command(client, environment_id, command)

examples/run_service.py

@@ -1,13 +1,14 @@
 #!/usr/bin/env python
 
+import os
 import sys
 import asyncio
 
 import gitpod.lib as util
 from gitpod import AsyncGitpod
 from gitpod.types.environment_spec_param import EnvironmentSpecParam
 from gitpod.types.environment_initializer_param import Spec
-
+from gitpod.lib.cli_scm_auth import verify_context_url
 
 # Examples:
 # - ./examples/run_service.py
@@ -16,8 +17,13 @@ async def main(cleanup: util.Disposables) -> None:
     client = AsyncGitpod()
 
     context_url = sys.argv[1] if len(sys.argv) > 1 else None
+    env_class_id = os.getenv("GITPOD_ENV_CLASS_ID")
 
-    env_class = await util.find_most_used_environment_class(client)
+    if not env_class_id:
+        env_class = await util.find_most_used_environment_class(client)
+    else:
+        env_class = await util.find_environment_class_by_id(client, env_class_id)
+    
     if not env_class:
         print("Error: No environment class found. Please create one first.")
         sys.exit(1)
@@ -36,15 +42,16 @@ async def main(cleanup: util.Disposables) -> None:
         }]
     }
     if context_url:
+        assert env_class.runner_id is not None
+        await verify_context_url(client, context_url, env_class.runner_id)
         spec["content"] = {
             "initializer": {"specs": [Spec(
-             context_url={
-                 "url": context_url
-             }
-        )]}
-    }
+                 context_url={
+                     "url": context_url
+                }
+            )]}
+        }
 
-    print("Creating environment")
     environment = (await client.environments.create(spec=spec)).environment
     assert environment is not None
     environment_id = environment.id
@@ -77,4 +84,4 @@ async def main(cleanup: util.Disposables) -> None:
 if __name__ == "__main__":
     disposables = util.Disposables()
     with disposables:
-        asyncio.run(main(disposables))
+        asyncio.run(main(disposables))

requirements-dev.lock

@@ -73,7 +73,6 @@ packaging==23.2
     # via nox
     # via pytest
 paramiko==3.5.1
-    # via gitpod-sdk
 platformdirs==3.11.0
     # via virtualenv
 pluggy==1.5.0

requirements.lock

@@ -14,16 +14,9 @@ annotated-types==0.6.0
 anyio==4.4.0
     # via gitpod-sdk
     # via httpx
-bcrypt==4.2.1
-    # via paramiko
 certifi==2023.7.22
     # via httpcore
     # via httpx
-cffi==1.17.1
-    # via cryptography
-    # via pynacl
-cryptography==44.0.1
-    # via paramiko
 distro==1.8.0
     # via gitpod-sdk
 exceptiongroup==1.2.2
@@ -37,16 +30,10 @@ httpx==0.28.1
 idna==3.4
     # via anyio
     # via httpx
-paramiko==3.5.1
-    # via gitpod-sdk
-pycparser==2.22
-    # via cffi
 pydantic==2.10.3
     # via gitpod-sdk
 pydantic-core==2.27.1
     # via pydantic
-pynacl==1.5.0
-    # via paramiko
 sniffio==1.3.0
     # via anyio
     # via gitpod-sdk

src/gitpod/lib/__init__.py

@@ -1,12 +1,15 @@
 from .automation import run_command, run_service
 from .disposables import Disposables
-from .environment import EnvironmentState, wait_for_environment_ready, find_most_used_environment_class
+from .environment import EnvironmentState, wait_for_environment_running, find_most_used_environment_class, find_environment_class_by_id
+from .runner import set_scm_pat
 
 __all__ = [
     'find_most_used_environment_class',
     'run_command',
     'run_service',
     'EnvironmentState',
     'Disposables',
-    'wait_for_environment_ready',
+    'wait_for_environment_running',
+    'find_environment_class_by_id',
+    'set_scm_pat',
 ] 

src/gitpod/lib/cli_scm_auth.py

@@ -0,0 +1,129 @@
+import sys
+from urllib.parse import urlparse
+
+import gitpod
+import gitpod.lib as util
+from gitpod import AsyncGitpod
+
+async def handle_pat_auth(client: AsyncGitpod, user_id: str, runner_id: str, host: str) -> None:
+    scm_info = {
+        "github.com": {
+            "create_url": f"https://{host}/settings/tokens/new?scopes=repo,workflow,read:user,user:email&description=gitpod",
+            "docs_url": "https://docs.github.com/en/authentication/keeping-your-account-and-data-secure/managing-your-personal-access-tokens",
+            "scopes": "repo, workflow, read:user, user:email"
+        },
+        "gitlab.com": {
+            "create_url": f"https://{host}/-/user_settings/personal_access_tokens?scopes=api,read_user,read_repository&name=gitpod",
+            "docs_url": "https://docs.gitlab.com/ee/user/profile/personal_access_tokens.html",
+            "scopes": "api, read_user, read_repository"
+        },
+        "dev.azure.com": {
+            "create_url": None,
+            "docs_url": "https://docs.microsoft.com/en-us/azure/devops/organizations/accounts/use-personal-access-tokens-to-authenticate",
+            "scopes": "Code (Read & Write)"
+        }
+    }.get(host, {
+        "create_url": None,
+        "docs_url": "https://www.gitpod.io/docs/flex/source-control",
+        "scopes": "repository access"
+    })
+
+    print("\nTo create a Personal Access Token:")
+    if scm_info["create_url"]:
+        print(f"1. Visit: {scm_info['create_url']}")
+    else:
+        print(f"1. Go to {host} > Settings > Developer Settings")
+    print(f"2. Create a new token with the following scopes: {scm_info['scopes']}")
+    print(f"3. Copy the generated token")
+    print(f"\nFor detailed instructions, visit: {scm_info['docs_url']}")
+
+    pat = input("\nEnter your Personal Access Token: ").strip()
+    if not pat:
+        return
+
+    await util.set_scm_pat(client, user_id, runner_id, host, pat)
+
+async def verify_context_url(client: AsyncGitpod, context_url: str, runner_id: str) -> None:
+    """Verify and handle authentication for a repository context URL."""
+    host = urlparse(context_url).hostname
+    if host is None:
+        print("Error: Invalid context URL")
+        sys.exit(1)
+
+    resp = await client.users.get_authenticated_user()
+    assert resp is not None
+    user = resp.user
+    assert user is not None
+    user_id = user.id
+    assert user_id is not None
+
+    # Main authentication loop
+    first_attempt = True
+    while True:
+        try:
+            # Try to access the context URL
+            await client.runners.parse_context_url(context_url=context_url, runner_id=runner_id)
+            print("\n✓ Authentication verified successfully")
+            return
+
+        except gitpod.APIError as e:
+            if e.code != "failed_precondition":
+                raise e
+            
+        # Show authentication required message only on first attempt
+        if first_attempt:
+            print(f"\nAuthentication required for {host}")
+            first_attempt = False
+
+        # Get authentication options for the host
+        auth_resp = await client.runners.check_authentication_for_host(
+            host=host,
+            runner_id=runner_id
+        )
+
+        # Handle re-authentication case
+        if auth_resp.authenticated and not first_attempt:
+            print("\nIt looks like you are already authenticated.")
+            if input("Would you like to re-authenticate? (y/n): ").lower().strip() != 'y':
+                print("\nAuthentication cancelled")
+                sys.exit(1)
+            else:
+                print("\nRetrying authentication...")
+                continue
+                
+        auth_methods: list[tuple[str, str]] = []
+        if auth_resp.authentication_url:
+            auth_methods.append(("OAuth", "Recommended"))
+        if auth_resp.pat_supported:
+            auth_methods.append(("Personal Access Token (PAT)", ""))
+
+        if not auth_methods:
+            print(f"\nError: No authentication method available for {host}")
+            sys.exit(1)
+
+        # Present authentication options
+        if len(auth_methods) > 1:
+            print("\nAvailable authentication methods:")
+            for i, (method, note) in enumerate(auth_methods, 1):
+                note_text = f" ({note})" if note else ""
+                print(f"{i}. {method}{note_text}")
+            
+            choice = input(f"\nChoose authentication method (1-{len(auth_methods)}): ").strip()
+            try:
+                method_index = int(choice) - 1
+                if not 0 <= method_index < len(auth_methods):
+                    raise ValueError()
+            except ValueError:
+                method_index = 0  # Default to OAuth if invalid input
+        else:
+            method_index = 0
+
+        # Handle chosen authentication method
+        chosen_method = auth_methods[method_index][0]
+        if chosen_method == "Personal Access Token (PAT)":
+            await handle_pat_auth(client, user_id, runner_id, host)
+        else:
+            print(f"\nPlease visit the following URL to authenticate:")
+            print(f"{auth_resp.authentication_url}")
+            print("\nWaiting for authentication to complete...")
+            input("Press Enter after completing authentication in your browser...") 

src/gitpod/lib/environment.py

@@ -135,7 +135,7 @@ def listener(env: Environment) -> None:
             await event.wait()
             if result is None:
                 raise RuntimeError("wait_until completed but result is None")
-            return result 
+            return result # type: ignore[unreachable] 
         finally:
             self._listeners.remove(listener)
 
@@ -144,11 +144,11 @@ def is_running(self, env: Environment) -> bool:
         if not env.status:
             return False
 
-        if env.status.phase in ["ENVIRONMENT_PHASE_STOPPING", "ENVIRONMENT_PHASE_STOPPED", 
+        if env.status.failure_message:
+            raise RuntimeError(f"Environment {env.id} failed: {'; '.join(env.status.failure_message)}")
+        elif env.status.phase in ["ENVIRONMENT_PHASE_STOPPING", "ENVIRONMENT_PHASE_STOPPED", 
                               "ENVIRONMENT_PHASE_DELETING", "ENVIRONMENT_PHASE_DELETED"]:
             raise RuntimeError(f"Environment {env.id} is in unexpected phase: {env.status.phase}")
-        elif env.status.failure_message:
-            raise RuntimeError(f"Environment {env.id} failed: {'; '.join(env.status.failure_message)}")
 
         return env.status.phase == "ENVIRONMENT_PHASE_RUNNING"
 
@@ -212,7 +212,7 @@ def check_key(env: Environment) -> Optional[bool]:
             return True if self.check_ssh_key_applied(env, key_id, key_value) else None
         await self.wait_until(check_key)
 
-async def wait_for_environment_ready(client: AsyncGitpod, environment_id: str) -> None:
+async def wait_for_environment_running(client: AsyncGitpod, environment_id: str) -> None:
     env = EnvironmentState(client, environment_id)
     try: 
         await env.wait_until_running()
@@ -240,6 +240,9 @@ async def find_most_used_environment_class(client: AsyncGitpod) -> Optional[Envi
     if not environment_class_id:
         return None
 
+    return await find_environment_class_by_id(client, environment_class_id)
+
+async def find_environment_class_by_id(client: AsyncGitpod, environment_class_id: str) -> Optional[EnvironmentClass]:
     classes_resp = await client.environments.classes.list(filter={"can_create_environments": True})
     while classes_resp:
         for cls in classes_resp.environment_classes: