10353957042

Committed 12 Aug 2024 02:35PM UTC coverage: 90.758% (+0.4%) from 90.398%

Build # 10353957042

Build Type

Pull #338

github

Committed by

web-flow

Commit Message

Merge 3a49eb6c2 into 8afb94949

Pull Request Pull Request #338: feat!: expand environments specification

Run Details

227 of 237 new or added lines in 7 files covered. (95.78%)

48 existing lines in 9 files now uncovered.

9202 of 10139 relevant lines covered (90.76%)

1.61 hits per line

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

80.26

/components/renku_data_services/secrets/core.py

"""Business logic for secrets storage."""

import asyncio
from base64 import b64encode

from cryptography.hazmat.primitives.asymmetric import rsa
from kubernetes import client as k8s_client
from prometheus_client import Counter, Enum
from sanic.log import logger
from ulid import ULID

from renku_data_services import base_models, errors
from renku_data_services.base_models.core import InternalServiceAdmin
from renku_data_services.k8s.client_interfaces import K8sCoreClientInterface
from renku_data_services.secrets.db import UserSecretsRepo
from renku_data_services.secrets.models import OwnerReference, Secret
from renku_data_services.utils.cryptography import (
    decrypt_rsa,
    decrypt_string,
    encrypt_rsa,
    encrypt_string,
    generate_random_encryption_key,
)


async def create_k8s_secret(
    user: base_models.APIUser,
    secret_name: str,
    namespace: str,
    secret_ids: list[ULID],
    owner_references: list[OwnerReference],
    secrets_repo: UserSecretsRepo,
    secret_service_private_key: rsa.RSAPrivateKey,
    previous_secret_service_private_key: rsa.RSAPrivateKey | None,
    core_client: K8sCoreClientInterface,
) -> None:
    """Creates a single k8s secret from a list of user secrets stored in the DB."""
    secrets = await secrets_repo.get_secrets_by_ids(requested_by=user, secret_ids=secret_ids)
    found_secret_ids = {str(s.id) for s in secrets}
    requested_secret_ids = set(map(str, secret_ids))
    missing_secret_ids = requested_secret_ids - found_secret_ids
    if len(missing_secret_ids) > 0:
        raise errors.MissingResourceError(message=f"Couldn't find secrets with ids {', '.join(missing_secret_ids)}")

    decrypted_secrets = {}
    try:
        for secret in secrets:
            try:
                decryption_key = decrypt_rsa(secret_service_private_key, secret.encrypted_key)
            except ValueError:
                if previous_secret_service_private_key is not None:
                    # If we're rotating keys right now, try the old key
                    decryption_key = decrypt_rsa(previous_secret_service_private_key, secret.encrypted_key)
                else:
                    raise

            decrypted_value = decrypt_string(decryption_key, user.id, secret.encrypted_value).encode()  # type: ignore
            decrypted_secrets[secret.name] = b64encode(decrypted_value).decode()
    except Exception as e:
        # don't wrap the error, we don't want secrets accidentally leaking.
        raise errors.SecretDecryptionError(message=f"An error occured decrypting secrets: {str(type(e))}")

    owner_refs = []
    if owner_references:
        owner_refs = [o.to_k8s() for o in owner_references]
    secret = k8s_client.V1Secret(
        data=decrypted_secrets,
        metadata=k8s_client.V1ObjectMeta(
            name=secret_name,
            namespace=namespace,
            owner_references=owner_refs,
        ),
    )

    try:
        core_client.create_namespaced_secret(namespace, secret)
    except k8s_client.ApiException as e:
        # don't wrap the error, we don't want secrets accidentally leaking.
        raise errors.SecretCreationError(message=f"An error occured creating secrets: {str(type(e))}")


async def rotate_encryption_keys(
    requested_by: InternalServiceAdmin,
    new_key: rsa.RSAPrivateKey,
    old_key: rsa.RSAPrivateKey,
    secrets_repo: UserSecretsRepo,
    batch_size: int = 100,
) -> None:
    """Rotate all secrets to a new private key.

    This method undoes the outer encryption and reencrypts with a new key, without touching the inner encryption.
    """
    processed_secrets_metrics = Counter(
        "secrets_rotation_count",
        "Number of secrets rotated",
    )
    runnning_metrics = Enum(
        "secrets_rotation_state", "State of secrets rotation", states=["running", "finished", "errored"]
    )
    runnning_metrics.state("running")
    try:
        async for batch in secrets_repo.get_all_secrets_batched(requested_by, batch_size):
            updated_secrets = []
            for secret, user_id in batch:
                new_secret = await rotate_single_encryption_key(secret, user_id, new_key, old_key)
                # we need to sleep, otherwise the async scheduler will never yield to other tasks like requests
                await asyncio.sleep(0.000001)
                if new_secret is not None:
                    updated_secrets.append(new_secret)

            await secrets_repo.update_secrets(requested_by, updated_secrets)
            processed_secrets_metrics.inc(len(updated_secrets))
    except:
        runnning_metrics.state("errored")
        raise
    else:
        runnning_metrics.state("finished")


async def rotate_single_encryption_key(
    secret: Secret, user_id: str, new_key: rsa.RSAPrivateKey, old_key: rsa.RSAPrivateKey
) -> Secret | None:
    """Rotate a single secret in place."""
    # try using new key first as a sanity check, in case it was already rotated
    try:
        decryption_key = decrypt_rsa(new_key, secret.encrypted_key)
    except ValueError:
        pass
    else:
        return None  # could decrypt with new key, nothing to do

    try:
        decryption_key = decrypt_rsa(old_key, secret.encrypted_key)
        decrypted_value = decrypt_string(decryption_key, user_id, secret.encrypted_value).encode()
        new_encryption_key = generate_random_encryption_key()
        secret.encrypted_value = encrypt_string(new_encryption_key, user_id, decrypted_value.decode())
        secret.encrypted_key = encrypt_rsa(new_key.public_key(), new_encryption_key)
    except Exception as e:
        logger.error(f"Couldn't decrypt secret {secret.name}({secret.id}): {e}")
        return None
    return secret

1	"""Business logic for secrets storage."""	2✔
2
3	import asyncio	2✔
4	from base64 import b64encode	2✔
5
6	from cryptography.hazmat.primitives.asymmetric import rsa	2✔
7	from kubernetes import client as k8s_client	2✔
8	from prometheus_client import Counter, Enum	2✔
9	from sanic.log import logger	2✔
10	from ulid import ULID	2✔
11
12	from renku_data_services import base_models, errors	2✔
13	from renku_data_services.base_models.core import InternalServiceAdmin	2✔
14	from renku_data_services.k8s.client_interfaces import K8sCoreClientInterface	2✔
15	from renku_data_services.secrets.db import UserSecretsRepo	2✔
16	from renku_data_services.secrets.models import OwnerReference, Secret	2✔
17	from renku_data_services.utils.cryptography import (	2✔
18	decrypt_rsa,
19	decrypt_string,
20	encrypt_rsa,
21	encrypt_string,
22	generate_random_encryption_key,
23	)
24
25
26	async def create_k8s_secret(	2✔
27	user: base_models.APIUser,
28	secret_name: str,
29	namespace: str,
30	secret_ids: list[ULID],
31	owner_references: list[OwnerReference],
32	secrets_repo: UserSecretsRepo,
33	secret_service_private_key: rsa.RSAPrivateKey,
34	previous_secret_service_private_key: rsa.RSAPrivateKey \| None,
35	core_client: K8sCoreClientInterface,
36	) -> None:
37	"""Creates a single k8s secret from a list of user secrets stored in the DB."""
38	secrets = await secrets_repo.get_secrets_by_ids(requested_by=user, secret_ids=secret_ids)	1✔
39	found_secret_ids = {str(s.id) for s in secrets}	1✔
40	requested_secret_ids = set(map(str, secret_ids))	1✔
41	missing_secret_ids = requested_secret_ids - found_secret_ids	1✔
42	if len(missing_secret_ids) > 0:	1✔
UNCOV 43	raise errors.MissingResourceError(message=f"Couldn't find secrets with ids {', '.join(missing_secret_ids)}")	×
44
45	decrypted_secrets = {}	1✔
46	try:	1✔
47	for secret in secrets:	1✔
48	try:	1✔
49	decryption_key = decrypt_rsa(secret_service_private_key, secret.encrypted_key)	1✔
50	except ValueError:	×
UNCOV 51	if previous_secret_service_private_key is not None:	×
52	# If we're rotating keys right now, try the old key
UNCOV 53	decryption_key = decrypt_rsa(previous_secret_service_private_key, secret.encrypted_key)	×
54	else:
UNCOV 55	raise	×
56
57	decrypted_value = decrypt_string(decryption_key, user.id, secret.encrypted_value).encode() # type: ignore	1✔
58	decrypted_secrets[secret.name] = b64encode(decrypted_value).decode()	1✔
UNCOV 59	except Exception as e:	×
60	# don't wrap the error, we don't want secrets accidentally leaking.
UNCOV 61	raise errors.SecretDecryptionError(message=f"An error occured decrypting secrets: {str(type(e))}")	×
62
63	owner_refs = []	1✔
64	if owner_references:	1✔
65	owner_refs = [o.to_k8s() for o in owner_references]	1✔
66	secret = k8s_client.V1Secret(	1✔
67	data=decrypted_secrets,
68	metadata=k8s_client.V1ObjectMeta(
69	name=secret_name,
70	namespace=namespace,
71	owner_references=owner_refs,
72	),
73	)
74
75	try:	1✔
76	core_client.create_namespaced_secret(namespace, secret)	1✔
UNCOV 77	except k8s_client.ApiException as e:	×
78	# don't wrap the error, we don't want secrets accidentally leaking.
UNCOV 79	raise errors.SecretCreationError(message=f"An error occured creating secrets: {str(type(e))}")	×
80
81
82	async def rotate_encryption_keys(	2✔
83	requested_by: InternalServiceAdmin,
84	new_key: rsa.RSAPrivateKey,
85	old_key: rsa.RSAPrivateKey,
86	secrets_repo: UserSecretsRepo,
87	batch_size: int = 100,
88	) -> None:
89	"""Rotate all secrets to a new private key.
90
91	This method undoes the outer encryption and reencrypts with a new key, without touching the inner encryption.
92	"""
93	processed_secrets_metrics = Counter(	1✔
94	"secrets_rotation_count",
95	"Number of secrets rotated",
96	)
97	runnning_metrics = Enum(	1✔
98	"secrets_rotation_state", "State of secrets rotation", states=["running", "finished", "errored"]
99	)
100	runnning_metrics.state("running")	1✔
101	try:	1✔
102	async for batch in secrets_repo.get_all_secrets_batched(requested_by, batch_size):	1✔
103	updated_secrets = []	1✔
104	for secret, user_id in batch:	1✔
105	new_secret = await rotate_single_encryption_key(secret, user_id, new_key, old_key)	1✔
106	# we need to sleep, otherwise the async scheduler will never yield to other tasks like requests
107	await asyncio.sleep(0.000001)	1✔
108	if new_secret is not None:	1✔
109	updated_secrets.append(new_secret)	1✔
110
111	await secrets_repo.update_secrets(requested_by, updated_secrets)	1✔
112	processed_secrets_metrics.inc(len(updated_secrets))	1✔
113	except:	×
114	runnning_metrics.state("errored")	×
UNCOV 115	raise	×
116	else:
117	runnning_metrics.state("finished")	1✔
118
119
120	async def rotate_single_encryption_key(	2✔
121	secret: Secret, user_id: str, new_key: rsa.RSAPrivateKey, old_key: rsa.RSAPrivateKey
122	) -> Secret \| None:
123	"""Rotate a single secret in place."""
124	# try using new key first as a sanity check, in case it was already rotated
125	try:	1✔
126	decryption_key = decrypt_rsa(new_key, secret.encrypted_key)	1✔
127	except ValueError:	1✔
128	pass	1✔
129	else:
130	return None # could decrypt with new key, nothing to do	1✔
131
132	try:	1✔
133	decryption_key = decrypt_rsa(old_key, secret.encrypted_key)	1✔
134	decrypted_value = decrypt_string(decryption_key, user_id, secret.encrypted_value).encode()	1✔
135	new_encryption_key = generate_random_encryption_key()	1✔
136	secret.encrypted_value = encrypt_string(new_encryption_key, user_id, decrypted_value.decode())	1✔
137	secret.encrypted_key = encrypt_rsa(new_key.public_key(), new_encryption_key)	1✔
138	except Exception as e:	×
139	logger.error(f"Couldn't decrypt secret {secret.name}({secret.id}): {e}")	×
UNCOV 140	return None	×
141	return secret	1✔

SwissDataScienceCenter / renku-data-services / 10353957042

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