Skip to content

Commit 0baff95

Browse files
author
Jussi Kukkonen
committed
Metadata API: Make Metadata Generic
The purpose is two-fold: 1. When we deserialize metadata, we usually know what signed type we expect: make it easy to enforce that 2. When we use Metadata, it is helpful if the specific signed type (and all of the classes attribute types are correctly annotated Making Metadata Generic over T, where T = TypeVar("T", "Root", "Timestamp", "Snapshot", "Targets") allows both of these cases to work. Using Generics is completely optional so all existing code still works. For case 1, the following calls will now raise a Deserialization error if the expected type is incorrect: md = Metadata.from_bytes(data, signed_type=Snapshot) md = Metadata.from_file(filename, signed_type=Snapshot) For case 2, the return value md of those calls is now of type "Metadata[Snapshot]", and md.signed is now of type "Snapshot" allowing IDE annotations and static type checking. Adding a type argument is an unconventional way to do this: the reason for it is that the specific type (e.g. Snapshot) is not otherwise available at runtime. A call like this works fine and md is annotated: md = Metadata[Snapshot].from_bytes(data) but it's not possible to validate that "data" contains a "Snapshot", because the value "Snapshot" is not defined at runtime at all, it is purely an annotation. So an actual argument is needed. Fixes #1433 Signed-off-by: Jussi Kukkonen <jkukkonen@vmware.com>
1 parent 39ed706 commit 0baff95

2 files changed

Lines changed: 44 additions & 9 deletions

File tree

tests/test_api.py

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,6 +134,23 @@ def test_generic_read(self):
134134
os.remove(bad_metadata_path)
135135

136136

137+
def test_typed_read(self):
138+
path = os.path.join(self.repo_dir, 'metadata', 'root.json')
139+
with open(path, 'rb') as f:
140+
data = f.read()
141+
142+
# Loading a root file as "Metadata[Root]" succeeds
143+
md = Metadata.from_bytes(data, signed_type=Root)
144+
md2 = Metadata.from_file(path, signed_type=Root)
145+
146+
# Loading the file fails with non-"Root" type constraints
147+
for expected_type in [Timestamp, Snapshot, Targets]:
148+
with self.assertRaises(DeserializationError):
149+
Metadata.from_bytes(data, signed_type=expected_type)
150+
with self.assertRaises(DeserializationError):
151+
Metadata.from_file(path, signed_type=expected_type)
152+
153+
137154
def test_compact_json(self):
138155
path = os.path.join(self.repo_dir, 'metadata', 'targets.json')
139156
metadata_obj = Metadata.from_file(path)

tuf/api/metadata.py

Lines changed: 27 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -25,12 +25,15 @@
2525
BinaryIO,
2626
ClassVar,
2727
Dict,
28+
Generic,
2829
List,
2930
Mapping,
3031
Optional,
3132
Tuple,
3233
Type,
34+
TypeVar,
3335
Union,
36+
cast,
3437
)
3538

3639
from securesystemslib import hash as sslib_hash
@@ -41,6 +44,7 @@
4144

4245
from tuf import exceptions
4346
from tuf.api.serialization import (
47+
DeserializationError,
4448
MetadataDeserializer,
4549
MetadataSerializer,
4650
SignedSerializer,
@@ -55,8 +59,11 @@
5559
# files to have the same major version (the first number) as ours.
5660
SPECIFICATION_VERSION = ["1", "0", "19"]
5761

62+
# T is a Generic type constraint for Metadata.signed
63+
T = TypeVar("T", "Root", "Timestamp", "Snapshot", "Targets")
5864

59-
class Metadata:
65+
66+
class Metadata(Generic[T]):
6067
"""A container for signed TUF metadata.
6168
6269
Provides methods to convert to and from dictionary, read and write to and
@@ -73,7 +80,8 @@ class Metadata:
7380
def __init__(
7481
self, signed: "Signed", signatures: "OrderedDict[str, Signature]"
7582
):
76-
self.signed = signed
83+
# init does not verify type constraint T: only from_bytes() does
84+
self.signed: T = cast(T, signed)
7785
self.signatures = signatures
7886

7987
@classmethod
@@ -123,13 +131,13 @@ def from_dict(cls, metadata: Dict[str, Any]) -> "Metadata":
123131
signatures=signatures,
124132
)
125133

126-
@classmethod
134+
@staticmethod
127135
def from_file(
128-
cls,
129136
filename: str,
130137
deserializer: Optional[MetadataDeserializer] = None,
131138
storage_backend: Optional[StorageBackendInterface] = None,
132-
) -> "Metadata":
139+
signed_type: Optional[Type[T]] = None,
140+
) -> "Metadata[T]":
133141
"""Loads TUF metadata from file storage.
134142
135143
Arguments:
@@ -140,6 +148,7 @@ def from_file(
140148
storage_backend: An object that implements
141149
securesystemslib.storage.StorageBackendInterface. Per default
142150
a (local) FilesystemBackend is used.
151+
signed_type: Optional; Expected type of deserialized signed object.
143152
144153
Raises:
145154
securesystemslib.exceptions.StorageError: The file cannot be read.
@@ -153,20 +162,22 @@ def from_file(
153162
if storage_backend is None:
154163
storage_backend = FilesystemBackend()
155164

156-
with storage_backend.get(filename) as file_obj:
157-
return cls.from_bytes(file_obj.read(), deserializer)
165+
with storage_backend.get(filename) as f:
166+
return Metadata.from_bytes(f.read(), deserializer, signed_type)
158167

159168
@staticmethod
160169
def from_bytes(
161170
data: bytes,
162171
deserializer: Optional[MetadataDeserializer] = None,
163-
) -> "Metadata":
172+
signed_type: Optional[Type[T]] = None,
173+
) -> "Metadata[T]":
164174
"""Loads TUF metadata from raw data.
165175
166176
Arguments:
167177
data: metadata content as bytes.
168178
deserializer: Optional; A MetadataDeserializer instance that
169179
implements deserialization. Default is JSONDeserializer.
180+
signed_type: Optional; Expected type of deserialized signed object.
170181
171182
Raises:
172183
tuf.api.serialization.DeserializationError:
@@ -183,7 +194,14 @@ def from_bytes(
183194

184195
deserializer = JSONDeserializer()
185196

186-
return deserializer.deserialize(data)
197+
md = deserializer.deserialize(data)
198+
199+
# Ensure deserialized signed type matches the requested type
200+
if signed_type is not None and signed_type != type(md.signed):
201+
raise DeserializationError(
202+
f"Expected {signed_type}, got {type(md.signed)}"
203+
)
204+
return md
187205

188206
def to_dict(self) -> Dict[str, Any]:
189207
"""Returns the dict representation of self."""

0 commit comments

Comments
 (0)