PYTHON / DOWNLOAD_FUNCTION
THE FINAL DOWNLOAD ACCESS GATE.
Identity, user ownership, Blob existence and clean scan metadata are verified before a short-lived read-only SAS URL is generated.
EASY AUTH PRINCIPALUSER-SCOPED PATHSCANSTATUS=CLEANREAD SAS / 15 MIN
PYTHONfunctions/download_function/__init__.py
READ ONLYdef main(req: func.HttpRequest) -> func.HttpResponse:
principal = parse_easy_auth_principal(req)
user_id = principal.get("user_id", "unknown")
if user_id == "unknown":
return func.HttpResponse(
"Unauthorized.",
status_code=401,
)
file_name = req.params.get("fileName")
if not file_name or not is_valid_filename(file_name):
return func.HttpResponse(
"Invalid file request.",
status_code=400,
)
blob_name = f"{user_id}/{file_name}"
credential = DefaultAzureCredential()
blob_service = BlobServiceClient(
account_url=os.environ["STORAGE_ACCOUNT_URL"],
credential=credential,
)
blob_client = blob_service.get_blob_client(
container=os.environ["SAFE_CONTAINER"],
blob=blob_name,
)
properties = blob_client.get_blob_properties()
metadata = properties.metadata or {}
scan_status = metadata.get(
"scanstatus",
metadata.get("scanStatus", "unknown"),
)
if scan_status != "clean":
return func.HttpResponse(
"File is not available for download.",
status_code=403,
)
now = datetime.datetime.now(datetime.timezone.utc)
expiry = now + datetime.timedelta(minutes=15)
delegation_key = blob_service.get_user_delegation_key(
key_start_time=now,
key_expiry_time=expiry,
)
sas_token = generate_blob_sas(
account_name=account_name,
container_name=safe_container,
blob_name=blob_name,
user_delegation_key=delegation_key,
permission=BlobSasPermissions(read=True),
expiry=expiry,
)
return func.HttpResponse(
json.dumps({"downloadUrl": sas_url}),
mimetype="application/json",
status_code=200,
)