SSH Key Management

def add_public_key(
    self,
    request: Optional[Union[cloudshell.AddPublicKeyRequest, dict]] = None,
    *,
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
    timeout: Union[float, object] = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
) -> operation.Operation:
    """
    Add public SSH key to environment. This is a long-running operation.
    
    Args:
        request: AddPublicKeyRequest object or dict
        retry: Retry configuration for the request
        timeout: Timeout for the request
        metadata: Additional metadata to send with the request
        
    Returns:
        Long-running operation that resolves to AddPublicKeyResponse
        
    Raises:
        google.api_core.exceptions.GoogleAPICallError: API call failed
        google.api_core.exceptions.InvalidArgument: Invalid SSH key format
    """

def remove_public_key(
    self,
    request: Optional[Union[cloudshell.RemovePublicKeyRequest, dict]] = None,
    *,
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
    timeout: Union[float, object] = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
) -> operation.Operation:
    """
    Remove public SSH key from environment. This is a long-running operation.
    
    Args:
        request: RemovePublicKeyRequest object or dict
        retry: Retry configuration for the request
        timeout: Timeout for the request
        metadata: Additional metadata to send with the request
        
    Returns:
        Long-running operation that resolves to RemovePublicKeyResponse
        
    Raises:
        google.api_core.exceptions.GoogleAPICallError: API call failed
        google.api_core.exceptions.NotFound: SSH key not found
    """

class AddPublicKeyRequest(proto.Message):
    """Request to add SSH public key to environment."""
    
    environment: str  # Environment name (e.g., users/me/environments/default)
    key: str         # SSH public key in supported format (OpenSSH format)

class RemovePublicKeyRequest(proto.Message):
    """Request to remove SSH public key from environment."""
    
    environment: str  # Environment name (e.g., users/me/environments/default)
    key: str         # SSH public key to remove (must match exactly)

class AddPublicKeyResponse(proto.Message):
    """Response from add public key operation."""
    
    key: str  # Key that was added (normalized format)

class RemovePublicKeyResponse(proto.Message):
    """Response from remove public key operation."""
    # Empty message

class AddPublicKeyMetadata(proto.Message):
    """Operation metadata for add key operations."""
    # Empty message

class RemovePublicKeyMetadata(proto.Message):
    """Operation metadata for remove key operations."""
    # Empty message

from google.cloud.shell import CloudShellServiceClient, AddPublicKeyRequest

client = CloudShellServiceClient()

# Read SSH public key from file
with open("~/.ssh/id_rsa.pub", "r") as f:
    public_key = f.read().strip()

request = AddPublicKeyRequest(
    environment="users/me/environments/default",
    key=public_key
)

operation = client.add_public_key(request=request)
print(f"Operation name: {operation.name}")

# Wait for completion
response = operation.result()
print(f"Added key: {response.key[:50]}...")

from google.cloud.shell import CloudShellServiceClient, RemovePublicKeyRequest

client = CloudShellServiceClient()

# Get current environment to see existing keys
environment = client.get_environment(
    name="users/me/environments/default"
)

if environment.public_keys:
    # Remove the first key
    key_to_remove = environment.public_keys[0]
    
    request = RemovePublicKeyRequest(
        environment="users/me/environments/default",
        key=key_to_remove
    )
    
    operation = client.remove_public_key(request=request)
    operation.result()
    print("SSH key removed successfully")
else:
    print("No SSH keys to remove")

from google.cloud.shell import CloudShellServiceClient
import os
import glob

client = CloudShellServiceClient()

# Add multiple SSH keys from ~/.ssh directory
ssh_dir = os.path.expanduser("~/.ssh")
pub_key_files = glob.glob(os.path.join(ssh_dir, "*.pub"))

for key_file in pub_key_files:
    with open(key_file, "r") as f:
        public_key = f.read().strip()
    
    operation = client.add_public_key(
        environment="users/me/environments/default",
        key=public_key
    )
    
    try:
        response = operation.result(timeout=60)
        print(f"Added key from {key_file}")
    except Exception as e:
        print(f"Failed to add key from {key_file}: {e}")

from google.cloud.shell import CloudShellServiceClient

client = CloudShellServiceClient()

environment = client.get_environment(
    name="users/me/environments/default"
)

print(f"Environment has {len(environment.public_keys)} SSH keys:")
for i, key in enumerate(environment.public_keys):
    # Show first and last 20 characters of each key
    key_preview = f"{key[:20]}...{key[-20:]}" if len(key) > 40 else key
    print(f"  {i+1}. {key_preview}")

import asyncio
from google.cloud.shell import CloudShellServiceAsyncClient, AddPublicKeyRequest

async def manage_ssh_keys():
    async with CloudShellServiceAsyncClient() as client:
        # Add key
        with open("~/.ssh/id_ed25519.pub", "r") as f:
            public_key = f.read().strip()
        
        add_operation = await client.add_public_key(
            AddPublicKeyRequest(
                environment="users/me/environments/default",
                key=public_key
            )
        )
        
        response = await add_operation.result()
        print(f"Added SSH key: {response.key[:50]}...")
        
        # Verify key was added
        environment = await client.get_environment(
            name="users/me/environments/default"
        )
        print(f"Environment now has {len(environment.public_keys)} SSH keys")

asyncio.run(manage_ssh_keys())

from google.cloud.shell import CloudShellServiceClient, AddPublicKeyRequest
from google.api_core import exceptions

client = CloudShellServiceClient()

try:
    # Try to add invalid SSH key
    operation = client.add_public_key(
        AddPublicKeyRequest(
            environment="users/me/environments/default",
            key="invalid-ssh-key-format"
        )
    )
    response = operation.result()
except exceptions.InvalidArgument as e:
    print(f"Invalid SSH key format: {e}")
except exceptions.GoogleAPICallError as e:
    print(f"API call failed: {e}")

ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgQC7YjmZH3+l8f... user@example.com

ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAILhLxNd8s6+m... user@example.com

ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTY... user@example.com