Module rugged.workers.root-worker

Expand source code 
from rugged.lib.config import get_config
from rugged.lib.logger import (
    get_rugged_logger,
    set_log_level_from_context,
)
from rugged.lib.task_queue import TaskQueue
from rugged.tuf.generate_keys import generate_keypair
from rugged.tuf.init_repo import init_repo
from rugged.workers.base_worker import BaseWorker

config = get_config()
log = get_rugged_logger()
worker = TaskQueue().get_task_queue()
queue = 'root-worker'
name = 'Root Worker'


class RootWorker(BaseWorker):
    """
    Rugged (Celery) worker that fulfills the TUF 'root' role.

    N.B. This worker has access to the root signing key.

    This worker should not be active during normal operations of the Rugged
    system. It should only be brought online by administrators, to perform
    'root' functions (eg. key rotation). Otherwise, it should be completely
    removed from the system.
    """

    @worker.task(name='initialize_task', queue=queue)
    def initialize_task(**context):
        """ Initialize a new TUF repository. """
        set_log_level_from_context(context)
        log.debug("Received 'initialize' task.")
        result, message = init_repo()
        if result:
            log.info(message)
        else:
            log.error(message)
        return (result, message)

    @worker.task(name='generate_keypair_task', queue=queue)
    def generate_keypair_task(key, role, **context):
        """ Generate keypairs for use in a TUF repository. """
        set_log_level_from_context(context)
        log.debug("Received 'generate_keypair' task.")
        log.info(f"Generating '{key}' keypair for '{ role }' role.")
        return generate_keypair(key, role)

Classes

class RootWorker

Rugged (Celery) worker that fulfills the TUF 'root' role.

N.B. This worker has access to the root signing key.

This worker should not be active during normal operations of the Rugged system. It should only be brought online by administrators, to perform 'root' functions (eg. key rotation). Otherwise, it should be completely removed from the system.

Expand source code 
class RootWorker(BaseWorker):
    """
    Rugged (Celery) worker that fulfills the TUF 'root' role.

    N.B. This worker has access to the root signing key.

    This worker should not be active during normal operations of the Rugged
    system. It should only be brought online by administrators, to perform
    'root' functions (eg. key rotation). Otherwise, it should be completely
    removed from the system.
    """

    @worker.task(name='initialize_task', queue=queue)
    def initialize_task(**context):
        """ Initialize a new TUF repository. """
        set_log_level_from_context(context)
        log.debug("Received 'initialize' task.")
        result, message = init_repo()
        if result:
            log.info(message)
        else:
            log.error(message)
        return (result, message)

    @worker.task(name='generate_keypair_task', queue=queue)
    def generate_keypair_task(key, role, **context):
        """ Generate keypairs for use in a TUF repository. """
        set_log_level_from_context(context)
        log.debug("Received 'generate_keypair' task.")
        log.info(f"Generating '{key}' keypair for '{ role }' role.")
        return generate_keypair(key, role)

Ancestors

Methods

def generate_keypair_task(key, role, **context)

Generate keypairs for use in a TUF repository.

Expand source code 
@worker.task(name='generate_keypair_task', queue=queue)
def generate_keypair_task(key, role, **context):
    """ Generate keypairs for use in a TUF repository. """
    set_log_level_from_context(context)
    log.debug("Received 'generate_keypair' task.")
    log.info(f"Generating '{key}' keypair for '{ role }' role.")
    return generate_keypair(key, role)
def initialize_task(**context)

Initialize a new TUF repository.

Expand source code 
@worker.task(name='initialize_task', queue=queue)
def initialize_task(**context):
    """ Initialize a new TUF repository. """
    set_log_level_from_context(context)
    log.debug("Received 'initialize' task.")
    result, message = init_repo()
    if result:
        log.info(message)
    else:
        log.error(message)
    return (result, message)

Inherited members