18123243513

Committed 30 Sep 2025 08:10AM UTC coverage: 86.702% (-0.01%) from 86.714%

Build # 18123243513

Build Type

Pull #1019

github

Committed by

web-flow

Commit Message

Merge e726c4543 into 0690bab65

Pull Request Pull Request #1019: feat: Attempt to support dockerhub private images

Run Details

70 of 101 new or added lines in 9 files covered. (69.31%)

106 existing lines in 6 files now uncovered.

22357 of 25786 relevant lines covered (86.7%)

1.52 hits per line

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

38.96

/components/renku_data_services/notebooks/image_check.py

"""Functions for checking access to images.

Access to docker images can fall into these cases:

1. The image is public and exists. It can be accessed anonymously
2. The image cannot be found. It may be absent or it requires credentials to access it

For the latter case, try to find out as much as possible:
- Look for credentials in the connected services
- If there are no connections defined for that user and registry, image is not accessible
- Try access it with the credentials, if it still fails the token could be invalid.
- Try to obtain the connected account that checks the token validity
"""

from __future__ import annotations

import base64
from dataclasses import dataclass, field
from typing import final

import httpx
from authlib.integrations.httpx_client import OAuthError

from renku_data_services.app_config import logging
from renku_data_services.base_models.core import APIUser
from renku_data_services.connected_services.db import ConnectedServicesRepository
from renku_data_services.connected_services.models import ImageProvider, OAuth2Client, OAuth2Connection
from renku_data_services.errors import errors
from renku_data_services.notebooks.api.classes.image import Image, ImageRepoDockerAPI

logger = logging.getLogger(__name__)


@final
@dataclass(frozen=True)
class CheckResult:
    """Result of checking access to an image."""

    accessible: bool
    response_code: int
    image_provider: ImageProvider | None = None
    token: str | None = field(default=None, repr=False)
    error: errors.UnauthorizedError | None = None

    def __str__(self) -> str:
        token = "***" if self.token else "None"
        error = "unauthorized" if self.error else "None"
        return (
            "CheckResult("
            f"accessible={self.accessible}/{self.response_code}, "
            f"provider={self.image_provider}, token={token}, error={error})"
        )

    @property
    def connection(self) -> OAuth2Connection | None:
        """Return the connection if present."""
        if self.image_provider is None:
            return None
        if self.image_provider.connected_user is None:
            return None
        return self.image_provider.connected_user.connection

    @property
    def client(self) -> OAuth2Client | None:
        """Return the OAuth2 client if present."""
        if self.image_provider is None:
            return None
        return self.image_provider.provider

    @property
    def user(self) -> APIUser | None:
        """Return the connected user if applicable."""
        if self.image_provider is None:
            return None
        if self.image_provider.connected_user is None:
            return None
        return self.image_provider.connected_user.user


async def check_image_path(
    image_path: str, user: APIUser, connected_services: ConnectedServicesRepository
) -> CheckResult:
    """Check access to the given image."""
    image = Image.from_path(image_path)
    return await check_image(image, user, connected_services)


async def check_image(image: Image, user: APIUser, connected_services: ConnectedServicesRepository) -> CheckResult:
    """Check access to the given image."""

    reg_api: ImageRepoDockerAPI = image.repo_api()  # public images
    unauth_error: errors.UnauthorizedError | None = None
    image_provider = await connected_services.get_provider_for_image(user, image)
    connected_user = image_provider.connected_user if image_provider is not None else None
    connection = connected_user.connection if connected_user is not None else None
    if image_provider is not None:
        try:
            reg_api = await connected_services.get_image_repo_client(image_provider)
        except errors.UnauthorizedError as e:
            logger.info(f"Error getting image repo client for image {image}: {e}")
            unauth_error = e
        except OAuthError as e:
            logger.info(f"Error getting image repo client for image {image}: {e}")
            unauth_error = errors.UnauthorizedError(message=f"OAuth error when getting repo client for image: {image}")
            unauth_error.__cause__ = e

    try:
        result = await reg_api.image_check(image)
    except httpx.HTTPError as e:
        logger.info(f"Error connecting {reg_api.scheme}://{reg_api.hostname}: {e}")
        result = 0

    if result != 200 and connection is not None:
        try:
            await connected_services.get_oauth2_connected_account(connection.id, user)
        except errors.UnauthorizedError as e:
            logger.info(f"Error getting connected account: {e}")
            unauth_error = e

    token = f"{reg_api.username}:{reg_api.oauth2_token}"
    token = base64.b64encode(token.encode()).decode()

    return CheckResult(
        accessible=result == 200,
        response_code=result,
        image_provider=image_provider,
        token=token,
        error=unauth_error,
    )

1	"""Functions for checking access to images.
2
3	Access to docker images can fall into these cases:
4
5	1. The image is public and exists. It can be accessed anonymously
6	2. The image cannot be found. It may be absent or it requires credentials to access it
7
8	For the latter case, try to find out as much as possible:
9	- Look for credentials in the connected services
10	- If there are no connections defined for that user and registry, image is not accessible
11	- Try access it with the credentials, if it still fails the token could be invalid.
12	- Try to obtain the connected account that checks the token validity
13	"""
14
15	from __future__ import annotations	2✔
16
17	import base64	2✔
18	from dataclasses import dataclass, field	2✔
19	from typing import final	2✔
20
21	import httpx	2✔
22	from authlib.integrations.httpx_client import OAuthError	2✔
23
24	from renku_data_services.app_config import logging	2✔
25	from renku_data_services.base_models.core import APIUser	2✔
26	from renku_data_services.connected_services.db import ConnectedServicesRepository	2✔
27	from renku_data_services.connected_services.models import ImageProvider, OAuth2Client, OAuth2Connection	2✔
28	from renku_data_services.errors import errors	2✔
29	from renku_data_services.notebooks.api.classes.image import Image, ImageRepoDockerAPI	2✔
30
31	logger = logging.getLogger(__name__)	2✔
32
33
34	@final	2✔
35	@dataclass(frozen=True)	2✔
36	class CheckResult:	2✔
37	"""Result of checking access to an image."""
38
39	accessible: bool	2✔
40	response_code: int	2✔
41	image_provider: ImageProvider \| None = None	2✔
42	token: str \| None = field(default=None, repr=False)	2✔
43	error: errors.UnauthorizedError \| None = None	2✔
44
45	def __str__(self) -> str:	2✔
46	token = "***" if self.token else "None"	×
47	error = "unauthorized" if self.error else "None"	×
UNCOV 48	return (	×
49	"CheckResult("
50	f"accessible={self.accessible}/{self.response_code}, "
51	f"provider={self.image_provider}, token={token}, error={error})"
52	)
53
54	@property	2✔
55	def connection(self) -> OAuth2Connection \| None:	2✔
56	"""Return the connection if present."""
57	if self.image_provider is None:	×
58	return None	×
59	if self.image_provider.connected_user is None:	×
60	return None	×
UNCOV 61	return self.image_provider.connected_user.connection	×
62
63	@property	2✔
64	def client(self) -> OAuth2Client \| None:	2✔
65	"""Return the OAuth2 client if present."""
66	if self.image_provider is None:	×
67	return None	×
UNCOV 68	return self.image_provider.provider	×
69
70	@property	2✔
71	def user(self) -> APIUser \| None:	2✔
72	"""Return the connected user if applicable."""
73	if self.image_provider is None:	×
74	return None	×
75	if self.image_provider.connected_user is None:	×
76	return None	×
UNCOV 77	return self.image_provider.connected_user.user	×
78
79
80	async def check_image_path(	2✔
81	image_path: str, user: APIUser, connected_services: ConnectedServicesRepository
82	) -> CheckResult:
83	"""Check access to the given image."""
84	image = Image.from_path(image_path)	×
UNCOV 85	return await check_image(image, user, connected_services)	×
86
87
88	async def check_image(image: Image, user: APIUser, connected_services: ConnectedServicesRepository) -> CheckResult:	2✔
89	"""Check access to the given image."""
90
91	reg_api: ImageRepoDockerAPI = image.repo_api() # public images	×
92	unauth_error: errors.UnauthorizedError \| None = None	×
93	image_provider = await connected_services.get_provider_for_image(user, image)	×
94	connected_user = image_provider.connected_user if image_provider is not None else None	×
95	connection = connected_user.connection if connected_user is not None else None	×
96	if image_provider is not None:	×
97	try:	×
98	reg_api = await connected_services.get_image_repo_client(image_provider)	×
99	except errors.UnauthorizedError as e:	×
100	logger.info(f"Error getting image repo client for image {image}: {e}")	×
UNCOV 101	unauth_error = e	×
102	except OAuthError as e:	×
103	logger.info(f"Error getting image repo client for image {image}: {e}")	×
104	unauth_error = errors.UnauthorizedError(message=f"OAuth error when getting repo client for image: {image}")	×
105	unauth_error.__cause__ = e	×
106
UNCOV 107	try:	×
108	result = await reg_api.image_check(image)	×
109	except httpx.HTTPError as e:	×
110	logger.info(f"Error connecting {reg_api.scheme}://{reg_api.hostname}: {e}")	×
111	result = 0	×
112
113	if result != 200 and connection is not None:	×
UNCOV 114	try:	×
NEW 115	await connected_services.get_oauth2_connected_account(connection.id, user)	×
NEW 116	except errors.UnauthorizedError as e:	×
NEW 117	logger.info(f"Error getting connected account: {e}")	×
UNCOV 118	unauth_error = e	×
119
UNCOV 120	token = f"{reg_api.username}:{reg_api.oauth2_token}"	×
UNCOV 121	token = base64.b64encode(token.encode()).decode()	×
122
UNCOV 123	return CheckResult(	×
124	accessible=result == 200,
125	response_code=result,
126	image_provider=image_provider,
127	token=token,
128	error=unauth_error,
129	)

SwissDataScienceCenter / renku-data-services / 18123243513

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