1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
|
# 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
|
