API Documentation

A single page with the entire API reference.

tufup.client:

class tufup.client.Client(app_name: str, app_install_dir: Path, current_version: str, metadata_dir: Path, metadata_base_url: str, target_dir: Path, target_base_url: str, extract_dir: Path | None = None, refresh_required: bool = False, session_auth: Dict[str, Tuple[str, str] | AuthBase] | None = None, binary_diff: type[BinaryDiff] | None = None, **kwargs)

__init__(app_name: str, app_install_dir: Path, current_version: str, metadata_dir: Path, metadata_base_url: str, target_dir: Path, target_base_url: str, extract_dir: Path | None = None, refresh_required: bool = False, session_auth: Dict[str, Tuple[str, str] | AuthBase] | None = None, binary_diff: type[BinaryDiff] | None = None, **kwargs)

The tufup.client.Client is a subclass of tuf.ngclient.Updater.

for session_auth formats, see docs for tufup.client.AuthRequestsFetcher

property trusted_target_metas: list

Return a list of available trusted targets, as TargetMeta objects.

This is convenient because TargetMeta objects can be sorted by version.

get_targetinfo(target_path: str | TargetMeta) → TargetFile | None: Extend Updater.get_targetinfo to handle TargetMeta input args.

property updates_available

download_and_apply_update(skip_confirmation: bool = False, install: Callable | None = None, progress_hook: Callable | None = None, **kwargs)

Download and apply available updates.

Note that check_for_updates must be called first.

This downloads the files found by check_for_updates, applies any patches, and extracts the resulting archive to the extract_dir. At that point, the update is ready to be installed (i.e. moved into place). This is done by calling install with the specified **kwargs.

The default install callable moves the content of extract_dir to app_install_dir, and exits the application (not necessarily in that order).

The **kwargs are passed on to the ‘install’ callable

The default install callable accepts two additional arguments:

purge_dst_dir (default False): if True, ALL content will be deleted from the app_install_dir

exclude_from_purge (default None): list of paths to exclude from purge

DANGER: Only set purge_dst_dir=True if your app is installed in its own separate directory, otherwise this will cause unrelated files and folders to be deleted.

check_for_updates(pre: str | None = None, patch: bool = True, ignore_required: bool = False) → TargetMeta | None

Check if any updates are available, based on current app version.

Returns latest archive meta, if a new archive is found.

Final releases are always included. Pre-releases are excluded by default. If pre is specified, pre-releases are included, down to the specified level. Pre-release identifiers follow the PEP440 specification, i.e. ‘a’, ‘b’, or ‘rc’, for alpha, beta, and release candidate, respectively.

If patch is False, a full update is enforced.

If a new release is marked as “required” (in its custom metadata) this release will take precedence over any non-required releases, even if the latter are newer. This may be useful e.g. in case of a configuration change. These “required” releases should be rare, and should preferably be avoided. However, in the exceedingly rare event that there are “required” updates, yet the user wants to treat them as non-required, they can specify ignore_required=True.

download_target(targetinfo: TargetFile, filepath: str | None = None, target_base_url: str | None = None) → str

Download the target file specified by targetinfo.

Args:

targetinfo: TargetFile from get_targetinfo(). filepath: Local path to download into. If None, the file is

downloaded into directory defined by target_dir constructor argument using a generated filename. If file already exists, it is overwritten.

target_base_url: Base URL used to form the final target: download URL. Default is the value provided in Updater()

Raises:

ValueError: Invalid arguments DownloadError: Download of the target file failed in some way RepositoryError: Downloaded target failed to be verified in some way OSError: Failed to write target to file

Returns:

Local path to downloaded file

find_cached_target(targetinfo: TargetFile, filepath: str | None = None) → str | None

Check whether a local file is an up to date target.

Args:: targetinfo: TargetFile from get_targetinfo(). filepath: Local path to file. If None, a file path is

generated based on target_dir constructor argument.
Raises:: ValueError: Incorrect arguments
Returns:: Local file path if the file is an up to date target file. None if file is not found or it is not up to date.

refresh() → None

Refresh top-level metadata.

Downloads, verifies, and loads metadata for the top-level roles in the specified order (root -> timestamp -> snapshot -> targets) implementing all the checks required in the TUF client workflow.

A refresh() can be done only once during the lifetime of an Updater. If refresh() has not been explicitly called before the first get_targetinfo() call, it will be done implicitly at that time.

The metadata for delegated roles is not updated by refresh(): that happens on demand during get_targetinfo(). However, if the repository uses consistent_snapshot, then all metadata downloaded by the Updater will use the same consistent repository state.

Raises:: OSError: New metadata could not be written to disk RepositoryError: Metadata failed to verify in some way DownloadError: Download of a metadata file failed in some way

class tufup.client.AuthRequestsFetcher(session_auth: Dict[str, Tuple[str, str] | AuthBase] | None = None)

__init__(session_auth: Dict[str, Tuple[str, str] | AuthBase] | None = None) → None

This extends the default tuf RequestsFetcher, so we can specify authentication tuples (or custom authentication objects) for each session.

session_auth (optional):

dict of the form {<scheme and server>: (<username>, <password>), …} or {<scheme and server>: <requests.auth.AuthBase>, …} or some combination of those

where <scheme and server> can be e.g. https://example.com or http://localhost:8000.

Note: <scheme and server> must not have a trailing slash

Naming follows [RFC 2396][1], which defines a generic uri as:

<scheme>://<authority><path>?<query>

where <authority> can be <server>.

Also see session authentication example in requests docs: [1][2][3]

[1]: https://datatracker.ietf.org/doc/html/rfc2396#section-3 [2]: https://docs.python-requests.org/en/master/user/advanced/#session-objects [3]: https://docs.python-requests.org/en/latest/user/advanced/#custom-authentication [4]: https://docs.python-requests.org/en/master/api/#sessionapi

attach_progress_hook(hook: Callable, bytes_expected: int)

Allow clients to attach a progress hook which gets called after every downloaded chunk.

The hook must accept two kwargs: bytes_downloaded and bytes_expected

download_bytes(url: str, max_length: int) → bytes

Download bytes from given url.

Returns the downloaded bytes, otherwise like download_file().

Args:

url: URL string that represents the location of the file. max_length: Upper bound of data size in bytes.

Raises:

exceptions.DownloadError: An error occurred during download. exceptions.DownloadLengthMismatchError: Downloaded bytes exceed

max_length.

exceptions.DownloadHTTPError: An HTTP error code was received.

Returns:

Content of the file in bytes.

download_file(url: str, max_length: int) → Iterator[IO]

Download file from given url.

It is recommended to use download_file() within a with block to guarantee that allocated file resources will always be released even if download fails.

Args:

url: URL string that represents the location of the file. max_length: Upper bound of file size in bytes.

Raises:

exceptions.DownloadError: An error occurred during download. exceptions.DownloadLengthMismatchError: Downloaded bytes exceed

max_length.

exceptions.DownloadHTTPError: An HTTP error code was received.

Yields:

TemporaryFile object that points to the contents of url.

fetch(url: str) → Iterator[bytes]

Fetch the contents of HTTP/HTTPS url from a remote server.

Args:: url: URL string that represents a file location.
Raises:: exceptions.DownloadError: An error occurred during download. exceptions.DownloadHTTPError: An HTTP error code was received.
Returns:: Bytes iterator

tufup.repo:

tufup.repo.in_(days: float) → datetime: Returns a timestamp for the specified number of days from now.

filename_pattern = '{key_name}'

__init__(dir_path: Path | str | None = None, encrypted: List[str] | None = None, key_map: RolesDict | None = None, thresholds: RolesDict | None = None): dir_path: directory where all key files are stored encrypted: names of the keys that are (to be) encrypted key_map: maps top-level role names to lists of key names

import_all_public_keys()

import_public_key(role_name: str, key_name: str | None = None): Import public key for specified role.

create()

static create_key_pair(private_key_path: Path, encrypted: bool) → Path

private_key_path(key_name: str) → Path

public_key_path(key_name: str) → Path

public()

roles()

classmethod find_private_key(key_name: str, key_dirs: List[Path | str]) → Path | None

recursively search key_dirs for a private key with specified key_name

returns path to first matching file (or None)

tufup.repo.make_gztar_archive(src_dir: Path | str, dst_dir: Path | str, app_name: str, version: str, tar_format: int = 2) → TargetMeta | None

Create a gzipped tar archive in the dst_dir, based on content of src_dir.

The PAX_FORMAT is currently the default tar format [1] used by the tarfile module. For improved portability [2] and reproducibility [3], this can be changed e.g. to USTAR_FORMAT.

[1]: https://www.gnu.org/software/tar/manual/html_node/Formats.html#Formats [2]: https://www.gnu.org/software/tar/manual/html_node/Portability.html#Portability [3]: https://www.gnu.org/software/tar/manual/html_node/Reproducibility.html#Reproducibility

High-level tools for repository management.

config_filename = '.tufup-repo-config'

__init__(app_name: str, app_version_attr: str | None = None, repo_dir: Path | str | None = None, keys_dir: Path | str | None = None, key_map: RolesDict | None = None, encrypted_keys: List[str] | None = None, expiration_days: RolesDict | None = None, thresholds: RolesDict | None = None, binary_diff: type[BinaryDiff] | None = None)

property config_dict: dict to be saved to configuration file.

property metadata_dir: Path

property targets_dir: Path

property app_version: str

classmethod get_config_file_path() → Path

save_config(): Save current configuration.

classmethod load_config() → dict: Load configuration dict from file.

classmethod from_config(): Create Repository instance from configuration file.

initialize(extra_key_dirs: List[Path] | None = None)

Initialize (or update) the local repository.

This includes:

create directories if they do not exist
import public keys from existing files, or create new key pairs
import roles from existing metadata files
create root metadata file if it does not exist

Safe to call for existing keys and roles.

refresh_expiration_date(role_name: str, days: int | None = None)

replace_key(old_key_name: str, new_public_key_path: Path | str, new_private_key_encrypted: bool)

Replace an existing key by a new one, e.g. after a key compromise.

Note the changes are not published yet: call publish_changes() for that.

add_key(role_name: str, public_key_path: Path | str, encrypted: bool)

Register a new public key for the specified role.

Note the changes are not published yet: call publish_changes() for that.

add_bundle(new_bundle_dir: Path | str, new_version: str | None = None, skip_patch: bool = False, custom_metadata: dict | None = None, required: bool = False)

Adds a new application bundle to the local repository.

An archive file is created from the app bundle, and this file is added to the tuf repository. If a previous archive version is found, a patch file is also created and added to the repository, unless skip_patch is True.

Optional custom_metadata can be specified as a dictionary.

If required=True (default is False), this release will always be installed, even if newer releases are available. For example, suppose an app is running at version 1.0, and version 2.0 is required, but version 3.0 is also available, then tufup will first update to version 2.0, before updating to 3.0 on the next run.

Note the changes are not published yet: call publish_changes() for that.

remove_latest_bundle()

Removes the latest app bundle from the local repository.

This deletes the bundle’s archive file and corresponding patch file from the targets directory, and updates the tuf repository metadata accordingly.

Note the changes are not published yet: call publish_changes() for that

publish_changes(private_key_dirs: List[Path | str])

Publish all modified roles. That is, if a role has changed w.r.t. to the version on disk:

update expiration date (if not yet updated)
bump version (if not yet bumped)
sign
save to disk

If a role has not been modified, it is skipped.

threshold_sign(role_name: str, private_key_dirs: List[Path | str]) → int

Sign the metadata file for a specific role, and save changes to disk.

Use this to sign and save without making any changes to the actual signed metadata.

In case of root key rotation, both the old private key and the new private key are required.

Returns the number of signatures created.

class tufup.repo.Roles(dir_path: Path | str | None = None)

filename_pattern = '{version}{role_name}{suffix}'

__init__(dir_path: Path | str | None = None)

TUF roles

root metadata tells us:
- all the known keys (key id and public key value)
- which keys belong to each role
- how many signatures are needed for each role
targets metadata tells us:
- which target files are available (filename, size, hash)
snapshots metatadata tells us:
- which version of the targets-metadata file to expect
timestamp metadata tells us:
- which version of the snapshot-metadata file to expect

initialize(keys: Keys, expiration_days: RolesDict | None = None)

add_or_update_target(local_path: Path | str, url_path_segments: List[str] | None = None, custom: CustomMetadataDict | None = None)

remove_target(local_path: Path | str) → bool

add_public_key(role_name: str, public_key_path: Path | str): Import a public key from file and add it to the specified role.

set_signature_threshold(role_name: str, threshold: int)

set_expiration_date(role_name: str, days: int)

sign_role(role_name: str, private_key_path: Path | str)

Sign role using specified private key.

We sign off on the role.signed part, and the signature is added to the role.signatures list.

file_path(role_name: str, version: int | None = None)

file_exists(role_name: str): Return True if any metadata file exists for the specified role, ignoring any versions in the filename.

persist_role(role_name: str)

Save specified role to corresponding metadata file.

In case of root, make sure “root.json” always represents the latest version (in addition to x.root.json).

get_latest_archive() → TargetMeta | None

Returns TargetMeta for latest archive.

Note that all pre-release versions are always included: On the repo side, there is no difference between final releases an pre-releases. Pre-release specifiers are only used on the Client side, to filter available updates).

tufup.common:

class tufup.common.CustomMetadataDict

explicitly separate custom metadata into user-specified metadata and metadata used by tufup internally

user: dict

tufup: dict

__init__(*args, **kwargs)

clear() → None. Remove all items from D.

copy() → a shallow copy of D

fromkeys(value=None, /): Create a new dictionary with keys from iterable and values set to value.

get(key, default=None, /): Return the value for key if key is in the dictionary, else default.

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

pop(k[, d]) → v, remove specified key and return the corresponding value.: If key is not found, default is returned if given, otherwise KeyError is raised

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

setdefault(key, default=None, /)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update([E, ]**F) → None. Update D from dict/iterable E and F.: If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values() → an object providing a view on D's values

filename_pattern = '{name}-{version}{suffix}'

filename_regex = re.compile('^(?P<name>[\\w-]+)-(?P<version>.+)(?P<suffix>\\.tar\\.gz|\\.patch)$')

Initialize either with target_path, or with name, version, archive.

BEWARE: whitespace is not allowed in the filename, nor in the name or version arguments

property custom: dict: returns user-specified custom metadata dict

property custom_internal: dict: returns tufup-internal custom metadata dict

property filename

property name: str | None: The app name.

property version: Version | None

property suffix: str | None: Returns the filename suffix, either ‘.tar.gz’, ‘.patch’, or None.

property is_archive: bool

property is_patch: bool

property is_other: bool

classmethod parse_filename(filename: str) → dict

Parse a filename to extract app name, version, and suffix.

We do not impose any versioning requirements yet, such as defined in packaging.version.VERSION_PATTERN.

classmethod compose_filename(name: str, version: str, is_archive: bool)

class tufup.common.BinaryDiff

BinaryDiff represents an interface for overriding the binary diff/patch functions

abstract static diff(*, src_bytes: bytes, dst_bytes: bytes) → bytes: Create patch as the binary difference between source and destination data

abstract static patch(*, src_bytes: bytes, patch_bytes: bytes) → bytes: Apply binary patch to source data to recover destination data

class tufup.common.DefaultBinaryDiff

The default implementation of binary differencing and patching functions

diff(src_bytes, dst_bytes) → bytes: Return a BSDIFF4-format patch (from src_bytes to dst_bytes) as bytes.

patch(src_bytes, patch_bytes) → bytes: Apply the BSDIFF4-format patch_bytes to src_bytes and return the bytes.

class tufup.common.Patcher

DEFAULT_HASH_ALGORITHM = 'sha256'

classmethod diff_and_hash(src_path: Path, dst_path: Path, patch_path: Path, binary_diff: type[BinaryDiff] | None = None) → dict

Creates a patch file from the binary difference between source and destination .tar archives. The source and destination files are expected to be gzip-compressed tar archives (.tar.gz).

The binary differencing method can be customized by implementing a BinaryDiff subclass and passing this in via the binary_diff argument.

Returns a dict with size and hash of the uncompressed destination archive.

classmethod patch_and_verify(src_path: Path, dst_path: Path, patch_targets: Dict[TargetMeta, Path], binary_diff: type[BinaryDiff] | None = None) → None

Applies one or more binary patch files to a source file in order to reconstruct a destination file.

Source file and destination file are gzip-compressed tar archives, but the patches are applied to the uncompressed tar archives. The reason is that small changes in uncompressed data can cause (very) large differences in gzip compressed data, leading to excessively large patch files (see #69).

The integrity of the patched .tar archive is verified using expected length and hash (from custom tuf metadata), similar to python-tuf’s download verification. If the patched archive fails this check, the destination file is not written.

The binary patching method can be customized by implementing a BinaryDiff subclass and passing this in via the binary_diff argument.