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 blob = bucket.get_blob(f"""{fileSkel["dlkey"]}/source/{file_name}""") 

109 if not blob: 

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

111 return 

112 fileData = io.BytesIO() 

113 blob.download_to_file(fileData) 

114 resList = [] 

115 for sizeDict in params: 

116 fileData.seek(0) 

117 outData = io.BytesIO() 

118 try: 

119 img = PIL.Image.open(fileData) 

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

121 return [] 

122 iccProfile = img.info.get('icc_profile') 

123 if iccProfile: 

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

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

126 f = io.BytesIO(iccProfile) 

127 src_profile = PIL.ImageCms.ImageCmsProfile(f) 

128 dst_profile = PIL.ImageCms.createProfile('sRGB') 

129 try: 

130 img = PIL.ImageCms.profileToProfile( 

131 img, 

132 inputProfile=src_profile, 

133 outputProfile=dst_profile, 

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

135 except Exception as e: 

136 logging.exception(e) 

137 continue 

138 fileExtension = sizeDict.get("fileExtension", "webp") 

139 if "width" in sizeDict and "height" in sizeDict: 

140 width = sizeDict["width"] 

141 height = sizeDict["height"] 

142 targetName = f"thumbnail-{width}-{height}.{fileExtension}" 

143 elif "width" in sizeDict: 

144 width = sizeDict["width"] 

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

146 targetName = f"thumbnail-w{width}.{fileExtension}" 

147 else: # No default fallback - ignore 

148 continue 

149 mimeType = sizeDict.get("mimeType", "image/webp") 

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

151 img.save(outData, fileExtension) 

152 outSize = outData.tell() 

153 outData.seek(0) 

154 targetBlob = bucket.blob(f"""{fileSkel["dlkey"]}/derived/{targetName}""") 

155 targetBlob.upload_from_file(outData, content_type=mimeType) 

156 resList.append((targetName, outSize, mimeType, {"mimetype": mimeType, "width": width, "height": height})) 

157 return resList 

158 

159 

160def cloudfunction_thumbnailer(fileSkel, existingFiles, params): 

161 """External Thumbnailer for images. 

162 

163 The corresponding cloudfunction can be found here . 

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

165 

166 You can use it like so: 

167 main.py: 

168 

169 .. code-block:: python 

170 

171 from viur.core.modules.file import cloudfunction_thumbnailer 

172 

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

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

175 

176 conf.derives_pdf = { 

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

178 } 

179 

180 skeletons/xxx.py: 

181 .. code-block:: python 

182 

183 test = FileBone(derive=conf.derives_pdf) 

184 """ 

185 

186 if not conf.file_thumbnailer_url: 

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

188 

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

190 

191 def getsignedurl(): 

192 if conf.instance.is_dev_server: 

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

194 else: 

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

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

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

198 return None 

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

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

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

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

203 signedUrl = blob.generate_signed_url( 

204 expiresAt, 

205 credentials=signing_credentials, 

206 response_disposition=content_disposition, 

207 version="v4") 

208 return signedUrl 

209 

210 def make_request(): 

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

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

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

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

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

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

217 match resp.status_code: 

218 case 302: 

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

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

221 logging.error("Cloudfunction not found") 

222 case 404: 

223 logging.error("Cloudfunction not found") 

224 case 403: 

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

226 case _: 

227 logging.error( 

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

229 return 

230 

231 try: 

232 response_data = resp.json() 

233 except Exception as e: 

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

235 return 

236 if "error" in response_data: 

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

238 return 

239 

240 return response_data 

241 

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

243 

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

245 return 

246 dataDict = { 

247 "url": url, 

248 "name": fileSkel["name"], 

249 "params": params, 

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

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

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

253 "nameOnly": True 

254 } 

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

256 return 

257 

258 uploadUrls = {} 

259 for data in derivedData["values"]: 

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

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

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

263 content_type=data["mimeType"]) 

264 

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

266 return 

267 

268 dataDict["url"] = url 

269 dataDict["nameOnly"] = False 

270 dataDict["uploadUrls"] = uploadUrls 

271 

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

273 return 

274 reslist = [] 

275 try: 

276 for derived in derivedData["values"]: 

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

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

279 

280 except Exception as e: 

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

282 return reslist 

283 

284 

285class DownloadUrlBone(BaseBone): 

286 """ 

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

288 """ 

289 

290 def unserialize(self, skel, name): 

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

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

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

294 ) 

295 return True 

296 

297 return False 

298 

299 

300class FileLeafSkel(TreeSkel): 

301 """ 

302 Default file leaf skeleton. 

303 """ 

304 kindName = "file" 

305 

306 name = StringBone( 

307 descr="Filename", 

308 caseSensitive=False, 

309 searchable=True, 

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

311 ) 

312 

313 alt = StringBone( 

314 descr=i18n.translate( 

315 "viur.core.image.alt", 

316 defaultText="Alternative description", 

317 ), 

318 searchable=True, 

319 languages=conf.i18n.available_languages, 

320 ) 

321 

322 size = NumericBone( 

323 descr="Filesize in Bytes", 

324 readOnly=True, 

325 searchable=True, 

326 ) 

327 

328 dlkey = StringBone( 

329 descr="Download-Key", 

330 readOnly=True, 

331 ) 

332 

333 mimetype = StringBone( 

334 descr="MIME-Type", 

335 readOnly=True, 

336 ) 

337 

338 weak = BooleanBone( 

339 descr="Weak reference", 

340 readOnly=True, 

341 visible=False, 

342 ) 

343 

344 pending = BooleanBone( 

345 descr="Pending upload", 

346 readOnly=True, 

347 visible=False, 

348 defaultValue=False, 

349 ) 

350 

351 width = NumericBone( 

352 descr="Width", 

353 readOnly=True, 

354 searchable=True, 

355 ) 

356 

357 height = NumericBone( 

358 descr="Height", 

359 readOnly=True, 

360 searchable=True, 

361 ) 

362 

363 downloadUrl = DownloadUrlBone( 

364 descr="Download-URL", 

365 readOnly=True, 

366 visible=False, 

367 ) 

368 

369 derived = JsonBone( 

370 descr="Derived Files", 

371 readOnly=True, 

372 visible=False, 

373 ) 

374 

375 pendingparententry = KeyBone( 

376 descr="Pending key Reference", 

377 readOnly=True, 

378 visible=False, 

379 ) 

380 

381 crc32c_checksum = StringBone( 

382 descr="CRC32C checksum", 

383 readOnly=True, 

384 ) 

385 

386 md5_checksum = StringBone( 

387 descr="MD5 checksum", 

388 readOnly=True, 

389 ) 

390 

391 public = BooleanBone( 

392 descr="Public File", 

393 readOnly=True, 

394 defaultValue=False, 

395 ) 

396 

397 serving_url = StringBone( 

398 descr="Serving-URL", 

399 readOnly=True, 

400 params={ 

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

402 } 

403 ) 

404 

405 @classmethod 

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

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

408 if ( 

409 skel["public"] 

410 and skel["mimetype"] 

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

412 and not skel["serving_url"] 

413 ): 

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

415 filename = f"/gs/{bucket.name}/{skel['dlkey']}/source/{skel['name']}" 

416 

417 # Trying this on local development server will raise a 

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

419 if conf.instance.is_dev_server: 

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

421 return 

422 

423 try: 

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

425 

426 except Exception as e: 

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

428 logging.exception(e) 

429 

430 def preProcessBlobLocks(self, locks): 

431 """ 

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

433 """ 

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

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

436 return locks 

437 

438 @classmethod 

439 def refresh(cls, skel): 

440 super().refresh(skel) 

441 if conf.viur2import_blobsource: 

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

443 if importData: 

444 if not skel["downloadUrl"]: 

445 skel["downloadUrl"] = importData 

446 skel["pendingparententry"] = None 

447 

448 cls._inject_serving_url(skel) 

449 

450 @classmethod 

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

452 cls._inject_serving_url(skel) 

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

454 

455 

456class FileNodeSkel(TreeSkel): 

457 """ 

458 Default file node skeleton. 

459 """ 

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

461 

462 name = StringBone( 

463 descr="Name", 

464 required=True, 

465 searchable=True 

466 ) 

467 

468 rootNode = BooleanBone( 

469 descr="Is RootNode", 

470 defaultValue=False, 

471 readOnly=True, 

472 visible=False, 

473 ) 

474 

475 public = BooleanBone( 

476 descr="Is public?", 

477 defaultValue=False, 

478 readOnly=True, 

479 visible=False, 

480 ) 

481 

482 viurCurrentSeoKeys = None 

483 

484 

485class File(Tree): 

486 PENDING_POSTFIX = " (pending)" 

487 DOWNLOAD_URL_PREFIX = "/file/download/" 

488 INTERNAL_SERVING_URL_PREFIX = "/file/serve/" 

489 MAX_FILENAME_LEN = 256 

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

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

492 Default: 10 MiB""" 

493 

494 leafSkelCls = FileLeafSkel 

495 nodeSkelCls = FileNodeSkel 

496 

497 handler = "tree.simple.file" 

498 adminInfo = { 

499 "icon": "folder-fill", 

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

501 } 

502 

503 roles = { 

504 "*": "view", 

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

506 "admin": "*", 

507 } 

508 

509 default_order = "name" 

510 

511 # Helper functions currently resist here 

512 

513 @staticmethod 

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

515 """ 

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

517 """ 

518 global _public_bucket 

519 if dlkey and dlkey.endswith(PUBLIC_DLKEY_SUFFIX): 

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

521 return _public_bucket 

522 

523 raise ValueError( 

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

525 ) 

526 

527 return _private_bucket 

528 

529 @classmethod 

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

531 """ 

532 Verifies a valid filename. 

533 

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

535 It should not be longer than MAX_FILENAME_LEN chars. 

536 

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

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

539 """ 

540 if not filename.strip(): 

541 return False 

542 

543 if len(filename) > cls.MAX_FILENAME_LEN: 

544 return False 

545 

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

547 

548 @staticmethod 

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

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

551 if not isinstance(data, bytes): 

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

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

554 

555 @classmethod 

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

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

558 

559 @classmethod 

560 def create_internal_serving_url( 

561 cls, 

562 serving_url: str, 

563 size: int = 0, 

564 filename: str = "", 

565 options: str = "", 

566 download: bool = False 

567 ) -> str: 

568 """ 

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

570 

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

572 legal requirement like GDPR. 

573 

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

575 :param size: Optional size setting 

576 :param filename: Optonal filename setting 

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

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

579 """ 

580 

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

582 res = re.match( 

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

584 serving_url 

585 ) 

586 

587 if not res: 

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

589 

590 # Create internal serving URL 

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

592 

593 # Append additional parameters 

594 if params := { 

595 k: v for k, v in { 

596 "download": download, 

597 "filename": filename, 

598 "options": options, 

599 "size": size, 

600 }.items() if v 

601 }: 

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

603 

604 return serving_url 

605 

606 @classmethod 

607 def create_download_url( 

608 cls, 

609 dlkey: str, 

610 filename: str, 

611 derived: bool = False, 

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

613 download_filename: t.Optional[str] = None 

614 ) -> str: 

615 """ 

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

617 

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

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

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

621 :param expires: 

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

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

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

625 filename 

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

627 """ 

628 if isinstance(expires, int): 

629 expires = datetime.timedelta(minutes=expires) 

630 

631 # Undo escaping on ()= performed on fileNames 

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

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

634 

635 if download_filename: 

636 if not cls.is_valid_filename(download_filename): 

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

638 

639 download_filename = urlquote(download_filename) 

640 

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

642 

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

644 sig = cls.hmac_sign(data) 

645 

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

647 

648 @classmethod 

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

650 """ 

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

652 

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

654 

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

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

657 """ 

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

659 return None 

660 

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

662 sig = sig.removeprefix("sig=") 

663 

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

665 # Invalid signature 

666 return None 

667 

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

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

670 

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

672 case 2: 

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

674 case 1: 

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

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

677 case _: 

678 # Invalid path 

679 return None 

680 

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

682 # Signature expired 

683 return None 

684 

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

686 # Invalid path 

687 return None 

688 

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

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

691 

692 @classmethod 

693 def create_src_set( 

694 cls, 

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

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

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

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

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

700 ) -> str: 

701 """ 

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

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

704 viewport without upscaling. 

705 

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

707 :param expires: 

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

709 otherwise it's lifetime in seconds 

710 :param width: 

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

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

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

714 it will be skipped. 

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

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

717 """ 

718 if not width and not height: 

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

720 return "" 

721 

722 if isinstance(file, str): 

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

724 

725 if not file: 

726 return "" 

727 

728 if isinstance(file, i18n.LanguageWrapper): 

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

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

731 return "" 

732 

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

734 file = file["dest"] 

735 

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

737 

738 if not ( 

739 isinstance(file, (SkeletonInstance, dict)) 

740 and "dlkey" in file 

741 and "derived" in file 

742 ): 

743 logging.error("Invalid file supplied") 

744 return "" 

745 

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

747 logging.error("No derives available") 

748 return "" 

749 

750 src_set = [] 

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

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

753 

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

755 src_set.append( 

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

757 ) 

758 

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

760 src_set.append( 

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

762 ) 

763 

764 return ", ".join(src_set) 

765 

766 def write( 

767 self, 

768 filename: str, 

769 content: t.Any, 

770 mimetype: str = "text/plain", 

771 *, 

772 width: int = None, 

773 height: int = None, 

774 public: bool = False, 

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

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

777 ) -> db.Key: 

778 """ 

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

780 

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

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

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

784 

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

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

787 

788 :param filename: Filename to be written. 

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

790 :param mimetype: The file's mimetype. 

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

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

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

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

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

796 

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

798 """ 

799 # logging.info(f"{filename=} {mimetype=} {width=} {height=} {public=}") 

800 if not self.is_valid_filename(filename): 

801 raise ValueError(f"{filename=} is invalid") 

802 

803 # Folder mode? 

804 if folder: 

805 # Validate correct folder naming 

806 if isinstance(folder, str): 

807 folder = folder, # make it a tuple 

808 

809 for foldername in folder: 

810 if not self.is_valid_filename(foldername): 

811 raise ValueError(f"{foldername=} is invalid") 

812 

813 # When in folder-mode, a rootnode must exist! 

814 if rootnode is None: 

815 rootnode = self.ensureOwnModuleRootNode() 

816