4510044419

Build Type

github-actions

Committed by GitHub

Commit Message

Merge f7fc67340 into 2108f65d0

Pull Request Pull Request #3377: feat(cli): get image repository host on login

Run Details

10 of 12 new or added lines in 2 files covered. (83.33%)

25623 of 29826 relevant lines covered (85.91%)

4.57 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.33

/renku/core/login.py

# Copyright Swiss Data Science Center (SDSC). A partnership between
# École Polytechnique Fédérale de Lausanne (EPFL) and
# Eidgenössische Technische Hochschule Zürich (ETHZ).
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Logging in to a Renku deployment."""

import os
import time
import urllib
import webbrowser
from typing import TYPE_CHECKING, Optional, cast

from pydantic import validate_arguments

from renku.core import errors
from renku.core.config import get_value, remove_value, set_value
from renku.core.util import communication
from renku.core.util.git import (
    RENKU_BACKUP_PREFIX,
    create_backup_remote,
    get_remote,
    get_renku_repo_url,
    set_git_credential_helper,
)
from renku.core.util.urls import parse_authentication_endpoint
from renku.domain_model.enums import ConfigFilter
from renku.domain_model.project_context import project_context

if TYPE_CHECKING:
    from renku.infrastructure.repository import Repository


CONFIG_SECTION = "http"
KEYCLOAK_REALM = "Renku"
CLIENT_ID = "renku-cli"


@validate_arguments(config=dict(arbitrary_types_allowed=True))
def login(endpoint: Optional[str], git_login: bool, yes: bool):
    """Log into a Renku deployment."""
    from renku.core.util import requests

    try:
        repository = project_context.repository
    except ValueError:
        repository = None

    parsed_endpoint = _parse_endpoint(endpoint)

    remote_name, remote_url = None, None
    if git_login:
        if repository is None:
            communication.warn("Cannot log in to git outside a project.")
        else:
            remote = get_remote(repository)
            if remote:
                remote_name, remote_url = remote.name, remote.url

            if remote_name and remote_url:
                show_login_warning = get_value("renku", "show_login_warning")
                if not yes and (show_login_warning is None or show_login_warning.lower() == "true"):
                    message = (
                        "Remote URL will be changed. Do you want to continue "
                        "(to disable this warning, pass '--yes' or run 'renku config set show_login_warning False')?"
                    )
                    communication.confirm(message, abort=True, warning=True)
            else:
                raise errors.ParameterError("Cannot find a unique remote URL for project.")

    auth_server_url = _get_url(
        parsed_endpoint, path=f"auth/realms/{KEYCLOAK_REALM}/protocol/openid-connect/auth/device"
    )

    try:
        response = requests.post(auth_server_url, data={"client_id": CLIENT_ID})
    except errors.RequestError as e:
        raise errors.RequestError(f"Cannot connect to authorization server at {auth_server_url}.") from e

    requests.check_response(response=response)
    data = response.json()

    verification_uri = data.get("verification_uri")
    user_code = data.get("user_code")
    verification_uri_complete = f"{verification_uri}?user_code={user_code}"

    communication.echo(
        f"Please grant access to '{CLIENT_ID}' in your browser.\n"
        f"If a browser window does not open automatically, go to {verification_uri_complete}"
    )

    webbrowser.open_new_tab(verification_uri_complete)

    polling_interval = min(data.get("interval", 5), 5)
    token_url = _get_url(parsed_endpoint, path=f"auth/realms/{KEYCLOAK_REALM}/protocol/openid-connect/token")
    device_code = data.get("device_code")

    while True:
        time.sleep(polling_interval)

        response = requests.post(
            token_url,
            data={
                "device_code": device_code,
                "client_id": CLIENT_ID,
                "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
            },
        )
        status_code = response.status_code
        if status_code == 200:
            break
        elif status_code == 400:
            error = response.json().get("error")

            if error == "authorization_pending":
                continue
            elif error == "slow_down":
                polling_interval += 1
            elif error == "access_denied":
                raise errors.AuthenticationError("Access denied")
            elif error == "expired_token":
                raise errors.AuthenticationError("Session expired, try again")
            else:
                raise errors.AuthenticationError(f"Invalid error message from server: {response.json()}")
        else:
            raise errors.AuthenticationError(f"Invalid status code from server: {status_code} - {response.content}")

    image_regsitry_host_req_url = _get_url(parsed_endpoint, path="/api/config/imageRegistries")
    image_registry_host_res = requests.get(image_regsitry_host_req_url)
    if image_registry_host_res.status_code != 200:
        raise errors.ConfigurationError(
            f"Cannot get the image registry host from {image_regsitry_host_req_url}, "
            f"got unexpected status code {image_registry_host_res.status_code}"
        )
    image_registry_host = image_registry_host_res.json().get("default")
    if not image_registry_host:
        raise errors.ConfigurationError(
            f"Cannot get the image registry host from the response {image_registry_host_res.text}"
        )
    set_value(
        section=CONFIG_SECTION,
        key=f"{parsed_endpoint.netloc}_image_registry_host",
        value=image_registry_host,
        global_only=True,
    )

    access_token = response.json().get("access_token")
    _store_token(parsed_endpoint.netloc, access_token)

    if git_login and repository:
        set_git_credential_helper(repository=cast("Repository", repository), hostname=parsed_endpoint.netloc)
        backup_remote_name, backup_exists, remote = create_backup_remote(
            repository=repository, remote_name=remote_name, url=remote_url  # type:ignore
        )
        if backup_exists:
            communication.echo(f"Backup remote '{backup_remote_name}' already exists.")

        if not remote and not backup_exists:
            communication.error(f"Cannot create backup remote '{backup_remote_name}' for '{remote_url}'")
        else:
            _set_renku_url_for_remote(
                repository=cast("Repository", repository),
                remote_name=remote_name,  # type:ignore
                remote_url=remote_url,  # type:ignore
                hostname=parsed_endpoint.netloc,
            )


def _parse_endpoint(endpoint, use_remote=False):
    parsed_endpoint = parse_authentication_endpoint(endpoint=endpoint, use_remote=use_remote)
    if not parsed_endpoint:
        raise errors.ParameterError("Parameter 'endpoint' is missing.")

    return parsed_endpoint


def _get_url(parsed_endpoint, path, **query_args) -> str:
    query = urllib.parse.urlencode(query_args)
    return parsed_endpoint._replace(path=path, query=query).geturl()


def _store_token(netloc, access_token):
    set_value(section=CONFIG_SECTION, key=netloc, value=access_token, global_only=True)
    os.chmod(project_context.global_config_path, 0o600)


def _set_renku_url_for_remote(repository: "Repository", remote_name: str, remote_url: str, hostname: str):
    """Set renku repository URL for ``remote_name``.

    Args:
        repository("Repository"): Current ``Repository``.
        remote_name(str): Name of the remote.
        remote_url(str): Url of the remote.
        hostname(str): Hostname.
    Raises:
        errors.GitCommandError: If remote doesn't exist.
    """
    new_remote_url = get_renku_repo_url(remote_url, deployment_hostname=hostname)

    try:
        repository.remotes[remote_name].set_url(url=new_remote_url)
    except errors.GitCommandError as e:
        raise errors.GitError(f"Cannot change remote url for '{remote_name}' to '{new_remote_url}'") from e


def read_renku_token(endpoint: str, get_endpoint_from_remote=False) -> str:
    """Read renku token from renku config file.

    Args:
        endpoint(str):  Endpoint to get token for.
    Keywords:
        get_endpoint_from_remote: if no endpoint is specified, use the repository remote to infer one

    Returns:
        Token for endpoint.
    """
    try:
        parsed_endpoint = _parse_endpoint(endpoint, use_remote=get_endpoint_from_remote)
    except errors.ParameterError:
        return ""
    if not parsed_endpoint:
        return ""

    return _read_renku_token_for_hostname(hostname=parsed_endpoint.netloc)


def _read_renku_token_for_hostname(hostname):
    return get_value(section=CONFIG_SECTION, key=hostname, config_filter=ConfigFilter.GLOBAL_ONLY)


@validate_arguments(config=dict(arbitrary_types_allowed=True))
def logout(endpoint: Optional[str]):
    """Log out from one or all Renku deployments."""
    if endpoint:
        parsed_endpoint = parse_authentication_endpoint(endpoint=endpoint)
        key = parsed_endpoint.netloc
    else:
        key = "*"

    try:
        repository = project_context.repository
    except ValueError:
        repository = None

    remove_value(section=CONFIG_SECTION, key=key, global_only=True)
    _remove_git_credential_helper(repository=repository)
    _restore_git_remote(repository=repository)


def _remove_git_credential_helper(repository):
    if not repository:  # Outside a renku project
        return

    with repository.get_configuration(writable=True) as config:
        try:
            config.remove_value("credential", "helper")
        except errors.GitError:  # NOTE: If already logged out, an exception is raised
            pass


def _restore_git_remote(repository):
    if not repository:  # Outside a renku project
        return

    backup_remotes = [r.name for r in repository.remotes if r.name.startswith(RENKU_BACKUP_PREFIX)]
    for backup_remote in backup_remotes:
        remote_name = backup_remote.replace(f"{RENKU_BACKUP_PREFIX}-", "")
        remote_url = repository.remotes[backup_remote].url

        try:
            repository.remotes[remote_name].set_url(remote_url)
        except errors.GitCommandError:
            raise errors.GitError(f"Cannot restore remote url for '{remote_name}' to {remote_url}")

        try:
            repository.remotes.remove(backup_remote)
        except errors.GitCommandError:
            communication.error(f"Cannot delete backup remote '{backup_remote}'")


@validate_arguments(config=dict(arbitrary_types_allowed=True))
def credentials(command: str, hostname: Optional[str]):
    """Credentials helper for git."""
    if command != "get":
        return

    # NOTE: hostname comes from the credential helper we set up and has proper format
    hostname = hostname or ""
    token = _read_renku_token_for_hostname(hostname=hostname) or ""

    communication.echo("username=renku")
    communication.echo(f"password={token}")

1	# Copyright Swiss Data Science Center (SDSC). A partnership between
2	# École Polytechnique Fédérale de Lausanne (EPFL) and
3	# Eidgenössische Technische Hochschule Zürich (ETHZ).
4	#
5	# Licensed under the Apache License, Version 2.0 (the "License");
6	# you may not use this file except in compliance with the License.
7	# You may obtain a copy of the License at
8	#
9	# http://www.apache.org/licenses/LICENSE-2.0
10	#
11	# Unless required by applicable law or agreed to in writing, software
12	# distributed under the License is distributed on an "AS IS" BASIS,
13	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	# See the License for the specific language governing permissions and
15	# limitations under the License.
16	"""Logging in to a Renku deployment."""	10✔
17
18	import os	10✔
19	import time	10✔
20	import urllib	10✔
21	import webbrowser	10✔
22	from typing import TYPE_CHECKING, Optional, cast	10✔
23
24	from pydantic import validate_arguments	10✔
25
26	from renku.core import errors	10✔
27	from renku.core.config import get_value, remove_value, set_value	10✔
28	from renku.core.util import communication	10✔
29	from renku.core.util.git import (	10✔
30	RENKU_BACKUP_PREFIX,
31	create_backup_remote,
32	get_remote,
33	get_renku_repo_url,
34	set_git_credential_helper,
35	)
36	from renku.core.util.urls import parse_authentication_endpoint	10✔
37	from renku.domain_model.enums import ConfigFilter	10✔
38	from renku.domain_model.project_context import project_context	10✔
39
40	if TYPE_CHECKING:	10✔
41	from renku.infrastructure.repository import Repository	×
42
43
44	CONFIG_SECTION = "http"	10✔
45	KEYCLOAK_REALM = "Renku"	10✔
46	CLIENT_ID = "renku-cli"	10✔
47
48
49	@validate_arguments(config=dict(arbitrary_types_allowed=True))	10✔
50	def login(endpoint: Optional[str], git_login: bool, yes: bool):	10✔
51	"""Log into a Renku deployment."""
52	from renku.core.util import requests	2✔
53
54	try:	2✔
55	repository = project_context.repository	2✔
56	except ValueError:	2✔
57	repository = None	2✔
58
59	parsed_endpoint = _parse_endpoint(endpoint)	2✔
60
61	remote_name, remote_url = None, None	2✔
62	if git_login:	2✔
63	if repository is None:	2✔
64	communication.warn("Cannot log in to git outside a project.")	2✔
65	else:
66	remote = get_remote(repository)	2✔
67	if remote:	2✔
68	remote_name, remote_url = remote.name, remote.url	2✔
69
70	if remote_name and remote_url:	2✔
71	show_login_warning = get_value("renku", "show_login_warning")	2✔
72	if not yes and (show_login_warning is None or show_login_warning.lower() == "true"):	2✔
73	message = (	2✔
74	"Remote URL will be changed. Do you want to continue "
75	"(to disable this warning, pass '--yes' or run 'renku config set show_login_warning False')?"
76	)
77	communication.confirm(message, abort=True, warning=True)	2✔
78	else:
79	raise errors.ParameterError("Cannot find a unique remote URL for project.")	2✔
80
81	auth_server_url = _get_url(	2✔
82	parsed_endpoint, path=f"auth/realms/{KEYCLOAK_REALM}/protocol/openid-connect/auth/device"
83	)
84
85	try:	2✔
86	response = requests.post(auth_server_url, data={"client_id": CLIENT_ID})	2✔
87	except errors.RequestError as e:	×
88	raise errors.RequestError(f"Cannot connect to authorization server at {auth_server_url}.") from e	×
89
90	requests.check_response(response=response)	2✔
91	data = response.json()	2✔
92
93	verification_uri = data.get("verification_uri")	2✔
94	user_code = data.get("user_code")	2✔
95	verification_uri_complete = f"{verification_uri}?user_code={user_code}"	2✔
96
97	communication.echo(	2✔
98	f"Please grant access to '{CLIENT_ID}' in your browser.\n"
99	f"If a browser window does not open automatically, go to {verification_uri_complete}"
100	)
101
102	webbrowser.open_new_tab(verification_uri_complete)	2✔
103
104	polling_interval = min(data.get("interval", 5), 5)	2✔
105	token_url = _get_url(parsed_endpoint, path=f"auth/realms/{KEYCLOAK_REALM}/protocol/openid-connect/token")	2✔
106	device_code = data.get("device_code")	2✔
107
108	while True:	2✔
109	time.sleep(polling_interval)	2✔
110
111	response = requests.post(	2✔
112	token_url,
113	data={
114	"device_code": device_code,
115	"client_id": CLIENT_ID,
116	"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
117	},
118	)
119	status_code = response.status_code	2✔
120	if status_code == 200:	2✔
121	break	2✔
122	elif status_code == 400:	×
123	error = response.json().get("error")	×
124
125	if error == "authorization_pending":	×
126	continue	×
127	elif error == "slow_down":	×
128	polling_interval += 1	×
129	elif error == "access_denied":	×
130	raise errors.AuthenticationError("Access denied")	×
131	elif error == "expired_token":	×
132	raise errors.AuthenticationError("Session expired, try again")	×
133	else:
134	raise errors.AuthenticationError(f"Invalid error message from server: {response.json()}")	×
135	else:
136	raise errors.AuthenticationError(f"Invalid status code from server: {status_code} - {response.content}")	×
137
138	image_regsitry_host_req_url = _get_url(parsed_endpoint, path="/api/config/imageRegistries")	2✔
139	image_registry_host_res = requests.get(image_regsitry_host_req_url)	2✔
140	if image_registry_host_res.status_code != 200:	2✔
NEW 141	raise errors.ConfigurationError(	×
142	f"Cannot get the image registry host from {image_regsitry_host_req_url}, "
143	f"got unexpected status code {image_registry_host_res.status_code}"
144	)
145	image_registry_host = image_registry_host_res.json().get("default")	2✔
146	if not image_registry_host:	2✔
NEW 147	raise errors.ConfigurationError(	×
148	f"Cannot get the image registry host from the response {image_registry_host_res.text}"
149	)
150	set_value(	2✔
151	section=CONFIG_SECTION,
152	key=f"{parsed_endpoint.netloc}_image_registry_host",
153	value=image_registry_host,
154	global_only=True,
155	)
156
157	access_token = response.json().get("access_token")	2✔
158	_store_token(parsed_endpoint.netloc, access_token)	2✔
159
160	if git_login and repository:	2✔
161	set_git_credential_helper(repository=cast("Repository", repository), hostname=parsed_endpoint.netloc)	2✔
162	backup_remote_name, backup_exists, remote = create_backup_remote(	2✔
163	repository=repository, remote_name=remote_name, url=remote_url # type:ignore
164	)
165	if backup_exists:	2✔
166	communication.echo(f"Backup remote '{backup_remote_name}' already exists.")	2✔
167
168	if not remote and not backup_exists:	2✔
169	communication.error(f"Cannot create backup remote '{backup_remote_name}' for '{remote_url}'")	×
170	else:
171	_set_renku_url_for_remote(	2✔
172	repository=cast("Repository", repository),
173	remote_name=remote_name, # type:ignore
174	remote_url=remote_url, # type:ignore
175	hostname=parsed_endpoint.netloc,
176	)
177
178
179	def _parse_endpoint(endpoint, use_remote=False):	10✔
180	parsed_endpoint = parse_authentication_endpoint(endpoint=endpoint, use_remote=use_remote)	4✔
181	if not parsed_endpoint:	4✔
182	raise errors.ParameterError("Parameter 'endpoint' is missing.")	2✔
183
184	return parsed_endpoint	4✔
185
186
187	def _get_url(parsed_endpoint, path, **query_args) -> str:	10✔
188	query = urllib.parse.urlencode(query_args)	2✔
189	return parsed_endpoint._replace(path=path, query=query).geturl()	2✔
190
191
192	def _store_token(netloc, access_token):	10✔
193	set_value(section=CONFIG_SECTION, key=netloc, value=access_token, global_only=True)	2✔
194	os.chmod(project_context.global_config_path, 0o600)	2✔
195
196
197	def _set_renku_url_for_remote(repository: "Repository", remote_name: str, remote_url: str, hostname: str):	10✔
198	"""Set renku repository URL for ``remote_name``.
199
200	Args:
201	repository("Repository"): Current ``Repository``.
202	remote_name(str): Name of the remote.
203	remote_url(str): Url of the remote.
204	hostname(str): Hostname.
205	Raises:
206	errors.GitCommandError: If remote doesn't exist.
207	"""
208	new_remote_url = get_renku_repo_url(remote_url, deployment_hostname=hostname)	2✔
209
210	try:	2✔
211	repository.remotes[remote_name].set_url(url=new_remote_url)	2✔
212	except errors.GitCommandError as e:	×
213	raise errors.GitError(f"Cannot change remote url for '{remote_name}' to '{new_remote_url}'") from e	×
214
215
216	def read_renku_token(endpoint: str, get_endpoint_from_remote=False) -> str:	10✔
217	"""Read renku token from renku config file.
218
219	Args:
220	endpoint(str): Endpoint to get token for.
221	Keywords:
222	get_endpoint_from_remote: if no endpoint is specified, use the repository remote to infer one
223
224	Returns:
225	Token for endpoint.
226	"""
227	try:	4✔
228	parsed_endpoint = _parse_endpoint(endpoint, use_remote=get_endpoint_from_remote)	4✔
229	except errors.ParameterError:	×
230	return ""	×
231	if not parsed_endpoint:	4✔
232	return ""	×
233
234	return _read_renku_token_for_hostname(hostname=parsed_endpoint.netloc)	4✔
235
236
237	def _read_renku_token_for_hostname(hostname):	10✔
238	return get_value(section=CONFIG_SECTION, key=hostname, config_filter=ConfigFilter.GLOBAL_ONLY)	4✔
239
240
241	@validate_arguments(config=dict(arbitrary_types_allowed=True))	10✔
242	def logout(endpoint: Optional[str]):	10✔
243	"""Log out from one or all Renku deployments."""
244	if endpoint:	2✔
245	parsed_endpoint = parse_authentication_endpoint(endpoint=endpoint)	2✔
246	key = parsed_endpoint.netloc	2✔
247	else:
248	key = "*"	2✔
249
250	try:	2✔
251	repository = project_context.repository	2✔
252	except ValueError:	2✔
253	repository = None	2✔
254
255	remove_value(section=CONFIG_SECTION, key=key, global_only=True)	2✔
256	_remove_git_credential_helper(repository=repository)	2✔
257	_restore_git_remote(repository=repository)	2✔
258
259
260	def _remove_git_credential_helper(repository):	10✔
261	if not repository: # Outside a renku project	2✔
262	return	2✔
263
264	with repository.get_configuration(writable=True) as config:	2✔
265	try:	2✔
266	config.remove_value("credential", "helper")	2✔
267	except errors.GitError: # NOTE: If already logged out, an exception is raised	2✔
268	pass	2✔
269
270
271	def _restore_git_remote(repository):	10✔
272	if not repository: # Outside a renku project	2✔
273	return	2✔
274
275	backup_remotes = [r.name for r in repository.remotes if r.name.startswith(RENKU_BACKUP_PREFIX)]	2✔
276	for backup_remote in backup_remotes:	2✔
277	remote_name = backup_remote.replace(f"{RENKU_BACKUP_PREFIX}-", "")	2✔
278	remote_url = repository.remotes[backup_remote].url	2✔
279
280	try:	2✔
281	repository.remotes[remote_name].set_url(remote_url)	2✔
282	except errors.GitCommandError:	×
283	raise errors.GitError(f"Cannot restore remote url for '{remote_name}' to {remote_url}")	×
284
285	try:	2✔
286	repository.remotes.remove(backup_remote)	2✔
287	except errors.GitCommandError:	×
288	communication.error(f"Cannot delete backup remote '{backup_remote}'")	×
289
290
291	@validate_arguments(config=dict(arbitrary_types_allowed=True))	10✔
292	def credentials(command: str, hostname: Optional[str]):	10✔
293	"""Credentials helper for git."""
294	if command != "get":	2✔
295	return	2✔
296
297	# NOTE: hostname comes from the credential helper we set up and has proper format
298	hostname = hostname or ""	2✔
299	token = _read_renku_token_for_hostname(hostname=hostname) or ""	2✔
300
301	communication.echo("username=renku")	2✔
302	communication.echo(f"password={token}")	2✔

SwissDataScienceCenter / renku-python / 4510044419

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous