diff --git a/src/viur/core/modules/file.py b/src/viur/core/modules/file.py index 063310253..a8a8e0d0e 100644 --- a/src/viur/core/modules/file.py +++ b/src/viur/core/modules/file.py @@ -53,7 +53,7 @@ def importBlobFromViur2(dlKey, fileName): - bucket = File.get_bucket(dlKey) + bucket = conf.main_app.file.get_bucket(dlKey) if not conf.viur2import_blobsource: return False @@ -96,14 +96,14 @@ def importBlobFromViur2(dlKey, fileName): marker["success"] = True marker["old_src_key"] = dlKey marker["old_src_name"] = fileName - marker["dlurl"] = File.create_download_url(dlKey, fileName, False, None) + marker["dlurl"] = conf.main_app.file.create_download_url(dlKey, fileName, False, None) db.Put(marker) return marker["dlurl"] def thumbnailer(fileSkel, existingFiles, params): file_name = html.unescape(fileSkel["name"]) - bucket = File.get_bucket(fileSkel["dlkey"]) + bucket = conf.main_app.file.get_bucket(fileSkel["dlkey"]) blob = bucket.get_blob(f"""{fileSkel["dlkey"]}/source/{file_name}""") if not blob: logging.warning(f"""Blob {fileSkel["dlkey"]}/source/{file_name} is missing from cloud storage!""") @@ -185,11 +185,11 @@ def cloudfunction_thumbnailer(fileSkel, existingFiles, params): if not conf.file_thumbnailer_url: raise ValueError("conf.file_thumbnailer_url is not set") - bucket = File.get_bucket(fileSkel["dlkey"]) + bucket = conf.main_app.file.get_bucket(fileSkel["dlkey"]) def getsignedurl(): if conf.instance.is_dev_server: - signedUrl = File.create_download_url(fileSkel["dlkey"], fileSkel["name"]) + signedUrl = conf.main_app.file.create_download_url(fileSkel["dlkey"], fileSkel["name"]) else: path = f"""{fileSkel["dlkey"]}/source/{file_name}""" if not (blob := bucket.get_blob(path)): @@ -209,7 +209,7 @@ def getsignedurl(): def make_request(): headers = {"Content-Type": "application/json"} data_str = base64.b64encode(json.dumps(dataDict).encode("UTF-8")) - sig = File.hmac_sign(data_str) + sig = conf.main_app.file.hmac_sign(data_str) datadump = json.dumps({"dataStr": data_str.decode('ASCII'), "sign": sig}) resp = requests.post(conf.file_thumbnailer_url, data=datadump, headers=headers, allow_redirects=False) if resp.status_code != 200: # Error Handling @@ -256,7 +256,7 @@ def make_request(): uploadUrls = {} for data in derivedData["values"]: - fileName = File.sanitize_filename(data["name"]) + fileName = conf.main_app.file.sanitize_filename(data["name"]) blob = bucket.blob(f"""{fileSkel["dlkey"]}/derived/{fileName}""") uploadUrls[fileSkel["dlkey"] + fileName] = blob.create_resumable_upload_session(timeout=60, content_type=data["mimeType"]) @@ -288,7 +288,7 @@ class DownloadUrlBone(BaseBone): def unserialize(self, skel, name): if "dlkey" in skel.dbEntity and "name" in skel.dbEntity: - skel.accessedValues[name] = File.create_download_url( + skel.accessedValues[name] = conf.main_app.file.create_download_url( skel["dlkey"], skel["name"], expires=conf.render_json_download_url_expiration ) return True @@ -317,7 +317,7 @@ class FileLeafSkel(TreeSkel): descr="Filename", caseSensitive=False, searchable=True, - vfunc=lambda val: None if File.is_valid_filename(val) else "Invalid filename provided", + vfunc=lambda val: None if conf.main_app.file.is_valid_filename(val) else "Invalid filename provided", ) mimetype = StringBone( @@ -469,8 +469,8 @@ def get_bucket(dlkey: str) -> google.cloud.storage.bucket.Bucket: return _private_bucket - @staticmethod - def is_valid_filename(filename: str) -> bool: + @classmethod + def is_valid_filename(cls, filename: str) -> bool: """ Verifies a valid filename. @@ -480,7 +480,7 @@ def is_valid_filename(filename: str) -> bool: Rule set: https://stackoverflow.com/a/31976060/3749896 Regex test: https://regex101.com/r/iBYpoC/1 """ - if len(filename) > File.MAX_FILENAME_LEN: + if len(filename) > cls.MAX_FILENAME_LEN: return False return bool(re.match(VALID_FILENAME_REGEX, filename)) @@ -492,12 +492,13 @@ def hmac_sign(data: t.Any) -> str: data = str(data).encode("UTF-8") return hmac.new(conf.file_hmac_key, msg=data, digestmod=hashlib.sha3_384).hexdigest() - @staticmethod - def hmac_verify(data: t.Any, signature: str) -> bool: - return hmac.compare_digest(File.hmac_sign(data.encode("ASCII")), signature) + @classmethod + def hmac_verify(cls, data: t.Any, signature: str) -> bool: + return hmac.compare_digest(cls.hmac_sign(data.encode("ASCII")), signature) - @staticmethod + @classmethod def create_internal_serving_url( + cls, serving_url: str, size: int = 0, filename: str = "", @@ -527,7 +528,7 @@ def create_internal_serving_url( raise ValueError(f"Invalid {serving_url=!r} provided") # Create internal serving URL - serving_url = File.INTERNAL_SERVING_URL_PREFIX + "/".join(res.groups()) + serving_url = cls.INTERNAL_SERVING_URL_PREFIX + "/".join(res.groups()) # Append additional parameters if params := { @@ -542,8 +543,9 @@ def create_internal_serving_url( return serving_url - @staticmethod + @classmethod def create_download_url( + cls, dlkey: str, filename: str, derived: bool = False, @@ -571,7 +573,7 @@ def create_download_url( filepath = f"""{dlkey}/{"derived" if derived else "source"}/{filename}""" if download_filename: - if not File.is_valid_filename(download_filename): + if not cls.is_valid_filename(download_filename): raise errors.UnprocessableEntity(f"Invalid download_filename {download_filename!r} provided") download_filename = urlquote(download_filename) @@ -579,12 +581,12 @@ def create_download_url( expires = (datetime.datetime.now() + expires).strftime("%Y%m%d%H%M") if expires else 0 data = base64.urlsafe_b64encode(f"""{filepath}\0{expires}\0{download_filename or ""}""".encode("UTF-8")) - sig = File.hmac_sign(data) + sig = cls.hmac_sign(data) - return f"""{File.DOWNLOAD_URL_PREFIX}{data.decode("ASCII")}?sig={sig}""" + return f"""{cls.DOWNLOAD_URL_PREFIX}{data.decode("ASCII")}?sig={sig}""" - @staticmethod - def parse_download_url(url) -> t.Optional[FilePath]: + @classmethod + def parse_download_url(cls, url) -> t.Optional[FilePath]: """ Parses a file download URL in the format `/file/download/xxxx?sig=yyyy` into its FilePath. @@ -593,13 +595,13 @@ def parse_download_url(url) -> t.Optional[FilePath]: :param url: The file download URL to be parsed. :return: A FilePath on success, None otherwise. """ - if not url.startswith(File.DOWNLOAD_URL_PREFIX) or "?" not in url: + if not url.startswith(cls.DOWNLOAD_URL_PREFIX) or "?" not in url: return None - data, sig = url.removeprefix(File.DOWNLOAD_URL_PREFIX).split("?", 1) # Strip "/file/download/" and split on "?" + data, sig = url.removeprefix(cls.DOWNLOAD_URL_PREFIX).split("?", 1) # Strip "/file/download/" and split on "?" sig = sig.removeprefix("sig=") - if not File.hmac_verify(data, sig): + if not cls.hmac_verify(data, sig): # Invalid signature return None @@ -627,8 +629,9 @@ def parse_download_url(url) -> t.Optional[FilePath]: dlkey, derived, filename = dlpath.split("/", 3) return FilePath(dlkey, derived != "source", filename) - @staticmethod + @classmethod def create_src_set( + cls, file: t.Union["SkeletonInstance", dict, str], expires: t.Optional[datetime.timedelta | int] = datetime.timedelta(hours=1), width: t.Optional[int] = None, @@ -664,7 +667,7 @@ def create_src_set( if isinstance(file, LanguageWrapper): language = language or current.language.get() - if not language or not (file := file.get(language)): + if not language or not (file := cls.get(language)): return "" if "dlkey" not in file and "dest" in file: @@ -690,12 +693,12 @@ def create_src_set( if width and customData.get("width") in width: src_set.append( - f"""{File.create_download_url(file["dlkey"], filename, True, expires)} {customData["width"]}w""" + f"""{cls.create_download_url(file["dlkey"], filename, True, expires)} {customData["width"]}w""" ) if height and customData.get("height") in height: src_set.append( - f"""{File.create_download_url(file["dlkey"], filename, True, expires)} {customData["height"]}h""" + f"""{cls.create_download_url(file["dlkey"], filename, True, expires)} {customData["height"]}h""" ) return ", ".join(src_set) @@ -720,7 +723,7 @@ def write( :param public: True if the file should be publicly accessible. :return: Returns the key of the file object written. This can be associated e.g. with a FileBone. """ - if not File.is_valid_filename(filename): + if not self.is_valid_filename(filename): raise ValueError(f"{filename=} is invalid") dl_key = utils.string.random() @@ -728,7 +731,7 @@ def write( if public: dl_key += PUBLIC_DLKEY_SUFFIX # mark file as public - bucket = File.get_bucket(dl_key) + bucket = self.get_bucket(dl_key) blob = bucket.blob(f"{dl_key}/source/{filename}") blob.upload_from_file(io.BytesIO(content), content_type=mimetype) @@ -774,9 +777,9 @@ def read( else: path = f"""{skel["dlkey"]}/source/{skel["name"]}""" - bucket = File.get_bucket(skel["dlkey"]) + bucket = self.get_bucket(skel["dlkey"]) else: - bucket = File.get_bucket(path.split("/", 1)[0]) # path's first part is dlkey plus eventual postfix + bucket = self.get_bucket(path.split("/", 1)[0]) # path's first part is dlkey plus eventual postfix blob = bucket.blob(path) return io.BytesIO(blob.download_as_bytes()), blob.content_type @@ -811,7 +814,7 @@ def getUploadURL( ): filename = fileName.strip() # VIUR4 FIXME: just for compatiblity of the parameter names - if not File.is_valid_filename(filename): + if not self.is_valid_filename(filename): raise errors.UnprocessableEntity(f"Invalid filename {filename!r} provided") # Validate the mimetype from the client seems legit @@ -869,7 +872,7 @@ def getUploadURL( if public: dlkey += PUBLIC_DLKEY_SUFFIX # mark file as public - blob = File.get_bucket(dlkey).blob(f"{dlkey}/source/{filename}") + blob = self.get_bucket(dlkey).blob(f"{dlkey}/source/{filename}") upload_url = blob.create_resumable_upload_session(content_type=mimeType, size=size, timeout=60) # Create a corresponding file-lock object early, otherwise we would have to ensure that the file-lock object @@ -920,7 +923,7 @@ def download(self, blobKey: str, fileName: str = "", download: bool = False, sig :param download: Set header to attachment retrival, set explictly to "1" if download is wanted. """ if filename := fileName.strip(): - if not File.is_valid_filename(filename): + if not self.is_valid_filename(filename): raise errors.UnprocessableEntity(f"The provided filename {filename!r} is invalid!") download_filename = "" @@ -932,7 +935,7 @@ def download(self, blobKey: str, fileName: str = "", download: bool = False, sig dlPath, validUntil = base64.urlsafe_b64decode(blobKey).decode( "UTF-8").split("\0") - bucket = File.get_bucket(dlPath.split("/", 1)[0]) + bucket = self.get_bucket(dlPath.split("/", 1)[0]) if not sig: # Check if the current user has the right to download *any* blob present in this application. @@ -1129,7 +1132,7 @@ def add(self, skelType: SkelType, node: db.Key | int | str | None = None, *args, session.markChanged() # Now read the blob from the dlkey folder - bucket = File.get_bucket(skel["dlkey"]) + bucket = self.get_bucket(skel["dlkey"]) blobs = list(bucket.list_blobs(prefix=f"""{skel["dlkey"]}/""")) if len(blobs) != 1: @@ -1172,7 +1175,7 @@ def onEdit(self, skelType: SkelType, skel: SkeletonInstance): old_path = f"{skel['dlkey']}/source/{html.unescape(old_skel['name'])}" new_path = f"{skel['dlkey']}/source/{html.unescape(skel['name'])}" - bucket = File.get_bucket(skel['dlkey']) + bucket = self.get_bucket(skel['dlkey']) if not (old_blob := bucket.get_blob(old_path)): raise errors.Gone() @@ -1211,7 +1214,7 @@ def inject_serving_url(self, skel: SkeletonInstance) -> None: and skel["mimetype"].startswith("image/") and not skel["serving_url"]: try: - bucket = File.get_bucket(skel['dlkey']) + bucket = self.get_bucket(skel['dlkey']) skel["serving_url"] = images.get_serving_url( None, secure_url=True, @@ -1290,7 +1293,7 @@ def doCleanupDeletedFiles(cursor=None): else: if file["itercount"] > maxIterCount: logging.info(f"""Finally deleting, {file["dlkey"]}""") - bucket = File.get_bucket(file["dlkey"]) + bucket = conf.main_app.file.get_bucket(file["dlkey"]) blobs = bucket.list_blobs(prefix=f"""{file["dlkey"]}/""") for blob in blobs: blob.delete() @@ -1300,7 +1303,7 @@ def doCleanupDeletedFiles(cursor=None): f.delete() if f["serving_url"]: - bucket = File.get_bucket(f["dlkey"]) + bucket = conf.main_app.file.get_bucket(f["dlkey"]) blob_key = blobstore.create_gs_key( f"/gs/{bucket.name}/{f['dlkey']}/source/{f['name']}" ) @@ -1331,7 +1334,7 @@ def start_delete_pending_files(): def __getattr__(attr: str) -> object: if entry := { # stuff prior viur-core < 3.7 - "GOOGLE_STORAGE_BUCKET": ("File.get_bucket()", _private_bucket), + "GOOGLE_STORAGE_BUCKET": ("conf.main_app.file.get_bucket()", _private_bucket), }.get(attr): msg = f"{attr} was replaced by {entry[0]}" warnings.warn(msg, DeprecationWarning, stacklevel=2) diff --git a/src/viur/core/render/html/env/viur.py b/src/viur/core/render/html/env/viur.py index 1162dfc96..5f6636608 100644 --- a/src/viur/core/render/html/env/viur.py +++ b/src/viur/core/render/html/env/viur.py @@ -676,7 +676,7 @@ def downloadUrlFor( return "" if derived: - return file.File.create_download_url( + return conf.main_app.file.create_download_url( fileObj["dlkey"], filename=derived, derived=True, @@ -684,7 +684,7 @@ def downloadUrlFor( download_filename=downloadFileName, ) - return file.File.create_download_url( + return conf.main_app.file.create_download_url( fileObj["dlkey"], filename=fileObj["name"], expires=expires, @@ -719,7 +719,7 @@ def srcSetFor( :param language: Language overwrite if fileObj has multiple languages and we want to explicitly specify one :return: The srctag generated or an empty string if a invalid file object was supplied """ - return file.File.create_src_set(fileObj, expires, width, height, language) + return conf.main_app.file.create_src_set(fileObj, expires, width, height, language) @jinjaGlobalFunction @@ -727,7 +727,7 @@ def serving_url_for(render: Render, *args, **kwargs): """ Jinja wrapper for File.create_internal_serving_url(), see there for parameter information. """ - return file.File.create_internal_serving_url(*args, **kwargs) + return conf.main_app.file.create_internal_serving_url(*args, **kwargs) @jinjaGlobalFunction def seoUrlForEntry(render: Render, *args, **kwargs): diff --git a/src/viur/core/utils/__init__.py b/src/viur/core/utils/__init__.py index 5ff2ecd83..995d86d72 100644 --- a/src/viur/core/utils/__init__.py +++ b/src/viur/core/utils/__init__.py @@ -143,13 +143,13 @@ def ensure_iterable( "currentRequest": ("current.request", current.request), "currentRequestData": ("current.request_data", current.request_data), "currentSession": ("current.session", current.session), - "downloadUrlFor": ("modules.file.File.create_download_url", "viur.core.modules.file.File.create_download_url"), + "downloadUrlFor": ("conf.main_app.file.create_download_url", lambda: conf.main_app.file.create_download_url), "escapeString": ("utils.string.escape", string.escape), "generateRandomString": ("utils.string.random", string.random), "getCurrentUser": ("current.user.get", current.user.get), "is_prefix": ("utils.string.is_prefix", string.is_prefix), "parse_bool": ("utils.parse.bool", parse.bool), - "srcSetFor": ("modules.file.File.create_src_set", "viur.core.modules.file.File.create_src_set"), + "srcSetFor": ("conf.main_app.file.create_src_set", lambda: conf.main_app.file.create_src_set), } @@ -164,16 +164,9 @@ def __getattr__(attr): msg = f"Use of `utils.{attr}` is deprecated; Use `{replace[0]}` instead!" warnings.warn(msg, DeprecationWarning, stacklevel=3) logging.warning(msg, stacklevel=3) - - ret = replace[1] - - # When this is a string, try to resolve by dynamic import - if isinstance(ret, str): - mod, item, attr = ret.rsplit(".", 2) - mod = __import__(mod, fromlist=(item,)) - item = getattr(mod, item) - ret = getattr(item, attr) - - return ret + res = replace[1] + if isinstance(res, t.Callable): + res = res() + return res return super(__import__(__name__).__class__).__getattribute__(attr)