Skip to content

refactor(updater): simplify digest verification flows #549

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
magrhino merged 2 commits into from
Jul 30, 2026
+334 −94
Merged

refactor(updater): simplify digest verification flows #549

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
652a502
refactor(updater): simplify digest verification flows
magrhino Jul 21, 2026
9ea252b
test(updater): cover digest failure continuation
magrhino Jul 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
241 changes: 147 additions & 94 deletions src/wudup/updater_lifecycle_digest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,47 +67,77 @@ def _verify_expected_digests(
for match in matches
if (expected_digest := _expected_digest_requirement(match))
}
for line_no, target, expected_image, allow_repo, expected in sorted(requirements):
matched = False
digest_result: DigestCheckResult | None = None
stale_result: DigestCheckResult | None = None
for image in images:
if not image_matches_resolved_target(image, expected_image, allow_repo):
continue
matched = True
digest_result = self.digest_verifier.verify(image, expected)
if digest_result.reason == "stale-digest":
stale_result = digest_result
if digest_result.ok:
break
if digest_result is not None and digest_result.status == "untrusted":
self.log.warn(
f"[{stack.name}] Digest verification was inconclusive for line {line_no} ({target}): wanted {expected}"
)
self._log_digest_untrusted(stack.name, digest_result)
continue
if digest_result is None or not digest_result.ok:
for requirement in sorted(requirements):
if not self._verify_expected_digest_requirement(
stack,
requirement,
images,
):
ok = False
self.log.error(
f"[{stack.name}] Expected digest not reached for line {line_no} ({target}): wanted {expected}"
)
if digest_result is not None:
self._log_digest_mismatch(stack.name, digest_result)
if stale_result is not None:
self._mark_stale_pending_digest(
stack,
line_no,
target,
expected,
stale_result,
)
if not matched:
self.log.plain(
"ERROR",
f"[{stack.name}] No compose image matched line {line_no} while checking expected digest",
)
return ok

def _verify_expected_digest_requirement(
self,
stack: ComposeStack,
requirement: tuple[int, str, str, bool, str],
images: Sequence[str],
) -> bool:
line_no, target, expected_image, allow_repo, expected = requirement
digest_result, stale_result, matched = self._verify_expected_digest_images(
expected_image,
allow_repo,
expected,
images,
)
if digest_result is not None and digest_result.status == "untrusted":
self.log.warn(
f"[{stack.name}] Digest verification was inconclusive for line {line_no} ({target}): wanted {expected}"
)
self._log_digest_untrusted(stack.name, digest_result)
return True
if digest_result is not None and digest_result.ok:
return True
self.log.error(
f"[{stack.name}] Expected digest not reached for line {line_no} ({target}): wanted {expected}"
)
if digest_result is not None:
self._log_digest_mismatch(stack.name, digest_result)
if stale_result is not None:
self._mark_stale_pending_digest(
stack,
line_no,
target,
expected,
stale_result,
)
if not matched:
self.log.plain(
"ERROR",
f"[{stack.name}] No compose image matched line {line_no} while checking expected digest",
)
return False

def _verify_expected_digest_images(
self,
expected_image: str,
allow_repo: bool,
expected: str,
images: Sequence[str],
) -> tuple[DigestCheckResult | None, DigestCheckResult | None, bool]:
matched = False
digest_result: DigestCheckResult | None = None
stale_result: DigestCheckResult | None = None
for image in images:
if not image_matches_resolved_target(image, expected_image, allow_repo):
continue
matched = True
digest_result = self.digest_verifier.verify(image, expected)
if digest_result.reason == "stale-digest":
stale_result = digest_result
if digest_result.ok:
break
return digest_result, stale_result, matched

def _mark_stale_pending_digest(
self,
stack: ComposeStack,
Expand Down Expand Up @@ -144,65 +174,88 @@ def _verify_digest_pin_updates(
) -> bool:
ok = True
for update in updates:
current = self._verify_digest_pin_update_target(update)
if not current.ok:
if not self._verify_digest_pin_update(stack, update, images):
ok = False
if current.reason == "stale-digest":
current_digest = normalize_digest(current.digest)
suffix = f", current {current_digest}" if current_digest else ""
self.log.plain(
"ERROR",
f"[{stack.name}] Digest-pin target moved for "
f"{update.resolved_image}: planned {update.planned_digest}"
f"{suffix}",
)
else:
self.log.error(
f"[{stack.name}] Could not re-resolve digest-pin target "
f"{update.resolved_image}: {current.reason}"
)
if current.error:
self.log.plain(
"ERROR",
f"[{stack.name}] Digest resolution error: {updater_logging.sanitize_stream(current.error)}",
)
continue
matched = False
digest_result: DigestCheckResult | None = None
for image in images:
if not image_matches_resolved_target(
image,
update.resolved_image,
False,
):
continue
matched = True
digest_result = self.digest_verifier.verify(
image,
update.planned_digest,
)
if digest_result.ok:
break
if digest_result is not None and digest_result.ok:
self.log.info(
f"[{stack.name}] Verified digest-pin target: "
f"{update.resolved_image} -> {update.planned_digest}"
)
return ok

def _verify_digest_pin_update(
self,
stack: ComposeStack,
update: DigestPinUpdate,
images: Sequence[str],
) -> bool:
current = self._verify_digest_pin_update_target(update)
if not current.ok:
self._log_digest_pin_resolution_failure(stack, update, current)
return False
digest_result, matched = self._verify_digest_pin_images(update, images)
if digest_result is not None and digest_result.ok:
self.log.info(
f"[{stack.name}] Verified digest-pin target: "
f"{update.resolved_image} -> {update.planned_digest}"
)
return True
self.log.error(
f"[{stack.name}] Digest-pin target did not verify for "
f"{update.resolved_image}: wanted {update.planned_digest}"
)
if digest_result is not None:
self._log_digest_mismatch(stack.name, digest_result)
if not matched:
self.log.plain(
"ERROR",
f"[{stack.name}] No compose image matched digest-pin target "
f"{update.resolved_image}",
)
return False

def _log_digest_pin_resolution_failure(
self,
stack: ComposeStack,
update: DigestPinUpdate,
current: DigestResolveResult,
) -> None:
if current.reason == "stale-digest":
current_digest = normalize_digest(current.digest)
suffix = f", current {current_digest}" if current_digest else ""
self.log.plain(
"ERROR",
f"[{stack.name}] Digest-pin target moved for "
f"{update.resolved_image}: planned {update.planned_digest}{suffix}",
)
return
self.log.error(
f"[{stack.name}] Could not re-resolve digest-pin target "
f"{update.resolved_image}: {current.reason}"
)
if current.error:
self.log.plain(
"ERROR",
f"[{stack.name}] Digest resolution error: {updater_logging.sanitize_stream(current.error)}",
)

def _verify_digest_pin_images(
self,
update: DigestPinUpdate,
images: Sequence[str],
) -> tuple[DigestCheckResult | None, bool]:
matched = False
digest_result: DigestCheckResult | None = None
for image in images:
if not image_matches_resolved_target(
image,
Comment thread
sourcery-ai[bot] marked this conversation as resolved.
update.resolved_image,
False,
):
continue
ok = False
self.log.error(
f"[{stack.name}] Digest-pin target did not verify for "
f"{update.resolved_image}: wanted {update.planned_digest}"
matched = True
digest_result = self.digest_verifier.verify(
image,
update.planned_digest,
)
if digest_result is not None:
self._log_digest_mismatch(stack.name, digest_result)
if not matched:
self.log.plain(
"ERROR",
f"[{stack.name}] No compose image matched digest-pin target "
f"{update.resolved_image}",
)
return ok
if digest_result.ok:
break
return digest_result, matched

def _log_digest_untrusted(
self,
Expand Down
50 changes: 50 additions & 0 deletions tests/test_python_update_from_wud_core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,13 @@

import json
import unittest
from unittest import mock

from wudup.compose import (
ComposeStack,
ServiceImage,
)
from wudup.digest_verifier import DigestCheckResult
from wudup.updater_lifecycle_health import _updated_images
from wudup.updater_models import (
ImageState,
Expand Down Expand Up @@ -90,6 +92,54 @@ def test_expected_digest_failure_reason_requires_all_matches_stale(self) -> None
runner.lifecycle._expected_digest_failure_reason(stack, matches),
STALE_PENDING_DIGEST_REASON,
)
def test_expected_digest_verification_checks_all_requirements(self) -> None:
stack_dir = self.make_stack(
"app",
[
("first", "repo/first:latest", "cid-first"),
("second", "repo/second:latest", "cid-second"),
],
)
stack = ComposeStack(
index=1,
Comment thread
sourcery-ai[bot] marked this conversation as resolved.
directory=stack_dir,
file="docker-compose.yml",
name="app",
images=("repo/first:latest", "repo/second:latest"),
service_images=(
ServiceImage("first", "repo/first:latest"),
ServiceImage("second", "repo/second:latest"),
),
)
targets = parse_wud_text(
"repo/first:latest@sha256:first\n"
"repo/second:latest@sha256:second\n"
).targets
matches = tuple(
Match(
stack=stack,
target=target,
resolved=stack.images[index],
compose_image=stack.images[index],
service=stack.service_images[index].service,
)
for index, target in enumerate(targets)
)
digest_verifier = mock.Mock()
digest_verifier.verify.side_effect = (
DigestCheckResult(False, "mismatch", "digest-mismatch"),
DigestCheckResult(True, "verified", "digest-match"),
)
runner = self.make_runner(digest_verifier=digest_verifier)

verified = runner.lifecycle._verify_expected_digests(
stack,
matches,
stack.images,
)

self.assertFalse(verified)
self.assertEqual(digest_verifier.verify.call_count, 2)
def test_wrapper_default_dry_run_plans_without_mutation(self) -> None:
self.wud_file.write_text("repo/app:latest\n", encoding="utf-8")
self.make_stack("app", [("app", "repo/app:latest", "cid-app")])
Expand Down
Loading