From 23c14754dbc68cd2a1acf999e51c7d6523473dee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabriel=20PEZ=C3=89?= Date: Thu, 29 Oct 2026 15:01:48 +0100 Subject: [PATCH] feat(credentials): add sovereign cloud process (#8762) --- pyoaev/credential/materializer.py | 61 +++++- pyoaev/credential/resolved.py | 18 ++ .../test_credential_materializer.py | 204 ++++++++++++++++++ test/credential/test_credential_resolved.py | 121 +++++++++++ 4 files changed, 399 insertions(+), 5 deletions(-) diff --git a/pyoaev/credential/materializer.py b/pyoaev/credential/materializer.py index 3851c8b..2494520 100644 --- a/pyoaev/credential/materializer.py +++ b/pyoaev/credential/materializer.py @@ -13,7 +13,7 @@ from contextlib import contextmanager from dataclasses import dataclass, field from pathlib import Path -from typing import Callable, Dict, Iterator, List, Optional +from typing import Callable, Dict, Iterator, List, Optional, Union from pyoaev.credential.errors import ( CredentialErrorCode, @@ -125,6 +125,13 @@ def _aws_region_env(region: Optional[str]) -> Dict[str, str]: return {"AWS_REGION": region, "AWS_DEFAULT_REGION": region} +def _aws_endpoint_env(endpoint_url: Optional[str]) -> Dict[str, str]: + # Sovereign or mocked cloud: the same endpoint serves the STS calls. + if not endpoint_url: + return {} + return {"AWS_ENDPOINT_URL": endpoint_url, "AWS_ENDPOINT_URL_STS": endpoint_url} + + def _aws_access_key( secret: AwsAccessKeySecret, workspace: _Workspace ) -> Dict[str, str]: @@ -132,6 +139,7 @@ def _aws_access_key( **_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, + **_aws_endpoint_env(secret.aws_endpoint_url), } if secret.aws_session_token: env["AWS_SESSION_TOKEN"] = secret.aws_session_token @@ -144,6 +152,7 @@ def _aws_assume_role( env = { **_aws_region_env(secret.aws_default_region), "AWS_PROFILE": AWS_TARGET_PROFILE, + **_aws_endpoint_env(secret.aws_endpoint_url), } profile: Dict[str, Optional[str]] = {"role_arn": secret.aws_role_arn} if secret.aws_source_identity_type == AwsSourceIdentityType.INSTANCE_DEFAULT: @@ -181,6 +190,36 @@ def _azure_authority_host(environment: str) -> str: raise _UnsupportedAzureEnvironmentError() from None +def _azure_sovereign_env( + secret: Union[AzureServicePrincipalSecret, AzureManagedIdentitySecret], + workspace: _Workspace, +) -> Dict[str, str]: + """Authority host and trusted CA of the credential, shared by both methods. + + ``REQUESTS_CA_BUNDLE`` (Python SDK) and ``SSL_CERT_FILE`` (Go, stratus + ignores the former) replace the default trust store of their client: the + certificate must contain every CA needed by the authority host. + """ + # The environment is always validated, even when the host is overridden. + env = {"AZURE_AUTHORITY_HOST": _azure_authority_host(secret.azure_environment)} + if secret.azure_authority_host_uri: + # The Go SDK requires the trailing slash. + authority_host = secret.azure_authority_host_uri + if not authority_host.endswith("/"): + authority_host += "/" + env["AZURE_AUTHORITY_HOST"] = authority_host + env["AZURE_POD_IDENTITY_AUTHORITY_HOST"] = authority_host + if secret.azure_requests_certificate: + bundle = str( + workspace.write( + secret.azure_requests_certificate, prefix="azure-ca-", suffix=".pem" + ) + ) + env["REQUESTS_CA_BUNDLE"] = bundle + env["SSL_CERT_FILE"] = bundle + return env + + def _azure_service_principal( secret: AzureServicePrincipalSecret, workspace: _Workspace ) -> Dict[str, str]: @@ -188,7 +227,7 @@ def _azure_service_principal( "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), + **_azure_sovereign_env(secret, workspace), } if secret.azure_subscription_id: env["AZURE_SUBSCRIPTION_ID"] = secret.azure_subscription_id @@ -198,7 +237,7 @@ def _azure_service_principal( def _azure_managed_identity( secret: AzureManagedIdentitySecret, workspace: _Workspace ) -> Dict[str, str]: - env = {"AZURE_AUTHORITY_HOST": _azure_authority_host(secret.azure_environment)} + env = _azure_sovereign_env(secret, workspace) # Without a client id, the system-assigned identity is used. if secret.azure_client_id: env["AZURE_CLIENT_ID"] = secret.azure_client_id @@ -225,7 +264,12 @@ def _gcp_service_account( 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) + env = _gcp_env(credentials_file, secret.gcp_scope, secret.gcp_project_id) + if secret.gcp_custom_resource_manager_uri: + env["GOOGLE_RESOURCE_MANAGER_CUSTOM_ENDPOINT"] = ( + secret.gcp_custom_resource_manager_uri + ) + return env def _gcp_oauth2(secret: GcpOAuth2Secret, workspace: _Workspace) -> Dict[str, str]: @@ -241,7 +285,14 @@ def _gcp_oauth2(secret: GcpOAuth2Secret, workspace: _Workspace) -> Dict[str, str prefix="gcp-credentials-", suffix=".json", ) - return _gcp_env(credentials_file, secret.gcp_scope, secret.gcp_project_id) + env = _gcp_env(credentials_file, secret.gcp_scope, secret.gcp_project_id) + if secret.gcp_custom_token_uri: + env["GOOGLE_OAUTH2_TOKEN_URI_OVERRIDE"] = secret.gcp_custom_token_uri + if secret.gcp_custom_iam_uri: + env["GOOGLE_IAM_CUSTOM_ENDPOINT"] = secret.gcp_custom_iam_uri + if secret.gcp_custom_oauth_access_token: + env["GOOGLE_OAUTH_ACCESS_TOKEN"] = secret.gcp_custom_oauth_access_token + return env _MATERIALIZERS: Dict[SecretType, Callable[..., Dict[str, str]]] = { diff --git a/pyoaev/credential/resolved.py b/pyoaev/credential/resolved.py index 59e26d2..6d382cb 100644 --- a/pyoaev/credential/resolved.py +++ b/pyoaev/credential/resolved.py @@ -85,6 +85,8 @@ class AwsAccessKeySecret(ResolvedSecret): aws_secret_access_key: str = _secret() aws_default_region: Optional[str] = None aws_session_token: Optional[str] = _secret(default=None) + aws_account_id: Optional[str] = None + aws_endpoint_url: Optional[str] = None @dataclass(frozen=True) @@ -99,6 +101,8 @@ class AwsAssumeRoleSecret(ResolvedSecret): 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) + aws_account_id: Optional[str] = None + aws_endpoint_url: Optional[str] = None def __post_init__(self) -> None: # A static source identity cannot assume the role without its keys. @@ -119,6 +123,11 @@ class AzureServicePrincipalSecret(ResolvedSecret): azure_client_secret: str = _secret() azure_tenant_id: str azure_subscription_id: Optional[str] = None + azure_authority_host_uri: Optional[str] = None + # Java ``byte[]``, sent base64 encoded by the platform. + azure_requests_certificate: Optional[bytes] = _parsed_with( + _decode_base64, repr=False, default=None + ) @dataclass(frozen=True) @@ -128,6 +137,11 @@ class AzureManagedIdentitySecret(ResolvedSecret): azure_environment: str azure_client_id: Optional[str] = None azure_subscription_id: Optional[str] = None + azure_authority_host_uri: Optional[str] = None + # Java ``byte[]``, sent base64 encoded by the platform. + azure_requests_certificate: Optional[bytes] = _parsed_with( + _decode_base64, repr=False, default=None + ) @dataclass(frozen=True) @@ -138,6 +152,7 @@ class GcpServiceAccountSecret(ResolvedSecret): # 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 + gcp_custom_resource_manager_uri: Optional[str] = None @dataclass(frozen=True) @@ -149,6 +164,9 @@ class GcpOAuth2Secret(ResolvedSecret): gcp_oauth_client_secret: str = _secret() gcp_oauth_refresh_token: str = _secret() gcp_project_id: Optional[str] = None + gcp_custom_token_uri: Optional[str] = None + gcp_custom_iam_uri: Optional[str] = None + gcp_custom_oauth_access_token: Optional[str] = _secret(default=None) _RESOLVED_SECRET_CLASSES: Dict[SecretType, Type[ResolvedSecret]] = { diff --git a/test/credential/test_credential_materializer.py b/test/credential/test_credential_materializer.py index af0023b..7f9dab5 100644 --- a/test/credential/test_credential_materializer.py +++ b/test/credential/test_credential_materializer.py @@ -4,6 +4,7 @@ import stat import tempfile import unittest +from dataclasses import replace from pathlib import Path from unittest import mock @@ -28,6 +29,9 @@ ) SECRET = "super-secret-value" +ACCESS_TOKEN = "custom-oauth-access-token" +AWS_ENDPOINT_URL = "http://mock-aws:4566" +CERTIFICATE = b"-----BEGIN CERTIFICATE-----\nMIIBsovereign\n-----END CERTIFICATE-----\n" GCP_KEY_JSON = b'{"type": "service_account", "private_key": "super-secret-value"}' AWS_ACCESS_KEY = AwsAccessKeySecret( @@ -319,6 +323,206 @@ def test_without_project_id(self): self.assertNotIn("GOOGLE_PROJECT", credential.env) +class SovereignCloudMaterializerTest(MaterializerTestCase): + def test_aws_endpoint_url_sets_both_variables(self): + for resolved in ( + replace(AWS_ACCESS_KEY, aws_endpoint_url=AWS_ENDPOINT_URL), + replace(AWS_ASSUME_ROLE_STATIC, aws_endpoint_url=AWS_ENDPOINT_URL), + replace(AWS_ASSUME_ROLE_INSTANCE, aws_endpoint_url=AWS_ENDPOINT_URL), + ): + with self.subTest(secret_type=resolved.secret_type): + with self.materialize(resolved) as credential: + self.assertEqual( + credential.env["AWS_ENDPOINT_URL"], AWS_ENDPOINT_URL + ) + self.assertEqual( + credential.env["AWS_ENDPOINT_URL_STS"], AWS_ENDPOINT_URL + ) + + def test_aws_account_id_has_no_environment_variable(self): + for resolved in (AWS_ACCESS_KEY, AWS_ASSUME_ROLE_STATIC): + with self.subTest(secret_type=resolved.secret_type): + with self.materialize( + replace(resolved, aws_account_id="123456789012") + ) as credential: + with self.materialize(resolved) as reference: + self.assertEqual(credential.env.keys(), reference.env.keys()) + self.assertNotIn("123456789012", credential.env.values()) + + def test_gcp_oauth2_overrides(self): + resolved = replace( + GCP_OAUTH2, + gcp_custom_token_uri="http://mock-gcp/token", + gcp_custom_iam_uri="http://mock-gcp/iam", + gcp_custom_oauth_access_token=ACCESS_TOKEN, + ) + + with self.materialize(resolved) as credential: + env = credential.env + self.assertEqual( + env["GOOGLE_OAUTH2_TOKEN_URI_OVERRIDE"], "http://mock-gcp/token" + ) + self.assertEqual(env["GOOGLE_IAM_CUSTOM_ENDPOINT"], "http://mock-gcp/iam") + self.assertEqual(env["GOOGLE_OAUTH_ACCESS_TOKEN"], ACCESS_TOKEN) + self.assertNotIn("GOOGLE_RESOURCE_MANAGER_CUSTOM_ENDPOINT", env) + self.assertNotIn(ACCESS_TOKEN, repr(credential)) + + def test_gcp_oauth2_overrides_are_independent(self): + resolved = replace(GCP_OAUTH2, gcp_custom_iam_uri="http://mock-gcp/iam") + + with self.materialize(resolved) as credential: + self.assertEqual( + set(credential.env) - set(self._env(GCP_OAUTH2)), + {"GOOGLE_IAM_CUSTOM_ENDPOINT"}, + ) + + def test_gcp_service_account_resource_manager_override(self): + resolved = replace( + GCP_SERVICE_ACCOUNT, gcp_custom_resource_manager_uri="http://mock-gcp/rm" + ) + + with self.materialize(resolved) as credential: + self.assertEqual( + credential.env["GOOGLE_RESOURCE_MANAGER_CUSTOM_ENDPOINT"], + "http://mock-gcp/rm", + ) + self.assertEqual( + set(credential.env) - set(self._env(GCP_SERVICE_ACCOUNT)), + {"GOOGLE_RESOURCE_MANAGER_CUSTOM_ENDPOINT"}, + ) + + def test_azure_authority_host_overrides_the_environment_one(self): + for resolved in (AZURE_SERVICE_PRINCIPAL, AZURE_MANAGED_IDENTITY): + for authority_host in ("https://mock-azure", "https://mock-azure/"): + with self.subTest( + secret_type=resolved.secret_type, authority_host=authority_host + ): + with self.materialize( + replace(resolved, azure_authority_host_uri=authority_host) + ) as credential: + self.assertEqual( + credential.env["AZURE_AUTHORITY_HOST"], + "https://mock-azure/", + ) + self.assertEqual( + credential.env["AZURE_POD_IDENTITY_AUTHORITY_HOST"], + "https://mock-azure/", + ) + + def test_azure_unsupported_environment_still_fails_with_an_authority_host(self): + resolved = AzureManagedIdentitySecret( + azure_environment="AzureGermanCloud", + azure_authority_host_uri="https://mock-azure/", + ) + + with self.assertRaises(CredentialResolutionError) as context: + with self.materialize(resolved): + self.fail("must not be reached") + + self.assertIs( + context.exception.code, CredentialErrorCode.CREDENTIAL_INCOMPATIBLE + ) + + def test_azure_certificate_is_written_and_referenced_by_both_variables(self): + for resolved in (AZURE_SERVICE_PRINCIPAL, AZURE_MANAGED_IDENTITY): + with self.subTest(secret_type=resolved.secret_type): + with self.materialize( + replace(resolved, azure_requests_certificate=CERTIFICATE) + ) as credential: + env = credential.env + self.assertEqual(env["REQUESTS_CA_BUNDLE"], env["SSL_CERT_FILE"]) + certificate_file = Path(env["SSL_CERT_FILE"]) + self.assertEqual(credential.files, [certificate_file]) + self.assertTrue(certificate_file.name.startswith("azure-ca-")) + self.assertEqual(certificate_file.suffix, ".pem") + self.assertEqual(certificate_file.read_bytes(), CERTIFICATE) + # Only the path travels in the environment. + self.assertNotIn(CERTIFICATE.decode(), env.values()) + self.assertNotIn(CERTIFICATE.decode(), repr(credential)) + + @unittest.skipIf(os.name == "nt", "POSIX permissions only") + def test_azure_certificate_posix_permissions(self): + with self.materialize( + replace(AZURE_SERVICE_PRINCIPAL, azure_requests_certificate=CERTIFICATE) + ) as credential: + path = credential.files[0] + self.assertEqual(stat.S_IMODE(path.parent.stat().st_mode), 0o700) + self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600) + + def test_azure_certificate_is_removed_after_the_block(self): + resolved = replace( + AZURE_SERVICE_PRINCIPAL, azure_requests_certificate=CERTIFICATE + ) + + with self.materialize(resolved) as credential: + files = credential.files + + self.assertEqual(len(files), 1) + self.assertFalse(files[0].exists()) + self.assert_root_is_empty() + + def test_azure_certificate_is_removed_when_the_body_raises(self): + resolved = replace( + AZURE_MANAGED_IDENTITY, azure_requests_certificate=CERTIFICATE + ) + + with self.assertRaises(RuntimeError): + with self.materialize(resolved) as credential: + files = credential.files + raise RuntimeError("execution failed") + + self.assertFalse(files[0].exists()) + self.assert_root_is_empty() + + def test_azure_certificate_removal_failure_is_reported(self): + resolved = replace( + AZURE_SERVICE_PRINCIPAL, azure_requests_certificate=CERTIFICATE + ) + + with mock.patch("pyoaev.credential.materializer.shutil.rmtree"): + with self.assertRaises(CredentialCleanupError) as context: + with self.materialize(resolved) as credential: + directory = credential.files[0].parent + + self.assertNotIn(CERTIFICATE.decode(), str(context.exception)) + for path in directory.iterdir(): + path.unlink() + directory.rmdir() + + def test_azure_without_sovereign_fields_is_unchanged(self): + for resolved in (AZURE_SERVICE_PRINCIPAL, AZURE_MANAGED_IDENTITY): + with self.subTest(secret_type=resolved.secret_type): + with self.materialize(resolved) as credential: + for name in ( + "AZURE_POD_IDENTITY_AUTHORITY_HOST", + "REQUESTS_CA_BUNDLE", + "SSL_CERT_FILE", + ): + self.assertNotIn(name, credential.env) + self.assertEqual(credential.files, []) + + def test_absent_overrides_leave_the_environment_unchanged(self): + for resolved in ALL_CLOUD_SECRETS: + with self.subTest(secret_type=resolved.secret_type): + env = self._env(resolved) + for name in ( + "AWS_ENDPOINT_URL", + "AWS_ENDPOINT_URL_STS", + "GOOGLE_OAUTH2_TOKEN_URI_OVERRIDE", + "GOOGLE_IAM_CUSTOM_ENDPOINT", + "GOOGLE_OAUTH_ACCESS_TOKEN", + "GOOGLE_RESOURCE_MANAGER_CUSTOM_ENDPOINT", + "AZURE_POD_IDENTITY_AUTHORITY_HOST", + "REQUESTS_CA_BUNDLE", + "SSL_CERT_FILE", + ): + self.assertNotIn(name, env) + + def _env(self, resolved): + with self.materialize(resolved) as credential: + return dict(credential.env) + + class MaterializerLifecycleTest(MaterializerTestCase): def test_identity_secrets_are_not_supported(self): for resolved in ( diff --git a/test/credential/test_credential_resolved.py b/test/credential/test_credential_resolved.py index cb4562d..c1dd590 100644 --- a/test/credential/test_credential_resolved.py +++ b/test/credential/test_credential_resolved.py @@ -25,6 +25,27 @@ REFERENCE = "credential-reference-id" SECRET = "super-secret-value" GCP_KEY_JSON = b'{"type": "service_account", "private_key": "super-secret-value"}' +ACCESS_TOKEN = "custom-oauth-access-token" +CERTIFICATE = b"-----BEGIN CERTIFICATE-----\nMIIBsovereign\n-----END CERTIFICATE-----\n" + +SOVEREIGN_FIELDS = { + SecretType.AWS_ACCESS_KEY: ("aws_account_id", "aws_endpoint_url"), + SecretType.AWS_ASSUME_ROLE: ("aws_account_id", "aws_endpoint_url"), + SecretType.AZURE_SERVICE_PRINCIPAL: ( + "azure_authority_host_uri", + "azure_requests_certificate", + ), + SecretType.AZURE_MANAGED_IDENTITY: ( + "azure_authority_host_uri", + "azure_requests_certificate", + ), + SecretType.GCP_SERVICE_ACCOUNT: ("gcp_custom_resource_manager_uri",), + SecretType.GCP_OAUTH2: ( + "gcp_custom_token_uri", + "gcp_custom_iam_uri", + "gcp_custom_oauth_access_token", + ), +} PAYLOADS = { SecretType.USERNAME_PASSWORD: { @@ -241,6 +262,106 @@ def test_gcp_oauth2(self): ), ) + def test_aws_sovereign_fields(self): + for secret_type in (SecretType.AWS_ACCESS_KEY, SecretType.AWS_ASSUME_ROLE): + with self.subTest(secret_type=secret_type): + resolved = _parse( + secret_type, + aws_account_id="000000000000", + aws_endpoint_url="http://mock-aws:4566", + ) + + self.assertEqual(resolved.aws_account_id, "000000000000") + self.assertEqual(resolved.aws_endpoint_url, "http://mock-aws:4566") + + def test_gcp_oauth2_sovereign_fields(self): + resolved = _parse( + SecretType.GCP_OAUTH2, + gcp_custom_token_uri="http://mock-gcp/token", + gcp_custom_iam_uri="http://mock-gcp/iam", + gcp_custom_oauth_access_token=SECRET, + ) + + self.assertEqual(resolved.gcp_custom_token_uri, "http://mock-gcp/token") + self.assertEqual(resolved.gcp_custom_iam_uri, "http://mock-gcp/iam") + self.assertEqual(resolved.gcp_custom_oauth_access_token, SECRET) + + def test_gcp_service_account_sovereign_fields(self): + resolved = _parse( + SecretType.GCP_SERVICE_ACCOUNT, + gcp_custom_resource_manager_uri="http://mock-gcp/rm", + ) + + self.assertEqual(resolved.gcp_custom_resource_manager_uri, "http://mock-gcp/rm") + + def test_azure_sovereign_fields(self): + for secret_type in ( + SecretType.AZURE_SERVICE_PRINCIPAL, + SecretType.AZURE_MANAGED_IDENTITY, + ): + with self.subTest(secret_type=secret_type): + resolved = _parse( + secret_type, + azure_authority_host_uri="https://mock-azure/", + azure_requests_certificate=base64.b64encode(CERTIFICATE).decode(), + ) + + self.assertEqual( + resolved.azure_authority_host_uri, "https://mock-azure/" + ) + self.assertEqual(resolved.azure_requests_certificate, CERTIFICATE) + + def test_sovereign_fields_default_to_none_when_absent_or_empty(self): + for secret_type, names in SOVEREIGN_FIELDS.items(): + for absent in (None, ""): + with self.subTest(secret_type=secret_type, absent=absent): + resolved = _parse(secret_type, **{name: absent for name in names}) + + for name in names: + self.assertIsNone(getattr(resolved, name)) + + def test_sovereign_fields_are_none_in_payloads_without_them(self): + for secret_type, names in SOVEREIGN_FIELDS.items(): + with self.subTest(secret_type=secret_type): + resolved = _parse(secret_type) + + for name in names: + self.assertIsNone(getattr(resolved, name)) + + def test_azure_certificate_with_invalid_base64(self): + for secret_type in ( + SecretType.AZURE_SERVICE_PRINCIPAL, + SecretType.AZURE_MANAGED_IDENTITY, + ): + with self.subTest(secret_type=secret_type): + with self.assertRaises(InvalidResolvedSecretError) as context: + _parse(secret_type, azure_requests_certificate="not base64!") + + self.assertIsNone(context.exception.__cause__) + self.assertIsNone(context.exception.__context__) + + def test_repr_and_str_never_expose_the_token_nor_the_certificate(self): + resolved_secrets = ( + _parse( + SecretType.GCP_OAUTH2, + gcp_custom_oauth_access_token=ACCESS_TOKEN, + ), + _parse( + SecretType.AZURE_SERVICE_PRINCIPAL, + azure_requests_certificate=base64.b64encode(CERTIFICATE).decode(), + ), + _parse( + SecretType.AZURE_MANAGED_IDENTITY, + azure_requests_certificate=base64.b64encode(CERTIFICATE).decode(), + ), + ) + for resolved in resolved_secrets: + with self.subTest(secret_type=resolved.secret_type): + for rendered in (repr(resolved), str(resolved)): + self.assertNotIn(ACCESS_TOKEN, rendered) + self.assertNotIn(CERTIFICATE.decode(), rendered) + self.assertNotIn(base64.b64encode(CERTIFICATE).decode(), rendered) + 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):