1 files changed, 104 insertions, 0 deletions
diff --git a/paste/s3.py b/paste/s3.py
new file mode 100644
index 0000000..a9b2b21
--- /dev/null
+++ b/paste/s3.py
@@ -0,0 +1,104 @@
+from datetime import datetime, UTC
+from typing import Dict
+from dataclasses import dataclass
+from bozo4 import s3v4_sign_request, s3v4_datetime_string
+
+
+@dataclass
+class Request:
+    url: str
+    headers: Dict[str, str]
+
+
+@dataclass
+class Bucket:
+    endpoint: str
+    region: str
+    bucket: str
+    access_key: str
+    secret_key: str
+
+    def get(self, key: str) -> Request:
+        now = datetime.now(UTC)
+
+        headers = {
+            "Host": self.endpoint,
+            "X-Amz-Date": s3v4_datetime_string(now),
+            "X-Amz-Content-SHA256": "UNSIGNED-PAYLOAD",
+        }
+
+        auth = s3v4_sign_request(
+            endpoint=self.endpoint,
+            region=self.region,
+            access_key=self.access_key,
+            secret_key=self.secret_key,
+            request_method="GET",
+            date=now,
+            payload_hash="UNSIGNED-PAYLOAD",
+            uri=f"/{self.bucket}/{key}",
+            parameters={},
+            headers=headers,
+            service="s3",
+        )
+
+        headers["Authorization"] = auth
+
+        return Request(f"https://{self.endpoint}/{self.bucket}/{key}", headers)
+
+    def put(
+        self, key: str, content_length: int, content_type: str, payload_hash: str
+    ) -> Request:
+        now = datetime.now(UTC)
+
+        headers = {
+            "Host": self.endpoint,
+            "X-Amz-Date": s3v4_datetime_string(now),
+            "X-Amz-Content-SHA256": payload_hash,
+            "Content-Length": str(content_length),
+            "Content-Type": content_type,
+        }
+
+        auth = s3v4_sign_request(
+            endpoint=self.endpoint,
+            region=self.region,
+            access_key=self.access_key,
+            secret_key=self.secret_key,
+            request_method="PUT",
+            date=now,
+            payload_hash=payload_hash,
+            uri=f"/{self.bucket}/{key}",
+            parameters={},
+            headers=headers,
+            service="s3",
+        )
+
+        headers["Authorization"] = auth
+
+        return Request(f"https://{self.endpoint}/{self.bucket}/{key}", headers)
+
+    def delete(self, key: str) -> Request:
+        now = datetime.now(UTC)
+
+        headers = {
+            "Host": self.endpoint,
+            "X-Amz-Date": s3v4_datetime_string(now),
+            "X-Amz-Content-SHA256": "UNSIGNED-PAYLOAD",
+        }
+
+        auth = s3v4_sign_request(
+            endpoint=self.endpoint,
+            region=self.region,
+            access_key=self.access_key,
+            secret_key=self.secret_key,
+            request_method="DELETE",
+            date=now,
+            payload_hash="UNSIGNED-PAYLOAD",
+            uri=f"/{self.bucket}/{key}",
+            parameters={},
+            headers=headers,
+            service="s3",
+        )
+
+        headers["Authorization"] = auth
+
+        return Request(f"https://{self.endpoint}/{self.bucket}/{key}", headers)