Module rugged.tuf.target_files

Functions

def delete_empty_target_dirs(root_dir: str, target: str) ‑> None

Expand source code

def delete_empty_target_dirs(root_dir: str, target: str) -> None:
    """ Delete any intermediate (empty) directories for a target path. """
    if root_dir not in target:
        target = path.join(root_dir, target)
    target_dir = path.dirname(target)
    if target_dir == root_dir:
        return   # This target is the root directory, so stop.
    try:
        if listdir(target_dir):
            return   # We're only cleaning up empty directories.
        rmdir(target_dir)
        log.debug(f"Cleaned up empty directory '{target_dir}'.")
    except Exception as e:
        if config['delete_targets_after_signing'].get():
            # It's worth trying to delete the directory, even if this
            # config is set, in case it was configured after initializing
            # the repo. So there may be cruft. But if not, we can ignore
            # the error. See: https://gitlab.com/rugged/rugged/-/issues/179
            return
        log_exception(e)
        log.warning(f"Failed to clean up empty directory '{target_dir}'.")
    finally:
        # Recurse until we hit the root directory.
        delete_empty_target_dirs(root_dir, target_dir)

Delete any intermediate (empty) directories for a target path.

def delete_removed_target(removed_target: str) ‑> None

Expand source code

def delete_removed_target(removed_target: str) -> None:
    """ Delete the file for the target that we removed from the repo. """
    repo_targets_path = config['repo_targets_path'].get()
    target_file = path.join(repo_targets_path, removed_target)
    try:
        remove(target_file)
        log.info(f"Deleted target file '{target_file}'.")
    except Exception as e:
        if config['delete_targets_after_signing'].get():
            # It's worth trying to delete the file, even if this config is
            # set, in case it was configured after initializing the repo.
            # So there may be cruft. But if not, we can ignore the error.
            # See: https://gitlab.com/rugged/rugged/-/issues/179
            return
        log_exception(e)
        log.warning(f"Failed to delete target file '{target_file}'.")
    finally:
        delete_empty_target_dirs(repo_targets_path, removed_target)

Delete the file for the target that we removed from the repo.

def delete_target_after_signing(target: str) ‑> None

Expand source code

def delete_target_after_signing(target: str) -> None:
    """ Delete a given target file after it has been signed. """
    try:
        remove(target)
        log.debug(f"Deleted '{target}' after signing.")
    except Exception as e:
        log_exception(e)
        log.warning(f"Failed to delete target '{target}' after signing.")

Delete a given target file after it has been signed.

def get_inbound_targets() ‑> List[str]

Expand source code

def get_inbound_targets() -> List[str]:
    """ Scan the inbound directory for files to add to the repository. """
    inbound_targets_dir = config['inbound_targets_path'].get()
    log.debug(f"Scanning for inbound targets in '{inbound_targets_dir}'")
    chdir(inbound_targets_dir)
    inbound_targets = []
    for inbound_target in glob('**', recursive=True):
        if path.isdir(inbound_target):
            # We only want files, not intermediate directories.
            continue
        if inbound_target.startswith(RUGGED_MONITOR_TUF_PROCESSING_PREFIX):
            # Do not process targets that are still processing.
            continue
        if inbound_target.startswith(RUGGED_MONITOR_TUF_READY_PREFIX):
            # Do not process ready targets either.
            continue
        log.debug(f"Found target: {inbound_target}")
        inbound_targets.append(inbound_target)
    return inbound_targets

Scan the inbound directory for files to add to the repository.

def move_inbound_target_to_targets_dir(inbound_target: str) ‑> str

Expand source code

def move_inbound_target_to_targets_dir(inbound_target: str) -> str:
    """ Move an inbound target to the repo targets directory. """
    inbound_targets_path = config['inbound_targets_path'].get()
    inbound_target_path = path.join(inbound_targets_path, inbound_target)
    moved_target_path = path.join(
        config['repo_targets_path'].get(),
        inbound_target,
    )
    try:
        makedirs(path.dirname(moved_target_path), exist_ok=True)
        move(inbound_target_path, moved_target_path)
        log.debug(f"Moved '{inbound_target_path}' to '{moved_target_path}'")
    except Exception as e:
        log_exception(e)
        log.warning(f"Failed to move target '{inbound_target}' to the targets directory.")
    delete_empty_target_dirs(inbound_targets_path, inbound_target)
    log.info(f"Moved inbound target '{inbound_target}' to targets directory.")
    return moved_target_path

Move an inbound target to the repo targets directory.