/usr/lib/python3/dist-packages/DistUpgrade/DistUpgradeViewNonInteractive.py is in python3-distupgrade 1:16.04.12.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 | # DistUpgradeView.py
#
# Copyright (c) 2004,2005 Canonical
#
# Author: Michael Vogt <michael.vogt@ubuntu.com>
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation; either version 2 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
# USA
import apt
import apt_pkg
import logging
import locale
import time
import sys
import os
import pty
import select
import subprocess
import copy
import apt.progress
from configparser import NoSectionError, NoOptionError
from subprocess import PIPE, Popen
from .DistUpgradeView import DistUpgradeView, InstallProgress, AcquireProgress
from .DistUpgradeConfigParser import DistUpgradeConfig
class NonInteractiveAcquireProgress(AcquireProgress):
def update_status(self, uri, descr, shortDescr, status):
AcquireProgress.update_status(self, uri, descr, shortDescr, status)
#logging.debug("Fetch: updateStatus %s %s" % (uri, status))
if status == apt_pkg.STAT_DONE:
print("fetched %s (%.2f/100) at %sb/s" % (
uri, self.percent, apt_pkg.size_to_str(int(self.current_cps))))
if sys.stdout.isatty():
sys.stdout.flush()
class NonInteractiveInstallProgress(InstallProgress):
"""
Non-interactive version of the install progress class
This ensures that conffile prompts are handled and that
hanging scripts are killed after a (long) timeout via ctrl-c
"""
def __init__(self, logdir):
InstallProgress.__init__(self)
logging.debug("setting up environ for non-interactive use")
if "DEBIAN_FRONTEND" not in os.environ:
os.environ["DEBIAN_FRONTEND"] = "noninteractive"
os.environ["APT_LISTCHANGES_FRONTEND"] = "none"
os.environ["RELEASE_UPRADER_NO_APPORT"] = "1"
self.config = DistUpgradeConfig(".")
self.logdir = logdir
self.install_run_number = 0
try:
if self.config.getWithDefault("NonInteractive","ForceOverwrite", False):
apt_pkg.config.set("DPkg::Options::","--force-overwrite")
except (NoSectionError, NoOptionError):
pass
# more debug
#apt_pkg.config.set("Debug::pkgOrderList","true")
#apt_pkg.config.set("Debug::pkgDPkgPM","true")
# default to 2400 sec timeout
self.timeout = 2400
try:
self.timeout = self.config.getint("NonInteractive","TerminalTimeout")
except Exception:
pass
def error(self, pkg, errormsg):
logging.error("got a error from dpkg for pkg: '%s': '%s'" % (pkg, errormsg))
# check if re-run of maintainer script is requested
if not self.config.getWithDefault(
"NonInteractive","DebugBrokenScripts", False):
return
# re-run maintainer script with sh -x/perl debug to get a better
# idea what went wrong
#
# FIXME: this is just a approximation for now, we also need
# to pass:
# - a version after remove (if upgrade to new version)
#
# not everything is a shell or perl script
#
# if the new preinst fails, its not yet in /var/lib/dpkg/info
# so this is inaccurate as well
environ = copy.copy(os.environ)
environ["PYCENTRAL"] = "debug"
cmd = []
# find what maintainer script failed
if "post-installation" in errormsg:
prefix = "/var/lib/dpkg/info/"
name = "postinst"
argument = "configure"
maintainer_script = "%s/%s.%s" % (prefix, pkg, name)
elif "pre-installation" in errormsg:
prefix = "/var/lib/dpkg/tmp.ci/"
#prefix = "/var/lib/dpkg/info/"
name = "preinst"
argument = "install"
maintainer_script = "%s/%s" % (prefix, name)
elif "pre-removal" in errormsg:
prefix = "/var/lib/dpkg/info/"
name = "prerm"
argument = "remove"
maintainer_script = "%s/%s.%s" % (prefix, pkg, name)
elif "post-removal" in errormsg:
prefix = "/var/lib/dpkg/info/"
name = "postrm"
argument = "remove"
maintainer_script = "%s/%s.%s" % (prefix, pkg, name)
else:
print("UNKNOWN (trigger?) dpkg/script failure for %s (%s) " % (pkg, errormsg))
return
# find out about the interpreter
if not os.path.exists(maintainer_script):
logging.error("can not find failed maintainer script '%s' " % maintainer_script)
return
with open(maintainer_script) as f:
interp = f.readline()[2:].strip().split()[0]
if ("bash" in interp) or ("/bin/sh" in interp):
debug_opts = ["-ex"]
elif ("perl" in interp):
debug_opts = ["-d"]
environ["PERLDB_OPTS"] = "AutoTrace NonStop"
else:
logging.warning("unknown interpreter: '%s'" % interp)
# check if debconf is used and fiddle a bit more if it is
with open(maintainer_script) as f:
maintainer_script_text = f.read()
if ". /usr/share/debconf/confmodule" in maintainer_script_text:
environ["DEBCONF_DEBUG"] = "developer"
environ["DEBIAN_HAS_FRONTEND"] = "1"
interp = "/usr/share/debconf/frontend"
debug_opts = ["sh","-ex"]
# build command
cmd.append(interp)
cmd.extend(debug_opts)
cmd.append(maintainer_script)
cmd.append(argument)
# check if we need to pass a version
if name == "postinst":
version = Popen("dpkg-query -s %s|grep ^Config-Version" % pkg,
shell=True, stdout=PIPE,
universal_newlines=True).communicate()[0]
if version:
cmd.append(version.split(":",1)[1].strip())
elif name == "preinst":
pkg = os.path.basename(pkg)
pkg = pkg.split("_")[0]
version = Popen("dpkg-query -s %s|grep ^Version" % pkg,
shell=True, stdout=PIPE,
universal_newlines=True).communicate()[0]
if version:
cmd.append(version.split(":",1)[1].strip())
logging.debug("re-running '%s' (%s)" % (cmd, environ))
ret = subprocess.call(cmd, env=environ)
logging.debug("%s script returned: %s" % (name,ret))
def conffile(self, current, new):
logging.warning("got a conffile-prompt from dpkg for file: '%s'" % current)
# looks like we have a race here *sometimes*
time.sleep(5)
try:
# don't overwrite
os.write(self.master_fd,"n\n")
except Exception as e:
logging.error("error '%s' when trying to write to the conffile"%e)
def start_update(self):
InstallProgress.start_update(self)
self.last_activity = time.time()
progress_log = self.config.getWithDefault("NonInteractive","DpkgProgressLog", False)
if progress_log:
fullpath = os.path.join(self.logdir, "dpkg-progress.%s.log" % self.install_run_number)
logging.debug("writing dpkg progress log to '%s'" % fullpath)
self.dpkg_progress_log = open(fullpath, "w")
else:
self.dpkg_progress_log = open(os.devnull, "w")
self.dpkg_progress_log.write("%s: Start\n" % time.time())
def finish_update(self):
InstallProgress.finish_update(self)
self.dpkg_progress_log.write("%s: Finished\n" % time.time())
self.dpkg_progress_log.close()
self.install_run_number += 1
def status_change(self, pkg, percent, status_str):
self.dpkg_progress_log.write("%s:%s:%s:%s\n" % (time.time(),
percent,
pkg,
status_str))
def update_interface(self):
InstallProgress.update_interface(self)
if self.statusfd == None:
return
if (self.last_activity + self.timeout) < time.time():
logging.warning("no activity %s seconds (%s) - sending ctrl-c" % (
self.timeout, self.status))
# ctrl-c
os.write(self.master_fd,chr(3))
# read master fd and write to stdout so that terminal output
# actualy works
res = select.select([self.master_fd],[],[],0.1)
while len(res[0]) > 0:
self.last_activity = time.time()
try:
s = os.read(self.master_fd, 1)
sys.stdout.write("%s" % s.decode(locale.getpreferredencoding()))
except OSError:
# happens after we are finished because the fd is closed
return
res = select.select([self.master_fd],[],[],0.1)
sys.stdout.flush()
def fork(self):
logging.debug("doing a pty.fork()")
# some maintainer scripts fail without
os.environ["TERM"] = "dumb"
# unset PAGER so that we can do "diff" in the dpkg prompt
os.environ["PAGER"] = "true"
(self.pid, self.master_fd) = pty.fork()
if self.pid != 0:
logging.debug("pid is: %s" % self.pid)
return self.pid
class DistUpgradeViewNonInteractive(DistUpgradeView):
" non-interactive version of the upgrade view "
def __init__(self, datadir=None, logdir=None):
DistUpgradeView.__init__(self)
self.config = DistUpgradeConfig(".")
self._acquireProgress = NonInteractiveAcquireProgress()
self._installProgress = NonInteractiveInstallProgress(logdir)
self._opProgress = apt.progress.base.OpProgress()
sys.__excepthook__ = self.excepthook
def excepthook(self, type, value, traceback):
" on uncaught exceptions -> print error and reboot "
logging.exception("got exception '%s': %s " % (type, value))
#sys.excepthook(type, value, traceback)
self.confirmRestart()
def getOpCacheProgress(self):
" return a OpProgress() subclass for the given graphic"
return self._opProgress
def getAcquireProgress(self):
" return an acquire progress object "
return self._acquireProgress
def getInstallProgress(self, cache=None):
" return a install progress object "
return self._installProgress
def updateStatus(self, msg):
""" update the current status of the distUpgrade based
on the current view
"""
pass
def setStep(self, step):
""" we have 5 steps current for a upgrade:
1. Analyzing the system
2. Updating repository information
3. Performing the upgrade
4. Post upgrade stuff
5. Complete
"""
pass
def confirmChanges(self, summary, changes, demotions, downloadSize,
actions=None, removal_bold=True):
DistUpgradeView.confirmChanges(self, summary, changes, demotions,
downloadSize, actions)
logging.debug("toinstall: '%s'" % [p.name for p in self.toInstall])
logging.debug("toupgrade: '%s'" % [p.name for p in self.toUpgrade])
logging.debug("toremove: '%s'" % [p.name for p in self.toRemove])
return True
def askYesNoQuestion(self, summary, msg, default='No'):
" ask a Yes/No question and return True on 'Yes' "
# if this gets enabled upgrades over ssh with the non-interactive
# frontend will no longer work
#if default.lower() == "no":
# return False
return True
def confirmRestart(self):
" generic ask about the restart, can be overridden "
logging.debug("confirmRestart() called")
# rebooting here makes sense if we run e.g. in qemu
return self.config.getWithDefault("NonInteractive","RealReboot", False)
def error(self, summary, msg, extended_msg=None):
" display a error "
logging.error("%s %s (%s)" % (summary, msg, extended_msg))
def abort(self):
logging.error("view.abort called")
if __name__ == "__main__":
view = DistUpgradeViewNonInteractive()
ap = NonInteractiveAcquireProgress()
ip = NonInteractiveInstallProgress()
#ip.error("linux-image-2.6.17-10-generic","post-installation script failed")
ip.error("xserver-xorg","pre-installation script failed")
cache = apt.Cache()
for pkg in sys.argv[1:]:
#if cache[pkg].is_installed:
# cache[pkg].mark_delete()
#else:
cache[pkg].mark_install()
cache.commit(ap, ip)
time.sleep(2)
sys.exit(0)
|