Coverage for src/updates2mqtt/integrations/docker_enrich.py: 83%
522 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-09 06:17 +0000
1import re
2import typing
3from abc import abstractmethod
4from typing import Any, cast
6import structlog
7from docker.auth import resolve_repository_name
8from docker.models.containers import Container
9from omegaconf import MissingMandatoryValue, OmegaConf, ValidationError
11from updates2mqtt.helpers import (
12 APIStatsCounter,
13 CacheMetadata,
14 ThrottledError,
15 Throttler,
16 fetch_url,
17 httpx_json_content,
18 validate_url,
19)
20from updates2mqtt.model import DiscoveryArtefactDetail, DiscoveryInstallationDetail, ReleaseDetail
22if typing.TYPE_CHECKING:
23 from docker.models.images import RegistryData
24 from httpx import Response
25 from omegaconf.dictconfig import DictConfig
26from http import HTTPStatus
28import docker
29import docker.errors
31from updates2mqtt.config import (
32 PKG_INFO_FILE,
33 SOURCE_PLATFORM_GITHUB,
34 SOURCE_PLATFORM_GITLAB,
35 SOURCE_PLATFORMS,
36 CommonPackages,
37 DockerConfig,
38 DockerPackageUpdateInfo,
39 MetadataSourceConfig,
40 PackageUpdateInfo,
41 RegistryConfig,
42 VersionPolicy,
43 docker_image_names,
44)
46log: Any = structlog.get_logger()
48DIFF_URL_TEMPLATES = {
49 SOURCE_PLATFORM_GITHUB: "{repo}/commit/{revision}",
50}
51RELEASE_URL_TEMPLATES = {
52 SOURCE_PLATFORM_GITHUB: "{repo}/releases/tag/{version}",
53}
54UNKNOWN_RELEASE_URL_TEMPLATES = {
55 SOURCE_PLATFORM_GITHUB: "{repo}/releases",
56 SOURCE_PLATFORM_GITLAB: "{repo}/container_registry",
57}
58MISSING_VAL = "**MISSING**"
59UNKNOWN_REGISTRY = "**UNKNOWN_REGISTRY**"
60UNKNOWN_NAME = "**UNKNOWN_NAME**"
62HEADER_DOCKER_DIGEST = "docker-content-digest"
63HEADER_DOCKER_API = "docker-distribution-api-version"
65TOKEN_URL_TEMPLATE = "https://{auth_host}/token?scope=repository:{image_name}:pull&service={service}" # nosec
67REGISTRY_GHCR = "ghcr.io"
68REGISTRY_DOCKER = "docker.io"
69REGISTRY_MCR = "mcr.microsoft.com"
70REGISTRY_QUAY = "quay.io"
71REGISTRY_LSCR = "lscr.io"
72REGISTRY_CODEBERG = "codeberg.org"
73REGISTRY_GITLAB = "registry.gitlab.com"
74REGISTRY_BITBUCKET = "crg.apkg.io"
77class RegistryInfo(typing.NamedTuple):
78 auth_host: str | None
79 api_host: str
80 service: str
81 url_template: str | None
82 repo_template: str | None
85REGISTRIES: dict[str, RegistryInfo] = {
86 REGISTRY_DOCKER: RegistryInfo("auth.docker.io", "registry-1.docker.io", "registry.docker.io", TOKEN_URL_TEMPLATE, None),
87 REGISTRY_MCR: RegistryInfo(None, "mcr.microsoft.com", "mcr.microsoft.com", None, None),
88 REGISTRY_QUAY: RegistryInfo(None, "quay.io", "quay.io", TOKEN_URL_TEMPLATE, None),
89 REGISTRY_GHCR: RegistryInfo("ghcr.io", "ghcr.io", "ghcr.io", TOKEN_URL_TEMPLATE, "https://github.com/{image_name}"),
90 REGISTRY_LSCR: RegistryInfo("ghcr.io", "lscr.io", "ghcr.io", TOKEN_URL_TEMPLATE, None),
91 REGISTRY_CODEBERG: RegistryInfo(
92 "codeberg.org",
93 "codeberg.org",
94 "container_registry",
95 TOKEN_URL_TEMPLATE,
96 "https://codeberg.org/{image_name}",
97 ),
98 REGISTRY_BITBUCKET: RegistryInfo("crg.apkg.io", "crg.apkg.io", "crg.apkg.io", TOKEN_URL_TEMPLATE, repo_template=None),
99 REGISTRY_GITLAB: RegistryInfo(
100 "www.gitlab.com",
101 "registry.gitlab.com",
102 "container_registry",
103 "https://{auth_host}/jwt/auth?service={service}&scope=repository:{image_name}:pull&offline_token=true&client_id=docker",
104 "https://gitlab.com/{image_name}",
105 ),
106}
108# source: https://specs.opencontainers.org/distribution-spec/?v=v1.0.0#pull
109OCI_NAME_RE = r"[a-z0-9]+((\.|_|__|-+)[a-z0-9]+)*(\/[a-z0-9]+((\.|_|__|-+)[a-z0-9]+)*)*"
110OCI_TAG_RE = r"[a-zA-Z0-9_][a-zA-Z0-9._-]{0,127}"
113class DockerImageInfo(DiscoveryArtefactDetail):
114 """Normalize and shlep around the bits of an image def
116 index_name: aka index_name, e.g. ghcr.io
117 name: image ref without index name or tag, e.g. nginx, or librenms/librenms
118 tag: tag or digest
119 untagged_ref: combined index name and package name
120 """
122 def __init__(
123 self,
124 ref: str, # ref with optional index name and tag or digest, index:name:tag_or_digest
125 image_digest: str | None = None,
126 tags: list[str] | None = None,
127 attributes: dict[str, Any] | None = None,
128 annotations: dict[str, Any] | None = None,
129 platform: str | None = None, # test harness simplification
130 version: str | None = None, # test harness simplification
131 created: str | None = None,
132 ) -> None:
133 super().__init__()
134 self.ref: str = ref
135 self.version: str | None = version
136 self.image_digest: str | None = image_digest
137 self.short_digest: str | None = None
138 self.repo_digest: str | None = None # the single RepoDigest known to match registry
139 self.git_digest: str | None = None
140 self.index_name: str | None = None
141 self.name: str | None = None
142 self.tag: str | None = None
143 self.pinned_digest: str | None = None
144 # untagged ref using combined index and remote name used only for pattern matching common pkg info
145 self.untagged_ref: str | None = None # index_name/remote_name used for pkg match
146 self.tag_or_digest: str | None = None # index_name/remote_name:**tag_or_digest**
147 self.tags = tags
148 self.attributes: dict[str, Any] = attributes or {}
149 self.annotations: dict[str, Any] = annotations or {}
150 self.throttled: bool = False
151 self.origin: str | None = None
152 self.error: str | None = None
153 self.platform: str | None = platform
154 self.custom: dict[str, str | float | int | bool | None] = {}
155 self.created: str | None = created
157 self.local_build: bool = not self.repo_digests
158 self.index_name, remote_name = resolve_repository_name(ref)
160 self.name = remote_name
162 if remote_name and ":" in remote_name and ("@" not in remote_name or remote_name.index("@") > remote_name.index(":")):
163 # name:tag format
164 self.name, self.tag_or_digest = remote_name.split(":", 1)
165 self.untagged_ref = ref.split(":", 1)[0]
166 self.tag = self.tag_or_digest
168 elif remote_name and "@" in remote_name:
169 # name@digest format
170 self.name, self.tag_or_digest = remote_name.split("@", 1)
171 self.untagged_ref = ref.split("@", 1)[0]
172 self.pinned_digest = self.tag_or_digest
174 if self.tag and "@" in self.tag:
175 # name:tag@digest format
176 # for pinned tags, care only about the digest part
177 self.tag, self.tag_or_digest = self.tag.split("@", 1)
178 self.pinned_digest = self.tag_or_digest
179 if self.tag_or_digest is None:
180 self.tag_or_digest = "latest"
181 self.untagged_ref = ref
182 self.tag = self.tag_or_digest
184 if self.repo_digest is None and len(self.repo_digests) == 1:
185 # definite known RepoDigest
186 # if its ambiguous, the final version selection will handle it
187 self.repo_digest = self.repo_digests[0]
189 if self.index_name == "docker.io" and "/" not in self.name:
190 # "official Docker images have an abbreviated library/foo name"
191 self.name = f"library/{self.name}"
192 if self.name is not None and not re.match(OCI_NAME_RE, self.name):
193 log.warning("Invalid OCI image name: %s", self.name)
194 if self.tag and not re.match(OCI_TAG_RE, self.tag):
195 log.warning("Invalid OCI image tag: %s", self.tag)
196 if "/" in self.name:
197 self.unqualified_name: str = self.name.split("/", 1)[1]
198 else:
199 self.unqualified_name = self.name
201 if self.os and self.arch:
202 self.platform = "/".join(
203 filter(
204 None,
205 [self.os, self.arch, self.variant],
206 ),
207 )
209 if self.image_digest is not None:
210 self.image_digest = self.condense_digest(self.image_digest, short=False)
211 self.short_digest = self.condense_digest(self.image_digest) # type: ignore[arg-type]
213 @property
214 def repo_digests(self) -> list[str]:
215 if self.repo_digest:
216 return [self.repo_digest]
217 # RepoDigest in image inspect, Registry Config object
218 digests = [v.split("@", 1)[1] if "@" in v else v for v in self.attributes.get("RepoDigests", [])]
219 return digests or []
221 @property
222 def pinned(self) -> bool:
223 """Check if this is pinned and installed version consistent with pin"""
224 return bool(self.pinned_digest and self.pinned_digest in self.repo_digests)
226 @property
227 def os(self) -> str | None:
228 return self.attributes.get("Os")
230 @property
231 def arch(self) -> str | None:
232 return self.attributes.get("Architecture")
234 @property
235 def variant(self) -> str | None:
236 return self.attributes.get("Variant")
238 def condense_digest(self, digest: str, short: bool = True) -> str | None:
239 try:
240 digest = digest.split("@")[1] if "@" in digest else digest # fully qualified RepoDigest
241 if short:
242 digest = digest.split(":")[1] if ":" in digest else digest # remove digest type prefix
243 return digest[0:12]
244 return digest
245 except Exception as e:
246 log.warning("Unable to condense digest %s: %s", digest, e)
247 return None
249 def reuse(self) -> "DockerImageInfo":
250 cloned = DockerImageInfo(
251 self.ref, self.image_digest, self.tags, self.attributes, self.annotations, self.version, self.created
252 )
253 cloned.origin = "REUSED"
254 return cloned
256 def as_dict(self, minimal: bool = True) -> dict[str, str | list | dict | bool | int | None]:
257 result: dict[str, str | list | dict | bool | int | None] = {
258 "captured": self.captured.isoformat(),
259 "image_ref": self.ref,
260 "name": self.name,
261 "version": self.version,
262 "image_digest": self.image_digest,
263 "repo_digest": self.repo_digest,
264 "repo_digests": self.repo_digest,
265 "git_digest": self.git_digest,
266 "index_name": self.index_name,
267 "tag": self.tag,
268 "pinned_digest": self.pinned_digest,
269 "tag_or_digest": self.tag_or_digest,
270 "tags": self.tags,
271 "origin": self.origin,
272 "platform": self.platform,
273 "local_build": self.local_build,
274 "error": self.error,
275 "throttled": self.throttled,
276 "custom": self.custom,
277 }
278 if not minimal: 278 ↛ 281line 278 didn't jump to line 281 because the condition on line 278 was always true
279 result["attributes"] = self.attributes
280 result["annotations"] = self.annotations
281 return result
284def id_source_platform(source: str | None) -> str | None:
285 candidates: list[str] = [platform for platform, pattern in SOURCE_PLATFORMS.items() if re.match(pattern, source or "")]
286 return candidates[0] if candidates else None
289def _select_annotation(
290 name: str, key: str, local_info: DockerImageInfo | None = None, registry_info: DockerImageInfo | None = None
291) -> dict[str, str | None]:
292 result: dict[str, str | None] = {}
293 if registry_info:
294 v: Any | None = registry_info.annotations.get(key)
295 if v is not None:
296 result[name] = v
297 elif local_info: 297 ↛ 301line 297 didn't jump to line 301 because the condition on line 297 was always true
298 v = local_info.annotations.get(key)
299 if v is not None: 299 ↛ 300line 299 didn't jump to line 300 because the condition on line 299 was never true
300 result[name] = v
301 return result
304def cherrypick_annotations(
305 local_info: DockerImageInfo, registry_info: DockerImageInfo | None
306) -> dict[str, str | float | int | bool | None]:
307 """https://github.com/opencontainers/image-spec/blob/main/annotations.md"""
308 results: dict[str, str | float | int | bool | None] = {}
309 for either_name, either_label in [
310 ("documentation_url", "org.opencontainers.image.documentation"),
311 ("description", "org.opencontainers.image.description"),
312 ("licences", "org.opencontainers.image.licenses"),
313 ("image_base", "org.opencontainers.image.base.name"),
314 ("image_created", "org.opencontainers.image.created"),
315 ("image_version", "org.opencontainers.image.version"),
316 ("image_revision", "org.opencontainers.image.revision"),
317 ("ref_name", "org.opencontainers.image.ref.name"),
318 ("title", "org.opencontainers.image.title"),
319 ("vendor", "org.opencontainers.image.vendor"),
320 ("source", "org.opencontainers.image.source"),
321 ]:
322 results.update(_select_annotation(either_name, either_label, local_info, registry_info))
323 if ( 323 ↛ 329line 323 didn't jump to line 329 because the condition on line 323 was never true
324 results.get("ref_name") == "ubuntu"
325 and local_info.name != "ubuntu"
326 and results.get("image_version")
327 and re.fullmatch(r"^2\d\.\d\d$", str(results["image_version"]))
328 ):
329 log.debug(
330 "Suppressing %s base %s version leaking into image version: %s",
331 local_info.name,
332 results["ref_name"],
333 results["image_version"],
334 )
335 del results["image_version"]
336 return results
339class DockerServiceDetails(DiscoveryInstallationDetail):
340 def __init__(
341 self,
342 container_name: str | None = None,
343 compose_path: str | None = None,
344 compose_version: str | None = None,
345 compose_service: str | None = None,
346 git_repo_path: str | None = None,
347 ) -> None:
348 self.container_name: str | None = container_name
349 self.compose_path: str | None = compose_path
350 self.compose_version: str | None = compose_version
351 self.compose_service: str | None = compose_service
352 self.git_repo_path: str | None = git_repo_path
353 self.git_local_timestamp: str | None = None
355 def as_dict(self) -> dict[str, str | list | dict | bool | int | None]:
356 results: dict[str, str | list | dict | bool | int | None] = {
357 "container_name": self.container_name,
358 "compose_path": self.compose_path,
359 "compose_service": self.compose_service,
360 "compose_version": self.compose_version,
361 }
362 if self.git_local_timestamp:
363 results["git_local_timestamp"] = self.git_local_timestamp
364 if self.git_repo_path:
365 results["git_repo_path"] = self.git_repo_path
366 return results
369class LocalContainerInfo:
370 def build_image_info(self, container: Container) -> tuple[DockerImageInfo, DockerServiceDetails]:
371 """Image contents equiv to `docker inspect image <image_ref>`"""
372 # container image can be none if someone ran `docker rmi -f`
373 # so although this could be sourced from image, like `container.image.tags[0]`
374 # use the container ref instead, which survives monkeying about with images
375 image_ref: str = container.attrs.get("Config", {}).get("Image") or ""
376 image_digest = container.attrs.get("Image")
378 image_info: DockerImageInfo = DockerImageInfo(
379 image_ref,
380 image_digest=image_digest,
381 tags=container.image.tags if container and container.image else None,
382 annotations=container.image.labels if container.image else None,
383 attributes=container.image.attrs if container.image else None,
384 )
385 service_info: DockerServiceDetails = DockerServiceDetails(
386 container.name,
387 compose_path=container.labels.get("com.docker.compose.project.working_dir"),
388 compose_service=container.labels.get("com.docker.compose.service"),
389 compose_version=container.labels.get("com.docker.compose.version"),
390 )
392 labels: dict[str, str | float | int | bool | None] = cherrypick_annotations(image_info, None)
393 # capture container labels/annotations, not image ones
394 labels = labels or {}
395 if container.image and container.image.attrs: 395 ↛ 397line 395 didn't jump to line 397 because the condition on line 395 was always true
396 image_info.created = container.image.attrs.get("Created")
397 image_info.custom = labels
398 image_info.version = cast("str|None", labels.get("image_version"))
400 return image_info, service_info
403class PackageEnricher:
404 def __init__(self, docker_cfg: DockerConfig, packages: dict[str, PackageUpdateInfo] | None = None) -> None:
405 self.pkgs: dict[str, PackageUpdateInfo] = packages or {}
406 self.cfg: DockerConfig = docker_cfg
407 self.log: Any = structlog.get_logger().bind(integration="docker")
409 def initialize(self) -> None:
410 pass
412 def enrich(self, image_info: DockerImageInfo) -> PackageUpdateInfo | None:
413 def match(pkg: PackageUpdateInfo) -> bool:
414 if pkg is not None and pkg.docker is not None and pkg.docker.image_name is not None: 414 ↛ 420line 414 didn't jump to line 420 because the condition on line 414 was always true
415 image_names = docker_image_names(pkg.docker)
416 if image_info.untagged_ref is not None and image_info.untagged_ref in image_names:
417 return True
418 if image_info.ref is not None and image_info.ref in image_names:
419 return True
420 return False
422 if image_info.untagged_ref is not None and image_info.ref is not None: 422 ↛ 432line 422 didn't jump to line 432 because the condition on line 422 was always true
423 for pkg in self.pkgs.values():
424 if match(pkg):
425 self.log.debug(
426 "Found common package",
427 image_name=pkg.docker.image_name, # type: ignore [union-attr]
428 logo_url=pkg.logo_url,
429 relnotes_url=pkg.release_notes_url,
430 )
431 return pkg
432 return None
435class DefaultPackageEnricher(PackageEnricher):
436 def enrich(self, image_info: DockerImageInfo) -> PackageUpdateInfo | None:
437 self.log.debug("Default pkg info", image_name=image_info.untagged_ref, image_ref=image_info.ref)
438 return PackageUpdateInfo(
439 DockerPackageUpdateInfo(image_info.untagged_ref or image_info.ref, version_policy=VersionPolicy.AUTO),
440 logo_url=self.cfg.default_entity_picture_url,
441 release_notes_url=None,
442 )
445class CommonPackageEnricher(PackageEnricher):
446 def initialize(self) -> None:
447 base_cfg: DictConfig = OmegaConf.structured(CommonPackages)
448 if PKG_INFO_FILE.exists(): 448 ↛ 455line 448 didn't jump to line 455 because the condition on line 448 was always true
449 self.log.debug("Loading common package update info", path=PKG_INFO_FILE)
450 cfg: DictConfig = typing.cast("DictConfig", OmegaConf.merge(base_cfg, OmegaConf.load(PKG_INFO_FILE)))
452 OmegaConf.to_container(cfg, throw_on_missing=True)
453 OmegaConf.set_readonly(cfg, True)
454 else:
455 self.log.warning("No common package update info found", path=PKG_INFO_FILE)
456 cfg = base_cfg
457 try:
458 common_config: CommonPackages = typing.cast("CommonPackages", cfg)
459 # omegaconf broken-ness on optional fields and converting to backclasses
460 self.pkgs = common_config.common_packages
461 # self.pkgs: dict[str, PackageUpdateInfo] = {
462 # pkg: PackageUpdateInfo(**pkg_cfg) for pkg, pkg_cfg in cfg.common_packages.items() if pkg not in self.pkgs
463 # }
464 except (MissingMandatoryValue, ValidationError) as e:
465 self.log.serror("Configuration error %s", e, path=PKG_INFO_FILE.as_posix())
466 raise
469class LinuxServerIOPackageEnricher(PackageEnricher):
470 def initialize(self) -> None:
471 cfg: MetadataSourceConfig | None = self.cfg.discover_metadata.get("linuxserver.io")
472 if cfg is None or not cfg.enabled:
473 return
475 self.log.debug(f"Fetching linuxserver.io metadata from API, cache_ttl={cfg.cache_ttl}")
476 response: Response | None = fetch_url(
477 "https://api.linuxserver.io/api/v1/images?include_config=false&include_deprecated=false",
478 cache_ttl=cfg.cache_ttl,
479 )
480 if response and response.is_success:
481 api_data: Any = response.json()
482 repos: list = api_data.get("data", {}).get("repositories", {}).get("linuxserver", [])
483 else:
484 return
486 added = 0
487 for repo in repos:
488 image_name = repo.get("name")
489 if image_name and image_name not in self.pkgs: 489 ↛ 487line 489 didn't jump to line 487 because the condition on line 489 was always true
490 github_url: str | None = repo.get("github_url")
491 self.pkgs[image_name] = PackageUpdateInfo(
492 DockerPackageUpdateInfo(f"lscr.io/linuxserver/{image_name}"),
493 logo_url=repo.get("project_logo"),
494 release_notes_url=f"{github_url}/releases" if github_url else None,
495 )
496 added += 1
497 self.log.debug(f"Added {added} linuxserver.io package details")
500class SourceReleaseEnricher:
501 def __init__(self) -> None:
502 self.log: Any = structlog.get_logger().bind(integration="docker")
504 def enrich(
505 self, registry_info: DockerImageInfo, source_repo_url: str | None = None, notes_url: str | None = None
506 ) -> ReleaseDetail | None:
507 detail = ReleaseDetail(registry_info.name or UNKNOWN_NAME)
509 detail.notes_url = notes_url
510 detail.version = registry_info.annotations.get("org.opencontainers.image.version")
511 detail.revision = registry_info.annotations.get("org.opencontainers.image.revision")
512 # explicit source_repo_url overrides container, e.g. where container source is only the docker wrapper
513 detail.source_url = source_repo_url or registry_info.annotations.get("org.opencontainers.image.source")
515 if detail.source_url is None and registry_info is not None and registry_info.index_name is not None:
516 registry_config: RegistryInfo | None = REGISTRIES.get(registry_info.index_name)
517 repo_template: str | None = registry_config.repo_template if registry_config else None
518 if repo_template:
519 source_url = repo_template.format(image_name=registry_info.name)
520 if validate_url(source_url, cache_ttl=86400): 520 ↛ 524line 520 didn't jump to line 524 because the condition on line 520 was always true
521 detail.source_url = source_url
522 self.log.info("Implied source from registry: %s", detail.source_url)
524 if detail.source_url is None and detail.notes_url is None and detail.revision is None and detail.version is None:
525 return None
527 if detail.source_url and "#" in detail.source_url:
528 detail.source_repo_url = detail.source_url.split("#", 1)[0]
529 self.log.debug("Simplifying %s from %s", detail.source_repo_url, detail.source_url)
530 else:
531 detail.source_repo_url = detail.source_url
533 detail.source_platform = id_source_platform(detail.source_repo_url)
534 if not detail.source_platform:
535 self.log.debug("No known source platform found on container", source=detail.source_repo_url)
536 return detail
538 template_vars: dict[str, str | None] = {
539 "version": detail.version or MISSING_VAL,
540 "revision": detail.revision or MISSING_VAL,
541 "repo": detail.source_repo_url or MISSING_VAL,
542 "source": detail.source_url or MISSING_VAL,
543 }
545 try:
546 diff_url_template: str | None = DIFF_URL_TEMPLATES.get(detail.source_platform)
547 diff_url: str | None = diff_url_template.format(**template_vars) if diff_url_template else None
548 if diff_url and MISSING_VAL not in diff_url and validate_url(diff_url, cache_ttl=3600): 548 ↛ 549line 548 didn't jump to line 549 because the condition on line 548 was never true
549 detail.diff_url = diff_url
550 else:
551 diff_url = None
553 if detail.notes_url is None and detail.source_platform in RELEASE_URL_TEMPLATES:
554 platform_notes_url: str | None = RELEASE_URL_TEMPLATES[detail.source_platform].format(**template_vars)
555 if ( 555 ↛ 560line 555 didn't jump to line 560 because the condition on line 555 was never true
556 platform_notes_url
557 and MISSING_VAL not in platform_notes_url
558 and validate_url(platform_notes_url, cache_ttl=86400)
559 ):
560 self.log.debug("Setting default known release notes url: %s", platform_notes_url)
561 detail.notes_url = platform_notes_url
563 if detail.notes_url is None and detail.source_platform in UNKNOWN_RELEASE_URL_TEMPLATES:
564 platform_notes_url = UNKNOWN_RELEASE_URL_TEMPLATES[detail.source_platform].format(**template_vars)
565 if (
566 platform_notes_url
567 and MISSING_VAL not in platform_notes_url
568 and validate_url(platform_notes_url, cache_ttl=86400)
569 ):
570 self.log.debug("Setting default unknown release notes url: %s", platform_notes_url)
571 detail.notes_url = platform_notes_url
572 except Exception as e:
573 self.log.error("Failed formatting enriched URLs with %s: %s", template_vars, e)
575 return detail
578class AuthError(Exception):
579 pass
582class VersionLookup:
583 def __init__(self) -> None:
584 self.log: Any = structlog.get_logger().bind(integration="docker", tool="version_lookup")
586 @abstractmethod
587 def lookup(self, local_image_info: DockerImageInfo, **kwargs) -> DockerImageInfo:
588 pass
591class ContainerDistributionAPIVersionLookup(VersionLookup):
592 def __init__(self, throttler: Throttler, cfg: RegistryConfig) -> None:
593 self.throttler: Throttler = throttler
594 self.cfg: RegistryConfig = cfg
595 self.log: Any = structlog.get_logger().bind(integration="docker", tool="version_lookup")
596 self.api_stats = APIStatsCounter()
598 def fetch_token(self, registry: str, image_name: str) -> str | None:
599 default_info = RegistryInfo(registry, registry, registry, TOKEN_URL_TEMPLATE, None)
600 registry_info_: RegistryInfo = REGISTRIES.get(registry, default_info)
601 auth_host: str | None = registry_info_.auth_host
602 if auth_host is None:
603 return None
605 service: str = registry_info_.service
606 url_template: str | None = registry_info_.url_template
607 auth_url: str | None = (
608 url_template.format(auth_host=auth_host, image_name=image_name, service=service) if url_template else None
609 )
610 if auth_url is None: 610 ↛ 611line 610 didn't jump to line 611 because the condition on line 610 was never true
611 return None
612 response: Response | None = fetch_url(
613 auth_url, cache_ttl=self.cfg.token_cache_ttl, follow_redirects=True, api_stats_counter=self.api_stats
614 )
616 if response and response.is_success:
617 api_data = httpx_json_content(response, {})
618 token: str | None = api_data.get("token") if api_data else None
619 if token: 619 ↛ 621line 619 didn't jump to line 621 because the condition on line 619 was always true
620 return token
621 self.log.warning("No token found in response for %s", auth_url)
622 raise AuthError(f"No token found in response for {image_name}")
624 self.log.debug(
625 "Non-success response at %s fetching token: %s",
626 auth_url,
627 (response and response.status_code) or None,
628 )
629 if response and response.status_code == 404:
630 self.log.debug(
631 "Default token URL %s not found, calling /v2 endpoint to validate OCI API and provoke auth", auth_url
632 )
633 response = fetch_url(
634 f"https://{auth_host}/v2",
635 follow_redirects=True,
636 allow_stale=False,
637 cache_ttl=0,
638 api_stats_counter=self.api_stats,
639 )
641 if response and response.status_code == 401: 641 ↛ 662line 641 didn't jump to line 662 because the condition on line 641 was always true
642 auth = response.headers.get("www-authenticate")
643 if not auth:
644 self.log.warning("No www-authenticate header found in 401 response for %s", auth_url)
645 raise AuthError(f"No www-authenticate header found on 401 for {image_name}")
646 match = re.search(r'realm="([^"]+)",service="([^"]+)",scope="([^"]+)"', auth)
647 if not match: 647 ↛ 648line 647 didn't jump to line 648 because the condition on line 647 was never true
648 self.log.warning("No realm/service/scope found in www-authenticate header for %s", auth_url)
649 raise AuthError(f"No realm/service/scope found on 401 headers for {image_name}")
651 realm, service, scope = match.groups()
652 auth_url = f"{realm}?service={service}&scope={scope}"
653 response = fetch_url(auth_url, follow_redirects=True, api_stats_counter=self.api_stats)
655 if response and response.is_success: 655 ↛ 659line 655 didn't jump to line 659 because the condition on line 655 was always true
656 token_data = response.json()
657 self.log.debug("Fetched registry token from %s", auth_url)
658 return token_data.get("token")
659 self.log.warning(
660 "Alternative auth %s with status %s has no token", auth_url, (response and response.status_code) or None
661 )
662 elif response:
663 self.log.warning("Auth %s failed with status %s", auth_url, (response and response.status_code) or None)
665 raise AuthError(f"Failed to fetch token for {image_name} at {auth_url}")
667 def fetch_index(
668 self, api_host: str, local_image_info: DockerImageInfo, token: str | None
669 ) -> tuple[Any | None, str | None, CacheMetadata | None]:
670 if local_image_info.tag: 670 ↛ 674line 670 didn't jump to line 674 because the condition on line 670 was always true
671 api_url: str = f"https://{api_host}/v2/{local_image_info.name}/manifests/{local_image_info.tag}"
672 cache_ttl: int | None = self.cfg.mutable_cache_ttl
673 else:
674 api_url = f"https://{api_host}/v2/{local_image_info.name}/manifests/{local_image_info.pinned_digest}"
675 cache_ttl = self.cfg.immutable_cache_ttl
677 response: Response | None = fetch_url(
678 api_url,
679 cache_ttl=cache_ttl,
680 bearer_token=token,
681 response_type=[
682 "application/vnd.oci.image.index.v1+json",
683 "application/vnd.docker.distribution.manifest.list.v2+json",
684 ],
685 api_stats_counter=self.api_stats,
686 )
688 if response is None: 688 ↛ 689line 688 didn't jump to line 689 because the condition on line 688 was never true
689 self.log.warning("Empty response for manifest for image at %s", api_url)
690 elif response.status_code == 429: 690 ↛ 691line 690 didn't jump to line 691 because the condition on line 690 was never true
691 self.throttler.throttle(local_image_info.index_name, raise_exception=True)
692 elif not response.is_success:
693 api_data = httpx_json_content(response, {})
694 self.log.warning(
695 "Failed to fetch index from %s: %s",
696 api_url,
697 api_data.get("errors") if api_data else response.text,
698 )
699 else:
700 index = response.json()
701 self.log.debug(
702 "INDEX %s manifests, %s annotations, api: %s, header digest: %s",
703 len(index.get("manifests", [])),
704 len(index.get("annotations", [])),
705 response.headers.get(HEADER_DOCKER_API, "N/A"),
706 response.headers.get(HEADER_DOCKER_DIGEST, "N/A"),
707 )
708 return index, response.headers.get(HEADER_DOCKER_DIGEST), CacheMetadata(response)
709 return None, None, None
711 def fetch_object(
712 self,
713 api_host: str,
714 local_image_info: DockerImageInfo,
715 media_type: str,
716 digest: str,
717 token: str | None,
718 follow_redirects: bool = False,
719 api_type: str = "manifests",
720 ) -> tuple[Any | None, CacheMetadata | None]:
721 api_url = f"https://{api_host}/v2/{local_image_info.name}/{api_type}/{digest}"
722 response = fetch_url(
723 api_url,
724 cache_ttl=self.cfg.immutable_cache_ttl,
725 bearer_token=token,
726 response_type=media_type,
727 allow_stale=True,
728 follow_redirects=follow_redirects,
729 api_stats_counter=self.api_stats,
730 )
732 if response and response.is_success:
733 obj = httpx_json_content(response, None)
734 if obj: 734 ↛ 761line 734 didn't jump to line 761 because the condition on line 734 was always true
735 self.log.debug(
736 "%s, header digest:%s, api: %s, %s annotations",
737 api_type.upper(),
738 response.headers.get(HEADER_DOCKER_DIGEST, "N/A"),
739 response.headers.get(HEADER_DOCKER_API, "N/A"),
740 len(obj.get("annotations", [])),
741 )
742 return obj, CacheMetadata(response)
743 elif response and response.status_code == 429: 743 ↛ 744line 743 didn't jump to line 744 because the condition on line 743 was never true
744 self.throttler.throttle(local_image_info.index_name, raise_exception=True)
745 elif response and not response.is_success: 745 ↛ 760line 745 didn't jump to line 760 because the condition on line 745 was always true
746 api_data = httpx_json_content(response, {})
747 if response: 747 ↛ 755line 747 didn't jump to line 755 because the condition on line 747 was always true
748 self.log.warning(
749 "Failed to fetch obj from %s: %s %s",
750 api_url,
751 response.status_code,
752 api_data.get("errors") if api_data else response.text,
753 )
754 else:
755 self.log.warning(
756 "Failed to fetch obj from %s: No Response, %s", api_url, api_data.get("errors") if api_data else None
757 )
759 else:
760 self.log.error("Empty response from %s", api_url)
761 return None, None
763 def lookup(
764 self,
765 local_image_info: DockerImageInfo,
766 token: str | None = None,
767 minimal: bool = False,
768 **kwargs,
769 ) -> DockerImageInfo:
770 result: DockerImageInfo = DockerImageInfo(local_image_info.ref)
771 if not local_image_info.name or not local_image_info.index_name: 771 ↛ 772line 771 didn't jump to line 772 because the condition on line 771 was never true
772 self.log.debug("No local pkg name or registry index name to check")
773 return result
775 if self.throttler.check_throttle(local_image_info.index_name): 775 ↛ 776line 775 didn't jump to line 776 because the condition on line 775 was never true
776 result.throttled = True
777 return result
779 if token: 779 ↛ 780line 779 didn't jump to line 780 because the condition on line 779 was never true
780 self.log.debug("Using provided token to fetch manifest for image %s", local_image_info.ref)
781 else:
782 try:
783 token = self.fetch_token(local_image_info.index_name, local_image_info.name)
784 except AuthError as e:
785 self.log.warning("Authentication error prevented Docker Registry enrichment: %s", e)
786 result.error = str(e)
787 return result
789 index: Any | None = None
790 index_digest: str | None = None # fetched from header, should be the image digest
791 index_cache_metadata: CacheMetadata | None = None
792 manifest_cache_metadata: CacheMetadata | None = None
793 config_cache_metadata: CacheMetadata | None = None
794 idx = local_image_info.index_name
795 api_host: str | None = REGISTRIES.get(
796 idx,
797 RegistryInfo(idx, idx, idx, TOKEN_URL_TEMPLATE, None),
798 ).api_host
799 if api_host is None: 799 ↛ 800line 799 didn't jump to line 800 because the condition on line 799 was never true
800 self.log("No API host can be determined for %s", local_image_info.index_name)
801 return result
802 try:
803 index, index_digest, index_cache_metadata = self.fetch_index(api_host, local_image_info, token)
804 except ThrottledError:
805 result.throttled = True
806 index = None
808 if index:
809 result.annotations = index.get("annotations", {})
810 for m in index.get("manifests", []):
811 try:
812 platform_info = m.get("platform", {})
813 except Exception as e:
814 self.log.warning("Failed analyzing manifest data: %s: %s", m, e)
815 continue
816 if (
817 platform_info.get("os") == local_image_info.os
818 and platform_info.get("architecture") == local_image_info.arch
819 and ("Variant" not in platform_info or platform_info.get("Variant") == local_image_info.variant)
820 ):
821 if index_digest:
822 result.image_digest = index_digest
823 result.short_digest = result.condense_digest(index_digest)
824 self.log.debug("Setting %s image digest %s", result.name, result.short_digest)
826 digest: str | None = m.get("digest")
827 media_type = m.get("mediaType")
828 manifest: Any | None = None
830 if digest: 830 ↛ 838line 830 didn't jump to line 838 because the condition on line 830 was always true
831 try:
832 manifest, manifest_cache_metadata = self.fetch_object(
833 api_host, local_image_info, media_type, digest, token
834 )
835 except ThrottledError:
836 result.throttled = True
838 if manifest:
839 manifest_config: dict[str, Any] = manifest.get("config", {})
840 digest = manifest_config.get("digest")
841 if digest is None: 841 ↛ 842line 841 didn't jump to line 842 because the condition on line 841 was never true
842 self.log.warning("Empty digest for %s %s %s", api_host, digest, media_type)
843 else:
844 result.repo_digest = result.condense_digest(digest, short=False)
845 self.log.debug("Setting %s repo digest: %s", result.name, result.repo_digest)
847 if manifest.get("annotations"): 847 ↛ 850line 847 didn't jump to line 850 because the condition on line 847 was always true
848 result.annotations.update(manifest.get("annotations", {}))
849 else:
850 self.log.debug("No annotations found in manifest: %s", manifest)
852 if (
853 not minimal
854 and manifest_config
855 and manifest_config.get("mediaType")
856 and manifest_config.get("digest")
857 ):
858 try:
859 img_config, config_cache_metadata = self.fetch_object(
860 api_host=api_host,
861 local_image_info=local_image_info,
862 media_type=manifest_config["mediaType"],
863 digest=manifest_config["digest"],
864 token=token,
865 follow_redirects=True,
866 api_type="blobs",
867 )
868 if img_config: 868 ↛ 878line 868 didn't jump to line 878 because the condition on line 868 was always true
869 config = img_config.get("config") or img_config.get("Config")
870 try:
871 if config and "Labels" in config: 871 ↛ 873line 871 didn't jump to line 873 because the condition on line 871 was always true
872 result.annotations.update(config.get("Labels") or {})
873 result.annotations.update(img_config.get("annotations") or {})
874 except Exception as e:
875 self.log.warning("Failure handling labels/annotations %s: %s", config, e)
876 result.created = config.get("created") or config.get("Created")
877 else:
878 self.log.debug("No config found: %s", manifest)
879 except Exception as e:
880 self.log.warning("Failed to extract %s image info from config: %s", local_image_info.ref, e)
882 if not result.annotations:
883 self.log.debug("No annotations found from registry data")
885 labels: dict[str, str | float | int | bool | None] = cherrypick_annotations(local_image_info, result)
886 result.custom = labels or {}
887 if index_cache_metadata:
888 result.custom["index_cache_age"] = index_cache_metadata.age
889 if manifest_cache_metadata:
890 result.custom["manifest_cache_age"] = manifest_cache_metadata.age
891 if config_cache_metadata:
892 result.custom["config_cache_age"] = config_cache_metadata.age
893 result.version = cast("str|None", labels.get("image_version"))
894 result.origin = "OCI_V2" if not minimal else "OCI_V2_MINIMAL"
896 self.log.debug(
897 "OCI_V2 Lookup for %s: short_digest:%s, repo_digest:%s, version: %s",
898 local_image_info.name,
899 result.short_digest,
900 result.repo_digest,
901 result.version,
902 )
903 return result
906class DockerClientVersionLookup(VersionLookup):
907 """Query remote registry via local Docker API
909 No auth needed, however uses the old v1 APIs, and only Index available via API
910 """
912 def __init__(self, client: docker.DockerClient, throttler: Throttler, cfg: RegistryConfig, api_backoff: int = 30) -> None:
913 self.client: docker.DockerClient = client
914 self.throttler: Throttler = throttler
915 self.cfg: RegistryConfig = cfg
916 self.api_backoff: int = api_backoff
917 self.log: Any = structlog.get_logger().bind(integration="docker", tool="version_lookup")
919 def lookup(self, local_image_info: DockerImageInfo, retries: int = 3, **kwargs) -> DockerImageInfo:
920 retries_left = retries
921 retry_secs: int = self.api_backoff
922 reg_data: RegistryData | None = None
924 result = DockerImageInfo(local_image_info.ref)
925 if local_image_info.index_name is None or local_image_info.ref is None: 925 ↛ 926line 925 didn't jump to line 926 because the condition on line 925 was never true
926 return result
928 while reg_data is None and retries_left > 0:
929 if self.throttler.check_throttle(local_image_info.index_name):
930 result.throttled = True
931 break
932 try:
933 self.log.debug("Fetching registry data", image_ref=local_image_info.ref)
934 reg_data = self.client.images.get_registry_data(local_image_info.ref)
935 self.log.debug(
936 "Registry Data: id:%s,image:%s, attrs:%s",
937 reg_data.id,
938 reg_data.image_name,
939 reg_data.attrs,
940 )
941 if reg_data: 941 ↛ 928line 941 didn't jump to line 928 because the condition on line 941 was always true
942 result.short_digest = result.condense_digest(reg_data.short_id)
943 result.image_digest = result.condense_digest(reg_data.id, short=False)
944 # result.name = reg_data.image_name
945 result.attributes = reg_data.attrs
946 result.annotations = reg_data.attrs.get("Config", {}).get("Labels") or {}
947 result.error = None
949 except docker.errors.APIError as e:
950 if e.status_code == HTTPStatus.TOO_MANY_REQUESTS: 950 ↛ 959line 950 didn't jump to line 959 because the condition on line 950 was always true
951 retry_secs = round(retry_secs**1.5)
952 try:
953 retry_secs = int(e.response.headers.get("Retry-After", -1)) # type: ignore[union-attr]
954 except Exception as e2:
955 self.log.debug("Failed to access headers for retry info: %s", e2)
956 self.throttler.throttle(local_image_info.index_name, retry_secs, e.explanation)
957 result.throttled = True
958 return result
959 result.error = str(e)
960 retries_left -= 1
961 if retries_left == 0 or e.is_client_error():
962 self.log.warning("Failed to fetch registry data: [%s] %s", e.errno, e.explanation)
963 else:
964 self.log.debug("Failed to fetch registry data, retrying: %s", e)
966 labels: dict[str, str | float | int | bool | None] = cherrypick_annotations(local_image_info, result)
967 result.custom = labels or {}
968 result.version = cast("str|None", labels.get("image_version"))
969 result.created = cast("str|None", labels.get("image_created"))
970 result.origin = "DOCKER_CLIENT"
971 return result