Setting the file. One moment.
Archive · Launch With AWS · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page 69
Creating Amazon Aurora Db Cluster With Instances
92
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Debugging Lambda Timeouts
277
def _is_symlink_entry
— line 277
This file
Number 45.2
Position 2 of 7
Type Python
Size 16 KB
Lines 495 scripts/ archive.py
Python · 495 lines · 16 KB
urllib.request
15 import zipfile
16 from typing import Optional, Set, Tuple
17 from urllib.parse import urlparse
18
19 from launch_config import (
20 GITHUB_ZIPBALL_TIMEOUT_SECS ,
21 MAX_ARCHIVE_BYTES ,
22 MAX_ARCHIVE_ENTRIES ,
23 MAX_COMPRESSION_RATIO ,
24 MAX_ENTRY_UNCOMPRESSED_BYTES ,
25 MAX_UNCOMPRESSED_BYTES ,
26 )
27
28 logger = logging.getLogger( __name__ )
29
30 # Chunk size for reading the GitHub download.
31 _DOWNLOAD_CHUNK_BYTES = 64 * 1024
32
33 # Hosts a GitHub zipball request may be redirected through: api.github.com
34 # redirects to codeload and then to its S3-backed object store.
35 _ALLOWED_REDIRECT_HOST_SUFFIXES = (
36 ".github.com" ,
37 ".githubusercontent.com" ,
38 ".amazonaws.com" ,
39 )
40 _ALLOWED_REDIRECT_HOSTS = frozenset ({ "github.com" })
41
42 _SKIP_DIRS = frozenset (
43 {
44 ".git" ,
45 ".hg" ,
46 ".svn" ,
47 "node_modules" ,
48 ".next" ,
49 ".nuxt" ,
50 ".turbo" ,
51 ".cache" ,
52 ".parcel-cache" ,
53 "dist" ,
54 "build" ,
55 "out" ,
56 "coverage" ,
57 ".venv" ,
58 "venv" ,
59 "__pycache__" ,
60 ".pytest_cache" ,
61 ".mypy_cache" ,
62 ".ruff_cache" ,
63 ".idea" ,
64 ".vscode" ,
65 # Credential/secret directories
66 ".aws" ,
67 ".ssh" ,
68 ".gnupg" ,
69 ".gcp" ,
70 ".azure" ,
71 ".docker" ,
72 ".kube" ,
73 }
74 )
75
76 _SKIP_FILES = frozenset (
77 {
78 ".env" ,
79 ".env.local" ,
80 ".env.production" ,
81 ".env.development" ,
82 ".env.test" ,
83 ".env.staging" ,
84 "credentials" ,
85 "credentials.json" ,
86 ".git-credentials" ,
87 ".netrc" ,
88 ".pypirc" ,
89 ".npmrc" ,
90 ".htpasswd" ,
91 }
92 )
93
94 # Best-effort filtering of common credential files, not a security boundary.
95 # A filtered archive is not guaranteed to be secret-free.
96 _SKIP_SUFFIXES = ( ".pem" , ".key" , ".p12" , ".pfx" , ".jks" , ".keystore" )
97
98 _SKIP_NAME_PREFIXES = ( "id_rsa" , "id_ed25519" , "id_ecdsa" , "id_dsa" )
99
100 _GITHUB_URL_RE = re.compile(
101 r " ^ https://github \. com/ ([\w .- ] + ) / ([\w .- ] +? )(?: \. git ) ? / ? $ " ,
102 )
103
104
105 class ArchiveError ( Exception ):
106 """Raised when a repo archive cannot be produced."""
107
108
109 def _decode_git_paths (output: bytes ) -> Set[ str ]:
110 """Decode NUL-delimited Git paths using the filesystem encoding."""
111 return {os.fsdecode(path) for path in output.split( b " \0 " ) if path}
112
113
114 def _run_git (root: str , args: list[ str ]) -> subprocess.CompletedProcess:
115 """Run a Git file-enumeration command with consistent safeguards."""
116 return subprocess.run(
117 [ "git" , * args],
118 cwd = root,
119 capture_output = True ,
120 timeout = 30 ,
121 )
122
123
124 def _has_gitignore (root: str ) -> bool :
125 """Return whether the source tree contains an applicable .gitignore file."""
126 for _, dirnames, filenames in os.walk(root):
127 dirnames[:] = [d for d in dirnames if d not in _SKIP_DIRS ]
128 if ".gitignore" in filenames:
129 return True
130 return False
131
132
133 def _gitignore_files_without_repo (root: str ) -> Set[ str ]:
134 """Apply Git ignore rules to a directory that is not a Git worktree."""
135 with tempfile.TemporaryDirectory( prefix = "launch-with-aws-git-" ) as git_dir:
136 initialized = _run_git(root, [ "init" , "--bare" , "--quiet" , git_dir])
137 if initialized.returncode != 0 :
138 raise ArchiveError( "Git could not initialize temporary metadata to apply ignore rules." )
139
140 result = _run_git(
141 root,
142 [
143 f "--git-dir= { git_dir } " ,
144 f "--work-tree= { root } " ,
145 "ls-files" ,
146 "--others" ,
147 "--exclude-standard" ,
148 "-z" ,
149 ],
150 )
151 if result.returncode != 0 :
152 raise ArchiveError(
153 "Git could not determine which files are safe to upload. "
154 "Resolve the Git error and try again."
155 )
156 return _decode_git_paths(result.stdout)
157
158
159 def _git_included_files (root: str ) -> Optional[Set[ str ]]:
160 """Return paths selected by Git ignore rules, or None when no rules apply.
161
162 NUL-delimited output preserves unusual filenames. Tracked files are combined
163 with untracked-but-not-ignored files so new local files are included.
164 """
165 try :
166 result = _run_git(
167 root,
168 [
169 "ls-files" ,
170 "--cached" ,
171 "--others" ,
172 "--exclude-standard" ,
173 "-z" ,
174 ],
175 )
176 if result.returncode == 0 :
177 ignored_tracked = _run_git(
178 root,
179 [
180 "ls-files" ,
181 "--cached" ,
182 "--ignored" ,
183 "--exclude-standard" ,
184 "-z" ,
185 ],
186 )
187 if ignored_tracked.returncode != 0 :
188 raise ArchiveError(
189 "Git could not determine which tracked files match ignore rules."
190 )
191 return _decode_git_paths(result.stdout) - _decode_git_paths(ignored_tracked.stdout)
192 if not _has_git_metadata(root):
193 if _has_gitignore(root):
194 return _gitignore_files_without_repo(root)
195 return None
196 raise ArchiveError(
197 "Git could not determine which files are safe to upload. "
198 "Resolve the Git error and try again."
199 )
200 except subprocess.TimeoutExpired as err:
201 raise ArchiveError(
202 "Git timed out while determining which files are safe to upload."
203 ) from err
204 except OSError as err:
205 if _has_git_metadata(root) or _has_gitignore(root):
206 raise ArchiveError(
207 "Git is required to apply this source directory's ignore rules before upload."
208 ) from err
209 return None
210
211
212 def _has_git_metadata (root: str ) -> bool :
213 """Return whether root is at or below a directory containing Git metadata."""
214 current = os.path.abspath(root)
215 while True :
216 if os.path.exists(os.path.join(current, ".git" )):
217 return True
218 parent = os.path.dirname(current)
219 if parent == current:
220 return False
221 current = parent
222
223
224 def parse_github_url (url: str ) -> Optional[Tuple[ str , str ]]:
225 """Return (owner, repo) for a GitHub HTTPS URL, or None.
226
227 Only https://github.com/<owner>/<repo> URLs are accepted, and `.` / `..`
228 segments are rejected.
229 """
230 match = _GITHUB_URL_RE .match(url.strip())
231 if not match:
232 return None
233 owner, repo = match.group( 1 ), match.group( 2 )
234 if any (segment in ( "." , ".." ) for segment in (owner, repo)):
235 return None
236 return owner, repo
237
238
239 def sanitize_root_name (name: str ) -> str :
240 segments = [s for s in re.split( r " [ \\ / ] " , name.strip()) if s]
241 base = (segments[ - 1 ] if segments else "" ).strip().lstrip( "." )
242 cleaned = re.sub( r " [ ^ \w .- ] " , "-" , base).strip( "-" )
243 return cleaned or "app"
244
245
246 def _is_secret_file (filename: str ) -> bool :
247 """Return True if the filename matches a known secret pattern."""
248 lower = filename.lower()
249 if lower in _SKIP_FILES :
250 return True
251 if lower.startswith( ".env." ):
252 return True
253 if lower.endswith( _SKIP_SUFFIXES ):
254 return True
255 if any (lower.startswith(p) for p in _SKIP_NAME_PREFIXES ):
256 return True
257 return False
258
259
260 def _is_safe_entry_name (name: str ) -> bool :
261 """Return True if a ZIP entry name is a plain relative path.
262
263 Rejects absolute paths, Windows drive/UNC prefixes, `..` segments, and
264 embedded NUL bytes.
265 """
266 if not name or " \x00 " in name:
267 return False
268 if name.startswith( "/" ) or name.startswith( " \\ " ):
269 return False
270 # Windows drive letter (e.g. "C:\") or UNC ("\\host").
271 if re.match( r " ^[ A-Za-z ] :" , name) or name.startswith( " \\\\ " ):
272 return False
273 normalized = name.replace( " \\ " , "/" )
274 return not any (segment == ".." for segment in normalized.split( "/" ))
275
276
277 def _is_symlink_entry (info: zipfile.ZipInfo) -> bool :
278 """Return True if a ZIP entry encodes a Unix symlink."""
279 # The high 16 bits of external_attr hold the Unix mode for Unix-created
280 # entries.
281 mode = (info.external_attr >> 16 ) & 0x FFFF
282 return stat.S_ISLNK(mode)
283
284
285 def validate_archive_bytes (data: bytes ) -> None :
286 """Validate ZIP entry names, shape, and size limits.
287
288 All checks read the central-directory metadata only (nothing is
289 decompressed). Raises ArchiveError on the first violation.
290 """
291 try :
292 zf = zipfile.ZipFile(io.BytesIO(data))
293 except zipfile.BadZipFile as err:
294 raise ArchiveError( "Downloaded file is not a valid zip archive." ) from err
295
296 with zf:
297 infos = zf.infolist()
298
299 if len (infos) > MAX_ARCHIVE_ENTRIES :
300 raise ArchiveError(
301 f "Archive has { len (infos) } entries, exceeding the "
302 f " { MAX_ARCHIVE_ENTRIES } entry limit."
303 )
304
305 total_uncompressed = 0
306 roots: Set[ str ] = set ()
307 for info in infos:
308 name = info.filename
309
310 if not _is_safe_entry_name(name):
311 raise ArchiveError( f "Archive contains an unsafe entry path: { name !r} " )
312 if _is_symlink_entry(info):
313 raise ArchiveError( f "Archive contains a symlink entry: { name !r} " )
314
315 size = info.file_size
316 if size > MAX_ENTRY_UNCOMPRESSED_BYTES :
317 raise ArchiveError(
318 f "Archive entry { name !r} is { size // ( 1024 * 1024 ) } MiB "
319 f "uncompressed, exceeding the "
320 f " { MAX_ENTRY_UNCOMPRESSED_BYTES // ( 1024 * 1024 ) } MiB per-entry limit."
321 )
322 if info.compress_size > 0 and size / info.compress_size > MAX_COMPRESSION_RATIO :
323 raise ArchiveError(
324 f "Archive entry { name !r} exceeds the { MAX_COMPRESSION_RATIO } :1 "
325 "compression-ratio limit."
326 )
327
328 total_uncompressed += size
329 if total_uncompressed > MAX_UNCOMPRESSED_BYTES :
330 raise ArchiveError(
331 f "Archive decompresses to over "
332 f " { MAX_UNCOMPRESSED_BYTES // ( 1024 * 1024 * 1024 ) } GiB, "
333 "exceeding the decompressed-size limit."
334 )
335
336 top = info.filename.replace( " \\ " , "/" ).split( "/" , 1 )[ 0 ]
337 if top:
338 roots.add(top)
339
340 if len (roots) > 1 :
341 raise ArchiveError(
342 f "Archive must contain a single top-level directory; found { len (roots) } ."
343 )
344
345
346 def zip_local_repo (path: str , root_name: str ) -> bytes :
347 """Zip a local repo directory, skipping VCS/build artifacts and secrets.
348
349 When the directory is a git repository, .gitignore rules are respected
350 via `git ls-files`. The hardcoded secret exclusions still apply on top
351 as a defense-in-depth measure.
352 """
353 root = os.path.abspath(os.path.expanduser(path))
354 if not os.path.isdir(root):
355 raise ArchiveError( f "Not a directory: { path } " )
356
357 prefix = sanitize_root_name(root_name)
358 skipped: list[ str ] = []
359 git_files = _git_included_files(root)
360
361 if git_files is not None :
362 logger.info( "Using git to determine included files (.gitignore respected)" )
363
364 buffer = io.BytesIO()
365 file_count = 0
366 with zipfile.ZipFile(buffer, "w" , zipfile. ZIP_DEFLATED ) as zf:
367 for dirpath, dirnames, filenames in os.walk(root):
368 dirnames[:] = [d for d in dirnames if d not in _SKIP_DIRS ]
369 for filename in filenames:
370 abs_path = os.path.join(dirpath, filename)
371 if os.path.islink(abs_path):
372 continue
373 rel = os.path.relpath(abs_path, root).replace(os.sep, "/" )
374 if git_files is not None and rel not in git_files:
375 continue
376 if _is_secret_file(filename):
377 skipped.append(rel)
378 continue
379 zf.write(abs_path, f " { prefix } / { rel } " )
380 file_count += 1
381
382 if skipped:
383 logger.info(
384 "Excluded %d file(s) matching secret patterns: %s " ,
385 len (skipped),
386 ", " .join(skipped),
387 )
388
389 if file_count == 0 :
390 raise ArchiveError( f "No files found under { path } " )
391
392 data = buffer.getvalue()
393 if len (data) > MAX_ARCHIVE_BYTES :
394 raise ArchiveError(
395 f "Archive is { len (data) // ( 1024 * 1024 ) } MiB, exceeding the "
396 f " { MAX_ARCHIVE_BYTES // ( 1024 * 1024 ) } MiB limit. Remove large "
397 "files or build artifacts and try again."
398 )
399 validate_archive_bytes(data)
400 return data
401
402
403 def _is_allowed_redirect_host (hostname: Optional[ str ]) -> bool :
404 """Return True if a redirect target host is on the allowlist."""
405 if not hostname:
406 return False
407 hostname = hostname.lower()
408 if hostname in _ALLOWED_REDIRECT_HOSTS :
409 return True
410 return any (hostname.endswith(suffix) for suffix in _ALLOWED_REDIRECT_HOST_SUFFIXES )
411
412
413 class _AllowlistRedirectHandler ( urllib . request . HTTPRedirectHandler ):
414 """Redirect handler that only follows redirects to allowlisted hosts.
415
416 GitHub zipball downloads 302 from api.github.com to codeload and then to an
417 S3 object store; redirects to any other host are refused.
418 """
419
420 def redirect_request (self, req, fp, code, msg, headers, newurl):
421 parsed = urlparse(newurl)
422 if parsed.scheme != "https" :
423 raise ArchiveError( f "Refusing redirect to non-HTTPS URL during download: { newurl !r} " )
424 if not _is_allowed_redirect_host(parsed.hostname):
425 raise ArchiveError(
426 f "Refusing redirect to disallowed host during download: { parsed.hostname !r} "
427 )
428 return super ().redirect_request(req, fp, code, msg, headers, newurl)
429
430
431 def _read_capped (resp) -> bytes :
432 """Read a response body, aborting if it exceeds MAX_ARCHIVE_BYTES.
433
434 Pre-checks the declared Content-Length when present, then reads in chunks
435 against a running total so bodies without a Content-Length are also bounded.
436 """
437 declared = resp.headers.get( "Content-Length" )
438 if declared is not None :
439 try :
440 if int (declared) > MAX_ARCHIVE_BYTES :
441 raise ArchiveError(
442 f "Downloaded archive is { int (declared) // ( 1024 * 1024 ) } MiB, "
443 f "exceeding the { MAX_ARCHIVE_BYTES // ( 1024 * 1024 ) } MiB limit."
444 )
445 except ValueError :
446 pass
447
448 chunks: list[ bytes ] = []
449 total = 0
450 while True :
451 chunk = resp.read( _DOWNLOAD_CHUNK_BYTES )
452 if not chunk:
453 break
454 total += len (chunk)
455 if total > MAX_ARCHIVE_BYTES :
456 raise ArchiveError(
457 f "Downloaded archive exceeds the "
458 f " { MAX_ARCHIVE_BYTES // ( 1024 * 1024 ) } MiB limit."
459 )
460 chunks.append(chunk)
461 return b "" .join(chunks)
462
463
464 def download_github_zip (url: str ) -> bytes :
465 """Download and validate a public GitHub repo's default-branch archive.
466
467 Follows only allowlisted redirects, caps the body at MAX_ARCHIVE_BYTES, and
468 validates the archive before returning the bytes.
469 """
470 parsed = parse_github_url(url)
471 if not parsed:
472 raise ArchiveError( f "Not a GitHub repository URL: { url } " )
473 owner, repo = parsed
474
475 zipball = f "https://api.github.com/repos/ { owner } / { repo } /zipball"
476 req = urllib.request.Request(zipball, headers = { "Accept" : "application/vnd.github+json" })
477 opener = urllib.request.build_opener(_AllowlistRedirectHandler())
478 try :
479 with opener.open(req, timeout = GITHUB_ZIPBALL_TIMEOUT_SECS ) as resp:
480 data = _read_capped(resp)
481 except ArchiveError:
482 raise
483 except urllib.error.HTTPError as err:
484 if err.code in ( 401 , 403 , 404 ):
485 raise ArchiveError(
486 f "Could not download { owner } / { repo } . Only public repositories can "
487 "be fetched by URL — for a private repo, clone it locally and pass "
488 "the local path instead."
489 ) from err
490 raise ArchiveError( f "GitHub returned { err.code } downloading { owner } / { repo } ." ) from err
491 except (urllib.error.URLError, OSError ) as err:
492 raise ArchiveError( f "Failed to download { url } : { err } " ) from err
493
494 validate_archive_bytes(data)
495 return data