univention.updater package

Contents

univention.updater package#

Python tools for Updater and repository maintenance.

class univention.updater.UniventionUpdater(check_access: bool = True)[source]#

Bases: object

Handle UCS package repositories.

Create new updater with settings from UCR.

Parameters:

check_access (bool) – Check if repository server is reachable on init.

Raises:

ConfigurationError – if configured server is not available immediately.

static call_sh_files(scripts: Iterable[tuple[_UCSServer, _UCSRepo, str | None, str, bytes]], logname: str, *args: str) Iterator[tuple[str, str]][source]#

Get pre- and postup.sh files and call them in the right order:

u = UniventionUpdater()
ver = u.get_next_version(u.current_version)
scripts = u.get_sh_files(ver, ver)
for phase, order in u.call_sh_files(scripts, '/var/log/univention/updater.log', ver):
  if (phase, order) == ('update', 'main'):
    pass
Parameters:
  • scripts – A generator returning the script to call, e.g. get_sh_files()

  • logname (str) – The file name of the log file.

  • args – Additional arguments to pass through to the scripts.

Returns:

A generator returning 2-tuples (phase, part)

component(name: str) Component[source]#
component_update_get_packages() tuple[list[tuple[str, str]], list[tuple[str, str, str]], list[tuple[str, str]]][source]#

Return tuple with list of (new, upgradeable, removed) packages.

Returns:

A 3-tuple (new, upgraded, removed).

Return type:

tuple(list[str], list[str], list[str])

config_repository() None[source]#

Retrieve configuration to access repository. Overridden by univention.updater.UniventionMirror.

get_all_available_release_updates(ucs_version: UCS_Version | None = None) tuple[list[UCS_Version], set[str] | None][source]#

Returns a list of all available release updates - the function takes required components into account and stops if a required component is missing

Parameters:

ucs_version (UCS_Version or None) – starts travelling through available version from version.

Returns:

a list of 2-tuple (versions, blocking_component), where versions is a list of UCS release and blocking_component is the first missing component blocking the update.

Return type:

tuple(list[str], str or None)

get_components(only_localmirror_enabled: bool = False, all: bool = False, only_current: bool = False) set[Component][source]#

Retrieve all (enabled) components from registry as set(). By default, only “enabled” components will be returned (repository/online/component/%s=$TRUE).

Parameters:
  • only_localmirror_enabled (bool) – Only the components enabled for local mirroring. If only_`localmirror`_enabled is True, then all components with repository/online/component/%s/localmirror=$TRUE will be returned. If repository/online/component/%s/localmirror is not set, then the value of repository/online/component/%s is used for backward compatibility.

  • all (bool) – Also return not enabled components.

  • only_current (bool) – Only return components marked as “current”.

Returns:

The set of (enabled) components.

get_next_version(version: UCS_Version, components: Iterable[Component] = [], errorsto: Literal['stderr', 'exception', 'none'] = 'stderr') UCS_Version | None[source]#

Check if a new patchlevel, minor or major release is available for the given version. Components must be available for the same major.minor version.

Parameters:
  • version (UCS_Version) – A UCS release version.

  • components – A list of components, which must be available for the next release.

  • errorsto (str) – Select method of reporting errors; on of ‘stderr’, ‘exception’, ‘none’.

Returns:

The next UCS release or None.

Return type:

UCS_Version or None

Raises:

RequiredComponentError – if a required component is missing

get_releases(start: UCS_Version | None = None, end: UCS_Version | None = None) Iterator[tuple[UCS_Version, dict[str, Any]]][source]#

Return UCS releases in range.

Parameters:
  • start – Minimum requried version.

  • end – Maximum allowed version.

Returns:

Iterator of 2-tuples (UCS_Version, data).

get_sh_files(start: UCS_Version, end: UCS_Version, mirror: bool = False) Iterator[tuple[_UCSServer, _UCSRepo, str | None, str, bytes]][source]#

Return all preup- and postup-scripts of repositories.

Parameters:
  • start (UCS_Version) – The UCS release to start from.

  • end (UCS_Version) – The UCS release where to stop.

  • mirror (bool) – Use the settings for mirroring.

Returns:

iteratable (server, struct, phase, path, script)

Raises:

VerificationError – if the PGP signature is invalid.

See call_sh_files() for an example.

print_component_repositories(clean: bool = False, start: UCS_Version | None = None, end: UCS_Version | None = None, for_mirror_list: bool = False) str[source]#

Return a string of Debian repository statements for all enabled components.

Parameters:
  • clean (bool) – Add additional clean statements for apt-mirror if enabled by UCRV repository/online/component/%s/clean.

  • start (UCS_Version) – optional smallest UCS release to return.

  • end (UCS_Version) – optional largest UCS release to return.

  • for_mirror_list (bool) – component entries for mirror.list will be returned, otherwise component entries for local sources.list.

Returns:

A string with APT statement lines.

Return type:

str

release_update_available(ucs_version: UCS_Version | None = None, errorsto: Literal['stderr', 'exception', 'none'] = 'stderr') UCS_Version | None[source]#

Check if an update is available for the ucs_version.

Parameters:
  • ucs_version (str) – The UCS release to check.

  • errorsto (str) – Select method of reporting errors; on of ‘stderr’, ‘exception’, ‘none’.

Returns:

The next UCS release or None.

Return type:

str or None

release_update_temporary_sources_list(version: UCS_Version) list[str][source]#

Return list of Debian repository statements for the release update including all enabled components.

Parameters:

version – The UCS release.

Returns:

A list of Debian APT sources.list lines.

Return type:

list[str]

run_dist_upgrade() int[source]#

Run apt-get dist-upgrade command.

Returns:

a 3-tuple (return_code, stdout, stderr)

Return type:

tuple(int, str, str)

ucr_reinit() None[source]#

Re-initialize settings.

Subpackages#

Submodules#

univention.updater.commands module#

Common commands to manage Debian packages.

univention.updater.commands.cmd_update = 'apt-get update'#

Update package cache.

univention.updater.commands.cmd_show = 'apt-cache show'#

Show package information.

univention.updater.commands.cmd_upgrade = 'apt-get -o DPkg::Options::=--force-confold -o DPkg::Options::=--force-overwrite -o DPkg::Options::=--force-overwrite-dir --trivial-only=no --assume-yes --quiet=1 upgrade'#

Upgrade only installed packages

univention.updater.commands.cmd_upgrade_sim = 'apt-get -o DPkg::Options::=--force-confold -o DPkg::Options::=--force-overwrite -o DPkg::Options::=--force-overwrite-dir --trivial-only=no --assume-yes --quiet=1 -s upgrade'#

Simulate upgrade only installed packages

univention.updater.commands.cmd_dist_upgrade = 'apt-get -o DPkg::Options::=--force-confold -o DPkg::Options::=--force-overwrite -o DPkg::Options::=--force-overwrite-dir --trivial-only=no --assume-yes --quiet=1 dist-upgrade'#

Upgrade system, may install new packages to satisfy dependencies

univention.updater.commands.cmd_dist_upgrade_sim = 'apt-get -o DPkg::Options::=--force-confold -o DPkg::Options::=--force-overwrite -o DPkg::Options::=--force-overwrite-dir --trivial-only=no --assume-yes --quiet=1 -s dist-upgrade'#

Simulate upgrade system, may install new packages to satisfy dependencies

univention.updater.commands.cmd_install = 'apt-get -o DPkg::Options::=--force-confold -o DPkg::Options::=--force-overwrite -o DPkg::Options::=--force-overwrite-dir --trivial-only=no --assume-yes --quiet=1 install'#

Install packages

univention.updater.commands.cmd_remove = 'apt-get --yes remove'#

Remove packages

univention.updater.commands.cmd_config = 'dpkg --configure -a'#

Configure all pending packages

univention.updater.errors module#

Univention Updater exceptions.

exception univention.updater.errors.UpdaterException[source]#

Bases: Exception

The root of all updater exceptions.

>>> raise UpdaterException()
Traceback (most recent call last):
    ...
univention.updater.errors.UpdaterException
exception univention.updater.errors.RequiredComponentError(version: str, components: set[str])[source]#

Bases: UpdaterException

Signal required component not available.

Parameters:
  • version (str) – The UCS release version.

  • components (set(str)) – A collection of components.

exception univention.updater.errors.PreconditionError(phase: str, order: str, component: str, script: str)[source]#

Bases: UpdaterException

Signal abort by release or component pre-/post-update script.

Parameters:
  • phase (str) – either preup or postup.

  • order (str) – either pre or main or post.

  • component (str) – The name of the component or None.

  • script (str) – The name of the failing script.

>>> raise PreconditionError('preup', 'main', None, 'preup.sh')
Traceback (most recent call last):
    ...
univention.updater.errors.PreconditionError: ('preup', 'main', None, 'preup.sh')
exception univention.updater.errors.DownloadError[source]#

Bases: UpdaterException

Signal temporary error in network communication.

>>> raise DownloadError("file:///preup.sh", 404)
Traceback (most recent call last):
    ...
univention.updater.errors.DownloadError: Error downloading file:///preup.sh: 404
exception univention.updater.errors.ConfigurationError[source]#

Bases: UpdaterException

Signal permanent error in configuration.

>>> raise ConfigurationError("file:///preup.sh", "not found")
Traceback (most recent call last):
    ...
univention.updater.errors.ConfigurationError: Configuration error: not found
exception univention.updater.errors.VerificationError[source]#

Bases: ConfigurationError

Signal permanent error in script verification.

>>> raise VerificationError("file:///preup.sh", "not signed")
Traceback (most recent call last):
    ...
univention.updater.errors.VerificationError: Verification error: not signed
exception univention.updater.errors.CannotResolveComponentServerError(component: str, for_mirror_list: bool)[source]#

Bases: ConfigurationError

Signal permanent error in component configuration.

Parameters:
  • component (str) – The name of the component.

  • for_mirror_list (bool) – True if the error happened while generating the list of repositories to mirror, False while generating the list of repositories for the server itself.

>>> raise CannotResolveComponentServerError("comp", False)
Traceback (most recent call last):
    ...
univention.updater.errors.CannotResolveComponentServerError: Cannot resolve component server for disabled component 'comp' (mirror_list=False).
exception univention.updater.errors.ProxyError[source]#

Bases: ConfigurationError

Signal permanent error in proxy configuration.

>>> raise ProxyError("file:///preup.sh", "blocked")
Traceback (most recent call last):
    ...
univention.updater.errors.ProxyError: Proxy configuration error: blocked file:///preup.sh
exception univention.updater.errors.UnmetDependencyError[source]#

Bases: UpdaterException

Signal unmet package dependencies

>>> raise UnmetDependencyError("stderr")
Traceback (most recent call last):
    ...
univention.updater.errors.UnmetDependencyError: You have unmet dependencies stderr

univention.updater.locking module#

Univention Updater locking

exception univention.updater.locking.LockingError[source]#

Bases: UpdaterException

Signal other updater process running.

>>> raise LockingError(1, "Invalid PID")
Traceback (most recent call last):
    ...
univention.updater.locking.LockingError: Another updater process 1 is currently running according to ...: Invalid PID
class univention.updater.locking.UpdaterLock(timeout: int = 0)[source]#

Bases: object

Context wrapper for updater-lock /var/lock/univention-updater.

updater_lock_acquire() int[source]#

Acquire the updater-lock.

Returns:

0 if it could be acquired within <timeout> seconds, >= 1 if locked by parent.

Return type:

int

Raises:
updater_lock_release() bool[source]#

Release the updater-lock.

Returns:

True if it has been unlocked (or decremented when nested), False if it was already unlocked.

Return type:

bool

univention.updater.locking.apt_lock(timeout=300, out=<colorama.ansitowin32.StreamWrapper object>)[source]#

Acquire and release lock for APT.

Parameters:
  • timeout – Time to wait.

  • out – Output stream for progress and error messages.

univention.updater.mirror module#

Mirror Univention repository server.

univention.updater.mirror.makedirs(dirname: str, mode: int = 493) None[source]#

Recursively create directory hierarchy will all parent directories.

Parameters:
  • dirname (str) – Name of the directory to create.

  • mode (int) – Directory permissions.

univention.updater.mirror.filter_releases_json(releases: Any, start: UCS_Version, end: UCS_Version) None[source]#

Filter releases that are not mirrored from the upstream repository Filtering is done inplace!

Parameters:
  • releases – The upstream releases object from ucs-releases.json.

  • start (UCS_Version) – First UCS version that is being mirrored.

  • end (UCS_Version) – Last UCS version that is being mirrored.

class univention.updater.mirror.UniventionMirror(check_access: bool = True)[source]#

Bases: UniventionUpdater

Create new mirror with settings from UCR.

Parameters:

check_access (bool) – Check if repository server is reachable on init.

Raises:

ConfigurationError – if configured server is not available immediately.

config_repository() None[source]#

Retrieve configuration to access repository. Overrides univention.updater.UniventionUpdater.

release_update_available(ucs_version: str | None = None, errorsto: Literal['stderr', 'exception', 'none'] = 'stderr') UCS_Version | None[source]#

Check if an update is available for the ucs_version.

Parameters:
  • ucs_version (str) – The UCS release to check.

  • errorsto (str) – Select method of reporting errors; on of ‘stderr’, ‘exception’, ‘none’.

Returns:

The next UCS release or None.

Return type:

str or None

mirror_repositories() int[source]#

Uses apt-mirror to copy a repository.

mirror_update_scripts() None[source]#

Mirrors the preup.sh and postup.sh scripts.

write_releases_json()[source]#

Write a ucs-releases.json including only the mirrored releases.

run()[source]#

starts the mirror process.

univention.updater.repo_url module#

Univention Updater: UCR Repository Server URL

class univention.updater.repo_url.UcsRepoUrl(ucr: ConfigRegistry, component: str, default: str | UcsRepoUrl | None = None)[source]#

Bases: object

UCS repository server base URL.

>>> UcsRepoUrl({'_/server': 'hostname'}, '_').path
''
>>> UcsRepoUrl({'_/server': 'hostname', '_/prefix': '/p'}, '_').path
'/p/'
>>> UcsRepoUrl({'_/server': 'hostname', '_/prefix': 'path'}, '_').path
'/path/'
>>> UcsRepoUrl({}, '', UcsRepoUrl({'_/server': 'https://hostname/'}, '_')).private()
'https://hostname/'
>>> UcsRepoUrl({'_/server': 'other'}, '_', UcsRepoUrl({'_/server': 'https://hostname:80/'}, '_')).private()
'http://other/'
>>> UcsRepoUrl({'_/server': 'other'}, '_', UcsRepoUrl.DEFAULT).private()
'http://other/'
>>> UcsRepoUrl({}, '').private() == UcsRepoUrl.DEFAULT
True
DEFAULT = 'https://updates.software-univention.de/'#
property cred: str#
public() str[source]#

URI without credentials.

>>> UcsRepoUrl({'_/server': 'hostname'}, '_').public()
'http://hostname/'
>>> UcsRepoUrl({'_/server': 'hostname', '_/username': 'user', '_/password': 'pass'}, '_').public()
'http://hostname/'
>>> UcsRepoUrl({'_/server': 'https://hostname'}, '_').public()
'https://hostname/'
>>> UcsRepoUrl({'_/server': 'https://user:pass@hostname'}, '_').public()
'https://hostname/'
private() str[source]#

URI with credentials.

>>> UcsRepoUrl({'_/server': 'hostname'}, '_').private()
'http://hostname/'
>>> UcsRepoUrl({'_/server': 'hostname', '_/username': 'user', '_/password': 'pass'}, '_').private()
'http://user:pass@hostname/'
>>> UcsRepoUrl({'_/server': 'https://hostname'}, '_').private()
'https://hostname/'
>>> UcsRepoUrl({'_/server': 'https://user:pass@hostname'}, '_').private()
'https://user:pass@hostname/'

univention.updater.repository module#

Univention Updater helper functions for managing a local repository.

class univention.updater.repository.TeeFile(fds: list[IO[str]] = [])[source]#

Bases: object

Writes the given string to several files at once. Could by used with the print statement

Register multiple file descriptors, to which the data is written.

Parameters:

fds (list(File)) – A list of opened files.

write(data: str) None[source]#

Write string to all registered files.

Parameters:

data (str) – The string to write.

univention.updater.repository.gzip_file(filename: str) int[source]#

Compress file.

Parameters:

filename (str) – The file name of the file to compress.

Returns:

the process exit code.

Return type:

int

univention.updater.repository.copy_package_files(source_dir: str, dest_dir: str) None[source]#

Copy all Debian binary package files and signed updater scripts from source_dir to dest_dir.

Parameters:
  • source_dir (str) – Source directory.

  • dest_dir (str) – Destination directory.

univention.updater.repository.gen_indexes(base: str, version: UCS_Version) None[source]#

Re-generate Debian Packages files from file:dists/ file.

Parameters:

base (str) – Base directory, which contains the per architecture sub directories.

univention.updater.repository.get_repo_basedir(packages_dir: str) str[source]#

Check if a file path is a UCS package repository.

Parameters:

package_dir (str) – A directory path.

Returns:

The canonicalized path without the architecture sub directory.

Return type:

str

univention.updater.repository.assert_local_repository(out: ~typing.IO[str] = <colorama.ansitowin32.StreamWrapper object>) None[source]#

Exit with error if the local repository is not enabled.

Parameters:

out (file) – Override error output. Defaults to sys.stderr.

univention.updater.tools module#

Univention Update tools.

univention.updater.tools.verify_script(script: bytes, signature: bytes) bytes | None[source]#

Verify detached signature of script:

Parameters:
  • script (str) – The script text to verify.

  • signature (str) – The detached signature.

Returns:

None or the error output.

Return type:

None or str

class univention.updater.tools.UCSRepoPool5(release: UCS_Version = None, **kwargs: Any)[source]#

Bases: _UCSRepo

APT repository using the debian pool structure (ucs5 and above).

Parameters:

version (list(int) or tuple(int) or str or UCS_Version) – must a str matching the pattern X.Y-Z or a triple with major, minor and patchlevel.

Raises:

TypeError – if the version cannot be parsed.

>>> v = UCS_Version((2,3,1))
>>> UCS_Version([2,3,1]) == v
True
>>> UCS_Version("2.3-1") == v
True
>>> UCS_Version(v) == v
True
deb(server: _UCSServer, type: str = 'deb', mirror: bool = False) str[source]#

Format for /etc/apt/sources.list.

Parameters:
  • server (str) – The URL of the repository server.

  • type (str) – The repository type, e.g. deb for a binary and deb-src for source package repository.

  • mirror (bool) – Also mirror files for Debian installer.

Returns:

The APT repository stanza.

Return type:

str

>>> r=UCSRepoPool5(major=5, minor=1, patchlevel=0)
>>> r.deb('https://updates.software-univention.de/')
'deb https://updates.software-univention.de/ ucs510 main'
>>> r.deb('https://updates.software-univention.de/', mirror=True)
'deb https://updates.software-univention.de/ ucs510 main main/debian-installer'
>>> r=UCSRepoPool5(major=5, minor=1, patchlevel=0, errata=True)
>>> r.deb('https://updates.software-univention.de/')
'deb https://updates.software-univention.de/ errata510 main'
path(filename: str | None = None) str[source]#

Format pool for directory/file access.

Parameters:

filename – The name of a file in the repository.

Returns:

relative path.

Return type:

str

>>> UCSRepoPool5(major=5, minor=1, patchlevel=0).path()
'dists/ucs510/InRelease'
>>> UCSRepoPool5(major=5, minor=1, patchlevel=0, errata=True).path()
'dists/errata510/InRelease'
class univention.updater.tools.UCSRepoPool(**kw: Any)[source]#

Bases: _UCSRepo

Flat Debian APT repository.

Parameters:

version (list(int) or tuple(int) or str or UCS_Version) – must a str matching the pattern X.Y-Z or a triple with major, minor and patchlevel.

Raises:

TypeError – if the version cannot be parsed.

>>> v = UCS_Version((2,3,1))
>>> UCS_Version([2,3,1]) == v
True
>>> UCS_Version("2.3-1") == v
True
>>> UCS_Version(v) == v
True
deb(server: _UCSServer, type: str = 'deb') str[source]#

Format for /etc/apt/sources.list.

Parameters:
  • server (str) – The URL of the repository server.

  • type (str) – The repository type, e.g. deb for a binary and deb-src for source package repository.

Returns:

The APT repository stanza.

Return type:

str

>>> r=UCSRepoPool(major=2,minor=3,patchlevel=1,part='maintained',arch='amd64')
>>> r.deb('https://updates.software-univention.de/')
'deb https://updates.software-univention.de/2.3/maintained/ 2.3-1/amd64/'
path(filename: str | None = None) str[source]#

Format pool for directory/file access.

Parameters:

filename – The name of a file in the repository.

Returns:

relative path.

Return type:

str

>>> UCSRepoPool(major=2,minor=3).path()
'2.3/'
>>> UCSRepoPool(major=2,minor=3,part='maintained').path()
'2.3/maintained/'
>>> UCSRepoPool(major=2,minor=3,patchlevel=1,part='maintained').path()
'2.3/maintained/2.3-1/'
>>> UCSRepoPool(major=2,minor=3,patchlevel=1,part='maintained',arch='amd64').path()
'2.3/maintained/2.3-1/amd64/Packages.gz'
clean(server: _UCSServer) str[source]#

Format for /etc/apt/mirror.list

Parameters:

server (str) – The URL of the repository server.

Returns:

The APT repository stanza.

Return type:

str

class univention.updater.tools.UCSRepoPoolNoArch(**kw: Any)[source]#

Bases: _UCSRepo

Flat Debian APT repository without explicit architecture subdirectory.

Parameters:

version (list(int) or tuple(int) or str or UCS_Version) – must a str matching the pattern X.Y-Z or a triple with major, minor and patchlevel.

Raises:

TypeError – if the version cannot be parsed.

>>> v = UCS_Version((2,3,1))
>>> UCS_Version([2,3,1]) == v
True
>>> UCS_Version("2.3-1") == v
True
>>> UCS_Version(v) == v
True
ARCHS = {''}#
deb(server: _UCSServer, type: str = 'deb') str[source]#

Format for /etc/apt/sources.list.

Parameters:
  • server (str) – The URL of the repository server.

  • type (str) – The repository type, e.g. deb for a binary and deb-src for source package repository.

Returns:

The APT repository stanza.

Return type:

str

>>> r=UCSRepoPoolNoArch(major=2,minor=3,patch='comp',part='maintained/component',arch='all')
>>> r.deb('https://updates.software-univention.de/')
'deb https://updates.software-univention.de/2.3/maintained/component/comp/ ./'
path(filename: str | None = None) str[source]#

Format pool for directory/file access. Returns relative path.

Parameters:

filename – The name of a file in the repository.

Returns:

relative path.

Return type:

str

>>> UCSRepoPoolNoArch(major=2,minor=3).path()
'2.3/'
>>> UCSRepoPoolNoArch(major=2,minor=3,part='maintained/component').path()
'2.3/maintained/component/'
>>> UCSRepoPoolNoArch(major=2,minor=3,part='maintained/component',patch='comp').path()
'2.3/maintained/component/comp/Packages.gz'
>>> UCSRepoPoolNoArch(major=2,minor=3,part='maintained/component',patch='comp',arch='all').path()
'2.3/maintained/component/comp/Packages.gz'
clean(server: _UCSServer) str[source]#

Format for /etc/apt/mirror.list

Parameters:

server (str) – The URL of the repository server.

Returns:

The APT repository stanza.

Return type:

str

class univention.updater.tools.UCSHttpServer(baseurl: UcsRepoUrl, user_agent: str | None = None, timeout: float | None = None)[source]#

Bases: _UCSServer

Access to UCS compatible remote update server.

Setup URL handler for accessing a UCS repository server.

Parameters:
  • baseurl (UcsRepoUrl) – the base URL.

  • user_agent (str) – optional user agent string.

  • timeout (int) – optional timeout for network access.

class HTTPHeadHandler[source]#

Bases: BaseHandler

Handle fallback from HEAD to GET if unimplemented.

http_error_501(req: Request, fp: Any, code: int, msg: str, headers: dict) Any[source]#
log = <Logger updater.UCSHttp (DEBUG)>#
http_method = 'HEAD'#
head_handler = <univention.updater.tools.UCSHttpServer.HTTPHeadHandler object>#
password_manager = <urllib.request.HTTPPasswordMgrWithDefaultRealm object>#
auth_handler = <urllib.request.HTTPBasicAuthHandler object>#
proxy_handler = <urllib.request.ProxyHandler object>#
opener = <urllib.request.OpenerDirector object>#
failed_hosts: set[str] = {}#
property prefix: str#
classmethod reinit() None[source]#

Reload proxy settings and reset failed hosts.

classmethod load_credentials(ucr: ConfigRegistry) None[source]#

Load credentials from UCR.

Parameters:

ucr (ConfigRegistry) – An UCR instance.

join(rel: str) str[source]#

Return joined URI without credential.

Parameters:

rel (str) – relative URI.

Returns:

The joined URI.

Return type:

str

access(repo: _UCSRepo | None, filename: str | None = None, get: bool = False) tuple[int, int, bytes][source]#

Access URI and optionally get data.

Parameters:
  • repo (_UCSRepo) – the URI to access as an instance of _UCSRepo.

  • filename (str) – An optional relative path.

  • get (bool) – Fetch data if True - otherwise check only.

Returns:

a 3-tuple (code, size, content)

Return type:

tuple(int, int, bytes)

Raises:
  • DownloadError – if the server is unreachable.

  • ValueError – if the credentials use an invalid encoding.

  • ConfigurationError – if a permanent error in the configuration occurs, e.g. the credentials are invalid or the host is unresolvable.

  • ProxyError – if the HTTP proxy returned an error.

class univention.updater.tools.UCSLocalServer(prefix: str)[source]#

Bases: _UCSServer

Access to UCS compatible local update server.

Setup URL handler for accessing a UCS repository server.

Parameters:

prefix (str) – The local path of the repository.

property prefix: str#
join(rel: str) str[source]#

Return joined URI without credential.

Parameters:

rel (str) – relative URI.

Returns:

The joined URI.

Return type:

str

access(repo: _UCSRepo | None, filename: str | None = None, get: bool = False) tuple[int, int, bytes][source]#

Access URI and optionally get data.

Parameters:
  • repo (_UCSRepo) – the URI to access as an instance of _UCSRepo.

  • filename (str) – An optional relative path.

  • get (bool) – Fetch data if True - otherwise check only.

Returns:

a 3-tuple (code, size, content)

Return type:

tuple(int, int, bytes)

Raises:
  • DownloadError – if the server is unreachable.

  • ValueError – if the credentials use an invalid encoding.

  • ConfigurationError – if a permanent error in the configuration occurs, e.g. the credentials are invalid or the host is unresolvable.

  • ProxyError – if the HTTP proxy returned an error.

class univention.updater.tools.Component(updater: UniventionUpdater, name: str)[source]#

Bases: object

FN_APTSOURCES = '/etc/apt/sources.list.d/20_ucs-online-component.list'#
UCRV = 'repository/online/component/{}/{}'#
AVAILABLE = 'available'#
NOT_FOUND = 'not_found'#
DISABLED = 'disabled'#
UNKNOWN = 'unknown'#
PERMISSION_DENIED = 'permission_denied'#
ucrv(key: str = '') str[source]#
property current: bool#
property default_packages: set[str]#

Returns a set of (meta) package names to be installed for this component.

Returns:

a set of package names.

defaultpackage_installed(ignore_invalid_package_names: bool = True) bool | None[source]#

Returns installation status of component’s default packages

Parameters:

ignore_invalid_package_names (bool) – Ignore invalid package names.

Returns:

On of the values:

None

no default packages are defined

True

all default packages are installed

False

at least one package is not installed

Return type:

None or bool

Raises:

ValueError – if UCR variable contains invalid package names if ignore_invalid_package_names=False

baseurl(for_mirror_list: bool = False) UcsRepoUrl[source]#

Calculate the base URL for a component.

Parameters:

for_mirror_list (bool) – Use external or local repository.

CS (component server)

value of repository/online/component/%s/server

MS (mirror server)

value of repository/mirror/server

RS (repository server)

value of repository/online/server

-

value is unset or no entry

/blank/

value is irrelevant

UCR configuration

Result

isRepoServer

enabled

localmirror

server

sources.list mirror.list

False

False

False

-

-

-

no local repository mirror

True

-

RS

-

True

CS

CS

-

True

False

False

-

-

local repository mirror

False

True

-

-

MS

False

True

CS

-

CS

True

False

-

MS

-

True

False

CS

CS

-

True

True

-

RS

MS

True

True

CS

RS

CS

False

-

-

-

-

backward compabibility [1]

True

-

RS

MS

True

CS

RS

CS

server(for_mirror_list: bool = False) UCSHttpServer[source]#

Return UCSHttpServer for component as configures via UCR.

Parameters:

for_mirror_list (bool) – component entries for mirror.list will be returned, otherwise component entries for local sources.list.

Returns:

The repository server for the component.

Return type:

UCSHttpServer

Raises:

ConfigurationError – if the configured server is not usable.

versions(start: UCS_Version, end: UCS_Version, for_mirror_list: bool = False) Iterator[tuple[UCSHttpServer, _UCSRepo]][source]#

Iterate component versions.

Parameters:
  • start – Minimum requried version.

  • end – Maximum allowed version.

  • clean (bool) – Add additional clean statements for apt-mirror.

  • for_mirror_list (bool) – component entries for mirror.list will be returned, otherwise component entries for local sources.list.

Returns:

A iterator returning 2-tuples (server, ver).

repositories(start: UCS_Version, end: UCS_Version, clean: bool = False, for_mirror_list: bool = False, failed: set[tuple[Component, str]] | None = None) Iterator[str][source]#

Return list of Debian repository statements for requested component.

Parameters:
  • start – Minimum requried version.

  • end – Maximum allowed version.

  • clean (bool) – Add additional clean statements for apt-mirror.

  • for_mirror_list (bool) – component entries for mirror.list will be returned, otherwise component entries for local sources.list.

  • failed – A set to recive the failed component names.

Returns:

A list of strings with APT statements.

status() str[source]#

Returns the current status of specified component based on /etc/apt/sources.list.d/20_ucs-online-component.list

Returns:

One of the strings:

DISABLED

component has been disabled via UCR

AVAILABLE

component is enabled and at least one valid repo string has been found in .list file

NOT_FOUND

component is enabled but no valid repo string has been found in .list file

PERMISSION_DENIED

component is enabled but authentication failed

UNKNOWN

component’s status is unknown

Return type:

str

property layout: type[_UCSRepo]#
class univention.updater.tools.UniventionUpdater(check_access: bool = True)[source]#

Bases: object

Handle UCS package repositories.

Create new updater with settings from UCR.

Parameters:

check_access (bool) – Check if repository server is reachable on init.

Raises:

ConfigurationError – if configured server is not available immediately.

config_repository() None[source]#

Retrieve configuration to access repository. Overridden by univention.updater.UniventionMirror.

ucr_reinit() None[source]#

Re-initialize settings.

get_releases(start: UCS_Version | None = None, end: UCS_Version | None = None) Iterator[tuple[UCS_Version, dict[str, Any]]][source]#

Return UCS releases in range.

Parameters:
  • start – Minimum requried version.

  • end – Maximum allowed version.

Returns:

Iterator of 2-tuples (UCS_Version, data).

get_next_version(version: UCS_Version, components: Iterable[Component] = [], errorsto: Literal['stderr', 'exception', 'none'] = 'stderr') UCS_Version | None[source]#

Check if a new patchlevel, minor or major release is available for the given version. Components must be available for the same major.minor version.

Parameters:
  • version (UCS_Version) – A UCS release version.

  • components – A list of components, which must be available for the next release.

  • errorsto (str) – Select method of reporting errors; on of ‘stderr’, ‘exception’, ‘none’.

Returns:

The next UCS release or None.

Return type:

UCS_Version or None

Raises:

RequiredComponentError – if a required component is missing

get_all_available_release_updates(ucs_version: UCS_Version | None = None) tuple[list[UCS_Version], set[str] | None][source]#

Returns a list of all available release updates - the function takes required components into account and stops if a required component is missing

Parameters:

ucs_version (UCS_Version or None) – starts travelling through available version from version.

Returns:

a list of 2-tuple (versions, blocking_component), where versions is a list of UCS release and blocking_component is the first missing component blocking the update.

Return type:

tuple(list[str], str or None)

release_update_available(ucs_version: UCS_Version | None = None, errorsto: Literal['stderr', 'exception', 'none'] = 'stderr') UCS_Version | None[source]#

Check if an update is available for the ucs_version.

Parameters:
  • ucs_version (str) – The UCS release to check.

  • errorsto (str) – Select method of reporting errors; on of ‘stderr’, ‘exception’, ‘none’.

Returns:

The next UCS release or None.

Return type:

str or None

release_update_temporary_sources_list(version: UCS_Version) list[str][source]#

Return list of Debian repository statements for the release update including all enabled components.

Parameters:

version – The UCS release.

Returns:

A list of Debian APT sources.list lines.

Return type:

list[str]

component(name: str) Component[source]#
get_components(only_localmirror_enabled: bool = False, all: bool = False, only_current: bool = False) set[Component][source]#

Retrieve all (enabled) components from registry as set(). By default, only “enabled” components will be returned (repository/online/component/%s=$TRUE).

Parameters:
  • only_localmirror_enabled (bool) – Only the components enabled for local mirroring. If only_`localmirror`_enabled is True, then all components with repository/online/component/%s/localmirror=$TRUE will be returned. If repository/online/component/%s/localmirror is not set, then the value of repository/online/component/%s is used for backward compatibility.

  • all (bool) – Also return not enabled components.

  • only_current (bool) – Only return components marked as “current”.

Returns:

The set of (enabled) components.

component_update_get_packages() tuple[list[tuple[str, str]], list[tuple[str, str, str]], list[tuple[str, str]]][source]#

Return tuple with list of (new, upgradeable, removed) packages.

Returns:

A 3-tuple (new, upgraded, removed).

Return type:

tuple(list[str], list[str], list[str])

run_dist_upgrade() int[source]#

Run apt-get dist-upgrade command.

Returns:

a 3-tuple (return_code, stdout, stderr)

Return type:

tuple(int, str, str)

print_component_repositories(clean: bool = False, start: UCS_Version | None = None, end: UCS_Version | None = None, for_mirror_list: bool = False) str[source]#

Return a string of Debian repository statements for all enabled components.

Parameters:
  • clean (bool) – Add additional clean statements for apt-mirror if enabled by UCRV repository/online/component/%s/clean.

  • start (UCS_Version) – optional smallest UCS release to return.

  • end (UCS_Version) – optional largest UCS release to return.

  • for_mirror_list (bool) – component entries for mirror.list will be returned, otherwise component entries for local sources.list.

Returns:

A string with APT statement lines.

Return type:

str

static call_sh_files(scripts: Iterable[tuple[_UCSServer, _UCSRepo, str | None, str, bytes]], logname: str, *args: str) Iterator[tuple[str, str]][source]#

Get pre- and postup.sh files and call them in the right order:

u = UniventionUpdater()
ver = u.get_next_version(u.current_version)
scripts = u.get_sh_files(ver, ver)
for phase, order in u.call_sh_files(scripts, '/var/log/univention/updater.log', ver):
  if (phase, order) == ('update', 'main'):
    pass
Parameters:
  • scripts – A generator returning the script to call, e.g. get_sh_files()

  • logname (str) – The file name of the log file.

  • args – Additional arguments to pass through to the scripts.

Returns:

A generator returning 2-tuples (phase, part)

get_sh_files(start: UCS_Version, end: UCS_Version, mirror: bool = False) Iterator[tuple[_UCSServer, _UCSRepo, str | None, str, bytes]][source]#

Return all preup- and postup-scripts of repositories.

Parameters:
  • start (UCS_Version) – The UCS release to start from.

  • end (UCS_Version) – The UCS release where to stop.

  • mirror (bool) – Use the settings for mirroring.

Returns:

iteratable (server, struct, phase, path, script)

Raises:

VerificationError – if the PGP signature is invalid.

See call_sh_files() for an example.

class univention.updater.tools.LocalUpdater[source]#

Bases: UniventionUpdater

Direct file access to local repository.

Create new updater with settings from UCR.

Parameters:

check_access (bool) – Check if repository server is reachable on init.

Raises:

ConfigurationError – if configured server is not available immediately.