API Reference: CertMonitor

Use this page when you know the operation you need and want its exact signature, arguments, or return value. For a guided first check, start with Basic Usage.

The monitor

The reference below comes directly from the code's docstrings. Certificate collection, validation, raw formats, connection cleanup, and snapshot refresh all belong to the same monitor.

certmonitor.core.CertMonitor

CertMonitor(host: str, port: int = 443, enabled_validators: list[str] | None = None, *, connection_host: str | None = None, server_hostname: str | None = None, timeout: float = 10, cafile: str | None = None, capath: str | None = None, client_cert: str | None = None, client_key: str | None = None, starttls: str | None = None, proxy: str | None = None)

Class for monitoring and retrieving certificate details from a given host.

Initialize a monitor for a host without opening a connection.

Use a context manager to connect and close automatically. Retrieval and validation methods can also connect lazily. By default, validation checks expiration, SAN-based hostname identity, and trust through a separate verified handshake. Collection itself is permissive.

Parameters:

Name	Type	Description	Default
host	str	The identity the certificate is checked against. Also the default TCP destination and TLS SNI name.	required
port	int	Target TCP port. Defaults to 443.	443
enabled_validators	list[str] | None	Names to run. None uses the environment-backed configuration; an empty list disables all checks.	None
connection_host	str | None	Override the TCP destination, such as a backend IP.	None
server_hostname	str | None	Override the TLS SNI name sent to the server.	None
timeout	float	Positive timeout in seconds for each network operation, including each connection attempt made while collecting the certificate. This is not a whole-scan deadline; platform DNS resolution cannot be interrupted by this timeout.	10
cafile	str | None	PEM CA bundle for the separate verified trust handshake.	None
capath	str | None	OpenSSL-compatible CA directory for the verified trust handshake.	None
client_cert	str | None	Client certificate chain file for mutual TLS.	None
client_key	str | None	Separate client private-key file, if needed.	None
starttls	str | None	Application protocol whose STARTTLS preamble runs before the TLS handshake: "smtp", "imap", "pop3", "ftp", "postgres", or "ldap". Leave it unset and CertMonitor discovers the service itself, without looking at the port number: a plaintext greeting names SMTP, FTP, IMAP, POP3, or SSH, and a silent service is asked the PostgreSQL and LDAP StartTLS requests in turn. Discovery runs only when the port turns out not to speak TLS directly, and is bounded by timeout. Passing a protocol skips detection and discovery entirely. Collection, trust verification, and the post-quantum probe all run the preamble.	None
proxy	str | None	Route every connection through http://[user:pass@]host:port (HTTP CONNECT) or socks5://[user:pass@]host:port. The proxy resolves the target name. Results record the route with the password removed. Collection, trust verification, and the post-quantum probe all go through the tunnel.	None

Raises:

Type	Description
ValueError	If timeout is not positive or starttls is not a supported protocol name.

Example

with CertMonitor("example.com") as monitor:
    print(monitor.validate())

Source code in certmonitor/core.py

def __init__(
    self,
    host: str,
    port: int = 443,
    enabled_validators: list[str] | None = None,
    *,
    connection_host: str | None = None,
    server_hostname: str | None = None,
    timeout: float = 10,
    cafile: str | None = None,
    capath: str | None = None,
    client_cert: str | None = None,
    client_key: str | None = None,
    starttls: str | None = None,
    proxy: str | None = None,
):
    """Initialize a monitor for a host without opening a connection.

    Use a context manager to connect and close automatically. Retrieval and
    validation methods can also connect lazily. By default, validation checks
    expiration, SAN-based hostname identity, and trust through a separate
    verified handshake. Collection itself is permissive.

    Args:
        host: The identity the certificate is checked against. Also the
            default TCP destination and TLS SNI name.
        port: Target TCP port. Defaults to 443.
        enabled_validators: Names to run. `None` uses the environment-backed
            configuration; an empty list disables all checks.
        connection_host: Override the TCP destination, such as a backend IP.
        server_hostname: Override the TLS SNI name sent to the server.
        timeout: Positive timeout in seconds for each network operation,
            including each connection attempt made while collecting the
            certificate. This is not a whole-scan deadline; platform DNS
            resolution cannot be interrupted by this timeout.
        cafile: PEM CA bundle for the separate verified trust handshake.
        capath: OpenSSL-compatible CA directory for the verified trust handshake.
        client_cert: Client certificate chain file for mutual TLS.
        client_key: Separate client private-key file, if needed.
        starttls: Application protocol whose STARTTLS preamble runs before the
            TLS handshake: `"smtp"`, `"imap"`, `"pop3"`, `"ftp"`, `"postgres"`,
            or `"ldap"`. Leave it unset and CertMonitor discovers the service
            itself, without looking at the port number: a plaintext greeting
            names SMTP, FTP, IMAP, POP3, or SSH, and a silent service is asked
            the PostgreSQL and LDAP StartTLS requests in turn. Discovery runs
            only when the port turns out not to speak TLS directly, and is
            bounded by `timeout`. Passing a protocol skips detection and
            discovery entirely. Collection, trust verification, and the
            post-quantum probe all run the preamble.
        proxy: Route every connection through `http://[user:pass@]host:port`
            (HTTP CONNECT) or `socks5://[user:pass@]host:port`. The proxy resolves
            the target name. Results record the route with the password removed.
            Collection, trust verification, and the post-quantum probe all
            go through the tunnel.

    Raises:
        ValueError: If `timeout` is not positive or `starttls` is not a
            supported protocol name.

    Example:
        ```python
        with CertMonitor("example.com") as monitor:
            print(monitor.validate())
        ```
    """
    if timeout <= 0:
        raise ValueError("timeout must be positive")
    if starttls is not None and starttls not in starttls_negotiation.PROTOCOLS:
        raise ValueError(
            f"starttls must be one of {', '.join(starttls_negotiation.PROTOCOLS)}, not {starttls!r}"
        )
    self.starttls = starttls
    self.proxy: ProxyConfig | None = parse_proxy(proxy) if proxy else None
    self.connection_host = connection_host or host
    self.server_hostname = server_hostname or host
    self.timeout = timeout
    self.cafile, self.capath = cafile, capath
    self.client_cert, self.client_key = client_cert, client_key
    self.snapshot_at: str | None = None
    self._verify_contexts: dict[bool, ssl.SSLContext] = {}
    self._trust_verdict: tuple[bytes, dict[str, Any]] | None = None
    self._certificate_source: dict[str, Any] | None = None
    self._offline_bytes: bytes | None = None
    self.host = host
    self.port = port
    self.is_ip = self._is_ip_address(host)
    self.der: bytes | None = None
    self.pem: str | None = None
    self.cert_info = None
    self.cert_data: dict[str, Any] = {}
    self.public_key_der = None
    self.public_key_pem = None
    self.public_key_info: dict[str, Any] | None = None
    self.validators = VALIDATORS
    self.enabled_validators = (
        enabled_validators
        if enabled_validators is not None
        else config.ENABLED_VALIDATORS
    )
    self.error_handler = ErrorHandler()
    self.handler: BaseProtocolHandler | None = None
    self.protocol: str | None = None
    self.connected = False

cert_data instance-attribute

cert_data: dict[str, Any] = {}

cert_info instance-attribute

cert_info = None

connected instance-attribute

connected = False

connection_host instance-attribute

connection_host = connection_host or host

der instance-attribute

der: bytes | None = None

enabled_validators instance-attribute

enabled_validators = enabled_validators if enabled_validators is not None else ENABLED_VALIDATORS

error_handler instance-attribute

error_handler = ErrorHandler()

fingerprint_sha256 property

fingerprint_sha256: str | None

Lowercase hex SHA-256 of the collected leaf DER, or None before collection.

The same value openssl x509 -fingerprint -sha256 prints without the colons, so it can be compared with what a CA, a load balancer, or a previous scan recorded. A changed fingerprint means the certificate was replaced.

handler instance-attribute

handler: BaseProtocolHandler | None = None

host instance-attribute

host = host

is_ip instance-attribute

is_ip = _is_ip_address(host)

offline property

offline: bool

True when the certificate comes from a file or bytes, not a connection.

pem instance-attribute

pem: str | None = None

port instance-attribute

port = port

protocol instance-attribute

protocol: str | None = None

proxy instance-attribute

proxy: ProxyConfig | None = parse_proxy(proxy) if proxy else None

public_key_der instance-attribute

public_key_der = None

public_key_info instance-attribute

public_key_info: dict[str, Any] | None = None

public_key_pem instance-attribute

public_key_pem = None

server_hostname instance-attribute

server_hostname = server_hostname or host

snapshot_at instance-attribute

snapshot_at: str | None = None

starttls instance-attribute

starttls = starttls

timeout instance-attribute

timeout = timeout

validators instance-attribute

validators = VALIDATORS

__enter__

__enter__() -> CertMonitor

Enter the runtime context related to this object.

Source code in certmonitor/core.py

def __enter__(self) -> "CertMonitor":
    """Enter the runtime context related to this object."""
    self.connect()
    return self

__exit__

__exit__(exc_type: Any, exc_value: Any, traceback: Any) -> None

Exit the runtime context related to this object.

Source code in certmonitor/core.py

def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
    """Exit the runtime context related to this object."""
    self.close()

close

close() -> None

Close the connection, retaining the last snapshot for inspection.

Source code in certmonitor/core.py

def close(self) -> None:
    """Close the connection, retaining the last snapshot for inspection."""
    try:
        if self.handler:
            self.handler.close()
    finally:
        self.handler = None
        self.connected = False

connect

connect() -> dict[str, Any] | None

Establishes a connection to the host if not already connected.

Source code in certmonitor/core.py

def connect(self) -> dict[str, Any] | None:
    """Establishes a connection to the host if not already connected."""
    if self.connected:
        logging.debug("Already connected, skipping connection attempt")
        return None
    if self.offline:
        self.connected = True
        return None

    # An explicit STARTTLS protocol is the user's override: no detection,
    # no discovery. Otherwise detection may already name a STARTTLS service
    # from the greeting it peeked at.
    protocol_result = "ssl" if self.starttls else self.detect_protocol()
    if isinstance(protocol_result, dict) and "error" in protocol_result:
        return protocol_result
    protocol = cast(str, protocol_result)
    if protocol.startswith("starttls:"):
        self.starttls = protocol.partition(":")[2]
        protocol = "ssl"
    self.protocol = protocol

    unsupported = self._build_handler()
    if unsupported is not None:
        return unsupported

    connection_result: dict[str, Any] | None = cast(Any, self.handler).connect()
    if (
        connection_result is not None
        and self.protocol == "ssl"
        and self.starttls is None
    ):
        # The port did not speak TLS directly. It may be a STARTTLS service
        # (or SSH) that detection could not see because it had not greeted
        # yet; ask it, then retry once with what discovery found.
        discovered = self._discover_service()
        if discovered is not None:
            if discovered == "ssh":
                self.protocol = "ssh"
            else:
                self.starttls = discovered
            self._build_handler()
            connection_result = cast(Any, self.handler).connect()
    if connection_result is not None:  # This means there was an error
        return connection_result

    self.connected = True
    logging.debug(f"Successfully connected to {self.host}:{self.port}")
    return None

describe_validators

describe_validators() -> dict[str, dict[str, Any]]

Describe every registered validator and the user args it accepts.

Reads each validator's cached _user_params (built by BaseCertValidator.__init_subclass__ / BaseCipherValidator.__init_subclass__ at class definition time) and renders a serializable description suitable for printing, logging, or feeding into a CLI --help page.

Returns:

Name	Type	Description
dict	dict[str, dict[str, Any]]	Keyed by validator name. Each value contains: validator_type: "cert" or "cipher". doc: the validator class docstring (first line). args: dict keyed by user arg name, each with annotation (string), default (the literal default value), and required (always False, every user arg must declare a default).

Example

with CertMonitor("example.com") as monitor:
    for name, info in monitor.describe_validators().items():
        print(name, info["args"])

Source code in certmonitor/core.py

def describe_validators(self) -> dict[str, dict[str, Any]]:
    """Describe every registered validator and the user args it accepts.

    Reads each validator's cached `_user_params` (built by
    `BaseCertValidator.__init_subclass__` / `BaseCipherValidator.__init_subclass__`
    at class definition time) and renders a serializable description suitable
    for printing, logging, or feeding into a CLI `--help` page.

    Returns:
        dict: Keyed by validator name. Each value contains:

            - `validator_type`: `"cert"` or `"cipher"`.
            - `doc`: the validator class docstring (first line).
            - `args`: dict keyed by user arg name, each with `annotation`
              (string), `default` (the literal default value), and
              `required` (always `False`, every user arg must declare a
              default).

    Example:
        ```python
        with CertMonitor("example.com") as monitor:
            for name, info in monitor.describe_validators().items():
                print(name, info["args"])
        ```
    """
    import inspect

    described: dict[str, dict[str, Any]] = {}
    for name, validator in self.validators.items():
        user_params = getattr(validator, "_user_params", {}) or {}
        args_info: dict[str, dict[str, Any]] = {}
        for param_name, param in user_params.items():
            # `str()` renders both plain classes and parameterized
            # generics; only plain classes need the `<class 'X'>` wrapper
            # unwrapped. Enforcement in __init_subclass__ guarantees every
            # user param has an annotation, so no empty-annotation path.
            rendered = str(param.annotation)
            if rendered.startswith("<class '") and rendered.endswith("'>"):
                rendered = rendered[len("<class '") : -len("'>")]
            args_info[param_name] = {
                "annotation": rendered.replace("typing.", ""),
                "default": param.default,
                "required": False,
            }

        doc = inspect.getdoc(validator.__class__) or ""
        described[name] = {
            "validator_type": getattr(validator, "validator_type", "cert"),
            "doc": doc.splitlines()[0] if doc else "",
            "args": args_info,
        }
    return described

detect_protocol

detect_protocol() -> str | dict[str, Any]

Detect the protocol used by the host.

Returns "ssh", "ssl", or "starttls:<protocol>" when the peeked greeting belongs to a STARTTLS service, or an error dict.

Source code in certmonitor/core.py

def detect_protocol(self) -> str | dict[str, Any]:
    """Detect the protocol used by the host.

    Returns `"ssh"`, `"ssl"`, or `"starttls:<protocol>"` when the peeked
    greeting belongs to a STARTTLS service, or an error dict.
    """
    try:
        found = detection.detect(
            self.connection_host, self.port, self.timeout, connect=self._connect
        )
    except detection.ProtocolDetectionError as exc:
        return cast(
            dict[str, Any],
            self.error_handler.handle_error(
                "ProtocolDetectionError", str(exc), self.host, self.port
            ),
        )
    except Exception as e:
        return cast(
            dict[str, Any],
            self.error_handler.handle_error(
                "ConnectionError", str(e), self.host, self.port
            ),
        )
    if found.starttls:
        return f"starttls:{found.starttls}"
    return found.protocol

from_bytes classmethod

from_bytes(data: bytes | str, *, host: str | None = None, port: int = 443, enabled_validators: list[str] | None = None) -> CertMonitor

Build a monitor from PEM text or DER bytes already in memory.

Behaves like from_file(); use it for certificates fetched from an API, a secrets store, or a database. refresh() re-parses the same bytes.

Parameters:

Name	Type	Description	Default
data	bytes | str	PEM text (str or bytes) or DER bytes.	required
host	str | None	The identity the certificate should be valid for.	None
port	int	Port to report alongside the host. Defaults to 443.	443
enabled_validators	list[str] | None	Names to run. None uses the environment-backed configuration.	None

Source code in certmonitor/core.py

@classmethod
def from_bytes(
    cls,
    data: bytes | str,
    *,
    host: str | None = None,
    port: int = 443,
    enabled_validators: list[str] | None = None,
) -> "CertMonitor":
    """Build a monitor from PEM text or DER bytes already in memory.

    Behaves like `from_file()`; use it for certificates fetched from an
    API, a secrets store, or a database. `refresh()` re-parses the same
    bytes.

    Args:
        data: PEM text (str or bytes) or DER bytes.
        host: The identity the certificate should be valid for.
        port: Port to report alongside the host. Defaults to 443.
        enabled_validators: Names to run. `None` uses the environment-backed
            configuration.
    """
    monitor = cls(host or "", port, enabled_validators)
    monitor._offline_bytes = data.encode() if isinstance(data, str) else bytes(data)
    monitor._certificate_source = {"type": "bytes"}
    monitor.protocol = "ssl"
    return monitor

from_file classmethod

from_file(path: str | PathLike[str], *, host: str | None = None, port: int = 443, enabled_validators: list[str] | None = None) -> CertMonitor

Build a monitor from a certificate file instead of a connection.

The file may be PEM (a single certificate or a chain, leaf first) or DER. Everything that only needs certificate data works as it does for a connected monitor: get_cert_info(), the public key helpers, validate(), and refresh(), which re-reads the file. Checks that need a live connection (tls_version, weak_cipher, root_certificate, pq_key_exchange) report status: unsupported with a reason.

Parameters:

Name	Type	Description	Default
path	str | PathLike[str]	Path to the PEM or DER file.	required
host	str | None	The identity the certificate should be valid for, used by hostname and subject_alt_names. Without it those two checks report unsupported rather than guessing.	None
port	int	Port to report alongside the host. Defaults to 443.	443
enabled_validators	list[str] | None	Names to run. None uses the environment-backed configuration.	None

Example

with CertMonitor.from_file("service.pem", host="service.example.com") as monitor:
    print(monitor.validate()["expiration"])

Source code in certmonitor/core.py

@classmethod
def from_file(
    cls,
    path: str | os.PathLike[str],
    *,
    host: str | None = None,
    port: int = 443,
    enabled_validators: list[str] | None = None,
) -> "CertMonitor":
    """Build a monitor from a certificate file instead of a connection.

    The file may be PEM (a single certificate or a chain, leaf first) or
    DER. Everything that only needs certificate data works as it does for
    a connected monitor: `get_cert_info()`, the public key helpers,
    `validate()`, and `refresh()`, which re-reads the file. Checks that
    need a live connection (`tls_version`, `weak_cipher`, `root_certificate`,
    `pq_key_exchange`) report `status: unsupported` with a reason.

    Args:
        path: Path to the PEM or DER file.
        host: The identity the certificate should be valid for, used by
            `hostname` and `subject_alt_names`. Without it those two
            checks report `unsupported` rather than guessing.
        port: Port to report alongside the host. Defaults to 443.
        enabled_validators: Names to run. `None` uses the environment-backed
            configuration.

    Example:
        ```python
        with CertMonitor.from_file("service.pem", host="service.example.com") as monitor:
            print(monitor.validate()["expiration"])
        ```
    """
    monitor = cls(host or "", port, enabled_validators)
    monitor._certificate_source = {"type": "file", "path": os.fspath(path)}
    monitor.protocol = "ssl"
    return monitor

get_cert_info

get_cert_info() -> dict[str, Any]

Retrieves and structures the certificate details.

Source code in certmonitor/core.py

def get_cert_info(self) -> dict[str, Any]:
    """Retrieves and structures the certificate details."""
    if not self.cert_info:
        try:
            connection_result = self._ensure_connection()
            if connection_result is not None:  # Connection failed
                return connection_result

            cert_data = self._fetch_raw_cert()

            if isinstance(cert_data, dict) and "error" in cert_data:
                logging.error(f"Error in fetching raw certificate: {cert_data}")
                return cert_data

            # The _fetch_raw_cert already sets self.cert_data, self.public_key_der, self.public_key_pem
            # We just need to structure the cert_info part
            structured = self._to_structured_dict(cert_data["cert_info"])
            if self.fingerprint_sha256:
                structured["fingerprint_sha256"] = self.fingerprint_sha256
            self.cert_info = structured
            # Update the cert_data with the structured version
            if not hasattr(self, "cert_data") or not self.cert_data:
                self.cert_data = {}
            self.cert_data["cert_info"] = self.cert_info
            logging.debug("Certificate info retrieved and structured")
        except Exception as e:
            logging.error(f"Error while getting certificate info: {e}")
            return cast(
                dict[str, Any],
                self.error_handler.handle_error(
                    "UnknownError", str(e), self.host, self.port
                ),
            )

    return self.cert_info if self.cert_info is not None else {}

get_cipher_info

get_cipher_info() -> dict[str, Any]

Retrieve and structure the cipher information of the SSL/TLS connection.

Source code in certmonitor/core.py

def get_cipher_info(self) -> dict[str, Any]:
    """Retrieve and structure the cipher information of the SSL/TLS connection."""
    raw_cipher = self._fetch_raw_cipher()

    # Check if raw_cipher is an error response
    if isinstance(raw_cipher, dict) and "error" in raw_cipher:
        return raw_cipher

    # If raw_cipher is not an error, it should be a tuple of 3 elements
    if not isinstance(raw_cipher, tuple) or len(raw_cipher) != 3:
        return self.error_handler.handle_error(
            "CipherInfoError", "Unexpected cipher info format", self.host, self.port
        )

    cipher_suite, protocol_version, key_bit_length = raw_cipher
    parsed_cipher: dict[str, str] = parse_cipher_suite(cipher_suite)

    result: dict[str, Any] = {
        "cipher_suite": {
            "name": cipher_suite,
            "encryption_algorithm": parsed_cipher["encryption"],
            "message_authentication_code": parsed_cipher["mac"],
        },
        "protocol_version": protocol_version,
        "key_bit_length": key_bit_length,
    }

    if protocol_version == "TLSv1.3":
        result["cipher_suite"]["key_exchange_algorithm"] = (
            "Not applicable (TLS 1.3 uses ephemeral key exchange by default)"
        )
    else:
        result["cipher_suite"]["key_exchange_algorithm"] = parsed_cipher[
            "key_exchange"
        ]

    return result

get_enabled_validators

get_enabled_validators() -> list[str]

Get the list of validators enabled for this CertMonitor instance.

Returns:

Type	Description
list[str]	List[str]: A list of enabled validator names for this instance.

Source code in certmonitor/core.py

def get_enabled_validators(self) -> list[str]:
    """
    Get the list of validators enabled for this CertMonitor instance.

    Returns:
        List[str]: A list of enabled validator names for this instance.
    """
    return (
        self.enabled_validators.copy()
    )  # Return a copy to prevent external modification

get_public_key_der

get_public_key_der() -> bytes | dict[str, Any] | None

Return the public key in DER format.

Source code in certmonitor/core.py

def get_public_key_der(self) -> bytes | dict[str, Any] | None:
    """Return the public key in DER format."""
    if self.protocol != "ssl":
        return self.error_handler.handle_error(
            "ProtocolError",
            "Public key extraction is only available for SSL/TLS connections",
            self.host,
            self.port,
        )

    connection_result = self._ensure_connection()
    if connection_result is not None:  # Connection failed
        return connection_result

    if self.public_key_der is None:
        # Trigger certificate fetching which will also extract public keys
        cert_data = self._fetch_raw_cert()
        if isinstance(cert_data, dict) and "error" in cert_data:
            return cert_data

    return self.public_key_der

get_public_key_pem

get_public_key_pem() -> str | dict[str, Any] | None

Return the public key in PEM format.

Source code in certmonitor/core.py

def get_public_key_pem(self) -> str | dict[str, Any] | None:
    """Return the public key in PEM format."""
    if self.protocol != "ssl":
        return self.error_handler.handle_error(
            "ProtocolError",
            "Public key extraction is only available for SSL/TLS connections",
            self.host,
            self.port,
        )

    connection_result = self._ensure_connection()
    if connection_result is not None:  # Connection failed
        return connection_result

    if self.public_key_pem is None:
        # Trigger certificate fetching which will also extract public keys
        cert_data = self._fetch_raw_cert()
        if isinstance(cert_data, dict) and "error" in cert_data:
            return cert_data

    return self.public_key_pem

get_raw_der

get_raw_der() -> bytes | dict[str, Any]

Return the raw DER format of the certificate.

Source code in certmonitor/core.py

def get_raw_der(self) -> bytes | dict[str, Any]:
    """Return the raw DER format of the certificate."""
    connection_result = self._ensure_connection()
    if connection_result is not None:  # Connection failed
        return connection_result

    if self.protocol != "ssl":
        return cast(
            dict[str, Any],
            self.error_handler.handle_error(
                "ProtocolError",
                "DER format is only available for SSL/TLS connections",
                self.host,
                self.port,
            ),
        )

    if self.der is None:
        cert_data = self._fetch_raw_cert()
        if isinstance(cert_data, dict) and "error" in cert_data:
            return cert_data

    # Return the DER or empty bytes if None
    return self.der if self.der is not None else b""

get_raw_pem

get_raw_pem() -> str | dict[str, Any]

Return the raw PEM format of the certificate.

Source code in certmonitor/core.py

def get_raw_pem(self) -> str | dict[str, Any]:
    """Return the raw PEM format of the certificate."""
    connection_result = self._ensure_connection()
    if connection_result is not None:  # Connection failed
        return connection_result

    if self.protocol != "ssl":
        return cast(
            dict[str, Any],
            self.error_handler.handle_error(
                "ProtocolError",
                "PEM format is only available for SSL/TLS connections",
                self.host,
                self.port,
            ),
        )

    if self.pem is None:
        cert_data = self._fetch_raw_cert()
        if isinstance(cert_data, dict) and "error" in cert_data:
            return cert_data

    # Return the PEM or empty string if None
    return self.pem if self.pem is not None else ""

list_validators

list_validators() -> list[str]

Get a list of all available validators that can be used.

Returns:

Type	Description
list[str]	List[str]: A list of all registered validator names.

Source code in certmonitor/core.py

def list_validators(self) -> list[str]:
    """
    Get a list of all available validators that can be used.

    Returns:
        List[str]: A list of all registered validator names.
    """
    from .validators import list_validators as _list_validators

    return _list_validators()

refresh

refresh() -> dict[str, Any]

Close the old connection and collect a new timestamped snapshot.

Source code in certmonitor/core.py

def refresh(self) -> dict[str, Any]:
    """Close the old connection and collect a new timestamped snapshot."""
    self.close()
    self._clear_snapshot()
    return self.get_cert_info()

validate

validate(validator_args: dict[str, Any] | None = None) -> dict[str, Any]

Validates the target host by running all enabled validators.

This method: 1. Checks if all requested validators are implemented. 2. Separates validators into cert-based and cipher-based groups. 3. Fetches cert_info and cipher_info as needed. 4. Runs each validator with the appropriate arguments. 5. Returns a dictionary of validation results.

Parameters:

Name	Type	Description	Default
validator_args	dict	Additional arguments for specific validators. Example: { "subject_alt_names": {"alternate_names": ["www.example.com"]} }	None

Returns:

Name	Type	Description
dict	dict[str, Any]	A dictionary keyed by validator name, each value being the result of that validator.

Example

with CertMonitor("example.com", enabled_validators=["expiration", "weak_cipher"]) as monitor: results = monitor.validate() print(results["expiration"]) print(results["weak_cipher"])

Source code in certmonitor/core.py

def validate(self, validator_args: dict[str, Any] | None = None) -> dict[str, Any]:
    """
    Validates the target host by running all enabled validators.

    This method:
    1. Checks if all requested validators are implemented.
    2. Separates validators into cert-based and cipher-based groups.
    3. Fetches cert_info and cipher_info as needed.
    4. Runs each validator with the appropriate arguments.
    5. Returns a dictionary of validation results.

    Args:
        validator_args (dict, optional): Additional arguments for specific validators.
            Example:
            {
                "subject_alt_names": {"alternate_names": ["www.example.com"]}
            }

    Returns:
        dict: A dictionary keyed by validator name, each value being the result of that validator.

    Example:
        with CertMonitor("example.com", enabled_validators=["expiration", "weak_cipher"]) as monitor:
            results = monitor.validate()
            print(results["expiration"])
            print(results["weak_cipher"])
    """
    results: dict[str, Any] = {}

    # Check for unknown validators
    for requested_validator in self.enabled_validators:
        if requested_validator not in self.validators:
            results[requested_validator] = {
                "is_valid": False,
                "status": "error",
                "error": "UnknownValidator",
                "reason": f"Validator '{requested_validator}' is not implemented.",
            }

    # Active validators: enabled, implemented, and not already flagged
    # unknown above. Order follows the registry.
    active = [
        validator
        for name, validator in self.validators.items()
        if name in self.enabled_validators and name not in results
    ]

    # Each named data source is fetched at most once per call and
    # shared across every validator that requires it (e.g. cipher_info
    # and a future tls_probe).
    source_cache: dict[str, Any] = {}

    def resolve_source(source_name: str) -> Any:
        if source_name not in source_cache:
            source_cache[source_name] = self._fetch_source(
                source_name, resolve_source
            )
        return source_cache[source_name]

    for validator in active:
        # `requires` is authoritative when a validator declares it as
        # a real tuple; otherwise fall back to the legacy
        # `validator_type` mapping (also what test doubles use).
        requires = getattr(validator, "requires", None)
        if not isinstance(requires, tuple):
            vtype = getattr(validator, "validator_type", "cert")
            requires = ("cipher_info",) if vtype == "cipher" else ("cert_data",)

        resolved: list[Any] = []
        source_error: dict[str, Any] | None = None
        for source_name in requires:
            value = resolve_source(source_name)
            source_error = self._source_error(source_name, value)
            if source_error is not None:
                break
            resolved.append(value)

        # Uniform rule: if any required source could not be obtained,
        # the validator still appears in the results with a structured
        # error, never silently omitted.
        if source_error is not None:
            results[validator.name] = source_error
            continue

        results[validator.name] = self._invoke_validator(
            validator,
            (*resolved, self.host, self.port),
            validator_args,
        )

    for name, result in results.items():
        result.setdefault(
            "status",
            "error"
            if result.get("error")
            else "fail"
            if not result.get("is_valid")
            else "warn"
            if result.get("warnings")
            else "pass",
        )
        result.setdefault("code", f"{name}.{result['status']}")
    return results

Scan multiple hosts

scan_hosts() creates an independent monitor per worker and yields results in completion order. See Performance Tips for a complete example and timeout limits.

certmonitor.scanning.scan_hosts

scan_hosts(hosts: Iterable[Endpoint], *, port: int = 443, max_workers: int = 8, timeout: float = 10, enabled_validators: list[str] | None = None, validator_args: dict[str, Any] | None = None, cafile: str | None = None, capath: str | None = None, client_cert: str | None = None, client_key: str | None = None, starttls: str | None = None, proxy: str | None = None) -> Iterator[dict[str, Any]]

Yield completed scans with at most max_workers endpoints in flight.

Each result is a dict with host, port, results (the validate() output), snapshot_at, fingerprint_sha256, the parsed certificate, and public_key_info, so two runs can be handed to compare_snapshots(). If a scan raises, the dict carries an error (exception class name) and message instead of aborting the whole scan, so one bad host never hides the rest. Results arrive in completion order. Stopping iteration early returns promptly; scans that were still queued are cancelled and in-flight ones finish in the background.

timeout bounds individual network operations; platform DNS resolution is not interruptible.

Parameters:

Name	Type	Description	Default
hosts	Iterable[Endpoint]	Endpoints to scan, consumed lazily. Each entry is a host name or IP address, a (host, port) pair, or a dict with host plus any of port, connection_host, server_hostname, timeout, cafile, capath, client_cert, client_key, starttls, and proxy for endpoints that need their own connection settings.	required
port	int	TCP port for entries that do not carry one. Defaults to 443.	443
max_workers	int	Maximum number of concurrent scans. Defaults to 8.	8
timeout	float	Per-operation network timeout in seconds. Defaults to 10.	10
enabled_validators	list[str] | None	Validator names to run; None uses the defaults.	None
validator_args	dict[str, Any] | None	Per-validator keyword arguments applied to every host, in the same shape validate() accepts.	None
cafile	str | None	PEM CA bundle for trust verification on every endpoint. An endpoint dict may override it.	None
capath	str | None	CA directory for trust verification on every endpoint.	None
client_cert	str | None	Client certificate for mutual TLS on every endpoint.	None
client_key	str | None	Client private key, if separate from client_cert.	None
starttls	str | None	STARTTLS protocol name applied to every endpoint ("smtp", "imap", "pop3", "ftp", "postgres", "ldap"); endpoint dicts may set their own. Unset, each monitor discovers the service itself when the port does not speak TLS directly.	None
proxy	str | None	http:// or socks5:// proxy URL applied to every endpoint; endpoint dicts may set their own.	None

Raises:

Type	Description
ValueError	If max_workers or timeout is not positive.

Example

from certmonitor import scan_hosts

targets = [
    "example.com",
    ("legacy.example.net", 8443),
    {"host": "api.example.com", "connection_host": "192.0.2.10"},
]
for scan in scan_hosts(targets, max_workers=4, cafile="/path/to/private-ca.pem"):
    print(scan["host"], scan["port"], scan["results"]["expiration"]["status"])

Source code in certmonitor/scanning.py

def scan_hosts(
    hosts: Iterable[Endpoint],
    *,
    port: int = 443,
    max_workers: int = 8,
    timeout: float = 10,
    enabled_validators: list[str] | None = None,
    validator_args: dict[str, Any] | None = None,
    cafile: str | None = None,
    capath: str | None = None,
    client_cert: str | None = None,
    client_key: str | None = None,
    starttls: str | None = None,
    proxy: str | None = None,
) -> Iterator[dict[str, Any]]:
    """Yield completed scans with at most `max_workers` endpoints in flight.

    Each result is a dict with `host`, `port`, `results` (the `validate()`
    output), `snapshot_at`, `fingerprint_sha256`, the parsed `certificate`, and
    `public_key_info`, so two runs can be handed to `compare_snapshots()`. If a scan raises, the dict carries an
    `error` (exception class name) and `message` instead of aborting the
    whole scan, so one bad host never hides the rest. Results arrive in
    completion order. Stopping iteration early returns promptly; scans that
    were still queued are cancelled and in-flight ones finish in the
    background.

    `timeout` bounds individual network operations; platform DNS resolution
    is not interruptible.

    Args:
        hosts: Endpoints to scan, consumed lazily. Each entry is a host name or
            IP address, a `(host, port)` pair, or a dict with `host` plus any of
            `port`, `connection_host`, `server_hostname`, `timeout`, `cafile`,
            `capath`, `client_cert`, `client_key`, `starttls`, and `proxy` for endpoints that need
            their own connection settings.
        port: TCP port for entries that do not carry one. Defaults to 443.
        max_workers: Maximum number of concurrent scans. Defaults to 8.
        timeout: Per-operation network timeout in seconds. Defaults to 10.
        enabled_validators: Validator names to run; `None` uses the defaults.
        validator_args: Per-validator keyword arguments applied to every host,
            in the same shape `validate()` accepts.
        cafile: PEM CA bundle for trust verification on every endpoint. An
            endpoint dict may override it.
        capath: CA directory for trust verification on every endpoint.
        client_cert: Client certificate for mutual TLS on every endpoint.
        client_key: Client private key, if separate from `client_cert`.
        starttls: STARTTLS protocol name applied to every endpoint (`"smtp"`,
            `"imap"`, `"pop3"`, `"ftp"`, `"postgres"`, `"ldap"`); endpoint
            dicts may set their own. Unset, each monitor discovers the service
            itself when the port does not speak TLS directly.
        proxy: `http://` or `socks5://` proxy URL applied to every endpoint;
            endpoint dicts may set their own.

    Raises:
        ValueError: If `max_workers` or `timeout` is not positive.

    Example:
        ```python
        from certmonitor import scan_hosts

        targets = [
            "example.com",
            ("legacy.example.net", 8443),
            {"host": "api.example.com", "connection_host": "192.0.2.10"},
        ]
        for scan in scan_hosts(targets, max_workers=4, cafile="/path/to/private-ca.pem"):
            print(scan["host"], scan["port"], scan["results"]["expiration"]["status"])
        ```
    """
    if max_workers < 1 or timeout <= 0:
        raise ValueError("max_workers and timeout must be positive")
    endpoints = iter(hosts)

    shared = {
        "timeout": timeout,
        "cafile": cafile,
        "capath": capath,
        "client_cert": client_cert,
        "client_key": client_key,
        "starttls": starttls,
        "proxy": proxy,
    }

    def describe(entry: Endpoint) -> tuple[str, int, dict[str, Any]]:
        if isinstance(entry, str):
            return entry, port, {}
        if isinstance(entry, dict):
            unknown = set(entry) - ENDPOINT_OPTIONS - {"host", "port"}
            if "host" not in entry or unknown:
                raise ValueError(
                    f"endpoint dict needs 'host' and may only set {sorted(ENDPOINT_OPTIONS)}"
                )
            options = {k: v for k, v in entry.items() if k in ENDPOINT_OPTIONS}
            return str(entry["host"]), int(entry.get("port", port)), options
        host, entry_port = entry
        return host, entry_port, {}

    def scan(entry: Endpoint) -> dict[str, Any]:
        host, entry_port = (entry if isinstance(entry, str) else str(entry)), port
        try:
            host, entry_port, options = describe(entry)
            with CertMonitor(
                host, entry_port, enabled_validators, **{**shared, **options}
            ) as monitor:
                report = {
                    "host": host,
                    "port": entry_port,
                    "results": monitor.validate(validator_args),
                    "snapshot_at": monitor.snapshot_at,
                    "fingerprint_sha256": monitor.fingerprint_sha256,
                    "certificate": monitor.cert_info,
                    "public_key_info": monitor.public_key_info,
                }
                if monitor.connection_host != host:
                    report["connection_host"] = monitor.connection_host
                return report
        except Exception as exc:  # noqa: BLE001
            return {
                "host": host,
                "port": entry_port,
                "results": {},
                "snapshot_at": None,
                "error": type(exc).__name__,
                "message": str(exc),
            }

    executor = ThreadPoolExecutor(max_workers=max_workers)
    try:
        pending = set()
        for _ in range(max_workers):
            host = next(endpoints, None)
            if host is None:
                break
            pending.add(executor.submit(scan, host))
        while pending:
            done, pending = wait(pending, return_when=FIRST_COMPLETED)
            for future in done:
                yield future.result()
                host = next(endpoints, None)
                if host is not None:
                    pending.add(executor.submit(scan, host))
    finally:
        executor.shutdown(wait=False, cancel_futures=True)