from __future__ import annotations import base64 import hashlib import hmac import os import time from datetime import datetime, timezone from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import quote, urlencode from urllib.request import Request, urlopen class AliyunOssSignedUploadService: storage_provider = "aliyun_oss" def __init__( self, *, endpoint: str, bucket: str, region: str, access_key_id: str, access_key_secret: str, now_epoch: int | None = None, ) -> None: _require_text(endpoint, "ALIYUN_OSS_ENDPOINT") _require_text(bucket, "ALIYUN_OSS_BUCKET") _require_text(region, "ALIYUN_OSS_REGION") _require_text(access_key_id, "ALIYUN_OSS_ACCESS_KEY_ID") _require_text(access_key_secret, "ALIYUN_OSS_ACCESS_KEY_SECRET") self.endpoint = endpoint.removeprefix("https://").removeprefix("http://").strip("/") self.bucket = bucket.strip() self.storage_region = region.strip() self._access_key_id = access_key_id.strip() self._access_key_secret = access_key_secret.strip() self._now_epoch = now_epoch @classmethod def from_env(cls) -> "AliyunOssSignedUploadService": missing = [ name for name in ( "ALIYUN_OSS_ENDPOINT", "ALIYUN_OSS_BUCKET", "ALIYUN_OSS_REGION", "ALIYUN_OSS_ACCESS_KEY_ID", "ALIYUN_OSS_ACCESS_KEY_SECRET", ) if not os.environ.get(name, "").strip() ] if missing: raise ValueError("Missing OSS upload configuration: " + ", ".join(missing)) return cls( endpoint=os.environ["ALIYUN_OSS_ENDPOINT"], bucket=os.environ["ALIYUN_OSS_BUCKET"], region=os.environ["ALIYUN_OSS_REGION"], access_key_id=os.environ["ALIYUN_OSS_ACCESS_KEY_ID"], access_key_secret=os.environ["ALIYUN_OSS_ACCESS_KEY_SECRET"], ) def create_upload( self, *, object_key: str, mime_type: str, size_bytes: int, expires_in_seconds: int, ) -> dict[str, Any]: _validate_object_key(object_key) _require_text(mime_type, "mime_type") if size_bytes <= 0: raise ValueError("size_bytes must be greater than zero") if expires_in_seconds <= 0 or expires_in_seconds > 900: raise ValueError("expires_in_seconds must be between 1 and 900") now = self._now_epoch if self._now_epoch is not None else int(time.time()) expires = now + expires_in_seconds canonical_resource = f"/{self.bucket}/{object_key}" string_to_sign = f"PUT\n\n{mime_type}\n{expires}\n{canonical_resource}" signature = base64.b64encode( hmac.new( self._access_key_secret.encode("utf-8"), string_to_sign.encode("utf-8"), hashlib.sha1, ).digest() ).decode("ascii") query = urlencode( { "OSSAccessKeyId": self._access_key_id, "Expires": str(expires), "Signature": signature, } ) encoded_key = quote(object_key, safe="/") return { "method": "PUT", "url": f"https://{self.bucket}.{self.endpoint}/{encoded_key}?{query}", "headers": {"Content-Type": mime_type}, "expires_at": datetime.fromtimestamp(expires, tz=timezone.utc) .replace(microsecond=0) .isoformat() .replace("+00:00", "Z"), } def verify_upload( self, *, object_key: str, mime_type: str, size_bytes: int, sha256: str, ) -> dict[str, Any]: _validate_object_key(object_key) _require_text(mime_type, "mime_type") if size_bytes <= 0: raise ValueError("size_bytes must be greater than zero") if len(str(sha256)) != 64: raise ValueError("sha256 must be a 64-character digest") head_status, head_headers, _ = self._request_signed_object(method="HEAD", object_key=object_key) if head_status != 200: raise ValueError("uploaded object is not available") content_length = int(head_headers.get("Content-Length") or head_headers.get("content-length") or -1) if content_length != size_bytes: raise ValueError("uploaded object size does not match the expected file size") content_type = (head_headers.get("Content-Type") or head_headers.get("content-type") or "").split(";", 1)[0].strip() if content_type and content_type.lower() != mime_type.lower(): raise ValueError("uploaded object content type does not match the expected file type") get_status, _, body = self._request_signed_object(method="GET", object_key=object_key) if get_status != 200: raise ValueError("uploaded object could not be read for integrity verification") actual_sha256 = hashlib.sha256(body).hexdigest() if actual_sha256 != sha256: raise ValueError("uploaded object checksum does not match the expected file") return { "size_bytes": len(body), "sha256": actual_sha256, "content_type": content_type or mime_type, } def _request_signed_object(self, *, method: str, object_key: str) -> tuple[int, dict[str, str], bytes]: url = self._signed_object_url(method=method, object_key=object_key) request = Request(url, method=method) try: with urlopen(request, timeout=30) as response: body = response.read() if method != "HEAD" else b"" return response.status, dict(response.headers), body except HTTPError as exc: body = exc.read() if method != "HEAD" and exc.fp else b"" return exc.code, dict(exc.headers), body except URLError as exc: raise ValueError("uploaded object is not available") from exc def _signed_object_url(self, *, method: str, object_key: str, expires_in_seconds: int = 600) -> str: _validate_object_key(object_key) now = self._now_epoch if self._now_epoch is not None else int(time.time()) expires = now + expires_in_seconds canonical_resource = f"/{self.bucket}/{object_key}" string_to_sign = f"{method}\n\n\n{expires}\n{canonical_resource}" signature = base64.b64encode( hmac.new( self._access_key_secret.encode("utf-8"), string_to_sign.encode("utf-8"), hashlib.sha1, ).digest() ).decode("ascii") query = urlencode( { "OSSAccessKeyId": self._access_key_id, "Expires": str(expires), "Signature": signature, } ) return f"https://{self.bucket}.{self.endpoint}/{quote(object_key, safe='/')}?{query}" def _require_text(value: str | None, field_name: str) -> None: if not str(value or "").strip(): raise ValueError(f"{field_name} is required") def _validate_object_key(object_key: str) -> None: cleaned = str(object_key or "").strip() if not cleaned: raise ValueError("object_key is required") if cleaned.startswith("/") or ".." in cleaned.split("/"): raise ValueError("object_key is invalid") if not cleaned.startswith("users/"): raise ValueError("object_key must be in the users namespace")