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

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

658 expires = datetime.timedelta(minutes=expires) 

659 

660 # Undo escaping on ()= performed on fileNames 

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

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

663 

664 if download_filename: 

665 if not cls.is_valid_filename(download_filename): 

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

667 

668 download_filename = urlquote(download_filename) 

669 

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

671 

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

673 sig = cls.hmac_sign(data) 

674 

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

676 

677 @classmethod 

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

679 """ 

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

681 

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

683 

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

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

686 """ 

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

688 return None 

689 

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

691 sig = sig.removeprefix("sig=") 

692 

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

694 # Invalid signature 

695 return None 

696 

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

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

699 

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

701 case 2: 

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

703 case 1: 

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

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

706 case _: 

707 # Invalid path 

708 return None 

709 

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

711 # Signature expired 

712 return None 

713 

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

715 # Invalid path 

716 return None 

717 

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

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

720 

721 @classmethod 

722 def create_src_set( 

723 cls, 

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

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

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

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

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

729 ) -> str: 

730 """ 

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

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

733 viewport without upscaling. 

734 

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

736 :param expires: 

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

738 otherwise it's lifetime in seconds 

739 :param width: 

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

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

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

743 it will be skipped. 

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

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

746 """ 

747 if not width and not height: 

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

749 return "" 

750 

751 if isinstance(file, str): 

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

753 

754 if not file: 

755 return "" 

756 

757 if isinstance(file, i18n.LanguageWrapper): 

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

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

760 return "" 

761 

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

763 file = file["dest"] 

764 

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

766 

767 if not ( 

768 isinstance(file, (SkeletonInstance, dict)) 

769 and "dlkey" in file 

770 and "derived" in file 

771 ): 

772 logging.error("Invalid file supplied") 

773 return "" 

774 

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

776 logging.error("No derives available") 

777 return "" 

778 

779 src_set = [] 

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

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

782 

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

784 src_set.append( 

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

786 ) 

787 

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

789 src_set.append( 

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

791 ) 

792 

793 return ", ".join(src_set) 

794 

795 def write( 

796 self, 

797 filename: str, 

798 content: t.Any, 

799 mimetype: str = "text/plain", 

800 *, 

801 width: int = None, 

802 height: int = None, 

803 public: bool = False, 

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

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

806 ) -> db.Key: 

807 """ 

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

809 

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

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

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

813 

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

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

816 

817 :param filename: Filename to be written. 

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

819 :param mimetype: The file's mimetype. 

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

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