Save uploaded file to disk quickly (plain) and return size. Speed notes: - Prefer Werkzeug's built-in `save()` which streams efficiently to disk. - Fallback to a large-buffer copy for environments where `save()` misbehaves.
(file_storage, tmp_path: str)
| 502 | pass |
| 503 | return key |
| 504 | |
| 505 | MASTER_KEY = load_or_create_master_key() |
| 506 | |
| 507 | def aesgcm_encrypt_stream(src_fp, dst_path: str): |
| 508 | """File format: MAGIC || NONCE || TAG || CIPHERTEXT""" |
| 509 | nonce = os.urandom(NONCE_LEN) |
| 510 | cipher = Cipher(algorithms.AES(MASTER_KEY), modes.GCM(nonce)) |
| 511 | encryptor = cipher.encryptor() |
| 512 | |
| 513 | os.makedirs(os.path.dirname(dst_path), exist_ok=True) |
| 514 | with open(dst_path, "wb") as out: |
| 515 | out.write(MAGIC) |
| 516 | out.write(nonce) |
| 517 | out.write(b"\x00" * TAG_LEN) # tag placeholder |
| 518 | |
| 519 | while True: |
| 520 | chunk = src_fp.read(CHUNK) |
| 521 | if not chunk: |
| 522 | break |
| 523 | out.write(encryptor.update(chunk)) |
| 524 | encryptor.finalize() |
| 525 | tag = encryptor.tag |
| 526 | |
| 527 | out.seek(len(MAGIC) + NONCE_LEN) |
| 528 | out.write(tag) |
| 529 | |
| 530 | def _filestorage_size(file_storage) -> Optional[int]: |
| 531 | """Return the size (bytes) of an uploaded FileStorage if possible. |
| 532 | |
| 533 | We prefer stream seeking because Content-Length is not always available for |
| 534 | multipart parts on all clients. |
| 535 | """ |
| 536 | if file_storage is None: |
| 537 | return None |
no test coverage detected