Normalize plugins Certificates output

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

tests/plugins/apps/webserver/test_apache.py

@@ -484,12 +484,16 @@ def test_apache_hosts_certificates(target_unix: Target, fs_unix: VirtualFilesyst
     </VirtualHost>
     """
     fs_unix.map_file_fh("/etc/apache2/sites-available/example.conf", BytesIO(textwrap.dedent(site).encode()))
-    fs_unix.map_file("/path/to/cert.crt", absolute_path("_data/plugins/apps/webserver/example.crt"))
-    fs_unix.map_file("/path/to/cert.key", absolute_path("_data/plugins/apps/webserver/example.key"))
+    fs_unix.map_file("/path/to/cert.crt", absolute_path("_data/plugins/apps/webserver/certificates/example.crt"))
+    fs_unix.map_file("/path/to/cert.key", absolute_path("_data/plugins/apps/webserver/certificates/example.key"))
 
     # Map a default location too
-    fs_unix.map_file("/etc/apache2/ssl/example/cert.crt", absolute_path("_data/plugins/apps/webserver/example.crt"))
-    fs_unix.map_file("/etc/apache2/ssl/example/cert.key", absolute_path("_data/plugins/apps/webserver/example.key"))
+    fs_unix.map_file(
+        "/etc/apache2/ssl/example/cert.crt", absolute_path("_data/plugins/apps/webserver/certificates/example.crt")
+    )
+    fs_unix.map_file(
+        "/etc/apache2/ssl/example/cert.key", absolute_path("_data/plugins/apps/webserver/certificates/example.key")
+    )
 
     target_unix.add_plugin(ApachePlugin)
 
@@ -501,8 +505,131 @@ def test_apache_hosts_certificates(target_unix: Target, fs_unix: VirtualFilesyst
     assert records[0].fingerprint.sha1 == "6566d8ebea1feb4eb3d12d9486cddb69e4e9e827"
     assert records[0].fingerprint.sha256 == "7221d881743505f13b7bfe854bdf800d7f0cd22d34307ed7157808a295299471"
     assert records[0].serial_number == 21067204948278457910649605551283467908287726794
+    assert records[0].serial_number_hex == "03b0afa702c33e37fffd40e0c402b2120c1284ca"
     assert records[0].not_valid_before == datetime(2025, 11, 27, 15, 31, 20, tzinfo=timezone.utc)
     assert records[0].not_valid_after == datetime(2026, 11, 27, 15, 31, 20, tzinfo=timezone.utc)
     assert records[0].issuer_dn == "C=AU,ST=Some-State,O=Internet Widgits Pty Ltd,CN=example.com"
     assert records[0].host == "example.com"
     assert records[0].source == "/etc/apache2/ssl/example/cert.crt"
+
+
+@pytest.mark.parametrize(
+    ("crt_name", "serial_number", "serial_number_hex"),
+    [
+        pytest.param(
+            "negative_serial_high.crt",
+            -21067204948278457910649605551283467908287726794,
+            "fc4f5058fd3cc1c80002bf1f3bfd4dedf3ed7b36",
+            id="high_number",
+        ),
+        pytest.param(
+            "negative_serial_high_2.crt",
+            -210672049482784579106496055512834679082877267940,
+            "db192379e45f91d0001b773857e50b4b8746d01c",
+            id="high_number_158_bits",
+        ),
+        pytest.param(
+            "negative_serial_high_3.crt",
+            -421344098965569158212992111025669358165754535880,
+            "b63246f3c8bf23a00036ee70afca16970e8da038",
+            id="high_number_159_bits",
+        ),
+        pytest.param(
+            "negative_serial_high_4.crt",
+            -842688197931138316425984222051338716331509071760,
+            "ff6c648de7917e4740006ddce15f942d2e1d1b4070",
+            id="high_number_160_bits",
+        ),
+        pytest.param(
+            "negative_serial.crt",
+            -1337,
+            "fac7",
+            id="small_number",
+        ),
+    ],
+)
+def test_apache_hosts_certificates_negative_serial_number(
+    target_unix: Target, fs_unix: VirtualFilesystem, crt_name: str, serial_number: int, serial_number_hex: str
+) -> None:
+    """Test if we can parse Apache ``VirtualHost`` certificates, with a certificate using a negative serial number.
+
+    Each test uses a number with a different length in binary format, to ensure the representation matches
+    the one from navigator.
+
+    Generated using::
+
+        openssl genrsa -out negative_serial.key 2048
+        openssl req -new -x509 -key negative_serial.key \
+            -out negative_serial.crt -days 365 -set_serial <serial_number> -config openssl.cnf
+
+    Where openssl.cnf has the following content::
+
+        [ req ]
+        default_bits       = 2048
+        distinguished_name = req_distinguished_name
+        x509_extensions    = v3_ca
+        prompt             = no
+
+        [ req_distinguished_name ]
+        C  = FR
+        ST = RHONE
+        L  = Lyon
+        O  = Dissect
+        OU = Demo
+        CN = docs.dissect.tools
+
+        [ v3_ca ]
+        subjectKeyIdentifier   = hash
+        authorityKeyIdentifier = keyid:always,issuer
+        basicConstraints       = critical,CA:TRUE
+        keyUsage              = critical,digitalSignature,keyEncipherment
+        subjectAltName        = @alt_names
+
+        [ alt_names ]
+        DNS.1 = monserveur.example.com
+        DNS.2 = www.monserveur.example.com
+        IP.1  = 192.168.1.100
+    """
+
+    fs_unix.map_file_fh("/etc/apache2/apache2.conf", BytesIO(b'ServerRoot "/etc/apache2"\n'))
+
+    site = r"""
+    <VirtualHost *:443>
+        ServerName example.com
+        ServerAlias www.example.com
+        DocumentRoot /var/www/html
+        ErrorLog ${APACHE_LOG_DIR}/error.log
+        CustomLog ${APACHE_LOG_DIR}/access.log combined
+        SSLEngine on
+        SSLCertificateFile /path/to/cert.crt
+        SSLCertificateKeyFile /path/to/cert.key
+    </VirtualHost>
+    """
+    fs_unix.map_file_fh("/etc/apache2/sites-available/example.conf", BytesIO(textwrap.dedent(site).encode()))
+    fs_unix.map_file("/path/to/cert.crt", absolute_path(f"_data/plugins/apps/webserver/certificates/{crt_name}"))
+    fs_unix.map_file(
+        "/path/to/cert.key", absolute_path("_data/plugins/apps/webserver/certificates/negative_serial.key")
+    )
+
+    # Map a default location too
+    fs_unix.map_file(
+        "/etc/apache2/ssl/example/cert.crt", absolute_path(f"_data/plugins/apps/webserver/certificates/{crt_name}")
+    )
+    fs_unix.map_file(
+        "/etc/apache2/ssl/example/cert.key",
+        absolute_path("_data/plugins/apps/webserver/certificates/negative_serial.key"),
+    )
+
+    target_unix.add_plugin(ApachePlugin)
+
+    records = sorted(target_unix.apache.certificates(), key=lambda r: r.source)
+    assert len(records) == 2
+
+    assert records[0].webserver == "apache"
+    assert records[0].serial_number == serial_number
+    # openssl display the following for negative numbers : Serial Number: -1337 (-0x539)
+    # But navigators show FA:C7, we keep this representation
+    assert records[0].serial_number_hex == serial_number_hex
+    assert records[0].issuer_dn == "C=FR,ST=RHONE,L=Lyon,O=Dissect,OU=Demo,CN=docs.dissect.tools"
+    assert records[0].host == "example.com"
+    assert records[0].source == "/etc/apache2/ssl/example/cert.crt"