Coverage for scripts / live_release_validation / inventory / scanners.py: 100.00%

1"""Per-service read-only scanners for project-owned resources. 

2 

3Each ``_list_*`` helper answers one question — which resources of this 

4service does this project own in this Region — and fails closed by 

5returning nothing rather than guessing when a tag or name is ambiguous. 

6``project.collect_project_resources`` fans these out.""" 

7 

8from __future__ import annotations 

9 

10from collections.abc import Iterable 

11from typing import Any 

12from urllib.parse import urlparse 

13 

14from botocore.exceptions import ClientError 

15 

16from ._shared import ( 

17 _GLOBAL_ACCELERATOR_CONTROL_REGIONS, 

18 _arn_is_project_owned, 

19 _ec2_resource_is_project_owned, 

20 _iam_resource_is_project_owned, 

21 _mapping_tags, 

22 _name_or_path_is_project_owned, 

23 _project_owned_name, 

24 _tags_are_project_owned, 

25 _tags_to_dict, 

26) 

27from .ecr import ( 

28 collect_ecr_inventory, 

29) 

30 

31 

32def _list_eks_clusters( 

33 session: Any, 

34 region: str, 

35 project_name: str | None, 

36) -> list[str]: 

37 """List all clusters, optionally narrowing the authoritative result to the project.""" 

38 client = session.client("eks", region_name=region) 

39 names: set[str] = set() 

40 for page in client.get_paginator("list_clusters").paginate(): 

41 for raw_name in page.get("clusters", []): 

42 name = str(raw_name or "") 

43 if not name: 

44 raise RuntimeError(f"EKS returned a cluster without a name in {region}") 

45 names.add(name) 

46 if project_name is None: 

47 return sorted(names) 

48 return sorted(name for name in names if _project_owned_name(name, project_name)) 

49 

50 

51def _list_sqs_queues(session: Any, region: str, project_name: str) -> list[str]: 

52 client = session.client("sqs", region_name=region) 

53 urls: list[str] = [] 

54 for page in client.get_paginator("list_queues").paginate(QueueNamePrefix=project_name): 

55 for queue_url in page.get("QueueUrls", []): 

56 queue_name = urlparse(str(queue_url)).path.rsplit("/", 1)[-1] 

57 if _project_owned_name(queue_name, project_name): 

58 urls.append(str(queue_url)) 

59 return sorted(set(urls)) 

60 

61 

62def _list_dynamodb_tables(session: Any, region: str, project_name: str) -> list[str]: 

63 client = session.client("dynamodb", region_name=region) 

64 names: list[str] = [] 

65 for page in client.get_paginator("list_tables").paginate(): 

66 names.extend( 

67 str(name) 

68 for name in page.get("TableNames", []) 

69 if _project_owned_name(str(name), project_name) 

70 ) 

71 return sorted(set(names)) 

72 

73 

74def _list_load_balancers(session: Any, region: str, project_name: str) -> list[str]: 

75 client = session.client("elbv2", region_name=region) 

76 load_balancers: list[dict[str, Any]] = [] 

77 for page in client.get_paginator("describe_load_balancers").paginate(): 

78 load_balancers.extend(page.get("LoadBalancers", [])) 

79 

80 owned: list[str] = [] 

81 for start in range(0, len(load_balancers), 20): 

82 batch = load_balancers[start : start + 20] 

83 arns = [str(item["LoadBalancerArn"]) for item in batch] 

84 tags_by_arn = { 

85 str(item["ResourceArn"]): _tags_to_dict(item.get("Tags", [])) 

86 for item in client.describe_tags(ResourceArns=arns).get("TagDescriptions", []) 

87 } 

88 for load_balancer in batch: 

89 arn = str(load_balancer["LoadBalancerArn"]) 

90 name = str(load_balancer.get("LoadBalancerName") or "") 

91 if _project_owned_name(name, project_name) or _tags_are_project_owned( 

92 tags_by_arn.get(arn, {}), project_name 

93 ): 

94 owned.append(arn) 

95 return sorted(set(owned)) 

96 

97 

98def _list_target_groups(session: Any, region: str, project_name: str) -> list[str]: 

99 """Return ELBv2 target groups owned by a project or its EKS controller.""" 

100 client = session.client("elbv2", region_name=region) 

101 target_groups: list[dict[str, Any]] = [] 

102 for page in client.get_paginator("describe_target_groups").paginate(): 

103 target_groups.extend(page.get("TargetGroups", [])) 

104 

105 owned: set[str] = set() 

106 for start in range(0, len(target_groups), 20): 

107 batch = target_groups[start : start + 20] 

108 arns = [str(item.get("TargetGroupArn") or "") for item in batch] 

109 if any(not arn for arn in arns): 

110 raise RuntimeError(f"ELBv2 returned a target group without an ARN in {region}") 

111 tags_by_arn = { 

112 str(item["ResourceArn"]): _tags_to_dict(item.get("Tags", [])) 

113 for item in client.describe_tags(ResourceArns=arns).get("TagDescriptions", []) 

114 } 

115 for target_group, arn in zip(batch, arns, strict=True): 

116 tags = tags_by_arn.get(arn, {}) 

117 cluster_names = { 

118 tags.get("elbv2.k8s.aws/cluster", ""), 

119 tags.get("eks:eks-cluster-name", ""), 

120 } 

121 name = str(target_group.get("TargetGroupName") or "") 

122 if ( 

123 _project_owned_name(name, project_name) 

124 or _tags_are_project_owned(tags, project_name) 

125 or any(_project_owned_name(cluster, project_name) for cluster in cluster_names) 

126 ): 

127 owned.add(arn) 

128 return sorted(owned) 

129 

130 

131def _list_instance_inventory( 

132 session: Any, 

133 region: str, 

134 project_name: str, 

135) -> tuple[list[str], list[str]]: 

136 """Return project-owned and all active EC2 instance IDs separately.""" 

137 client = session.client("ec2", region_name=region) 

138 state_filter = { 

139 "Name": "instance-state-name", 

140 "Values": ["pending", "running", "stopping", "stopped", "shutting-down"], 

141 } 

142 project_instance_ids: set[str] = set() 

143 all_instance_ids: set[str] = set() 

144 paginator = client.get_paginator("describe_instances") 

145 for page in paginator.paginate(Filters=[state_filter]): 

146 for reservation in page.get("Reservations", []): 

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

148 instance_id = str(instance.get("InstanceId") or "") 

149 if not instance_id: 

150 raise RuntimeError(f"EC2 returned an instance without an ID in {region}") 

151 all_instance_ids.add(instance_id) 

152 if _ec2_resource_is_project_owned(instance, project_name): 

153 project_instance_ids.add(instance_id) 

154 return sorted(project_instance_ids), sorted(all_instance_ids) 

155 

156 

157def _list_instances(session: Any, region: str, project_name: str) -> list[str]: 

158 return _list_instance_inventory(session, region, project_name)[0] 

159 

160 

161def _list_project_kms_keys( 

162 session: Any, 

163 region: str, 

164 project_name: str, 

165 validation_run_id: str | None = None, 

166) -> list[dict[str, Any]]: 

167 """List project keys while isolating prior validation runs pending deletion.""" 

168 client = session.client("kms", region_name=region) 

169 keys: list[dict[str, Any]] = [] 

170 for page in client.get_paginator("list_keys").paginate(): 

171 for summary in page.get("Keys", []): 

172 key_id = str(summary.get("KeyId") or "") 

173 if not key_id: 

174 continue 

175 metadata = client.describe_key(KeyId=key_id).get("KeyMetadata", {}) 

176 if metadata.get("KeyManager") != "CUSTOMER": 

177 continue 

178 tags: dict[str, str] = {} 

179 marker: str | None = None 

180 while True: 

181 kwargs = {"KeyId": key_id} 

182 if marker: 

183 kwargs["Marker"] = marker 

184 response = client.list_resource_tags(**kwargs) 

185 tags.update( 

186 { 

187 str(tag["TagKey"]): str(tag.get("TagValue") or "") 

188 for tag in response.get("Tags", []) 

189 if tag.get("TagKey") is not None 

190 } 

191 ) 

192 marker = response.get("NextMarker") if response.get("Truncated") else None 

193 if not marker: 

194 break 

195 project_owned = _tags_are_project_owned(tags, project_name) 

196 validation_owner = tags.get("GcoLiveValidationRun") 

197 state = str(metadata.get("KeyState") or "") 

198 if validation_run_id: 

199 if validation_owner and validation_owner != validation_run_id: 

200 if state == "PendingDeletion": 

201 # Successful runs leave exact keys pending for seven days. 

202 # They are neither baseline contamination nor authority for 

203 # this run; active keys from another run still fail closed. 

204 continue 

205 elif not validation_owner and not project_owned: 

206 continue 

207 elif not project_owned: 

208 continue 

209 deletion_date = metadata.get("DeletionDate") 

210 keys.append( 

211 { 

212 "key_id": key_id, 

213 "arn": str(metadata.get("Arn") or ""), 

214 "state": str(metadata.get("KeyState") or ""), 

215 "description": str(metadata.get("Description") or ""), 

216 "deletion_date": ( 

217 deletion_date.isoformat() if deletion_date is not None else None 

218 ), 

219 "tags": tags, 

220 } 

221 ) 

222 return sorted(keys, key=lambda item: (item["arn"], item["key_id"])) 

223 

224 

225def _list_project_ecr_repositories( 

226 session: Any, 

227 region: str, 

228 project_name: str, 

229) -> list[str]: 

230 repositories = collect_ecr_inventory(session, [region])[region] 

231 return sorted( 

232 item["name"] for item in repositories if _project_owned_name(item["name"], project_name) 

233 ) 

234 

235 

236def _global_accelerator_control_region(session: Any, seed_region: str) -> str | None: 

237 """Return the partition's supported Global Accelerator control Region.""" 

238 partition = session.get_partition_for_region(seed_region) 

239 control_region = _GLOBAL_ACCELERATOR_CONTROL_REGIONS.get(str(partition)) 

240 if control_region is None: 

241 return None 

242 service_regions = set( 

243 session.get_available_regions("globalaccelerator", partition_name=partition) 

244 ) 

245 if control_region not in service_regions: 

246 raise RuntimeError( 

247 "AWS SDK does not advertise the required Global Accelerator control Region " 

248 f"{control_region} for partition {partition}" 

249 ) 

250 return control_region 

251 

252 

253def _list_global_accelerators( 

254 session: Any, 

255 control_region: str | None, 

256 project_name: str, 

257) -> list[str]: 

258 if control_region is None: 

259 return [] 

260 client = session.client("globalaccelerator", region_name=control_region) 

261 accelerators: list[dict[str, Any]] = [] 

262 token: str | None = None 

263 while True: 

264 kwargs = {"NextToken": token} if token else {} 

265 response = client.list_accelerators(**kwargs) 

266 accelerators.extend(response.get("Accelerators", [])) 

267 token = response.get("NextToken") 

268 if not token: 

269 break 

270 

271 owned: list[str] = [] 

272 for accelerator in accelerators: 

273 arn = str(accelerator.get("AcceleratorArn") or "") 

274 name = str(accelerator.get("Name") or "") 

275 tags = _tags_to_dict(client.list_tags_for_resource(ResourceArn=arn).get("Tags", [])) 

276 if _project_owned_name(name, project_name) or _tags_are_project_owned(tags, project_name): 

277 owned.append(arn) 

278 return sorted(set(owned)) 

279 

280 

281def _list_project_tagged_resources( 

282 session: Any, 

283 region: str, 

284 project_name: str, 

285) -> list[dict[str, Any]]: 

286 """List project-scoped resources exposed by the regional Tagging API.""" 

287 client = session.client("resourcegroupstaggingapi", region_name=region) 

288 resources: dict[str, dict[str, Any]] = {} 

289 for page in client.get_paginator("get_resources").paginate(): 

290 for mapping in page.get("ResourceTagMappingList", []): 

291 arn = str(mapping.get("ResourceARN") or "") 

292 if not arn: 

293 raise RuntimeError(f"Resource Groups Tagging API omitted an ARN in {region}") 

294 tags = _tags_to_dict(mapping.get("Tags", [])) 

295 if _tags_are_project_owned(tags, project_name) or _arn_is_project_owned( 

296 arn, project_name 

297 ): 

298 resources[arn] = {"arn": arn, "tags": tags} 

299 return [resources[arn] for arn in sorted(resources)] 

300 

301 

302def _ec2_items(client: Any, operation: str, response_key: str) -> list[dict[str, Any]]: 

303 items: list[dict[str, Any]] = [] 

304 for page in client.get_paginator(operation).paginate(): 

305 items.extend(page.get(response_key, [])) 

306 return items 

307 

308 

309_CLUSTER_TAG_PREFIX = "kubernetes.io/cluster/" 

310 

311#: Volume states that mean EC2 is already removing the volume, so it is not 

312#: residual. Everything else -- ``available``, ``in-use``, ``creating``, 

313#: ``error`` -- counts against the all-zero teardown gate. 

314_VOLUME_TERMINAL_STATES = frozenset({"deleting", "deleted"}) 

315 

316 

317def _list_cluster_volumes(session: Any, region: str, project_name: str) -> list[str]: 

318 """Return EBS volumes tagged for a project cluster by the EKS CSI driver. 

319 

320 These are the only project resources whose sole ownership marker is a 

321 Kubernetes tag: the CSI driver writes ``kubernetes.io/cluster/<cluster>`` 

322 and never the CloudFormation or ``gco:project`` tags every other scanner 

323 matches on. Deleting a cluster does not delete the PersistentVolumes its 

324 driver provisioned, so without this scanner a teardown could strand 

325 billable volumes and still pass the all-zero final-inventory gate. 

326 

327 Ownership is decided from the cluster name inside the tag key, which is the 

328 regional stack name, so a volume belonging to another project's cluster in 

329 the same account is never claimed. 

330 

331 Volumes EC2 is already removing (``deleting``/``deleted``) are not counted. 

332 The cluster tag is not exclusive to CSI-provisioned PersistentVolumes: EKS 

333 Auto Mode writes it onto node root volumes too, which live and die with 

334 their instances. Those pass through ``deleting`` while a cluster tears 

335 down, and a run that observed one mid-transition would fail the all-zero 

336 gate for a volume that was already on its way out. Every other state -- 

337 ``available``, ``in-use``, ``creating``, ``error`` -- still counts, because 

338 after teardown nothing should be holding a cluster-tagged volume at all. 

339 

340 Enumerates unfiltered and matches client-side, like the other EC2 scanners 

341 here. A server-side ``tag-key`` filter would need wildcard semantics to 

342 match the cluster-name suffix, and depending on that in the gate that 

343 authorizes calling a teardown clean is not worth the round-trip saved. 

344 """ 

345 client = session.client("ec2", region_name=region) 

346 volume_ids: set[str] = set() 

347 for volume in _ec2_items(client, "describe_volumes", "Volumes"): 

348 volume_id = str(volume.get("VolumeId") or "") 

349 if not volume_id: 

350 raise RuntimeError(f"EC2 returned a volume without an ID in {region}") 

351 if str(volume.get("State") or "") in _VOLUME_TERMINAL_STATES: 

352 continue 

353 for key in _tags_to_dict(volume.get("Tags", [])): 

354 if not key.startswith(_CLUSTER_TAG_PREFIX): 

355 continue 

356 cluster_name = key[len(_CLUSTER_TAG_PREFIX) :] 

357 if cluster_name and _project_owned_name(cluster_name, project_name): 

358 volume_ids.add(volume_id) 

359 break 

360 return sorted(volume_ids) 

361 

362 

363def _list_project_ec2_networking( 

364 session: Any, 

365 region: str, 

366 project_name: str, 

367 project_instance_ids: Iterable[str], 

368) -> tuple[dict[str, list[str]], dict[str, list[str]]]: 

369 """Return project-owned networking and unfiltered live ID authority separately.""" 

370 client = session.client("ec2", region_name=region) 

371 

372 vpc_ids: set[str] = set() 

373 all_vpc_ids: set[str] = set() 

374 for vpc in _ec2_items(client, "describe_vpcs", "Vpcs"): 

375 vpc_id = str(vpc.get("VpcId") or "") 

376 if not vpc_id: 

377 raise RuntimeError(f"EC2 returned a VPC without an ID in {region}") 

378 all_vpc_ids.add(vpc_id) 

379 if _ec2_resource_is_project_owned(vpc, project_name): 

380 vpc_ids.add(vpc_id) 

381 

382 subnet_ids: set[str] = set() 

383 all_subnet_ids: set[str] = set() 

384 for subnet in _ec2_items(client, "describe_subnets", "Subnets"): 

385 subnet_id = str(subnet.get("SubnetId") or "") 

386 if not subnet_id: 

387 raise RuntimeError(f"EC2 returned a subnet without an ID in {region}") 

388 all_subnet_ids.add(subnet_id) 

389 if ( 

390 _ec2_resource_is_project_owned(subnet, project_name) 

391 or str(subnet.get("VpcId") or "") in vpc_ids 

392 ): 

393 subnet_ids.add(subnet_id) 

394 

395 nat_gateway_ids: set[str] = set() 

396 all_nat_gateway_ids: set[str] = set() 

397 for nat_gateway in _ec2_items(client, "describe_nat_gateways", "NatGateways"): 

398 nat_gateway_id = str(nat_gateway.get("NatGatewayId") or "") 

399 if not nat_gateway_id: 

400 raise RuntimeError(f"EC2 returned a NAT gateway without an ID in {region}") 

401 if str(nat_gateway.get("State") or "") == "deleted": 

402 continue 

403 all_nat_gateway_ids.add(nat_gateway_id) 

404 if ( 

405 _ec2_resource_is_project_owned(nat_gateway, project_name) 

406 or str(nat_gateway.get("VpcId") or "") in vpc_ids 

407 or str(nat_gateway.get("SubnetId") or "") in subnet_ids 

408 ): 

409 nat_gateway_ids.add(nat_gateway_id) 

410 

411 security_group_ids: set[str] = set() 

412 all_security_group_ids: set[str] = set() 

413 for security_group in _ec2_items(client, "describe_security_groups", "SecurityGroups"): 

414 group_id = str(security_group.get("GroupId") or "") 

415 if not group_id: 

416 raise RuntimeError(f"EC2 returned a security group without an ID in {region}") 

417 all_security_group_ids.add(group_id) 

418 if ( 

419 _ec2_resource_is_project_owned(security_group, project_name) 

420 or _project_owned_name(str(security_group.get("GroupName") or ""), project_name) 

421 or str(security_group.get("VpcId") or "") in vpc_ids 

422 ): 

423 security_group_ids.add(group_id) 

424 

425 instance_ids = set(project_instance_ids) 

426 network_interface_ids: set[str] = set() 

427 all_network_interface_ids: set[str] = set() 

428 for interface in _ec2_items( 

429 client, 

430 "describe_network_interfaces", 

431 "NetworkInterfaces", 

432 ): 

433 interface_id = str(interface.get("NetworkInterfaceId") or "") 

434 if not interface_id: 

435 raise RuntimeError(f"EC2 returned a network interface without an ID in {region}") 

436 all_network_interface_ids.add(interface_id) 

437 group_ids = {str(group.get("GroupId") or "") for group in interface.get("Groups", [])} 

438 attachment_instance_id = str((interface.get("Attachment") or {}).get("InstanceId") or "") 

439 if ( 

440 _ec2_resource_is_project_owned(interface, project_name) 

441 or str(interface.get("VpcId") or "") in vpc_ids 

442 or str(interface.get("SubnetId") or "") in subnet_ids 

443 or bool(group_ids & security_group_ids) 

444 or attachment_instance_id in instance_ids 

445 ): 

446 network_interface_ids.add(interface_id) 

447 

448 flow_log_ids: set[str] = set() 

449 all_flow_log_ids: set[str] = set() 

450 project_network_resource_ids = vpc_ids | subnet_ids | network_interface_ids | instance_ids 

451 for flow_log in _ec2_items(client, "describe_flow_logs", "FlowLogs"): 

452 flow_log_id = str(flow_log.get("FlowLogId") or "") 

453 if not flow_log_id: 

454 raise RuntimeError(f"EC2 returned a flow log without an ID in {region}") 

455 all_flow_log_ids.add(flow_log_id) 

456 if ( 

457 _ec2_resource_is_project_owned(flow_log, project_name) 

458 or str(flow_log.get("ResourceId") or "") in project_network_resource_ids 

459 ): 

460 flow_log_ids.add(flow_log_id) 

461 

462 elastic_ip_ids: set[str] = set() 

463 all_elastic_ip_ids: set[str] = set() 

464 for address in client.describe_addresses().get("Addresses", []): 

465 identifier = str(address.get("AllocationId") or address.get("PublicIp") or "") 

466 if not identifier: 

467 raise RuntimeError(f"EC2 returned an Elastic IP without an identity in {region}") 

468 all_elastic_ip_ids.add(identifier) 

469 if ( 

470 _ec2_resource_is_project_owned(address, project_name) 

471 or str(address.get("NetworkInterfaceId") or "") in network_interface_ids 

472 or str(address.get("InstanceId") or "") in instance_ids 

473 ): 

474 elastic_ip_ids.add(identifier) 

475 

476 project_resources = { 

477 "vpcs": sorted(vpc_ids), 

478 "subnets": sorted(subnet_ids), 

479 "nat_gateways": sorted(nat_gateway_ids), 

480 "flow_logs": sorted(flow_log_ids), 

481 "network_interfaces": sorted(network_interface_ids), 

482 "security_groups": sorted(security_group_ids), 

483 "elastic_ips": sorted(elastic_ip_ids), 

484 } 

485 authoritative_resources = { 

486 "vpcs": sorted(all_vpc_ids), 

487 "subnets": sorted(all_subnet_ids), 

488 "nat_gateways": sorted(all_nat_gateway_ids), 

489 "flow_logs": sorted(all_flow_log_ids), 

490 "network_interfaces": sorted(all_network_interface_ids), 

491 "security_groups": sorted(all_security_group_ids), 

492 "elastic_ips": sorted(all_elastic_ip_ids), 

493 } 

494 return project_resources, authoritative_resources 

495 

496 

497def _list_lambda_functions(session: Any, region: str, project_name: str) -> list[str]: 

498 client = session.client("lambda", region_name=region) 

499 functions: set[str] = set() 

500 for page in client.get_paginator("list_functions").paginate(): 

501 for function in page.get("Functions", []): 

502 name = str(function.get("FunctionName") or "") 

503 arn = str(function.get("FunctionArn") or "") 

504 if not name or not arn: 

505 raise RuntimeError(f"Lambda returned a function without identity in {region}") 

506 tags = _mapping_tags(client.list_tags(Resource=arn).get("Tags")) 

507 if _project_owned_name(name, project_name) or _tags_are_project_owned( 

508 tags, project_name 

509 ): 

510 functions.add(arn) 

511 return sorted(functions) 

512 

513 

514def _list_api_gateway_v1_apis(session: Any, region: str, project_name: str) -> list[str]: 

515 client = session.client("apigateway", region_name=region) 

516 apis: set[str] = set() 

517 for page in client.get_paginator("get_rest_apis").paginate(): 

518 for api in page.get("items", []): 

519 api_id = str(api.get("id") or "") 

520 name = str(api.get("name") or "") 

521 tags = _mapping_tags(api.get("tags")) 

522 if _project_owned_name(name, project_name) or _tags_are_project_owned( 

523 tags, project_name 

524 ): 

525 if not api_id: 

526 raise RuntimeError(f"API Gateway v1 returned an API without an ID in {region}") 

527 apis.add(api_id) 

528 return sorted(apis) 

529 

530 

531def _list_api_gateway_v2_apis(session: Any, region: str, project_name: str) -> list[str]: 

532 client = session.client("apigatewayv2", region_name=region) 

533 apis: set[str] = set() 

534 for page in client.get_paginator("get_apis").paginate(): 

535 for api in page.get("Items", []): 

536 api_id = str(api.get("ApiId") or "") 

537 name = str(api.get("Name") or "") 

538 tags = _mapping_tags(api.get("Tags")) 

539 if _project_owned_name(name, project_name) or _tags_are_project_owned( 

540 tags, project_name 

541 ): 

542 if not api_id: 

543 raise RuntimeError(f"API Gateway v2 returned an API without an ID in {region}") 

544 apis.add(api_id) 

545 return sorted(apis) 

546 

547 

548def _list_cloudwatch_log_groups( 

549 session: Any, 

550 region: str, 

551 project_name: str, 

552) -> list[str]: 

553 client = session.client("logs", region_name=region) 

554 log_groups: set[str] = set() 

555 for page in client.get_paginator("describe_log_groups").paginate(): 

556 for log_group in page.get("logGroups", []): 

557 name = str(log_group.get("logGroupName") or "") 

558 arn = str(log_group.get("logGroupArn") or log_group.get("arn") or "").removesuffix(":*") 

559 if not name or not arn: 

560 raise RuntimeError( 

561 f"CloudWatch Logs returned a log group without identity in {region}" 

562 ) 

563 tags = _mapping_tags(client.list_tags_for_resource(resourceArn=arn).get("tags")) 

564 if _name_or_path_is_project_owned(name, project_name) or _tags_are_project_owned( 

565 tags, project_name 

566 ): 

567 log_groups.add(name) 

568 return sorted(log_groups) 

569 

570 

571def _list_secrets(session: Any, region: str, project_name: str) -> list[str]: 

572 client = session.client("secretsmanager", region_name=region) 

573 secrets: set[str] = set() 

574 for page in client.get_paginator("list_secrets").paginate(IncludePlannedDeletion=True): 

575 for secret in page.get("SecretList", []): 

576 name = str(secret.get("Name") or "") 

577 arn = str(secret.get("ARN") or "") 

578 tags = _tags_to_dict(secret.get("Tags", [])) 

579 if _project_owned_name(name, project_name) or _tags_are_project_owned( 

580 tags, project_name 

581 ): 

582 if not arn: 

583 raise RuntimeError( 

584 f"Secrets Manager returned a project-owned secret without an ARN in {region}" 

585 ) 

586 secrets.add(arn) 

587 return sorted(secrets) 

588 

589 

590def _list_s3_bucket_tags(client: Any, bucket_name: str) -> dict[str, str]: 

591 try: 

592 response = client.get_bucket_tagging(Bucket=bucket_name) 

593 except ClientError as exc: 

594 if exc.response.get("Error", {}).get("Code") in {"NoSuchTagSet", "NoSuchTagSetError"}: 

595 return {} 

596 raise 

597 return _tags_to_dict(response.get("TagSet", [])) 

598 

599 

600def _list_project_s3_buckets(session: Any, seed_region: str, project_name: str) -> list[str]: 

601 client = session.client("s3", region_name=seed_region) 

602 buckets: set[str] = set() 

603 for bucket in client.list_buckets().get("Buckets", []): 

604 name = str(bucket.get("Name") or "") 

605 if not name: 

606 raise RuntimeError("S3 returned a bucket without a name") 

607 tags = _list_s3_bucket_tags(client, name) 

608 if _project_owned_name(name, project_name) or _tags_are_project_owned(tags, project_name): 

609 buckets.add(name) 

610 return sorted(buckets) 

611 

612 

613def _list_iam_tags( 

614 client: Any, 

615 operation: str, 

616 identifier_name: str, 

617 identifier: str, 

618) -> dict[str, str]: 

619 tags: dict[str, str] = {} 

620 marker: str | None = None 

621 while True: 

622 kwargs: dict[str, Any] = {identifier_name: identifier} 

623 if marker: 

624 kwargs["Marker"] = marker 

625 response = getattr(client, operation)(**kwargs) 

626 tags.update(_tags_to_dict(response.get("Tags", []))) 

627 if not response.get("IsTruncated"): 

628 return tags 

629 marker = str(response.get("Marker") or "") 

630 if not marker: 

631 raise RuntimeError(f"IAM {operation} truncated its response without a Marker") 

632 

633 

634def _list_project_iam_resources( 

635 session: Any, 

636 seed_region: str, 

637 project_name: str, 

638) -> dict[str, list[str]]: 

639 client = session.client("iam", region_name=seed_region) 

640 resources: dict[str, set[str]] = { 

641 "iam_roles": set(), 

642 "iam_policies": set(), 

643 "iam_instance_profiles": set(), 

644 "iam_users": set(), 

645 "iam_groups": set(), 

646 } 

647 

648 for page in client.get_paginator("list_roles").paginate(): 

649 for role in page.get("Roles", []): 

650 name = str(role.get("RoleName") or "") 

651 arn = str(role.get("Arn") or "") 

652 if not name or not arn: 

653 raise RuntimeError("IAM returned a role without identity") 

654 tags = _list_iam_tags(client, "list_role_tags", "RoleName", name) 

655 if _iam_resource_is_project_owned( 

656 name, str(role.get("Path") or ""), tags, project_name 

657 ): 

658 resources["iam_roles"].add(arn) 

659 

660 for page in client.get_paginator("list_policies").paginate(Scope="Local"): 

661 for policy in page.get("Policies", []): 

662 name = str(policy.get("PolicyName") or "") 

663 arn = str(policy.get("Arn") or "")