Coverage for src/ezcompiler/adapters/_r2_uploader.py: 75.26%

1# /////////////////////////////////////////////////////////////// 

2# R2_UPLOADER - Cloudflare R2 (S3-compatible) uploader implementation 

3# Project: ezcompiler 

4# /////////////////////////////////////////////////////////////// 

5 

6"""R2 uploader - bidirectional S3-compatible backend for the TUF update channel. 

7 

8Credentials are read from environment variables only and are never logged 

9nor included in error messages. boto3 is imported lazily (extra ``[r2]``). 

10""" 

11 

12from __future__ import annotations 

13 

14# /////////////////////////////////////////////////////////////// 

15# IMPORTS 

16# /////////////////////////////////////////////////////////////// 

17# Standard library imports 

18import os 

19from pathlib import Path 

20from typing import Any 

21 

22# Local imports 

23from ..shared.exceptions import UploadError 

24from ..utils._uploader_utils import UploaderUtils 

25from .base_uploader import BaseUploader 

26 

27# /////////////////////////////////////////////////////////////// 

28# CLASSES 

29# /////////////////////////////////////////////////////////////// 

30 

31 

32class R2Uploader(BaseUploader): 

33 """Uploader for Cloudflare R2 (S3-compatible). 

34 

35 Configuration keys: 

36 bucket (str): Target R2 bucket name (required). 

37 

38 Environment variables (required, write credentials): 

39 R2_ENDPOINT or R2_ACCOUNT_ID, R2_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY. 

40 """ 

41 

42 # //////////////////////////////////////////////// 

43 # INITIALIZATION 

44 # //////////////////////////////////////////////// 

45 

46 def __init__(self, config: dict[str, Any] | None = None) -> None: 

47 """Initialize the R2 uploader and build the S3 client. 

48 

49 Args: 

50 config: Configuration dictionary (requires ``bucket``). 

51 """ 

52 UploaderUtils.reject_unknown_config_keys(config, ("bucket",), uploader="r2") 

53 super().__init__(config or {}) 

54 self._bucket = self._config["bucket"] 

55 self._endpoint = self._resolve_endpoint() 

56 self._client = self._build_client() 

57 

58 # //////////////////////////////////////////////// 

59 # PUBLIC METHODS 

60 # //////////////////////////////////////////////// 

61 

62 def get_uploader_name(self) -> str: 

63 """Get the name of this uploader. 

64 

65 Returns: 

66 str: Name of the uploader. 

67 """ 

68 return "R2 Uploader" 

69 

70 def upload(self, source_path: Path, destination: str) -> None: 

71 """Upload a file or directory tree under the ``destination`` prefix. 

72 

73 For a directory, each file is uploaded with its POSIX path relative 

74 to ``source_path`` appended to the prefix. Metadata is uploaded last 

75 to keep the published tree TUF-consistent. 

76 

77 Args: 

78 source_path: Path to the source file or directory. 

79 destination: Object key prefix under the bucket. 

80 

81 Raises: 

82 UploadError: If any object upload fails. 

83 """ 

84 try: 

85 self._validate_source_path(source_path) 

86 prefix = destination.strip("/") 

87 if source_path.is_file(): 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true

88 self._put(source_path, f"{prefix}/{source_path.name}") 

89 return 

90 

91 files = sorted(p for p in source_path.rglob("*") if p.is_file()) 

92 # targets/zip first, metadata last (TUF consistency) 

93 files.sort(key=lambda p: "metadata/" in p.as_posix()) 

94 for file_path in files: 

95 rel = file_path.relative_to(source_path).as_posix() 

96 self._put(file_path, f"{prefix}/{rel}") 

97 except UploadError: 

98 raise 

99 except Exception as e: 

100 raise UploadError(f"R2 upload failed: {e}") from e 

101 

102 def download(self, remote_source: str, local_dir: Path) -> None: 

103 """Download every object under the ``remote_source`` prefix. 

104 

105 Args: 

106 remote_source: Object key prefix to fetch. 

107 local_dir: Local directory to populate. 

108 

109 Raises: 

110 UploadError: If a download fails. 

111 

112 Note: 

113 An empty prefix (first run) is a no-op. 

114 """ 

115 try: 

116 prefix = remote_source.strip("/") 

117 paginator = self._client.get_paginator("list_objects_v2") 

118 for page in paginator.paginate(Bucket=self._bucket, Prefix=prefix): 

119 for obj in page.get("Contents", []): 

120 key = obj["Key"] 

121 rel = key[len(prefix) :].lstrip("/") 

122 dest = self._safe_join(local_dir, rel) 

123 dest.parent.mkdir(parents=True, exist_ok=True) 

124 self._client.download_file( 

125 Bucket=self._bucket, Key=key, Filename=str(dest) 

126 ) 

127 except UploadError: 

128 raise 

129 except Exception as e: 

130 raise UploadError(f"R2 download failed: {e}") from e 

131 

132 # //////////////////////////////////////////////// 

133 # PRIVATE METHODS 

134 # //////////////////////////////////////////////// 

135 

136 @staticmethod 

137 def _safe_join(root: Path, rel: str) -> Path: 

138 """Join ``rel`` under ``root``, rejecting path traversal. 

139 

140 Object keys come from the remote bucket listing; a crafted key must 

141 never let the downloaded file escape ``root``. 

142 

143 Raises: 

144 UploadError: If ``rel`` resolves outside ``root``. 

145 """ 

146 root_resolved = root.resolve() 

147 candidate = (root / rel).resolve() 

148 if root_resolved != candidate and root_resolved not in candidate.parents: 148 ↛ 149line 148 didn't jump to line 149 because the condition on line 148 was never true

149 raise UploadError(f"Unsafe object key rejected: {rel}") 

150 return candidate 

151 

152 def _put(self, source_path: Path, key: str) -> None: 

153 """Upload a single file to the bucket under ``key``.""" 

154 self._client.upload_file( 

155 Filename=str(source_path), Bucket=self._bucket, Key=key 

156 ) 

157 

158 def _resolve_endpoint(self) -> str: 

159 """Resolve the R2 endpoint URL from env vars. 

160 

161 Raises: 

162 UploadError: If neither R2_ENDPOINT nor R2_ACCOUNT_ID is set. 

163 """ 

164 endpoint = os.environ.get("R2_ENDPOINT") 

165 if endpoint: 165 ↛ 167line 165 didn't jump to line 167 because the condition on line 165 was always true

166 return endpoint 

167 account = os.environ.get("R2_ACCOUNT_ID") 

168 if not account: 

169 raise UploadError("Missing R2_ENDPOINT or R2_ACCOUNT_ID env var") 

170 return f"https://{account}.r2.cloudflarestorage.com" 

171 

172 def _build_client(self) -> Any: 

173 """Build the boto3 S3 client (lazy import, extra ``[r2]``).""" 

174 try: 

175 import boto3 # noqa: PLC0415 # pyright: ignore[reportMissingImports] 

176 except ImportError as exc: 

177 raise UploadError("boto3 is not installed; install ezcompiler[r2]") from exc 

178 

179 return boto3.client( 

180 "s3", 

181 endpoint_url=self._endpoint, 

182 aws_access_key_id=self._require_env("R2_ACCESS_KEY_ID"), 

183 aws_secret_access_key=self._require_env("R2_SECRET_ACCESS_KEY"), 

184 region_name="auto", 

185 ) 

186 

187 @staticmethod 

188 def _require_env(name: str) -> str: 

189 """Return the value of env var ``name`` or raise. 

190 

191 Raises: 

192 UploadError: If the variable is missing or empty. 

193 """ 

194 value = os.environ.get(name) 

195 if not value: 

196 raise UploadError(f"Missing required env var: {name}") 

197 return value 

198 

199 # //////////////////////////////////////////////// 

200 # VALIDATION METHODS 

201 # //////////////////////////////////////////////// 

202 

203 def _validate_config(self) -> None: 

204 """Validate R2 uploader configuration. 

205 

206 Raises: 

207 UploadError: If the required ``bucket`` key is missing. 

208 """ 

209 if not self._config.get("bucket"): 209 ↛ 210line 209 didn't jump to line 210 because the condition on line 209 was never true

210 raise UploadError("Missing required configuration key: bucket")