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

1import base64 

2import datetime 

3import google.auth 

4import hashlib 

5import hmac 

6import html 

7import io 

8import json 

9import logging 

10import PIL 

11import PIL.Image 

12import PIL.ImageCms 

13import re 

14import requests 

15import string 

16import typing as t 

17import warnings 

18 

19from collections import namedtuple 

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

21from urllib.request import urlopen 

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 ( 

28 BaseBone, 

29 BooleanBone, 

30 Compute, 

31 ComputeInterval, 

32 ComputeMethod, 

33 JsonBone, 

34 KeyBone, 

35 NumericBone, 

36 StringBone, 

37) 

38from viur.core.decorators import exposed, force_post, force_ssl, skey 

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

40from viur.core.skeleton import SkeletonInstance, skeletonByKind 

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

42 

43# Globals for connectivity 

44 

45VALID_FILENAME_REGEX = re.compile( 

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

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

48 re.IGNORECASE 

49) 

50 

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

52GOOGLE_STORAGE_CLIENT = storage.Client(_PROJECT_ID, _CREDENTIALS) 

53 

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

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

56PUBLIC_DLKEY_SUFFIX = "_pub" 

57 

58_private_bucket = GOOGLE_STORAGE_CLIENT.lookup_bucket(PRIVATE_BUCKET_NAME) 

59_public_bucket = None 

60 

61# FilePath is a descriptor for ViUR file components 

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

63 

64 

65def importBlobFromViur2(dlKey, fileName): 

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

67 

68 if not conf.viur2import_blobsource: 

69 return False 

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

71 if existingImport: 

72 if existingImport["success"]: 

73 return existingImport["dlurl"] 

74 return False 

75 if conf.viur2import_blobsource["infoURL"]: 

76 try: 

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

78 except Exception as e: 

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

80 marker["success"] = False 

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

82 db.put(marker) 

83 return False 

84 if importDataReq.status != 200: 

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

86 marker["success"] = False 

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

88 db.put(marker) 

89 return False 

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

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

92 srcBlob = storage.Blob(bucket=bucket, 

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

94 else: 

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

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

97 if not srcBlob.exists(): 

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

99 marker["success"] = False 

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

101 marker["oldBlobName"] = oldBlobName 

102 db.put(marker) 

103 return False 

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

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

106 marker["success"] = True 

107 marker["old_src_key"] = dlKey 

108 marker["old_src_name"] = fileName 

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

110 db.put(marker) 

111 return marker["dlurl"] 

112 

113 

114def thumbnailer(fileSkel, existingFiles, params): 

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

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

117 

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

119 if not blob: 

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

121 return 

122 

123 source = io.BytesIO() 

124 blob.download_to_file(source) 

125 

126 result = [] 

127 

128 for info in params: 

129 # Read the image into PIL 

130 try: 

131 source.seek(0) 

132 img = PIL.Image.open(source) 

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

134 break 

135 

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

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

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

139 f = io.BytesIO(icc_profile) 

140 src_profile = PIL.ImageCms.ImageCmsProfile(f) 

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

142 try: 

143 img = PIL.ImageCms.profileToProfile( 

144 img, 

145 inputProfile=src_profile, 

146 outputProfile=dst_profile, 

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

148 except Exception as e: 

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

150 logging.exception(e) 

151 continue 

152 

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

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

155 

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

157 width = info["width"] 

158 height = info["height"] 

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

160 

161 elif "width" in info: 

162 width = info["width"] 

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

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

165 

166 else: # No default fallback - ignore 

167 continue 

168 

169 # Create resized version of the source 

170 target = io.BytesIO() 

171 

172 try: 

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

174 except ValueError as e: 

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

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

177 logging.exception(e) 

178 break 

179 

180 img.save(target, file_extension) 

181 

182 # Safe derived target file 

183 target_size = target.tell() 

184 target.seek(0) 

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

186 target_blob.upload_from_file(target, content_type=mimetype) 

187 

188 result.append( 

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

190 ) 

191 

192 return result 

193 

194 

195def cloudfunction_thumbnailer(fileSkel, existingFiles, params): 

196 """External Thumbnailer for images. 

197 

198 The corresponding cloudfunction can be found here . 

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

200 

201 You can use it like so: 

202 main.py: 

203 

204 .. code-block:: python 

205 

206 from viur.core.modules.file import cloudfunction_thumbnailer 

207 

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

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

210 

211 conf.derives_pdf = { 

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

213 } 

214 

215 skeletons/xxx.py: 

216 .. code-block:: python 

217 

218 test = FileBone(derive=conf.derives_pdf) 

219 """ 

220 

221 if not conf.file_thumbnailer_url: 

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

223 

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

225 

226 def getsignedurl(): 

227 if conf.instance.is_dev_server: 

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

229 else: 

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

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

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

233 return None 

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

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

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

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

238 signedUrl = blob.generate_signed_url( 

239 expiresAt, 

240 credentials=signing_credentials, 

241 response_disposition=content_disposition, 

242 version="v4") 

243 return signedUrl 

244 

245 def make_request(): 

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

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

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

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

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

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

252 match resp.status_code: 

253 case 302: 

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

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

256 logging.error("Cloudfunction not found") 

257 case 404: 

258 logging.error("Cloudfunction not found") 

259 case 403: 

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

261 case _: 

262 logging.error( 

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

264 return 

265 

266 try: 

267 response_data = resp.json() 

268 except Exception as e: 

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

270 return 

271 if "error" in response_data: 

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

273 return 

274 

275 return response_data 

276 

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

278 

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

280 return 

281 dataDict = { 

282 "url": url, 

283 "name": fileSkel["name"], 

284 "params": params, 

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

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

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

288 "nameOnly": True 

289 } 

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

291 return 

292 

293 uploadUrls = {} 

294 for data in derivedData["values"]: 

295 if not conf.main_app.file.is_valid_filename(data["name"]): 

296 raise errors.UnprocessableEntity(f"""Invalid derived filename {data["name"]!r} provided""") 

297 

298 fileName = urlquote(data["name"]) 

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

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

301 content_type=data["mimeType"]) 

302 

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

304 return 

305 

306 dataDict["url"] = url 

307 dataDict["nameOnly"] = False 

308 dataDict["uploadUrls"] = uploadUrls 

309 

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

311 return 

312 reslist = [] 

313 try: 

314 for derived in derivedData["values"]: 

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

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

317 

318 except Exception as e: 

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

320 return reslist 

321 

322 

323class DownloadUrlBone(BaseBone): 

324 """ 

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

326 """ 

327 

328 def unserialize(self, skel, name): 

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

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

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

332 ) 

333 return True 

334 

335 return False 

336 

337 

338class FileLeafSkel(TreeSkel): 

339 """ 

340 Default file leaf skeleton. 

341 """ 

342 kindName = "file" 

343 

344 name = StringBone( 

345 descr="Filename", 

346 caseSensitive=False, 

347 searchable=True, 

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

349 ) 

350 

351 alt = StringBone( 

352 descr=i18n.translate( 

353 "viur.core.image.alt", 

354 defaultText="Alternative description", 

355 ), 

356 searchable=True, 

357 languages=conf.i18n.available_languages, 

358 ) 

359 

360 size = NumericBone( 

361 descr="Filesize in Bytes", 

362 readOnly=True, 

363 searchable=True, 

364 ) 

365 

366 dlkey = StringBone( 

367 descr="Download-Key", 

368 readOnly=True, 

369 ) 

370 

371 mimetype = StringBone( 

372 descr="MIME-Type", 

373 readOnly=True, 

374 ) 

375 

376 weak = BooleanBone( 

377 descr="Weak reference", 

378 readOnly=True, 

379 visible=False, 

380 ) 

381 

382 pending = BooleanBone( 

383 descr="Pending upload", 

384 readOnly=True, 

385 visible=False, 

386 defaultValue=False, 

387 ) 

388 

389 width = NumericBone( 

390 descr="Width", 

391 readOnly=True, 

392 searchable=True, 

393 ) 

394 

395 height = NumericBone( 

396 descr="Height", 

397 readOnly=True, 

398 searchable=True, 

399 ) 

400 

401 downloadUrl = DownloadUrlBone( 

402 descr="Download-URL", 

403 readOnly=True, 

404 visible=False, 

405 ) 

406 

407 derived = JsonBone( 

408 descr="Derived Files", 

409 readOnly=True, 

410 visible=False, 

411 ) 

412 

413 pendingparententry = KeyBone( 

414 descr="Pending key Reference", 

415 readOnly=True, 

416 visible=False, 

417 ) 

418 

419 crc32c_checksum = StringBone( 

420 descr="CRC32C checksum", 

421 readOnly=True, 

422 ) 

423 

424 md5_checksum = StringBone( 

425 descr="MD5 checksum", 

426 readOnly=True, 

427 ) 

428 

429 public = BooleanBone( 

430 descr="Public File", 

431 readOnly=True, 

432 defaultValue=False, 

433 ) 

434 

435 def _compute_serving_url(skel: SkeletonInstance): 

436 """ 

437 Standard compute function for generating a Google Image Manipulation API serving URL for public files. 

438 """ 

439 logging.info("_compute_serving_url") 

440 if ( 

441 skel["public"] 

442 and skel["mimetype"] 

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

444 ): 

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

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

447 

448 # Trying this on local development server will raise a 

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

450 if conf.instance.is_dev_server: 

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

452 return 

453 

454 try: 

455 # Generate Google Image Manipulation API Serving-URL 

456 return images.get_serving_url(None, secure_url=True, filename=filename) 

457 

458 except Exception as e: 

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

460 logging.exception(e) 

461 

462 return None 

463 

464 serving_url = StringBone( 

465 descr="Serving-URL", 

466 params={ 

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

468 }, 

469 compute=Compute(_compute_serving_url, ComputeInterval(ComputeMethod.Once)), 

470 ) 

471 

472 def preProcessBlobLocks(self, locks): 

473 """ 

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

475 """ 

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

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

478 

479 return locks 

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_download_url( 

590 cls, 

591 dlkey: str, 

592 filename: str, 

593 derived: bool = False, 

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

595 download_filename: t.Optional[str] = None 

596 ) -> str: 

597 """ 

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

599 

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

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

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

603 :param expires: 

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

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

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

607 filename 

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

609 """ 

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

611 expires = datetime.timedelta(minutes=expires) 

612 

613 filename = html.unescape(filename) 

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

615 

616 if download_filename: 

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

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

619 

620 download_filename = urlquote(download_filename) 

621 

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

623 

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

625 sig = cls.hmac_sign(data) 

626 

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

628 

629 @classmethod 

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

631 """ 

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

633 

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

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

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

637 

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

639 

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

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

642 """ 

643 parsed_url = urlsplit(url) 

644 

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

646 return None 

647 

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

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

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

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

652 

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

654 # Invalid signature 

655 return None 

656 

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

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

659 

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

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

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

663 case 1: 

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

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

666 case _: 

667 # Invalid path 

668 return None 

669 

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

671 # Signature expired 

672 return None 

673 

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

675 # Invalid path 

676 return None 

677 

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

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

680 

681 @classmethod 

682 def create_src_set( 

683 cls, 

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

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

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

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

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

689 ) -> str: 

690 """ 

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

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

693 viewport without upscaling. 

694 

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

696 :param expires: 

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

698 otherwise it's lifetime in seconds 

699 :param width: 

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

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

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

703 it will be skipped. 

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

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

706 """ 

707 if not width and not height: 707 ↛ 708line 707 didn't jump to line 708 because the condition on line 707 was never true

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

709 return "" 

710 

711 if isinstance(file, str): 711 ↛ 712line 711 didn't jump to line 712 because the condition on line 711 was never true

712 file = db.Query("file").filter("dlkey =", file).order( 

713 db.QueryOrder("creationdate")).getEntry() 

714 

715 if not file: 715 ↛ 716line 715 didn't jump to line 716 because the condition on line 715 was never true

716 return "" 

717 

718 if isinstance(file, i18n.LanguageWrapper): 718 ↛ 723line 718 didn't jump to line 723 because the condition on line 718 was always true

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

720 if not language or not (file := file.get(language)): 

721 return "" 

722 

723 if "dlkey" not in file and "dest" in file: 723 ↛ 724line 723 didn't jump to line 724 because the condition on line 723 was never true

724 file = file["dest"] 

725 

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

727 

728 if not ( 728 ↛ 733line 728 didn't jump to line 733 because the condition on line 728 was never true

729 isinstance(file, (SkeletonInstance, dict)) 

730 and "dlkey" in file 

731 and "derived" in file 

732 ): 

733 logging.error("Invalid file supplied") 

734 return "" 

735 

736 if not isinstance(file["derived"], dict): 736 ↛ 737line 736 didn't jump to line 737 because the condition on line 736 was never true

737 logging.error("No derives available") 

738 return "" 

739 

740 src_set = [] 

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

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

743 

744 if width and customData.get("width") in width: 744 ↛ 749line 744 didn't jump to line 749 because the condition on line 744 was always true

745 src_set.append( 

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

747 ) 

748 

749 if height and customData.get("height") in height: 749 ↛ 750line 749 didn't jump to line 750 because the condition on line 749 was never true

750 src_set.append( 

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

752 ) 

753 

754 return ", ".join(src_set) 

755 

756 def write( 

757 self, 

758 filename: str, 

759 content: t.Any, 

760 mimetype: str = "text/plain", 

761 *, 

762 width: int = None, 

763 height: int = None, 

764 public: bool = False, 

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

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

767 ) -> db.Key: 

768 """ 

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

770 

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

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

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

774 

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

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

777 

778 :param filename: Filename to be written. 

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

780 :param mimetype: The file's mimetype. 

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

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

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

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

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

786 

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

788 """ 

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

790 if not self.is_valid_filename(filename): 790 ↛ 791line 790 didn't jump to line 791 because the condition on line 790 was never true

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

792 

793 # Folder mode? 

794 if folder: 

795 # Validate correct folder naming 

796 if isinstance(folder, str): 796 ↛ 799line 796 didn't jump to line 799 because the condition on line 796 was always true

797 folder = folder, # make it a tuple 

798 

799 for foldername in folder: 

800 if not self.is_valid_filename(foldername): 800 ↛ 801line 800 didn't jump to line 801 because the condition on line 800 was never true

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

802 

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

804 if rootnode is None: 804 ↛ 807line 804 didn't jump to line 807 because the condition on line 804 was always true

805 rootnode = self.ensureOwnModuleRootNode() 

806 

807 parentrepokey = rootnode.key 

808 parentfolderkey = rootnode.key 

809 

810 for foldername in folder: 

811 query = self.addSkel("node").all() 

812 query.filter("parentrepo", parentrepokey) 

813 query.filter("parententry", parentfolderkey) 

814 query.filter("name", foldername) 

815 

816 if folder_skel := query.getSkel(): 816 ↛ 821line 816 didn't jump to line 821 because the condition on line 816 was always true

817 # Skip existing folder 

818 parentfolderkey = folder_skel["key"]