From f088634a1414bc586cd162c44923d781010e7164 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabriel=20PEZ=C3=89?= Date: Wed, 7 Oct 2026 15:45:41 +0200 Subject: [PATCH 1/8] feat(credential): add api call for secret by reference (#8101) --- pyoaev/apis/inject.py | 46 ++++++++++++++++++++++++++++++++++- pyoaev/credential/__init__.py | 8 ++++++ 2 files changed, 53 insertions(+), 1 deletion(-) diff --git a/pyoaev/apis/inject.py b/pyoaev/apis/inject.py index 7fb6add1..2f4f21c4 100644 --- a/pyoaev/apis/inject.py +++ b/pyoaev/apis/inject.py @@ -1,7 +1,14 @@ -from typing import Any, Dict +from typing import Any, Dict, Optional + +import requests from pyoaev import exceptions as exc from pyoaev.base import RESTManager, RESTObject +from pyoaev.credential.errors import ( + CredentialErrorCode, + CredentialResolutionError, + credential_error_code_from_http, +) class Inject(RESTObject): @@ -27,3 +34,40 @@ def execution_reception( path = f"{self.path}/execution/reception/{inject_id}" result = self.openaev.http_post(path, post_data=data, **kwargs) return result + + def resolve_attachment_secret( + self, inject_id: str, attachment_id: str, authorisation: str, **kwargs: Any + ) -> Dict[str, Any]: + """Resolve the secret of a credential attached to an inject. + + Must be called while the inject is being executed, with the + authorisation code received with the job. The tenant prefix is added + by the client when a tenant id is configured. + + Raises: + CredentialResolutionError: on any failure. The request and response + bodies are never logged nor kept on the error. + """ + path = f"{self.path}/{inject_id}/attachment/secret" + data = {"attachment_id": attachment_id, "authorisation": authorisation} + error_code: Optional[CredentialErrorCode] = None + response_code: Optional[int] = None + # The error is raised outside of the except blocks so the original + # exception, which holds the response body, is not chained to it. + try: + result = self.openaev.http_post(path, post_data=data, **kwargs) + except exc.OpenAEVError as e: + response_code = e.response_code + error_code = credential_error_code_from_http( + e.response_code, e.error_message + ) + except requests.RequestException: + error_code = CredentialErrorCode.CREDENTIAL_ACCESS_DENIED + else: + if not isinstance(result, dict): + error_code = CredentialErrorCode.CREDENTIAL_ACCESS_DENIED + if error_code is not None: + raise CredentialResolutionError( + error_code, reference=attachment_id, response_code=response_code + ) + return result diff --git a/pyoaev/credential/__init__.py b/pyoaev/credential/__init__.py index 5086830e..b4688d93 100644 --- a/pyoaev/credential/__init__.py +++ b/pyoaev/credential/__init__.py @@ -1,3 +1,8 @@ +from .errors import ( + CredentialErrorCode, + CredentialResolutionError, + credential_error_code_from_http, +) from .types import CredentialType @@ -10,6 +15,9 @@ def build_single_referenced_credential_element(provider_name: str): __all__ = [ + "CredentialErrorCode", + "CredentialResolutionError", "CredentialType", + "credential_error_code_from_http", "build_single_referenced_credential_element", ] From 6e7451e13cf9991a5d6b8a6e79717405940a3e1e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabriel=20PEZ=C3=89?= Date: Wed, 7 Oct 2026 15:55:55 +0200 Subject: [PATCH 2/8] feat(credential): add api call for secret by reference (#8101) --- pyoaev/credential/errors.py | 93 ++++++++++ test/apis/inject/__init__.py | 0 test/apis/inject/test_inject.py | 206 ++++++++++++++++++++++ test/credential/test_credential_errors.py | 119 +++++++++++++ 4 files changed, 418 insertions(+) create mode 100644 pyoaev/credential/errors.py create mode 100644 test/apis/inject/__init__.py create mode 100644 test/apis/inject/test_inject.py create mode 100644 test/credential/test_credential_errors.py diff --git a/pyoaev/credential/errors.py b/pyoaev/credential/errors.py new file mode 100644 index 00000000..eb9eb029 --- /dev/null +++ b/pyoaev/credential/errors.py @@ -0,0 +1,93 @@ +"""Errors raised when an injector resolves a credential reference. + +The platform resolution endpoint answers with a closed set of error codes. +Each code carries a fixed, human-readable message meant to be reported as is +in the inject execution trace. Messages only ever mention the credential +reference id, never a secret value, and the raw HTTP request and response +bodies are deliberately never kept on the error. +""" + +from enum import Enum +from typing import Optional + +from pyoaev.exceptions import OpenAEVError + + +class CredentialErrorCode(str, Enum): + CREDENTIAL_NOT_FOUND = "CREDENTIAL_NOT_FOUND" + CREDENTIAL_INACTIVE = "CREDENTIAL_INACTIVE" + CREDENTIAL_ACCESS_DENIED = "CREDENTIAL_ACCESS_DENIED" + + +_CREDENTIAL_ERROR_MESSAGES = { + CredentialErrorCode.CREDENTIAL_NOT_FOUND: ( + "The credential configured on this inject ({reference}) no longer exists " + "at the moment of execution. Select an existing credential on the inject, " + "then run it again." + ), + CredentialErrorCode.CREDENTIAL_INACTIVE: ( + "The credential {reference} is inactive at the moment of execution. " + "Update or replace this credential, then run the inject again." + ), + CredentialErrorCode.CREDENTIAL_ACCESS_DENIED: ( + "This execution is not entitled to use the credential configured on this " + "inject. Contact your Cloud platform administrator" + ), +} + + +def credential_error_code_from_http( + response_code: Optional[int], error_message: Optional[str] +) -> CredentialErrorCode: + """Map a failed resolution call to its credential error code. + + The code sent by the platform in the response ``message`` wins. A 404 + without a known code (the inject itself does not exist) is reported as + ``CREDENTIAL_NOT_FOUND``. Any other failure falls back to + ``CREDENTIAL_ACCESS_DENIED`` so that nothing about the credential leaks. + """ + if error_message: + try: + return CredentialErrorCode(error_message.strip()) + except ValueError: + pass + if response_code == 404: + return CredentialErrorCode.CREDENTIAL_NOT_FOUND + return CredentialErrorCode.CREDENTIAL_ACCESS_DENIED + + +class CredentialResolutionError(OpenAEVError): + """A credential reference could not be resolved for the current execution. + + ``message`` holds the fixed human-readable message of ``code``. The HTTP + response body is never stored, so it cannot surface in ``str()``. + """ + + def __init__( + self, + code: CredentialErrorCode, + reference: Optional[str] = None, + response_code: Optional[int] = None, + ) -> None: + self.code = CredentialErrorCode(code) + self.reference = reference + message = _CREDENTIAL_ERROR_MESSAGES[self.code].format( + reference=reference or "unknown reference" + ) + super().__init__( + error_message=message, response_code=response_code, response_body=None + ) + + @property + def message(self) -> str: + return self.error_message + + def __str__(self) -> str: + return f"{self.code.value}: {self.error_message}" + + +__all__ = [ + "CredentialErrorCode", + "CredentialResolutionError", + "credential_error_code_from_http", +] diff --git a/test/apis/inject/__init__.py b/test/apis/inject/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/test/apis/inject/test_inject.py b/test/apis/inject/test_inject.py new file mode 100644 index 00000000..88abc258 --- /dev/null +++ b/test/apis/inject/test_inject.py @@ -0,0 +1,206 @@ +import json +import logging +from typing import Any, Optional +from unittest import TestCase, main, mock +from uuid import UUID + +import requests + +from pyoaev import OpenAEV +from pyoaev.credential.errors import CredentialErrorCode, CredentialResolutionError + +INJECT_ID = "inject-id" +ATTACHMENT_ID = "credential-reference-id" +AUTHORISATION = "authorisation-code" +TENANT_ID = UUID("2cffad3a-0001-4078-b0e2-ef74274022c3") +SECRET_VALUE = "super-secret-value" + + +def build_response( + status_code: int, + body: Optional[Any] = None, + reason: str = "", +) -> requests.Response: + response = requests.Response() + response.status_code = status_code + response.reason = reason + if body is None: + response._content = b"" + else: + response._content = json.dumps(body).encode() + response.headers["Content-Type"] = "application/json" + return response + + +class TestResolveAttachmentSecret(TestCase): + def _resolve(self, response, tenant_id=None): + api_client = OpenAEV("url", "token", tenant_id=tenant_id) + with mock.patch( + "requests.Session.request", return_value=response + ) as mock_request: + try: + result = api_client.inject.resolve_attachment_secret( + INJECT_ID, ATTACHMENT_ID, AUTHORISATION + ) + except CredentialResolutionError as error: + return mock_request, error + return mock_request, result + + def _assert_resolution_error( + self, response, expected_code: CredentialErrorCode + ) -> CredentialResolutionError: + _, error = self._resolve(response) + + self.assertIsInstance(error, CredentialResolutionError) + self.assertEqual(error.code, expected_code) + return error + + def test_success_returns_resolved_secret(self): + payload = { + "type": "AWS_ACCESS_KEY", + "value": {"aws_access_key_id": "AKIA", "aws_secret_access_key": "s"}, + } + + mock_request, result = self._resolve(build_response(200, payload)) + + self.assertEqual(result, payload) + mock_request.assert_called_once() + call = mock_request.call_args.kwargs + self.assertEqual(call["method"], "post") + self.assertEqual(call["url"], f"url/api/injects/{INJECT_ID}/attachment/secret") + self.assertEqual( + call["json"], + {"attachment_id": ATTACHMENT_ID, "authorisation": AUTHORISATION}, + ) + + def test_success_with_tenant_uses_tenant_route(self): + payload = {"type": "AWS_ACCESS_KEY", "value": {}} + + mock_request, _ = self._resolve( + build_response(200, payload), tenant_id=TENANT_ID + ) + + self.assertEqual( + mock_request.call_args.kwargs["url"], + f"url/api/tenants/{TENANT_ID}/injects/{INJECT_ID}/attachment/secret", + ) + + def test_404_credential_not_found(self): + error = self._assert_resolution_error( + build_response(404, {"message": "CREDENTIAL_NOT_FOUND"}, "Not Found"), + CredentialErrorCode.CREDENTIAL_NOT_FOUND, + ) + + self.assertEqual(error.response_code, 404) + self.assertEqual(error.reference, ATTACHMENT_ID) + self.assertIn(ATTACHMENT_ID, error.message) + + def test_400_credential_inactive(self): + error = self._assert_resolution_error( + build_response(400, {"message": "CREDENTIAL_INACTIVE"}, "Bad Request"), + CredentialErrorCode.CREDENTIAL_INACTIVE, + ) + + self.assertIn(ATTACHMENT_ID, error.message) + self.assertIn("Update or replace this credential", error.message) + + def test_403_credential_access_denied(self): + error = self._assert_resolution_error( + build_response(403, {"message": "CREDENTIAL_ACCESS_DENIED"}, "Forbidden"), + CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, + ) + + self.assertNotIn(ATTACHMENT_ID, error.message) + + def test_404_with_empty_message_is_credential_not_found(self): + self._assert_resolution_error( + build_response(404, None, "Not Found"), + CredentialErrorCode.CREDENTIAL_NOT_FOUND, + ) + self._assert_resolution_error( + build_response(404, {"message": ""}, "Not Found"), + CredentialErrorCode.CREDENTIAL_NOT_FOUND, + ) + + def test_403_without_credential_code_is_access_denied(self): + self._assert_resolution_error( + build_response(403, {"message": "Access denied"}, "Forbidden"), + CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, + ) + + def test_unknown_failures_fall_back_to_access_denied(self): + for response in ( + build_response(401, {"message": "Unauthorized"}, "Unauthorized"), + build_response(400, {"message": "attachment_id: must not be blank"}), + build_response(500, {"message": "boom"}, "Internal Server Error"), + build_response(502, None, "Bad Gateway"), + ): + with self.subTest(status_code=response.status_code): + self._assert_resolution_error( + response, CredentialErrorCode.CREDENTIAL_ACCESS_DENIED + ) + + def test_non_json_success_is_access_denied(self): + response = requests.Response() + response.status_code = 200 + response._content = b"not json" + response.headers["Content-Type"] = "text/plain" + + self._assert_resolution_error( + response, CredentialErrorCode.CREDENTIAL_ACCESS_DENIED + ) + + def test_network_error_is_access_denied(self): + api_client = OpenAEV("url", "token") + with mock.patch( + "requests.Session.request", + side_effect=requests.ConnectionError("connection refused"), + ): + with self.assertRaises(CredentialResolutionError) as context: + api_client.inject.resolve_attachment_secret( + INJECT_ID, ATTACHMENT_ID, AUTHORISATION + ) + + self.assertEqual( + context.exception.code, CredentialErrorCode.CREDENTIAL_ACCESS_DENIED + ) + + def test_error_never_exposes_response_body(self): + body = {"message": "CREDENTIAL_INACTIVE", "detail": SECRET_VALUE} + + error = self._assert_resolution_error( + build_response(400, body, "Bad Request"), + CredentialErrorCode.CREDENTIAL_INACTIVE, + ) + + self.assertIsNone(error.response_body) + self.assertIsNone(error.__cause__) + self.assertIsNone(error.__context__) + self.assertNotIn(SECRET_VALUE, str(error)) + self.assertNotIn(SECRET_VALUE, repr(error)) + self.assertNotIn(AUTHORISATION, str(error)) + + def test_request_and_response_bodies_are_never_logged(self): + payload = {"type": "AWS_ACCESS_KEY", "value": {"secret": SECRET_VALUE}} + records = [] + handler = logging.Handler(level=logging.DEBUG) + handler.emit = records.append + root_logger = logging.getLogger() + previous_level = root_logger.level + root_logger.addHandler(handler) + root_logger.setLevel(logging.DEBUG) + try: + self._resolve(build_response(200, payload)) + self._resolve(build_response(403, {"message": SECRET_VALUE})) + finally: + root_logger.removeHandler(handler) + root_logger.setLevel(previous_level) + + for record in records: + message = record.getMessage() + self.assertNotIn(SECRET_VALUE, message) + self.assertNotIn(AUTHORISATION, message) + + +if __name__ == "__main__": + main() diff --git a/test/credential/test_credential_errors.py b/test/credential/test_credential_errors.py new file mode 100644 index 00000000..a7ed7462 --- /dev/null +++ b/test/credential/test_credential_errors.py @@ -0,0 +1,119 @@ +import unittest + +from pyoaev.credential import ( + CredentialErrorCode, + CredentialResolutionError, + credential_error_code_from_http, +) +from pyoaev.exceptions import OpenAEVError + + +class CredentialErrorCodeTest(unittest.TestCase): + def test_codes_are_the_closed_platform_set(self): + self.assertEqual( + {code.value for code in CredentialErrorCode}, + { + "CREDENTIAL_NOT_FOUND", + "CREDENTIAL_INACTIVE", + "CREDENTIAL_ACCESS_DENIED", + }, + ) + + def test_platform_code_in_message_wins(self): + self.assertEqual( + credential_error_code_from_http(404, "CREDENTIAL_NOT_FOUND"), + CredentialErrorCode.CREDENTIAL_NOT_FOUND, + ) + self.assertEqual( + credential_error_code_from_http(400, "CREDENTIAL_INACTIVE"), + CredentialErrorCode.CREDENTIAL_INACTIVE, + ) + self.assertEqual( + credential_error_code_from_http(403, "CREDENTIAL_ACCESS_DENIED"), + CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, + ) + + def test_404_without_code_is_not_found(self): + self.assertEqual( + credential_error_code_from_http(404, None), + CredentialErrorCode.CREDENTIAL_NOT_FOUND, + ) + self.assertEqual( + credential_error_code_from_http(404, "Not Found"), + CredentialErrorCode.CREDENTIAL_NOT_FOUND, + ) + + def test_unknown_failure_falls_back_to_access_denied(self): + for response_code, message in ( + (400, "Bad Request"), + (401, "Unauthorized"), + (403, None), + (500, "boom"), + (None, None), + ): + with self.subTest(response_code=response_code, message=message): + self.assertEqual( + credential_error_code_from_http(response_code, message), + CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, + ) + + +class CredentialResolutionErrorTest(unittest.TestCase): + def test_not_found_message_identifies_the_reference(self): + error = CredentialResolutionError( + CredentialErrorCode.CREDENTIAL_NOT_FOUND, reference="ref-1" + ) + + self.assertEqual( + error.message, + "The credential configured on this inject (ref-1) no longer exists at " + "the moment of execution. Select an existing credential on the inject, " + "then run it again.", + ) + + def test_inactive_message_identifies_the_reference(self): + error = CredentialResolutionError( + CredentialErrorCode.CREDENTIAL_INACTIVE, reference="ref-1" + ) + + self.assertEqual( + error.message, + "The credential ref-1 is inactive at the moment of execution. Update or " + "replace this credential, then run the inject again.", + ) + + def test_access_denied_message_discloses_nothing(self): + error = CredentialResolutionError( + CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, reference="ref-1" + ) + + self.assertEqual( + error.message, + "This execution is not entitled to use the credential configured on this " + "inject. Contact your Cloud platform administrator", + ) + + def test_str_carries_the_code_and_the_message(self): + error = CredentialResolutionError( + CredentialErrorCode.CREDENTIAL_INACTIVE, + reference="ref-1", + response_code=400, + ) + + self.assertEqual(str(error), f"CREDENTIAL_INACTIVE: {error.message}") + self.assertEqual(error.response_code, 400) + self.assertIsNone(error.response_body) + + def test_accepts_code_as_string(self): + error = CredentialResolutionError("CREDENTIAL_INACTIVE", reference="ref-1") + + self.assertIs(error.code, CredentialErrorCode.CREDENTIAL_INACTIVE) + + def test_is_an_openaev_error(self): + error = CredentialResolutionError(CredentialErrorCode.CREDENTIAL_NOT_FOUND) + + self.assertIsInstance(error, OpenAEVError) + + +if __name__ == "__main__": + unittest.main() From f637202b24c70959fd23c2c8406d0729591a5988 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabriel=20PEZ=C3=89?= Date: Wed, 7 Oct 2026 16:56:03 +0200 Subject: [PATCH 3/8] feat(credential): chk 2 - typed resolved and compatibility check (#8101) --- pyoaev/credential/__init__.py | 33 +- pyoaev/credential/errors.py | 44 ++- pyoaev/credential/resolved.py | 263 +++++++++++++ pyoaev/credential/types.py | 27 ++ test/credential/test_credential_errors.py | 20 +- test/credential/test_credential_resolved.py | 393 ++++++++++++++++++++ 6 files changed, 775 insertions(+), 5 deletions(-) create mode 100644 pyoaev/credential/resolved.py create mode 100644 test/credential/test_credential_resolved.py diff --git a/pyoaev/credential/__init__.py b/pyoaev/credential/__init__.py index b4688d93..0a2b3648 100644 --- a/pyoaev/credential/__init__.py +++ b/pyoaev/credential/__init__.py @@ -1,9 +1,24 @@ from .errors import ( CredentialErrorCode, CredentialResolutionError, + InvalidResolvedSecretError, + UnsupportedSecretTypeError, credential_error_code_from_http, ) -from .types import CredentialType +from .resolved import ( + AwsAccessKeySecret, + AwsAssumeRoleSecret, + AzureManagedIdentitySecret, + AzureServicePrincipalSecret, + GcpOAuth2Secret, + GcpServiceAccountSecret, + HashSecret, + ResolvedSecret, + UsernamePasswordSecret, + ensure_compatible, + parse_resolved_secret, +) +from .types import AwsSourceIdentityType, CredentialType, HashAlgorithm, SecretType def build_single_referenced_credential_element(provider_name: str): @@ -15,9 +30,25 @@ def build_single_referenced_credential_element(provider_name: str): __all__ = [ + "AwsAccessKeySecret", + "AwsAssumeRoleSecret", + "AwsSourceIdentityType", + "AzureManagedIdentitySecret", + "AzureServicePrincipalSecret", "CredentialErrorCode", "CredentialResolutionError", "CredentialType", + "GcpOAuth2Secret", + "GcpServiceAccountSecret", + "HashAlgorithm", + "HashSecret", + "InvalidResolvedSecretError", + "ResolvedSecret", + "SecretType", + "UnsupportedSecretTypeError", + "UsernamePasswordSecret", "credential_error_code_from_http", "build_single_referenced_credential_element", + "ensure_compatible", + "parse_resolved_secret", ] diff --git a/pyoaev/credential/errors.py b/pyoaev/credential/errors.py index eb9eb029..c8e60a64 100644 --- a/pyoaev/credential/errors.py +++ b/pyoaev/credential/errors.py @@ -1,7 +1,8 @@ """Errors raised when an injector resolves a credential reference. The platform resolution endpoint answers with a closed set of error codes. -Each code carries a fixed, human-readable message meant to be reported as is +``CREDENTIAL_INCOMPATIBLE`` is detected by the injector itself, when the +resolved credential does not satisfy the contract field. Each code carries a fixed, human-readable message meant to be reported as is in the inject execution trace. Messages only ever mention the credential reference id, never a secret value, and the raw HTTP request and response bodies are deliberately never kept on the error. @@ -17,6 +18,14 @@ class CredentialErrorCode(str, Enum): CREDENTIAL_NOT_FOUND = "CREDENTIAL_NOT_FOUND" CREDENTIAL_INACTIVE = "CREDENTIAL_INACTIVE" CREDENTIAL_ACCESS_DENIED = "CREDENTIAL_ACCESS_DENIED" + CREDENTIAL_INCOMPATIBLE = "CREDENTIAL_INCOMPATIBLE" + + +_PLATFORM_CREDENTIAL_ERROR_CODES = { + CredentialErrorCode.CREDENTIAL_NOT_FOUND, + CredentialErrorCode.CREDENTIAL_INACTIVE, + CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, +} _CREDENTIAL_ERROR_MESSAGES = { @@ -33,6 +42,10 @@ class CredentialErrorCode(str, Enum): "This execution is not entitled to use the credential configured on this " "inject. Contact your Cloud platform administrator" ), + CredentialErrorCode.CREDENTIAL_INCOMPATIBLE: ( + "The credential {reference} is not compatible with this inject. Select a " + "credential of the type expected by the inject, then run it again." + ), } @@ -48,9 +61,11 @@ def credential_error_code_from_http( """ if error_message: try: - return CredentialErrorCode(error_message.strip()) + code = CredentialErrorCode(error_message.strip()) except ValueError: - pass + code = None + if code in _PLATFORM_CREDENTIAL_ERROR_CODES: + return code if response_code == 404: return CredentialErrorCode.CREDENTIAL_NOT_FOUND return CredentialErrorCode.CREDENTIAL_ACCESS_DENIED @@ -86,8 +101,31 @@ def __str__(self) -> str: return f"{self.code.value}: {self.error_message}" +class UnsupportedSecretTypeError(CredentialResolutionError): + """The platform resolved a secret type this client does not know.""" + + def __init__( + self, secret_type: Optional[str] = None, reference: Optional[str] = None + ) -> None: + self.secret_type = secret_type + super().__init__(CredentialErrorCode.CREDENTIAL_INCOMPATIBLE, reference) + + +class InvalidResolvedSecretError(CredentialResolutionError): + """The resolved secret payload does not match its declared type. + + Reported as ``CREDENTIAL_ACCESS_DENIED``, like any unexpected failure, so + that nothing about the payload leaks in the trace. + """ + + def __init__(self, reference: Optional[str] = None) -> None: + super().__init__(CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, reference) + + __all__ = [ "CredentialErrorCode", "CredentialResolutionError", + "InvalidResolvedSecretError", + "UnsupportedSecretTypeError", "credential_error_code_from_http", ] diff --git a/pyoaev/credential/resolved.py b/pyoaev/credential/resolved.py new file mode 100644 index 00000000..714ce343 --- /dev/null +++ b/pyoaev/credential/resolved.py @@ -0,0 +1,263 @@ +"""Typed secrets resolved from a credential reference at execution time. + +``parse_resolved_secret`` turns the payload returned by the platform resolution +endpoint into one frozen dataclass per ``SecretType``. Secret material is +excluded from ``repr``/``str`` so a resolved secret can never end up in a log +line by accident. + +``ensure_compatible`` is the injector-side check that the resolved secret +satisfies the credential type declared on the contract field. +""" + +import base64 +from dataclasses import MISSING, dataclass, field, fields +from typing import Any, ClassVar, Dict, Optional, Type, TypeVar, Union + +from pyoaev.credential.errors import ( + CredentialErrorCode, + CredentialResolutionError, + InvalidResolvedSecretError, + UnsupportedSecretTypeError, +) +from pyoaev.credential.types import ( + AwsSourceIdentityType, + CredentialType, + HashAlgorithm, + SecretType, +) + +_SECRET_TYPE_TO_CREDENTIAL_TYPE = { + SecretType.USERNAME_PASSWORD: CredentialType.IDENTITY, + SecretType.HASH: CredentialType.IDENTITY, + SecretType.AWS_ACCESS_KEY: CredentialType.CLOUD_AWS, + SecretType.AWS_ASSUME_ROLE: CredentialType.CLOUD_AWS, + SecretType.AZURE_SERVICE_PRINCIPAL: CredentialType.CLOUD_AZURE, + SecretType.AZURE_MANAGED_IDENTITY: CredentialType.CLOUD_AZURE, + SecretType.GCP_SERVICE_ACCOUNT: CredentialType.CLOUD_GCP, + SecretType.GCP_OAUTH2: CredentialType.CLOUD_GCP, +} + + +def _secret(**kwargs: Any) -> Any: + """A field holding secret material: hidden from ``repr``/``str``.""" + return field(repr=False, **kwargs) + + +def _parsed_with(parser, **kwargs: Any) -> Any: + """A field whose raw string value is converted by ``parser``.""" + return field(metadata={"parser": parser}, **kwargs) + + +def _decode_base64(value: str) -> bytes: + return base64.b64decode(value, validate=True) + + +@dataclass(frozen=True) +class ResolvedSecret: + secret_type: ClassVar[SecretType] + + @property + def credential_type(self) -> CredentialType: + return _SECRET_TYPE_TO_CREDENTIAL_TYPE[self.secret_type] + + +@dataclass(frozen=True) +class UsernamePasswordSecret(ResolvedSecret): + secret_type: ClassVar[SecretType] = SecretType.USERNAME_PASSWORD + + username: str + password: str = _secret() + + +@dataclass(frozen=True) +class HashSecret(ResolvedSecret): + secret_type: ClassVar[SecretType] = SecretType.HASH + + hash_algorithm: HashAlgorithm = _parsed_with(HashAlgorithm) + hash: str = _secret() + + +@dataclass(frozen=True) +class AwsAccessKeySecret(ResolvedSecret): + secret_type: ClassVar[SecretType] = SecretType.AWS_ACCESS_KEY + + aws_access_key_id: str = _secret() + aws_secret_access_key: str = _secret() + aws_default_region: Optional[str] = None + aws_session_token: Optional[str] = _secret(default=None) + + +@dataclass(frozen=True) +class AwsAssumeRoleSecret(ResolvedSecret): + secret_type: ClassVar[SecretType] = SecretType.AWS_ASSUME_ROLE + + aws_role_arn: str + aws_source_identity_type: AwsSourceIdentityType = _parsed_with( + AwsSourceIdentityType + ) + aws_default_region: Optional[str] = None + aws_external_id: Optional[str] = _secret(default=None) + aws_source_profile_access_key_id: Optional[str] = _secret(default=None) + aws_source_profile_secret_access_key: Optional[str] = _secret(default=None) + + def __post_init__(self) -> None: + # A static source identity cannot assume the role without its keys. + if self.aws_source_identity_type == AwsSourceIdentityType.STATIC_ACCESS_KEY: + if not ( + self.aws_source_profile_access_key_id + and self.aws_source_profile_secret_access_key + ): + raise ValueError("Missing source profile keys") + + +@dataclass(frozen=True) +class AzureServicePrincipalSecret(ResolvedSecret): + secret_type: ClassVar[SecretType] = SecretType.AZURE_SERVICE_PRINCIPAL + + azure_environment: str + azure_client_id: str + azure_client_secret: str = _secret() + azure_tenant_id: str + azure_subscription_id: Optional[str] = None + + +@dataclass(frozen=True) +class AzureManagedIdentitySecret(ResolvedSecret): + secret_type: ClassVar[SecretType] = SecretType.AZURE_MANAGED_IDENTITY + + azure_environment: str + azure_client_id: Optional[str] = None + azure_subscription_id: Optional[str] = None + + +@dataclass(frozen=True) +class GcpServiceAccountSecret(ResolvedSecret): + secret_type: ClassVar[SecretType] = SecretType.GCP_SERVICE_ACCOUNT + + gcp_scope: str + # Java ``byte[]``, sent base64 encoded by the platform. + gcp_private_key_json: bytes = _parsed_with(_decode_base64, repr=False) + gcp_project_id: Optional[str] = None + + +@dataclass(frozen=True) +class GcpOAuth2Secret(ResolvedSecret): + secret_type: ClassVar[SecretType] = SecretType.GCP_OAUTH2 + + gcp_scope: str + gcp_oauth_client_id: str + gcp_oauth_client_secret: str = _secret() + gcp_oauth_refresh_token: str = _secret() + gcp_project_id: Optional[str] = None + + +_RESOLVED_SECRET_CLASSES: Dict[SecretType, Type[ResolvedSecret]] = { + cls.secret_type: cls + for cls in ( + UsernamePasswordSecret, + HashSecret, + AwsAccessKeySecret, + AwsAssumeRoleSecret, + AzureServicePrincipalSecret, + AzureManagedIdentitySecret, + GcpServiceAccountSecret, + GcpOAuth2Secret, + ) +} + +_R = TypeVar("_R", bound=ResolvedSecret) + + +def _build(cls: Type[_R], value: Dict[str, Any]) -> _R: + kwargs: Dict[str, Any] = {} + for secret_field in fields(cls): + raw = value.get(secret_field.name) + # The platform omits null fields; an empty string is as good as absent. + if raw is None or raw == "": + if secret_field.default is MISSING: + raise ValueError(f"Missing required field {secret_field.name}") + continue + if not isinstance(raw, str): + raise ValueError(f"Unexpected value type for {secret_field.name}") + parser = secret_field.metadata.get("parser") + kwargs[secret_field.name] = parser(raw) if parser else raw + return cls(**kwargs) + + +def parse_resolved_secret( + payload: Any, reference: Optional[str] = None +) -> ResolvedSecret: + """Build the typed secret from a resolution endpoint payload. + + Raises: + UnsupportedSecretTypeError: the ``type`` is not a known ``SecretType``. + InvalidResolvedSecretError: the payload does not match its ``type`` + (missing required key, wrong value type, invalid base64...). + """ + if not isinstance(payload, dict): + raise InvalidResolvedSecretError(reference) + raw_type = payload.get("type") + try: + secret_type = SecretType(raw_type) + except ValueError: + raise UnsupportedSecretTypeError( + raw_type if isinstance(raw_type, str) else None, reference + ) from None + + value = payload.get("value") + if not isinstance(value, dict): + raise InvalidResolvedSecretError(reference) + error: Optional[CredentialResolutionError] = None + # Raised outside of the except block so that the original error, which may + # quote a secret value, is not chained to it. + try: + return _build(_RESOLVED_SECRET_CLASSES[secret_type], value) + except (ValueError, TypeError): + error = InvalidResolvedSecretError(reference) + raise error + + +def ensure_compatible( + resolved: _R, + expected: Union[CredentialType, str], + reference: Optional[str] = None, +) -> _R: + """Check that ``resolved`` satisfies the credential type of the contract. + + ``expected`` is either a ``CredentialType`` or a provider name (``aws``, + ``eks``, ``azure``, ``gcp``), mapped with the same table as the one used to + declare the contract field. + + Raises: + CredentialResolutionError: ``CREDENTIAL_INCOMPATIBLE`` on mismatch. + ValueError: ``expected`` is a provider name with no credential type. + """ + # Imported here: utils depends on contracts, which import this package. + from pyoaev.credential.utils import _resolve_credential_type + + if isinstance(expected, CredentialType): + expected_type = expected + else: + expected_type = _resolve_credential_type(expected) + if expected_type is None: + raise ValueError(f"No credential type is mapped to provider {expected!r}") + if resolved.credential_type != expected_type: + raise CredentialResolutionError( + CredentialErrorCode.CREDENTIAL_INCOMPATIBLE, reference + ) + return resolved + + +__all__ = [ + "AwsAccessKeySecret", + "AwsAssumeRoleSecret", + "AzureManagedIdentitySecret", + "AzureServicePrincipalSecret", + "GcpOAuth2Secret", + "GcpServiceAccountSecret", + "HashSecret", + "ResolvedSecret", + "UsernamePasswordSecret", + "ensure_compatible", + "parse_resolved_secret", +] diff --git a/pyoaev/credential/types.py b/pyoaev/credential/types.py index f2600a4d..4db1c460 100644 --- a/pyoaev/credential/types.py +++ b/pyoaev/credential/types.py @@ -14,3 +14,30 @@ class CredentialType(str, Enum): CLOUD_AWS = "CLOUD_AWS" CLOUD_AZURE = "CLOUD_AZURE" CLOUD_GCP = "CLOUD_GCP" + + +class SecretType(str, Enum): + """Mirror of the platform ``Secret.SECRET_TYPE`` enum, label for label.""" + + USERNAME_PASSWORD = "USERNAME_PASSWORD" + HASH = "HASH" + AWS_ACCESS_KEY = "AWS_ACCESS_KEY" + AWS_ASSUME_ROLE = "AWS_ASSUME_ROLE" + AZURE_SERVICE_PRINCIPAL = "AZURE_SERVICE_PRINCIPAL" + AZURE_MANAGED_IDENTITY = "AZURE_MANAGED_IDENTITY" + GCP_SERVICE_ACCOUNT = "GCP_SERVICE_ACCOUNT" + GCP_OAUTH2 = "GCP_OAUTH2" + + +class AwsSourceIdentityType(str, Enum): + """Mirror of the platform ``AwsAssumeRoleSecret.AWS_SOURCE_IDENTITY_TYPE`` enum.""" + + STATIC_ACCESS_KEY = "STATIC_ACCESS_KEY" + INSTANCE_DEFAULT = "INSTANCE_DEFAULT" + + +class HashAlgorithm(str, Enum): + """Mirror of the platform ``HashSecret.HASH_ALGORITHM`` enum.""" + + SHA = "SHA" + NTLM = "NTLM" diff --git a/test/credential/test_credential_errors.py b/test/credential/test_credential_errors.py index a7ed7462..50240225 100644 --- a/test/credential/test_credential_errors.py +++ b/test/credential/test_credential_errors.py @@ -9,16 +9,23 @@ class CredentialErrorCodeTest(unittest.TestCase): - def test_codes_are_the_closed_platform_set(self): + def test_codes_are_a_closed_set(self): self.assertEqual( {code.value for code in CredentialErrorCode}, { "CREDENTIAL_NOT_FOUND", "CREDENTIAL_INACTIVE", "CREDENTIAL_ACCESS_DENIED", + "CREDENTIAL_INCOMPATIBLE", }, ) + def test_injector_side_code_is_never_taken_from_the_platform(self): + self.assertEqual( + credential_error_code_from_http(400, "CREDENTIAL_INCOMPATIBLE"), + CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, + ) + def test_platform_code_in_message_wins(self): self.assertEqual( credential_error_code_from_http(404, "CREDENTIAL_NOT_FOUND"), @@ -93,6 +100,17 @@ def test_access_denied_message_discloses_nothing(self): "inject. Contact your Cloud platform administrator", ) + def test_incompatible_message_identifies_the_reference(self): + error = CredentialResolutionError( + CredentialErrorCode.CREDENTIAL_INCOMPATIBLE, reference="ref-1" + ) + + self.assertEqual( + error.message, + "The credential ref-1 is not compatible with this inject. Select a " + "credential of the type expected by the inject, then run it again.", + ) + def test_str_carries_the_code_and_the_message(self): error = CredentialResolutionError( CredentialErrorCode.CREDENTIAL_INACTIVE, diff --git a/test/credential/test_credential_resolved.py b/test/credential/test_credential_resolved.py new file mode 100644 index 00000000..69d70494 --- /dev/null +++ b/test/credential/test_credential_resolved.py @@ -0,0 +1,393 @@ +import base64 +import unittest + +from pyoaev.credential import ( + AwsAccessKeySecret, + AwsAssumeRoleSecret, + AwsSourceIdentityType, + AzureManagedIdentitySecret, + AzureServicePrincipalSecret, + CredentialErrorCode, + CredentialResolutionError, + CredentialType, + GcpOAuth2Secret, + GcpServiceAccountSecret, + HashAlgorithm, + HashSecret, + InvalidResolvedSecretError, + SecretType, + UnsupportedSecretTypeError, + UsernamePasswordSecret, + ensure_compatible, + parse_resolved_secret, +) + +REFERENCE = "credential-reference-id" +SECRET = "super-secret-value" +GCP_KEY_JSON = b'{"type": "service_account", "private_key": "super-secret-value"}' + +PAYLOADS = { + SecretType.USERNAME_PASSWORD: { + "username": "alice", + "password": SECRET, + }, + SecretType.HASH: { + "hash_algorithm": "NTLM", + "hash": SECRET, + }, + SecretType.AWS_ACCESS_KEY: { + "aws_default_region": "eu-west-3", + "aws_access_key_id": "AKIAEXAMPLE", + "aws_secret_access_key": SECRET, + "aws_session_token": SECRET, + }, + SecretType.AWS_ASSUME_ROLE: { + "aws_default_region": "us-east-1", + "aws_role_arn": "arn:aws:iam::123456789012:role/srt", + "aws_source_identity_type": "STATIC_ACCESS_KEY", + "aws_external_id": "external-id", + "aws_source_profile_access_key_id": "AKIASOURCE", + "aws_source_profile_secret_access_key": SECRET, + }, + SecretType.AZURE_SERVICE_PRINCIPAL: { + "azure_environment": "AzureCloud", + "azure_client_id": "client-id", + "azure_client_secret": SECRET, + "azure_tenant_id": "tenant-id", + "azure_subscription_id": "subscription-id", + }, + SecretType.AZURE_MANAGED_IDENTITY: { + "azure_environment": "AzureChinaCloud", + "azure_client_id": "client-id", + "azure_subscription_id": "subscription-id", + }, + SecretType.GCP_SERVICE_ACCOUNT: { + "gcp_scope": "googleapis.com", + "gcp_project_id": "project-id", + "gcp_private_key_json": base64.b64encode(GCP_KEY_JSON).decode(), + }, + SecretType.GCP_OAUTH2: { + "gcp_scope": "googleapis.com", + "gcp_project_id": "project-id", + "gcp_oauth_client_id": "client-id", + "gcp_oauth_client_secret": SECRET, + "gcp_oauth_refresh_token": SECRET, + }, +} + +EXPECTED_CREDENTIAL_TYPES = { + SecretType.USERNAME_PASSWORD: CredentialType.IDENTITY, + SecretType.HASH: CredentialType.IDENTITY, + SecretType.AWS_ACCESS_KEY: CredentialType.CLOUD_AWS, + SecretType.AWS_ASSUME_ROLE: CredentialType.CLOUD_AWS, + SecretType.AZURE_SERVICE_PRINCIPAL: CredentialType.CLOUD_AZURE, + SecretType.AZURE_MANAGED_IDENTITY: CredentialType.CLOUD_AZURE, + SecretType.GCP_SERVICE_ACCOUNT: CredentialType.CLOUD_GCP, + SecretType.GCP_OAUTH2: CredentialType.CLOUD_GCP, +} + + +def _parse(secret_type, **overrides): + value = {**PAYLOADS[secret_type], **overrides} + value = {key: item for key, item in value.items() if item is not None} + return parse_resolved_secret( + {"type": secret_type.value, "value": value}, reference=REFERENCE + ) + + +class SecretTypeTest(unittest.TestCase): + def test_secret_type_mirrors_platform_labels(self): + self.assertEqual( + [secret_type.value for secret_type in SecretType], + [ + "USERNAME_PASSWORD", + "HASH", + "AWS_ACCESS_KEY", + "AWS_ASSUME_ROLE", + "AZURE_SERVICE_PRINCIPAL", + "AZURE_MANAGED_IDENTITY", + "GCP_SERVICE_ACCOUNT", + "GCP_OAUTH2", + ], + ) + + +class ParseResolvedSecretTest(unittest.TestCase): + def test_username_password(self): + resolved = _parse(SecretType.USERNAME_PASSWORD) + + self.assertEqual(resolved, UsernamePasswordSecret("alice", SECRET)) + + def test_hash(self): + resolved = _parse(SecretType.HASH) + + self.assertEqual(resolved, HashSecret(HashAlgorithm.NTLM, SECRET)) + + def test_aws_access_key(self): + resolved = _parse(SecretType.AWS_ACCESS_KEY) + + self.assertEqual( + resolved, + AwsAccessKeySecret( + aws_access_key_id="AKIAEXAMPLE", + aws_secret_access_key=SECRET, + aws_default_region="eu-west-3", + aws_session_token=SECRET, + ), + ) + + def test_aws_access_key_optional_fields_default_to_none(self): + resolved = _parse( + SecretType.AWS_ACCESS_KEY, aws_default_region=None, aws_session_token=None + ) + + self.assertIsNone(resolved.aws_default_region) + self.assertIsNone(resolved.aws_session_token) + + def test_aws_assume_role_with_static_access_key(self): + resolved = _parse(SecretType.AWS_ASSUME_ROLE) + + self.assertEqual( + resolved, + AwsAssumeRoleSecret( + aws_role_arn="arn:aws:iam::123456789012:role/srt", + aws_source_identity_type=AwsSourceIdentityType.STATIC_ACCESS_KEY, + aws_default_region="us-east-1", + aws_external_id="external-id", + aws_source_profile_access_key_id="AKIASOURCE", + aws_source_profile_secret_access_key=SECRET, + ), + ) + + def test_aws_assume_role_with_instance_default_needs_no_source_keys(self): + resolved = _parse( + SecretType.AWS_ASSUME_ROLE, + aws_source_identity_type="INSTANCE_DEFAULT", + aws_external_id=None, + aws_source_profile_access_key_id=None, + aws_source_profile_secret_access_key=None, + ) + + self.assertEqual( + resolved.aws_source_identity_type, AwsSourceIdentityType.INSTANCE_DEFAULT + ) + self.assertIsNone(resolved.aws_source_profile_access_key_id) + self.assertIsNone(resolved.aws_source_profile_secret_access_key) + + def test_aws_assume_role_with_static_access_key_requires_source_keys(self): + with self.assertRaises(InvalidResolvedSecretError): + _parse( + SecretType.AWS_ASSUME_ROLE, aws_source_profile_secret_access_key=None + ) + + def test_azure_service_principal(self): + resolved = _parse(SecretType.AZURE_SERVICE_PRINCIPAL) + + self.assertEqual( + resolved, + AzureServicePrincipalSecret( + azure_environment="AzureCloud", + azure_client_id="client-id", + azure_client_secret=SECRET, + azure_tenant_id="tenant-id", + azure_subscription_id="subscription-id", + ), + ) + + def test_azure_managed_identity(self): + resolved = _parse(SecretType.AZURE_MANAGED_IDENTITY) + + self.assertEqual( + resolved, + AzureManagedIdentitySecret( + azure_environment="AzureChinaCloud", + azure_client_id="client-id", + azure_subscription_id="subscription-id", + ), + ) + + def test_azure_managed_identity_without_client_id(self): + resolved = _parse(SecretType.AZURE_MANAGED_IDENTITY, azure_client_id=None) + + self.assertIsNone(resolved.azure_client_id) + + def test_gcp_service_account_decodes_the_private_key_json(self): + resolved = _parse(SecretType.GCP_SERVICE_ACCOUNT) + + self.assertEqual( + resolved, + GcpServiceAccountSecret( + gcp_scope="googleapis.com", + gcp_private_key_json=GCP_KEY_JSON, + gcp_project_id="project-id", + ), + ) + + def test_gcp_service_account_with_invalid_base64(self): + with self.assertRaises(InvalidResolvedSecretError): + _parse(SecretType.GCP_SERVICE_ACCOUNT, gcp_private_key_json="not base64!") + + def test_gcp_oauth2(self): + resolved = _parse(SecretType.GCP_OAUTH2) + + self.assertEqual( + resolved, + GcpOAuth2Secret( + gcp_scope="googleapis.com", + gcp_oauth_client_id="client-id", + gcp_oauth_client_secret=SECRET, + gcp_oauth_refresh_token=SECRET, + gcp_project_id="project-id", + ), + ) + + def test_every_type_exposes_its_secret_and_credential_type(self): + for secret_type, credential_type in EXPECTED_CREDENTIAL_TYPES.items(): + with self.subTest(secret_type=secret_type): + resolved = _parse(secret_type) + + self.assertIs(resolved.secret_type, secret_type) + self.assertIs(resolved.credential_type, credential_type) + + def test_repr_and_str_never_expose_secret_material(self): + for secret_type in SecretType: + with self.subTest(secret_type=secret_type): + resolved = _parse(secret_type) + + self.assertNotIn(SECRET, repr(resolved)) + self.assertNotIn(SECRET, str(resolved)) + + def test_resolved_secret_is_frozen(self): + resolved = _parse(SecretType.AWS_ACCESS_KEY) + + with self.assertRaises(AttributeError): + resolved.aws_secret_access_key = "other" + + def test_unknown_extra_keys_are_ignored(self): + resolved = _parse(SecretType.USERNAME_PASSWORD, domain="corp") + + self.assertEqual(resolved, UsernamePasswordSecret("alice", SECRET)) + + def test_missing_required_key(self): + for secret_type, key in ( + (SecretType.AWS_ACCESS_KEY, "aws_secret_access_key"), + (SecretType.AWS_ASSUME_ROLE, "aws_role_arn"), + (SecretType.AZURE_SERVICE_PRINCIPAL, "azure_tenant_id"), + (SecretType.AZURE_MANAGED_IDENTITY, "azure_environment"), + (SecretType.GCP_SERVICE_ACCOUNT, "gcp_private_key_json"), + (SecretType.GCP_OAUTH2, "gcp_oauth_refresh_token"), + ): + with self.subTest(secret_type=secret_type, key=key): + with self.assertRaises(InvalidResolvedSecretError) as context: + _parse(secret_type, **{key: None}) + + self.assertEqual( + context.exception.code, + CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, + ) + + def test_empty_required_value_is_missing(self): + with self.assertRaises(InvalidResolvedSecretError): + _parse(SecretType.AWS_ACCESS_KEY, aws_access_key_id="") + + def test_non_string_value(self): + with self.assertRaises(InvalidResolvedSecretError): + _parse(SecretType.AWS_ACCESS_KEY, aws_access_key_id=123) + + def test_unknown_enum_value(self): + with self.assertRaises(InvalidResolvedSecretError): + _parse(SecretType.AWS_ASSUME_ROLE, aws_source_identity_type="FEDERATED") + + def test_invalid_error_never_exposes_secret_material(self): + with self.assertRaises(InvalidResolvedSecretError) as context: + _parse(SecretType.GCP_SERVICE_ACCOUNT, gcp_private_key_json=SECRET) + + error = context.exception + self.assertEqual(error.reference, REFERENCE) + self.assertIsNone(error.__cause__) + self.assertIsNone(error.__context__) + self.assertNotIn(SECRET, str(error)) + + def test_payload_is_not_a_dict(self): + for payload in (None, [], "AWS_ACCESS_KEY"): + with self.subTest(payload=payload): + with self.assertRaises(InvalidResolvedSecretError): + parse_resolved_secret(payload) + + def test_value_is_not_a_dict(self): + for value in (None, [], "secret"): + with self.subTest(value=value): + with self.assertRaises(InvalidResolvedSecretError): + parse_resolved_secret({"type": "AWS_ACCESS_KEY", "value": value}) + + def test_unknown_type(self): + with self.assertRaises(UnsupportedSecretTypeError) as context: + parse_resolved_secret( + {"type": "KERBEROS_TICKET", "value": {}}, reference=REFERENCE + ) + + error = context.exception + self.assertEqual(error.secret_type, "KERBEROS_TICKET") + self.assertEqual(error.code, CredentialErrorCode.CREDENTIAL_INCOMPATIBLE) + self.assertIn(REFERENCE, error.message) + + def test_missing_type(self): + with self.assertRaises(UnsupportedSecretTypeError) as context: + parse_resolved_secret({"value": {}}) + + self.assertIsNone(context.exception.secret_type) + + +class EnsureCompatibleTest(unittest.TestCase): + def test_matching_credential_type_returns_the_secret(self): + for secret_type, credential_type in EXPECTED_CREDENTIAL_TYPES.items(): + with self.subTest(secret_type=secret_type): + resolved = _parse(secret_type) + + self.assertIs(ensure_compatible(resolved, credential_type), resolved) + + def test_mismatching_credential_type_is_incompatible(self): + for secret_type, credential_type in EXPECTED_CREDENTIAL_TYPES.items(): + for expected in CredentialType: + if expected == credential_type: + continue + with self.subTest(secret_type=secret_type, expected=expected): + with self.assertRaises(CredentialResolutionError) as context: + ensure_compatible( + _parse(secret_type), expected, reference=REFERENCE + ) + + error = context.exception + self.assertEqual( + error.code, CredentialErrorCode.CREDENTIAL_INCOMPATIBLE + ) + self.assertIn(REFERENCE, error.message) + self.assertNotIn(SECRET, str(error)) + + def test_provider_name_uses_the_contract_mapping(self): + for provider, secret_type in ( + ("aws", SecretType.AWS_ACCESS_KEY), + ("eks", SecretType.AWS_ASSUME_ROLE), + ("AZURE", SecretType.AZURE_SERVICE_PRINCIPAL), + ("gcp", SecretType.GCP_OAUTH2), + ): + with self.subTest(provider=provider): + resolved = _parse(secret_type) + + self.assertIs(ensure_compatible(resolved, provider), resolved) + + def test_provider_name_mismatch_is_incompatible(self): + with self.assertRaises(CredentialResolutionError) as context: + ensure_compatible(_parse(SecretType.GCP_SERVICE_ACCOUNT), "aws") + + self.assertEqual( + context.exception.code, CredentialErrorCode.CREDENTIAL_INCOMPATIBLE + ) + + def test_unknown_provider_name_is_a_programming_error(self): + with self.assertRaises(ValueError): + ensure_compatible(_parse(SecretType.AWS_ACCESS_KEY), "kubernetes") + + +if __name__ == "__main__": + unittest.main() From 03397ec6d4266bb7b254f6c664f3e1d8ce2c57d5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabriel=20PEZ=C3=89?= Date: Wed, 7 Oct 2026 17:22:36 +0200 Subject: [PATCH 4/8] feat(credential): chk 3 - read the attached creds (#8101) --- pyoaev/credential/__init__.py | 8 + pyoaev/credential/attachment.py | 114 +++++++++ test/credential/test_credential_attachment.py | 220 ++++++++++++++++++ 3 files changed, 342 insertions(+) create mode 100644 pyoaev/credential/attachment.py create mode 100644 test/credential/test_credential_attachment.py diff --git a/pyoaev/credential/__init__.py b/pyoaev/credential/__init__.py index 0a2b3648..a221f711 100644 --- a/pyoaev/credential/__init__.py +++ b/pyoaev/credential/__init__.py @@ -1,3 +1,8 @@ +from .attachment import ( + CredentialAttachment, + get_credential_attachment, + resolve_inject_credential, +) from .errors import ( CredentialErrorCode, CredentialResolutionError, @@ -35,6 +40,7 @@ def build_single_referenced_credential_element(provider_name: str): "AwsSourceIdentityType", "AzureManagedIdentitySecret", "AzureServicePrincipalSecret", + "CredentialAttachment", "CredentialErrorCode", "CredentialResolutionError", "CredentialType", @@ -50,5 +56,7 @@ def build_single_referenced_credential_element(provider_name: str): "credential_error_code_from_http", "build_single_referenced_credential_element", "ensure_compatible", + "get_credential_attachment", "parse_resolved_secret", + "resolve_inject_credential", ] diff --git a/pyoaev/credential/attachment.py b/pyoaev/credential/attachment.py new file mode 100644 index 00000000..f0425f44 --- /dev/null +++ b/pyoaev/credential/attachment.py @@ -0,0 +1,114 @@ +"""Read the credential attachment of an inject job and resolve it. + +The platform adds an ``attachments`` object to the job message when the inject +carries a credential reference:: + + {"credential_references": [""], "authorisation_code": ""} + +``attachments`` is ``null`` when the inject has no reference: the injector then +keeps its legacy credential fields and no resolution request is sent. + +``resolve_inject_credential`` is the single entry point injectors call. +""" + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, Dict, Optional, Union + +from pyoaev.credential.errors import CredentialErrorCode, CredentialResolutionError +from pyoaev.credential.resolved import ( + ResolvedSecret, + ensure_compatible, + parse_resolved_secret, +) +from pyoaev.credential.types import CredentialType + +if TYPE_CHECKING: + from pyoaev.client import OpenAEV + + +@dataclass(frozen=True) +class CredentialAttachment: + reference: str + # Only valid for the current job, but still a bearer token: keep it out of logs. + authorisation_code: str = field(repr=False) + + +def _is_filled_string(value: Any) -> bool: + return isinstance(value, str) and value.strip() != "" + + +def get_credential_attachment(data: Dict[str, Any]) -> Optional[CredentialAttachment]: + """Return the credential attachment of a job message, if any. + + Only the first credential reference is used. + + Returns: + ``None`` when ``attachments`` is missing, ``null`` or has no reference, + meaning the injector must use its legacy credential fields. + + Raises: + CredentialResolutionError: ``CREDENTIAL_ACCESS_DENIED`` when a reference + is present without an authorisation code, or when the attachment + is malformed. + """ + attachments = data.get("attachments") + if attachments is None: + return None + if not isinstance(attachments, dict): + raise CredentialResolutionError(CredentialErrorCode.CREDENTIAL_ACCESS_DENIED) + + references = attachments.get("credential_references") + if not references: + return None + if not isinstance(references, list) or not _is_filled_string(references[0]): + raise CredentialResolutionError(CredentialErrorCode.CREDENTIAL_ACCESS_DENIED) + reference = references[0] + + authorisation_code = attachments.get("authorisation_code") + if not _is_filled_string(authorisation_code): + raise CredentialResolutionError( + CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, reference + ) + return CredentialAttachment(reference, authorisation_code) + + +def resolve_inject_credential( + api: "OpenAEV", + inject_id: str, + data: Dict[str, Any], + expected_type: Union[CredentialType, str], +) -> Optional[ResolvedSecret]: + """Resolve the credential attached to an inject job, just in time. + + Must be called after ``execution_reception`` and before + ``execution_callback``, while the inject is in progress. + + Args: + api: the OpenAEV client of the injector. + inject_id: the id of the inject being executed. + data: the job message received by the injector. + expected_type: the credential type declared on the contract field, + or the provider name it was declared from (see ``ensure_compatible``). + + Returns: + ``None`` when the job carries no credential reference (legacy path), + otherwise the resolved secret, checked against ``expected_type``. + + Raises: + CredentialResolutionError: with the code to report in the trace. + """ + attachment = get_credential_attachment(data) + if attachment is None: + return None + payload = api.inject.resolve_attachment_secret( + inject_id, attachment.reference, attachment.authorisation_code + ) + resolved = parse_resolved_secret(payload, reference=attachment.reference) + return ensure_compatible(resolved, expected_type, reference=attachment.reference) + + +__all__ = [ + "CredentialAttachment", + "get_credential_attachment", + "resolve_inject_credential", +] diff --git a/test/credential/test_credential_attachment.py b/test/credential/test_credential_attachment.py new file mode 100644 index 00000000..99d65dad --- /dev/null +++ b/test/credential/test_credential_attachment.py @@ -0,0 +1,220 @@ +import json +import unittest +from unittest import mock + +import requests + +from pyoaev import OpenAEV +from pyoaev.credential import ( + AwsAccessKeySecret, + CredentialAttachment, + CredentialErrorCode, + CredentialResolutionError, + CredentialType, + get_credential_attachment, + resolve_inject_credential, +) + +INJECT_ID = "inject-id" +REFERENCE = "credential-reference-id" +AUTHORISATION_CODE = "authorisation-code" +SECRET = "super-secret-value" + +AWS_PAYLOAD = { + "type": "AWS_ACCESS_KEY", + "value": { + "aws_default_region": "eu-west-3", + "aws_access_key_id": "AKIAEXAMPLE", + "aws_secret_access_key": SECRET, + }, +} + + +def _message(attachments=mock.sentinel.missing): + data = {"injection": {"inject_id": INJECT_ID, "inject_content": {}}} + if attachments is not mock.sentinel.missing: + data["attachments"] = attachments + return data + + +def _attachments(references=(REFERENCE,), authorisation_code=AUTHORISATION_CODE): + return { + "credential_references": list(references), + "authorisation_code": authorisation_code, + } + + +class GetCredentialAttachmentTest(unittest.TestCase): + def test_no_attachments_is_legacy_path(self): + self.assertIsNone(get_credential_attachment(_message())) + + def test_null_attachments_is_legacy_path(self): + self.assertIsNone(get_credential_attachment(_message(None))) + + def test_empty_references_is_legacy_path(self): + self.assertIsNone(get_credential_attachment(_message(_attachments([])))) + + def test_null_references_is_legacy_path(self): + self.assertIsNone( + get_credential_attachment( + _message({"credential_references": None, "authorisation_code": "c"}) + ) + ) + + def test_single_reference(self): + attachment = get_credential_attachment(_message(_attachments())) + + self.assertEqual( + attachment, CredentialAttachment(REFERENCE, AUTHORISATION_CODE) + ) + + def test_several_references_uses_the_first_one(self): + attachment = get_credential_attachment( + _message(_attachments([REFERENCE, "second-reference"])) + ) + + self.assertEqual(attachment.reference, REFERENCE) + + def test_missing_authorisation_code_is_access_denied(self): + for authorisation_code in (None, "", " "): + with self.subTest(authorisation_code=authorisation_code): + with self.assertRaises(CredentialResolutionError) as context: + get_credential_attachment( + _message(_attachments(authorisation_code=authorisation_code)) + ) + + self.assertEqual( + context.exception.code, + CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, + ) + self.assertEqual(context.exception.reference, REFERENCE) + + def test_authorisation_code_key_absent_is_access_denied(self): + with self.assertRaises(CredentialResolutionError) as context: + get_credential_attachment(_message({"credential_references": [REFERENCE]})) + + self.assertEqual( + context.exception.code, CredentialErrorCode.CREDENTIAL_ACCESS_DENIED + ) + + def test_malformed_attachments_is_access_denied(self): + for attachments in ( + "not a dict", + ["not", "a", "dict"], + _attachments(references=[None]), + _attachments(references=[""]), + {"credential_references": REFERENCE, "authorisation_code": "c"}, + ): + with self.subTest(attachments=attachments): + with self.assertRaises(CredentialResolutionError) as context: + get_credential_attachment(_message(attachments)) + + self.assertEqual( + context.exception.code, + CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, + ) + + def test_repr_never_exposes_the_authorisation_code(self): + attachment = get_credential_attachment(_message(_attachments())) + + self.assertNotIn(AUTHORISATION_CODE, repr(attachment)) + self.assertNotIn(AUTHORISATION_CODE, str(attachment)) + + +class ResolveInjectCredentialTest(unittest.TestCase): + def setUp(self): + self.api = mock.MagicMock() + self.api.inject.resolve_attachment_secret.return_value = AWS_PAYLOAD + + def test_no_reference_sends_no_resolution_request(self): + for data in (_message(), _message(None), _message(_attachments([]))): + with self.subTest(data=data): + self.assertIsNone( + resolve_inject_credential( + self.api, INJECT_ID, data, CredentialType.CLOUD_AWS + ) + ) + + self.api.inject.resolve_attachment_secret.assert_not_called() + + def test_reference_is_resolved_parsed_and_checked(self): + resolved = resolve_inject_credential( + self.api, INJECT_ID, _message(_attachments()), CredentialType.CLOUD_AWS + ) + + self.api.inject.resolve_attachment_secret.assert_called_once_with( + INJECT_ID, REFERENCE, AUTHORISATION_CODE + ) + self.assertIsInstance(resolved, AwsAccessKeySecret) + self.assertEqual(resolved.aws_secret_access_key, SECRET) + + def test_expected_type_can_be_a_provider_name(self): + resolved = resolve_inject_credential( + self.api, INJECT_ID, _message(_attachments()), "aws" + ) + + self.assertIsInstance(resolved, AwsAccessKeySecret) + + def test_incompatible_credential(self): + with self.assertRaises(CredentialResolutionError) as context: + resolve_inject_credential( + self.api, INJECT_ID, _message(_attachments()), CredentialType.CLOUD_GCP + ) + + self.assertEqual( + context.exception.code, CredentialErrorCode.CREDENTIAL_INCOMPATIBLE + ) + self.assertEqual(context.exception.reference, REFERENCE) + + def test_missing_authorisation_code_sends_no_resolution_request(self): + with self.assertRaises(CredentialResolutionError): + resolve_inject_credential( + self.api, + INJECT_ID, + _message(_attachments(authorisation_code=None)), + CredentialType.CLOUD_AWS, + ) + + self.api.inject.resolve_attachment_secret.assert_not_called() + + def test_platform_error_is_propagated(self): + self.api.inject.resolve_attachment_secret.side_effect = ( + CredentialResolutionError( + CredentialErrorCode.CREDENTIAL_INACTIVE, REFERENCE, 400 + ) + ) + + with self.assertRaises(CredentialResolutionError) as context: + resolve_inject_credential( + self.api, INJECT_ID, _message(_attachments()), CredentialType.CLOUD_AWS + ) + + self.assertEqual( + context.exception.code, CredentialErrorCode.CREDENTIAL_INACTIVE + ) + + def test_end_to_end_with_the_openaev_client(self): + response = requests.Response() + response.status_code = 200 + response._content = json.dumps(AWS_PAYLOAD).encode() + response.headers["Content-Type"] = "application/json" + api = OpenAEV("url", "token") + + with mock.patch( + "requests.Session.request", return_value=response + ) as mock_request: + resolved = resolve_inject_credential( + api, INJECT_ID, _message(_attachments()), CredentialType.CLOUD_AWS + ) + + call = mock_request.call_args.kwargs + self.assertEqual(call["url"], f"url/api/injects/{INJECT_ID}/attachment/secret") + self.assertEqual( + call["json"], + {"attachment_id": REFERENCE, "authorisation": AUTHORISATION_CODE}, + ) + self.assertEqual(resolved.aws_access_key_id, "AKIAEXAMPLE") + + +if __name__ == "__main__": + unittest.main() From 9a974d349b4fabaa3c9fb01fa68417fe06a079ae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabriel=20PEZ=C3=89?= Date: Thu, 8 Oct 2026 10:16:09 +0200 Subject: [PATCH 5/8] feat(credential): chk 4 - generic creds materializer (#7882) --- pyoaev/credential/__init__.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pyoaev/credential/__init__.py b/pyoaev/credential/__init__.py index a221f711..7f8595f7 100644 --- a/pyoaev/credential/__init__.py +++ b/pyoaev/credential/__init__.py @@ -10,6 +10,7 @@ UnsupportedSecretTypeError, credential_error_code_from_http, ) +from .materializer import CredentialCleanupError, MaterializedCredential, materialize from .resolved import ( AwsAccessKeySecret, AwsAssumeRoleSecret, @@ -41,6 +42,7 @@ def build_single_referenced_credential_element(provider_name: str): "AzureManagedIdentitySecret", "AzureServicePrincipalSecret", "CredentialAttachment", + "CredentialCleanupError", "CredentialErrorCode", "CredentialResolutionError", "CredentialType", @@ -49,6 +51,7 @@ def build_single_referenced_credential_element(provider_name: str): "HashAlgorithm", "HashSecret", "InvalidResolvedSecretError", + "MaterializedCredential", "ResolvedSecret", "SecretType", "UnsupportedSecretTypeError", @@ -57,6 +60,7 @@ def build_single_referenced_credential_element(provider_name: str): "build_single_referenced_credential_element", "ensure_compatible", "get_credential_attachment", + "materialize", "parse_resolved_secret", "resolve_inject_credential", ] From 6057a64fdc74c6b0577dd322ff7e502028f18fd1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabriel=20PEZ=C3=89?= Date: Thu, 8 Oct 2026 16:23:27 +0200 Subject: [PATCH 6/8] feat(credential): chk 4 - refactor after tests (#7882) --- pyoaev/credential/errors.py | 27 +- pyoaev/credential/materializer.py | 281 +++++++++++++ pyoaev/credential/resolved.py | 7 +- test/credential/test_credential_attachment.py | 1 + test/credential/test_credential_errors.py | 27 +- .../test_credential_materializer.py | 386 ++++++++++++++++++ test/credential/test_credential_resolved.py | 4 + 7 files changed, 727 insertions(+), 6 deletions(-) create mode 100644 pyoaev/credential/materializer.py create mode 100644 test/credential/test_credential_materializer.py diff --git a/pyoaev/credential/errors.py b/pyoaev/credential/errors.py index c8e60a64..0a9ccba2 100644 --- a/pyoaev/credential/errors.py +++ b/pyoaev/credential/errors.py @@ -11,6 +11,7 @@ from enum import Enum from typing import Optional +from pyoaev.credential.types import CredentialType from pyoaev.exceptions import OpenAEVError @@ -44,10 +45,17 @@ class CredentialErrorCode(str, Enum): ), CredentialErrorCode.CREDENTIAL_INCOMPATIBLE: ( "The credential {reference} is not compatible with this inject. Select a " - "credential of the type expected by the inject, then run it again." + "credential of type {type} on the inject, then run it again." ), } +# Used when the expected credential type is not known where the error is raised +# (for example an unknown secret type returned by the platform). +_CREDENTIAL_INCOMPATIBLE_UNKNOWN_TYPE_MESSAGE = ( + "The credential {reference} is not compatible with this inject. Select a " + "credential of the type expected by the inject, then run it again." +) + def credential_error_code_from_http( response_code: Optional[int], error_message: Optional[str] @@ -76,6 +84,8 @@ class CredentialResolutionError(OpenAEVError): ``message`` holds the fixed human-readable message of ``code``. The HTTP response body is never stored, so it cannot surface in ``str()``. + ``expected_type`` is the credential type the inject expects, only used by + the ``CREDENTIAL_INCOMPATIBLE`` message. """ def __init__( @@ -83,11 +93,22 @@ def __init__( code: CredentialErrorCode, reference: Optional[str] = None, response_code: Optional[int] = None, + expected_type: Optional[CredentialType] = None, ) -> None: self.code = CredentialErrorCode(code) self.reference = reference - message = _CREDENTIAL_ERROR_MESSAGES[self.code].format( - reference=reference or "unknown reference" + self.expected_type = ( + CredentialType(expected_type) if expected_type is not None else None + ) + template = _CREDENTIAL_ERROR_MESSAGES[self.code] + if ( + self.code == CredentialErrorCode.CREDENTIAL_INCOMPATIBLE + and self.expected_type is None + ): + template = _CREDENTIAL_INCOMPATIBLE_UNKNOWN_TYPE_MESSAGE + message = template.format( + reference=reference or "unknown reference", + type=self.expected_type.value if self.expected_type else None, ) super().__init__( error_message=message, response_code=response_code, response_body=None diff --git a/pyoaev/credential/materializer.py b/pyoaev/credential/materializer.py new file mode 100644 index 00000000..7b4216fe --- /dev/null +++ b/pyoaev/credential/materializer.py @@ -0,0 +1,281 @@ +"""Turn a resolved secret into the environment and files a cloud SDK expects. + +``materialize`` is a context manager: the files it writes live in a private, +per-run temporary directory (``0700``, each file ``0600`` on POSIX) which is +removed on exit, even when the execution fails. Nothing is written to ``~/.aws`` +or any other shared location. +""" + +import json +import os +import shutil +import tempfile +from contextlib import contextmanager +from dataclasses import dataclass, field +from pathlib import Path +from typing import Callable, Dict, Iterator, List, Optional + +from pyoaev.credential.errors import ( + InvalidResolvedSecretError, + UnsupportedSecretTypeError, +) +from pyoaev.credential.resolved import ( + AwsAccessKeySecret, + AwsAssumeRoleSecret, + AzureManagedIdentitySecret, + AzureServicePrincipalSecret, + GcpOAuth2Secret, + GcpServiceAccountSecret, + ResolvedSecret, +) +from pyoaev.credential.types import AwsSourceIdentityType, SecretType + +AWS_SOURCE_PROFILE = "srt-source" +AWS_TARGET_PROFILE = "srt-target" + +# Scheme and trailing slash are required by the Go SDK (stratus) and accepted +# by the Python SDK. +AZURE_AUTHORITY_HOSTS = { + "AzureCloud": "https://login.microsoftonline.com/", + "AzureChinaCloud": "https://login.chinacloudapi.cn/", + "AzureUSGovernment": "https://login.microsoftonline.us/", +} + + +class CredentialCleanupError(RuntimeError): + """Temporary credential files could not be removed.""" + + def __init__(self) -> None: + super().__init__("temporary credential cleanup failed") + + +@dataclass(frozen=True) +class MaterializedCredential: + """Environment and files to apply to the process using the credential. + + ``secret_type`` lets callers pick the authentication mode (for example a + CLI flag) without re-reading the secret. + """ + + secret_type: SecretType + env: Dict[str, str] = field(repr=False) + files: List[Path] + + +class _Workspace: + """Private per-run directory, only created when a file must be written.""" + + def __init__(self, temporary_root: Optional[Path]) -> None: + self._temporary_root = temporary_root + self.directory: Optional[Path] = None + self.files: List[Path] = [] + + def write(self, content: bytes, *, prefix: str, suffix: str = "") -> Path: + if self.directory is None: + # mkdtemp creates the directory 0700 on POSIX. + self.directory = Path( + tempfile.mkdtemp(prefix="openaev-credential-", dir=self._temporary_root) + ) + # mkstemp creates the file 0600 on POSIX. + descriptor, name = tempfile.mkstemp( + prefix=prefix, suffix=suffix, dir=self.directory + ) + path = Path(name) + self.files.append(path) + try: + credential_file = os.fdopen(descriptor, "wb") + except BaseException: + # An open descriptor would prevent the removal on Windows. + os.close(descriptor) + raise + with credential_file: + credential_file.write(content) + return path + + def cleanup(self) -> None: + if self.directory is None: + return + shutil.rmtree(self.directory, ignore_errors=True) + if self.directory.exists(): + raise CredentialCleanupError() + + +def _ini_value(value: str) -> str: + # A line break would let a value inject keys or sections in the file. + if "\n" in value or "\r" in value: + raise InvalidResolvedSecretError() + return value + + +def _ini(section: str, values: Dict[str, Optional[str]]) -> bytes: + lines = [f"[{section}]"] + lines += [f"{key} = {_ini_value(value)}" for key, value in values.items() if value] + return ("\n".join(lines) + "\n").encode() + + +def _aws_region_env(region: Optional[str]) -> Dict[str, str]: + if not region: + return {} + return {"AWS_REGION": region, "AWS_DEFAULT_REGION": region} + + +def _aws_access_key( + secret: AwsAccessKeySecret, workspace: _Workspace +) -> Dict[str, str]: + env = { + **_aws_region_env(secret.aws_default_region), + "AWS_ACCESS_KEY_ID": secret.aws_access_key_id, + "AWS_SECRET_ACCESS_KEY": secret.aws_secret_access_key, + } + if secret.aws_session_token: + env["AWS_SESSION_TOKEN"] = secret.aws_session_token + return env + + +def _aws_assume_role( + secret: AwsAssumeRoleSecret, workspace: _Workspace +) -> Dict[str, str]: + env = {"AWS_PROFILE": AWS_TARGET_PROFILE} + profile: Dict[str, Optional[str]] = {"role_arn": secret.aws_role_arn} + if secret.aws_source_identity_type == AwsSourceIdentityType.INSTANCE_DEFAULT: + profile["credential_source"] = "Ec2InstanceMetadata" + else: + credentials = workspace.write( + _ini( + AWS_SOURCE_PROFILE, + { + "aws_access_key_id": secret.aws_source_profile_access_key_id, + "aws_secret_access_key": ( + secret.aws_source_profile_secret_access_key + ), + }, + ), + prefix="aws-credentials-", + ) + env["AWS_SHARED_CREDENTIALS_FILE"] = str(credentials) + profile["source_profile"] = AWS_SOURCE_PROFILE + profile["external_id"] = secret.aws_external_id + profile["region"] = secret.aws_default_region + config = workspace.write( + _ini(f"profile {AWS_TARGET_PROFILE}", profile), prefix="aws-config-" + ) + env["AWS_CONFIG_FILE"] = str(config) + return env + + +def _azure_authority_host(environment: str) -> str: + try: + return AZURE_AUTHORITY_HOSTS[environment] + except KeyError: + raise InvalidResolvedSecretError() from None + + +def _azure_service_principal( + secret: AzureServicePrincipalSecret, workspace: _Workspace +) -> Dict[str, str]: + env = { + "AZURE_CLIENT_ID": secret.azure_client_id, + "AZURE_CLIENT_SECRET": secret.azure_client_secret, + "AZURE_TENANT_ID": secret.azure_tenant_id, + "AZURE_AUTHORITY_HOST": _azure_authority_host(secret.azure_environment), + } + if secret.azure_subscription_id: + env["AZURE_SUBSCRIPTION_ID"] = secret.azure_subscription_id + return env + + +def _azure_managed_identity( + secret: AzureManagedIdentitySecret, workspace: _Workspace +) -> Dict[str, str]: + env = {"AZURE_AUTHORITY_HOST": _azure_authority_host(secret.azure_environment)} + # Without a client id, the system-assigned identity is used. + if secret.azure_client_id: + env["AZURE_CLIENT_ID"] = secret.azure_client_id + if secret.azure_subscription_id: + env["AZURE_SUBSCRIPTION_ID"] = secret.azure_subscription_id + return env + + +def _gcp_env( + credentials_file: Path, scope: str, project_id: Optional[str] +) -> Dict[str, str]: + env = { + "GOOGLE_APPLICATION_CREDENTIALS": str(credentials_file), + "GOOGLE_CLOUD_UNIVERSE_DOMAIN": scope, + } + if project_id: + env["GOOGLE_PROJECT"] = project_id + return env + + +def _gcp_service_account( + secret: GcpServiceAccountSecret, workspace: _Workspace +) -> Dict[str, str]: + credentials_file = workspace.write( + secret.gcp_private_key_json, prefix="gcp-credentials-", suffix=".json" + ) + return _gcp_env(credentials_file, secret.gcp_scope, secret.gcp_project_id) + + +def _gcp_oauth2(secret: GcpOAuth2Secret, workspace: _Workspace) -> Dict[str, str]: + authorized_user = { + "type": "authorized_user", + "client_id": secret.gcp_oauth_client_id, + "client_secret": secret.gcp_oauth_client_secret, + "refresh_token": secret.gcp_oauth_refresh_token, + "universe_domain": secret.gcp_scope, + } + credentials_file = workspace.write( + json.dumps(authorized_user).encode(), + prefix="gcp-credentials-", + suffix=".json", + ) + return _gcp_env(credentials_file, secret.gcp_scope, secret.gcp_project_id) + + +_MATERIALIZERS: Dict[SecretType, Callable[..., Dict[str, str]]] = { + SecretType.AWS_ACCESS_KEY: _aws_access_key, + SecretType.AWS_ASSUME_ROLE: _aws_assume_role, + SecretType.AZURE_SERVICE_PRINCIPAL: _azure_service_principal, + SecretType.AZURE_MANAGED_IDENTITY: _azure_managed_identity, + SecretType.GCP_SERVICE_ACCOUNT: _gcp_service_account, + SecretType.GCP_OAUTH2: _gcp_oauth2, +} + + +@contextmanager +def materialize( + resolved: ResolvedSecret, temporary_root: Optional[Path] = None +) -> Iterator[MaterializedCredential]: + """Materialize ``resolved`` for the duration of the ``with`` block. + + Args: + resolved: the secret returned by ``resolve_inject_credential``. + temporary_root: parent of the private directory, the system temporary + directory when omitted. + + Raises: + UnsupportedSecretTypeError: the secret type is not a cloud credential + (``USERNAME_PASSWORD``, ``HASH``). + InvalidResolvedSecretError: a value cannot be materialized safely. + CredentialCleanupError: the temporary files could not be removed. + """ + materializer = _MATERIALIZERS.get(resolved.secret_type) + if materializer is None: + raise UnsupportedSecretTypeError(resolved.secret_type.value) + workspace = _Workspace(temporary_root) + try: + env = materializer(resolved, workspace) + yield MaterializedCredential( + secret_type=resolved.secret_type, env=env, files=list(workspace.files) + ) + finally: + workspace.cleanup() + + +__all__ = [ + "AZURE_AUTHORITY_HOSTS", + "CredentialCleanupError", + "MaterializedCredential", + "materialize", +] diff --git a/pyoaev/credential/resolved.py b/pyoaev/credential/resolved.py index 714ce343..b72987cb 100644 --- a/pyoaev/credential/resolved.py +++ b/pyoaev/credential/resolved.py @@ -212,7 +212,7 @@ def parse_resolved_secret( # quote a secret value, is not chained to it. try: return _build(_RESOLVED_SECRET_CLASSES[secret_type], value) - except (ValueError, TypeError): + except (ValueError, TypeError) as e: error = InvalidResolvedSecretError(reference) raise error @@ -241,9 +241,12 @@ def ensure_compatible( expected_type = _resolve_credential_type(expected) if expected_type is None: raise ValueError(f"No credential type is mapped to provider {expected!r}") + if resolved.credential_type != expected_type: raise CredentialResolutionError( - CredentialErrorCode.CREDENTIAL_INCOMPATIBLE, reference + CredentialErrorCode.CREDENTIAL_INCOMPATIBLE, + reference, + expected_type=expected_type, ) return resolved diff --git a/test/credential/test_credential_attachment.py b/test/credential/test_credential_attachment.py index 99d65dad..c0750707 100644 --- a/test/credential/test_credential_attachment.py +++ b/test/credential/test_credential_attachment.py @@ -165,6 +165,7 @@ def test_incompatible_credential(self): context.exception.code, CredentialErrorCode.CREDENTIAL_INCOMPATIBLE ) self.assertEqual(context.exception.reference, REFERENCE) + self.assertIn("CLOUD_GCP", context.exception.message) def test_missing_authorisation_code_sends_no_resolution_request(self): with self.assertRaises(CredentialResolutionError): diff --git a/test/credential/test_credential_errors.py b/test/credential/test_credential_errors.py index 50240225..032192ca 100644 --- a/test/credential/test_credential_errors.py +++ b/test/credential/test_credential_errors.py @@ -3,6 +3,7 @@ from pyoaev.credential import ( CredentialErrorCode, CredentialResolutionError, + CredentialType, credential_error_code_from_http, ) from pyoaev.exceptions import OpenAEVError @@ -100,17 +101,41 @@ def test_access_denied_message_discloses_nothing(self): "inject. Contact your Cloud platform administrator", ) - def test_incompatible_message_identifies_the_reference(self): + def test_incompatible_message_identifies_the_expected_type(self): + error = CredentialResolutionError( + CredentialErrorCode.CREDENTIAL_INCOMPATIBLE, + reference="ref-1", + expected_type=CredentialType.CLOUD_AWS, + ) + + self.assertIs(error.expected_type, CredentialType.CLOUD_AWS) + self.assertEqual( + error.message, + "The credential ref-1 is not compatible with this inject. Select a " + "credential of type CLOUD_AWS on the inject, then run it again.", + ) + + def test_incompatible_message_without_expected_type(self): error = CredentialResolutionError( CredentialErrorCode.CREDENTIAL_INCOMPATIBLE, reference="ref-1" ) + self.assertIsNone(error.expected_type) self.assertEqual( error.message, "The credential ref-1 is not compatible with this inject. Select a " "credential of the type expected by the inject, then run it again.", ) + def test_expected_type_is_ignored_by_other_codes(self): + error = CredentialResolutionError( + CredentialErrorCode.CREDENTIAL_INACTIVE, + reference="ref-1", + expected_type=CredentialType.CLOUD_AWS, + ) + + self.assertNotIn("CLOUD_AWS", error.message) + def test_str_carries_the_code_and_the_message(self): error = CredentialResolutionError( CredentialErrorCode.CREDENTIAL_INACTIVE, diff --git a/test/credential/test_credential_materializer.py b/test/credential/test_credential_materializer.py new file mode 100644 index 00000000..ca0a0965 --- /dev/null +++ b/test/credential/test_credential_materializer.py @@ -0,0 +1,386 @@ +import configparser +import json +import os +import stat +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +from pyoaev.credential import ( + AwsAccessKeySecret, + AwsAssumeRoleSecret, + AwsSourceIdentityType, + AzureManagedIdentitySecret, + AzureServicePrincipalSecret, + CredentialCleanupError, + GcpOAuth2Secret, + GcpServiceAccountSecret, + HashAlgorithm, + HashSecret, + InvalidResolvedSecretError, + SecretType, + UnsupportedSecretTypeError, + UsernamePasswordSecret, + materialize, +) + +SECRET = "super-secret-value" +GCP_KEY_JSON = b'{"type": "service_account", "private_key": "super-secret-value"}' + +AWS_ACCESS_KEY = AwsAccessKeySecret( + aws_access_key_id="AKIAEXAMPLE", + aws_secret_access_key=SECRET, + aws_default_region="eu-west-3", + aws_session_token="session-token", +) +AWS_ASSUME_ROLE_STATIC = AwsAssumeRoleSecret( + aws_role_arn="arn:aws:iam::123456789012:role/srt", + aws_source_identity_type=AwsSourceIdentityType.STATIC_ACCESS_KEY, + aws_default_region="us-east-1", + aws_external_id="external-id", + aws_source_profile_access_key_id="AKIASOURCE", + aws_source_profile_secret_access_key=SECRET, +) +AWS_ASSUME_ROLE_INSTANCE = AwsAssumeRoleSecret( + aws_role_arn="arn:aws:iam::123456789012:role/srt", + aws_source_identity_type=AwsSourceIdentityType.INSTANCE_DEFAULT, +) +AZURE_SERVICE_PRINCIPAL = AzureServicePrincipalSecret( + azure_environment="AzureCloud", + azure_client_id="client-id", + azure_client_secret=SECRET, + azure_tenant_id="tenant-id", + azure_subscription_id="subscription-id", +) +AZURE_MANAGED_IDENTITY = AzureManagedIdentitySecret( + azure_environment="AzureUSGovernment", + azure_client_id="client-id", + azure_subscription_id="subscription-id", +) +GCP_SERVICE_ACCOUNT = GcpServiceAccountSecret( + gcp_scope="googleapis.com", + gcp_private_key_json=GCP_KEY_JSON, + gcp_project_id="project-id", +) +GCP_OAUTH2 = GcpOAuth2Secret( + gcp_scope="googleapis.com", + gcp_oauth_client_id="client-id", + gcp_oauth_client_secret=SECRET, + gcp_oauth_refresh_token="refresh-token", + gcp_project_id="project-id", +) + +ALL_CLOUD_SECRETS = ( + AWS_ACCESS_KEY, + AWS_ASSUME_ROLE_STATIC, + AWS_ASSUME_ROLE_INSTANCE, + AZURE_SERVICE_PRINCIPAL, + AZURE_MANAGED_IDENTITY, + GCP_SERVICE_ACCOUNT, + GCP_OAUTH2, +) +SECRETS_WITH_FILES = ( + AWS_ASSUME_ROLE_STATIC, + AWS_ASSUME_ROLE_INSTANCE, + GCP_SERVICE_ACCOUNT, + GCP_OAUTH2, +) + + +def _read_ini(path: str) -> configparser.ConfigParser: + parser = configparser.ConfigParser() + parser.read(path, encoding="utf-8") + return parser + + +class MaterializerTestCase(unittest.TestCase): + def setUp(self): + temporary_directory = tempfile.TemporaryDirectory() + self.addCleanup(temporary_directory.cleanup) + self.root = Path(temporary_directory.name) + + def materialize(self, resolved): + return materialize(resolved, temporary_root=self.root) + + def assert_root_is_empty(self): + self.assertEqual(list(self.root.iterdir()), []) + + +class AwsMaterializerTest(MaterializerTestCase): + def test_access_key(self): + with self.materialize(AWS_ACCESS_KEY) as credential: + self.assertEqual(credential.secret_type, SecretType.AWS_ACCESS_KEY) + self.assertEqual( + credential.env, + { + "AWS_REGION": "eu-west-3", + "AWS_DEFAULT_REGION": "eu-west-3", + "AWS_ACCESS_KEY_ID": "AKIAEXAMPLE", + "AWS_SECRET_ACCESS_KEY": SECRET, + "AWS_SESSION_TOKEN": "session-token", + }, + ) + self.assertEqual(credential.files, []) + self.assert_root_is_empty() + + def test_access_key_without_region_nor_session_token(self): + resolved = AwsAccessKeySecret( + aws_access_key_id="AKIAEXAMPLE", aws_secret_access_key=SECRET + ) + + with self.materialize(resolved) as credential: + self.assertEqual( + credential.env, + { + "AWS_ACCESS_KEY_ID": "AKIAEXAMPLE", + "AWS_SECRET_ACCESS_KEY": SECRET, + }, + ) + + def test_assume_role_with_static_access_key(self): + with self.materialize(AWS_ASSUME_ROLE_STATIC) as credential: + env = credential.env + self.assertEqual(credential.secret_type, SecretType.AWS_ASSUME_ROLE) + self.assertEqual( + set(env), + {"AWS_PROFILE", "AWS_SHARED_CREDENTIALS_FILE", "AWS_CONFIG_FILE"}, + ) + self.assertEqual(env["AWS_PROFILE"], "srt-target") + self.assertEqual( + credential.files, + [ + Path(env["AWS_SHARED_CREDENTIALS_FILE"]), + Path(env["AWS_CONFIG_FILE"]), + ], + ) + + credentials = _read_ini(env["AWS_SHARED_CREDENTIALS_FILE"]) + self.assertEqual(credentials.sections(), ["srt-source"]) + self.assertEqual( + dict(credentials["srt-source"]), + { + "aws_access_key_id": "AKIASOURCE", + "aws_secret_access_key": SECRET, + }, + ) + + config = _read_ini(env["AWS_CONFIG_FILE"]) + self.assertEqual(config.sections(), ["profile srt-target"]) + self.assertEqual( + dict(config["profile srt-target"]), + { + "role_arn": "arn:aws:iam::123456789012:role/srt", + "source_profile": "srt-source", + "external_id": "external-id", + "region": "us-east-1", + }, + ) + + def test_assume_role_with_instance_default(self): + with self.materialize(AWS_ASSUME_ROLE_INSTANCE) as credential: + env = credential.env + self.assertEqual(set(env), {"AWS_PROFILE", "AWS_CONFIG_FILE"}) + self.assertEqual(credential.files, [Path(env["AWS_CONFIG_FILE"])]) + + config = _read_ini(env["AWS_CONFIG_FILE"]) + self.assertEqual( + dict(config["profile srt-target"]), + { + "role_arn": "arn:aws:iam::123456789012:role/srt", + "credential_source": "Ec2InstanceMetadata", + }, + ) + + def test_line_break_in_a_file_value_is_rejected(self): + resolved = AwsAssumeRoleSecret( + aws_role_arn="arn:aws:iam::123456789012:role/srt\n[profile other]", + aws_source_identity_type=AwsSourceIdentityType.INSTANCE_DEFAULT, + ) + + with self.assertRaises(InvalidResolvedSecretError): + with self.materialize(resolved): + self.fail("must not be reached") + + self.assert_root_is_empty() + + +class AzureMaterializerTest(MaterializerTestCase): + def test_service_principal(self): + with self.materialize(AZURE_SERVICE_PRINCIPAL) as credential: + self.assertEqual(credential.secret_type, SecretType.AZURE_SERVICE_PRINCIPAL) + self.assertEqual( + credential.env, + { + "AZURE_CLIENT_ID": "client-id", + "AZURE_CLIENT_SECRET": SECRET, + "AZURE_TENANT_ID": "tenant-id", + "AZURE_SUBSCRIPTION_ID": "subscription-id", + "AZURE_AUTHORITY_HOST": "https://login.microsoftonline.com/", + }, + ) + self.assertEqual(credential.files, []) + + def test_managed_identity(self): + with self.materialize(AZURE_MANAGED_IDENTITY) as credential: + self.assertEqual(credential.secret_type, SecretType.AZURE_MANAGED_IDENTITY) + self.assertEqual( + credential.env, + { + "AZURE_CLIENT_ID": "client-id", + "AZURE_SUBSCRIPTION_ID": "subscription-id", + "AZURE_AUTHORITY_HOST": "https://login.microsoftonline.us/", + }, + ) + + def test_system_assigned_managed_identity(self): + resolved = AzureManagedIdentitySecret(azure_environment="AzureChinaCloud") + + with self.materialize(resolved) as credential: + self.assertEqual( + credential.env, + {"AZURE_AUTHORITY_HOST": "https://login.chinacloudapi.cn/"}, + ) + + def test_unknown_environment_is_rejected(self): + resolved = AzureManagedIdentitySecret(azure_environment="AzureGermanCloud") + + with self.assertRaises(InvalidResolvedSecretError): + with self.materialize(resolved): + self.fail("must not be reached") + + +class GcpMaterializerTest(MaterializerTestCase): + def test_service_account(self): + with self.materialize(GCP_SERVICE_ACCOUNT) as credential: + env = credential.env + self.assertEqual(credential.secret_type, SecretType.GCP_SERVICE_ACCOUNT) + self.assertEqual( + set(env), + { + "GOOGLE_APPLICATION_CREDENTIALS", + "GOOGLE_PROJECT", + "GOOGLE_CLOUD_UNIVERSE_DOMAIN", + }, + ) + self.assertEqual(env["GOOGLE_PROJECT"], "project-id") + self.assertEqual(env["GOOGLE_CLOUD_UNIVERSE_DOMAIN"], "googleapis.com") + credentials_file = Path(env["GOOGLE_APPLICATION_CREDENTIALS"]) + self.assertEqual(credential.files, [credentials_file]) + self.assertEqual(credentials_file.suffix, ".json") + self.assertEqual(credentials_file.read_bytes(), GCP_KEY_JSON) + + def test_oauth2(self): + with self.materialize(GCP_OAUTH2) as credential: + env = credential.env + self.assertEqual(credential.secret_type, SecretType.GCP_OAUTH2) + self.assertEqual(env["GOOGLE_PROJECT"], "project-id") + self.assertEqual(env["GOOGLE_CLOUD_UNIVERSE_DOMAIN"], "googleapis.com") + credentials_file = Path(env["GOOGLE_APPLICATION_CREDENTIALS"]) + self.assertEqual(credential.files, [credentials_file]) + self.assertEqual( + json.loads(credentials_file.read_text(encoding="utf-8")), + { + "type": "authorized_user", + "client_id": "client-id", + "client_secret": SECRET, + "refresh_token": "refresh-token", + "universe_domain": "googleapis.com", + }, + ) + + def test_without_project_id(self): + resolved = GcpOAuth2Secret( + gcp_scope="googleapis.com", + gcp_oauth_client_id="client-id", + gcp_oauth_client_secret=SECRET, + gcp_oauth_refresh_token="refresh-token", + ) + + with self.materialize(resolved) as credential: + self.assertNotIn("GOOGLE_PROJECT", credential.env) + + +class MaterializerLifecycleTest(MaterializerTestCase): + def test_identity_secrets_are_not_supported(self): + for resolved in ( + UsernamePasswordSecret("alice", SECRET), + HashSecret(HashAlgorithm.NTLM, SECRET), + ): + with self.subTest(secret_type=resolved.secret_type): + with self.assertRaises(UnsupportedSecretTypeError): + with self.materialize(resolved): + self.fail("must not be reached") + + def test_files_are_written_in_a_private_directory_under_the_root(self): + for resolved in SECRETS_WITH_FILES: + with self.subTest(secret_type=resolved.secret_type): + with self.materialize(resolved) as credential: + directories = list(self.root.iterdir()) + self.assertEqual(len(directories), 1) + for path in credential.files: + self.assertEqual(path.parent, directories[0]) + + @unittest.skipIf(os.name == "nt", "POSIX permissions only") + def test_posix_permissions(self): + for resolved in SECRETS_WITH_FILES: + with self.subTest(secret_type=resolved.secret_type): + with self.materialize(resolved) as credential: + directory = credential.files[0].parent + self.assertEqual(stat.S_IMODE(directory.stat().st_mode), 0o700) + for path in credential.files: + self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600) + + def test_cleanup_on_success(self): + for resolved in ALL_CLOUD_SECRETS: + with self.subTest(secret_type=resolved.secret_type): + with self.materialize(resolved) as credential: + files = credential.files + + for path in files: + self.assertFalse(path.exists()) + self.assert_root_is_empty() + + def test_cleanup_on_exception(self): + for resolved in ALL_CLOUD_SECRETS: + with self.subTest(secret_type=resolved.secret_type): + with self.assertRaises(RuntimeError): + with self.materialize(resolved) as credential: + files = credential.files + raise RuntimeError("execution failed") + + for path in files: + self.assertFalse(path.exists()) + self.assert_root_is_empty() + + def test_cleanup_when_a_write_fails(self): + with mock.patch( + "pyoaev.credential.materializer.os.fdopen", side_effect=OSError("disk full") + ): + with self.assertRaises(OSError): + with self.materialize(GCP_SERVICE_ACCOUNT): + self.fail("must not be reached") + + self.assert_root_is_empty() + + def test_cleanup_failure_is_reported_without_details(self): + with mock.patch("pyoaev.credential.materializer.shutil.rmtree"): + with self.assertRaises(CredentialCleanupError) as context: + with self.materialize(GCP_SERVICE_ACCOUNT) as credential: + directory = credential.files[0].parent + + self.assertEqual(str(context.exception), "temporary credential cleanup failed") + # Remove what the mocked cleanup left behind. + for path in directory.iterdir(): + path.unlink() + directory.rmdir() + + def test_repr_never_exposes_secret_material(self): + for resolved in ALL_CLOUD_SECRETS: + with self.subTest(secret_type=resolved.secret_type): + with self.materialize(resolved) as credential: + self.assertNotIn(SECRET, repr(credential)) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/credential/test_credential_resolved.py b/test/credential/test_credential_resolved.py index 69d70494..cb4562d7 100644 --- a/test/credential/test_credential_resolved.py +++ b/test/credential/test_credential_resolved.py @@ -362,6 +362,8 @@ def test_mismatching_credential_type_is_incompatible(self): error.code, CredentialErrorCode.CREDENTIAL_INCOMPATIBLE ) self.assertIn(REFERENCE, error.message) + self.assertIs(error.expected_type, expected) + self.assertIn(expected.value, error.message) self.assertNotIn(SECRET, str(error)) def test_provider_name_uses_the_contract_mapping(self): @@ -383,6 +385,8 @@ def test_provider_name_mismatch_is_incompatible(self): self.assertEqual( context.exception.code, CredentialErrorCode.CREDENTIAL_INCOMPATIBLE ) + self.assertIs(context.exception.expected_type, CredentialType.CLOUD_AWS) + self.assertIn("CLOUD_AWS", context.exception.message) def test_unknown_provider_name_is_a_programming_error(self): with self.assertRaises(ValueError): From af95d81e0c40b69fc692656fe517b104e6655d8d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabriel=20PEZ=C3=89?= Date: Fri, 9 Oct 2026 15:15:32 +0200 Subject: [PATCH 7/8] feat(credential): fix pr issue (#7882) --- pyoaev/credential/resolved.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyoaev/credential/resolved.py b/pyoaev/credential/resolved.py index b72987cb..59e26d2b 100644 --- a/pyoaev/credential/resolved.py +++ b/pyoaev/credential/resolved.py @@ -212,7 +212,7 @@ def parse_resolved_secret( # quote a secret value, is not chained to it. try: return _build(_RESOLVED_SECRET_CLASSES[secret_type], value) - except (ValueError, TypeError) as e: + except (ValueError, TypeError): error = InvalidResolvedSecretError(reference) raise error From 072ab487cb5cc53c6d2be37bbc87372b4c35d763 Mon Sep 17 00:00:00 2001 From: Christophe Melchior Date: Wed, 14 Oct 2026 11:38:25 +0200 Subject: [PATCH 8/8] feat(credential): split error summary and guidance, map platform failures to unavailable (#7882) --- pyoaev/apis/inject.py | 3 +- pyoaev/credential/__init__.py | 6 + pyoaev/credential/attachment.py | 22 +++- pyoaev/credential/errors.py | 111 +++++++++++++----- pyoaev/credential/materializer.py | 34 +++++- test/apis/inject/test_inject.py | 18 ++- test/credential/test_credential_attachment.py | 29 +++++ test/credential/test_credential_errors.py | 63 +++++++++- .../test_credential_materializer.py | 15 ++- 9 files changed, 256 insertions(+), 45 deletions(-) diff --git a/pyoaev/apis/inject.py b/pyoaev/apis/inject.py index 2f4f21c4..829e8e05 100644 --- a/pyoaev/apis/inject.py +++ b/pyoaev/apis/inject.py @@ -62,7 +62,8 @@ def resolve_attachment_secret( e.response_code, e.error_message ) except requests.RequestException: - error_code = CredentialErrorCode.CREDENTIAL_ACCESS_DENIED + # Connection error or timeout: the platform never answered. + error_code = CredentialErrorCode.CREDENTIAL_RESOLUTION_UNAVAILABLE else: if not isinstance(result, dict): error_code = CredentialErrorCode.CREDENTIAL_ACCESS_DENIED diff --git a/pyoaev/credential/__init__.py b/pyoaev/credential/__init__.py index 7f8595f7..574dfd01 100644 --- a/pyoaev/credential/__init__.py +++ b/pyoaev/credential/__init__.py @@ -1,6 +1,7 @@ from .attachment import ( CredentialAttachment, get_credential_attachment, + resolve_credential_attachment, resolve_inject_credential, ) from .errors import ( @@ -9,6 +10,8 @@ InvalidResolvedSecretError, UnsupportedSecretTypeError, credential_error_code_from_http, + credential_error_guidance, + credential_error_summary, ) from .materializer import CredentialCleanupError, MaterializedCredential, materialize from .resolved import ( @@ -57,10 +60,13 @@ def build_single_referenced_credential_element(provider_name: str): "UnsupportedSecretTypeError", "UsernamePasswordSecret", "credential_error_code_from_http", + "credential_error_guidance", + "credential_error_summary", "build_single_referenced_credential_element", "ensure_compatible", "get_credential_attachment", "materialize", "parse_resolved_secret", + "resolve_credential_attachment", "resolve_inject_credential", ] diff --git a/pyoaev/credential/attachment.py b/pyoaev/credential/attachment.py index f0425f44..791d015d 100644 --- a/pyoaev/credential/attachment.py +++ b/pyoaev/credential/attachment.py @@ -8,7 +8,9 @@ ``attachments`` is ``null`` when the inject has no reference: the injector then keeps its legacy credential fields and no resolution request is sent. -``resolve_inject_credential`` is the single entry point injectors call. +``resolve_inject_credential`` is the single entry point injectors call. An +injector that reads the attachment while validating its input and resolves it +later calls ``get_credential_attachment`` then ``resolve_credential_attachment``. """ from dataclasses import dataclass, field @@ -100,6 +102,23 @@ def resolve_inject_credential( attachment = get_credential_attachment(data) if attachment is None: return None + return resolve_credential_attachment(api, inject_id, attachment, expected_type) + + +def resolve_credential_attachment( + api: "OpenAEV", + inject_id: str, + attachment: CredentialAttachment, + expected_type: Union[CredentialType, str], +) -> ResolvedSecret: + """Resolve an attachment already read with ``get_credential_attachment``. + + Same contract as ``resolve_inject_credential``, for injectors that read + the attachment and resolve it at two different stages. + + Raises: + CredentialResolutionError: with the code to report in the trace. + """ payload = api.inject.resolve_attachment_secret( inject_id, attachment.reference, attachment.authorisation_code ) @@ -110,5 +129,6 @@ def resolve_inject_credential( __all__ = [ "CredentialAttachment", "get_credential_attachment", + "resolve_credential_attachment", "resolve_inject_credential", ] diff --git a/pyoaev/credential/errors.py b/pyoaev/credential/errors.py index 0a9ccba2..ab54f1d9 100644 --- a/pyoaev/credential/errors.py +++ b/pyoaev/credential/errors.py @@ -2,10 +2,17 @@ The platform resolution endpoint answers with a closed set of error codes. ``CREDENTIAL_INCOMPATIBLE`` is detected by the injector itself, when the -resolved credential does not satisfy the contract field. Each code carries a fixed, human-readable message meant to be reported as is -in the inject execution trace. Messages only ever mention the credential -reference id, never a secret value, and the raw HTTP request and response -bodies are deliberately never kept on the error. +resolved credential does not satisfy the contract field, and +``CREDENTIAL_RESOLUTION_UNAVAILABLE`` when the platform could not be reached or +failed to answer. Each code carries a fixed, human-readable message meant to be +reported as is in the inject execution trace. Messages only ever mention the +credential reference id, never a secret value, and the raw HTTP request and +response bodies are deliberately never kept on the error. + +A message is made of a summary (what happened) followed by a guidance (what to +do). Both parts are exposed through ``credential_error_summary`` and +``credential_error_guidance`` so injectors that report them separately reuse +the exact same texts. """ from enum import Enum @@ -20,6 +27,7 @@ class CredentialErrorCode(str, Enum): CREDENTIAL_INACTIVE = "CREDENTIAL_INACTIVE" CREDENTIAL_ACCESS_DENIED = "CREDENTIAL_ACCESS_DENIED" CREDENTIAL_INCOMPATIBLE = "CREDENTIAL_INCOMPATIBLE" + CREDENTIAL_RESOLUTION_UNAVAILABLE = "CREDENTIAL_RESOLUTION_UNAVAILABLE" _PLATFORM_CREDENTIAL_ERROR_CODES = { @@ -28,35 +36,75 @@ class CredentialErrorCode(str, Enum): CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, } +UNKNOWN_REFERENCE = "unknown reference" -_CREDENTIAL_ERROR_MESSAGES = { +_CREDENTIAL_ERROR_SUMMARIES = { CredentialErrorCode.CREDENTIAL_NOT_FOUND: ( "The credential configured on this inject ({reference}) no longer exists " - "at the moment of execution. Select an existing credential on the inject, " - "then run it again." + "at the moment of execution." ), CredentialErrorCode.CREDENTIAL_INACTIVE: ( - "The credential {reference} is inactive at the moment of execution. " - "Update or replace this credential, then run the inject again." + "The credential {reference} is inactive at the moment of execution." ), CredentialErrorCode.CREDENTIAL_ACCESS_DENIED: ( "This execution is not entitled to use the credential configured on this " - "inject. Contact your Cloud platform administrator" + "inject." ), CredentialErrorCode.CREDENTIAL_INCOMPATIBLE: ( - "The credential {reference} is not compatible with this inject. Select a " - "credential of type {type} on the inject, then run it again." + "The credential {reference} is not compatible with this inject." + ), + CredentialErrorCode.CREDENTIAL_RESOLUTION_UNAVAILABLE: ( + "The credential {reference} could not be resolved: the OpenAEV platform " + "was unavailable at the moment of execution." + ), +} + +_CREDENTIAL_ERROR_GUIDANCES = { + CredentialErrorCode.CREDENTIAL_NOT_FOUND: ( + "Select an existing credential on the inject, then run it again." + ), + CredentialErrorCode.CREDENTIAL_INACTIVE: ( + "Update or replace this credential, then run the inject again." + ), + CredentialErrorCode.CREDENTIAL_ACCESS_DENIED: ( + "Contact your Cloud platform administrator" + ), + CredentialErrorCode.CREDENTIAL_INCOMPATIBLE: ( + "Select a credential of type {type} on the inject, then run it again." + ), + CredentialErrorCode.CREDENTIAL_RESOLUTION_UNAVAILABLE: ( + "Check that the injector can reach the platform, then run the inject again." ), } # Used when the expected credential type is not known where the error is raised # (for example an unknown secret type returned by the platform). -_CREDENTIAL_INCOMPATIBLE_UNKNOWN_TYPE_MESSAGE = ( - "The credential {reference} is not compatible with this inject. Select a " - "credential of the type expected by the inject, then run it again." +_CREDENTIAL_INCOMPATIBLE_UNKNOWN_TYPE_GUIDANCE = ( + "Select a credential of the type expected by the inject, then run it again." ) +def credential_error_summary( + code: CredentialErrorCode, reference: Optional[str] = None +) -> str: + """Return what happened, mentioning only the reference id.""" + return _CREDENTIAL_ERROR_SUMMARIES[CredentialErrorCode(code)].format( + reference=reference or UNKNOWN_REFERENCE + ) + + +def credential_error_guidance( + code: CredentialErrorCode, expected_type: Optional[CredentialType] = None +) -> str: + """Return what the operator should do about ``code``.""" + code = CredentialErrorCode(code) + if code == CredentialErrorCode.CREDENTIAL_INCOMPATIBLE and expected_type is None: + return _CREDENTIAL_INCOMPATIBLE_UNKNOWN_TYPE_GUIDANCE + return _CREDENTIAL_ERROR_GUIDANCES[code].format( + type=CredentialType(expected_type).value if expected_type else None + ) + + def credential_error_code_from_http( response_code: Optional[int], error_message: Optional[str] ) -> CredentialErrorCode: @@ -64,7 +112,8 @@ def credential_error_code_from_http( The code sent by the platform in the response ``message`` wins. A 404 without a known code (the inject itself does not exist) is reported as - ``CREDENTIAL_NOT_FOUND``. Any other failure falls back to + ``CREDENTIAL_NOT_FOUND``, and a server error as + ``CREDENTIAL_RESOLUTION_UNAVAILABLE``. Any other failure falls back to ``CREDENTIAL_ACCESS_DENIED`` so that nothing about the credential leaks. """ if error_message: @@ -76,16 +125,19 @@ def credential_error_code_from_http( return code if response_code == 404: return CredentialErrorCode.CREDENTIAL_NOT_FOUND + if response_code is not None and response_code >= 500: + return CredentialErrorCode.CREDENTIAL_RESOLUTION_UNAVAILABLE return CredentialErrorCode.CREDENTIAL_ACCESS_DENIED class CredentialResolutionError(OpenAEVError): """A credential reference could not be resolved for the current execution. - ``message`` holds the fixed human-readable message of ``code``. The HTTP - response body is never stored, so it cannot surface in ``str()``. - ``expected_type`` is the credential type the inject expects, only used by - the ``CREDENTIAL_INCOMPATIBLE`` message. + ``message`` holds the fixed human-readable message of ``code``, which is + ``summary`` followed by ``guidance``. The HTTP response body is never + stored, so it cannot surface in ``str()``. ``expected_type`` is the + credential type the inject expects, only used by the + ``CREDENTIAL_INCOMPATIBLE`` message. """ def __init__( @@ -100,18 +152,12 @@ def __init__( self.expected_type = ( CredentialType(expected_type) if expected_type is not None else None ) - template = _CREDENTIAL_ERROR_MESSAGES[self.code] - if ( - self.code == CredentialErrorCode.CREDENTIAL_INCOMPATIBLE - and self.expected_type is None - ): - template = _CREDENTIAL_INCOMPATIBLE_UNKNOWN_TYPE_MESSAGE - message = template.format( - reference=reference or "unknown reference", - type=self.expected_type.value if self.expected_type else None, - ) + self.summary = credential_error_summary(self.code, reference) + self.guidance = credential_error_guidance(self.code, self.expected_type) super().__init__( - error_message=message, response_code=response_code, response_body=None + error_message=f"{self.summary} {self.guidance}", + response_code=response_code, + response_body=None, ) @property @@ -144,9 +190,12 @@ def __init__(self, reference: Optional[str] = None) -> None: __all__ = [ + "UNKNOWN_REFERENCE", "CredentialErrorCode", "CredentialResolutionError", "InvalidResolvedSecretError", "UnsupportedSecretTypeError", "credential_error_code_from_http", + "credential_error_guidance", + "credential_error_summary", ] diff --git a/pyoaev/credential/materializer.py b/pyoaev/credential/materializer.py index 7b4216fe..f4a485c1 100644 --- a/pyoaev/credential/materializer.py +++ b/pyoaev/credential/materializer.py @@ -16,6 +16,8 @@ from typing import Callable, Dict, Iterator, List, Optional from pyoaev.credential.errors import ( + CredentialErrorCode, + CredentialResolutionError, InvalidResolvedSecretError, UnsupportedSecretTypeError, ) @@ -42,6 +44,10 @@ } +class _UnsupportedAzureEnvironmentError(Exception): + """The Azure cloud of the credential has no known authority host.""" + + class CredentialCleanupError(RuntimeError): """Temporary credential files could not be removed.""" @@ -167,7 +173,9 @@ def _azure_authority_host(environment: str) -> str: try: return AZURE_AUTHORITY_HOSTS[environment] except KeyError: - raise InvalidResolvedSecretError() from None + # A valid platform value (e.g. the retired AzureGermanCloud) the + # injectors cannot target: the credential is incompatible, not invalid. + raise _UnsupportedAzureEnvironmentError() from None def _azure_service_principal( @@ -245,7 +253,10 @@ def _gcp_oauth2(secret: GcpOAuth2Secret, workspace: _Workspace) -> Dict[str, str @contextmanager def materialize( - resolved: ResolvedSecret, temporary_root: Optional[Path] = None + resolved: ResolvedSecret, + temporary_root: Optional[Path] = None, + *, + reference: Optional[str] = None, ) -> Iterator[MaterializedCredential]: """Materialize ``resolved`` for the duration of the ``with`` block. @@ -253,19 +264,34 @@ def materialize( resolved: the secret returned by ``resolve_inject_credential``. temporary_root: parent of the private directory, the system temporary directory when omitted. + reference: the credential reference id, only used in error messages. Raises: UnsupportedSecretTypeError: the secret type is not a cloud credential (``USERNAME_PASSWORD``, ``HASH``). + CredentialResolutionError: ``CREDENTIAL_INCOMPATIBLE`` when the Azure + cloud of the credential cannot be targeted. InvalidResolvedSecretError: a value cannot be materialized safely. CredentialCleanupError: the temporary files could not be removed. """ materializer = _MATERIALIZERS.get(resolved.secret_type) if materializer is None: - raise UnsupportedSecretTypeError(resolved.secret_type.value) + raise UnsupportedSecretTypeError(resolved.secret_type.value, reference) workspace = _Workspace(temporary_root) + error: Optional[CredentialResolutionError] = None try: - env = materializer(resolved, workspace) + # Raised outside of the except blocks so that no original error, which + # may quote a secret value, is chained to the reported one. + try: + env = materializer(resolved, workspace) + except _UnsupportedAzureEnvironmentError: + error = CredentialResolutionError( + CredentialErrorCode.CREDENTIAL_INCOMPATIBLE, reference + ) + except InvalidResolvedSecretError: + error = InvalidResolvedSecretError(reference) + if error is not None: + raise error yield MaterializedCredential( secret_type=resolved.secret_type, env=env, files=list(workspace.files) ) diff --git a/test/apis/inject/test_inject.py b/test/apis/inject/test_inject.py index 88abc258..e0d310ae 100644 --- a/test/apis/inject/test_inject.py +++ b/test/apis/inject/test_inject.py @@ -132,14 +132,23 @@ def test_unknown_failures_fall_back_to_access_denied(self): for response in ( build_response(401, {"message": "Unauthorized"}, "Unauthorized"), build_response(400, {"message": "attachment_id: must not be blank"}), - build_response(500, {"message": "boom"}, "Internal Server Error"), - build_response(502, None, "Bad Gateway"), ): with self.subTest(status_code=response.status_code): self._assert_resolution_error( response, CredentialErrorCode.CREDENTIAL_ACCESS_DENIED ) + def test_server_errors_are_resolution_unavailable(self): + for response in ( + build_response(500, {"message": "boom"}, "Internal Server Error"), + build_response(502, None, "Bad Gateway"), + ): + with self.subTest(status_code=response.status_code): + error = self._assert_resolution_error( + response, CredentialErrorCode.CREDENTIAL_RESOLUTION_UNAVAILABLE + ) + self.assertNotIn("boom", error.message) + def test_non_json_success_is_access_denied(self): response = requests.Response() response.status_code = 200 @@ -150,7 +159,7 @@ def test_non_json_success_is_access_denied(self): response, CredentialErrorCode.CREDENTIAL_ACCESS_DENIED ) - def test_network_error_is_access_denied(self): + def test_network_error_is_resolution_unavailable(self): api_client = OpenAEV("url", "token") with mock.patch( "requests.Session.request", @@ -162,7 +171,8 @@ def test_network_error_is_access_denied(self): ) self.assertEqual( - context.exception.code, CredentialErrorCode.CREDENTIAL_ACCESS_DENIED + context.exception.code, + CredentialErrorCode.CREDENTIAL_RESOLUTION_UNAVAILABLE, ) def test_error_never_exposes_response_body(self): diff --git a/test/credential/test_credential_attachment.py b/test/credential/test_credential_attachment.py index c0750707..c966e74b 100644 --- a/test/credential/test_credential_attachment.py +++ b/test/credential/test_credential_attachment.py @@ -12,6 +12,7 @@ CredentialResolutionError, CredentialType, get_credential_attachment, + resolve_credential_attachment, resolve_inject_credential, ) @@ -217,5 +218,33 @@ def test_end_to_end_with_the_openaev_client(self): self.assertEqual(resolved.aws_access_key_id, "AKIAEXAMPLE") +class ResolveCredentialAttachmentTest(unittest.TestCase): + def setUp(self): + self.api = mock.MagicMock() + self.api.inject.resolve_attachment_secret.return_value = AWS_PAYLOAD + self.attachment = CredentialAttachment(REFERENCE, AUTHORISATION_CODE) + + def test_attachment_read_earlier_is_resolved_parsed_and_checked(self): + resolved = resolve_credential_attachment( + self.api, INJECT_ID, self.attachment, "aws" + ) + + self.api.inject.resolve_attachment_secret.assert_called_once_with( + INJECT_ID, REFERENCE, AUTHORISATION_CODE + ) + self.assertIsInstance(resolved, AwsAccessKeySecret) + + def test_incompatible_credential(self): + with self.assertRaises(CredentialResolutionError) as context: + resolve_credential_attachment( + self.api, INJECT_ID, self.attachment, CredentialType.CLOUD_AZURE + ) + + self.assertEqual( + context.exception.code, CredentialErrorCode.CREDENTIAL_INCOMPATIBLE + ) + self.assertEqual(context.exception.reference, REFERENCE) + + if __name__ == "__main__": unittest.main() diff --git a/test/credential/test_credential_errors.py b/test/credential/test_credential_errors.py index 032192ca..a1d879fb 100644 --- a/test/credential/test_credential_errors.py +++ b/test/credential/test_credential_errors.py @@ -5,6 +5,8 @@ CredentialResolutionError, CredentialType, credential_error_code_from_http, + credential_error_guidance, + credential_error_summary, ) from pyoaev.exceptions import OpenAEVError @@ -18,6 +20,7 @@ def test_codes_are_a_closed_set(self): "CREDENTIAL_INACTIVE", "CREDENTIAL_ACCESS_DENIED", "CREDENTIAL_INCOMPATIBLE", + "CREDENTIAL_RESOLUTION_UNAVAILABLE", }, ) @@ -56,7 +59,6 @@ def test_unknown_failure_falls_back_to_access_denied(self): (400, "Bad Request"), (401, "Unauthorized"), (403, None), - (500, "boom"), (None, None), ): with self.subTest(response_code=response_code, message=message): @@ -65,6 +67,20 @@ def test_unknown_failure_falls_back_to_access_denied(self): CredentialErrorCode.CREDENTIAL_ACCESS_DENIED, ) + def test_server_error_is_resolution_unavailable(self): + for response_code in (500, 502, 503, 504): + with self.subTest(response_code=response_code): + self.assertEqual( + credential_error_code_from_http(response_code, "boom"), + CredentialErrorCode.CREDENTIAL_RESOLUTION_UNAVAILABLE, + ) + + def test_platform_code_wins_over_server_error(self): + self.assertEqual( + credential_error_code_from_http(500, "CREDENTIAL_INACTIVE"), + CredentialErrorCode.CREDENTIAL_INACTIVE, + ) + class CredentialResolutionErrorTest(unittest.TestCase): def test_not_found_message_identifies_the_reference(self): @@ -158,5 +174,50 @@ def test_is_an_openaev_error(self): self.assertIsInstance(error, OpenAEVError) +class CredentialErrorMessagePartsTest(unittest.TestCase): + def test_message_is_summary_then_guidance_for_every_code(self): + for code in CredentialErrorCode: + with self.subTest(code=code): + error = CredentialResolutionError(code, reference="ref-1") + + self.assertEqual(error.message, f"{error.summary} {error.guidance}") + self.assertEqual(error.summary, credential_error_summary(code, "ref-1")) + self.assertEqual(error.guidance, credential_error_guidance(code)) + + def test_not_found_message_is_unchanged(self): + error = CredentialResolutionError( + CredentialErrorCode.CREDENTIAL_NOT_FOUND, reference="ref-1" + ) + + self.assertEqual( + error.message, + "The credential configured on this inject (ref-1) no longer exists at " + "the moment of execution. Select an existing credential on the inject, " + "then run it again.", + ) + + def test_incompatible_guidance_names_the_expected_type(self): + self.assertEqual( + credential_error_guidance( + CredentialErrorCode.CREDENTIAL_INCOMPATIBLE, CredentialType.CLOUD_GCP + ), + "Select a credential of type CLOUD_GCP on the inject, then run it again.", + ) + + def test_summary_without_reference_uses_the_sentinel(self): + self.assertIn( + "unknown reference", + credential_error_summary(CredentialErrorCode.CREDENTIAL_INACTIVE), + ) + + def test_resolution_unavailable_mentions_the_platform(self): + error = CredentialResolutionError( + CredentialErrorCode.CREDENTIAL_RESOLUTION_UNAVAILABLE, reference="ref-1" + ) + + self.assertIn("ref-1", error.message) + self.assertIn("platform", error.message) + + if __name__ == "__main__": unittest.main() diff --git a/test/credential/test_credential_materializer.py b/test/credential/test_credential_materializer.py index ca0a0965..d9a42c83 100644 --- a/test/credential/test_credential_materializer.py +++ b/test/credential/test_credential_materializer.py @@ -14,6 +14,8 @@ AzureManagedIdentitySecret, AzureServicePrincipalSecret, CredentialCleanupError, + CredentialErrorCode, + CredentialResolutionError, GcpOAuth2Secret, GcpServiceAccountSecret, HashAlgorithm, @@ -242,13 +244,20 @@ def test_system_assigned_managed_identity(self): {"AZURE_AUTHORITY_HOST": "https://login.chinacloudapi.cn/"}, ) - def test_unknown_environment_is_rejected(self): + def test_unsupported_environment_is_incompatible(self): resolved = AzureManagedIdentitySecret(azure_environment="AzureGermanCloud") - with self.assertRaises(InvalidResolvedSecretError): - with self.materialize(resolved): + with self.assertRaises(CredentialResolutionError) as context: + with materialize(resolved, self.root, reference="ref-1"): self.fail("must not be reached") + self.assertIs( + context.exception.code, CredentialErrorCode.CREDENTIAL_INCOMPATIBLE + ) + self.assertEqual(context.exception.reference, "ref-1") + self.assertIsNone(context.exception.__cause__) + self.assertIsNone(context.exception.__context__) + class GcpMaterializerTest(MaterializerTestCase): def test_service_account(self):