1+ import sys
2+ from urllib .parse import urlparse
3+
4+ import gitpod
5+ import gitpod .lib as util
6+ from gitpod import AsyncGitpod
7+ from gitpod .types .runner_check_authentication_for_host_response import SupportsPat
8+
9+ async def handle_pat_auth (client : AsyncGitpod , user_id : str , runner_id : str , host : str , supports_pat : SupportsPat ) -> None :
10+ print ("\n To create a Personal Access Token:" )
11+ create_url = supports_pat .create_url
12+
13+ if create_url :
14+ print (f"1. Visit: { create_url } )
15+ else :
16+ print (f"1. Go to { host } )
17+
18+ if supports_pat .required_scopes and len (supports_pat .required_scopes ) > 0 :
19+ required_scopes = ", " .join (supports_pat .required_scopes )
20+ print (f"2. Create a new token with the following scopes: { required_scopes } )
21+ else :
22+ print (f"2. Create a new token" )
23+
24+ if supports_pat .example :
25+ print (f"3. Copy the generated token (example: { supports_pat .example } )
26+ else :
27+ print (f"3. Copy the generated token" )
28+
29+ if supports_pat .docs_url :
30+ print (f"\n For detailed instructions, visit: { supports_pat .docs_url } )
31+
32+ pat = input ("\n Enter your Personal Access Token: " ).strip ()
33+ if not pat :
34+ return
35+
36+ await util .set_scm_pat (client , user_id , runner_id , host , pat )
37+
38+ async def verify_context_url (client : AsyncGitpod , context_url : str , runner_id : str ) -> None :
39+ """Verify and handle authentication for a repository context URL.
40+
41+ This function checks if the user has access to the specified repository and manages
42+ the authentication process if needed. Git access to the repository is required for
43+ environments to function properly.
44+
45+ As an alternative, you can authenticate once via the Gitpod dashboard:
46+ 1. Start a new environment
47+ 2. Complete the browser-based authentication flow
48+
49+ See https://www.gitpod.io/docs/flex/source-control for more details.
50+ """
51+ host = urlparse (context_url ).hostname
52+ if host is None :
53+ print ("Error: Invalid context URL" )
54+ sys .exit (1 )
55+
56+ resp = await client .users .get_authenticated_user ()
57+ user = resp .user
58+ assert user is not None
59+ user_id = user .id
60+ assert user_id is not None
61+
62+ # Main authentication loop
63+ first_attempt = True
64+ while True :
65+ try :
66+ # Try to access the context URL
67+ await client .runners .parse_context_url (context_url = context_url , runner_id = runner_id )
68+ print ("\n ✓ Authentication verified successfully" )
69+ return
70+
71+ except gitpod .APIError as e :
72+ if e .code != "failed_precondition" :
73+ raise e
74+
75+ # Show authentication required message only on first attempt
76+ if first_attempt :
77+ print (f"\n Authentication required for { host } )
78+ first_attempt = False
79+
80+ # Get authentication options for the host
81+ auth_resp = await client .runners .check_authentication_for_host (
82+ host = host ,
83+ runner_id = runner_id
84+ )
85+
86+ # Handle re-authentication case
87+ if auth_resp .authenticated and not first_attempt :
88+ print ("\n It looks like you are already authenticated." )
89+ if input ("Would you like to re-authenticate? (y/n): " ).lower ().strip () != 'y' :
90+ print ("\n Authentication cancelled" )
91+ sys .exit (1 )
92+ else :
93+ print ("\n Retrying authentication..." )
94+ continue
95+
96+ auth_methods : list [tuple [str , str ]] = []
97+ if auth_resp .supports_oauth2 :
98+ auth_methods .append (("OAuth" , "Recommended" ))
99+ if auth_resp .supports_pat :
100+ auth_methods .append (("Personal Access Token (PAT)" , "" ))
101+
102+ if not auth_methods :
103+ print (f"\n Error: No authentication method available for { host } )
104+ sys .exit (1 )
105+
106+ # Present authentication options
107+ if len (auth_methods ) > 1 :
108+ print ("\n Available authentication methods:" )
109+ for i , (method , note ) in enumerate (auth_methods , 1 ):
110+ note_text = f" ({ note } if note else ""
111+ print (f"{ i } { method } { note_text } )
112+
113+ choice = input (f"\n Choose authentication method (1-{ len (auth_methods )} ).strip ()
114+ try :
115+ method_index = int (choice ) - 1
116+ if not 0 <= method_index < len (auth_methods ):
117+ raise ValueError ()
118+ except ValueError :
119+ method_index = 0 # Default to OAuth if invalid input
120+ else :
121+ method_index = 0
122+
123+ # Handle chosen authentication method
124+ chosen_method = auth_methods [method_index ][0 ]
125+ if chosen_method == "Personal Access Token (PAT)" :
126+ assert auth_resp .supports_pat
127+ await handle_pat_auth (client , user_id , runner_id , host , auth_resp .supports_pat )
128+ else :
129+ assert auth_resp .supports_oauth2
130+ print (f"\n Please visit the following URL to authenticate:" )
131+ print (f"{ auth_resp .supports_oauth2 .auth_url } )
132+ if auth_resp .supports_oauth2 .docs_url :
133+ print (f"\n For detailed instructions, visit: { auth_resp .supports_oauth2 .docs_url } )
134+ print ("\n Waiting for authentication to complete..." )
135+ input ("Press Enter after completing authentication in your browser..." )
0 commit comments