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

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 .base_uploader import BaseUploader 

25 

26# /////////////////////////////////////////////////////////////// 

27# CLASSES 

28# /////////////////////////////////////////////////////////////// 

29 

30 

31class R2Uploader(BaseUploader): 

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

33 

34 Configuration keys: 

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

36 

37 Environment variables (required, write credentials): 

38 R2_ENDPOINT or R2_ACCOUNT_ID, R2_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY. 

39 """ 

40 

41 # //////////////////////////////////////////////// 

42 # INITIALIZATION 

43 # //////////////////////////////////////////////// 

44 

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

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

47 

48 Args: 

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

50 """ 

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

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

53 self._endpoint = self._resolve_endpoint() 

54 self._client = self._build_client() 

55 

56 # //////////////////////////////////////////////// 

57 # PUBLIC METHODS 

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

59 

60 def get_uploader_name(self) -> str: 

61 """Get the name of this uploader. 

62 

63 Returns: 

64 str: Name of the uploader. 

65 """ 

66 return "R2 Uploader" 

67 

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

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

70 

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

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

73 to keep the published tree TUF-consistent. 

74 

75 Args: 

76 source_path: Path to the source file or directory. 

77 destination: Object key prefix under the bucket. 

78 

79 Raises: 

80 UploadError: If any object upload fails. 

81 """ 

82 try: 

83 self._validate_source_path(source_path) 

84 prefix = destination.strip("/") 

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

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

87 return 

88 

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

90 # targets/zip d'abord, metadata en dernier (cohérence TUF) 

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

92 for file_path in files: 

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

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

95 except UploadError: 

96 raise 

97 except Exception as e: 

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

99 

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

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

102 

103 Args: 

104 remote_source: Object key prefix to fetch. 

105 local_dir: Local directory to populate. 

106 

107 Raises: 

108 UploadError: If a download fails. 

109 

110 Note: 

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

112 """ 

113 try: 

114 prefix = remote_source.strip("/") 

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

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

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

118 key = obj["Key"] 

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

120 dest = self._safe_join(local_dir, rel) 

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

122 self._client.download_file( 

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

124 ) 

125 except UploadError: 

126 raise 

127 except Exception as e: 

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

129 

130 # //////////////////////////////////////////////// 

131 # PRIVATE METHODS 

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

133 

134 @staticmethod 

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

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

137 

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

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

140 

141 Raises: 

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

143 """ 

144 root_resolved = root.resolve() 

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

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

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

148 return candidate 

149 

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

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

152 self._client.upload_file( 

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

154 ) 

155 

156 def _resolve_endpoint(self) -> str: 

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

158 

159 Raises: 

160 UploadError: If neither R2_ENDPOINT nor R2_ACCOUNT_ID is set. 

161 """ 

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

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

164 return endpoint 

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

166 if not account: 

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

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

169 

170 def _build_client(self) -> Any: 

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

172 try: 

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

174 except ImportError as exc: 

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

176 

177 return boto3.client( 

178 "s3", 

179 endpoint_url=self._endpoint, 

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

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

182 region_name="auto", 

183 ) 

184 

185 @staticmethod 

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

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

188 

189 Raises: 

190 UploadError: If the variable is missing or empty. 

191 """ 

192 value = os.environ.get(name) 

193 if not value: 

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

195 return value 

196 

197 # //////////////////////////////////////////////// 

198 # VALIDATION METHODS 

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

200 

201 def _validate_config(self) -> None: 

202 """Validate R2 uploader configuration. 

203 

204 Raises: 

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

206 """ 

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

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