storages.py 1.9 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758
  1. import base64
  2. import datetime
  3. import hmac
  4. import json
  5. import os
  6. from django.conf import settings
  7. from django.core.files.storage import FileSystemStorage, default_storage
  8. from django.utils._os import safe_join
  9. class S3MockStorage(FileSystemStorage):
  10. @property
  11. def location(self):
  12. return safe_join(os.path.abspath(self.base_location), get_aws_location())
  13. class connection:
  14. class meta:
  15. class client:
  16. @staticmethod
  17. def generate_presigned_post(bucket_name, key, **policy):
  18. policy = json.dumps(policy).encode()
  19. policy_b64 = base64.b64encode(policy).decode()
  20. date = datetime.datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
  21. aws_id = getattr(
  22. settings,
  23. "AWS_ACCESS_KEY_ID",
  24. "AWS_ACCESS_KEY_ID",
  25. )
  26. fields = {
  27. "x-amz-algorithm": "AWS4-HMAC-SHA256",
  28. "x-amz-date": date,
  29. "x-amz-credential": aws_id,
  30. "policy": policy_b64,
  31. "key": key,
  32. }
  33. signature = hmac.new(
  34. settings.SECRET_KEY.encode(),
  35. policy + date.encode(),
  36. "sha256",
  37. ).digest()
  38. signature = base64.b64encode(signature).decode()
  39. return {
  40. "url": "/__s3_mock__/",
  41. "fields": {"x-amz-signature": signature, **fields},
  42. }
  43. class bucket:
  44. name = "test-bucket"
  45. local_dev = isinstance(default_storage, FileSystemStorage)
  46. storage = default_storage if not local_dev else S3MockStorage()
  47. def get_aws_location():
  48. return getattr(settings, "AWS_LOCATION", "")