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

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 quote as urlquote, urlencode 

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/{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(): 

567 return False 

568 

569 if len(filename) > cls.MAX_FILENAME_LEN: 

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): 

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 return hmac.compare_digest(cls.hmac_sign(data.encode("ASCII")), signature) 

584 

585 @classmethod 

586 def create_internal_serving_url( 

587 cls, 

588 serving_url: str, 

589 size: int = 0, 

590 filename: str = "", 

591 options: str = "", 

592 download: bool = False 

593 ) -> str: 

594 """ 

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

596 

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

598 legal requirement like GDPR. 

599 

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

601 :param size: Optional size setting 

602 :param filename: Optonal filename setting 

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

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

605 """ 

606 

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

608 res = re.match( 

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

610 serving_url 

611 ) 

612 

613 if not res: 

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

615 

616 # Create internal serving URL 

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

618 

619 # Append additional parameters 

620 if params := { 

621 k: v for k, v in { 

622 "download": download, 

623 "filename": filename, 

624 "options": options, 

625 "size": size, 

626 }.items() if v 

627 }: 

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

629 

630 return serving_url 

631 

632 @classmethod 

633 def create_download_url( 

634 cls, 

635 dlkey: str, 

636 filename: str, 

637 derived: bool = False, 

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

639 download_filename: t.Optional[str] = None 

640 ) -> str: 

641 """ 

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

643 

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

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

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

647 :param expires: 

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

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

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

651 filename 

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

653 """ 

654 if isinstance(expires, int): 

655 expires = datetime.timedelta(minutes=expires) 

656 

657 # Undo escaping on ()= performed on fileNames 

658 filename = filename.replace("&#040;", "(").replace("&#041;", ")").replace("&#061;", "=") 

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

660 

661 if download_filename: 

662 if not cls.is_valid_filename(download_filename): 

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

664 

665 download_filename = urlquote(download_filename) 

666 

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

668 

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

670 sig = cls.hmac_sign(data) 

671 

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

673 

674 @classmethod 

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

676 """ 

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

678 

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

680 

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

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

683 """ 

684 if not url.startswith(cls.DOWNLOAD_URL_PREFIX) or "?" not in url: 

685 return None 

686 

687 data, sig = url.removeprefix(cls.DOWNLOAD_URL_PREFIX).split("?", 1) # Strip "/file/download/" and split on "?" 

688 sig = sig.removeprefix("sig=") 

689 

690 if not cls.hmac_verify(data, sig): 

691 # Invalid signature 

692 return None 

693 

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

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

696 

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

698 case 2: 

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

700 case 1: 

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

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

703 case _: 

704 # Invalid path 

705 return None 

706 

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

708 # Signature expired 

709 return None 

710 

711 if dlpath.count("/") != 2: 

712 # Invalid path 

713 return None 

714 

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

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

717 

718 @classmethod 

719 def create_src_set( 

720 cls, 

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

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

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

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

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

726 ) -> str: 

727 """ 

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

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

730 viewport without upscaling. 

731 

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

733 :param expires: 

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

735 otherwise it's lifetime in seconds 

736 :param width: 

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

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

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

740 it will be skipped. 

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

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

743 """ 

744 if not width and not height: 

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

746 return "" 

747 

748 if isinstance(file, str): 

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

750 

751 if not file: 

752 return "" 

753 

754 if isinstance(file, i18n.LanguageWrapper): 

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

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

757 return "" 

758 

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

760 file = file["dest"] 

761 

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

763 

764 if not ( 

765 isinstance(file, (SkeletonInstance, dict)) 

766 and "dlkey" in file 

767 and "derived" in file 

768 ): 

769 logging.error("Invalid file supplied") 

770 return "" 

771 

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

773 logging.error("No derives available") 

774 return "" 

775 

776 src_set = [] 

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

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

779 

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

781 src_set.append( 

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

783 ) 

784 

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

786 src_set.append( 

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

788 ) 

789 

790 return ", ".join(src_set) 

791 

792 def write( 

793 self, 

794 filename: str, 

795 content: t.Any, 

796 mimetype: str = "text/plain", 

797 *, 

798 width: int = None, 

799 height: int = None, 

800 public: bool = False, 

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

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

803 ) -> db.Key: 

804 """ 

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

806 

807 If *folder* and *rootnode* are both set, the file is added to the repository in that folder. 

808 If only *folder* is set, the file is added to the default repository in that folder. 

809 If only *rootnode* is set, the file is added to that repository in the root folder. 

810 

811 If both are not set, the file is added without a path or repository as a weak file. 

812 It will not be visible in admin in this case. 

813 

814 :param filename: Filename to be written. 

815 :param content: The file content to be written, as bytes-like object. 

816 :param mimetype: The file's mimetype. 

817 :param width: Optional width information for the file. 

818 :param height: Optional height information for the file. 

819 :param public: True if the file should be publicly accessible. 

820 :param rootnode: Optional root-node of the repository to add the file to 

821 :param folder: Optional folder the file should be written into. 

822 

823 :return: Returns the key of the file object written. This can be associated e.g. with a FileBone. 

824 """