API Reference: Protocol Handlers

Protocol handlers are how CertMonitor talks to a host. When you connect, CertMonitor detects whether the endpoint speaks SSL/TLS or SSH and hands off to the matching handler, which knows how to fetch the certificate (and, for TLS, the cipher information). The pieces every connection shares live in the same package: opening streams, running and discovering STARTTLS preambles, and detection itself. You normally won't use any of this directly, since CertMonitor drives it for you. It's documented here for contributors and for anyone writing a custom handler.

See Protocol Detection for how the right handler gets chosen at connect time.

Base handler

The shared interface every handler implements.

certmonitor.protocol_handlers.base

BaseProtocolHandler

BaseProtocolHandler(host: str, port: int, error_handler: Any)

Bases: ABC

Source code in certmonitor/protocol_handlers/base.py

def __init__(self, host: str, port: int, error_handler: Any) -> None:
    self.host = host
    self.port = port
    self.socket: socket.socket | None = None
    self.secure_socket: ssl.SSLSocket | None = None
    self.error_handler = error_handler
    self.timeout: float = 10.0
    self.proxy: ProxyConfig | None = None

error_handler instance-attribute

error_handler = error_handler

host instance-attribute

host = host

port instance-attribute

port = port

proxy instance-attribute

proxy: ProxyConfig | None = None

secure_socket instance-attribute

secure_socket: SSLSocket | None = None

socket instance-attribute

socket: socket | None = None

timeout instance-attribute

timeout: float = 10.0

close abstractmethod

close() -> None

Source code in certmonitor/protocol_handlers/base.py

@abstractmethod
def close(self) -> None:
    pass

connect abstractmethod

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

Source code in certmonitor/protocol_handlers/base.py

@abstractmethod
def connect(self) -> dict[str, Any] | None:
    pass

fetch_raw_cert abstractmethod

fetch_raw_cert() -> dict[str, Any]

Source code in certmonitor/protocol_handlers/base.py

@abstractmethod
def fetch_raw_cert(self) -> dict[str, Any]:
    pass

SSL/TLS handler

Handles SSL/TLS endpoints: the handshake, certificate retrieval, and cipher info.

certmonitor.protocol_handlers.ssl_handler

SSLHandler

SSLHandler(host: str, port: int, error_handler: Any)

Bases: BaseProtocolHandler

Source code in certmonitor/protocol_handlers/ssl_handler.py

def __init__(self, host: str, port: int, error_handler: Any) -> None:
    super().__init__(host, port, error_handler)
    self.socket: socket.socket | None = None
    self.secure_socket: ssl.SSLSocket | None = None
    self.server_hostname = host
    self.timeout = 10.0
    self.client_cert: str | None = None
    self.client_key: str | None = None
    self.starttls: str | None = None
    self.tls_version: str | None = None

client_cert instance-attribute

client_cert: str | None = None

client_key instance-attribute

client_key: str | None = None

secure_socket instance-attribute

secure_socket: SSLSocket | None = None

server_hostname instance-attribute

server_hostname = host

socket instance-attribute

socket: socket | None = None

starttls instance-attribute

starttls: str | None = None

timeout instance-attribute

timeout = 10.0

tls_version instance-attribute

tls_version: str | None = None

check_connection

check_connection() -> bool

Source code in certmonitor/protocol_handlers/ssl_handler.py

def check_connection(self) -> bool:
    if self.secure_socket:
        try:
            self.secure_socket.getpeername()
            return True
        except Exception as e:
            logging.error(f"Error checking connection: {e}")
            return False
    return False

close

close() -> None

Source code in certmonitor/protocol_handlers/ssl_handler.py

def close(self) -> None:
    if self.secure_socket:
        self.secure_socket.close()
    if self.socket:
        self.socket.close()
    self.secure_socket = None
    self.socket = None
    self.tls_version = None

connect

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

Negotiate a permissive TLS session for certificate collection.

The first attempt offers every protocol version the local build supports. If that fails, a second attempt caps the offer at TLS 1.2 for servers that mishandle a TLS 1.3 ClientHello. A server that demands legacy renegotiation gets one retry with that option enabled. Every attempt gets its own timeout.

Source code in certmonitor/protocol_handlers/ssl_handler.py

def connect(self) -> dict[str, Any] | None:
    """Negotiate a permissive TLS session for certificate collection.

    The first attempt offers every protocol version the local build
    supports. If that fails, a second attempt caps the offer at TLS 1.2 for
    servers that mishandle a TLS 1.3 ClientHello. A server that demands
    legacy renegotiation gets one retry with that option enabled. Every
    attempt gets its own `timeout`.
    """
    attempts: list[dict[str, Any]] = [
        {},
        {"maximum_version": ssl.TLSVersion.TLSv1_2},
    ]
    for options in attempts:
        error = self._attempt(**options)
        if error is None:
            return None
        if "UNSAFE_LEGACY_RENEGOTIATION_DISABLED" in error.upper().replace(
            " ", "_"
        ):
            if self._attempt(**options, legacy_server_connect=True) is None:
                return None
    return cast(
        dict[str, Any],
        self.error_handler.handle_error(
            "SSLError",
            "Failed to establish SSL connection with any protocol",
            self.host,
            self.port,
        ),
    )

fetch_raw_cert

fetch_raw_cert() -> dict[str, Any]

Source code in certmonitor/protocol_handlers/ssl_handler.py

def fetch_raw_cert(self) -> dict[str, Any]:
    if not self.secure_socket:
        return cast(
            dict[str, Any],
            self.error_handler.handle_error(
                "ConnectionError",
                "SSL connection not established",
                self.host,
                self.port,
            ),
        )
    try:
        cert = self.secure_socket.getpeercert(binary_form=True)
        if cert is None:
            return cast(
                dict[str, Any],
                self.error_handler.handle_error(
                    "CertificateError",
                    "No certificate available",
                    self.host,
                    self.port,
                ),
            )
        chain_der, chain_error = self._fetch_chain_der()
        return {
            "cert_info": self.secure_socket.getpeercert(),
            "der": cert,
            "pem": ssl.DER_cert_to_PEM_cert(cert),
            "chain_der": chain_der,
            "chain_error": chain_error,
        }
    except Exception as e:
        return cast(
            dict[str, Any],
            self.error_handler.handle_error(
                "CertificateError", str(e), self.host, self.port
            ),
        )

fetch_raw_cipher

fetch_raw_cipher() -> tuple[str, str, int | None] | dict[str, Any]

Source code in certmonitor/protocol_handlers/ssl_handler.py

def fetch_raw_cipher(self) -> tuple[str, str, int | None] | dict[str, Any]:
    if not self.secure_socket:
        return cast(
            dict[str, Any],
            self.error_handler.handle_error(
                "ConnectionError",
                "SSL connection not established",
                self.host,
                self.port,
            ),
        )
    cipher_info = self.secure_socket.cipher()
    if cipher_info is None:
        return cast(
            dict[str, Any],
            self.error_handler.handle_error(
                "CipherError",
                "No cipher information available",
                self.host,
                self.port,
            ),
        )
    # cipher_info should be a 3-tuple when not None, but we check to be safe
    if isinstance(cipher_info, tuple) and len(cipher_info) == 3:
        return cipher_info
    # This should not happen in practice, but we handle it defensively
    return cast(  # type: ignore[unreachable]
        dict[str, Any],
        self.error_handler.handle_error(
            "CipherError", "Cipher information is not a tuple", self.host, self.port
        ),
    )

get_protocol_version

get_protocol_version() -> str

Source code in certmonitor/protocol_handlers/ssl_handler.py

def get_protocol_version(self) -> str:
    return self.tls_version or "Unknown"

SSH handler

Reads SSH version banners. It does not retrieve or validate SSH host keys or SSH certificates.

certmonitor.protocol_handlers.ssh_handler

SSHHandler

SSHHandler(host: str, port: int, error_handler: Any)

Bases: BaseProtocolHandler

Source code in certmonitor/protocol_handlers/base.py

def __init__(self, host: str, port: int, error_handler: Any) -> None:
    self.host = host
    self.port = port
    self.socket: socket.socket | None = None
    self.secure_socket: ssl.SSLSocket | None = None
    self.error_handler = error_handler
    self.timeout: float = 10.0
    self.proxy: ProxyConfig | None = None

check_connection

check_connection() -> bool

Source code in certmonitor/protocol_handlers/ssh_handler.py

def check_connection(self) -> bool:
    if not self.socket:
        return False
    try:
        self.socket.getpeername()
        return True
    except OSError:
        return False

close

close() -> None

Source code in certmonitor/protocol_handlers/ssh_handler.py

def close(self) -> None:
    if self.socket:
        self.socket.close()
        self.socket = None

connect

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

Source code in certmonitor/protocol_handlers/ssh_handler.py

def connect(self) -> dict[str, Any] | None:
    try:
        self.socket = open_stream(
            self.host, self.port, self.timeout, proxy=self.proxy
        )
        return None
    except OSError as e:
        return cast(
            dict[str, Any] | None,
            self.error_handler.handle_error(
                "SocketError", str(e), self.host, self.port
            ),
        )
    except Exception as e:
        return cast(
            dict[str, Any] | None,
            self.error_handler.handle_error(
                "UnknownError", str(e), self.host, self.port
            ),
        )

fetch_raw_cert

fetch_raw_cert() -> dict[str, Any]

Source code in certmonitor/protocol_handlers/ssh_handler.py

def fetch_raw_cert(self) -> dict[str, Any]:
    try:
        if not self.socket:
            return cast(
                dict[str, Any],
                self.error_handler.handle_error(
                    "ConnectionError", "Socket not connected", self.host, self.port
                ),
            )

        ssh_banner = self.socket.recv(1024).decode("ascii", errors="ignore").strip()
        match = re.match(r"^SSH-(\d+\.\d+)-(.*)$", ssh_banner)
        if match:
            return {
                "protocol": "ssh",
                "ssh_version_string": ssh_banner,
                "protocol_version": match.group(1),
                "software_version": match.group(2),
            }
        else:
            return cast(
                dict[str, Any],
                self.error_handler.handle_error(
                    "SSHError", "Invalid SSH banner", self.host, self.port
                ),
            )
    except Exception as e:
        return cast(
            dict[str, Any],
            self.error_handler.handle_error(
                "SSHError", str(e), self.host, self.port
            ),
        )

Connections

Every socket CertMonitor opens comes from here: a plaintext stream with any STARTTLS preamble already negotiated, or a TLS stream handshaken with the caller's context.

certmonitor.protocol_handlers.connection

Open the plaintext or TLS stream a handler needs, in one place.

Every connection CertMonitor makes takes the same steps: reach the host, through a proxy tunnel when one is configured, run a STARTTLS preamble when the service needs one, and, for TLS, wrap the socket with the caller's context. open_stream and open_tls_stream perform those steps so the handlers, protocol detection, service discovery, and the verified trust handshake never assemble a connection on their own.

open_stream

open_stream(host: str, port: int, timeout: float, *, starttls: str | None = None, proxy: ProxyConfig | None = None) -> socket.socket

Return a connected plaintext socket, with the STARTTLS preamble done.

Parameters:

Name	Type	Description	Default
host	str	Address to connect to.	required
port	int	TCP port.	required
timeout	float	Timeout in seconds for the connection and each preamble step.	required
starttls	str | None	One of starttls.PROTOCOLS to negotiate before returning, or None for a bare connection.	None
proxy	ProxyConfig | None	Tunnel to reach the host through, or None to connect directly.	None

Raises:

Type	Description
OSError	If the host cannot be reached, the proxy refuses the tunnel (proxy.ProxyError), or the server refuses STARTTLS (starttls.StartTLSError); all are OSError subclasses.

Source code in certmonitor/protocol_handlers/connection.py

def open_stream(
    host: str,
    port: int,
    timeout: float,
    *,
    starttls: str | None = None,
    proxy: ProxyConfig | None = None,
) -> socket.socket:
    """Return a connected plaintext socket, with the STARTTLS preamble done.

    Args:
        host: Address to connect to.
        port: TCP port.
        timeout: Timeout in seconds for the connection and each preamble step.
        starttls: One of `starttls.PROTOCOLS` to negotiate before returning,
            or `None` for a bare connection.
        proxy: Tunnel to reach the host through, or `None` to connect directly.

    Raises:
        OSError: If the host cannot be reached, the proxy refuses the tunnel
            (`proxy.ProxyError`), or the server refuses STARTTLS
            (`starttls.StartTLSError`); all are `OSError` subclasses.
    """
    sock = open_connection(host, port, timeout, proxy)
    try:
        if starttls:
            starttls_negotiation.negotiate(sock, starttls)
    except BaseException:
        sock.close()
        raise
    return sock

open_tls_stream

open_tls_stream(host: str, port: int, timeout: float, context: SSLContext, *, server_hostname: str | None, starttls: str | None = None, proxy: ProxyConfig | None = None) -> ssl.SSLSocket

Return a TLS socket handshaken with context over a fresh stream.

The plaintext socket is closed if the handshake fails, so a failed attempt never leaks a connection.

Source code in certmonitor/protocol_handlers/connection.py

def open_tls_stream(
    host: str,
    port: int,
    timeout: float,
    context: ssl.SSLContext,
    *,
    server_hostname: str | None,
    starttls: str | None = None,
    proxy: ProxyConfig | None = None,
) -> ssl.SSLSocket:
    """Return a TLS socket handshaken with `context` over a fresh stream.

    The plaintext socket is closed if the handshake fails, so a failed attempt
    never leaks a connection.
    """
    sock = open_stream(host, port, timeout, starttls=starttls, proxy=proxy)
    try:
        return context.wrap_socket(sock, server_hostname=server_hostname)
    except BaseException:
        sock.close()
        raise

Proxies

HTTP CONNECT and SOCKS5 tunnels, with authentication, that open_stream routes through when a monitor has a proxy. See Proxies for usage.

certmonitor.protocol_handlers.proxy

Outbound proxies: HTTP CONNECT tunnels and SOCKS5, standard library only.

open_connection() is the one place CertMonitor opens a TCP connection. With no proxy it is socket.create_connection(); with one it connects to the proxy, negotiates a tunnel to the target, and returns the socket ready for a TLS handshake or a STARTTLS preamble, exactly as a direct connection would be.

SCHEMES module-attribute

SCHEMES = ('http', 'socks5', 'socks5h')

ProxyConfig

Bases: NamedTuple

A parsed proxy URL. scheme is "http" or "socks5".

host instance-attribute

host: str

password class-attribute instance-attribute

password: str | None = None

port instance-attribute

port: int

redacted property

redacted: str

The proxy URL without its password, safe for results and logs.

scheme instance-attribute

scheme: str

username class-attribute instance-attribute

username: str | None = None

ProxyError

Bases: OSError

The proxy refused the tunnel, rejected the credentials, or misbehaved.

open_connection

open_connection(host: str, port: int, timeout: float, proxy: ProxyConfig | None = None) -> socket.socket

Return a connected socket to host:port, tunnelled through proxy when given.

Source code in certmonitor/protocol_handlers/proxy.py

def open_connection(
    host: str, port: int, timeout: float, proxy: ProxyConfig | None = None
) -> socket.socket:
    """Return a connected socket to `host:port`, tunnelled through `proxy` when given."""
    if proxy is None:
        return socket.create_connection((host, port), timeout=timeout)
    sock = socket.create_connection((proxy.host, proxy.port), timeout=timeout)
    try:
        if proxy.scheme == "http":
            _http_connect(sock, host, port, proxy)
        else:
            _socks5_connect(sock, host, port, proxy)
    except ProxyError:
        sock.close()
        raise
    except OSError as exc:
        sock.close()
        raise ProxyError(
            f"proxy {proxy.redacted} failed during negotiation: {exc}"
        ) from exc
    except BaseException:
        sock.close()
        raise
    return sock

parse_proxy

parse_proxy(url: str) -> ProxyConfig

Parse http://[user:pass@]host:port or socks5://[user:pass@]host:port.

socks5h:// is accepted as a synonym: the proxy always resolves the target name, so no DNS query leaves the scanning host either way.

Raises:

Type	Description
ValueError	If the scheme is unsupported or the host or port is missing.

Source code in certmonitor/protocol_handlers/proxy.py

def parse_proxy(url: str) -> ProxyConfig:
    """Parse `http://[user:pass@]host:port` or `socks5://[user:pass@]host:port`.

    `socks5h://` is accepted as a synonym: the proxy always resolves the target
    name, so no DNS query leaves the scanning host either way.

    Raises:
        ValueError: If the scheme is unsupported or the host or port is missing.
    """
    parts = urlsplit(url)
    scheme = parts.scheme.lower()
    if scheme not in SCHEMES:
        raise ValueError(
            f"unsupported proxy scheme {parts.scheme!r}; use http:// or socks5://"
        )
    if not parts.hostname:
        raise ValueError(f"proxy URL {url!r} has no host")
    try:
        port = parts.port
    except ValueError as exc:
        raise ValueError(f"proxy URL {url!r} has an invalid port") from exc
    if port is None:
        port = 1080 if scheme.startswith("socks") else 3128
    username = unquote(parts.username) if parts.username else None
    password = unquote(parts.password) if parts.password else None
    return ProxyConfig(
        "socks5" if scheme.startswith("socks") else "http",
        parts.hostname,
        port,
        username,
        password,
    )

Detection

Decides which handler a port needs from its first bytes, handing plaintext greetings to STARTTLS discovery.

certmonitor.protocol_handlers.detection

Work out which handler a port needs before any certificate is fetched.

Detection peeks at the first bytes a server sends. An SSH banner or a TLS record settles it at once, and a server that sends nothing is assumed to be waiting for a TLS ClientHello. A plaintext greeting means a STARTTLS service, so detection asks starttls.discover to name it rather than reporting an error. CertMonitor.detect_protocol() wraps this in the result envelope.

Connector module-attribute

Connector = Callable[[str, int, float], socket]

Opens a plaintext socket to (host, port, timeout); proxies plug in here.

Detected

Bases: NamedTuple

What a port speaks: the handler protocol and any preamble it needs.

protocol instance-attribute

protocol: str

"ssl" or "ssh".

starttls instance-attribute

starttls: str | None

STARTTLS protocol the SSL handler must negotiate first, if any.

ProtocolDetectionError

Bases: OSError

The port answered, but with something CertMonitor cannot name.

detect

detect(host: str, port: int, timeout: float, *, connect: Connector = open_stream) -> Detected

Name the protocol on host:port from its first bytes.

Parameters:

Name	Type	Description	Default
host	str	Address to connect to.	required
port	int	TCP port.	required
timeout	float	Timeout in seconds for the connection and for discovery.	required
connect	Connector	Opens the plaintext socket; the default connects directly.	open_stream

Raises:

Type	Description
ProtocolDetectionError	If the server greeted in plaintext and no STARTTLS service could be named.
OSError	If the host cannot be reached.

Source code in certmonitor/protocol_handlers/detection.py

def detect(
    host: str, port: int, timeout: float, *, connect: Connector = open_stream
) -> Detected:
    """Name the protocol on `host:port` from its first bytes.

    Args:
        host: Address to connect to.
        port: TCP port.
        timeout: Timeout in seconds for the connection and for discovery.
        connect: Opens the plaintext socket; the default connects directly.

    Raises:
        ProtocolDetectionError: If the server greeted in plaintext and no
            STARTTLS service could be named.
        OSError: If the host cannot be reached.
    """
    with connect(host, port, timeout) as sock:
        sock.setblocking(False)
        try:
            data = sock.recv(4, socket.MSG_PEEK)
        except OSError:
            # Nothing waiting: a TLS server speaks only after the ClientHello.
            return Detected("ssl", None)
        finally:
            sock.setblocking(True)
    if data.startswith(b"SSH-"):
        return Detected("ssh", None)
    if data and data[0] in _TLS_FIRST_BYTES:
        return Detected("ssl", None)
    # A plaintext greeting: name the service so its STARTTLS preamble can run
    # instead of a doomed handshake.
    try:
        discovered = starttls_negotiation.discover(host, port, timeout, connect=connect)
    except OSError:
        discovered = None
    if discovered == "ssh":
        return Detected("ssh", None)
    if discovered is not None:
        return Detected("ssl", discovered)
    raise ProtocolDetectionError(
        f"Unable to determine protocol. First bytes: {data.hex()}"
    )

STARTTLS

The preambles for SMTP, IMAP, POP3, FTP, PostgreSQL, and LDAP, and the discovery that names a service without looking at its port. See STARTTLS Services for usage.

certmonitor.protocol_handlers.starttls

Application-protocol preambles that upgrade a plain socket to TLS.

Some services start in plaintext and switch to TLS only after a short exchange (STARTTLS). Each function here performs that exchange on an already-connected socket and returns once the server has agreed to start TLS, leaving the socket ready for SSLContext.wrap_socket(). Nothing here imports beyond the standard library.

LDAP_STARTTLS_OID module-attribute

LDAP_STARTTLS_OID = b'1.3.6.1.4.1.1466.20037'

PROTOCOLS module-attribute

PROTOCOLS = ('smtp', 'imap', 'pop3', 'ftp', 'postgres', 'ldap')

StartTLSError

Bases: OSError

The server did not agree to start TLS, or the preamble was malformed.

discover

discover(host: str, port: int, timeout: float, *, client_name: str = 'certmonitor', connect: Callable[[str, int, float], socket] = _direct_connection) -> str | None

Name the plaintext service on host:port so the right STARTTLS preamble can run.

Nothing here looks at the port number, so services on non-standard ports are found just the same. A service that speaks first is named from its greeting: IMAP (* OK), POP3 (+OK), SSH (SSH-), and the 220 greeting shared by SMTP and FTP, which is settled by the greeting text or, failing that, by whether the server answers EHLO with 250. A service that stays silent is asked, in turn, the PostgreSQL SSLRequest and the LDAP StartTLS request, and is named from the reply. The whole exchange is bounded by timeout.

Parameters:

Name	Type	Description	Default
host	str	Address to connect to.	required
port	int	TCP port.	required
timeout	float	Total time budget in seconds for discovery.	required
client_name	str	Name announced in the EHLO used to tell SMTP from FTP.	'certmonitor'
connect	Callable[[str, int, float], socket]	Opens a plaintext socket to (host, port, timeout); the default connects directly, and a proxy-aware opener routes discovery too.	_direct_connection

Returns:

Type	Description
str | None	One of PROTOCOLS, "ssh" for an SSH banner, or None when the service
str | None	could not be named.

Raises:

Type	Description
OSError	If the first connection to the host fails.

Source code in certmonitor/protocol_handlers/starttls.py

def discover(
    host: str,
    port: int,
    timeout: float,
    *,
    client_name: str = "certmonitor",
    connect: Callable[[str, int, float], socket.socket] = _direct_connection,
) -> str | None:
    """Name the plaintext service on `host:port` so the right STARTTLS preamble can run.

    Nothing here looks at the port number, so services on non-standard ports are
    found just the same. A service that speaks first is named from its greeting:
    IMAP (`* OK`), POP3 (`+OK`), SSH (`SSH-`), and the `220` greeting shared by
    SMTP and FTP, which is settled by the greeting text or, failing that, by
    whether the server answers `EHLO` with `250`. A service that stays silent is
    asked, in turn, the PostgreSQL `SSLRequest` and the LDAP StartTLS request,
    and is named from the reply. The whole exchange is bounded by `timeout`.

    Args:
        host: Address to connect to.
        port: TCP port.
        timeout: Total time budget in seconds for discovery.
        client_name: Name announced in the `EHLO` used to tell SMTP from FTP.
        connect: Opens a plaintext socket to `(host, port, timeout)`; the default
            connects directly, and a proxy-aware opener routes discovery too.

    Returns:
        One of `PROTOCOLS`, `"ssh"` for an SSH banner, or `None` when the service
        could not be named.

    Raises:
        OSError: If the first connection to the host fails.
    """
    deadline = time.monotonic() + timeout
    with connect(host, port, timeout) as sock:
        greeting = _wait_for_greeting(sock, _remaining(deadline) / 2)
        if greeting is not None:
            return _name_greeting(sock, greeting, client_name)
        if _answers_ssl_request(sock, _remaining(deadline) / 2):
            return "postgres"
    try:
        with connect(host, port, _remaining(deadline)) as sock:
            if _answers_ldap_starttls(sock, _remaining(deadline)):
                return "ldap"
    except OSError:
        return None
    return None

ldap_starttls_request

ldap_starttls_request(message_id: int = 1) -> bytes

The LDAPMessage carrying an ExtendedRequest for the StartTLS OID (RFC 4511).

Source code in certmonitor/protocol_handlers/starttls.py

def ldap_starttls_request(message_id: int = 1) -> bytes:
    """The LDAPMessage carrying an ExtendedRequest for the StartTLS OID (RFC 4511)."""
    extended_request = _ber(
        0x77, _ber(0x80, LDAP_STARTTLS_OID)
    )  # [APPLICATION 23], [0] requestName
    return _ber(0x30, _ber(0x02, bytes([message_id])) + extended_request)

negotiate

negotiate(sock: socket, protocol: str, *, client_name: str = 'certmonitor') -> None

Run the STARTTLS preamble for protocol on sock.

Parameters:

Name	Type	Description	Default
sock	socket	A connected plaintext socket.	required
protocol	str	One of PROTOCOLS.	required
client_name	str	Name announced to servers that ask for one (SMTP EHLO).	'certmonitor'

Raises:

Type	Description
ValueError	If protocol is not supported.
StartTLSError	If the server refuses or answers unexpectedly.

Source code in certmonitor/protocol_handlers/starttls.py

def negotiate(
    sock: socket.socket, protocol: str, *, client_name: str = "certmonitor"
) -> None:
    """Run the STARTTLS preamble for `protocol` on `sock`.

    Args:
        sock: A connected plaintext socket.
        protocol: One of `PROTOCOLS`.
        client_name: Name announced to servers that ask for one (SMTP EHLO).

    Raises:
        ValueError: If `protocol` is not supported.
        StartTLSError: If the server refuses or answers unexpectedly.
    """
    handler = _HANDLERS.get(protocol)
    if handler is None:
        raise ValueError(
            f"unsupported STARTTLS protocol {protocol!r}; choose one of {', '.join(PROTOCOLS)}"
        )
    try:
        handler(sock, client_name)
    except ConnectionError as exc:
        # A reset or broken pipe mid-preamble means the server hung up on us,
        # which callers should see the same way as an orderly close.
        raise StartTLSError(
            f"connection closed during STARTTLS negotiation: {exc}"
        ) from exc