Cleanup & refactor faketime use

clients/go-tuf/cmd/download.go

@@ -15,7 +15,10 @@ import (
 	"fmt"
 	"net/url"
 	"os"
+	"os/exec"
 	"path/filepath"
+	"strconv"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -85,6 +88,21 @@ func RefreshAndDownloadCmd(targetName string,
 		return fmt.Errorf("failed to create Updater instance: %w", err)
 	}
 
+	// Test suite uses faketime to set the time client should consider "current".
+	// Golang does not use libc and bypasses the tricks faketime uses.
+	// use "date" command to get current time as this is affected by faketime
+	out, err := exec.Command("date", "+%s").Output()
+	if err != nil {
+		return fmt.Errorf("calling 'date' failed: %w", err)
+	}
+	timestamp_str := string(out[:len(out)-1])
+	timestamp, err := strconv.ParseInt(timestamp_str, 10, 64)
+	if err != nil {
+		return fmt.Errorf("failed to parse date output: %w", err)
+	}
+	ref_time := time.Unix(timestamp, 0)
+	up.UnsafeSetRefTime(ref_time)
+
 	// try to build the top-level metadata
 	err = up.Refresh()
 	if err != nil {

tuf_conformance/__init__.py

@@ -3,4 +3,4 @@
 __version__ = "2.2.0"
 
 # register pytest asserts before the imports happen in conftest.py
-pytest.register_assert_rewrite("tuf_conformance.client_runner")
+pytest.register_assert_rewrite("tuf_conformance._internal.client_runner")

tuf_conformance/client_runner.py → tuf_conformance/_internal/client_runner.py

@@ -1,14 +1,15 @@
 import glob
 import os
 import subprocess
+from datetime import datetime
 from tempfile import TemporaryDirectory
 
 from tuf.api.exceptions import StorageError
 from tuf.api.metadata import Metadata
 from tuf.api.serialization.json import JSONSerializer
 
-from tuf_conformance.metadata import MetadataTest
-from tuf_conformance.simulator_server import (
+from tuf_conformance._internal.metadata import MetadataTest
+from tuf_conformance._internal.simulator_server import (
     ClientInitData,
     SimulatorServer,
     StaticServer,
@@ -62,13 +63,13 @@ def init_client(self, data: ClientInitData) -> int:
         cmd = [*self._cmd, "--metadata-dir", self.metadata_dir, "init", trusted]
         return self._run(cmd)
 
-    def refresh(self, data: ClientInitData, days_in_future: int = 0) -> int:
+    def refresh(self, data: ClientInitData, fake_time: datetime | None = None) -> int:
         # dump a repository version for each client refresh (if configured to)
         self._server.debug_dump(self.test_name)
 
         cmd = self._cmd
-        if days_in_future:
-            cmd = ["faketime", "-f", f"+{days_in_future}d", *cmd]
+        if fake_time:
+            cmd = ["faketime", f"{fake_time}", *cmd]
 
         cmd = [
             *cmd,

tuf_conformance/repository_simulator.py → tuf_conformance/_internal/repository_simulator.py

@@ -52,7 +52,7 @@
 )
 from tuf.api.serialization.json import JSONSerializer
 
-from tuf_conformance.metadata import MetadataTest, RootTest
+from tuf_conformance._internal.metadata import MetadataTest, RootTest
 
 logger = logging.getLogger(__name__)
 

tuf_conformance/simulator_server.py → tuf_conformance/_internal/simulator_server.py

@@ -1,10 +1,11 @@
 import os
 from dataclasses import dataclass
+from datetime import UTC, datetime
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from os import path
 from urllib import parse
 
-from tuf_conformance.repository_simulator import RepositorySimulator
+from tuf_conformance._internal.repository_simulator import RepositorySimulator
 
 
 @dataclass
@@ -87,7 +88,9 @@ def debug_dump(self, test_name: str) -> None:
 class StaticServer(ThreadingHTTPServer):
     """Web server to serve static repositories"""
 
-    data_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static_data")
+    data_dir = os.path.join(
+        os.path.dirname(os.path.abspath(__file__)), "..", "static_data"
+    )
 
     @classmethod
     def static_test_names(cls) -> list[str]:
@@ -117,7 +120,7 @@ def do_GET(self) -> None:  # noqa: N802
         super().__init__(("127.0.0.1", 0), _StaticReqHandler)
         self.timeout = 0
 
-    def new_test(self, static_dir: str) -> tuple[ClientInitData, str]:
+    def new_test(self, static_dir: str) -> tuple[ClientInitData, str, datetime]:
         sub_dir = os.path.join(self.data_dir, static_dir)
         with open(os.path.join(sub_dir, "initial_root.json"), "rb") as f:
             initial_root = f.read()
@@ -133,7 +136,13 @@ def new_test(self, static_dir: str) -> tuple[ClientInitData, str]:
         with open(os.path.join(sub_dir, "targetpath")) as f:
             targetpath = f.readline().strip("\n")
 
-        return client_data, targetpath
+        try:
+            with open(os.path.join(sub_dir, "faketime")) as f:
+                faketime = datetime.fromisoformat(f.readline().strip("\n"))
+        except OSError:
+            faketime = datetime.now(UTC)
+
+        return client_data, targetpath, faketime
 
     def debug_dump(self, test_name: str) -> None:
         pass  # not implemented

tuf_conformance/conftest.py

@@ -3,8 +3,8 @@
 
 import pytest
 
-from tuf_conformance.client_runner import ClientRunner
-from tuf_conformance.simulator_server import SimulatorServer, StaticServer
+from tuf_conformance._internal.client_runner import ClientRunner
+from tuf_conformance._internal.simulator_server import SimulatorServer, StaticServer
 
 
 def pytest_addoption(parser: pytest.Parser) -> None:

tuf_conformance/static_data/README.md

@@ -3,11 +3,11 @@
 Subdirectories should contain complete repositories produced by a specific repository
 implementation. Each repository in a `<SUBDIR>` should
 * demonstrate all of the TUF features that the implementation uses
-* not expire for a very long time
 * Store metadata in `<SUBDIR/metadata>` and artifacts in `<SUBDIR/targets>` 
 * be ready to be published with just `python -m http.server <SUBDIR>` (in other words filenames
   should match the TUF http API)
 
 Additionally there should be 
   * A version of root in `<SUBDIR>/initial_root.json`: This will be used to initialize the client
   * `<SUBDIR>/targetpath` containing a targetpath of an artifact that exists in the repository
+  * optional `<SUBDIR>/faketime` containing a ISO 8601 date that will be used as the refresh time for the client

tuf_conformance/static_data/tuf-on-ci-0.11/faketime

@@ -0,0 +1 @@
+2025-02-09 09:17:23+00:00

tuf_conformance/test_basic.py

@@ -1,13 +1,14 @@
 import json
 import os
+from datetime import UTC, datetime, timedelta
 
 import pytest
 from securesystemslib.formats import encode_canonical
 from securesystemslib.hash import digest
 from tuf.api.metadata import Key, Metadata, MetaFile, Root, Snapshot, Targets, Timestamp
 
-from tuf_conformance.client_runner import ClientRunner
-from tuf_conformance.simulator_server import SimulatorServer
+from tuf_conformance._internal.client_runner import ClientRunner
+from tuf_conformance._internal.simulator_server import SimulatorServer
 
 
 def recalculate_keyid(key: Key) -> None:
@@ -414,3 +415,22 @@ def test_incorrect_metadata_type(client: ClientRunner, server: SimulatorServer)
     # Client should refuse timestamp that has incorrect type field
     assert client.refresh(init_data) == 1
     assert client.trusted_roles() == [(Root.type, 2)]
+
+
+def test_faketime(client: ClientRunner, server: SimulatorServer) -> None:
+    """Ensures that client supports the faketime setup in this test suite"""
+    init_data, repo = server.new_test(client.test_name)
+
+    assert client.init_client(init_data) == 0
+
+    # root v2 expires in 7 days
+    now = datetime.now(UTC)
+    repo.root.expires = now + timedelta(days=7)
+    repo.publish([Root.type])
+
+    # Refresh
+    assert client.refresh(init_data) == 0
+
+    # Mock time so that root has expired. If client unexpectedly succeeds here,
+    # it likely does not work with faketime
+    assert client.refresh(init_data, now + timedelta(days=8)) == 1

tuf_conformance/test_expiration.py

@@ -1,11 +1,10 @@
-import datetime
-from datetime import timezone
+from datetime import UTC, datetime, timedelta
 
 from tuf.api.metadata import Root, Snapshot, Targets, Timestamp
 
-from tuf_conformance import utils
-from tuf_conformance.client_runner import ClientRunner
-from tuf_conformance.simulator_server import SimulatorServer
+from tuf_conformance._internal import utils
+from tuf_conformance._internal.client_runner import ClientRunner
+from tuf_conformance._internal.simulator_server import SimulatorServer
 
 
 def test_root_expired(client: ClientRunner, server: SimulatorServer) -> None:
@@ -113,19 +112,19 @@ def test_expired_local_root(client: ClientRunner, server: SimulatorServer) -> No
     assert client.init_client(init_data) == 0
 
     # root v2 expires in 7 days
-    now = datetime.datetime.now(timezone.utc)
-    repo.root.expires = now + datetime.timedelta(days=7)
+    now = datetime.now(UTC)
+    repo.root.expires = now + timedelta(days=7)
     repo.publish([Root.type])
 
     # Refresh
     assert client.refresh(init_data) == 0
 
     # root v3 expires in 21 days
-    repo.root.expires = now + datetime.timedelta(days=21)
+    repo.root.expires = now + timedelta(days=21)
     repo.publish([Root.type])
 
     # Mocking time so that local root (v2) has expired but v3 from repo has not
-    assert client.refresh(init_data, days_in_future=18) == 0
+    assert client.refresh(init_data, now + timedelta(days=18)) == 0
     assert client.version(Root.type) == 3
 
 
@@ -145,21 +144,21 @@ def test_expired_local_timestamp(client: ClientRunner, server: SimulatorServer)
     assert client.init_client(init_data) == 0
 
     # Repo timestamp v1 expires in 7 days
-    now = datetime.datetime.now(timezone.utc)
-    repo.timestamp.expires = now + datetime.timedelta(days=7)
+    now = datetime.now(UTC)
+    repo.timestamp.expires = now + timedelta(days=7)
     repo.publish([Timestamp.type])  # v2
 
     # Refresh
     assert client.refresh(init_data) == 0
 
     # Bump targets + snapshot version
     # Set next version of repo timestamp to expire in 21 days
-    repo.timestamp.expires = now + datetime.timedelta(days=21)
+    repo.timestamp.expires = now + timedelta(days=21)
     repo.publish([Targets.type, Snapshot.type, Timestamp.type])  # v2, v2, v3
 
     # Mocking time so that local timestamp has expired
     # but the new timestamp has not
-    assert client.refresh(init_data, days_in_future=18) == 0
+    assert client.refresh(init_data, now + timedelta(days=18)) == 0
 
     # Assert final versions of timestamp/snapshot
     # which means a successful refresh is performed

tuf_conformance/test_file_download.py

@@ -1,9 +1,9 @@
 import pytest
 from tuf.api.metadata import Snapshot, TargetFile, Targets, Timestamp
 
-from tuf_conformance.client_runner import ClientRunner
-from tuf_conformance.repository_simulator import Artifact
-from tuf_conformance.simulator_server import SimulatorServer
+from tuf_conformance._internal.client_runner import ClientRunner
+from tuf_conformance._internal.repository_simulator import Artifact
+from tuf_conformance._internal.simulator_server import SimulatorServer
 
 
 def test_client_downloads_expected_file(

tuf_conformance/test_keys.py

@@ -3,8 +3,8 @@
 import pytest
 from tuf.api.metadata import Metadata, Root, Snapshot, Targets, Timestamp
 
-from tuf_conformance.client_runner import ClientRunner
-from tuf_conformance.simulator_server import SimulatorServer
+from tuf_conformance._internal.client_runner import ClientRunner
+from tuf_conformance._internal.simulator_server import SimulatorServer
 
 
 def test_snapshot_does_not_meet_threshold(

tuf_conformance/test_quoting_issues.py

@@ -3,8 +3,8 @@
 import pytest
 from tuf.api.metadata import DelegatedRole, Snapshot, Targets, Timestamp
 
-from tuf_conformance.client_runner import ClientRunner
-from tuf_conformance.simulator_server import SimulatorServer
+from tuf_conformance._internal.client_runner import ClientRunner
+from tuf_conformance._internal.simulator_server import SimulatorServer
 
 unusual_role_names = [
     "?",

tuf_conformance/test_repository_simulator.py

@@ -9,8 +9,8 @@
     Timestamp,
 )
 
-from tuf_conformance import utils
-from tuf_conformance.simulator_server import SimulatorServer
+from tuf_conformance._internal import utils
+from tuf_conformance._internal.simulator_server import SimulatorServer
 
 
 class TestRepositorySimulator(unittest.TestCase):

tuf_conformance/test_rollback.py

@@ -1,8 +1,8 @@
 import pytest
 from tuf.api.metadata import DelegatedRole, Root, Snapshot, Targets, Timestamp
 
-from tuf_conformance.client_runner import ClientRunner
-from tuf_conformance.simulator_server import SimulatorServer
+from tuf_conformance._internal.client_runner import ClientRunner
+from tuf_conformance._internal.simulator_server import SimulatorServer
 
 
 def test_new_timestamp_version_rollback(

tuf_conformance/test_static_repositories.py

@@ -1,7 +1,7 @@
 import pytest
 
-from tuf_conformance.client_runner import ClientRunner
-from tuf_conformance.simulator_server import StaticServer
+from tuf_conformance._internal.client_runner import ClientRunner
+from tuf_conformance._internal.simulator_server import StaticServer
 
 
 @pytest.mark.parametrize("static_repo", StaticServer.static_test_names())
@@ -13,8 +13,8 @@ def test_static_repository(
     This test is not a specification compliance test: It tests client compatibility
     with the repository format that a specific repository implementation produces.
     """
-    init_data, targetpath = static_server.new_test(static_repo)
+    init_data, targetpath, refresh_time = static_server.new_test(static_repo)
 
     assert static_client.init_client(init_data) == 0
-    assert static_client.refresh(init_data) == 0
+    assert static_client.refresh(init_data, refresh_time) == 0
     assert static_client.download_target(init_data, targetpath) == 0

tuf_conformance/test_updater_delegation_graphs.py

@@ -9,9 +9,9 @@
     Timestamp,
 )
 
-from tuf_conformance.client_runner import ClientRunner
-from tuf_conformance.repository_simulator import RepositorySimulator
-from tuf_conformance.simulator_server import SimulatorServer
+from tuf_conformance._internal.client_runner import ClientRunner
+from tuf_conformance._internal.repository_simulator import RepositorySimulator
+from tuf_conformance._internal.simulator_server import SimulatorServer
 
 
 @dataclass

tuf_conformance/test_updater_key_rotations.py

@@ -3,8 +3,8 @@
 import pytest
 from tuf.api.metadata import Root, Snapshot, Targets, Timestamp
 
-from tuf_conformance.client_runner import ClientRunner
-from tuf_conformance.simulator_server import SimulatorServer
+from tuf_conformance._internal.client_runner import ClientRunner
+from tuf_conformance._internal.simulator_server import SimulatorServer
 
 
 @dataclass