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

Author: akosyakov

examples/anthropic_tool_use.py

@@ -1,4 +1,5 @@
-#!/usr/bin/env python
+# export ANTHROPIC_API_KEY=...
+# python -m examples.anthropic_tool_use
 
 from __future__ import annotations
 
@@ -11,6 +12,8 @@
 from gitpod import AsyncGitpod
 from gitpod.types.environment_initializer_param import Spec
 
+from .scm_auth import verify_context_url  # type: ignore
+
 gpclient = AsyncGitpod()
 llmclient = Anthropic()
 
@@ -41,8 +44,7 @@ async def create_environment(args: dict[str, str], cleanup: util.Disposables) ->
     env_class = await util.find_most_used_environment_class(gpclient)
     if not env_class:
         raise Exception("No environment class found. Please create one first.")
-    env_class_id = env_class.id
-    assert env_class_id is not None
+    await verify_context_url(gpclient, args["context_url"], env_class.runner_id)
 
     environment = (await gpclient.environments.create(
         spec={
@@ -54,18 +56,15 @@ async def create_environment(args: dict[str, str], cleanup: util.Disposables) ->
                     }
                 )]},
             },
-            "machine": {"class": env_class_id},
+            "machine": {"class": env_class.id},
         }
     )).environment
-    assert environment is not None
-    environment_id = environment.id
-    assert environment_id is not None
-    cleanup.add(lambda: asyncio.run(gpclient.environments.delete(environment_id=environment_id)))    
+    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)
-    print(f"\nEnvironment is ready: {environment_id}")
-    return environment_id
+    print(f"\nCreated environment: {environment.id} - waiting for it to be ready...")
+    await util.wait_for_environment_running(gpclient, environment.id)
+    print(f"\nEnvironment is ready: {environment.id}")
+    return environment.id
 
 async def execute_command(args: dict[str, str]) -> str:
     lines_iter = await util.run_command(gpclient, args["environment_id"], args["command"])

examples/fs_access.py

@@ -1,5 +1,3 @@
-#!/usr/bin/env python
-
 import sys
 import asyncio
 from io import StringIO
@@ -11,10 +9,12 @@
 from gitpod.types.environment_spec_param import EnvironmentSpecParam
 from gitpod.types.environment_initializer_param import Spec
 
+from .scm_auth import verify_context_url  # type: ignore
+
 
 # Examples:
-# - ./examples/fs_access.py
-# - ./examples/fs_access.py https://github.com/gitpod-io/empty
+# - python -m examples.fs_access
+# - python -m examples.fs_access https://github.com/gitpod-io/empty
 async def main(cleanup: util.Disposables) -> None:
     client = AsyncGitpod()
 
@@ -25,8 +25,6 @@ async def main(cleanup: util.Disposables) -> None:
         print("Error: No environment class found. Please create one first.")
         sys.exit(1)
     print(f"Found environment class: {env_class.display_name} ({env_class.description})")
-    env_class_id = env_class.id
-    assert env_class_id is not None
 
     print("Generating SSH key pair")
     key = paramiko.RSAKey.generate(2048)
@@ -39,13 +37,14 @@ async def main(cleanup: util.Disposables) -> None:
     key_id = "fs-access-example"
     spec: EnvironmentSpecParam = {
         "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
-        "machine": {"class": env_class_id},
+        "machine": {"class": env_class.id},
         "ssh_public_keys": [{
             "id": key_id,
             "value": public_key
         }]
     }
     if context_url:
+        await verify_context_url(client, context_url, env_class.runner_id)
         spec["content"] = {
             "initializer": {"specs": [Spec(
                 context_url={
@@ -56,12 +55,9 @@ async def main(cleanup: util.Disposables) -> None:
 
     print("Creating environment")
     environment = (await client.environments.create(spec=spec)).environment
-    assert environment is not None
-    environment_id = environment.id
-    assert environment_id is not None
-    cleanup.add(lambda: asyncio.run(client.environments.delete(environment_id=environment_id)))
+    cleanup.add(lambda: asyncio.run(client.environments.delete(environment_id=environment.id)))
 
-    env = util.EnvironmentState(client, environment_id)
+    env = util.EnvironmentState(client, environment.id)
     cleanup.add(lambda: asyncio.run(env.close()))
 
     print("Waiting for environment to be running")

examples/run_command.py

@@ -1,5 +1,3 @@
-#!/usr/bin/env python
-
 import sys
 import asyncio
 
@@ -8,10 +6,12 @@
 from gitpod.types.environment_spec_param import EnvironmentSpecParam
 from gitpod.types.environment_initializer_param import Spec
 
+from .scm_auth import verify_context_url  # type: ignore
+
 
 # Examples:
-# - ./examples/run_command.py 'echo "Hello World!"'
-# - ./examples/run_command.py 'echo "Hello World!"' https://github.com/gitpod-io/empty
+# - python -m examples.run_command 'echo "Hello World!"'
+# - python -m examples.run_command 'echo "Hello World!"' https://github.com/gitpod-io/empty
 async def main(cleanup: util.Disposables) -> None:
     client = AsyncGitpod()
 
@@ -27,14 +27,13 @@ async def main(cleanup: util.Disposables) -> None:
         print("Error: No environment class found. Please create one first.")
         sys.exit(1)
     print(f"Found environment class: {env_class.display_name} ({env_class.description})")
-    env_class_id = env_class.id
-    assert env_class_id is not None
-
+    
     spec: EnvironmentSpecParam = {
         "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
-        "machine": {"class": env_class_id},
+        "machine": {"class": env_class.id},
     }
     if context_url:
+        await verify_context_url(client, context_url, env_class.runner_id)
         spec["content"] = {
             "initializer": {"specs": [Spec(
                 context_url={
@@ -45,16 +44,13 @@ async def main(cleanup: util.Disposables) -> None:
 
     print("Creating environment")
     environment = (await client.environments.create(spec=spec)).environment
-    assert environment is not None
-    environment_id = environment.id
-    assert environment_id is not None
-    cleanup.add(lambda: asyncio.run(client.environments.delete(environment_id=environment_id)))
+    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)
+    lines = await util.run_command(client, environment.id, command)
     async for line in lines:
         print(line)
 

examples/run_service.py

@@ -1,5 +1,3 @@
-#!/usr/bin/env python
-
 import sys
 import asyncio
 
@@ -8,56 +6,53 @@
 from gitpod.types.environment_spec_param import EnvironmentSpecParam
 from gitpod.types.environment_initializer_param import Spec
 
+from .scm_auth import verify_context_url  # type: ignore
+
 
 # Examples:
-# - ./examples/run_service.py
-# - ./examples/run_service.py https://github.com/gitpod-io/empty
+# - python -m examples.run_service
+# - python -m examples.run_service https://github.com/gitpod-io/empty
 async def main(cleanup: util.Disposables) -> None:
     client = AsyncGitpod()
 
     context_url = sys.argv[1] if len(sys.argv) > 1 else None
-
+    
     env_class = await util.find_most_used_environment_class(client)
     if not env_class:
         print("Error: No environment class found. Please create one first.")
         sys.exit(1)
     print(f"Found environment class: {env_class.display_name} ({env_class.description})")
-    env_class_id = env_class.id
-    assert env_class_id is not None
 
     port = 8888
     spec: EnvironmentSpecParam = {
         "desired_phase": "ENVIRONMENT_PHASE_RUNNING",
-        "machine": {"class": env_class_id},
+        "machine": {"class": env_class.id},
         "ports": [{
             "name": "Lama Service",
             "port": port,
             "admission": "ADMISSION_LEVEL_EVERYONE"
         }]
     }
     if context_url:
+        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
-    assert environment_id is not None
-    cleanup.add(lambda: asyncio.run(client.environments.delete(environment_id=environment_id)))
+    cleanup.add(lambda: asyncio.run(client.environments.delete(environment_id=environment.id)))
 
     print("Waiting for environment to be ready")
-    env = util.EnvironmentState(client, environment_id)
+    env = util.EnvironmentState(client, environment.id)
     cleanup.add(lambda: asyncio.run(env.close()))
     await env.wait_until_running()
 
     print("Starting Lama Service")
-    lines = await util.run_service(client, environment_id, {
+    lines = await util.run_service(client, environment.id, {
         "name":"Lama Service",
         "description":"Lama Service", 
         "reference":"lama-service"
@@ -77,4 +72,4 @@ async def main(cleanup: util.Disposables) -> None:
 if __name__ == "__main__":
     disposables = util.Disposables()
     with disposables:
-        asyncio.run(main(disposables))
+        asyncio.run(main(disposables))

examples/scm_auth.py

@@ -0,0 +1,133 @@
+import sys
+from urllib.parse import urlparse
+
+import gitpod
+import gitpod.lib as util
+from gitpod import AsyncGitpod
+from gitpod.types.runner_check_authentication_for_host_response import SupportsPat
+
+
+async def handle_pat_auth(client: AsyncGitpod, user_id: str, runner_id: str, host: str, supports_pat: SupportsPat) -> None:
+    print("\nTo create a Personal Access Token:")
+    create_url = supports_pat.create_url
+    
+    if create_url: 
+        print(f"1. Visit: {create_url}")
+    else:
+        print(f"1. Go to {host} > Settings > Developer Settings")
+    
+    if supports_pat.required_scopes and len(supports_pat.required_scopes) > 0:
+        required_scopes = ", ".join(supports_pat.required_scopes)
+        print(f"2. Create a new token with the following scopes: {required_scopes}")
+    else:
+        print(f"2. Create a new token")
+    
+    if supports_pat.example:
+        print(f"3. Copy the generated token (example: {supports_pat.example})")
+    else:
+        print(f"3. Copy the generated token")
+    
+    if supports_pat.docs_url:
+        print(f"\nFor detailed instructions, visit: {supports_pat.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.
+    
+    This function checks if the user has access to the specified repository and manages
+    the authentication process if needed. Git access to the repository is required for
+    environments to function properly.
+
+    As an alternative, you can authenticate once via the Gitpod dashboard:
+    1. Start a new environment 
+    2. Complete the browser-based authentication flow
+    
+    See https://www.gitpod.io/docs/flex/source-control for more details.
+    """
+    host = urlparse(context_url).hostname
+    if host is None:
+        print("Error: Invalid context URL")
+        sys.exit(1)
+
+    user = (await client.users.get_authenticated_user()).user
+    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.supports_oauth2:
+            auth_methods.append(("OAuth", "Recommended"))
+        if auth_resp.supports_pat:
+            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)":
+            assert auth_resp.supports_pat
+            await handle_pat_auth(client, user.id, runner_id, host, auth_resp.supports_pat)
+        else:
+            assert auth_resp.supports_oauth2
+            print(f"\nPlease visit the following URL to authenticate:")
+            print(f"{auth_resp.supports_oauth2.auth_url}")
+            if auth_resp.supports_oauth2.docs_url:
+                print(f"\nFor detailed instructions, visit: {auth_resp.supports_oauth2.docs_url}")
+            print("\nWaiting for authentication to complete...")
+            input("Press Enter after completing authentication in your browser...") 

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