Coverage for src/gitlabracadabra/containers/with

4# This program is free software: you can redistribute it and/or modify

5# it under the terms of the GNU Lesser General Public License as published by

6# the Free Software Foundation, either version 3 of the License, or

7# (at your option) any later version.

9# This program is distributed in the hope that it will be useful,

10# but WITHOUT ANY WARRANTY; without even the implied warranty of

11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

12# GNU Lesser General Public License for more details.

13#

14# You should have received a copy of the GNU Lesser General Public License

15# along with this program. If not, see <http://www.gnu.org/licenses/>.

17from __future__ import annotations

19from hashlib import sha256

20from logging import getLogger

21from os.path import getsize, isfile

22from shutil import copy, copyfileobj

23from tempfile import NamedTemporaryFile

24from typing import IO, TYPE_CHECKING, BinaryIO

25from urllib.parse import quote

27from requests import HTTPError, Response, codes

29from gitlabracadabra.containers.const import DIGEST_HEADER, DOCKER_MANIFEST_SCHEMA1_SIGNED

30from gitlabracadabra.containers.scope import PULL, Scope

31from gitlabracadabra.disk_cache import cache_dir

33if TYPE_CHECKING:

34 from typing import Self

36 from gitlabracadabra.containers.registry_importer import RegistryImporter

39logger = getLogger(__name__)

42class WithDigest:

43 """An object with a digest."""

45 supported_mime_types: tuple[str, ...] | None = None

47 def __init__(

48 self,

49 registry: RegistryImporter,

50 manifest_name: str,

51 digest: str | None = None,

52 *,

53 size: int | None = None,

54 mime_type: str | None = None,

55 ) -> None:

56 """Initialize an object with a digest.

58 Args:

59 registry: Registry.

60 manifest_name: Manifest name (Example: library/debian).

61 digest: Digest (Example: sha256:5890f8ba95f680c87fcf89e51190098641b4f646102ce7ca906e7f83c84874dc).

62 size: Size (Example: 42).

63 mime_type: Content-Type / mediaType.

64 """

65 self._registry = registry

66 self._manifest_name = manifest_name

67 self._digest = digest

68 self._size = size

69 self._mime_type = mime_type

70 self._exists: bool | None = None

71 self._fd: BinaryIO | None = None

72 self._retrieve_mehod = "head"

73 self.forced_digest = False

75 def __eq__(self, other: object) -> bool:

76 """Compare.

78 Args:

79 other: Compare

81 Returns:

82 True if registry, manifest name, digest, size and mime_types are equal.

83 """

84 return (isinstance(self, type(other)) or isinstance(other, type(self))) and self.__dict__ == other.__dict__

86 def __hash__(self) -> int:

87 """Hash.

89 Returns:

90 hash.

91 """

92 return hash(self.__dict__)

94 @property

95 def registry(self) -> RegistryImporter:

96 """Get the registry.

98 Returns:

99 The registry.

100 """

101 return self._registry

102

103 @property

104 def manifest_name(self) -> str:

105 """Get the manifest name.

106

107 Returns:

108 The manifest name.

109 """

110 return self._manifest_name

111

112 @property

113 def digest(self) -> str:

114 """Get the digest.

115

116 Returns:

117 The digest.

118

119 Raises:

120 ValueError: Unable to get digest.

121 """

122 if self._digest is None:

123 self._retrieve()

124 if self._digest is None: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true

125 msg = "Unable to get digest"

126 raise ValueError(msg)

127 return self._digest

128

129 @property

130 def size(self) -> int:

131 """Get the size.

132

133 Returns:

134 The size.

135

136 Raises:

137 ValueError: Unable to get size.

138 """

139 if self._size is None:

140 try:

141 self._size = getsize(self.cache_path)

142 except FileNotFoundError:

143 self._retrieve()

144 if self._size is None: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true

145 msg = "Unable to get size"

146 raise ValueError(msg)

147 return self._size

148

149 @property

150 def mime_type(self) -> str | None:

151 """Get the MIME type (mediaType).

152

153 Returns:

154 The MIME type.

155 """

156 if self._mime_type is None:

157 self._retrieve()

158 return self._mime_type

159

160 @property

161 def cache_path(self) -> str:

162 """Get the cache path (local).

163

164 Returns:

165 Local path.

166 """

167 return str(cache_dir("containers_cache") / quote(self.digest, safe=""))

168

169 @property

170 def registry_path(self) -> str:

171 """Get the registry path.

172

173 Raises:

174 NotImplementedError: Needs to be implemented in subclasses.

175 """

176 raise NotImplementedError

177

178 def __enter__(self) -> Self:

179 """Open the cached file.

180

181 Returns:

182 self.

183

184 Raises:

185 RuntimeError: File already opened.

186 """

187 self._ensure_cached()

188 if self._fd is not None: 188 ↛ 189line 188 didn't jump to line 189 because the condition on line 188 was never true

189 msg = "File already opened"

190 raise RuntimeError(msg)

191 self._fd = open(self.cache_path, "rb")

192 return self

193

194 def __exit__(self, exc_type, exc_val, exc_tb) -> None: # type: ignore

195 """Close the cached file.

196

197 Args:

198 exc_type: Exception type.

199 exc_val: Exception value.

200 exc_tb: Exception traceback.

201 """

202 if self._fd is not None: 202 ↛ exitline 202 didn't return from function '__exit__' because the condition on line 202 was always true

203 self._fd.close()

204 self._fd = None

205

206 def read(self, n: int = -1) -> bytes:

207 """Read the cached file.

208

209 Args:

210 n: buffer size.

211

212 Returns:

213 Bytes.

214

215 Raises:

216 ValueError: File is not opened.

217 """

218 if self._fd is None: 218 ↛ 219line 218 didn't jump to line 219 because the condition on line 218 was never true

219 msg = "File is not opened"

220 raise ValueError(msg)

221 return self._fd.read(n)

222

223 def scope(self, actions: str = PULL) -> Scope:

224 """Get a scope.

225

226 Args:

227 actions: Scope action.

228

229 Returns:

230 A scope.

231 """

232 return Scope(self.manifest_name, actions)

233

234 def exists(self) -> bool:

235 """Get Blob/Manifest existence in the associated registry.

236

237 Returns:

238 True or False.

239

240 Raises:

241 HTTPError: Error when fetching existence.

242 """

243 if self._exists is None:

244 try:

245 self._retrieve()

246 self._exists = True

247 except HTTPError as err:

248 if (err.response is None) or (err.response.status_code != codes["not_found"]): 248 ↛ 249line 248 didn't jump to line 249 because the condition on line 248 was never true

249 raise

250 self._exists = False

251 if self._exists:

252 self.register()

253 return self._exists

254

255 def register(self) -> None:

256 """Notify the registry that the Digest exists."""

257 # Overridden in Blob

258

259 def _ensure_cached(self) -> None:

260 if self._digest is None or not isfile(self.cache_path):

261 self._retrieve(with_content=True)

262

263 def _retrieve(self, *, with_content: bool = False) -> None:

264 method = self._retrieve_mehod

265 if with_content:

266 method = "get"

267 with self._request(method) as response:

268 if self._digest is None:

269 self._digest = response.headers.get(DIGEST_HEADER)

270 elif DIGEST_HEADER in response.headers and self._digest != response.headers.get(DIGEST_HEADER): 270 ↛ 271line 270 didn't jump to line 271 because the condition on line 270 was never true

271 msg = f"Retrieved digest does not match {response.headers.get(DIGEST_HEADER)} != {self._digest}"

272 raise ValueError(msg)

273 if "Content-Type" in response.headers: 273 ↛ 275line 273 didn't jump to line 275 because the condition on line 273 was always true

274 self._mime_type = response.headers.get("Content-Type")

275 if "Content-Length" in response.headers: 275 ↛ 277line 275 didn't jump to line 277 because the condition on line 275 was always true

276 self._size = int(response.headers["Content-Length"])

277 if method != "head":

278 self._download_and_verify(response)

279

280 def _request(self, method: str) -> Response:

281 return self.registry.request(

282 method,

283 self.registry_path,

284 scopes={self.scope()},

285 accept=self.supported_mime_types,

286 stream=True,

287 )

288

289 def _download_and_verify(self, response: Response) -> None:

290 with NamedTemporaryFile(dir=cache_dir("containers_cache")) as fp:

291 copyfileobj(response.raw, fp)

292 downloaded_digest = self._compute_digest(fp)

293 if self._digest is None:

294 self._digest = downloaded_digest

295 else:

296 self._verify_digest(downloaded_digest)

297 copy(fp.name, self.cache_path)

298

299 def _verify_digest(self, digest: str) -> None:

300 if digest != self._digest:

301 if self._mime_type == DOCKER_MANIFEST_SCHEMA1_SIGNED: 301 ↛ 311line 301 didn't jump to line 311 because the condition on line 301 was always true

302 # https://docs.docker.com/registry/spec/api/#content-digests

303 # "manifest body without the signature content, also known as the JWS payload"

304 logger.info(

305 "Ignoring checksum mismatch for signed manifest %s: %s ! %s",

306 str(self),

307 digest,

308 self._digest,

309 )

310 else:

311 msg = f"Checksum mismatch: {digest} != {self._digest}"

312 raise ValueError(msg)

313

314 def _compute_digest(self, fp: IO[bytes]) -> str:

315 sha256_hash = sha256()

316 buf_len = 4096

317 fp.seek(0)

318 for byte_block in iter(lambda: fp.read(buf_len), b""):

319 sha256_hash.update(byte_block)

320 return f"sha256:{sha256_hash.hexdigest()}"

Coverage for src/gitlabracadabra/containers/with_digest.py: 86%

138 statements