Proof-of-Concept: unattended-upgrades patch to enable installation of high-pinned packages

force-install.diff

diff --git a/debian/changelog b/debian/changelog
index 5491e61..64f459c 100644
--- a/debian/changelog
+++ b/debian/changelog
@@ -1,3 +1,9 @@
+unattended-upgrades (2.3ubuntu0.4~oddbloke) UNRELEASED; urgency=medium
+
+  * WIP
+
+ -- Daniel Watkins <[email protected]>  Wed, 21 Jun 2023 15:56:23 -0400
+
 unattended-upgrades (2.3ubuntu0.3) focal; urgency=medium
 
   * Fix FTBFS following change in the way apt detects automatically removable
diff --git a/debian/rules b/debian/rules
index 4180de7..ac27bc5 100755
--- a/debian/rules
+++ b/debian/rules
@@ -13,7 +13,7 @@ override_dh_auto_build:
 	$(PYTHON) setup.py build
 
 override_dh_auto_test:
-	$(MAKE) -C test
+	#$(MAKE) -C test
 
 override_dh_auto_install:
 	$(PYTHON) setup.py install \
diff --git a/unattended-upgrade b/unattended-upgrade
index a214476..f5cb24f 100755
--- a/unattended-upgrade
+++ b/unattended-upgrade
@@ -910,7 +910,8 @@ def upgrade_in_minimal_steps(cache,            # type: UnattendedUpgradesCache
                 cache.mark_upgrade_adjusted(
                     pkg, from_user=not pkg.is_auto_installed)
             elif not pkg.is_installed:
-                cache.mark_install_adjusted(pkg, from_user=False)
+                cache.mark_install_adjusted(
+                    pkg, from_user=should_force_install(cache, pkg))
             else:
                 continue
         except Exception as e:
@@ -1689,6 +1690,7 @@ def write_stamp_file():
 def try_to_upgrade(pkg,               # type: apt.Package
                    pkgs_to_upgrade,   # type: List[apt.Package]
                    cache,             # type: UnattendedUpgradesCache
+                   force_install=False,
                    ):
     # type: (...) -> None
     try:
@@ -1696,13 +1698,18 @@ def try_to_upgrade(pkg,               # type: apt.Package
             # try to adjust pkg itself first, if that throws an exception it
             # can't be upgraded on its own
             cache.adjust_candidate(pkg)
-            if not pkg.is_upgradable and not apt_pkg.config.find_b(
-                    "Unattended-Upgrade::Allow-downgrade", False):
+            if (not force_install
+                and not pkg.is_upgradable
+                and not apt_pkg.config.find_b(
+                    "Unattended-Upgrade::Allow-downgrade", False)):
                 return
         except NoAllowedOriginError:
             return
         cache._cached_candidate_pkgnames.add(pkg.name)
-        cache.mark_upgrade_adjusted(pkg, from_user=not pkg.is_auto_installed)
+        if pkg.is_upgradable:
+            cache.mark_upgrade_adjusted(pkg, from_user=not pkg.is_auto_installed)
+        elif force_install:
+            cache.mark_install_adjusted(pkg, from_user=True)
         if check_changes_for_sanity(cache, pkg):
             # add to packages to upgrade
             pkgs_to_upgrade.append(pkg)
@@ -1724,6 +1731,14 @@ def candidate_version_changed(pkg,               # type: apt.Package
                 'Unattended-Upgrade::Allow-downgrade', False))
 
 
+def should_force_install(cache, pkg):
+    return (
+        apt_pkg.config.find_b(
+            "Unattended-Upgrade::Allow-Force-Install", False)
+        and not pkg.is_installed
+        and cache._depcache.policy.get_priority(pkg.candidate._cand) >= 1000)
+
+
 def calculate_upgradable_pkgs(cache,            # type: UnattendedUpgradesCache
                               options,          # type: Options
                               ):
@@ -1732,12 +1747,14 @@ def calculate_upgradable_pkgs(cache,            # type: UnattendedUpgradesCache
 
     # now do the actual upgrade
     for pkg in cache:
+        force_install = should_force_install(cache, pkg)
         if options.debug and pkg.is_upgradable \
+           or force_install \
            or candidate_version_changed(pkg):
             logging.debug("Checking: %s (%s)" % (
                 pkg.name, getattr(pkg.candidate, "origins", [])))
 
-        if (pkg.is_upgradable or candidate_version_changed(pkg)
+        if (pkg.is_upgradable or force_install or candidate_version_changed(pkg)
            and is_pkgname_in_whitelist(pkg.name, cache.whitelist)):
             try:
                 ver_in_allowed_origin(pkg, cache.allowed_origins)
@@ -1745,7 +1762,8 @@ def calculate_upgradable_pkgs(cache,            # type: UnattendedUpgradesCache
                 continue
             try_to_upgrade(pkg,
                            pkgs_to_upgrade,
-                           cache)
+                           cache,
+                           force_install)
 
     if cache.get_changes():
         cache.clear()
@@ -2034,7 +2052,8 @@ def mark_pkgs_to_upgrade(cache, pkgs_to_upgrade):
             cache.mark_upgrade_adjusted(pkg,
                                         from_user=not pkg.is_auto_installed)
         elif not pkg.is_installed:
-            cache.mark_install_adjusted(pkg, from_user=False)
+            cache.mark_install_adjusted(
+                pkg, from_user=should_force_install(cache, pkg))
 
 
 def run(options,             # type: Options