Securely download a URL to a stream of chunks. If the integrity of the download fails, an IntegrityError is raised.
(url, size, sha256)
| 263 | |
| 264 | |
| 265 | def secure_download_stream(url, size, sha256): |
| 266 | """Securely download a URL to a stream of chunks. |
| 267 | |
| 268 | If the integrity of the download fails, an IntegrityError is |
| 269 | raised. |
| 270 | """ |
| 271 | h = hashlib.sha256() |
| 272 | length = 0 |
| 273 | |
| 274 | with urllib.request.urlopen(url) as fh: |
| 275 | if not url.endswith(".gz") and fh.info().get("Content-Encoding") == "gzip": |
| 276 | fh = gzip.GzipFile(fileobj=fh) |
| 277 | |
| 278 | while True: |
| 279 | chunk = fh.read(65536) |
| 280 | if not chunk: |
| 281 | break |
| 282 | |
| 283 | h.update(chunk) |
| 284 | length += len(chunk) |
| 285 | |
| 286 | yield chunk |
| 287 | |
| 288 | digest = h.hexdigest() |
| 289 | |
| 290 | if length != size or digest != sha256: |
| 291 | raise IntegrityError( |
| 292 | "integrity mismatch on %s: wanted size=%d, sha256=%s; got size=%d, sha256=%s" |
| 293 | % (url, size, sha256, length, digest), |
| 294 | length=length, |
| 295 | ) |
| 296 | |
| 297 | |
| 298 | def download_to_path(url: str, path: pathlib.Path, size: int, sha256: str): |
no test coverage detected