External Key Management

def list_ekm_connections(self, request: ListEkmConnectionsRequest) -> ListEkmConnectionsResponse:
    """
    Lists EkmConnections.
    
    Parameters:
    - request.parent: Required. Location path (projects/{project}/locations/{location})
    - request.page_size: Optional. Maximum results per page
    - request.page_token: Optional. Pagination token
    - request.filter: Optional. Filter expression
    - request.order_by: Optional. Sorting order
    
    Returns:
    List of EkmConnection objects with pagination
    """

def get_ekm_connection(self, request: GetEkmConnectionRequest) -> EkmConnection:
    """
    Returns metadata for a given EkmConnection.
    
    Parameters:
    - request.name: Required. EkmConnection resource name
    
    Returns:
    EkmConnection object with metadata
    """

def create_ekm_connection(self, request: CreateEkmConnectionRequest) -> EkmConnection:
    """
    Creates a new EkmConnection in a given Project and Location.
    
    Parameters:
    - request.parent: Required. Location path
    - request.ekm_connection_id: Required. Unique ID for EkmConnection
    - request.ekm_connection: Required. EkmConnection object to create
    
    Returns:
    Created EkmConnection object
    """

def update_ekm_connection(self, request: UpdateEkmConnectionRequest) -> EkmConnection:
    """
    Updates an EkmConnection's metadata.
    
    Parameters:
    - request.ekm_connection: Required. Updated EkmConnection object
    - request.update_mask: Required. Fields to update
    
    Returns:
    Updated EkmConnection object
    """

def get_ekm_config(self, request: GetEkmConfigRequest) -> EkmConfig:
    """
    Returns the EkmConfig singleton resource for a given project and location.
    
    Parameters:
    - request.name: Required. EkmConfig resource name
    
    Returns:
    EkmConfig object with configuration
    """

def update_ekm_config(self, request: UpdateEkmConfigRequest) -> EkmConfig:
    """
    Updates the EkmConfig singleton resource for a given project and location.
    
    Parameters:
    - request.ekm_config: Required. Updated EkmConfig object
    - request.update_mask: Required. Fields to update
    
    Returns:
    Updated EkmConfig object
    """

def verify_connectivity(self, request: VerifyConnectivityRequest) -> VerifyConnectivityResponse:
    """
    Verifies that Cloud KMS can successfully connect to the external key manager specified by an EkmConnection.
    
    Parameters:
    - request.name: Required. EkmConnection resource name
    
    Returns:
    VerifyConnectivityResponse with verification results
    """

class EkmConnection:
    """
    Individual EKM connection for creating external VPC keys.
    
    Attributes:
    - name: str - Resource name of the EkmConnection
    - create_time: Timestamp - Creation timestamp
    - service_resolvers: List[ServiceResolver] - EKM replica resolvers
    - etag: str - Entity tag for optimistic concurrency control
    - key_management_mode: KeyManagementMode - Key management mode
    - crypto_space_path: str - Path to crypto space in EKM
    """
    
    class ServiceResolver:
        """
        EKM replica resolver within an EkmConnection.
        
        Attributes:
        - service_directory_service: str - Service Directory service name
        - endpoint_filter: str - Filter for selecting endpoints
        - hostname: str - Hostname of EKM replica
        - server_certificates: List[Certificate] - TLS certificates
        """
    
    class KeyManagementMode:
        """
        Key management mode for EKM operations.
        
        Values:
        - KEY_MANAGEMENT_MODE_UNSPECIFIED: Not specified
        - MANUAL: EKM-side key management operations
        - CLOUD_KMS: Cloud KMS-initiated EKM operations
        """

class EkmConfig:
    """
    Singleton resource for EKM configuration parameters.
    
    Attributes:
    - name: str - Resource name of the EkmConfig
    - default_ekm_connection: str - Default EkmConnection for external keys
    """

class Certificate:
    """
    X.509 certificate for HTTPS connections to EKM replicas.
    
    Attributes:
    - raw_der: bytes - DER-encoded certificate
    - parsed: bool - Whether certificate was successfully parsed
    - issuer: str - Certificate issuer
    - subject: str - Certificate subject
    - subject_alternative_dns_names: List[str] - Subject alternative DNS names
    - not_before_time: Timestamp - Certificate validity start time
    - not_after_time: Timestamp - Certificate validity end time
    - serial_number: str - Certificate serial number
    - sha256_fingerprint: str - SHA256 fingerprint
    """

class CreateEkmConnectionRequest:
    """
    Request to create an EkmConnection.
    
    Attributes:
    - parent: str - Required. Location path
    - ekm_connection_id: str - Required. Unique ID for EkmConnection
    - ekm_connection: EkmConnection - Required. EkmConnection to create
    """

class UpdateEkmConnectionRequest:
    """
    Request to update an EkmConnection.
    
    Attributes:
    - ekm_connection: EkmConnection - Required. Updated EkmConnection
    - update_mask: FieldMask - Required. Fields to update
    """

class ListEkmConnectionsRequest:
    """
    Request to list EkmConnections.
    
    Attributes:
    - parent: str - Required. Location path
    - page_size: int - Optional. Maximum results per page
    - page_token: str - Optional. Pagination token
    - filter: str - Optional. Filter expression
    - order_by: str - Optional. Sorting order
    """

class ListEkmConnectionsResponse:
    """
    Response from listing EkmConnections.
    
    Attributes:
    - ekm_connections: List[EkmConnection] - List of EkmConnection objects
    - next_page_token: str - Token for next page
    - total_size: int - Total number of connections
    """

class UpdateEkmConfigRequest:
    """
    Request to update EkmConfig.
    
    Attributes:
    - ekm_config: EkmConfig - Required. Updated EkmConfig
    - update_mask: FieldMask - Required. Fields to update
    """

class VerifyConnectivityRequest:
    """
    Request to verify EKM connectivity.
    
    Attributes:
    - name: str - Required. EkmConnection resource name
    """

class VerifyConnectivityResponse:
    """
    Response from EKM connectivity verification.
    """

from google.cloud import kms

# Initialize the EKM Service client
client = kms.EkmServiceClient()

# Create an EKM connection
project_id = "my-project"
location_id = "us-central1"
connection_id = "my-ekm-connection"

# Define service resolver for EKM replica
service_resolver = kms.EkmConnection.ServiceResolver()
service_resolver.hostname = "ekm.example.com"
service_resolver.server_certificates = [
    kms.Certificate(raw_der=certificate_der_bytes)
]

# Create EKM connection
ekm_connection = kms.EkmConnection()
ekm_connection.service_resolvers = [service_resolver]
ekm_connection.key_management_mode = kms.EkmConnection.KeyManagementMode.MANUAL

parent = f"projects/{project_id}/locations/{location_id}"
created_connection = client.create_ekm_connection(
    request={
        "parent": parent,
        "ekm_connection_id": connection_id,
        "ekm_connection": ekm_connection,
    }
)

print(f"Created EKM connection: {created_connection.name}")

from google.cloud import kms
from google.api_core import exceptions

# Initialize the EKM Service client
client = kms.EkmServiceClient()

# Verify connectivity to EKM
connection_name = f"projects/{project_id}/locations/{location_id}/ekmConnections/{connection_id}"

try:
    verify_response = client.verify_connectivity(
        request={"name": connection_name}
    )
    print("EKM connectivity verified successfully")
except exceptions.FailedPrecondition as e:
    print(f"Connectivity verification failed: {e}")
except Exception as e:
    print(f"Error verifying connectivity: {e}")

from google.cloud import kms

# Initialize the EKM Service client
client = kms.EkmServiceClient()

# Set default EKM connection for a location
config_name = f"projects/{project_id}/locations/{location_id}/ekmConfig"
default_connection = f"projects/{project_id}/locations/{location_id}/ekmConnections/{connection_id}"

# Update EKM configuration
ekm_config = kms.EkmConfig()
ekm_config.name = config_name
ekm_config.default_ekm_connection = default_connection

updated_config = client.update_ekm_config(
    request={
        "ekm_config": ekm_config,
        "update_mask": {"paths": ["default_ekm_connection"]},
    }
)

print(f"Default EKM connection set to: {updated_config.default_ekm_connection}")

from google.cloud import kms

# Initialize both clients
kms_client = kms.KeyManagementServiceClient()
ekm_client = kms.EkmServiceClient()

# Create a crypto key that uses external key material
crypto_key = kms.CryptoKey()
crypto_key.purpose = kms.CryptoKey.CryptoKeyPurpose.ENCRYPT_DECRYPT

# Create version template for external key
version_template = kms.CryptoKeyVersionTemplate()
version_template.protection_level = kms.ProtectionLevel.EXTERNAL_VPC
version_template.algorithm = kms.CryptoKeyVersion.CryptoKeyVersionAlgorithm.EXTERNAL_SYMMETRIC_ENCRYPTION

crypto_key.version_template = version_template

# Create the crypto key
key_ring_name = f"projects/{project_id}/locations/{location_id}/keyRings/{key_ring_id}"
external_key = kms_client.create_crypto_key(
    request={
        "parent": key_ring_name,
        "crypto_key_id": "external-key",
        "crypto_key": crypto_key,
    }
)

print(f"Created external crypto key: {external_key.name}")

# Use the external key for encryption
plaintext = b"Data to encrypt with external key"
encrypt_response = kms_client.encrypt(
    request={
        "name": external_key.name,
        "plaintext": plaintext,
    }
)

print(f"Encrypted data using external key")

from google.cloud import kms

# Initialize the EKM Service client
client = kms.EkmServiceClient()

# List all EKM connections in a location
parent = f"projects/{project_id}/locations/{location_id}"
response = client.list_ekm_connections(
    request={
        "parent": parent,
        "page_size": 100,
    }
)

for connection in response.ekm_connections:
    print(f"EKM Connection: {connection.name}")
    print(f"  Key Management Mode: {connection.key_management_mode}")
    print(f"  Service Resolvers: {len(connection.service_resolvers)}")
    
    # Verify connectivity for each connection
    try:
        client.verify_connectivity(request={"name": connection.name})
        print(f"  Status: Connected")
    except Exception:
        print(f"  Status: Connection failed")

# When creating EKM connections, ensure certificates are properly configured
from google.cloud import kms

# Load certificate data (DER encoded)
with open("ekm-server-cert.der", "rb") as f:
    certificate_der_bytes = f.read()

# Create certificate object for EKM connection
certificate = kms.Certificate(raw_der=certificate_der_bytes)

# Note: Certificate validation should be performed by your organization's
# PKI infrastructure before deploying certificates to production EKM servers

# Configure EKM connection with proper network security
service_resolver = kms.EkmConnection.ServiceResolver()
service_resolver.hostname = "ekm.internal.example.com"  # Use internal hostname
service_resolver.endpoint_filter = "port=443"  # Ensure HTTPS

# Add multiple certificates for redundancy
service_resolver.server_certificates = [
    kms.Certificate(raw_der=primary_cert_der),
    kms.Certificate(raw_der=backup_cert_der),
]