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 

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

28from viur.core.decorators import * 

29from viur.core.i18n import LanguageWrapper 

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 = 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"] = 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 = 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 = File.get_bucket(fileSkel["dlkey"]) 

190 

191 def getsignedurl(): 

192 if conf.instance.is_dev_server: 

193 signedUrl = 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 = f"""filename={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 = 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 = 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] = 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 size = StringBone( 

307 descr="Size", 

308 readOnly=True, 

309 searchable=True, 

310 ) 

311 

312 dlkey = StringBone( 

313 descr="Download-Key", 

314 readOnly=True, 

315 ) 

316 

317 name = StringBone( 

318 descr="Filename", 

319 caseSensitive=False, 

320 searchable=True, 

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

322 ) 

323 

324 mimetype = StringBone( 

325 descr="MIME-Type", 

326 readOnly=True, 

327 ) 

328 

329 weak = BooleanBone( 

330 descr="Weak reference", 

331 readOnly=True, 

332 visible=False, 

333 ) 

334 

335 pending = BooleanBone( 

336 descr="Pending upload", 

337 readOnly=True, 

338 visible=False, 

339 defaultValue=False, 

340 ) 

341 

342 width = NumericBone( 

343 descr="Width", 

344 readOnly=True, 

345 searchable=True, 

346 ) 

347 

348 height = NumericBone( 

349 descr="Height", 

350 readOnly=True, 

351 searchable=True, 

352 ) 

353 

354 downloadUrl = DownloadUrlBone( 

355 descr="Download-URL", 

356 readOnly=True, 

357 visible=False, 

358 ) 

359 

360 derived = BaseBone( 

361 descr="Derived Files", 

362 readOnly=True, 

363 visible=False, 

364 ) 

365 

366 pendingparententry = KeyBone( 

367 descr="Pending key Reference", 

368 readOnly=True, 

369 visible=False, 

370 ) 

371 

372 crc32c_checksum = StringBone( 

373 descr="CRC32C checksum", 

374 readOnly=True, 

375 ) 

376 

377 md5_checksum = StringBone( 

378 descr="MD5 checksum", 

379 readOnly=True, 

380 ) 

381 

382 public = BooleanBone( 

383 descr="Public File", 

384 readOnly=True, 

385 defaultValue=False, 

386 ) 

387 

388 serving_url = StringBone( 

389 descr="Serving-URL", 

390 readOnly=True, 

391 params={ 

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

393 } 

394 ) 

395 

396 @classmethod 

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

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

399 if ( 

400 skel["public"] 

401 and skel["mimetype"] 

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

403 and not skel["serving_url"] 

404 ): 

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

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

407 

408 # Trying this on local development server will raise a 

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

410 if conf.instance.is_dev_server: 

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

412 return 

413 

414 try: 

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

416 

417 except Exception as e: 

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

419 logging.exception(e) 

420 

421 def preProcessBlobLocks(self, locks): 

422 """ 

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

424 """ 

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

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

427 return locks 

428 

429 @classmethod 

430 def refresh(cls, skel): 

431 super().refresh(skel) 

432 if conf.viur2import_blobsource: 

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

434 if importData: 

435 if not skel["downloadUrl"]: 

436 skel["downloadUrl"] = importData 

437 skel["pendingparententry"] = None 

438 

439 cls._inject_serving_url(skel) 

440 

441 @classmethod 

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

443 cls._inject_serving_url(skel) 

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

445 

446 

447class FileNodeSkel(TreeSkel): 

448 """ 

449 Default file node skeleton. 

450 """ 

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

452 

453 name = StringBone( 

454 descr="Name", 

455 required=True, 

456 searchable=True 

457 ) 

458 

459 rootNode = BooleanBone( 

460 descr="Is RootNode", 

461 defaultValue=False, 

462 readOnly=True, 

463 visible=False, 

464 ) 

465 

466 public = BooleanBone( 

467 descr="Is public?", 

468 defaultValue=False, 

469 readOnly=True, 

470 visible=False, 

471 ) 

472 

473 viurCurrentSeoKeys = None 

474 

475 

476class File(Tree): 

477 PENDING_POSTFIX = " (pending)" 

478 DOWNLOAD_URL_PREFIX = "/file/download/" 

479 INTERNAL_SERVING_URL_PREFIX = "/file/serve/" 

480 MAX_FILENAME_LEN = 256 

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

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

483 Default: 10 MiB""" 

484 

485 leafSkelCls = FileLeafSkel 

486 nodeSkelCls = FileNodeSkel 

487 

488 handler = "tree.simple.file" 

489 adminInfo = { 

490 "icon": "folder-fill", 

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

492 } 

493 

494 roles = { 

495 "*": "view", 

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

497 "admin": "*", 

498 } 

499 

500 default_order = "name" 

501 

502 # Helper functions currently resist here 

503 

504 @staticmethod 

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

506 """ 

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

508 """ 

509 global _public_bucket 

510 if dlkey and dlkey.endswith(PUBLIC_DLKEY_SUFFIX): 

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

512 return _public_bucket 

513 

514 raise ValueError( 

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

516 ) 

517 

518 return _private_bucket 

519 

520 @staticmethod 

521 def is_valid_filename(filename: str) -> bool: 

522 """ 

523 Verifies a valid filename. 

524 

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

526 It should not be longer than MAX_FILENAME_LEN chars. 

527 

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

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

530 """ 

531 if len(filename) > File.MAX_FILENAME_LEN: 

532 return False 

533 

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

535 

536 @staticmethod 

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

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

539 if not isinstance(data, bytes): 

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

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

542 

543 @staticmethod 

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

545 return hmac.compare_digest(File.hmac_sign(data.encode("ASCII")), signature) 

546 

547 @staticmethod 

548 def create_internal_serving_url( 

549 serving_url: str, 

550 size: int = 0, 

551 filename: str = "", 

552 options: str = "", 

553 download: bool = False 

554 ) -> str: 

555 """ 

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

557 

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

559 legal requirement like GDPR. 

560 

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

562 :param size: Optional size setting 

563 :param filename: Optonal filename setting 

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

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

566 """ 

567 

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

569 res = re.match( 

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

571 serving_url 

572 ) 

573 

574 if not res: 

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

576 

577 # Create internal serving URL 

578 serving_url = File.INTERNAL_SERVING_URL_PREFIX + "/".join(res.groups()) 

579 

580 # Append additional parameters 

581 if params := { 

582 k: v for k, v in { 

583 "download": download, 

584 "filename": filename, 

585 "options": options, 

586 "size": size, 

587 }.items() if v 

588 }: 

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

590 

591 return serving_url 

592 

593 @staticmethod 

594 def create_download_url( 

595 dlkey: str, 

596 filename: str, 

597 derived: bool = False, 

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

599 download_filename: t.Optional[str] = None 

600 ) -> str: 

601 """ 

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

603 

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

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

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

607 :param expires: 

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

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

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

611 filename 

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

613 """ 

614 if isinstance(expires, int): 

615 expires = datetime.timedelta(minutes=expires) 

616 

617 # Undo escaping on ()= performed on fileNames 

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

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

620 

621 if download_filename: 

622 if not File.is_valid_filename(download_filename): 

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

624 

625 download_filename = urlquote(download_filename) 

626 

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

628 

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

630 sig = File.hmac_sign(data) 

631 

632 return f"""{File.DOWNLOAD_URL_PREFIX}{data.decode("ASCII")}?sig={sig}""" 

633 

634 @staticmethod 

635 def parse_download_url(url) -> t.Optional[FilePath]: 

636 """ 

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

638 

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

640 

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

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

643 """ 

644 if not url.startswith(File.DOWNLOAD_URL_PREFIX) or "?" not in url: 

645 return None 

646 

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

648 sig = sig.removeprefix("sig=") 

649 

650 if not File.hmac_verify(data, sig): 

651 # Invalid signature 

652 return None 

653 

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

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

656 

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

658 case 2: 

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

660 case 1: 

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

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

663 case _: 

664 # Invalid path 

665 return None 

666 

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

668 # Signature expired 

669 return None 

670 

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

672 # Invalid path 

673 return None 

674 

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

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

677 

678 @staticmethod 

679 def create_src_set( 

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

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

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

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

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

685 ) -> str: 

686 """ 

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

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

689 viewport without upscaling. 

690 

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

692 :param expires: 

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

694 otherwise it's lifetime in seconds 

695 :param width: 

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

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

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

699 it will be skipped. 

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

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

702 """ 

703 if not width and not height: 

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

705 return "" 

706 

707 if isinstance(file, str): 

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

709 

710 if not file: 

711 return "" 

712 

713 if isinstance(file, LanguageWrapper): 

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

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

716 return "" 

717 

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

719 file = file["dest"] 

720 

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

722 

723 if not ( 

724 isinstance(file, (SkeletonInstance, dict)) 

725 and "dlkey" in file 

726 and "derived" in file 

727 ): 

728 logging.error("Invalid file supplied") 

729 return "" 

730 

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

732 logging.error("No derives available") 

733 return "" 

734 

735 src_set = [] 

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

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

738 

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

740 src_set.append( 

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

742 ) 

743 

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

745 src_set.append( 

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

747 ) 

748 

749 return ", ".join(src_set) 

750 

751 def write( 

752 self, 

753 filename: str, 

754 content: t.Any, 

755 mimetype: str = "text/plain", 

756 width: int = None, 

757 height: int = None, 

758 public: bool = False, 

759 ) -> db.Key: 

760 """ 

761 Write a file from any buffer into the file module. 

762 

763 :param filename: Filename to be written. 

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

765 :param mimetype: The file's mimetype. 

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

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

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

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

770 """ 

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

772 if not File.is_valid_filename(filename): 

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

774 

775 dl_key = utils.string.random() 

776 

777 if public: 

778 dl_key += PUBLIC_DLKEY_SUFFIX # mark file as public 

779 

780 bucket = File.get_bucket(dl_key) 

781 

782 blob = bucket.blob(f"{dl_key}/source/{filename}") 

783 blob.upload_from_file(io.BytesIO(content), content_type=mimetype) 

784 

785 skel = self.addSkel("leaf") 

786 skel["name"] = filename 

787 skel["size"] = blob.size 

788 skel["mimetype"] = mimetype 

789 skel["dlkey"] = dl_key 

790 skel["weak"] = True 

791 skel["public"] = public 

792 skel["width"] = width 

793 skel["height"] = height 

794 skel["crc32c_checksum"] = base64.b64decode(blob.crc32c).hex() 

795 skel["md5_checksum"] = base64.b64decode(blob.md5_hash).hex() 

796 

797 skel.write() 

798 return skel["key"] 

799 

800 def read( 

801 self, 

802 key: db.Key | int | str | None = None, 

803 path: str | None = None, 

804 ) -> tuple[io.BytesIO, str]: 

805 """ 

806 Read a file from the Cloud Storage. 

807 

808 If a key and a path are provided, the key is preferred. 

809 This means that the entry in the db is searched first and if this is not found, the path is used. 

810 

811 :param key: Key of the LeafSkel that contains the "dlkey" and the "name".