File size: 4,881 Bytes
6d63e5b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | import os
import tarfile
from hashlib import sha256
from pathlib import Path
from typing import List, Optional, Tuple
from urllib.parse import urlparse
from .common import PathOrStr, get_cache_dir
from .meta import Meta
def resource_to_filename(resource: PathOrStr, etag: Optional[str] = None) -> str:
"""
Convert a ``resource`` into a hashed filename in a repeatable way.
If ``etag`` is specified, append its hash to the resources', delimited
by a period.
THis is essentially the inverse of :func:`filename_to_url()`.
"""
resource_bytes = str(resource).encode("utf-8")
resource_hash = sha256(resource_bytes)
filename = resource_hash.hexdigest()
if etag:
etag_bytes = etag.encode("utf-8")
etag_hash = sha256(etag_bytes)
filename += "." + etag_hash.hexdigest()
return filename
def filename_to_url(
filename: str, cache_dir: Optional[PathOrStr] = None
) -> Tuple[str, Optional[str]]:
"""
Return the URL and etag (which may be ``None``) stored for ``filename``.
Raises :exc:`FileNotFoundError` if ``filename`` or its stored metadata do not exist.
This is essentially the inverse of :func:`resource_to_filename()`.
"""
cache_dir = cache_dir if cache_dir else get_cache_dir()
cache_path = os.path.join(cache_dir, filename)
if not os.path.exists(cache_path):
raise FileNotFoundError("file {} not found".format(cache_path))
meta_path = cache_path + ".json"
if not os.path.exists(meta_path):
raise FileNotFoundError("file {} not found".format(meta_path))
metadata = Meta.from_path(meta_path)
return metadata.resource, metadata.etag
def find_latest_cached(
url: str, cache_dir: Optional[PathOrStr] = None, verbose: bool = False
) -> Optional[Path]:
"""
Get the path to the latest cached version of a given resource.
"""
cache_dir = Path(cache_dir if cache_dir else get_cache_dir())
filename = resource_to_filename(url)
candidates: List[Tuple[Path, float]] = []
for path in cache_dir.glob(f"{filename}*"):
if verbose:
print(path, path.suffix, path.name)
if path.suffix in {".json", ".lock"} or path.name.endswith("-extracted"):
continue
mtime = path.stat().st_mtime
candidates.append((path, mtime))
# Sort candidates by modification time, newest first.
candidates.sort(key=lambda x: x[1], reverse=True)
if candidates:
return candidates[0][0]
return None
def check_tarfile(tar_file: tarfile.TarFile):
"""Tar files can contain files outside of the extraction directory, or symlinks that point
outside the extraction directory. We also don't want any block devices fifos, or other
weird file types extracted. This checks for those issues and throws an exception if there
is a problem."""
base_path = os.path.join("tmp", "pathtest")
base_path = os.path.normpath(base_path)
def normalize_path(path: str) -> str:
path = path.rstrip("/")
path = path.replace("/", os.sep)
path = os.path.join(base_path, path)
path = os.path.normpath(path)
return path
for tarinfo in tar_file:
if not (
tarinfo.isreg()
or tarinfo.isdir()
or tarinfo.isfile()
or tarinfo.islnk()
or tarinfo.issym()
):
raise ValueError(
f"Tar file {str(tar_file.name)} contains invalid member {tarinfo.name}."
)
target_path = normalize_path(tarinfo.name)
if os.path.commonprefix([base_path, target_path]) != base_path:
raise ValueError(
f"Tar file {str(tar_file.name)} is trying to create a file outside of its extraction directory."
)
if tarinfo.islnk() or tarinfo.issym():
target_path = normalize_path(tarinfo.linkname)
if os.path.commonprefix([base_path, target_path]) != base_path:
raise ValueError(
f"Tar file {str(tar_file.name)} is trying to link to a file "
"outside of its extraction directory."
)
def is_url_or_existing_file(url_or_filename: PathOrStr) -> bool:
"""
Given something that might be a URL or local path,
determine if it's actually a url or the path to an existing file.
"""
if url_or_filename is None:
return False
from .schemes import get_supported_schemes
url_or_filename = os.path.expanduser(str(url_or_filename))
parsed = urlparse(url_or_filename)
return parsed.scheme in get_supported_schemes() or os.path.exists(url_or_filename)
def _lock_file_path(cache_path: Path) -> Path:
return cache_path.parent / (cache_path.name + ".lock")
def _meta_file_path(cache_path: Path) -> Path:
return cache_path.parent / (cache_path.name + ".json")
|