16369691288

Committed 18 Jul 2025 11:43AM UTC coverage: 87.186% (-0.04%) from 87.229%

Build # 16369691288

Build Type

Pull #929

github

Committed by

web-flow

Commit Message

Merge 03ce98685 into 8be58eb79

Pull Request Pull Request #929: exp: session api proxy

Run Details

8 of 21 new or added lines in 3 files covered. (38.1%)

13 existing lines in 5 files now uncovered.

21486 of 24644 relevant lines covered (87.19%)

1.53 hits per line

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

62.37

/components/renku_data_services/notebooks/api/amalthea_patches/init_containers.py

"""Patches for init containers."""

from __future__ import annotations

import json
import os
from dataclasses import asdict
from pathlib import Path, PurePosixPath
from typing import TYPE_CHECKING, Any

from kubernetes import client

from renku_data_services.base_models.core import AnonymousAPIUser, AuthenticatedAPIUser
from renku_data_services.notebooks.api.amalthea_patches.utils import get_certificates_volume_mounts
from renku_data_services.notebooks.api.classes.repository import GitProvider, Repository
from renku_data_services.notebooks.config import NotebooksConfig
from renku_data_services.notebooks.crs import EmptyDir, ExtraVolume, ExtraVolumeMount, InitContainer, SecretAsVolume
from renku_data_services.project import constants as project_constants
from renku_data_services.project.models import SessionSecret

if TYPE_CHECKING:
    # NOTE: If these are directly imported then you get circular imports.
    from renku_data_services.notebooks.api.classes.server import UserServer


async def git_clone_container_v2(
    user: AuthenticatedAPIUser | AnonymousAPIUser,
    config: NotebooksConfig,
    repositories: list[Repository],
    git_providers: list[GitProvider],
    workspace_mount_path: PurePosixPath,
    work_dir: PurePosixPath,
    lfs_auto_fetch: bool = False,
    uid: int = 1000,
    gid: int = 1000,
) -> dict[str, Any] | None:
    """Returns the specification for the container that clones the user's repositories for new operator."""
    amalthea_session_work_volume: str = "amalthea-volume"
    if not repositories:
        return None

    etc_cert_volume_mount = get_certificates_volume_mounts(
        config,
        custom_certs=False,
        etc_certs=True,
        read_only_etc_certs=True,
    )

    prefix = "GIT_CLONE_"
    env = [
        {
            "name": f"{prefix}MOUNT_PATH",
            "value": work_dir.as_posix(),
        },
        {
            "name": f"{prefix}LFS_AUTO_FETCH",
            "value": "1" if lfs_auto_fetch else "0",
        },
        {
            "name": f"{prefix}USER__USERNAME",
            "value": user.email,
        },
        {
            "name": f"{prefix}USER__RENKU_TOKEN",
            "value": str(user.access_token),
        },
        {"name": f"{prefix}IS_GIT_PROXY_ENABLED", "value": "0" if user.is_anonymous else "1"},
        {
            "name": f"{prefix}SENTRY__ENABLED",
            "value": str(config.sessions.git_clone.sentry.enabled).lower(),
        },
        {
            "name": f"{prefix}SENTRY__DSN",
            "value": config.sessions.git_clone.sentry.dsn,
        },
        {
            "name": f"{prefix}SENTRY__ENVIRONMENT",
            "value": config.sessions.git_clone.sentry.env,
        },
        {
            "name": f"{prefix}SENTRY__SAMPLE_RATE",
            "value": str(config.sessions.git_clone.sentry.sample_rate),
        },
        {"name": "SENTRY_RELEASE", "value": os.environ.get("SENTRY_RELEASE")},
        {
            "name": "REQUESTS_CA_BUNDLE",
            "value": str(Path(etc_cert_volume_mount[0]["mountPath"]) / "ca-certificates.crt"),
        },
        {
            "name": "SSL_CERT_FILE",
            "value": str(Path(etc_cert_volume_mount[0]["mountPath"]) / "ca-certificates.crt"),
        },
    ]
    if user.is_authenticated:
        if user.email:
            env.append(
                {"name": f"{prefix}USER__EMAIL", "value": user.email},
            )
        full_name = user.get_full_name()
        if full_name:
            env.append(
                {
                    "name": f"{prefix}USER__FULL_NAME",
                    "value": full_name,
                },
            )

    # Set up git repositories
    for idx, repo in enumerate(repositories):
        obj_env = f"{prefix}REPOSITORIES_{idx}_"
        env.append(
            {
                "name": obj_env,
                "value": json.dumps(asdict(repo)),
            }
        )

    # Set up git providers
    required_provider_ids: set[str] = {r.provider for r in repositories if r.provider}
    required_git_providers = [p for p in git_providers if p.id in required_provider_ids]
    for idx, provider in enumerate(required_git_providers):
        obj_env = f"{prefix}GIT_PROVIDERS_{idx}_"
        data = dict(id=provider.id, access_token_url=provider.access_token_url)
        env.append(
            {
                "name": obj_env,
                "value": json.dumps(data),
            }
        )

    return {
        "image": config.sessions.git_clone.image,
        "name": "git-clone",
        "resources": {
            "requests": {
                "cpu": "100m",
                "memory": "100Mi",
            }
        },
        "securityContext": {
            "allowPrivilegeEscalation": False,
            "runAsGroup": gid,
            "runAsUser": uid,
            "runAsNonRoot": True,
            "capabilities": {"drop": ["ALL"]},
        },
        "volumeMounts": [
            {
                "mountPath": workspace_mount_path.as_posix(),
                "name": amalthea_session_work_volume,
            },
            *etc_cert_volume_mount,
        ],
        "env": env,
    }


async def git_clone_container(server: UserServer) -> dict[str, Any] | None:
    """Returns the specification for the container that clones the user's repositories."""
    repositories = await server.repositories()
    if not repositories:
        return None

    etc_cert_volume_mount = get_certificates_volume_mounts(
        server.config,
        custom_certs=False,
        etc_certs=True,
        read_only_etc_certs=True,
    )

    prefix = "GIT_CLONE_"
    env = [
        {
            "name": f"{prefix}MOUNT_PATH",
            "value": server.workspace_mount_path.as_posix(),
        },
        {
            "name": f"{prefix}LFS_AUTO_FETCH",
            "value": "1" if server.server_options.lfs_auto_fetch else "0",
        },
        {
            "name": f"{prefix}USER__USERNAME",
            "value": server.user.email,
        },
        {
            "name": f"{prefix}USER__RENKU_TOKEN",
            "value": str(server.user.access_token),
        },
        {"name": f"{prefix}IS_GIT_PROXY_ENABLED", "value": "0" if server.user.is_anonymous else "1"},
        {
            "name": f"{prefix}SENTRY__ENABLED",
            "value": str(server.config.sessions.git_clone.sentry.enabled).lower(),
        },
        {
            "name": f"{prefix}SENTRY__DSN",
            "value": server.config.sessions.git_clone.sentry.dsn,
        },
        {
            "name": f"{prefix}SENTRY__ENVIRONMENT",
            "value": server.config.sessions.git_clone.sentry.env,
        },
        {
            "name": f"{prefix}SENTRY__SAMPLE_RATE",
            "value": str(server.config.sessions.git_clone.sentry.sample_rate),
        },
        {"name": "SENTRY_RELEASE", "value": os.environ.get("SENTRY_RELEASE")},
        {
            "name": "REQUESTS_CA_BUNDLE",
            "value": str(Path(etc_cert_volume_mount[0]["mountPath"]) / "ca-certificates.crt"),
        },
        {
            "name": "SSL_CERT_FILE",
            "value": str(Path(etc_cert_volume_mount[0]["mountPath"]) / "ca-certificates.crt"),
        },
    ]
    if server.user.is_authenticated:
        if server.user.email:
            env.append(
                {"name": f"{prefix}USER__EMAIL", "value": server.user.email},
            )
        full_name = server.user.get_full_name()
        if full_name:
            env.append(
                {
                    "name": f"{prefix}USER__FULL_NAME",
                    "value": full_name,
                },
            )

    # Set up git repositories
    for idx, repo in enumerate(repositories):
        obj_env = f"{prefix}REPOSITORIES_{idx}_"
        env.append(
            {
                "name": obj_env,
                "value": json.dumps(asdict(repo)),
            }
        )

    # Set up git providers
    required_git_providers = await server.required_git_providers()
    for idx, provider in enumerate(required_git_providers):
        obj_env = f"{prefix}GIT_PROVIDERS_{idx}_"
        data = dict(id=provider.id, access_token_url=provider.access_token_url)
        env.append(
            {
                "name": obj_env,
                "value": json.dumps(data),
            }
        )

    return {
        "image": server.config.sessions.git_clone.image,
        "name": "git-clone",
        "resources": {
            "requests": {
                "cpu": "100m",
                "memory": "100Mi",
            }
        },
        "securityContext": {
            "allowPrivilegeEscalation": False,
            "runAsGroup": 100,
            "runAsUser": 1000,
            "runAsNonRoot": True,
        },
        "volumeMounts": [
            {
                "mountPath": server.workspace_mount_path.as_posix(),
                "name": "workspace",
            },
            *etc_cert_volume_mount,
        ],
        "env": env,
    }


async def git_clone(server: UserServer) -> list[dict[str, Any]]:
    """The patch for the init container that clones the git repository."""
    container = await git_clone_container(server)
    if not container:
        return []
    return [
        {
            "type": "application/json-patch+json",
            "patch": [
                {
                    "op": "add",
                    "path": "/statefulset/spec/template/spec/initContainers/-",
                    "value": container,
                },
            ],
        }
    ]


def certificates_container(config: NotebooksConfig) -> tuple[client.V1Container, list[client.V1Volume]]:
    """The specification for the container that setups self signed CAs."""
    init_container = client.V1Container(
        name="init-certificates",
        image=config.sessions.ca_certs.image,
        volume_mounts=get_certificates_volume_mounts(
            config,
            etc_certs=True,
            custom_certs=True,
            read_only_etc_certs=False,
        ),
        security_context=client.V1SecurityContext(
            allow_privilege_escalation=False,
            run_as_group=1000,
            run_as_user=1000,
            run_as_non_root=True,
            capabilities=client.V1Capabilities(drop=["ALL"]),
        ),
        resources={
            "requests": {
                "cpu": "50m",
                "memory": "50Mi",
            }
        },
    )
    volume_etc_certs = client.V1Volume(name="etc-ssl-certs", empty_dir=client.V1EmptyDirVolumeSource(medium="Memory"))
    volume_custom_certs = client.V1Volume(
        name="custom-ca-certs",
        projected=client.V1ProjectedVolumeSource(
            default_mode=440,
            sources=[
                {"secret": {"name": secret.get("secret")}}
                for secret in config.sessions.ca_certs.secrets
                if isinstance(secret, dict) and secret.get("secret") is not None
            ],
        ),
    )
    return (init_container, [volume_etc_certs, volume_custom_certs])


def certificates(config: NotebooksConfig) -> list[dict[str, Any]]:
    """Add a container that initializes custom certificate authorities for a session."""
    container, vols = certificates_container(config)
    api_client = client.ApiClient()
    patches = [
        {
            "type": "application/json-patch+json",
            "patch": [
                {
                    "op": "add",
                    "path": "/statefulset/spec/template/spec/initContainers/-",
                    "value": api_client.sanitize_for_serialization(container),
                },
            ],
        },
    ]
    for vol in vols:
        patches.append(
            {
                "type": "application/json-patch+json",
                "patch": [
                    {
                        "op": "add",
                        "path": "/statefulset/spec/template/spec/volumes/-",
                        "value": api_client.sanitize_for_serialization(vol),
                    },
                ],
            },
        )
    return patches


def download_image_container(server: UserServer) -> client.V1Container:
    """Adds a container that does not do anything but simply downloads the session image at startup."""
    container = client.V1Container(
        name="download-image",
        image=server.image,
        command=["sh", "-c"],
        args=["exit", "0"],
        resources={
            "requests": {
                "cpu": "50m",
                "memory": "50Mi",
            }
        },
    )
    return container


def download_image(server: UserServer) -> list[dict[str, Any]]:
    """Adds a container that does not do anything but simply downloads the session image at startup."""
    container = download_image_container(server)
    api_client = client.ApiClient()
    return [
        {
            "type": "application/json-patch+json",
            "patch": [
                {
                    "op": "add",
                    "path": "/statefulset/spec/template/spec/initContainers/-",
                    "value": api_client.sanitize_for_serialization(container),
                },
            ],
        },
    ]


def user_secrets_container(
    user: AuthenticatedAPIUser | AnonymousAPIUser,
    config: NotebooksConfig,
    secrets_mount_directory: str,
    k8s_secret_name: str,
    session_secrets: list[SessionSecret],
) -> tuple[InitContainer, list[ExtraVolume], list[ExtraVolumeMount]] | None:
    """The init container which decrypts user secrets to be mounted in the session."""
    if not session_secrets or user.is_anonymous:
        return None

    volume_k8s_secrets = ExtraVolume(
        name=f"{k8s_secret_name}-volume",
        secret=SecretAsVolume(
            secretName=k8s_secret_name,
        ),
    )
    volume_decrypted_secrets = ExtraVolume(name="user-secrets-volume", emptyDir=EmptyDir(medium="Memory"))

    decrypted_volume_mount = ExtraVolumeMount(
        name="user-secrets-volume",
        mountPath=secrets_mount_directory or project_constants.DEFAULT_SESSION_SECRETS_MOUNT_DIR.as_posix(),
        readOnly=True,
    )

    init_container = InitContainer.model_validate(
        dict(
            name="init-user-secrets",
            image=config.user_secrets.image,
            env=[
                dict(name="DATA_SERVICE_URL", value=config.data_service_url),
                dict(name="RENKU_ACCESS_TOKEN", value=user.access_token or ""),
                dict(name="ENCRYPTED_SECRETS_MOUNT_PATH", value="/encrypted"),
                dict(name="DECRYPTED_SECRETS_MOUNT_PATH", value="/decrypted"),
            ],
            volumeMounts=[
                dict(
                    name=f"{k8s_secret_name}-volume",
                    mountPath="/encrypted",
                    readOnly=True,
                ),
                dict(
                    name="user-secrets-volume",
                    mountPath="/decrypted",
                    readOnly=False,
                ),
            ],
            resources={
                "requests": {
                    "cpu": "50m",
                    "memory": "50Mi",
                }
            },
        )
    )

    return (
        init_container,
        [volume_k8s_secrets, volume_decrypted_secrets],
        [decrypted_volume_mount],
    )

1	"""Patches for init containers."""
2
3	from __future__ import annotations	2✔
4
5	import json	2✔
6	import os	2✔
7	from dataclasses import asdict	2✔
8	from pathlib import Path, PurePosixPath	2✔
9	from typing import TYPE_CHECKING, Any	2✔
10
11	from kubernetes import client	2✔
12
13	from renku_data_services.base_models.core import AnonymousAPIUser, AuthenticatedAPIUser	2✔
14	from renku_data_services.notebooks.api.amalthea_patches.utils import get_certificates_volume_mounts	2✔
15	from renku_data_services.notebooks.api.classes.repository import GitProvider, Repository	2✔
16	from renku_data_services.notebooks.config import NotebooksConfig	2✔
17	from renku_data_services.notebooks.crs import EmptyDir, ExtraVolume, ExtraVolumeMount, InitContainer, SecretAsVolume	2✔
18	from renku_data_services.project import constants as project_constants	2✔
19	from renku_data_services.project.models import SessionSecret	2✔
20
21	if TYPE_CHECKING:	2✔
22	# NOTE: If these are directly imported then you get circular imports.
23	from renku_data_services.notebooks.api.classes.server import UserServer	×
24
25
26	async def git_clone_container_v2(	2✔
27	user: AuthenticatedAPIUser \| AnonymousAPIUser,
28	config: NotebooksConfig,
29	repositories: list[Repository],
30	git_providers: list[GitProvider],
31	workspace_mount_path: PurePosixPath,
32	work_dir: PurePosixPath,
33	lfs_auto_fetch: bool = False,
34	uid: int = 1000,
35	gid: int = 1000,
36	) -> dict[str, Any] \| None:
37	"""Returns the specification for the container that clones the user's repositories for new operator."""
38	amalthea_session_work_volume: str = "amalthea-volume"	×
39	if not repositories:	×
40	return None	×
41
42	etc_cert_volume_mount = get_certificates_volume_mounts(	×
43	config,
44	custom_certs=False,
45	etc_certs=True,
46	read_only_etc_certs=True,
47	)
48
49	prefix = "GIT_CLONE_"	×
50	env = [	×
51	{
52	"name": f"{prefix}MOUNT_PATH",
53	"value": work_dir.as_posix(),
54	},
55	{
56	"name": f"{prefix}LFS_AUTO_FETCH",
57	"value": "1" if lfs_auto_fetch else "0",
58	},
59	{
60	"name": f"{prefix}USER__USERNAME",
61	"value": user.email,
62	},
63	{
64	"name": f"{prefix}USER__RENKU_TOKEN",
65	"value": str(user.access_token),
66	},
67	{"name": f"{prefix}IS_GIT_PROXY_ENABLED", "value": "0" if user.is_anonymous else "1"},
68	{
69	"name": f"{prefix}SENTRY__ENABLED",
70	"value": str(config.sessions.git_clone.sentry.enabled).lower(),
71	},
72	{
73	"name": f"{prefix}SENTRY__DSN",
74	"value": config.sessions.git_clone.sentry.dsn,
75	},
76	{
77	"name": f"{prefix}SENTRY__ENVIRONMENT",
78	"value": config.sessions.git_clone.sentry.env,
79	},
80	{
81	"name": f"{prefix}SENTRY__SAMPLE_RATE",
82	"value": str(config.sessions.git_clone.sentry.sample_rate),
83	},
84	{"name": "SENTRY_RELEASE", "value": os.environ.get("SENTRY_RELEASE")},
85	{
86	"name": "REQUESTS_CA_BUNDLE",
87	"value": str(Path(etc_cert_volume_mount[0]["mountPath"]) / "ca-certificates.crt"),
88	},
89	{
90	"name": "SSL_CERT_FILE",
91	"value": str(Path(etc_cert_volume_mount[0]["mountPath"]) / "ca-certificates.crt"),
92	},
93	]
94	if user.is_authenticated:	×
95	if user.email:	×
96	env.append(	×
97	{"name": f"{prefix}USER__EMAIL", "value": user.email},
98	)
99	full_name = user.get_full_name()	×
100	if full_name:	×
101	env.append(	×
102	{
103	"name": f"{prefix}USER__FULL_NAME",
104	"value": full_name,
105	},
106	)
107
108	# Set up git repositories
109	for idx, repo in enumerate(repositories):	×
110	obj_env = f"{prefix}REPOSITORIES_{idx}_"	×
111	env.append(	×
112	{
113	"name": obj_env,
114	"value": json.dumps(asdict(repo)),
115	}
116	)
117
118	# Set up git providers
119	required_provider_ids: set[str] = {r.provider for r in repositories if r.provider}	×
120	required_git_providers = [p for p in git_providers if p.id in required_provider_ids]	×
121	for idx, provider in enumerate(required_git_providers):	×
122	obj_env = f"{prefix}GIT_PROVIDERS_{idx}_"	×
123	data = dict(id=provider.id, access_token_url=provider.access_token_url)	×
124	env.append(	×
125	{
126	"name": obj_env,
127	"value": json.dumps(data),
128	}
129	)
130
131	return {	×
132	"image": config.sessions.git_clone.image,
133	"name": "git-clone",
134	"resources": {
135	"requests": {
136	"cpu": "100m",
137	"memory": "100Mi",
138	}
139	},
140	"securityContext": {
141	"allowPrivilegeEscalation": False,
142	"runAsGroup": gid,
143	"runAsUser": uid,
144	"runAsNonRoot": True,
145	"capabilities": {"drop": ["ALL"]},
146	},
147	"volumeMounts": [
148	{
149	"mountPath": workspace_mount_path.as_posix(),
150	"name": amalthea_session_work_volume,
151	},
152	*etc_cert_volume_mount,
153	],
154	"env": env,
155	}
156
157
158	async def git_clone_container(server: UserServer) -> dict[str, Any] \| None:	2✔
159	"""Returns the specification for the container that clones the user's repositories."""
160	repositories = await server.repositories()	1✔
161	if not repositories:	1✔
UNCOV 162	return None	×
163
164	etc_cert_volume_mount = get_certificates_volume_mounts(	1✔
165	server.config,
166	custom_certs=False,
167	etc_certs=True,
168	read_only_etc_certs=True,
169	)
170
171	prefix = "GIT_CLONE_"	1✔
172	env = [	1✔
173	{
174	"name": f"{prefix}MOUNT_PATH",
175	"value": server.workspace_mount_path.as_posix(),
176	},
177	{
178	"name": f"{prefix}LFS_AUTO_FETCH",
179	"value": "1" if server.server_options.lfs_auto_fetch else "0",
180	},
181	{
182	"name": f"{prefix}USER__USERNAME",
183	"value": server.user.email,
184	},
185	{
186	"name": f"{prefix}USER__RENKU_TOKEN",
187	"value": str(server.user.access_token),
188	},
189	{"name": f"{prefix}IS_GIT_PROXY_ENABLED", "value": "0" if server.user.is_anonymous else "1"},
190	{
191	"name": f"{prefix}SENTRY__ENABLED",
192	"value": str(server.config.sessions.git_clone.sentry.enabled).lower(),
193	},
194	{
195	"name": f"{prefix}SENTRY__DSN",
196	"value": server.config.sessions.git_clone.sentry.dsn,
197	},
198	{
199	"name": f"{prefix}SENTRY__ENVIRONMENT",
200	"value": server.config.sessions.git_clone.sentry.env,
201	},
202	{
203	"name": f"{prefix}SENTRY__SAMPLE_RATE",
204	"value": str(server.config.sessions.git_clone.sentry.sample_rate),
205	},
206	{"name": "SENTRY_RELEASE", "value": os.environ.get("SENTRY_RELEASE")},
207	{
208	"name": "REQUESTS_CA_BUNDLE",
209	"value": str(Path(etc_cert_volume_mount[0]["mountPath"]) / "ca-certificates.crt"),
210	},
211	{
212	"name": "SSL_CERT_FILE",
213	"value": str(Path(etc_cert_volume_mount[0]["mountPath"]) / "ca-certificates.crt"),
214	},
215	]
216	if server.user.is_authenticated:	1✔
217	if server.user.email:	1✔
218	env.append(	1✔
219	{"name": f"{prefix}USER__EMAIL", "value": server.user.email},
220	)
221	full_name = server.user.get_full_name()	1✔
222	if full_name:	1✔
223	env.append(	1✔
224	{
225	"name": f"{prefix}USER__FULL_NAME",
226	"value": full_name,
227	},
228	)
229
230	# Set up git repositories
231	for idx, repo in enumerate(repositories):	1✔
232	obj_env = f"{prefix}REPOSITORIES_{idx}_"	1✔
233	env.append(	1✔
234	{
235	"name": obj_env,
236	"value": json.dumps(asdict(repo)),
237	}
238	)
239
240	# Set up git providers
241	required_git_providers = await server.required_git_providers()	1✔
242	for idx, provider in enumerate(required_git_providers):	1✔
243	obj_env = f"{prefix}GIT_PROVIDERS_{idx}_"	×
244	data = dict(id=provider.id, access_token_url=provider.access_token_url)	×
UNCOV 245	env.append(	×
246	{
247	"name": obj_env,
248	"value": json.dumps(data),
249	}
250	)
251
252	return {	1✔
253	"image": server.config.sessions.git_clone.image,
254	"name": "git-clone",
255	"resources": {
256	"requests": {
257	"cpu": "100m",
258	"memory": "100Mi",
259	}
260	},
261	"securityContext": {
262	"allowPrivilegeEscalation": False,
263	"runAsGroup": 100,
264	"runAsUser": 1000,
265	"runAsNonRoot": True,
266	},
267	"volumeMounts": [
268	{
269	"mountPath": server.workspace_mount_path.as_posix(),
270	"name": "workspace",
271	},
272	*etc_cert_volume_mount,
273	],
274	"env": env,
275	}
276
277
278	async def git_clone(server: UserServer) -> list[dict[str, Any]]:	2✔
279	"""The patch for the init container that clones the git repository."""
280	container = await git_clone_container(server)	1✔
281	if not container:	1✔
UNCOV 282	return []	×
283	return [	1✔
284	{
285	"type": "application/json-patch+json",
286	"patch": [
287	{
288	"op": "add",
289	"path": "/statefulset/spec/template/spec/initContainers/-",
290	"value": container,
291	},
292	],
293	}
294	]
295
296
297	def certificates_container(config: NotebooksConfig) -> tuple[client.V1Container, list[client.V1Volume]]:	2✔
298	"""The specification for the container that setups self signed CAs."""
299	init_container = client.V1Container(	1✔
300	name="init-certificates",
301	image=config.sessions.ca_certs.image,
302	volume_mounts=get_certificates_volume_mounts(
303	config,
304	etc_certs=True,
305	custom_certs=True,
306	read_only_etc_certs=False,
307	),
308	security_context=client.V1SecurityContext(
309	allow_privilege_escalation=False,
310	run_as_group=1000,
311	run_as_user=1000,
312	run_as_non_root=True,
313	capabilities=client.V1Capabilities(drop=["ALL"]),
314	),
315	resources={
316	"requests": {
317	"cpu": "50m",
318	"memory": "50Mi",
319	}
320	},
321	)
322	volume_etc_certs = client.V1Volume(name="etc-ssl-certs", empty_dir=client.V1EmptyDirVolumeSource(medium="Memory"))	1✔
323	volume_custom_certs = client.V1Volume(	1✔
324	name="custom-ca-certs",
325	projected=client.V1ProjectedVolumeSource(
326	default_mode=440,
327	sources=[
328	{"secret": {"name": secret.get("secret")}}
329	for secret in config.sessions.ca_certs.secrets
330	if isinstance(secret, dict) and secret.get("secret") is not None
331	],
332	),
333	)
334	return (init_container, [volume_etc_certs, volume_custom_certs])	1✔
335
336
337	def certificates(config: NotebooksConfig) -> list[dict[str, Any]]:	2✔
338	"""Add a container that initializes custom certificate authorities for a session."""
339	container, vols = certificates_container(config)	1✔
340	api_client = client.ApiClient()	1✔
341	patches = [	1✔
342	{
343	"type": "application/json-patch+json",
344	"patch": [
345	{
346	"op": "add",
347	"path": "/statefulset/spec/template/spec/initContainers/-",
348	"value": api_client.sanitize_for_serialization(container),
349	},
350	],
351	},
352	]
353	for vol in vols:	1✔
354	patches.append(	1✔
355	{
356	"type": "application/json-patch+json",
357	"patch": [
358	{
359	"op": "add",
360	"path": "/statefulset/spec/template/spec/volumes/-",
361	"value": api_client.sanitize_for_serialization(vol),
362	},
363	],
364	},
365	)
366	return patches	1✔
367
368
369	def download_image_container(server: UserServer) -> client.V1Container:	2✔
370	"""Adds a container that does not do anything but simply downloads the session image at startup."""
371	container = client.V1Container(	1✔
372	name="download-image",
373	image=server.image,
374	command=["sh", "-c"],
375	args=["exit", "0"],
376	resources={
377	"requests": {
378	"cpu": "50m",
379	"memory": "50Mi",
380	}
381	},
382	)
383	return container	1✔
384
385
386	def download_image(server: UserServer) -> list[dict[str, Any]]:	2✔
387	"""Adds a container that does not do anything but simply downloads the session image at startup."""
388	container = download_image_container(server)	1✔
389	api_client = client.ApiClient()	1✔
390	return [	1✔
391	{
392	"type": "application/json-patch+json",
393	"patch": [
394	{
395	"op": "add",
396	"path": "/statefulset/spec/template/spec/initContainers/-",
397	"value": api_client.sanitize_for_serialization(container),
398	},
399	],
400	},
401	]
402
403
404	def user_secrets_container(	2✔
405	user: AuthenticatedAPIUser \| AnonymousAPIUser,
406	config: NotebooksConfig,
407	secrets_mount_directory: str,
408	k8s_secret_name: str,
409	session_secrets: list[SessionSecret],
410	) -> tuple[InitContainer, list[ExtraVolume], list[ExtraVolumeMount]] \| None:
411	"""The init container which decrypts user secrets to be mounted in the session."""
UNCOV 412	if not session_secrets or user.is_anonymous:	×
413	return None	×
414
415	volume_k8s_secrets = ExtraVolume(	×
416	name=f"{k8s_secret_name}-volume",
417	secret=SecretAsVolume(
418	secretName=k8s_secret_name,
419	),
420	)
421	volume_decrypted_secrets = ExtraVolume(name="user-secrets-volume", emptyDir=EmptyDir(medium="Memory"))	×
422
UNCOV 423	decrypted_volume_mount = ExtraVolumeMount(	×
424	name="user-secrets-volume",
425	mountPath=secrets_mount_directory or project_constants.DEFAULT_SESSION_SECRETS_MOUNT_DIR.as_posix(),
426	readOnly=True,
427	)
428
UNCOV 429	init_container = InitContainer.model_validate(	×
430	dict(
431	name="init-user-secrets",
432	image=config.user_secrets.image,
433	env=[
434	dict(name="DATA_SERVICE_URL", value=config.data_service_url),
435	dict(name="RENKU_ACCESS_TOKEN", value=user.access_token or ""),
436	dict(name="ENCRYPTED_SECRETS_MOUNT_PATH", value="/encrypted"),
437	dict(name="DECRYPTED_SECRETS_MOUNT_PATH", value="/decrypted"),
438	],
439	volumeMounts=[
440	dict(
441	name=f"{k8s_secret_name}-volume",
442	mountPath="/encrypted",
443	readOnly=True,
444	),
445	dict(
446	name="user-secrets-volume",
447	mountPath="/decrypted",
448	readOnly=False,
449	),
450	],
451	resources={
452	"requests": {
453	"cpu": "50m",
454	"memory": "50Mi",
455	}
456	},
457	)
458	)
459
UNCOV 460	return (	×
461	init_container,
462	[volume_k8s_secrets, volume_decrypted_secrets],
463	[decrypted_volume_mount],
464	)

SwissDataScienceCenter / renku-data-services / 16369691288

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