# Copyright 2021-2022 python-tuf contributors

# SPDX-License-Identifier: MIT OR Apache-2.0

"""Simple example of using the repository library to build a repository"""

from __future__ import annotations

import copy

import json

import logging

from collections import defaultdict

from datetime import datetime, timedelta, timezone

from securesystemslib.signer import CryptoSigner, Key, Signer

from tuf.api.exceptions import RepositoryError

from tuf.api.metadata import (

DelegatedRole,

Delegations,

Metadata,

MetaFile,

Root,

RootVerificationResult,

Signed,

Snapshot,

TargetFile,

Targets,

Timestamp,

VerificationResult,

)

from tuf.repository import Repository

logger = logging.getLogger(__name__)

_signed_init = {

Root.type: Root,

Snapshot.type: Snapshot,

Targets.type: Targets,

Timestamp.type: Timestamp,

}

class SimpleRepository(Repository):

"""Very simple in-memory repository implementation

This repository keeps the metadata for all versions of all roles in memory.

It also keeps all target content in memory.

Attributes:

role_cache: Every historical metadata version of every role in this

repository. Keys are role names and values are lists of Metadata

signer_cache: All signers available to the repository. Keys are role

names, values are lists of signers

target_cache: All target files served by the repository. Keys are

target paths and values are file contents as bytes.

"""

expiry_period = timedelta(days=1)

def __init__(self) -> None:

# all versions of all metadata

self.role_cache: dict[str, list[Metadata]] = defaultdict(list)

# all current keys

self.signer_cache: dict[str, list[Signer]] = defaultdict(list)

# all target content

self.target_cache: dict[str, bytes] = {}

# version cache for snapshot and all targets, updated in close().

# The 'defaultdict(lambda: ...)' trick allows close() to easily modify

# the version without always creating a new MetaFile

self._snapshot_info = MetaFile(1)

self._targets_infos: dict[str, MetaFile] = defaultdict(

lambda: MetaFile(1)

)

# setup a basic repository, generate signing key per top-level role

with self.edit_root() as root:

for role in ["root", "timestamp", "snapshot", "targets"]:

signer = CryptoSigner.generate_ecdsa()

self.signer_cache[role].append(signer)

root.add_key(signer.public_key, role)

for role in ["timestamp", "snapshot", "targets"]:

with self.edit(role):

pass

@property

def targets_infos(self) -> dict[str, MetaFile]:

return self._targets_infos

@property

def snapshot_info(self) -> MetaFile:

return self._snapshot_info

def _get_verification_result(

self, role: str, md: Metadata

) -> VerificationResult | RootVerificationResult:

"""Verify roles metadata using the existing repository metadata"""

if role == Root.type:

assert isinstance(md.signed, Root)

root = self.root()

previous = root if root.version > 0 else None

return md.signed.get_root_verification_result(

previous, md.signed_bytes, md.signatures

)

if role in [Timestamp.type, Snapshot.type, Targets.type]:

delegator: Signed = self.root()

else:

delegator = self.targets()

return delegator.get_verification_result(

role, md.signed_bytes, md.signatures

)

def open(self, role: str) -> Metadata:

"""Return current Metadata for role from 'storage'

(or create a new one)

"""

if role not in self.role_cache:

signed_init = _signed_init.get(role, Targets)

md = Metadata(signed_init())

# this makes version bumping in close() simpler

md.signed.version = 0

return md

# return latest metadata from storage (but don't return a reference)

return copy.deepcopy(self.role_cache[role][-1])

def close(self, role: str, md: Metadata) -> None:

"""Store a version of metadata. Handle version bumps, expiry, signing"""

md.signed.version += 1

md.signed.expires = datetime.now(timezone.utc) + self.expiry_period

md.signatures.clear()

for signer in self.signer_cache[role]:

md.sign(signer, append=True)

# Double check that we only write verified metadata

vr = self._get_verification_result(role, md)

if not vr:

raise ValueError(f"Role {role} failed to verify")

keyids = [keyid[:7] for keyid in vr.signed]

verify_str = f"verified with keys [{', '.join(keyids)}]"

logger.debug("Role %s v%d: %s", role, md.signed.version, verify_str)

# store new metadata version, update version caches

self.role_cache[role].append(md)

if role == "snapshot":

self._snapshot_info.version = md.signed.version

elif role not in ["root", "timestamp"]:

self._targets_infos[f"{role}.json"].version = md.signed.version

def add_target(self, path: str, content: str) -> None:

"""Add a target to top-level targets metadata"""

data = bytes(content, "utf-8")

# add content to cache for serving to clients

self.target_cache[path] = data

# add a target in the targets metadata

with self.edit_targets() as targets:

targets.targets[path] = TargetFile.from_data(path, data)

# update snapshot, timestamp

self.do_snapshot()

self.do_timestamp()

def submit_delegation(self, rolename: str, data: bytes) -> bool:

"""Add a delegation to a (offline signed) delegated targets metadata"""

try:

logger.debug("Processing new delegation to role %s", rolename)

keyid, keydict = next(iter(json.loads(data).items()))

key = Key.from_dict(keyid, keydict)

# add delegation and key

role = DelegatedRole(rolename, [], 1, True, [f"{rolename}/*"])

with self.edit_targets() as targets:

if targets.delegations is None:

targets.delegations = Delegations({}, {})

if targets.delegations.roles is None:

targets.delegations.roles = {}

targets.delegations.roles[rolename] = role

targets.add_key(key, rolename)

except (RepositoryError, json.JSONDecodeError) as e:

logger.info("Failed to add delegation for %s: %s", rolename, e)

return False

# update snapshot, timestamp

self.do_snapshot()

self.do_timestamp()

return True

def submit_role(self, role: str, data: bytes) -> bool:

"""Add a new version of a delegated roles metadata"""

try:

logger.debug("Processing new version for role %s", role)

if role in ["root", "snapshot", "timestamp", "targets"]:

raise ValueError("Only delegated targets are accepted")

md = Metadata.from_bytes(data)

for targetpath in md.signed.targets:

if not targetpath.startswith(f"{role}/"):

raise ValueError(f"targets allowed under {role}/ only")

if md.signed.version != self.targets(role).version + 1:

raise ValueError("Invalid version {md.signed.version}")

except (RepositoryError, ValueError) as e:

logger.info("Failed to add new version for %s: %s", role, e)

return False

# Check that we only write verified metadata

vr = self._get_verification_result(role, md)

if not vr:

logger.info("Role %s failed to verify", role)

return False

keyids = [keyid[:7] for keyid in vr.signed]

verify_str = f"verified with keys [{', '.join(keyids)}]"

logger.debug("Role %s v%d: %s", role, md.signed.version, verify_str)

# Checks passed: Add new delegated role version

self.role_cache[role].append(md)

self._targets_infos[f"{role}.json"].version = md.signed.version

# To keep it simple, target content is generated from targetpath

for targetpath in md.signed.targets:

self.target_cache[targetpath] = bytes(f"{targetpath}", "utf-8")

# update snapshot, timestamp

self.do_snapshot()

self.do_timestamp()

return True