Add compressed certs

Author: elibon99

tests/device/test_opaque.py

@@ -25,6 +25,27 @@
 import pytest
 
 
+def get_test_cert():
+    private_key = ec.generate_private_key(
+        ALGORITHM.EC_P256.to_curve(), default_backend()
+    )
+    name = x509.Name(
+        [x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, "Test Certificate")]
+    )
+    one_day = datetime.timedelta(1, 0, 0)
+    certificate = (
+        x509.CertificateBuilder()
+        .subject_name(name)
+        .issuer_name(name)
+        .not_valid_before(datetime.datetime.today() - one_day)
+        .not_valid_after(datetime.datetime.today() + one_day)
+        .serial_number(int(uuid.uuid4()))
+        .public_key(private_key.public_key())
+        .sign(private_key, hashes.SHA256(), default_backend())
+    )
+    return certificate
+
+
 def test_put_empty(session):
     # Can't put an empty object
     with pytest.raises(ValueError):
@@ -74,27 +95,37 @@ def test_put_too_big(session):
 
 
 def test_certificate(session):
-    private_key = ec.generate_private_key(
-        ALGORITHM.EC_P256.to_curve(), default_backend()
-    )
-    name = x509.Name(
-        [x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, "Test Certificate")]
-    )
-    one_day = datetime.timedelta(1, 0, 0)
-    certificate = (
-        x509.CertificateBuilder()
-        .subject_name(name)
-        .issuer_name(name)
-        .not_valid_before(datetime.datetime.today() - one_day)
-        .not_valid_after(datetime.datetime.today() + one_day)
-        .serial_number(int(uuid.uuid4()))
-        .public_key(private_key.public_key())
-        .sign(private_key, hashes.SHA256(), default_backend())
-    )
-
+    certificate = get_test_cert()
     certobj = Opaque.put_certificate(
         session, 0, "Test certificate Opaque", 1, CAPABILITY.NONE, certificate
     )
 
     assert certificate == certobj.get_certificate()
     certobj.delete()
+
+
+def test_compressed_certificate(session):
+    certificate = get_test_cert()
+
+    certobj = Opaque.put_certificate(
+        session,
+        0,
+        "Test certificate Opaque",
+        1,
+        CAPABILITY.NONE,
+        certificate,
+    )
+    compressed_certobj = Opaque.put_certificate(
+        session,
+        0,
+        "Test certificate Opaque Compressed",
+        1,
+        CAPABILITY.NONE,
+        certificate,
+        compress=True,
+    )
+    assert certobj.get_certificate() == compressed_certobj.get_certificate(
+        decompress=True
+    )
+    assert certobj.get() != compressed_certobj.get()
+    assert len(certobj.get()) > len(compressed_certobj.get())

yubihsm/objects.py

@@ -34,6 +34,7 @@
 from typing import ClassVar, Union, Optional, TypeVar, NamedTuple, Type
 import copy
 import struct
+import gzip
 
 
 LABEL_LENGTH = 40
@@ -357,6 +358,7 @@ def put_certificate(
         domains: int,
         capabilities: CAPABILITY,
         certificate: x509.Certificate,
+        compress: bool = False,
     ) -> "Opaque":
         """Import an X509 certificate into the YubiHSM as an Opaque.
 
@@ -370,6 +372,8 @@ def put_certificate(
         :return: A reference to the newly created object.
         """
         encoded_cert = certificate.public_bytes(Encoding.DER)
+        if compress:
+            encoded_cert = gzip.compress(encoded_cert)
         return cls.put(
             session,
             object_id,
@@ -380,12 +384,15 @@ def put_certificate(
             encoded_cert,
         )
 
-    def get_certificate(self) -> x509.Certificate:
+    def get_certificate(self, decompress: bool = False) -> x509.Certificate:
         """Read an Opaque object from the YubiHSM, parsed as a certificate.
 
         :return: The certificate stored for the object.
         """
-        return x509.load_der_x509_certificate(self.get(), default_backend())
+        cert_data = self.get()
+        if decompress:
+            cert_data = gzip.decompress(cert_data)
+        return x509.load_der_x509_certificate(cert_data, default_backend())
 
 
 class AuthenticationKey(YhsmObject):