repository_service_tuf_worker package

Subpackages

• repository_service_tuf_worker.models package
  • Subpackages
    • repository_service_tuf_worker.models.targets package
      • Submodules
      • repository_service_tuf_worker.models.targets.crud module
      • repository_service_tuf_worker.models.targets.models module
      • repository_service_tuf_worker.models.targets.schemas module
      • Module contents
  • Module contents
    • rstuf_db()
• repository_service_tuf_worker.services package
  • Subpackages
    • repository_service_tuf_worker.services.keyvault package
      • Submodules
      • repository_service_tuf_worker.services.keyvault.local module
      • Module contents
    • repository_service_tuf_worker.services.storage package
      • Submodules
      • repository_service_tuf_worker.services.storage.awss3 module
      • repository_service_tuf_worker.services.storage.local module
      • Module contents
  • Module contents

Submodules

repository_service_tuf_worker.interfaces module

class repository_service_tuf_worker.interfaces.IKeyVault

Bases: ABC

abstract classmethod configure(settings: LazySettings) → IKeyVault

Run actions to verify, configure and create object using the settings.

classmethod from_dynaconf(settings: LazySettings) → None

Run actions to verify and configure using the settings.

abstract get(public_key: Key) → Signer

Return a signer using the online key.

abstract classmethod settings() → List[ServiceSettings]

Define all the ServiceSettings required in settings.

class repository_service_tuf_worker.interfaces.IStorage

Bases: ABC

abstract classmethod configure(settings: LazySettings) → IStorage

Run actions to verify, configure and create object using the settings.

classmethod from_dynaconf(settings: LazySettings) → None

Run actions to verify and configure using the dynaconf settings.

abstract get(rolename: str, version: Optional[int]) → Metadata[T]

Return metadata from specific role name, optionally specific version (latest if None).

abstract put(file_data: bytes, filename: str) → None

Stores file bytes within a file with a specific filename.

abstract classmethod settings() → List[ServiceSettings]

Define all the ServiceSettings required in settings.

class repository_service_tuf_worker.interfaces.ServiceSettings(names: List[str], required: bool, default: Optional[Any] = None)

Bases: object

Dataclass for service settings.

default: Optional[Any] = None

names: List[str]

required: bool

repository_service_tuf_worker.repository module

class repository_service_tuf_worker.repository.MetadataRepository

Bases: object

A repository service to create and maintain TUF role metadata.

add_targets(payload: Dict[str, Any], update_state: update_state) → Optional[Dict[str, Any]]

Add or update the new target in the SQL DB and submit the task for update_targets

Check the target(s) in the SQL DB; if it doesn’t exist, create a new entry or update it as not published. After changing the SQL DB submit a new publish_target task. This function will wait until all the targets are published.

bootstrap(payload: Dict[str, Any], update_state: Optional[update_state] = None) → Dict[str, Any]

Bootstrap the Metadata Repository

bump_online_roles(force: Optional[bool] = False) → bool

Bump online roles (Snapshot, Timestamp, Targets and BINS).

Args:

force: force target roles bump if they don’t match the hours before

expire (self._hours_before_expire)

bump_snapshot(force: Optional[bool] = False)

Bumps version and expiration date of TUF ‘snapshot’ role metadata.

The version number is incremented by one, the expiration date renewed using a configured expiration interval, and the metadata is signed and persisted using the configured key and storage services.

Updating ‘snapshot’ also updates ‘timestamp’.

Args:

force: force snapshot bump if it doesn’t match the hours before

expire (self._hours_before_expire)

classmethod create_service() → MetadataRepository

Class Method for MetadataRepository service creation.

delete_sign_metadata(payload: Dict[str, Any], update_state: Optional[update_state] = None) → Dict[str, Any]

metadata_rotation(payload: Dict[Literal['metadata'], Dict[Literal['root'], Any]], update_state: Optional[update_state] = None) → Dict[str, Any]

metadata_update(payload: Dict[Literal['metadata'], Dict[Literal['root'], Any]], update_state: Optional[update_state] = None) → Dict[str, Any]

Update TUF metadata.

Args:

payload: contains new metadata

Supported metadata types: Root example: {“metadata”: {“root”: Any}}

update_state: not used, but required argument by app.py

publish_targets(payload: Optional[Dict[str, Any]] = None, update_state: Optional[update_state] = None)

Publish targets as persistent TUF Metadata in the backend storage, updating Snapshot and Timestamp.

refresh_settings(worker_settings: Optional[LazySettings] = None)

Refreshes the MetadataRepository settings.

remove_targets(payload: Dict[str, Any], update_state: update_state) → Dict[str, Any]

Remove targets from the metadata roles.

save_settings(root: Metadata[Root], settings: Dict[str, Any])

Save settings to the repository settings.

Args:

root: Root metadata settings: payload settings

sign_metadata(payload: Dict[str, Any], update_state: Optional[update_state] = None) → Dict[str, Any]

Add signature to metadata for pending signing event.

Add signature (from payload) to cached root metadata (from settings), if a signing event exists, and the signature is valid.

Signing event types are ‘bootstrap’ or ‘metadata update’.

If the signature threshold is reached, the signing event is finalized, otherwise it remains in pending state.

update_settings(payload: Dict[str, Any], update_state: Optional[update_state] = None) → Dict[str, Any]

Update repository settings with the new settings.

Supports only updating the expiration policy settings for online roles. Expiration parameters reference: https://repository-service-tuf.readthedocs.io/en/stable/devel/design.html#tuf-repository-settings # noqa

write_repository_settings(key: str, value: Any)

Writes repository settings.

Repository settings are stored in a dictionary like Dynaconf object and each of them has its own key and value. Additionally, repository settings are persisted in the Redis server so that they can be reused by multiple RSTUF Worker instances.

https://repository-service-tuf.readthedocs.io/en/latest/devel/design.html#rstuf-repository-settings-configuration # noqa

Args:

key: key name

value: value for the key

class repository_service_tuf_worker.repository.Roles(value)

Bases: Enum

An enumeration.

BINS = 'bins'

ROOT = 'root'

SNAPSHOT = 'snapshot'

TARGETS = 'targets'

TIMESTAMP = 'timestamp'

static online_roles() → List[str]

class repository_service_tuf_worker.repository.TaskName(value)

Bases: str, Enum

An enumeration.

ADD_TARGETS = 'add_targets'

BOOTSTRAP = 'bootstrap'

DELETE_SIGN_METADATA = 'delete_sign_metadata'

METADATA_UPDATE = 'metadata_update'

PUBLISH_TARGETS = 'publish_targets'

REMOVE_TARGETS = 'remove_targets'

SIGN_METADATA = 'sign_metadata'

UPDATE_SETTINGS = 'update_settings'

class repository_service_tuf_worker.repository.TaskResult(task: repository_service_tuf_worker.repository.TaskName, status: bool, last_update: datetime.datetime, message: str, error: Optional[str], details: Optional[Dict[str, Any]])

Bases: object

details: Optional[Dict[str, Any]]

error: Optional[str]

last_update: datetime

message: str

status: bool

task: TaskName

repository_service_tuf_worker.signer module

class repository_service_tuf_worker.signer.SignerStore(settings: LazySettings)

Bases: object

Generic signer store.

get(key: Key) → Signer

Return signer for passed key.

Module contents

repository_service_tuf_worker.get_repository_settings() → LazySettings

repository_service_tuf_worker.get_worker_settings() → LazySettings

repository_service_tuf_worker.parse_if_secret(env_var: str) → str