Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/modules/file.py: 17%

1import base64 

2import datetime 

3import hashlib 

4import hmac 

5import html 

6import io 

7import json 

8import logging 

9import re 

10import string 

11import typing as t 

12import warnings 

13from collections import namedtuple 

14from urllib.parse import parse_qs, quote as urlquote, unquote as urlunquote, urlencode, urlsplit 

15from urllib.request import urlopen 

16 

17import PIL 

18import PIL.ImageCms 

19import google.auth 

20import requests 

21from PIL import Image 

22from google.appengine.api import blobstore, images 

23from google.cloud import storage 

24from google.oauth2.service_account import Credentials as ServiceAccountCredentials 

25 

26from viur.core import conf, current, db, errors, utils, i18n 

27from viur.core.bones import BaseBone, BooleanBone, JsonBone, KeyBone, NumericBone, StringBone 

28 

29from viur.core.decorators import * 

30from viur.core.prototypes.tree import SkelType, Tree, TreeSkel 

31from viur.core.skeleton import SkeletonInstance, skeletonByKind 

32from viur.core.tasks import CallDeferred, DeleteEntitiesIter, PeriodicTask 

33 

34# Globals for connectivity 

35 

36VALID_FILENAME_REGEX = re.compile( 

37 # || MAY NOT BE THE NAME | MADE OF SPECIAL CHARS | SPECIAL CHARS + `. `|` 

38 r"^(?!^(CON|PRN|AUX|NUL|COM[1-9]|LPT[1-9])$)[^\x00-\x1F<>:\"\/\\|?*]*[^\x00-\x1F<>:\"\/\\|?*. ]$", 

39 re.IGNORECASE 

40) 

41 

42_CREDENTIALS, _PROJECT_ID = google.auth.default() 

43GOOGLE_STORAGE_CLIENT = storage.Client(_PROJECT_ID, _CREDENTIALS) 

44 

45PRIVATE_BUCKET_NAME = f"""{_PROJECT_ID}.appspot.com""" 

46PUBLIC_BUCKET_NAME = f"""public-dot-{_PROJECT_ID}""" 

47PUBLIC_DLKEY_SUFFIX = "_pub" 

48 

49_private_bucket = GOOGLE_STORAGE_CLIENT.lookup_bucket(PRIVATE_BUCKET_NAME) 

50_public_bucket = None 

51 

52# FilePath is a descriptor for ViUR file components 

53FilePath = namedtuple("FilePath", ("dlkey", "is_derived", "filename")) 

54 

55 

56def importBlobFromViur2(dlKey, fileName): 

57 bucket = conf.main_app.file.get_bucket(dlKey) 

58 

59 if not conf.viur2import_blobsource: 

60 return False 

61 existingImport = db.get(db.Key("viur-viur2-blobimport", dlKey)) 

62 if existingImport: 

63 if existingImport["success"]: 

64 return existingImport["dlurl"] 

65 return False 

66 if conf.viur2import_blobsource["infoURL"]: 

67 try: 

68 importDataReq = urlopen(conf.viur2import_blobsource["infoURL"] + dlKey) 

69 except Exception as e: 

70 marker = db.Entity(db.Key("viur-viur2-blobimport", dlKey)) 

71 marker["success"] = False 

72 marker["error"] = "Failed URL-FETCH 1" 

73 db.put(marker) 

74 return False 

75 if importDataReq.status != 200: 

76 marker = db.Entity(db.Key("viur-viur2-blobimport", dlKey)) 

77 marker["success"] = False 

78 marker["error"] = "Failed URL-FETCH 2" 

79 db.put(marker) 

80 return False 

81 importData = json.loads(importDataReq.read()) 

82 oldBlobName = conf.viur2import_blobsource["gsdir"] + "/" + importData["key"] 

83 srcBlob = storage.Blob(bucket=bucket, 

84 name=conf.viur2import_blobsource["gsdir"] + "/" + importData["key"]) 

85 else: 

86 oldBlobName = conf.viur2import_blobsource["gsdir"] + "/" + dlKey 

87 srcBlob = storage.Blob(bucket=bucket, name=conf.viur2import_blobsource["gsdir"] + "/" + dlKey) 

88 if not srcBlob.exists(): 

89 marker = db.Entity(db.Key("viur-viur2-blobimport", dlKey)) 

90 marker["success"] = False 

91 marker["error"] = "Local SRC-Blob missing" 

92 marker["oldBlobName"] = oldBlobName 

93 db.put(marker) 

94 return False 

95 bucket.rename_blob(srcBlob, f"{dlKey}/source/{fileName}") 

96 marker = db.Entity(db.Key("viur-viur2-blobimport", dlKey)) 

97 marker["success"] = True 

98 marker["old_src_key"] = dlKey 

99 marker["old_src_name"] = fileName 

100 marker["dlurl"] = conf.main_app.file.create_download_url(dlKey, fileName, False, None) 

101 db.put(marker) 

102 return marker["dlurl"] 

103 

104 

105def thumbnailer(fileSkel, existingFiles, params): 

106 file_name = html.unescape(fileSkel["name"]) 

107 bucket = conf.main_app.file.get_bucket(fileSkel["dlkey"]) 

108 

109 blob = bucket.get_blob(f"""{fileSkel["dlkey"]}/source/{file_name}""") 

110 if not blob: 

111 logging.warning(f"""Blob {fileSkel["dlkey"]}/source/{file_name} is missing from cloud storage!""") 

112 return 

113 

114 source = io.BytesIO() 

115 blob.download_to_file(source) 

116 

117 result = [] 

118 

119 for info in params: 

120 # Read the image into PIL 

121 try: 

122 source.seek(0) 

123 img = PIL.Image.open(source) 

124 except PIL.Image.UnidentifiedImageError: # Can't load this image; so there's no need to try other resolutions 

125 break 

126 

127 if icc_profile := img.info.get("icc_profile"): 

128 # JPEGs might be encoded with a non-standard color-profile; we need to compensate for this if we convert 

129 # to WEBp as we'll loose this color-profile information 

130 f = io.BytesIO(icc_profile) 

131 src_profile = PIL.ImageCms.ImageCmsProfile(f) 

132 dst_profile = PIL.ImageCms.createProfile("sRGB") 

133 try: 

134 img = PIL.ImageCms.profileToProfile( 

135 img, 

136 inputProfile=src_profile, 

137 outputProfile=dst_profile, 

138 outputMode="RGBA" if img.has_transparency_data else "RGB") 

139 except Exception as e: 

140 logging.debug(f"{info=}") 

141 logging.exception(e) 

142 continue 

143 

144 file_extension = info.get("fileExtension", "webp") 

145 mimetype = info.get("mimeType", "image/webp") 

146 

147 if "width" in info and "height" in info: 

148 width = info["width"] 

149 height = info["height"] 

150 target_filename = f"thumbnail-{width}-{height}.{file_extension}" 

151 

152 elif "width" in info: 

153 width = info["width"] 

154 height = int((float(img.size[1]) * float(width / float(img.size[0])))) 

155 target_filename = f"thumbnail-w{width}.{file_extension}" 

156 

157 else: # No default fallback - ignore 

158 continue 

159 

160 # Create resized version of the source 

161 target = io.BytesIO() 

162 

163 try: 

164 img = img.resize((width, height), PIL.Image.LANCZOS) 

165 except ValueError as e: 

166 # Usually happens to some files, like TIFF-images. 

167 logging.debug(f"{info=}") 

168 logging.exception(e) 

169 break 

170 

171 img.save(target, file_extension) 

172 

173 # Safe derived target file 

174 target_size = target.tell() 

175 target.seek(0) 

176 target_blob = bucket.blob(f"""{fileSkel["dlkey"]}/derived/{target_filename}""") 

177 target_blob.upload_from_file(target, content_type=mimetype) 

178 

179 result.append( 

180 (target_filename, target_size, mimetype, {"mimetype": mimetype, "width": width, "height": height}) 

181 ) 

182 

183 return result 

184 

185 

186def cloudfunction_thumbnailer(fileSkel, existingFiles, params): 

187 """External Thumbnailer for images. 

188 

189 The corresponding cloudfunction can be found here . 

190 https://github.com/viur-framework/viur-cloudfunctions/tree/main/thumbnailer 

191 

192 You can use it like so: 

193 main.py: 

194 

195 .. code-block:: python 

196 

197 from viur.core.modules.file import cloudfunction_thumbnailer 

198 

199 conf.file_thumbnailer_url = "https://xxxxx.cloudfunctions.net/imagerenderer" 

200 conf.file_derivations = {"thumbnail": cloudfunction_thumbnailer} 

201 

202 conf.derives_pdf = { 

203 "thumbnail": [{"width": 1920,"sites":"1,2"}] 

204 } 

205 

206 skeletons/xxx.py: 

207 .. code-block:: python 

208 

209 test = FileBone(derive=conf.derives_pdf) 

210 """ 

211 

212 if not conf.file_thumbnailer_url: 

213 raise ValueError("conf.file_thumbnailer_url is not set") 

214 

215 bucket = conf.main_app.file.get_bucket(fileSkel["dlkey"]) 

216 

217 def getsignedurl(): 

218 if conf.instance.is_dev_server: 

219 signedUrl = conf.main_app.file.create_download_url(fileSkel["dlkey"], fileSkel["name"]) 

220 else: 

221 path = f"""{fileSkel["dlkey"]}/source/{file_name}""" 

222 if not (blob := bucket.get_blob(path)): 

223 logging.warning(f"Blob {path} is missing from cloud storage!") 

224 return None 

225 authRequest = google.auth.transport.requests.Request() 

226 expiresAt = datetime.datetime.now() + datetime.timedelta(seconds=60) 

227 signing_credentials = google.auth.compute_engine.IDTokenCredentials(authRequest, "") 

228 content_disposition = utils.build_content_disposition_header(fileSkel["name"]) 

229 signedUrl = blob.generate_signed_url( 

230 expiresAt, 

231 credentials=signing_credentials, 

232 response_disposition=content_disposition, 

233 version="v4") 

234 return signedUrl 

235 

236 def make_request(): 

237 headers = {"Content-Type": "application/json"} 

238 data_str = base64.b64encode(json.dumps(dataDict).encode("UTF-8")) 

239 sig = conf.main_app.file.hmac_sign(data_str) 

240 datadump = json.dumps({"dataStr": data_str.decode('ASCII'), "sign": sig}) 

241 resp = requests.post(conf.file_thumbnailer_url, data=datadump, headers=headers, allow_redirects=False) 

242 if resp.status_code != 200: # Error Handling 

243 match resp.status_code: 

244 case 302: 

245 # The problem is Google resposen 302 to an auth Site when the cloudfunction was not found 

246 # https://cloud.google.com/functions/docs/troubleshooting#login 

247 logging.error("Cloudfunction not found") 

248 case 404: 

249 logging.error("Cloudfunction not found") 

250 case 403: 

251 logging.error("No permission for the Cloudfunction") 

252 case _: 

253 logging.error( 

254 f"cloudfunction_thumbnailer failed with code: {resp.status_code} and data: {resp.content}") 

255 return 

256 

257 try: 

258 response_data = resp.json() 

259 except Exception as e: 

260 logging.error(f"response could not be converted in json failed with: {e=}") 

261 return 

262 if "error" in response_data: 

263 logging.error(f"cloudfunction_thumbnailer failed with: {response_data.get('error')}") 

264 return 

265 

266 return response_data 

267 

268 file_name = html.unescape(fileSkel["name"]) 

269 

270 if not (url := getsignedurl()): 

271 return 

272 dataDict = { 

273 "url": url, 

274 "name": fileSkel["name"], 

275 "params": params, 

276 "minetype": fileSkel["mimetype"], 

277 "baseUrl": current.request.get().request.host_url.lower(), 

278 "targetKey": fileSkel["dlkey"], 

279 "nameOnly": True 

280 } 

281 if not (derivedData := make_request()): 

282 return 

283 

284 uploadUrls = {} 

285 for data in derivedData["values"]: 

286 fileName = conf.main_app.file.sanitize_filename(data["name"]) 

287 blob = bucket.blob(f"""{fileSkel["dlkey"]}/derived/{fileName}""") 

288 uploadUrls[fileSkel["dlkey"] + fileName] = blob.create_resumable_upload_session(timeout=60, 

289 content_type=data["mimeType"]) 

290 

291 if not (url := getsignedurl()): 

292 return 

293 

294 dataDict["url"] = url 

295 dataDict["nameOnly"] = False 

296 dataDict["uploadUrls"] = uploadUrls 

297 

298 if not (derivedData := make_request()): 

299 return 

300 reslist = [] 

301 try: 

302 for derived in derivedData["values"]: 

303 for key, value in derived.items(): 

304 reslist.append((key, value["size"], value["mimetype"], value["customData"])) 

305 

306 except Exception as e: 

307 logging.error(f"cloudfunction_thumbnailer failed with: {e=}") 

308 return reslist 

309 

310 

311class DownloadUrlBone(BaseBone): 

312 """ 

313 This bone is used to inject a freshly signed download url into a FileSkel. 

314 """ 

315 

316 def unserialize(self, skel, name): 

317 if "dlkey" in skel.dbEntity and "name" in skel.dbEntity: 

318 skel.accessedValues[name] = conf.main_app.file.create_download_url( 

319 skel["dlkey"], skel["name"], expires=conf.render_json_download_url_expiration 

320 ) 

321 return True 

322 

323 return False 

324 

325 

326class FileLeafSkel(TreeSkel): 

327 """ 

328 Default file leaf skeleton. 

329 """ 

330 kindName = "file" 

331 

332 name = StringBone( 

333 descr="Filename", 

334 caseSensitive=False, 

335 searchable=True, 

336 vfunc=lambda val: None if File.is_valid_filename(val) else "Invalid filename provided", 

337 ) 

338 

339 alt = StringBone( 

340 descr=i18n.translate( 

341 "viur.core.image.alt", 

342 defaultText="Alternative description", 

343 ), 

344 searchable=True, 

345 languages=conf.i18n.available_languages, 

346 ) 

347 

348 size = NumericBone( 

349 descr="Filesize in Bytes", 

350 readOnly=True, 

351 searchable=True, 

352 ) 

353 

354 dlkey = StringBone( 

355 descr="Download-Key", 

356 readOnly=True, 

357 ) 

358 

359 mimetype = StringBone( 

360 descr="MIME-Type", 

361 readOnly=True, 

362 ) 

363 

364 weak = BooleanBone( 

365 descr="Weak reference", 

366 readOnly=True, 

367 visible=False, 

368 ) 

369 

370 pending = BooleanBone( 

371 descr="Pending upload", 

372 readOnly=True, 

373 visible=False, 

374 defaultValue=False, 

375 ) 

376 

377 width = NumericBone( 

378 descr="Width", 

379 readOnly=True, 

380 searchable=True, 

381 ) 

382 

383 height = NumericBone( 

384 descr="Height", 

385 readOnly=True, 

386 searchable=True, 

387 ) 

388 

389 downloadUrl = DownloadUrlBone( 

390 descr="Download-URL", 

391 readOnly=True, 

392 visible=False, 

393 ) 

394 

395 derived = JsonBone( 

396 descr="Derived Files", 

397 readOnly=True, 

398 visible=False, 

399 ) 

400 

401 pendingparententry = KeyBone( 

402 descr="Pending key Reference", 

403 readOnly=True, 

404 visible=False, 

405 ) 

406 

407 crc32c_checksum = StringBone( 

408 descr="CRC32C checksum", 

409 readOnly=True, 

410 ) 

411 

412 md5_checksum = StringBone( 

413 descr="MD5 checksum", 

414 readOnly=True, 

415 ) 

416 

417 public = BooleanBone( 

418 descr="Public File", 

419 readOnly=True, 

420 defaultValue=False, 

421 ) 

422 

423 serving_url = StringBone( 

424 descr="Serving-URL", 

425 readOnly=True, 

426 params={ 

427 "tooltip": "The 'serving_url' is only available in public file repositories.", 

428 } 

429 ) 

430 

431 @classmethod 

432 def _inject_serving_url(cls, skel: SkeletonInstance) -> None: 

433 """Inject the serving url for public image files into a FileSkel""" 

434 if ( 

435 skel["public"] 

436 and skel["mimetype"] 

437 and skel["mimetype"].startswith("image/") 

438 and not skel["serving_url"] 

439 ): 

440 bucket = File.get_bucket(skel["dlkey"]) 

441 filename = f"/gs/{bucket.name}/{skel['dlkey']}/source/{utils.string.unescape(skel['name'])}" 

442 

443 # Trying this on local development server will raise a 

444 # `google.appengine.runtime.apiproxy_errors.RPCFailedError` 

445 if conf.instance.is_dev_server: 

446 logging.warning(f"Can't inject serving_url for {filename!r} on local development server") 

447 return 

448 

449 try: 

450 skel["serving_url"] = images.get_serving_url(None, secure_url=True, filename=filename) 

451 

452 except Exception as e: 

453 logging.warning(f"Failed to create serving_url for {filename!r} with exception {e!r}") 

454 logging.exception(e) 

455 

456 def preProcessBlobLocks(self, locks): 

457 """ 

458 Ensure that our dlkey is locked even if we don't have a filebone here 

459 """ 

460 if not self["weak"] and self["dlkey"]: 

461 locks.add(self["dlkey"]) 

462 return locks 

463 

464 @classmethod 

465 def refresh(cls, skel): 

466 super().refresh(skel) 

467 if conf.viur2import_blobsource: 

468 importData = importBlobFromViur2(skel["dlkey"], skel["name"]) 

469 if importData: 

470 if not skel["downloadUrl"]: 

471 skel["downloadUrl"] = importData 

472 skel["pendingparententry"] = None 

473 

474 cls._inject_serving_url(skel) 

475 

476 @classmethod 

477 def write(cls, skel, **kwargs): 

478 cls._inject_serving_url(skel) 

479 return super().write(skel, **kwargs) 

480 

481 

482class FileNodeSkel(TreeSkel): 

483 """ 

484 Default file node skeleton. 

485 """ 

486 kindName = "file_rootNode" # FIXME: VIUR4, don't use "_rootNode" kindname 

487 

488 name = StringBone( 

489 descr="Name", 

490 required=True, 

491 searchable=True 

492 ) 

493 

494 rootNode = BooleanBone( 

495 descr="Is RootNode", 

496 defaultValue=False, 

497 readOnly=True, 

498 visible=False, 

499 ) 

500 

501 public = BooleanBone( 

502 descr="Is public?", 

503 defaultValue=False, 

504 readOnly=True, 

505 visible=False, 

506 ) 

507 

508 viurCurrentSeoKeys = None 

509 

510 

511class File(Tree): 

512 PENDING_POSTFIX = " (pending)" 

513 DOWNLOAD_URL_PREFIX = "/file/download/" 

514 INTERNAL_SERVING_URL_PREFIX = "/file/serve/" 

515 MAX_FILENAME_LEN = 256 

516 IMAGE_META_MAX_SIZE: t.Final[int] = 10 * 1024 ** 2 

517 """Maximum size of image files that should be analysed in :meth:`set_image_meta`. 

518 Default: 10 MiB""" 

519 

520 leafSkelCls = FileLeafSkel 

521 nodeSkelCls = FileNodeSkel 

522 

523 handler = "tree.simple.file" 

524 adminInfo = { 

525 "icon": "folder-fill", 

526 "handler": handler, # fixme: Use static handler; Remove with VIUR4! 

527 } 

528 

529 roles = { 

530 "*": "view", 

531 "editor": ("add", "edit"), 

532 "admin": "*", 

533 } 

534 

535 default_order = "name" 

536 

537 # Helper functions currently resist here 

538 

539 @staticmethod 

540 def get_bucket(dlkey: str) -> google.cloud.storage.bucket.Bucket: 

541 """ 

542 Retrieves a Google Cloud Storage bucket for the given dlkey. 

543 """ 

544 global _public_bucket 

545 if dlkey and dlkey.endswith(PUBLIC_DLKEY_SUFFIX): 

546 if _public_bucket or (_public_bucket := GOOGLE_STORAGE_CLIENT.lookup_bucket(PUBLIC_BUCKET_NAME)): 

547 return _public_bucket 

548 

549 raise ValueError( 

550 f"""The bucket '{PUBLIC_BUCKET_NAME}' does not exist! Please create it with ACL access.""" 

551 ) 

552 

553 return _private_bucket 

554 

555 @classmethod 

556 def is_valid_filename(cls, filename: str) -> bool: 

557 """ 

558 Verifies a valid filename. 

559 

560 The filename should be valid on Linux, Mac OS and Windows. 

561 It should not be longer than MAX_FILENAME_LEN chars. 

562 

563 Rule set: https://stackoverflow.com/a/31976060/3749896 

564 Regex test: https://regex101.com/r/iBYpoC/1 

565 """ 

566 if not filename.strip(): 566 ↛ 567line 566 didn't jump to line 567 because the condition on line 566 was never true

567 return False 

568 

569 if len(filename) > cls.MAX_FILENAME_LEN: 569 ↛ 570line 569 didn't jump to line 570 because the condition on line 569 was never true

570 return False 

571 

572 return bool(re.match(VALID_FILENAME_REGEX, filename)) 

573 

574 @staticmethod 

575 def hmac_sign(data: t.Any) -> str: 

576 assert conf.file_hmac_key is not None, "No hmac-key set!" 

577 if not isinstance(data, bytes): 577 ↛ 578line 577 didn't jump to line 578 because the condition on line 577 was never true

578 data = str(data).encode("UTF-8") 

579 return hmac.new(conf.file_hmac_key, msg=data, digestmod=hashlib.sha3_384).hexdigest() 

580 

581 @classmethod 

582 def hmac_verify(cls, data: t.Any, signature: str) -> bool: 

583 try: 

584 return hmac.compare_digest(cls.hmac_sign(data.encode("ASCII")), signature) 

585 except (TypeError, UnicodeEncodeError): 

586 return False 

587 

588 @classmethod 

589 def create_internal_serving_url( 

590 cls, 

591 serving_url: str, 

592 size: int = 0, 

593 filename: str = "", 

594 options: str = "", 

595 download: bool = False 

596 ) -> str: 

597 """ 

598 Helper function to generate an internal serving url (endpoint: /file/serve) from a Google serving url. 

599 

600 This is needed to hide requests to Google as they are internally be routed, and can be the result of a 

601 legal requirement like GDPR. 

602 

603 :param serving_url: Is the original serving URL as generated from FileLeafSkel._inject_serving_url() 

604 :param size: Optional size setting 

605 :param filename: Optonal filename setting 

606 :param options: Additional options parameter-pass through to /file/serve 

607 :param download: Download parameter-pass through to /file/serve 

608 """ 

609 

610 # Split a serving URL into its components, used by serve function. 

611 res = re.match( 

612 r"^https:\/\/(.*?)\.googleusercontent\.com\/(.*?)$", 

613 serving_url 

614 ) 

615 

616 if not res: 

617 raise ValueError(f"Invalid {serving_url=!r} provided") 

618 

619 # Create internal serving URL 

620 serving_url = cls.INTERNAL_SERVING_URL_PREFIX + "/".join(res.groups()) 

621 

622 # Append additional parameters 

623 if params := { 

624 k: v for k, v in { 

625 "download": download, 

626 "filename": filename, 

627 "options": options, 

628 "size": size, 

629 }.items() if v 

630 }: 

631 serving_url += f"?{urlencode(params)}" 

632 

633 return serving_url 

634 

635 @classmethod 

636 def create_download_url( 

637 cls, 

638 dlkey: str, 

639 filename: str, 

640 derived: bool = False, 

641 expires: t.Optional[datetime.timedelta | int] = datetime.timedelta(hours=1), 

642 download_filename: t.Optional[str] = None 

643 ) -> str: 

644 """ 

645 Utility function that creates a signed download-url for the given folder/filename combination 

646 

647 :param folder: The GCS-Folder (= the download-key) for that file 

648 :param filename: The name of the file. Either the original filename or the name of a derived file. 

649 :param derived: True, if it points to a derived file, False if it points to the original uploaded file 

650 :param expires: 

651 None if the file is supposed to be public (which causes it to be cached on the google ede caches), 

652 otherwise a datetime.timedelta of how long that link should be valid 

653 :param download_filename: If set, browser is enforced to download this blob with the given alternate 

654 filename 

655 :return: The signed download-url relative to the current domain (eg /download/...) 

656 """ 

657 if isinstance(expires, int): 657 ↛ 658line 657 didn't jump to line 658 because the condition on line 657 was never true

658 expires = datetime.timedelta(minutes=expires) 

659 

660 filename = html.unescape(filename) 

661 filepath = f"""{dlkey}/{"derived" if derived else "source"}/{filename}""" 

662 

663 if download_filename: 

664 if not cls.is_valid_filename(download_filename): 664 ↛ 665line 664 didn't jump to line 665 because the condition on line 664 was never true

665 raise errors.UnprocessableEntity(f"Invalid download_filename {download_filename!r} provided") 

666 

667 download_filename = urlquote(download_filename) 

668 

669 expires = (datetime.datetime.now() + expires).strftime("%Y%m%d%H%M") if expires else 0 

670 

671 data = base64.urlsafe_b64encode(f"""{filepath}\0{expires}\0{download_filename or ""}""".encode("UTF-8")) 

672 sig = cls.hmac_sign(data) 

673 

674 return f"""{cls.DOWNLOAD_URL_PREFIX}{data.decode("ASCII")}?sig={sig}""" 

675 

676 @classmethod 

677 def parse_download_url(cls, url) -> t.Optional[FilePath]: 

678 """ 

679 Parses a file download URL in the format `/file/download/xxxx?sig=yyyy` into its FilePath. 

680 

681 The URL may be absolute, because admin frontends store it including scheme and host, and 

682 it may carry an additional path segment after the payload: :meth:`download` accepts the 

683 download file name that way, and a bare trailing slash occurs as well. 

684 

685 If the URL cannot be parsed, the function returns None. 

686 

687 :param url: The file download URL to be parsed. 

688 :return: A FilePath on success, None otherwise. 

689 """ 

690 parsed_url = urlsplit(url) 

691 

692 if not parsed_url.path.startswith(cls.DOWNLOAD_URL_PREFIX) or not parsed_url.query: 

693 return None 

694 

695 # Strip "/file/download/"; the payload is urlsafe-base64 and therefore never contains a 

696 # slash, so anything after the first one is the optional download file name. 

697 data = parsed_url.path.removeprefix(cls.DOWNLOAD_URL_PREFIX).split("/", 1)[0] 

698 sig = parse_qs(parsed_url.query).get("sig", [""])[0] 

699 

700 if not cls.hmac_verify(data, sig): 700 ↛ 702line 700 didn't jump to line 702 because the condition on line 700 was never true

701 # Invalid signature 

702 return None 

703 

704 # Split the blobKey into the individual fields it should contain 

705 data = base64.urlsafe_b64decode(data).decode("UTF-8") 

706 

707 match data.count("\0"): 

708 case 2: 708 ↛ 710line 708 didn't jump to line 710 because the pattern on line 708 always matched

709 dlpath, valid_until, _ = data.split("\0") 

710 case 1: 

711 # It's the old format, without an downloadFileName 

712 dlpath, valid_until = data.split("\0") 

713 case _: 

714 # Invalid path 

715 return None 

716 

717 if valid_until != "0" and datetime.datetime.strptime(valid_until, "%Y%m%d%H%M") < datetime.datetime.now(): 

718 # Signature expired 

719 return None 

720 

721 if dlpath.count("/") != 2: 721 ↛ 723line 721 didn't jump to line 723 because the condition on line 721 was never true

722 # Invalid path 

723 return None 

724 

725 dlkey, derived, filename = dlpath.split("/") 

726 return FilePath(dlkey, derived != "source", filename) 

727 

728 @classmethod 

729 def create_src_set( 

730 cls, 

731 file: t.Union["SkeletonInstance", dict, str], 

732 expires: t.Optional[datetime.timedelta | int] = datetime.timedelta(hours=1), 

733 width: t.Optional[int] = None, 

734 height: t.Optional[int] = None, 

735 language: t.Optional[str] = None, 

736 ) -> str: 

737 """ 

738 Generates a string suitable for use as the srcset tag in html. This functionality provides the browser 

739 with a list of images in different sizes and allows it to choose the smallest file that will fill it's 

740 viewport without upscaling. 

741 

742 :param file: The file skeleton (or if multiple=True a single value from it) to generate the srcset. 

743 :param expires: 

744 None if the file is supposed to be public (which causes it to be cached on the google edecaches), 

745 otherwise it's lifetime in seconds 

746 :param width: 

747 A list of widths that should be included in the srcset. 

748 If a given width is not available, it will be skipped. 

749 :param height: A list of heights that should be included in the srcset. If a given height is not available, 

750 it will be skipped. 

751 :param language: Language overwrite if file has multiple languages, and we want to explicitly specify one 

752 :return: The srctag generated or an empty string if a invalid file object was supplied 

753 """ 

754 if not width and not height: 

755 logging.error("Neither width or height supplied") 

756 return "" 

757 

758 if isinstance(file, str): 

759 file = db.Query("file").filter("dlkey =", file).order(("creationdate", db.SortOrder.Ascending)).getEntry() 

760 

761 if not file: 

762 return "" 

763 

764 if isinstance(file, i18n.LanguageWrapper): 

765 language = language or current.language.get() 

766 if not language or not (file := cls.get(language)): 

767 return "" 

768 

769 if "dlkey" not in file and "dest" in file: 

770 file = file["dest"] 

771 

772 from viur.core.skeleton import SkeletonInstance # avoid circular imports 

773 

774 if not ( 

775 isinstance(file, (SkeletonInstance, dict)) 

776 and "dlkey" in file 

777 and "derived" in file 

778 ): 

779 logging.error("Invalid file supplied") 

780 return "" 

781 

782 if not isinstance(file["derived"], dict): 

783 logging.error("No derives available") 

784 return "" 

785 

786 src_set = [] 

787 for filename, derivate in file["derived"]["files"].items(): 

788 customData = derivate.get("customData", {}) 

789 

790 if width and customData.get("width") in width: 

791 src_set.append( 

792 f"""{cls.create_download_url(file["dlkey"], filename, True, expires)} {customData["width"]}w""" 

793 ) 

794 

795 if height and customData.get("height") in height: 

796 src_set.append( 

797 f"""{cls.create_download_url(file["dlkey"], filename, True, expires)} {customData["height"]}h""" 

798 ) 

799 

800 return ", ".join(src_set) 

801 

802 def write( 

803 self, 

804 filename: str, 

805 content: t.Any, 

806 mimetype: str = "text/plain", 

807 *, 

808 width: int = None, 

809 height: int = None, 

810 public: bool = False, 

811 rootnode: t.Optional[db.Key] = None, 

812 folder: t.Iterable[str] | str = (), 

813 ) -> db.Key: 

814 """ 

815 Write a file from any bytes-like object into the file module. 

816