Normalize plugin certificates output (#1498)

Author: william-billaud

dissect/target/helpers/certificate.py

@@ -15,10 +15,10 @@
 except ImportError:
     HAS_ASN1 = False
 
-
 COMMON_CERTIFICATE_FIELDS = [
     ("digest", "fingerprint"),
     ("varint", "serial_number"),
+    ("string", "serial_number_hex"),
     ("datetime", "not_valid_before"),
     ("datetime", "not_valid_after"),
     ("string", "issuer_dn"),
@@ -75,6 +75,34 @@ def compute_pem_fingerprints(pem: str | bytes) -> tuple[str, str, str]:
     return md5, sha1, sha256
 
 
+def format_serial_number_as_hex(serial_number: int | None) -> str | None:
+    """Format serial_number from integer to hex.
+
+    Add a prefix 0 if output length is not pair, in order to be consistent with usual serial_number representation
+    (navigator, openssl etc...).
+    For negative number use the same representation as navigator, which differ from OpenSSL.
+
+    For example for -1337::
+
+        OpenSSL : Serial Number: -1337 (-0x539)
+        Navigator : FA C7
+
+    Args:
+        serial_number: The serial number to format as hex.
+    """
+    if serial_number is None:
+        return serial_number
+
+    if serial_number > 0:
+        serial_number_as_hex = f"{serial_number:x}"
+        if len(serial_number_as_hex) % 2 == 1:
+            serial_number_as_hex = f"0{serial_number_as_hex}"
+        return serial_number_as_hex
+    # Representation is always a multiple of 8 bits, we need to compute this size
+    output_bin_len = 8 - (serial_number.bit_length() % 8) + serial_number.bit_length()
+    return f"{serial_number & ((1 << output_bin_len) - 1):x}"
+
+
 def parse_x509(file: str | bytes | Path) -> CertificateRecord:
     """Parses a PEM file. Returns a CertificateREcord. Does not parse a public key embedded in a x509 certificate."""
 
@@ -112,5 +140,6 @@ def parse_x509(file: str | bytes | Path) -> CertificateRecord:
         subject_dn=",".join(subject),
         fingerprint=(md5, crt.sha1.hex(), crt.sha256.hex()),
         serial_number=crt.serial_number,
+        serial_number_hex=format_serial_number_as_hex(crt.serial_number),
         pem=crt.dump(),
     )

dissect/target/plugins/os/windows/certlog.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Union
+from typing import TYPE_CHECKING, Any, Union
 
 from dissect.database.ese.tools import certlog
 from dissect.database.exception import Error
@@ -10,7 +10,7 @@
 from dissect.target.plugin import Plugin, export
 
 if TYPE_CHECKING:
-    from collections.abc import Iterator
+    from collections.abc import Callable, Iterator
     from pathlib import Path
 
     from dissect.target.target import Target
@@ -43,12 +43,12 @@
 CertificateRecord = TargetRecordDescriptor(
     "filesystem/windows/certlog/certificate",
     [
-        ("string", "certificate_hash2"),
+        ("digest", "fingerprint"),
         ("string", "certificate_template"),
         ("string", "common_name"),
         ("string", "country"),
         ("string", "device_serial_number"),
-        ("string", "distinguished_name"),
+        ("string", "subject_dn"),
         ("string", "domain_component"),
         ("string", "email"),
         ("string", "given_name"),
@@ -58,6 +58,7 @@
         ("string", "organizational_unit"),
         ("string", "public_key_algorithm"),
         ("string", "serial_number_hex"),
+        ("varint", "serial_number"),
         ("string", "state_or_province"),
         ("string", "street_address"),
         ("string", "subject_key_identifier"),
@@ -69,8 +70,8 @@
         ("string", "enrollment_flags"),
         ("string", "general_flags"),
         ("string", "issuer_name_id"),
-        ("datetime", "not_after"),
-        ("datetime", "not_before"),
+        ("datetime", "not_valid_after"),
+        ("datetime", "not_valid_before"),
         ("string", "private_key_flags"),
         ("string", "public_key"),
         ("varint", "public_key_length"),
@@ -95,7 +96,7 @@
         ("string", "device_serial_number"),
         ("string", "disposition"),
         ("string", "disposition_message"),
-        ("string", "distinguished_name"),
+        ("string", "subject_dn"),
         ("string", "domain_component"),
         ("string", "email"),
         ("string", "endorsement_certificate_hash"),
@@ -172,13 +173,14 @@
     "$AttributeValue": "common_name",
     "$CRLPublishError": "crl_publish_error",
     "$CallerName": "caller_name",
-    "$CertificateHash2": "certificate_hash2",
+    "$CertificateHash": "fingerprint",
+    "$CertificateHash2": "fingerprint",
     "$CertificateTemplate": "certificate_template",
     "$CommonName": "common_name",
     "$Country": "country",
     "$DeviceSerialNumber": "device_serial_number",
     "$DispositionMessage": "disposition_message",
-    "$DistinguishedName": "distinguished_name",
+    "$DistinguishedName": "subject_dn",
     "$DomainComponent": "domain_component",
     "$EMail": "email",
     "$EndorsementCertificateHash": "endorsement_certificate_hash",
@@ -221,8 +223,8 @@
     "NameId": "name_id",
     "NextPublish": "next_publish",
     "NextUpdate": "next_update",
-    "NotAfter": "not_after",
-    "NotBefore": "not_before",
+    "NotAfter": "not_valid_after",
+    "NotBefore": "not_valid_before",
     "Number": "number",
     "PrivateKeyFlags": "private_key_flags",
     "PropagationComplete": "propagation_complete",
@@ -254,6 +256,44 @@
 }
 
 
+def format_fingerprint(input_hash: str | None) -> tuple[str | None, str | None, str | None]:
+    if input_hash:
+        input_hash = input_hash.replace(" ", "")
+        # hash is expected to be a sha1, but as it not documented, we make this function more flexible if hash is
+        # in another standard format (md5/sha256), especially in the future
+        match len(input_hash):
+            case 32:
+                return input_hash, None, None
+            case 40:
+                return None, input_hash, None
+            case 64:
+                return None, None, input_hash
+            case _:
+                raise ValueError(
+                    "Unexpected hash size found while processing certlog "
+                    f"$CertificateHash/$CertificateHash2 column: len {len(input_hash)}, content {input_hash}"
+                )
+    return None, None, None
+
+
+def format_serial_number(serial_number_as_hex: str | None) -> str | None:
+    if not serial_number_as_hex:
+        return None
+    return serial_number_as_hex.replace(" ", "")
+
+
+def serial_number_as_int(serial_number_as_hex: str | None) -> int | None:
+    if not serial_number_as_hex:
+        return None
+    return int(serial_number_as_hex, 16)
+
+
+FORMATING_FUNC: dict[str, Callable[[Any], Any]] = {
+    "fingerprint": format_fingerprint,
+    "serial_number_hex": format_serial_number,
+}
+
+
 class CertLogPlugin(Plugin):
     """Return all available data stored in CertLog databases.
 
@@ -302,8 +342,27 @@ def read_records(self, table_name: str, record_type: CertLogRecord) -> Iterator[
                 record_values = {}
                 for column, value in column_values:
                     new_column = FIELD_MAPPINGS.get(column)
-                    if new_column:
+                    if new_column in FORMATING_FUNC:
+                        try:
+                            value = FORMATING_FUNC[new_column](value)
+                        except Exception as e:
+                            self.target.log.warning("Error formatting column %s (%s): %s", new_column, column, value)
+                            self.target.log.debug("", exc_info=e)
+                            value = None
+                    if new_column and new_column not in record_values:
                         record_values[new_column] = value
+                        # Serial number is format as int and string, to ease search of a specific sn in both format
+                        if new_column == "serial_number_hex":
+                            record_values["serial_number"] = serial_number_as_int(value)
+                    elif new_column:
+                        self.target.log.debug(
+                            "Unexpected element while processing %s entries : %s column already exists "
+                            "(mapped from original column name %s). This may be cause by two column that were not"
+                            " expected to be present in the same time.",
+                            table_name,
+                            new_column,
+                            column,
+                        )
                     else:
                         self.target.log.debug(
                             "Unexpected column for table %s in CA %s: %s", table_name, ca_name, column

…_data/plugins/apps/webserver/example.crt …/apps/webserver/certificates/example.crttests/_data/plugins/apps/webserver/example.crt renamed to tests/_data/plugins/apps/webserver/certificates/example.crt tests/_data/plugins/apps/webserver/example.crt renamed to tests/_data/plugins/apps/webserver/certificates/example.crt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d6f36dde378f080a05358ae6f572d1ef6a42725e23caafa076a7a532ad576888
+size 1440

…_data/plugins/apps/webserver/example.key …/apps/webserver/certificates/example.keytests/_data/plugins/apps/webserver/example.key renamed to tests/_data/plugins/apps/webserver/certificates/example.key tests/_data/plugins/apps/webserver/example.key renamed to tests/_data/plugins/apps/webserver/certificates/example.key

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:c2950633419e7dcd5b2830d6a6a05ac8fa675c58e70f19cc1ae23c7fb739b6ea
+size 1704

tests/_data/plugins/apps/webserver/certificates/negative_serial.crt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:bdf320de0e673176fe131dab252bda76a1e0a8ae1f46a382e2da0caf3cfa8db4
+size 1464

tests/_data/plugins/apps/webserver/certificates/negative_serial.key

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:370814279fa55bb7968081924bf1e508495ae1d87079e651f4a9e32359ab2dc2
+size 1464

tests/_data/plugins/apps/webserver/certificates/negative_serial_high.crt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:6b5bff12a5eed82a63a815975e8e277b2400b993594e0d7a553ae5751e6b7805
+size 1464

tests/_data/plugins/apps/webserver/certificates/negative_serial_high_2.crt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:4a24d908118cf2e6d4818250fabdbadc63aae32f9bf4f1abb2ebeba24f6948e5
+size 1464