Fix: Prevent potential DoS

Previously, Connaisseur accepted all trust data files at first and then validated them. This was not an immediate security issue, since the root key could not be overwritten and since the KeyStore is write-once, so keys will only be used after they have been validated. However, Connaisseur would have pulled all delegations in a malicious targets.json without prior validation, which would have allowed an attacker to specify many non-existant delegations, potentially causing a denial of service. This PR fixes the issue by first validating and then processing the trust data files. In addition, the way Connaisseur previously validated trust data files would have allowed an attacker that compromised the long-term snapshot key to mount freeze attacks (i.e. ignoring the validation via timestamp key) by mounting a targeted collision attack instead of a 2nd-preimage attack against the DCT hash function.

SECURITY.md

@@ -2,7 +2,7 @@
 
 ## Supported Versions
 
-At the current stage of the project, we are not working with actual releases. Please make sure to frequently pull the latest project state, version and build of the Connaisseur image to ensure up-to-date security.
+The last known significant security vulnerability was in version 1.3.0 (Connaisseur not validating initContainers). However, since both Python packages and OS packages in the Connaisseur image may become known to be vulnerable over time, we suggest either frequently rebuilding the Connaisseur image from source yourself or updating to the latest Connaisseur image. We strictly stick to semantic versioning, so unless the major version changes, updating Conaisseur should never brake your installation.
 
 ## Reporting a Vulnerability
 

connaisseur/tests/test_keystore.py

@@ -84,7 +84,7 @@ def pub_key(path: str):
 
 @pytest.fixture
 def mock_trust_data(monkeypatch):
-    def _validate_expiry(self):
+    def validate_expiry(self):
         pass
 
     def trust_init(self, data: dict, role: str):
@@ -94,7 +94,7 @@ def trust_init(self, data: dict, role: str):
         self.signed = data["signed"]
         self.signatures = data["signatures"]
 
-    monkeypatch.setattr(TrustData, "_validate_expiry", _validate_expiry)
+    monkeypatch.setattr(TrustData, "validate_expiry", validate_expiry)
     monkeypatch.setattr(TargetsData, "__init__", trust_init)
     TrustData.schema_path = "res/{}_schema.json"
 

connaisseur/tests/test_mutate.py

@@ -314,7 +314,7 @@ def get_policy():
 
 @pytest.fixture
 def mock_trust_data(monkeypatch):
-    def _validate_expiry(self):
+    def validate_expiry(self):
         pass
 
     def trust_init(self, data: dict, role: str):
@@ -325,7 +325,7 @@ def trust_init(self, data: dict, role: str):
         self.signatures = data["signatures"]
 
     monkeypatch.setattr(
-        connaisseur.trust_data.TrustData, "_validate_expiry", _validate_expiry
+        connaisseur.trust_data.TrustData, "validate_expiry", validate_expiry
     )
     monkeypatch.setattr(connaisseur.trust_data.TargetsData, "__init__", trust_init)
     connaisseur.trust_data.TrustData.schema_path = "res/{}_schema.json"

connaisseur/tests/test_notary_api.py

@@ -105,7 +105,7 @@ def mock_get_request(**kwargs):
 
 @pytest.fixture
 def mock_trust_data(monkeypatch):
-    def _validate_expiry(self):
+    def validate_expiry(self):
         pass
 
     def trust_init(self, data: dict, role: str):
@@ -116,7 +116,7 @@ def trust_init(self, data: dict, role: str):
         self.signatures = data["signatures"]
 
     monkeypatch.setattr(
-        connaisseur.trust_data.TrustData, "_validate_expiry", _validate_expiry
+        connaisseur.trust_data.TrustData, "validate_expiry", validate_expiry
     )
     monkeypatch.setattr(connaisseur.trust_data.TargetsData, "__init__", trust_init)
     connaisseur.trust_data.TrustData.schema_path = "res/{}_schema.json"

connaisseur/tests/test_trust_data.py

@@ -351,7 +351,7 @@ def test_validate_schema(td, mock_schema_path, trustdata: dict, role: str):
 def test_validate_signature(td, mock_schema_path, mock_keystore, data: dict, role: str):
     ks = KeyStore()
     trust_data_ = td.TrustData(data, role)
-    assert trust_data_._validate_signature(ks) is None
+    assert trust_data_.validate_signature(ks) is None
 
 
 def test_validate_signature_error(td, mock_schema_path, mock_keystore):
@@ -361,7 +361,7 @@ def test_validate_signature_error(td, mock_schema_path, mock_keystore):
     ks = KeyStore()
 
     with pytest.raises(ValidationError) as err:
-        trust_data_._validate_signature(ks)
+        trust_data_.validate_signature(ks)
     assert "failed to verify signature of trust data." in str(err.value)
 
 
@@ -378,7 +378,7 @@ def test_validate_signature_error(td, mock_schema_path, mock_keystore):
 def test_validate_hash(td, mock_schema_path, mock_keystore, data: dict, role: str):
     ks = KeyStore()
     trust_data_ = td.TrustData(data, role)
-    assert trust_data_._validate_hash(ks) is None
+    assert trust_data_.validate_hash(ks) is None
 
 
 def test_validate_hash_error(td, mock_schema_path, mock_keystore):
@@ -388,7 +388,7 @@ def test_validate_hash_error(td, mock_schema_path, mock_keystore):
     ks = KeyStore()
 
     with pytest.raises(ValidationError) as err:
-        trust_data_._validate_hash(ks)
+        trust_data_.validate_hash(ks)
     assert "failed validating trust data hash." in str(err.value)
 
 
@@ -406,7 +406,7 @@ def test_validate_trust_data_expiry(td, mock_schema_path, data: dict, role: str)
     time_format = "%Y-%m-%dT%H:%M:%S.%f%z"
     trust_data_.signed["expires"] = time.strftime(time_format)
 
-    assert trust_data_._validate_expiry() is None
+    assert trust_data_.validate_expiry() is None
 
 
 @pytest.mark.parametrize(
@@ -420,7 +420,7 @@ def test_validate_trust_data_expiry_error(td, mock_schema_path, data: dict, role
     trust_data_.signed["expires"] = time.strftime(time_format)
 
     with pytest.raises(ValidationError) as err:
-        trust_data_._validate_expiry()
+        trust_data_.validate_expiry()
     assert "trust data expired." in str(err.value)
 
 

connaisseur/tests/test_validate.py

@@ -193,7 +193,7 @@ def init(self):
 
 @pytest.fixture
 def mock_trust_data(monkeypatch):
-    def _validate_expiry(self):
+    def validate_expiry(self):
         pass
 
     def trust_init(self, data: dict, role: str):
@@ -204,7 +204,7 @@ def trust_init(self, data: dict, role: str):
         self.signatures = data["signatures"]
 
     monkeypatch.setattr(
-        connaisseur.trust_data.TrustData, "_validate_expiry", _validate_expiry
+        connaisseur.trust_data.TrustData, "validate_expiry", validate_expiry
     )
     monkeypatch.setattr(connaisseur.trust_data.TargetsData, "__init__", trust_init)
     connaisseur.trust_data.TrustData.schema_path = "res/{}_schema.json"

connaisseur/trust_data.py

@@ -69,11 +69,11 @@ def validate(self, keystore: KeyStore):
         Validates the trust data's signature, expiry date and hash value, given
         the keys and hashes from a `keystore`.
         """
-        self._validate_signature(keystore)
-        self._validate_expiry()
-        self._validate_hash(keystore)
+        self.validate_signature(keystore)
+        self.validate_expiry()
+        self.validate_hash(keystore)
 
-    def _validate_expiry(self):
+    def validate_expiry(self):
         """
         Validates the expiry date of the trust data.
 
@@ -88,7 +88,7 @@ def _validate_expiry(self):
                 {"expire": str(expire), "trust_data_type": self.signed.get("_type")},
             )
 
-    def _validate_signature(self, keystore: KeyStore):
+    def validate_signature(self, keystore: KeyStore):
         """
         Validates the signature of the trust data, using keys from a
         `keystore`.
@@ -109,7 +109,7 @@ def _validate_signature(self, keystore: KeyStore):
                     {"key_id": key_id, "trust_data_type": self.signed.get("_type")},
                 )
 
-    def _validate_hash(self, keystore: KeyStore):
+    def validate_hash(self, keystore: KeyStore):
         """
         Validates the given hash from a `keystore` corresponds to the trust
         data's calculated hash.
@@ -166,7 +166,7 @@ def get_hashes(self):
 
 
 class TimestampData(TrustData):  # pylint: disable=abstract-method
-    def _validate_hash(self, keystore: KeyStore):
+    def validate_hash(self, keystore: KeyStore):
         pass
 
     def get_hashes(self):

connaisseur/validate.py

@@ -72,41 +72,80 @@ def process_chain_of_trust(
     """
     Processes the whole chain of trust, provided by the notary server (`host`)
     for any given `image`. The 'root', 'snapshot', 'timestamp', 'targets' and
-    potentially 'targets/releases' are requested in this order and afterwards
-    validated, also according to the `policy_rule`.
+    potentially 'targets/releases' are requested and validated.
+    Additionally, it is checked whether all required delegations are valid.
 
     Returns the signed image targets, which contain the digests.
 
     Raises `NotFoundExceptions` should no required delegetions be present in
     the trust data, or no image targets be found.
     """
-    tuf_roles = ["root", "snapshot", "timestamp", "targets"]
     trust_data = {}
     key_store = KeyStore()
 
-    # get all trust data and collect keys (from root and targets), as well as
-    # hashes (from snapshot and timestamp)
+    tuf_roles = ["root", "snapshot", "timestamp", "targets"]
+
+    # Load all trust data
     for role in tuf_roles:
-        trust_data[role] = get_trust_data(host, image, TUFRole(role))
-        key_store.update(trust_data[role])
+        role_trust_data = get_trust_data(host, image, TUFRole(role))
+        trust_data[role] = role_trust_data
+
+    # Validate signature and expiry data of and load root file
+    # This does NOT conclude the validation of the root file. To prevent roleback/freeze attacks,
+    # the hash still needs to be validated against the snapshot file
+    root_trust_data = get_trust_data(host, image, TUFRole("root"))
+    root_trust_data.validate_signature(key_store)
+    root_trust_data.validate_expiry()
+    trust_data["root"] = root_trust_data
+    key_store.update(root_trust_data)
+
+    # Validate timestamp file to prevent freeze attacks
+    # validates signature and expiry data
+    # there is no hash to verify it against since it is short lived
+    # TODO should we ensure short expiry duration here?
+    timestamp_trust_data = trust_data["timestamp"]
+    timestamp_trust_data.validate(key_store)
+
+    # Validate snapshot file signature against the key defined in the root file
+    # and its hash against the one from the timestamp file
+    # and validate expiry
+    snapshot_trust_data = trust_data["snapshot"]
+    snapshot_trust_data.validate_signature(key_store)
+
+    timestamp_key_store = KeyStore()
+    timestamp_key_store.update(timestamp_trust_data)
+    snapshot_trust_data.validate_hash(timestamp_key_store)
+
+    snapshot_trust_data.validate_expiry()
+
+    # Now snapshot and timestamp files are validated, we can be safe against
+    # roleback and freeze attacks if the root file matches the hash of the snapshot file
+    # (or the root key has been compromised, which Connaisseur cannot defend against)
+    snapshot_key_store = KeyStore()
+    snapshot_key_store.update(snapshot_trust_data)
+    root_trust_data.validate_hash(snapshot_key_store)
+
+    # If we are safe at this point, we can add the snapshot data to the main KeyStore
+    # and proceed with validating the targets file and (potentially) delegation files
+    key_store.update(snapshot_trust_data)
+    targets_trust_data = trust_data["targets"]
+    targets_trust_data.validate(key_store)
+    key_store.update(targets_trust_data)
 
     # if the 'targets.json' has delegation roles defined, get their trust data
     # as well
     if trust_data["targets"].has_delegations():
         for delegation in trust_data["targets"].get_delegations():
-            trust_data[delegation] = get_delegation_trust_data(
+            delegation_trust_data = get_delegation_trust_data(
                 host, image, TUFRole(delegation)
             )
+            # when delegations are added to the repository, but weren't yet used for signing, the
+            # delegation files don't exist yet and are `None`. in this case validation must be skipped
+            if delegation_trust_data is not None:
+                delegation_trust_data.validate(key_store)
+            trust_data[delegation] = delegation_trust_data
 
-    # validate all trust data's signatures, expiry dates and hashes.
-    # when delegations are added to the repository, but weren't yet used for signing, the
-    # delegation files don't exist yet and are `None`. in this case they can't be
-    # validated and must be skipped
-    for role in trust_data:
-        if trust_data[role] is not None:
-            trust_data[role].validate(key_store)
-
-    # validate needed delegations
+    # validate required delegations
     if req_delegations:
         if trust_data["targets"].has_delegations():
             delegations = trust_data["targets"].get_delegations()