Coverage for scripts / live_release_validation / ownership / vpc_endpoints.py: 100.00%

1"""Accepted-residue accounting for VPC endpoints EC2 has already deleted. 

2 

3The Resource Groups Tagging API is an index, not the source of truth, and it 

4lags resource deletion by minutes. Gateway VPC endpoints (the S3 and DynamoDB 

5endpoints the regional stack creates) are deleted synchronously with their 

6stack, yet the index kept returning both ARNs — still carrying their 

7``aws:cloudformation:stack-id`` tags — when ``final-inventory`` ran seconds 

8after the stack was gone. Observed live on 2026-09-12: two 

9``vpc-endpoint/vpce-…`` ARNs failed the all-zero teardown gate while 

10``DescribeVpcEndpoints`` already answered ``InvalidVpcEndpointId.NotFound`` for 

11both. The same stale entries would fail the next run's clean-account gate. 

12 

13``_strip_deleted_vpc_endpoints`` accepts exactly that shape and nothing else: a 

14``tagged_resources`` entry whose ARN parses as a VPC endpoint in the same 

15region and expected account is stripped **only after** EC2 itself confirms the 

16endpoint does not exist or is in a terminal ``deleting``/``deleted`` state. A 

17live endpoint keeps its entry as genuine residue. Every acceptance is returned 

18as evidence (region, endpoint id, the EC2 check performed, tags) so ``baseline`` 

19and ``final-inventory`` disclose what they tolerated rather than silently 

20ignoring it — the same posture as ``ownership/dynamodb_streams.py``. 

21""" 

22 

23from __future__ import annotations 

24 

25import copy 

26import re 

27from typing import Any 

28 

29from botocore.exceptions import ClientError 

30 

31from ..models import RunContext 

32 

33_VPC_ENDPOINT_ARN = re.compile( 

34 r"^arn:[^:]+:ec2:(?P<region>[a-z0-9-]+):(?P<account>\d{12})" 

35 r":vpc-endpoint/(?P<endpoint_id>vpce-[0-9a-f]+)$" 

36) 

37#: EC2 states in which the endpoint is already being removed and cannot be 

38#: residue; anything else (``available``, ``pending``, ``failed``, ...) is. 

39_TERMINAL_ENDPOINT_STATES = frozenset({"deleting", "deleted"}) 

40_NOT_FOUND_CODES = frozenset({"InvalidVpcEndpointId.NotFound", "InvalidVpcEndpoint.NotFound"}) 

41 

42 

43def _endpoint_state(ctx: RunContext, region: str, endpoint_id: str) -> str: 

44 """Return the live endpoint state, or ``ABSENT`` once EC2 no longer knows it.""" 

45 ec2 = ctx.session.client("ec2", region_name=region) 

46 try: 

47 response = ec2.describe_vpc_endpoints(VpcEndpointIds=[endpoint_id]) 

48 except ClientError as exc: 

49 if exc.response.get("Error", {}).get("Code") not in _NOT_FOUND_CODES: 

50 raise 

51 return "ABSENT" 

52 endpoints = response.get("VpcEndpoints") or [] 

53 if not endpoints: 

54 return "ABSENT" 

55 return str(endpoints[0].get("State") or "UNKNOWN") 

56 

57 

58def _strip_deleted_vpc_endpoints( 

59 ctx: RunContext, 

60 project_inventory: dict[str, Any], 

61) -> tuple[dict[str, Any], list[dict[str, Any]]]: 

62 """Strip tagged VPC endpoint ARNs whose endpoint EC2 proves gone.""" 

63 inventory = copy.deepcopy(project_inventory) 

64 accepted: list[dict[str, Any]] = [] 

65 for region, resources in list(inventory.get("regional", {}).items()): 

66 kept: list[dict[str, Any]] = [] 

67 for entry in resources.get("tagged_resources", []): 

68 arn = str(entry.get("arn") or "") 

69 match = _VPC_ENDPOINT_ARN.match(arn) 

70 if ( 

71 match is None 

72 or match.group("region") != region 

73 or match.group("account") != ctx.settings.expected_account 

74 ): 

75 kept.append(entry) 

76 continue 

77 endpoint_id = match.group("endpoint_id") 

78 state = _endpoint_state(ctx, region, endpoint_id) 

79 if state != "ABSENT" and state not in _TERMINAL_ENDPOINT_STATES: 

80 # EC2 still has it: genuine residue, keep it visible. 

81 kept.append(entry) 

82 continue 

83 accepted.append( 

84 { 

85 "region": region, 

86 "arn": arn, 

87 "endpoint_id": endpoint_id, 

88 "endpoint_state": state, 

89 "authority": "ec2:DescribeVpcEndpoints", 

90 "tags": dict(entry.get("tags") or {}), 

91 "note": ( 

92 "the Resource Groups Tagging API index lags endpoint deletion; " 

93 "EC2 is the authority for existence" 

94 ), 

95 } 

96 ) 

97 resources["tagged_resources"] = kept 

98 if not any(resources.values()): 

99 inventory["regional"].pop(region) 

100 return inventory, accepted