Skip to content

Commit

Permalink
Merge 1f074f8 into e005801
Browse files Browse the repository at this point in the history
  • Loading branch information
@MVrachev
MVrachev committed Nov 25, 2020
2 parents e005801 + 1f074f8 commit d1144c4
Show file tree
Hide file tree
Showing 2 changed files with 186 additions and 66 deletions.
44 changes: 29 additions & 15 deletions tests/test_api.py
View file
Expand Up @@ -33,9 +33,12 @@ def setUpModule():
import tuf.exceptions
from tuf.api.metadata import (
Metadata,
MetaFile,
Root,
Snapshot,
Timestamp,
Targets
Targets,
TargetFile
)

from securesystemslib.interface import (
Expand Down Expand Up @@ -92,6 +95,7 @@ def tearDownClass(cls):

def test_generic_read(self):
for metadata, inner_metadata_cls in [
('root', Root),
('snapshot', Snapshot),
('timestamp', Timestamp),
('targets', Targets)]:
Expand Down Expand Up @@ -137,7 +141,7 @@ def test_compact_json(self):


def test_read_write_read_compare(self):
for metadata in ['snapshot', 'timestamp', 'targets']:
for metadata in ['root', 'snapshot', 'timestamp', 'targets']:
path = os.path.join(self.repo_dir, 'metadata', metadata + '.json')
metadata_obj = Metadata.from_json_file(path)

Expand Down Expand Up @@ -222,13 +226,18 @@ def test_metadata_snapshot(self):
# Create a dict representing what we expect the updated data to be
fileinfo = copy.deepcopy(snapshot.signed.meta)
hashes = {'sha256': 'c2986576f5fdfd43944e2b19e775453b96748ec4fe2638a6d2f32f1310967095'}
fileinfo['role1.json']['version'] = 2
fileinfo['role1.json']['hashes'] = hashes
fileinfo['role1.json']['length'] = 123
fileinfo['role1.json'].version = 2
fileinfo['role1.json'].hashes = hashes
fileinfo['role1.json'].length = 123


self.assertNotEqual(snapshot.signed.meta, fileinfo)
snapshot.signed.update('role1', 2, 123, hashes)
snapshot.signed.update_meta_file('role1', 2, 123, hashes)
self.assertEqual(snapshot.signed.meta, fileinfo)

# Update only version. Length and hashes are optional.
snapshot.signed.update_meta_file('role1', 3)
fileinfo['role1.json'] = MetaFile(3)
self.assertEqual(snapshot.signed.meta, fileinfo)


Expand Down Expand Up @@ -257,14 +266,18 @@ def test_metadata_timestamp(self):
self.assertEqual(timestamp.signed.expires, datetime(2036, 1, 3, 0, 0))

hashes = {'sha256': '0ae9664468150a9aa1e7f11feecb32341658eb84292851367fea2da88e8a58dc'}
fileinfo = copy.deepcopy(timestamp.signed.meta['snapshot.json'])
fileinfo['hashes'] = hashes
fileinfo['version'] = 2
fileinfo['length'] = 520
fileinfo = copy.deepcopy(timestamp.signed.meta)
fileinfo['snapshot.json'].hashes = hashes
fileinfo['snapshot.json'].version = 2
fileinfo['snapshot.json'].length = 520
self.assertNotEqual(timestamp.signed.meta, fileinfo)
timestamp.signed.update_snapshot_meta(2, 520, hashes)
self.assertEqual(timestamp.signed.meta, fileinfo)

self.assertNotEqual(timestamp.signed.meta['snapshot.json'], fileinfo)
timestamp.signed.update(2, 520, hashes)
self.assertEqual(timestamp.signed.meta['snapshot.json'], fileinfo)
# Update only version. Length and hashes are optional.
timestamp.signed.update_snapshot_meta(3)
fileinfo['snapshot.json'] = MetaFile(version=3)
self.assertEqual(timestamp.signed.meta, fileinfo)


def test_metadata_root(self):
Expand Down Expand Up @@ -320,9 +333,10 @@ def test_metadata_targets(self):
# Assert that data is not aleady equal
self.assertNotEqual(targets.signed.targets[filename], fileinfo)
# Update an already existing fileinfo
targets.signed.update(filename, fileinfo)
targets.signed.update_target(filename, fileinfo)
expected_target_file = TargetFile(length=28, hashes=hashes)
# Verify that data is updated
self.assertEqual(targets.signed.targets[filename], fileinfo)
self.assertEqual(targets.signed.targets[filename], expected_target_file)

# Run unit test.
if __name__ == '__main__':
Expand Down
208 changes: 157 additions & 51 deletions tuf/api/metadata.py
View file
Expand Up @@ -415,53 +415,102 @@ def remove_key(self, role: str, keyid: str) -> None:
del self.keys[keyid]


class MetaFile:
"""A container with information about a particilur file.
Instances of Metafile are used as values in a dictionary called "meta" in
Timestamp and Snapshot:
meta: A dictionary that contains information about metadata files::
{
'meta_file_path_1': <MetaFile INSTANCE_1>.
'meta_file_path_2': <MetaFile INSTANCE_2>
...
}
Attributes:
version: An integer indicating the version of the metafile.
length: An optional integer indicating the length of the metafile.
hashes: An optional dictionary containing hash algorithms and the
hashes resulted from applying them over the metafile::
'hashes': {
'<HASH ALGO 1>': '<META FILE HASH 1>',
'<HASH ALGO 2>': '<META FILE HASH 2>',
...
}
"""

def __init__(self, version: int, length: Optional[int] = None,
hashes: Optional[JsonDict] = None) -> None:
self.version = version
self.length = length
self.hashes = hashes


def __eq__(self, other: Any) -> bool:
"""Used to compare objects by their values instead of by their
addresses."""
if not isinstance(other, MetaFile):
# don't attempt to compare against unrelated types
return NotImplemented

return self.version == other.version and \
self.length == other.length and \
self.hashes == other.hashes


def to_dict(self) -> JsonDict:
"""Returns the JSON-serializable dictionary representation of self. """
json_dict = {'version': self.version}

if self.length is not None:
json_dict['length'] = self.length

if self.hashes is not None:
json_dict['hashes'] = self.hashes

return json_dict


class Timestamp(Signed):
"""A container for the signed part of timestamp metadata.
Attributes:
meta: A dictionary that contains information about snapshot metadata::
{
'snapshot.json': {
'version': <SNAPSHOT METADATA VERSION NUMBER>,
'length': <SNAPSHOT METADATA FILE SIZE>, // optional
'hashes': {
'<HASH ALGO 1>': '<SNAPSHOT METADATA FILE HASH 1>',
'<HASH ALGO 2>': '<SNAPSHOT METADATA FILE HASH 2>',
...
}
}
'snapshot.json': <MetaFile INSTANCE>
}
"""
def __init__(
self, _type: str, version: int, spec_version: str,
expires: datetime, meta: JsonDict) -> None:
super().__init__(_type, version, spec_version, expires)
# TODO: Add class for meta
self.meta = meta

self.meta = {}
self.meta['snapshot.json'] = MetaFile(
meta['snapshot.json']['version'],
meta['snapshot.json'].get('length'),
meta['snapshot.json'].get('hashes'))


# Serialization.
def to_dict(self) -> JsonDict:
"""Returns the JSON-serializable dictionary representation of self. """
json_dict = super().to_dict()
json_dict.update({
'meta': self.meta
'meta': {
'snapshot.json': self.meta['snapshot.json'].to_dict()
}
})
return json_dict


# Modification.
def update(self, version: int, length: int, hashes: JsonDict) -> None:
def update_snapshot_meta(self, version: int, length: Optional[int] = None,
hashes: Optional[JsonDict] = None) -> None:
"""Assigns passed info about snapshot metadata to meta dict. """
self.meta['snapshot.json'] = {
'version': version,
'length': length,
'hashes': hashes
}

self.meta['snapshot.json'] = MetaFile(version, length, hashes)


class Snapshot(Signed):
Expand All @@ -471,15 +520,7 @@ class Snapshot(Signed):
meta: A dictionary that contains information about targets metadata::
{
'targets.json': {
'version': <TARGETS METADATA VERSION NUMBER>,
'length': <TARGETS METADATA FILE SIZE>, // optional
'hashes': {
'<HASH ALGO 1>': '<TARGETS METADATA FILE HASH 1>',
'<HASH ALGO 2>': '<TARGETS METADATA FILE HASH 2>',
...
} // optional
},
'targets.json': <MetaFile INSTANCE>
'<DELEGATED TARGETS ROLE 1>.json': {
...
},
Expand All @@ -494,32 +535,95 @@ def __init__(
self, _type: str, version: int, spec_version: str,
expires: datetime, meta: JsonDict) -> None:
super().__init__(_type, version, spec_version, expires)
# TODO: Add class for meta
self.meta = meta

self.meta = {}
for meta_path, meta_info in meta.items():
self.meta[meta_path] = MetaFile(meta_info['version'],
meta_info.get('length'), meta_info.get('hashes'))


# Serialization.
def to_dict(self) -> JsonDict:
"""Returns the JSON-serializable dictionary representation of self. """
json_dict = super().to_dict()
meta_dict = {}
for meta_path, meta_info in self.meta.items():
meta_dict[meta_path] = meta_info.to_dict()

json_dict.update({
'meta': self.meta
'meta': meta_dict
})
return json_dict


# Modification.
def update(
def update_meta_file(
self, rolename: str, version: int, length: Optional[int] = None,
hashes: Optional[JsonDict] = None) -> None:
"""Assigns passed (delegated) targets role info to meta dict. """
metadata_fn = f'{rolename}.json'

self.meta[metadata_fn] = {'version': version}
if length is not None:
self.meta[metadata_fn]['length'] = length
self.meta[metadata_fn] = MetaFile(version, length, hashes)


class TargetFile:
"""A container with information about a particilur target file.
Instances of TargetFile are used as values in a dictionary
called "targets" in Targets.
targets: A dictionary that contains information about target files::
{
'target_file_path_1': <TargetFile INSTANCE_1>,
'target_file_path_2': <TargetFile INSTANCE_2>,
...
}
Attributes:
length: An integer indicating the length of the target file.
hashes: An dictionary containing hash algorithms and the
hashes resulted from applying them over the target file::
'hashes': {
'<HASH ALGO 1>': '<TARGET FILE HASH 1>',
'<HASH ALGO 2>': '<TARGET FILE HASH 2>',
...
}
custom: An optional dictionary which may include version numbers,
dependencies, or any other data that the application wants
to include to describe the target file::
'custom': {
'type': 'metadata',
'file_permissions': '0644',
...
} // optional
"""

if hashes is not None:
self.meta[metadata_fn]['hashes'] = hashes
def __init__(self, length: int, hashes: JsonDict,
custom: Optional[JsonDict] = None) -> None:
self.length = length
self.hashes = hashes
self.custom = custom


def __eq__(self, other: Any) -> bool:
"""Used to compare objects by their values instead of by their
addresses."""
if not isinstance(other, TargetFile):
# don't attempt to compare against unrelated types
return NotImplemented

return self.length == other.length and \
self.hashes == other.hashes and \
self.custom == other.custom


def to_dict(self) -> JsonDict:
"""Returns the JSON-serializable dictionary representation of self. """
json_dict = {'length': self.length, 'hashes': self.hashes}

if self.custom is not None:
json_dict['custom'] = self.custom

return json_dict


class Targets(Signed):
Expand All @@ -529,15 +633,7 @@ class Targets(Signed):
targets: A dictionary that contains information about target files::
{
'<TARGET FILE NAME>': {
'length': <TARGET FILE SIZE>,
'hashes': {
'<HASH ALGO 1>': '<TARGET FILE HASH 1>',
'<HASH ALGO 2>': '<TARGETS FILE HASH 2>',
...
},
'custom': <CUSTOM OPAQUE DICT> // optional
},
'<TARGET FILE NAME>': <TargetFile INSTANCE>,
...
}
Expand Down Expand Up @@ -584,22 +680,32 @@ def __init__(
expires: datetime, targets: JsonDict, delegations: JsonDict
) -> None:
super().__init__(_type, version, spec_version, expires)
# TODO: Add class for meta
self.targets = targets

self.targets = {}
for target_path, target_dict in targets.items():
self.targets[target_path] = TargetFile(target_dict['length'],
target_dict['hashes'], target_dict.get('custom'))

# TO DO: Add Key and Role classes
self.delegations = delegations


# Serialization.
def to_dict(self) -> JsonDict:
"""Returns the JSON-serializable dictionary representation of self. """
json_dict = super().to_dict()
target_dict = {}
for target_path, target_file_obj in self.targets.items():
target_dict[target_path] = target_file_obj.to_dict()

json_dict.update({
'targets': self.targets,
'targets': target_dict,
'delegations': self.delegations,
})
return json_dict

# Modification.
def update(self, filename: str, fileinfo: JsonDict) -> None:
def update_target(self, filename: str, fileinfo: JsonDict) -> None:
"""Assigns passed target file info to meta dict. """
self.targets[filename] = fileinfo
self.targets[filename] = TargetFile(fileinfo['length'],
fileinfo['hashes'], fileinfo.get('custom'))

0 comments on commit d1144c4

Please sign in to comment.