Implement chain-of-trust verification

Author: glasserc

autograph_utils/__init__.py

@@ -18,6 +18,7 @@
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives.asymmetric import ec as cryptography_ec
+from cryptography.hazmat.primitives.asymmetric import padding
 from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature
 from cryptography.hazmat.primitives.hashes import SHA256, SHA384
 from cryptography.x509.oid import NameOID
@@ -154,6 +155,19 @@ def detail(self):
         )
 
 
+class CertificateChainBroken(BadCertificate):
+    def __init__(self, previous_cert, next_cert):
+        self.previous_cert = previous_cert
+        self.next_cert = next_cert
+
+    @property
+    def detail(self):
+        return (
+            "Certificate chain is not continuous. "
+            f"Expected {self.previous_cert!r} to sign {self.next_cert!r}"
+        )
+
+
 class BadSignature(Exception):
     detail = "Unknown signature problem"
 
@@ -274,6 +288,19 @@ async def verify_x5u(self, url):
         if root_hash != self.root_hash:
             raise CertificateHasWrongRoot(expected=self.root_hash, actual=root_hash)
 
+        current_cert = chain[0]
+        for next_cert in chain[1:]:
+            try:
+                current_cert.public_key().verify(
+                    next_cert.signature,
+                    next_cert.tbs_certificate_bytes,
+                    padding.PKCS1v15(),
+                    next_cert.signature_hash_algorithm,
+                )
+            except cryptography.exceptions.InvalidSignature:
+                raise CertificateChainBroken(current_cert, next_cert)
+            current_cert = next_cert
+
         leaf_subject_name = (
             certs[0].subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value
         )

tests/test_autograph_utils.py

@@ -236,6 +236,25 @@ async def test_verify_wrong_root_hash(aiohttp_session, mock_with_x5u, cache, now
     )
 
 
+async def test_verify_broken_chain(
+    aiohttp_session, mock_aioresponses, cache, now_fixed
+):
+    # Drop next-to-last cert in cert list
+    broken_chain = CERT_LIST[:1] + CERT_LIST[2:]
+    mock_aioresponses.get(FAKE_CERT_URL, status=200, body=b"\n".join(broken_chain))
+    s = SignatureVerifier(aiohttp_session, cache, DEV_ROOT_HASH)
+    with pytest.raises(autograph_utils.CertificateChainBroken) as excinfo:
+        await s.verify_x5u(FAKE_CERT_URL)
+
+    assert excinfo.value.detail.startswith("Certificate chain is not continuous. ")
+    assert excinfo.value.previous_cert == cryptography.x509.load_pem_x509_certificate(
+        CERT_LIST[2], backend=default_backend()
+    )
+    assert excinfo.value.next_cert == cryptography.x509.load_pem_x509_certificate(
+        CERT_LIST[0], backend=default_backend()
+    )
+
+
 def test_command_line_interface():
     """Test the CLI."""
     runner = CliRunner()