Coverage for scripts / live_release_validation / ownership / efs_automatic_backups.py: 100.00%
145 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Accepted residue for AWS-managed EFS automatic backup recovery points.
3Amazon EFS automatic backups are stored in the AWS-managed
4``aws/efs/automatic-backup-vault``. That vault denies manual recovery-point
5deletion and expires each point on its calculated lifecycle. A validation run
6that deletes its EFS file system can therefore leave one non-billable-by-stack,
7non-deletable recovery point for the retention window.
9This module removes only that exact shape from inventory after independently
10proving vault identity and policy, EFS resource identity and absence, project
11ownership, and scheduled deletion. It performs no AWS mutations.
12"""
14from __future__ import annotations
16import copy
17import re
18from collections.abc import Mapping
19from datetime import datetime
20from typing import Any
22from botocore.exceptions import ClientError
24from ..constants import _RUN_STACK_TAG
25from ..inventory._shared import _mapping_tags, _name_or_path_is_project_owned
26from ..json_utils import loads_without_duplicate_keys
27from ..models import RunContext
29_EFS_AUTOMATIC_BACKUP_VAULT = "aws/efs/automatic-backup-vault"
30_DELETE_RECOVERY_POINT_ACTION = "backup:DeleteRecoveryPoint"
31_RECOVERY_POINT_RESOURCE = re.compile(r"^recovery-point:[A-Za-z0-9-]+$")
32_EFS_FILE_SYSTEM_RESOURCE = re.compile(r"^file-system/(?P<id>fs-[0-9a-f]{8,40})$")
33_VALIDATION_RUN_TAG = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/+=@-]{0,255}$")
36def _string_values(value: object) -> list[str] | None:
37 if isinstance(value, str):
38 return [value]
39 if isinstance(value, list) and all(isinstance(item, str) for item in value):
40 return list(value)
41 return None
44def _all_principals(value: object) -> bool:
45 if value == "*":
46 return True
47 if not isinstance(value, Mapping) or set(value) != {"AWS"}:
48 return False
49 principals = _string_values(value["AWS"])
50 return principals is not None and principals == ["*"]
53def _policy_has_unconditional_delete_deny(policy_text: str, recovery_point_arn: str) -> bool:
54 try:
55 policy = loads_without_duplicate_keys(policy_text)
56 except ValueError as exc:
57 raise RuntimeError("EFS automatic backup vault policy is invalid JSON") from exc
58 if not isinstance(policy, dict):
59 raise RuntimeError("EFS automatic backup vault policy must be an object")
60 raw_statements = policy.get("Statement")
61 statements = raw_statements if isinstance(raw_statements, list) else [raw_statements]
62 for statement in statements:
63 if not isinstance(statement, dict) or statement.get("Effect") != "Deny":
64 continue
65 if any(
66 key in statement for key in ("Condition", "NotAction", "NotPrincipal", "NotResource")
67 ):
68 continue
69 actions = _string_values(statement.get("Action"))
70 resources = _string_values(statement.get("Resource"))
71 if (
72 actions is not None
73 and _DELETE_RECOVERY_POINT_ACTION in actions
74 and resources is not None
75 and ("*" in resources or recovery_point_arn in resources)
76 and _all_principals(statement.get("Principal"))
77 ):
78 return True
79 return False
82def _arn_parts(arn: str) -> tuple[str, str, str, str, str] | None:
83 parts = arn.split(":", 5)
84 if len(parts) != 6 or parts[0] != "arn":
85 return None
86 return parts[1], parts[2], parts[3], parts[4], parts[5]
89def _accepted_recovery_point(
90 ctx: RunContext,
91 *,
92 region: str,
93 recovery_point_arn: str,
94) -> dict[str, Any] | None:
95 partition = ctx.session.get_partition_for_region(region)
96 if not isinstance(partition, str) or not partition:
97 raise RuntimeError(f"Could not resolve AWS partition for automatic EFS backup in {region}")
98 expected_account = ctx.settings.expected_account
99 parsed = _arn_parts(recovery_point_arn)
100 if parsed is None:
101 return None
102 arn_partition, service, arn_region, account, resource = parsed
103 if (
104 arn_partition != partition
105 or service != "backup"
106 or arn_region != region
107 or account != expected_account
108 or _RECOVERY_POINT_RESOURCE.fullmatch(resource) is None
109 ):
110 return None
112 backup = ctx.session.client("backup", region_name=region)
113 description = backup.describe_recovery_point(
114 BackupVaultName=_EFS_AUTOMATIC_BACKUP_VAULT,
115 RecoveryPointArn=recovery_point_arn,
116 )
117 expected_vault_arn = (
118 f"arn:{partition}:backup:{region}:{expected_account}:"
119 f"backup-vault:{_EFS_AUTOMATIC_BACKUP_VAULT}"
120 )
121 if (
122 description.get("RecoveryPointArn") != recovery_point_arn
123 or description.get("BackupVaultName") != _EFS_AUTOMATIC_BACKUP_VAULT
124 or description.get("BackupVaultArn") != expected_vault_arn
125 ):
126 return None
127 source_vault_arn = str(description.get("SourceBackupVaultArn") or "")
128 if source_vault_arn and source_vault_arn != expected_vault_arn:
129 return None
130 if description.get("ResourceType") != "EFS" or description.get("Status") != "COMPLETED":
131 return None
133 resource_arn = str(description.get("ResourceArn") or "")
134 resource_parts = _arn_parts(resource_arn)
135 if resource_parts is None:
136 return None
137 resource_partition, resource_service, resource_region, resource_account, resource = (
138 resource_parts
139 )
140 file_system_match = _EFS_FILE_SYSTEM_RESOURCE.fullmatch(resource)
141 if (
142 resource_partition != partition
143 or resource_service != "elasticfilesystem"
144 or resource_region != region
145 or resource_account != expected_account
146 or file_system_match is None
147 ):
148 return None
149 resource_name = str(description.get("ResourceName") or "")
150 if not _name_or_path_is_project_owned(resource_name, ctx.config.project_name):
151 return None
153 lifecycle = description.get("CalculatedLifecycle")
154 delete_at = lifecycle.get("DeleteAt") if isinstance(lifecycle, Mapping) else None
155 if (
156 not isinstance(delete_at, datetime)
157 or delete_at.tzinfo is None
158 or delete_at.utcoffset() is None
159 ):
160 return None
162 tags = _mapping_tags(backup.list_tags(ResourceArn=recovery_point_arn).get("Tags"))
163 validation_run = str(tags.get(_RUN_STACK_TAG) or "")
164 if validation_run and _VALIDATION_RUN_TAG.fullmatch(validation_run) is None:
165 return None
167 vault = backup.describe_backup_vault(BackupVaultName=_EFS_AUTOMATIC_BACKUP_VAULT)
168 if (
169 vault.get("BackupVaultName") != _EFS_AUTOMATIC_BACKUP_VAULT
170 or vault.get("BackupVaultArn") != expected_vault_arn
171 ):
172 return None
173 policy_text = backup.get_backup_vault_access_policy(
174 BackupVaultName=_EFS_AUTOMATIC_BACKUP_VAULT
175 ).get("Policy")
176 if not isinstance(policy_text, str) or not _policy_has_unconditional_delete_deny(
177 policy_text, recovery_point_arn
178 ):
179 return None
181 file_system_id = file_system_match.group("id")
182 efs = ctx.session.client("efs", region_name=region)
183 try:
184 efs.describe_file_systems(FileSystemId=file_system_id)
185 except ClientError as exc:
186 if exc.response.get("Error", {}).get("Code") != "FileSystemNotFound":
187 raise
188 else:
189 return None
191 return {
192 "region": region,
193 "account": expected_account,
194 "partition": partition,
195 "recovery_point_arn": recovery_point_arn,
196 "backup_vault_name": _EFS_AUTOMATIC_BACKUP_VAULT,
197 "backup_vault_arn": expected_vault_arn,
198 "resource_type": "EFS",
199 "resource_name": resource_name,
200 "resource_arn": resource_arn,
201 "file_system_id": file_system_id,
202 "source_file_system_absent": True,
203 "source_absence_authority": "efs:DescribeFileSystems FileSystemNotFound",
204 "delete_at": delete_at.isoformat(),
205 "tags": tags,
206 "validation_run_tag": validation_run or None,
207 "vault_policy_unconditional_delete_deny": True,
208 "note": (
209 "AWS-managed EFS automatic backups deny manual deletion and expire "
210 "at their calculated lifecycle date"
211 ),
212 }
215def _strip_accepted_efs_automatic_backup_recovery_points(
216 ctx: RunContext,
217 project_inventory: dict[str, Any],
218) -> tuple[dict[str, Any], list[dict[str, Any]]]:
219 """Strip only proven, expiring AWS-managed backups of absent EFS sources."""
220 inventory = copy.deepcopy(project_inventory)
221 accepted: list[dict[str, Any]] = []
222 seen: set[tuple[str, str]] = set()
223 regional = inventory.get("regional")
224 if not isinstance(regional, dict):
225 return inventory, accepted
227 for raw_region, resources in list(regional.items()):
228 region = str(raw_region)
229 if not isinstance(resources, dict):
230 raise RuntimeError(f"Project inventory for {region} must be an object")
231 candidates = resources.get("backup_recovery_points", [])
232 if not isinstance(candidates, list):
233 raise RuntimeError(f"Backup recovery-point inventory for {region} must be a list")
234 kept: list[str] = []
235 accepted_arns: set[str] = set()
236 for candidate in candidates:
237 arn = str(candidate or "")
238 identity = (region, arn)
239 if not arn or identity in seen:
240 raise RuntimeError(f"Duplicate or empty backup recovery-point identity in {region}")
241 seen.add(identity)
242 evidence = _accepted_recovery_point(
243 ctx,
244 region=region,
245 recovery_point_arn=arn,
246 )
247 if evidence is None:
248 kept.append(arn)
249 continue
250 accepted_arns.add(arn)
251 accepted.append(evidence)
252 resources["backup_recovery_points"] = kept
254 tagged = resources.get("tagged_resources")
255 if tagged is not None:
256 if not isinstance(tagged, list):
257 raise RuntimeError(f"Tagged-resource inventory for {region} must be a list")
258 resources["tagged_resources"] = [
259 entry
260 for entry in tagged
261 if not (isinstance(entry, dict) and str(entry.get("arn") or "") in accepted_arns)
262 ]
263 if not any(resources.values()):
264 regional.pop(raw_region)
265 return inventory, accepted