feat(credentials): add sovereign cloud process (#8762)

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,13 +125,22 @@ 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]:
+    # Single host for every service: fits a mocked cloud or a proxy. A real
+    # AWS partition (GovCloud, China) is reached through its region alone.
+    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]:
     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,
+        **_aws_endpoint_env(secret.aws_endpoint_url),
     }
     if secret.aws_session_token:
         env["AWS_SESSION_TOKEN"] = secret.aws_session_token
@@ -144,6 +153,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,14 +191,48 @@ 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.
+
+    ``AZURE_POD_IDENTITY_AUTHORITY_HOST`` is deliberately not set: despite its
+    name, azure-identity uses it as the managed identity (IMDS) endpoint, not
+    as the Entra authority. Pointing it to a login server would break the
+    managed identity token requests.
+    """
+    # 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
+    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]:
     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),
+        **_azure_sovereign_env(secret, workspace),
     }
     if secret.azure_subscription_id:
         env["AZURE_SUBSCRIPTION_ID"] = secret.azure_subscription_id
@@ -198,7 +242,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
@@ -215,7 +259,9 @@ def _gcp_env(
         "GOOGLE_CLOUD_UNIVERSE_DOMAIN": scope,
     }
     if project_id:
+        # Stratus and Terraform read the former, Prowler the latter.
         env["GOOGLE_PROJECT"] = project_id
+        env["GOOGLE_CLOUD_PROJECT"] = project_id
     return env
 
 
@@ -225,7 +271,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 +292,17 @@ 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:
+        # Terraform (stratus) reads the former, Prowler and gcloud the latter.
+        # Prowler then prefers the token over the credentials file.
+        env["GOOGLE_OAUTH_ACCESS_TOKEN"] = secret.gcp_custom_oauth_access_token
+        env["CLOUDSDK_AUTH_ACCESS_TOKEN"] = secret.gcp_custom_oauth_access_token
+    return env
 
 
 _MATERIALIZERS: Dict[SecretType, Callable[..., Dict[str, str]]] = {

pyoaev/credential/resolved.py

@@ -11,7 +11,8 @@
 
 import base64
 from dataclasses import MISSING, dataclass, field, fields
-from typing import Any, ClassVar, Dict, Optional, Type, TypeVar, Union
+from typing import Any, ClassVar, Dict, Optional, Tuple, Type, TypeVar, Union
+from urllib.parse import urlsplit
 
 from pyoaev.credential.errors import (
     CredentialErrorCode,
@@ -52,6 +53,37 @@ def _decode_base64(value: str) -> bytes:
     return base64.b64decode(value, validate=True)
 
 
+def _decode_pem_certificate(value: str) -> bytes:
+    certificate = _decode_base64(value)
+    # A DER certificate (common for .cer/.crt files) written as .pem is
+    # rejected by OpenSSL and silently ignored by Go (empty root pool).
+    if b"-----BEGIN CERTIFICATE-----" not in certificate:
+        raise ValueError("The certificate is not PEM encoded")
+    return certificate
+
+
+def _endpoint_url(value: str, schemes: Tuple[str, ...] = ("http", "https")) -> str:
+    # The value is never quoted in the error: it may embed credentials.
+    try:
+        parsed = urlsplit(value)
+    except ValueError:
+        raise ValueError("Invalid endpoint URL") from None
+    if (
+        any(character.isspace() for character in value)
+        or parsed.scheme not in schemes
+        or not parsed.hostname
+        # A user info part would leak credentials in every request log.
+        or "@" in parsed.netloc
+    ):
+        raise ValueError("Invalid endpoint URL")
+    return value
+
+
+def _https_endpoint_url(value: str) -> str:
+    # azure-identity and the Go SDK refuse an authority that is not TLS.
+    return _endpoint_url(value, schemes=("https",))
+
+
 @dataclass(frozen=True)
 class ResolvedSecret:
     secret_type: ClassVar[SecretType]
@@ -85,6 +117,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] = _parsed_with(_endpoint_url, default=None)
 
 
 @dataclass(frozen=True)
@@ -99,6 +133,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] = _parsed_with(_endpoint_url, default=None)
 
     def __post_init__(self) -> None:
         # A static source identity cannot assume the role without its keys.
@@ -119,6 +155,13 @@ class AzureServicePrincipalSecret(ResolvedSecret):
     azure_client_secret: str = _secret()
     azure_tenant_id: str
     azure_subscription_id: Optional[str] = None
+    azure_authority_host_uri: Optional[str] = _parsed_with(
+        _https_endpoint_url, default=None
+    )
+    # Java ``byte[]``, sent base64 encoded by the platform; PEM content only.
+    azure_requests_certificate: Optional[bytes] = _parsed_with(
+        _decode_pem_certificate, repr=False, default=None
+    )
 
 
 @dataclass(frozen=True)
@@ -128,6 +171,13 @@ class AzureManagedIdentitySecret(ResolvedSecret):
     azure_environment: str
     azure_client_id: Optional[str] = None
     azure_subscription_id: Optional[str] = None
+    azure_authority_host_uri: Optional[str] = _parsed_with(
+        _https_endpoint_url, default=None
+    )
+    # Java ``byte[]``, sent base64 encoded by the platform; PEM content only.
+    azure_requests_certificate: Optional[bytes] = _parsed_with(
+        _decode_pem_certificate, repr=False, default=None
+    )
 
 
 @dataclass(frozen=True)
@@ -138,6 +188,9 @@ 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] = _parsed_with(
+        _endpoint_url, default=None
+    )
 
 
 @dataclass(frozen=True)
@@ -149,6 +202,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] = _parsed_with(_endpoint_url, default=None)
+    gcp_custom_iam_uri: Optional[str] = _parsed_with(_endpoint_url, default=None)
+    gcp_custom_oauth_access_token: Optional[str] = _secret(default=None)
 
 
 _RESOLVED_SECRET_CLASSES: Dict[SecretType, Type[ResolvedSecret]] = {