Coverage for src / osiris_cli / cloud_manager.py: 0%

1""" 

2Cloud CLI Manager for Osiris 

3 

4Unified interface for AWS, GCP, and Azure cloud operations. 

5Wraps existing cloud CLIs with AI-friendly abstractions. 

6""" 

7 

8import subprocess 

9import json 

10from typing import Dict, List, Any, Optional 

11from enum import Enum 

12from dataclasses import dataclass 

13 

14from .logger import get_logger 

15from .errors import CloudError, CloudAuthenticationError, CloudResourceNotFoundError 

16 

17logger = get_logger() 

18 

19 

20class CloudProvider(Enum): 

21 """Supported cloud providers""" 

22 AWS = "aws" 

23 GCP = "gcp" 

24 AZURE = "azure" 

25 

26 

27@dataclass 

28class CloudResource: 

29 """Generic cloud resource representation""" 

30 provider: CloudProvider 

31 resource_type: str 

32 resource_id: str 

33 name: Optional[str] = None 

34 status: Optional[str] = None 

35 region: Optional[str] = None 

36 metadata: Dict[str, Any] = None 

37 

38 

39class CloudManager: 

40 """ 

41 Unified cloud operations manager. 

42  

43 Features: 

44 - AWS CLI wrapper (boto3 + aws cli) 

45 - GCP CLI wrapper (gcloud) 

46 - Azure CLI wrapper (az) 

47 - Natural language to CLI translation 

48 - Resource management 

49 """ 

50 

51 def __init__(self): 

52 """Initialize cloud manager""" 

53 self.authenticated = { 

54 CloudProvider.AWS: False, 

55 CloudProvider.GCP: False, 

56 CloudProvider.AZURE: False 

57 } 

58 logger.info("Cloud manager initialized") 

59 

60 def check_cli_available(self, provider: CloudProvider) -> bool: 

61 """Check if cloud CLI is available""" 

62 cli_commands = { 

63 CloudProvider.AWS: "aws --version", 

64 CloudProvider.GCP: "gcloud --version", 

65 CloudProvider.AZURE: "az --version" 

66 } 

67 

68 try: 

69 cmd = cli_commands[provider] 

70 subprocess.run( 

71 cmd.split(), 

72 capture_output=True, 

73 check=True, 

74 timeout=5 

75 ) 

76 return True 

77 except Exception as e: 

78 logger.warning(f"{provider.value} CLI not available: {e}") 

79 return False 

80 

81 def check_authentication(self, provider: CloudProvider) -> bool: 

82 """Check if authenticated with cloud provider""" 

83 try: 

84 if provider == CloudProvider.AWS: 

85 result = subprocess.run( 

86 ["aws", "sts", "get-caller-identity"], 

87 capture_output=True, 

88 timeout=10 

89 ) 

90 self.authenticated[provider] = result.returncode == 0 

91 

92 elif provider == CloudProvider.GCP: 

93 result = subprocess.run( 

94 ["gcloud", "auth", "list"], 

95 capture_output=True, 

96 timeout=10 

97 ) 

98 self.authenticated[provider] = result.returncode == 0 

99 

100 elif provider == CloudProvider.AZURE: 

101 result = subprocess.run( 

102 ["az", "account", "show"], 

103 capture_output=True, 

104 timeout=10 

105 ) 

106 self.authenticated[provider] = result.returncode == 0 

107 

108 return self.authenticated[provider] 

109 

110 except Exception as e: 

111 logger.error(f"Auth check failed for {provider.value}: {e}") 

112 return False 

113 

114 # AWS Operations 

115 def aws_list_ec2(self, region: str = "us-east-1") -> Dict[str, Any]: 

116 """List AWS EC2 instances""" 

117 if not self.check_authentication(CloudProvider.AWS): 

118 raise CloudAuthenticationError("AWS") 

119 

120 try: 

121 result = subprocess.run( 

122 [ 

123 "aws", "ec2", "describe-instances", 

124 "--region", region, 

125 "--output", "json" 

126 ], 

127 capture_output=True, 

128 text=True, 

129 timeout=30, 

130 check=True 

131 ) 

132 

133 data = json.loads(result.stdout) 

134 instances = [] 

135 

136 for reservation in data.get("Reservations", []): 

137 for instance in reservation.get("Instances", []): 

138 instances.append({ 

139 "id": instance.get("InstanceId"), 

140 "type": instance.get("InstanceType"), 

141 "state": instance.get("State", {}).get("Name"), 

142 "public_ip": instance.get("PublicIpAddress"), 

143 "private_ip": instance.get("PrivateIpAddress"), 

144 "launch_time": instance.get("LaunchTime") 

145 }) 

146 

147 return { 

148 "success": True, 

149 "provider": "aws", 

150 "resource_type": "ec2_instances", 

151 "region": region, 

152 "count": len(instances), 

153 "instances": instances 

154 } 

155 

156 except subprocess.CalledProcessError as e: 

157 logger.error(f"AWS EC2 list failed: {e}") 

158 raise CloudError(f"Failed to list EC2 instances: {e.stderr}") 

159 except Exception as e: 

160 logger.error(f"AWS operation failed: {e}") 

161 raise CloudError(str(e)) 

162 

163 def aws_list_s3(self) -> Dict[str, Any]: 

164 """List AWS S3 buckets""" 

165 if not self.check_authentication(CloudProvider.AWS): 

166 raise CloudAuthenticationError("AWS") 

167 

168 try: 

169 result = subprocess.run( 

170 ["aws", "s3api", "list-buckets", "--output", "json"], 

171 capture_output=True, 

172 text=True, 

173 timeout=30, 

174 check=True 

175 ) 

176 

177 data = json.loads(result.stdout) 

178 buckets = [ 

179 { 

180 "name": b.get("Name"), 

181 "created": b.get("CreationDate") 

182 } 

183 for b in data.get("Buckets", []) 

184 ] 

185 

186 return { 

187 "success": True, 

188 "provider": "aws", 

189 "resource_type": "s3_buckets", 

190 "count": len(buckets), 

191 "buckets": buckets 

192 } 

193 

194 except Exception as e: 

195 logger.error(f"AWS S3 list failed: {e}") 

196 raise CloudError(str(e)) 

197 

198 def aws_list_lambda(self, region: str = "us-east-1") -> Dict[str, Any]: 

199 """List AWS Lambda functions""" 

200 if not self.check_authentication(CloudProvider.AWS): 

201 raise CloudAuthenticationError("AWS") 

202 

203 try: 

204 result = subprocess.run( 

205 [ 

206 "aws", "lambda", "list-functions", 

207 "--region", region, 

208 "--output", "json" 

209 ], 

210 capture_output=True, 

211 text=True, 

212 timeout=30, 

213 check=True 

214 ) 

215 

216 data = json.loads(result.stdout) 

217 functions = [ 

218 { 

219 "name": f.get("FunctionName"), 

220 "runtime": f.get("Runtime"), 

221 "memory": f.get("MemorySize"), 

222 "timeout": f.get("Timeout"), 

223 "last_modified": f.get("LastModified") 

224 } 

225 for f in data.get("Functions", []) 

226 ] 

227 

228 return { 

229 "success": True, 

230 "provider": "aws", 

231 "resource_type": "lambda_functions", 

232 "region": region, 

233 "count": len(functions), 

234 "functions": functions 

235 } 

236 

237 except Exception as e: 

238 logger.error(f"AWS Lambda list failed: {e}") 

239 raise CloudError(str(e)) 

240 

241 # GCP Operations 

242 def gcp_list_compute(self, project: str, zone: str = "us-central1-a") -> Dict[str, Any]: 

243 """List GCP Compute Engine instances""" 

244 if not self.check_authentication(CloudProvider.GCP): 

245 raise CloudAuthenticationError("GCP") 

246 

247 try: 

248 result = subprocess.run( 

249 [ 

250 "gcloud", "compute", "instances", "list", 

251 "--project", project, 

252 "--zones", zone, 

253 "--format", "json" 

254 ], 

255 capture_output=True, 

256 text=True, 

257 timeout=30, 

258 check=True 

259 ) 

260 

261 instances = json.loads(result.stdout) 

262 

263 parsed = [ 

264 { 

265 "name": i.get("name"), 

266 "machine_type": i.get("machineType", "").split("/")[-1], 

267 "status": i.get("status"), 

268 "zone": i.get("zone", "").split("/")[-1], 

269 "external_ip": i.get("networkInterfaces", [{}])[0].get("accessConfigs", [{}])[0].get("natIP") 

270 } 

271 for i in instances 

272 ] 

273 

274 return { 

275 "success": True, 

276 "provider": "gcp", 

277 "resource_type": "compute_instances", 

278 "project": project, 

279 "zone": zone, 

280 "count": len(parsed), 

281 "instances": parsed 

282 } 

283 

284 except Exception as e: 

285 logger.error(f"GCP Compute list failed: {e}") 

286 raise CloudError(str(e)) 

287 

288 def gcp_list_storage(self, project: str) -> Dict[str, Any]: 

289 """List GCP Cloud Storage buckets""" 

290 if not self.check_authentication(CloudProvider.GCP): 

291 raise CloudAuthenticationError("GCP") 

292 

293 try: 

294 result = subprocess.run( 

295 [ 

296 "gcloud", "storage", "buckets", "list", 

297 "--project", project, 

298 "--format", "json" 

299 ], 

300 capture_output=True, 

301 text=True, 

302 timeout=30, 

303 check=True 

304 ) 

305 

306 buckets = json.loads(result.stdout) 

307 

308 parsed = [ 

309 { 

310 "name": b.get("name"), 

311 "location": b.get("location"), 

312 "storage_class": b.get("storageClass"), 

313 "created": b.get("timeCreated") 

314 } 

315 for b in buckets 

316 ] 

317 

318 return { 

319 "success": True, 

320 "provider": "gcp", 

321 "resource_type": "storage_buckets", 

322 "project": project, 

323 "count": len(parsed), 

324 "buckets": parsed 

325 } 

326 

327 except Exception as e: 

328 logger.error(f"GCP Storage list failed: {e}") 

329 raise CloudError(str(e)) 

330 

331 # Azure Operations 

332 def azure_list_vms(self, resource_group: Optional[str] = None) -> Dict[str, Any]: 

333 """List Azure Virtual Machines""" 

334 if not self.check_authentication(CloudProvider.AZURE): 

335 raise CloudAuthenticationError("Azure") 

336 

337 try: 

338 cmd = ["az", "vm", "list", "--output", "json"] 

339 if resource_group: 

340 cmd.extend(["--resource-group", resource_group]) 

341 

342 result = subprocess.run( 

343 cmd, 

344 capture_output=True, 

345 text=True, 

346 timeout=30, 

347 check=True 

348 ) 

349 

350 vms = json.loads(result.stdout) 

351 

352 parsed = [ 

353 { 

354 "name": vm.get("name"), 

355 "resource_group": vm.get("resourceGroup"), 

356 "location": vm.get("location"), 

357 "vm_size": vm.get("hardwareProfile", {}).get("vmSize"), 

358 "os": vm.get("storageProfile", {}).get("osDisk", {}).get("osType"), 

359 "provisioning_state": vm.get("provisioningState") 

360 } 

361 for vm in vms 

362 ] 

363 

364 return { 

365 "success": True, 

366 "provider": "azure", 

367 "resource_type": "virtual_machines", 

368 "resource_group": resource_group, 

369 "count": len(parsed), 

370 "vms": parsed 

371 } 

372 

373 except Exception as e: 

374 logger.error(f"Azure VM list failed: {e}") 

375 raise CloudError(str(e)) 

376 

377 def azure_list_storage(self, resource_group: Optional[str] = None) -> Dict[str, Any]: 

378 """List Azure Storage accounts""" 

379 if not self.check_authentication(CloudProvider.AZURE): 

380 raise CloudAuthenticationError("Azure") 

381 

382 try: 

383 cmd = ["az", "storage", "account", "list", "--output", "json"] 

384 if resource_group: 

385 cmd.extend(["--resource-group", resource_group]) 

386 

387 result = subprocess.run( 

388 cmd, 

389 capture_output=True, 

390 text=True, 

391 timeout=30, 

392 check=True 

393 ) 

394 

395 accounts = json.loads(result.stdout) 

396 

397 parsed = [ 

398 { 

399 "name": acc.get("name"), 

400 "resource_group": acc.get("resourceGroup"), 

401 "location": acc.get("location"), 

402 "sku": acc.get("sku", {}).get("name"), 

403 "kind": acc.get("kind"), 

404 "created": acc.get("creationTime") 

405 } 

406 for acc in accounts 

407 ] 

408 

409 return { 

410 "success": True, 

411 "provider": "azure", 

412 "resource_type": "storage_accounts", 

413 "resource_group": resource_group, 

414 "count": len(parsed), 

415 "accounts": parsed 

416 } 

417 

418 except Exception as e: 

419 logger.error(f"Azure Storage list failed: {e}") 

420 raise CloudError(str(e)) 

421 

422 def get_provider_status(self) -> Dict[str, Any]: 

423 """Get status of all cloud providers""" 

424 status = {} 

425 

426 for provider in CloudProvider: 

427 status[provider.value] = { 

428 "cli_available": self.check_cli_available(provider), 

429 "authenticated": self.check_authentication(provider) 

430 } 

431 

432 return status 

433 

434 

435# Global cloud manager instance 

436cloud_manager = CloudManager()