Setting the file. One moment.
Input Validator · Amazon Elasticache · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page 10
Setup DevOps Agent
33
AWS Deployment
61.5
Recipe Gallery
254
def main
— line 254
This file
Number 61.69
Position 69 of 79
Type Python
Size 11 KB
Lines 284 scripts/ input_validator.py
Python · 284 lines · 11 KB
17 python input_validator.py cache_name=my-cache region=us-east-1 subnet_id=subnet-abc123
18 """
19
20 from __future__ import annotations
21
22 import re
23 import sys
24 from typing import Optional
25
26 # ---------------------------------------------------------------------------
27 # Individual validators
28 # ---------------------------------------------------------------------------
29 # Each function returns (is_valid: bool, error_message: str | None).
30 # A None error_message means the value is valid.
31 # ---------------------------------------------------------------------------
32
33
34 def validate_cache_name (name: str ) -> tuple[ bool , Optional[ str ]]:
35 """Validate an ElastiCache cache or cluster name.
36
37 Rules:
38 - 1 to 40 characters
39 - Alphanumeric characters and hyphens only
40 - Must start with a letter
41 - No consecutive hyphens
42 - Must not end with a hyphen
43
44 Note: Serverless cache names must be lowercase. Node-based cluster names
45 are case-insensitive (AWS lowercases them). This validator enforces
46 lowercase to ensure compatibility with both deployment types.
47 """
48 if not name:
49 return False , "Cache name must not be empty"
50 if len (name) > 40 :
51 return False , f "Cache name must be 1-40 characters (got { len (name) } )"
52 if not re.match( r ' ^[ a-z ] ' , name):
53 return False , "Cache name must start with a lowercase letter (serverless requires lowercase; node-based is case-insensitive but lowercased by AWS)"
54 if not re.match( r ' ^[ a-z0-9- ] + $ ' , name):
55 return False , "Cache name may only contain lowercase alphanumeric characters and hyphens"
56 if '--' in name:
57 return False , "Cache name must not contain consecutive hyphens"
58 if name.endswith( '-' ):
59 return False , "Cache name must not end with a hyphen"
60 return True , None
61
62
63 def validate_replication_group_id (rg_id: str ) -> tuple[ bool , Optional[ str ]]:
64 """Validate a replication group ID.
65
66 Same rules as cache name: 1-40 chars, alphanumeric + hyphens, starts with
67 a letter, no consecutive hyphens, no trailing hyphen.
68 """
69 if not rg_id:
70 return False , "Replication group ID must not be empty"
71 if len (rg_id) > 40 :
72 return False , f "Replication group ID must be 1-40 characters (got { len (rg_id) } )"
73 if not re.match( r ' ^[ a-zA-Z ] ' , rg_id):
74 return False , "Replication group ID must start with a letter"
75 if not re.match( r ' ^[ a-zA-Z0-9- ] + $ ' , rg_id):
76 return False , "Replication group ID may only contain alphanumeric characters and hyphens"
77 if '--' in rg_id:
78 return False , "Replication group ID must not contain consecutive hyphens"
79 if rg_id.endswith( '-' ):
80 return False , "Replication group ID must not end with a hyphen"
81 return True , None
82
83
84 def validate_subnet_id (subnet_id: str ) -> tuple[ bool , Optional[ str ]]:
85 """Validate an AWS subnet ID (subnet-<8-17 hex chars>)."""
86 if not subnet_id:
87 return False , "Subnet ID must not be empty"
88 if not re.match( r ' ^ subnet- [ 0-9a-f ] {8,17} $ ' , subnet_id):
89 return False , "Subnet ID must match pattern subnet-<8-17 hex characters> (e.g. subnet-0abc1234def56789a)"
90 return True , None
91
92
93 def validate_security_group_id (sg_id: str ) -> tuple[ bool , Optional[ str ]]:
94 """Validate an AWS security group ID (sg-<8-17 hex chars>)."""
95 if not sg_id:
96 return False , "Security group ID must not be empty"
97 if not re.match( r ' ^ sg- [ 0-9a-f ] {8,17} $ ' , sg_id):
98 return False , "Security group ID must match pattern sg-<8-17 hex characters> (e.g. sg-0abc1234def56789a)"
99 return True , None
100
101
102 def validate_region (region: str ) -> tuple[ bool , Optional[ str ]]:
103 """Validate an AWS region code (e.g. us-east-1, eu-west-2, ap-southeast-1)."""
104 if not region:
105 return False , "Region must not be empty"
106 if not re.match( r ' ^[ a-z ] {2} - [ a-z ] + - \d + $ ' , region):
107 return False , "Region must match AWS region pattern (e.g. us-east-1, eu-west-2)"
108 return True , None
109
110
111 def validate_engine_version (version: str ) -> tuple[ bool , Optional[ str ]]:
112 r """Validate an engine version string.
113
114 Accepted formats:
115 - Major only: "7" (used by serverless MajorEngineVersion)
116 - Major.minor: "7.1", "8.2"
117 - Major.minor.patch: "7.0.4", "7.1.0"
118 """
119 if not version:
120 return False , "Engine version must not be empty"
121 if not re.match( r ' ^\d + ( \. \d + ) {0,2} $ ' , version):
122 return False , "Engine version must match pattern <major>[.<minor>[.<patch>]] (e.g. 7, 7.1, 7.0.4)"
123 return True , None
124
125
126 def validate_vpc_id (vpc_id: str ) -> tuple[ bool , Optional[ str ]]:
127 """Validate an AWS VPC ID (vpc-<8-17 hex chars>)."""
128 if not vpc_id:
129 return False , "VPC ID must not be empty"
130 if not re.match( r ' ^ vpc- [ 0-9a-f ] {8,17} $ ' , vpc_id):
131 return False , "VPC ID must match pattern vpc-<8-17 hex characters> (e.g. vpc-0abc1234def56789a)"
132 return True , None
133
134
135 def validate_kms_key_id (key_id: str ) -> tuple[ bool , Optional[ str ]]:
136 """Validate a KMS key ID.
137
138 Accepts either:
139 - A UUID: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
140 - A KMS key ARN: arn:aws:kms:<region>:<account>:key/<key-id>
141 - A KMS alias ARN: arn:aws:kms:<region>:<account>:alias/<alias-name>
142 """
143 if not key_id:
144 return False , "KMS key ID must not be empty"
145
146 uuid_pattern = r ' ^[ 0-9a-f ] {8} - [ 0-9a-f ] {4} - [ 0-9a-f ] {4} - [ 0-9a-f ] {4} - [ 0-9a-f ] {12} $ '
147 arn_pattern = r ' ^ arn:aws:kms: [ a-z0-9- ] + : \d {12} : ( key | alias ) / [ a-zA-Z0-9/_- ] + $ '
148
149 if re.match(uuid_pattern, key_id):
150 return True , None
151 if re.match(arn_pattern, key_id):
152 return True , None
153
154 return False , (
155 "KMS key ID must be a UUID (e.g. 12345678-1234-1234-1234-123456789abc) "
156 "or a KMS ARN (e.g. arn:aws:kms:us-east-1:123456789012:key/<key-id>)"
157 )
158
159
160 def validate_snapshot_name (name: str , serverless: bool = False ) -> tuple[ bool , Optional[ str ]]:
161 """Validate a snapshot name.
162
163 Rules:
164 - 1 to 255 characters for serverless snapshots, 1 to 40 for node-based
165 - Alphanumeric characters and hyphens only
166 - Must start with a letter
167 - No consecutive hyphens
168 - Must not end with a hyphen
169 """
170 if not name:
171 return False , "Snapshot name must not be empty"
172 max_len = 255 if serverless else 40
173 if len (name) > max_len:
174 label = "serverless" if serverless else "node-based"
175 return False , f "Snapshot name must be 1- { max_len } characters for { label } (got { len (name) } )"
176 if not re.match( r ' ^[ a-zA-Z ] ' , name):
177 return False , "Snapshot name must start with a letter"
178 if not re.match( r ' ^[ a-zA-Z0-9- ] + $ ' , name):
179 return False , "Snapshot name may only contain alphanumeric characters and hyphens"
180 if '--' in name:
181 return False , "Snapshot name must not contain consecutive hyphens"
182 if name.endswith( '-' ):
183 return False , "Snapshot name must not end with a hyphen"
184 return True , None
185
186
187 # ---------------------------------------------------------------------------
188 # CLI sanitization safety net
189 # ---------------------------------------------------------------------------
190
191 # Shell metacharacters that could enable command injection
192 _SHELL_METACHARACTERS = re.compile( r ' [ ;&|`$(){}!<> \'\"\\\n\r\x00 ] ' )
193
194
195 def sanitize_for_cli (value: str ) -> str :
196 """Strip shell metacharacters from a value as a defense-in-depth safety net.
197
198 This is NOT a substitute for proper validation -- always validate inputs
199 first using the validate_* functions. This function provides an additional
200 layer of protection by removing characters that could be interpreted by
201 a shell if a value is accidentally interpolated into a command string.
202
203 Returns the sanitized string with all shell metacharacters removed.
204 """
205 return _SHELL_METACHARACTERS .sub( '' , value)
206
207
208 # ---------------------------------------------------------------------------
209 # Bulk validation
210 # ---------------------------------------------------------------------------
211
212 # Map of recognized parameter keys to their validator functions
213 _VALIDATORS = {
214 "cache_name" : validate_cache_name,
215 "replication_group_id" : validate_replication_group_id,
216 "subnet_id" : validate_subnet_id,
217 "security_group_id" : validate_security_group_id,
218 "region" : validate_region,
219 "engine_version" : validate_engine_version,
220 "vpc_id" : validate_vpc_id,
221 "kms_key_id" : validate_kms_key_id,
222 "snapshot_name" : validate_snapshot_name,
223 }
224
225
226 def validate_all (params: dict ) -> list[ str ]:
227 """Validate all recognized keys in a parameter dict.
228
229 Takes a dict like {"cache_name": "my-cache", "region": "us-east-1"} and
230 validates every key that has a known validator. Keys not recognized are
231 silently skipped (they may be application-specific parameters).
232
233 Returns a list of error messages. An empty list means all recognized
234 parameters passed validation.
235 """
236 errors = []
237 for key, value in params.items():
238 validator = _VALIDATORS .get(key)
239 if validator is None :
240 continue
241 if not isinstance (value, str ):
242 errors.append( f " { key } : expected a string, got { type (value). __name__ } " )
243 continue
244 is_valid, error_msg = validator(value) # type: ignore[operator]
245 if not is_valid:
246 errors.append( f " { key } : { error_msg } " )
247 return errors
248
249
250 # ---------------------------------------------------------------------------
251 # CLI entrypoint
252 # ---------------------------------------------------------------------------
253
254 def main ():
255 """Validate key=value pairs passed on the command line."""
256 if len (sys.argv) < 2 :
257 print ( "Usage: python input_validator.py key=value [key=value ...]" )
258 print ( "Example: python input_validator.py cache_name=my-cache region=us-east-1" )
259 print ()
260 print ( "Recognized keys:" , ", " .join( sorted ( _VALIDATORS .keys())))
261 sys.exit( 0 )
262
263 params = {}
264 for arg in sys.argv[ 1 :]:
265 if '=' not in arg:
266 print ( f "Error: argument ' { arg } ' is not in key=value format" )
267 sys.exit( 2 )
268 key, value = arg.split( '=' , 1 )
269 params[key] = value
270
271 errors = validate_all(params)
272
273 if errors:
274 print ( "Validation FAILED:" )
275 for err in errors:
276 print ( f " - { err } " )
277 sys.exit( 1 )
278 else :
279 print ( "All inputs valid." )
280 sys.exit( 0 )
281
282
283 if __name__ == "__main__" :
284 main()