Skip to content

Commit 3a1eb05

Browse files
KshitijaChoudhariKshitijaChoudhari
committed
Python TFE - ssh_key implementation
1 parent 0acddb3 commit 3a1eb05

7 files changed

Lines changed: 457 additions & 0 deletions

File tree

‎examples/ssh_keys.py‎

Lines changed: 174 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,174 @@
1+
#!/usr/bin/env python3
2+
"""SSH Keys Example Script.
3+
4+
This script demonstrates how to use the SSH Keys API to:
5+
1. List all SSH keys for an organization
6+
2. Create a new SSH key
7+
3. Read a specific SSH key
8+
4. Update an SSH key
9+
5. Delete an SSH key
10+
11+
IMPORTANT: SSH Keys API has special authentication requirements:
12+
- ❌ CANNOT use Organization Tokens (AT-*)
13+
- ✅ MUST use User Tokens or Team Tokens
14+
- ✅ MUST have 'manage VCS settings' permission
15+
16+
Before running this script:
17+
1. Create a User Token in Terraform Cloud:
18+
- Go to User Settings → Tokens (not Organization Settings)
19+
- Create new token with VCS management permissions
20+
2. Set TFE_TOKEN environment variable with your User token (not Organization token!)
21+
3. Set TFE_ORG environment variable with your organization name
22+
4. Set SSH_PRIVATE_KEY environment variable with your SSH private key
23+
"""
24+
25+
import os
26+
import sys
27+
28+
# Add the source directory to the path for direct execution
29+
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))
30+
31+
from tfe import TFEClient, TFEConfig
32+
from tfe.errors import NotFound, TFEError
33+
from tfe.models import SSHKeyCreateOptions, SSHKeyListOptions, SSHKeyUpdateOptions
34+
35+
# Configuration
36+
TFE_TOKEN = os.getenv("TFE_TOKEN")
37+
TFE_ORG = os.getenv("TFE_ORG")
38+
39+
# SSH private key from environment variable (API expects private key, not public)
40+
SSH_KEY_VALUE = os.getenv("SSH_PRIVATE_KEY")
41+
42+
43+
def check_token_type(token):
44+
"""Check and validate token type for SSH Keys API."""
45+
print("🔍 Token Analysis:")
46+
if token.startswith("AT-"):
47+
print(" Token Type: Organization Token (AT-*)")
48+
print(" ❌ SSH Keys API does NOT support Organization Tokens")
49+
print(" 💡 Please create a User Token instead")
50+
print("")
51+
print("🔧 To create a User Token:")
52+
print(" 1. Go to Terraform Cloud → User Settings → Tokens")
53+
print(" 2. Create new token with VCS management permissions")
54+
print(" 3. Replace TFE_TOKEN environment variable")
55+
return False
56+
elif token.startswith("TF-"):
57+
print(" Token Type: User Token (TF-*)")
58+
print(" ✅ SSH Keys API supports User Tokens")
59+
return True
60+
elif ".atlasv1." in token:
61+
print(" Token Type: User/Team Token (.atlasv1. format)")
62+
print(" ✅ SSH Keys API supports User/Team Tokens")
63+
return True
64+
else:
65+
print(f" Token Type: Unknown format ({token[:10]}...)")
66+
print(" 💡 Expected User Token (TF-*) or Team Token")
67+
return True # Allow unknown formats to try
68+
69+
70+
def main():
71+
"""Main function demonstrating SSH Keys API usage."""
72+
73+
# Validate environment variables
74+
if not TFE_TOKEN:
75+
print("❌ Error: TFE_TOKEN environment variable is required")
76+
print("💡 Create a User Token (not Organization Token) in Terraform Cloud")
77+
sys.exit(1)
78+
79+
if not TFE_ORG:
80+
print("❌ Error: TFE_ORG environment variable is required")
81+
sys.exit(1)
82+
83+
if not SSH_KEY_VALUE:
84+
print("❌ Error: SSH_PRIVATE_KEY environment variable is required")
85+
print("💡 Provide a valid SSH private key for testing")
86+
sys.exit(1)
87+
88+
# Check token type first
89+
if not check_token_type(TFE_TOKEN):
90+
sys.exit(1)
91+
92+
# Initialize the TFE client
93+
config = TFEConfig(token=TFE_TOKEN)
94+
client = TFEClient(config)
95+
96+
print(f"\nSSH Keys API Example for organization: {TFE_ORG}")
97+
print("=" * 50)
98+
99+
try:
100+
# 1. List existing SSH keys
101+
print("\n1. Listing SSH keys...")
102+
ssh_keys = client.ssh_keys.list(TFE_ORG)
103+
print(f"✅ Found {len(ssh_keys.items)} SSH keys:")
104+
for key in ssh_keys.items:
105+
print(f" - ID: {key.id}, Name: {key.name}")
106+
107+
# 2. Create a new SSH key
108+
print("\n2. Creating a new SSH key...")
109+
create_options = SSHKeyCreateOptions(
110+
name="Python TFE Example SSH Key", value=SSH_KEY_VALUE
111+
)
112+
113+
new_key = client.ssh_keys.create(TFE_ORG, create_options)
114+
print(f"✅ Created SSH key: {new_key.id} - {new_key.name}")
115+
116+
# 3. Read the SSH key we just created
117+
print("\n3. Reading the SSH key...")
118+
read_key = client.ssh_keys.read(new_key.id)
119+
print(f"✅ Read SSH key: {read_key.id} - {read_key.name}")
120+
121+
# 4. Update the SSH key
122+
print("\n4. Updating the SSH key...")
123+
update_options = SSHKeyUpdateOptions(name="Updated Python TFE Example SSH Key")
124+
125+
updated_key = client.ssh_keys.update(new_key.id, update_options)
126+
print(f"✅ Updated SSH key: {updated_key.id} - {updated_key.name}")
127+
128+
# 5. Delete the SSH key
129+
print("\n5. Deleting the SSH key...")
130+
client.ssh_keys.delete(new_key.id)
131+
print(f"✅ Deleted SSH key: {new_key.id}")
132+
133+
# 6. Verify deletion by listing again
134+
print("\n6. Verifying deletion...")
135+
ssh_keys_after = client.ssh_keys.list(TFE_ORG)
136+
print(f"✅ SSH keys after deletion: {len(ssh_keys_after.items)}")
137+
138+
# 7. Demonstrate pagination with options
139+
print("\n7. Demonstrating pagination options...")
140+
list_options = SSHKeyListOptions(page_size=5, page_number=1)
141+
paginated_keys = client.ssh_keys.list(TFE_ORG, list_options)
142+
print(f"✅ Page 1 with page size 5: {len(paginated_keys.items)} keys")
143+
print(f" Total pages: {paginated_keys.total_pages}")
144+
print(f" Total count: {paginated_keys.total_count}")
145+
146+
print("\n🎉 SSH Keys API example completed successfully!")
147+
148+
except NotFound as e:
149+
print(f"\n❌ SSH Keys API Error: {e}")
150+
print("\n💡 This error usually means:")
151+
print(" - Using Organization Token (not allowed)")
152+
print(" - SSH Keys feature not available")
153+
print(" - Insufficient permissions")
154+
print("\n🔧 Try using a User Token instead of Organization Token")
155+
sys.exit(1)
156+
157+
except TFEError as e:
158+
print(f"\n❌ TFE API Error: {e}")
159+
if hasattr(e, "status"):
160+
if e.status == 403:
161+
print("💡 Permission denied - check token type and permissions")
162+
elif e.status == 401:
163+
print("💡 Authentication failed - check token validity")
164+
elif e.status == 422:
165+
print("💡 Validation error - check SSH key format")
166+
sys.exit(1)
167+
168+
except Exception as e:
169+
print(f"\n❌ Unexpected error: {e}")
170+
sys.exit(1)
171+
172+
173+
if __name__ == "__main__":
174+
main()

‎src/tfe/client.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
from .resources.run_event import RunEvents
1818
from .resources.run_task import RunTasks
1919
from .resources.run_trigger import RunTriggers
20+
from .resources.ssh_keys import SSHKeys
2021
from .resources.state_version_outputs import StateVersionOutputs
2122
from .resources.state_versions import StateVersions
2223
from .resources.variable import Variables
@@ -69,5 +70,8 @@ def __init__(self, config: TFEConfig | None = None):
6970
self.query_runs = QueryRuns(self._transport)
7071
self.run_events = RunEvents(self._transport)
7172

73+
# SSH Keys
74+
self.ssh_keys = SSHKeys(self._transport)
75+
7276
def close(self) -> None:
7377
pass

‎src/tfe/errors.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,9 @@ class ErrStateVersionUploadNotSupported(TFEError): ...
9999
ERR_REQUIRED_PROJECT = "projects are required"
100100
ERR_PROJECT_MIN_LIMIT = "must specify at least one project"
101101

102+
# SSH Key Error Constants
103+
ERR_INVALID_SSH_KEY_ID = "invalid SSH key ID"
104+
102105

103106
class WorkspaceNotFound(NotFound): ...
104107

‎src/tfe/models/__init__.py‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,15 @@
111111
RegistryProviderReadOptions,
112112
)
113113

114+
# Re-export all SSH key types
115+
from .ssh_key import (
116+
SSHKey,
117+
SSHKeyCreateOptions,
118+
SSHKeyList,
119+
SSHKeyListOptions,
120+
SSHKeyUpdateOptions,
121+
)
122+
114123
# Define what should be available when importing with *
115124
__all__ = [
116125
# OAuth client types
@@ -124,6 +133,12 @@
124133
"OAuthClientRemoveProjectsOptions",
125134
"OAuthClientUpdateOptions",
126135
"ServiceProviderType",
136+
# SSH key types
137+
"SSHKey",
138+
"SSHKeyCreateOptions",
139+
"SSHKeyList",
140+
"SSHKeyListOptions",
141+
"SSHKeyUpdateOptions",
127142
# Agent and agent pool types
128143
"Agent",
129144
"AgentPool",

‎src/tfe/models/ssh_key.py‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
from __future__ import annotations
2+
3+
from pydantic import BaseModel, ConfigDict, Field
4+
5+
6+
class SSHKey(BaseModel):
7+
"""Represents an SSH key in Terraform Enterprise."""
8+
9+
model_config = ConfigDict(populate_by_name=True)
10+
11+
id: str = Field(..., description="The unique identifier for this SSH key")
12+
type: str = Field(default="ssh-keys", description="The type of this resource")
13+
name: str = Field(..., description="A name to identify the SSH key")
14+
15+
16+
class SSHKeyCreateOptions(BaseModel):
17+
"""Options for creating a new SSH key."""
18+
19+
model_config = ConfigDict(populate_by_name=True)
20+
21+
name: str = Field(..., description="A name to identify the SSH key")
22+
value: str = Field(..., description="The text of the SSH private key")
23+
24+
25+
class SSHKeyUpdateOptions(BaseModel):
26+
"""Options for updating an SSH key."""
27+
28+
model_config = ConfigDict(populate_by_name=True)
29+
30+
name: str | None = Field(None, description="A name to identify the SSH key")
31+
32+
33+
class SSHKeyListOptions(BaseModel):
34+
"""Options for listing SSH keys."""
35+
36+
model_config = ConfigDict(populate_by_name=True)
37+
38+
page_number: int | None = Field(
39+
None, alias="page[number]", description="Page number to retrieve", ge=1
40+
)
41+
page_size: int | None = Field(
42+
None, alias="page[size]", description="Number of items per page", ge=1, le=100
43+
)
44+
45+
46+
class SSHKeyList(BaseModel):
47+
"""Represents a paginated list of SSH keys."""
48+
49+
model_config = ConfigDict(populate_by_name=True)
50+
51+
items: list[SSHKey] = Field(default_factory=list, description="List of SSH keys")
52+
current_page: int | None = Field(None, description="Current page number")
53+
total_pages: int | None = Field(None, description="Total number of pages")
54+
prev_page: str | None = Field(None, description="URL of the previous page")
55+
next_page: str | None = Field(None, description="URL of the next page")
56+
total_count: int | None = Field(None, description="Total number of items")

0 commit comments

Comments
 (0)