if database_name == "firebase":
db = DATABASE_IDS.handlers().get(database_name)
db_handler = DBRecipeLoader(db)
+ db_handler.validate_input_recipe_path(filename)
recipe_id = file_path.split("/")[-1]
db_doc, _ = db_handler.collect_docs_by_id(
collection="recipes", id=recipe_id
diff --git a/_modules/cellpack/autopack/AWSHandler.html b/_modules/cellpack/autopack/AWSHandler.html
index 96ead53c..284823ff 100644
--- a/_modules/cellpack/autopack/AWSHandler.html
+++ b/_modules/cellpack/autopack/AWSHandler.html
@@ -40,6 +40,7 @@ Navigation
Source code for cellpack.autopack.AWSHandler
import logging
from pathlib import Path
+from urllib.parse import parse_qs, urlparse, urlunparse
import boto3
from botocore.exceptions import ClientError
@@ -84,7 +85,7 @@ Source code for cellpack.autopack.AWSHandler
[docs]
def get_aws_object_key(self, object_name):
if self.folder_name is not None:
- object_name = self.folder_name + object_name
+ object_name = f"{self.folder_name}/{object_name}"
else:
object_name = object_name
return object_name
@@ -126,29 +127,55 @@ Source code for cellpack.autopack.AWSHandler
"""
object_name = self.get_aws_object_key(object_name)
# Generate a presigned URL for the S3 object
+ # The response contains the presigned URL
+ # https://{self.bucket_name}.s3.{region}.amazonaws.com/{object_key}
try:
url = self.s3_client.generate_presigned_url(
"get_object",
Params={"Bucket": self.bucket_name, "Key": object_name},
ExpiresIn=expiration,
)
+ base_url = urlunparse(urlparse(url)._replace(query="", fragment=""))
+ return base_url
except ClientError as e:
- logging.error(e)
- return None
- # The response contains the presigned URL
- # https://{self.bucket_name}.s3.{region}.amazonaws.com/{object_key}
- return url
+ logging.error(f"Error generating presigned URL: {e}")
+ return None
-
-[docs]
- def save_file(self, file_path):
+
+[docs]
+ def is_url_valid(self, url):
+ """
+ Validate the url's scheme, bucket name, and query parameters, etc.
+ """
+ parsed_url = urlparse(url)
+ # Check the scheme
+ if parsed_url.scheme != "https":
+ return False
+ # Check the bucket name
+ if not parsed_url.path.startswith(f"/{self.bucket_name}/"):
+ return False
+ # Check unwanted query parameters
+ unwanted_query_params = ["AWSAccessKeyId", "Signature", "Expires"]
+ if parsed_url.query:
+ query_params = parse_qs(parsed_url.query)
+ for param in unwanted_query_params:
+ if param in query_params:
+ return False
+ return True
+
+
+
+[docs]
+ def save_file_and_get_url(self, file_path):
"""
- Uploads a file to S3 and returns the presigned url
+ Uploads a file to S3 and returns the base url
"""
file_name = self.upload_file(file_path)
- if file_name:
- return file_name, self.create_presigned_url(file_name)
+ base_url = self.create_presigned_url(file_name)
+ if file_name and base_url:
+ if self.is_url_valid(base_url):
+ return file_name, base_url
Uploads a file to S3 and returns the presigned url
+Validate the url’s scheme, bucket name, and query parameters, etc.
+Validates if the input path corresponds to a recipe path in the database. +Format of a recipe path: firebase:recipes/[RECIPE-ID]
+AWSHandler
DBRecipeLoader.collect_docs_by_id()DBRecipeLoader.compile_db_recipe_data()DBRecipeLoader.prep_db_doc_for_download()DBRecipeLoader.validate_input_recipe_path()DBUploaderFirebaseHandler.get_path_from_ref()FirebaseHandler.get_staging_creds()FirebaseHandler.get_username()FirebaseHandler.get_value()FirebaseHandler.is_firebase_obj()FirebaseHandler.is_reference()FirebaseHandler.set_doc()AWSHandler
DBRecipeLoader.collect_docs_by_id()
DBRecipeLoader.compile_db_recipe_data()DBRecipeLoader.prep_db_doc_for_download()DBRecipeLoader.validate_input_recipe_path()DBUploaderFirebaseHandler.get_path_from_ref()FirebaseHandler.get_staging_creds()FirebaseHandler.get_username()FirebaseHandler.get_value()FirebaseHandler.is_firebase_obj()FirebaseHandler.is_reference()FirebaseHandler.set_doc()