tests: Add target support to RepositorySimulator

* Add very simple targets support into simulator
* Add documentation for the simulator
* Add an example targets test

This might need to be tweaked and/or extended as we add tests but the
implementation should give a good indication of how to extend it.

As an example, non-consistent targets are not yet supported, but
making fetch() check for the consistent_snapshot state and respond
accordingly should be easy.

Signed-off-by: Jussi Kukkonen <[email protected]>

tests/repository_simulator.py

@@ -13,20 +13,51 @@
 as a way to "download" new metadata from remote: in practice no downloading,
 network connections or even file access happens as RepositorySimulator serves
 everything from memory.
+
+Metadata and targets "hosted" by the simulator are made available in URL paths
+"/metadata/..." and "/targets/..." respectively.
+
+Example::
+
+    # constructor creates repository with top-level metadata
+    sim = RepositorySimulator()
+
+    # metadata can be modified directly: it is immediately available to clients
+    sim.snapshot.version += 1
+
+    # As an exception, new root versions require explicit publishing
+    sim.root.version += 1
+    sim.publish_root()
+
+    # there are helper functions
+    sim.add_target("targets", b"content", "targetpath")
+    sim.targets.version += 1
+    sim.update_snapshot()
+
+    # Use the simulated repository from an Updater:
+    updater = Updater(
+        dir,
+        "https://example.com/metadata/",
+        "https://example.com/targets/",
+        sim
+    )
+    updater.refresh()
 """
 
 from collections import OrderedDict
+from dataclasses import dataclass
 from datetime import datetime, timedelta
 import logging
 import os
 import tempfile
+from securesystemslib.hash import digest
 from securesystemslib.keys import generate_ed25519_key
 from securesystemslib.signer import SSlibSigner
 from typing import Dict, Iterator, List, Optional, Tuple
 from urllib import parse
 
 from tuf.api.serialization.json import JSONSerializer
-from tuf.exceptions import FetcherHTTPError
+from tuf.exceptions import FetcherHTTPError, RepositoryError
 from tuf.api.metadata import (
     Key,
     Metadata,
@@ -35,6 +66,7 @@
     Root,
     SPECIFICATION_VERSION,
     Snapshot,
+    TargetFile,
     Targets,
     Timestamp,
 )
@@ -44,6 +76,11 @@
 
 SPEC_VER = ".".join(SPECIFICATION_VERSION)
 
+@dataclass
+class RepositoryTarget:
+    """Contains actual target data and the related target metadata"""
+    data: bytes
+    target_file: TargetFile
 
 class RepositorySimulator(FetcherInterface):
     def __init__(self):
@@ -60,6 +97,9 @@ def __init__(self):
         # signers are used on-demand at fetch time to sign metadata
         self.signers: Dict[str, List[SSlibSigner]] = {}
 
+        # target downloads are served from this dict
+        self.target_files: Dict[str, RepositoryTarget] = {}
+
         self.dump_dir = None
         self.dump_version = 0
 
@@ -136,10 +176,30 @@ def fetch(self, url: str) -> Iterator[bytes]:
                 version = None
                 role = parts[0]
             yield self._fetch_metadata(role, version)
+        elif spliturl.path.startswith("/targets/"):
+            # figure out the actual target path and the hash prefix
+            path = spliturl.path[len("/targets/") :]
+            dir_parts, sep , prefixed_filename = path.rpartition("/")
+            prefix, _, filename = prefixed_filename.partition(".")
+            target_path = f"{dir_parts}{sep}{filename}"
+
+            yield self._fetch_target(target_path, prefix)
         else:
             raise FetcherHTTPError(f"Unknown path '{spliturl.path}'", 404)
 
+    def _fetch_target(self, target_path: str, hash: Optional[str]) -> bytes:
+        """Return data for 'target_path', checking 'hash' if it is given"""
+        repo_target = self.target_files.get(target_path)
+        if repo_target is None:
+            raise FetcherHTTPError(f"No target {target_path}", 404)
+        if hash and hash not in repo_target.target_file.hashes.values():
+            raise FetcherHTTPError(f"hash mismatch for {target_path}", 404)
+
+        logger.debug("fetched target %s", target_path)
+        return repo_target.data
+
     def _fetch_metadata(self, role: str, version: Optional[int] = None) -> bytes:
+        """Return metadata for 'role', using 'version' if it is given"""
         if role == "root":
             # return a version previously serialized in publish_root()
             if version is None or version > len(self.signed_roots):
@@ -187,6 +247,16 @@ def update_snapshot(self):
         self.snapshot.version += 1
         self.update_timestamp()
 
+    def add_target(self, role: str, data: bytes, path: str):
+        if role == "targets":
+            targets = self.targets
+        else:
+            targets = self.md_delegates[role].signed
+
+        target = TargetFile.from_data(path, data, ["sha256"])
+        targets.targets[path] = target
+        self.target_files[path] = RepositoryTarget(data, target)
+
     def write(self):
         """Dump current repository metadata to self.dump_dir
 

tests/test_updater_with_simulator.py

@@ -24,11 +24,15 @@ class TestUpdater(unittest.TestCase):
     dump_dir:Optional[str] = None
 
     def setUp(self):
-        self.client_dir = tempfile.TemporaryDirectory()
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.metadata_dir = os.path.join(self.temp_dir.name, "metadata")
+        self.targets_dir = os.path.join(self.temp_dir.name, "targets")
+        os.mkdir(self.metadata_dir)
+        os.mkdir(self.targets_dir)
 
         # Setup the repository, bootstrap client root.json
         self.sim = RepositorySimulator()
-        with open(os.path.join(self.client_dir.name, "root.json"), "bw") as f:
+        with open(os.path.join(self.metadata_dir, "root.json"), "bw") as f:
             root = self.sim.download_bytes("https://example.com/metadata/1.root.json", 100000)
             f.write(root)
 
@@ -38,17 +42,21 @@ def setUp(self):
             self.sim.dump_dir = os.path.join(self.dump_dir, name)
             os.mkdir(self.sim.dump_dir)
 
-    def _run_refresh(self):
+    def tearDown(self):
+        self.temp_dir.cleanup()
+
+    def _run_refresh(self) -> Updater:
         if self.sim.dump_dir is not None:
             self.sim.write()
 
         updater = Updater(
-            self.client_dir.name,
+            self.metadata_dir,
             "https://example.com/metadata/",
             "https://example.com/targets/",
             self.sim
         )
         updater.refresh()
+        return updater
 
     def test_refresh(self):
         # Update top level metadata
@@ -71,6 +79,33 @@ def test_refresh(self):
 
         self._run_refresh()
 
+    def test_targets(self):
+        # target does not exist yet
+        updater = self._run_refresh()
+        self.assertIsNone(updater.get_one_valid_targetinfo("file"))
+
+        self.sim.targets.version += 1
+        self.sim.add_target("targets", b"content", "file")
+        self.sim.update_snapshot()
+
+        # target now exists, is not in cache yet
+        updater = self._run_refresh()
+        info = updater.get_one_valid_targetinfo("file")
+        self.assertIsNotNone(info)
+        self.assertEqual(
+            updater.updated_targets([info], self.targets_dir), [info]
+        )
+
+        # download target, assert it is in cache and content is correct
+        updater.download_target(info, self.targets_dir)
+        self.assertEqual(updater.updated_targets([info], self.targets_dir), [])
+        with open(os.path.join(self.targets_dir, "file"), "rb") as f:
+            self.assertEqual(f.read(), b"content")
+
+        # TODO: run the same download tests for
+        #     self.sim.add_target("targets", b"more content", "dir/file2")
+        # This currently fails because issue #1576
+
     def test_keys_and_signatures(self):
         """Example of the two trickiest test areas: keys and root updates"""
 
@@ -110,9 +145,6 @@ def test_keys_and_signatures(self):
 
         self._run_refresh()
 
-    def tearDown(self):
-        self.client_dir.cleanup()
-
 if __name__ == "__main__":
     if "--dump" in sys.argv:
         TestUpdater.dump_dir = tempfile.mkdtemp()