acemq_amqp.security

  1# Copyright 2026 AceMQ.
  2#
  3# Licensed under the Apache License, Version 2.0 (the "License");
  4# you may not use this file except in compliance with the License.
  5# You may obtain a copy of the License at
  6#
  7#     https://www.apache.org/licenses/LICENSE-2.0
  8#
  9# Unless required by applicable law or agreed to in writing, software
 10# distributed under the License is distributed on an "AS IS" BASIS,
 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12# See the License for the specific language governing permissions and
 13# limitations under the License.
 14
 15"""How this library reaches a broker: which authority it trusts, what
 16certificate it presents, and where the password comes from.
 17
 18Two things are decided here and they are deliberately not the same knob. **The
 19URL scheme decides whether the connection is encrypted** — ``amqps://`` is, and
 20``amqp://`` is not — because that is the part somebody reads in a configuration
 21file and believes. Everything in this module describes *how* an encrypted
 22connection is checked and who it logs in as. A :class:`Security` carrying TLS
 23settings against an ``amqp://`` URL is refused rather than ignored: settings that
 24look like TLS and quietly do nothing are how a service ends up in plaintext with
 25a certificate file sitting next to it.
 26
 27The defaults are the ones to want. ``amqps://`` with no :class:`Security` at all
 28verifies the broker against the machine's trust store and will not speak
 29anything older than TLS 1.2::
 30
 31    mq = await connect("amqps://broker.internal:5671/")
 32
 33A broker with its own authority — which is most brokers, because a public
 34certificate authority does not issue certificates for ``broker.internal`` — needs
 35that authority named, and then it is the *only* one trusted::
 36
 37    mq = await connect(
 38        "amqps://broker.internal:5671/",
 39        security=Security(
 40            certificate_authority="certs/ca.crt",
 41            credentials=credentials_from_environment(),
 42        ),
 43    )
 44
 45Credentials are separate from the URL on purpose. A password in a connection
 46string reaches every log line, crash report and ``ps`` listing that URL ever
 47appears in, and :class:`Credentials` never renders its secret however it is
 48printed.
 49
 50**A certificate marked as a development one is refused however trust is
 51configured**, including by :func:`without_verifying_the_broker`. The generator in
 52:mod:`acemq_amqp.devcerts` stamps everything it writes with
 53:data:`DEVELOPMENT_MARKER`, and this module is the other half of that bargain:
 54the mark is what makes a certificate whose private key is in a git repository
 55unusable against anything real. Java, Go, Ruby and .NET all do the same, and all
 56five give the same way out — say so deliberately, with
 57``Security(allow_development_certificates=True)``.
 58
 59Nothing here needs a dependency: :mod:`ssl` is in the standard library, and the
 60context this module builds is handed to aio-pika rather than being reimplemented
 61around it. The generator does need one, which is why it is an extra and lives in
 62a module of its own.
 63"""
 64
 65from __future__ import annotations
 66
 67import base64
 68import contextlib
 69import os
 70import ssl
 71from collections.abc import Callable, Sequence
 72from dataclasses import dataclass, field
 73from enum import Enum
 74from typing import Any, TypeAlias
 75from urllib.parse import quote, urlsplit, urlunsplit
 76
 77from .errors import SecurityError
 78
 79#: A path to a file, in either of the two shapes Python people write one.
 80FilePath: TypeAlias = "str | os.PathLike[str]"
 81
 82#: Stamped into the subject of every certificate :mod:`acemq_amqp.devcerts`
 83#: writes, and refused by this module wherever it appears.
 84#:
 85#: The same string in Java, Go, Ruby and .NET, byte for byte, so a certificate
 86#: generated by any of the five libraries is refused by all of them. That is the
 87#: entire mechanism: a development certificate is not kept out of production by
 88#: anybody remembering to delete it, but by being recognisable and by every
 89#: client refusing it on sight.
 90DEVELOPMENT_MARKER = "ACEMQ DEVELOPMENT ONLY - DO NOT TRUST"
 91
 92_MARKER_BYTES = DEVELOPMENT_MARKER.encode("ascii")
 93_MARKER_BYTES_LOWER = _MARKER_BYTES.lower()
 94
 95
 96def is_development_certificate(certificate: bytes) -> bool:
 97    """Whether a certificate carries :data:`DEVELOPMENT_MARKER`.
 98
 99    Reads DER, and reads PEM by decoding it first — the marker is plain ASCII
100    inside a DER certificate and is nowhere to be seen in the Base64 that wraps
101    it, so a PEM file searched as it stands would come back clean every time. A
102    PEM holding several certificates is read in full, so a chain file is checked
103    entry by entry.
104
105    The marker is looked for in the encoded certificate as a whole rather than in
106    a parsed subject, which is deliberate and is what lets this live in the
107    dependency-free core: an X.509 parser is not in the standard library, and the
108    answer would be the same one. Any distinguished name carrying the marker —
109    subject or issuer, ``O=`` as Go and .NET write it or ``OU=`` as Java does —
110    puts these bytes in the encoding, and there is nothing else a certificate
111    could plausibly contain them for.
112
113    Erring towards refusing is the right direction for this particular question.
114    A certificate that somehow contains the marker without being a development
115    one is refused, which is an inconvenience; the failure in the other direction
116    is a production system trusting an authority whose private key is on GitHub.
117
118    :param certificate: the certificate, DER-encoded or PEM-encoded
119    :returns: whether it is marked as a development certificate
120    """
121    if _MARKER_BYTES_LOWER in certificate.lower():
122        return True
123    return any(_MARKER_BYTES_LOWER in block.lower() for block in _der_blocks_in(certificate))
124
125
126def _der_blocks_in(pem: bytes) -> list[bytes]:
127    """The DER bodies of every ``CERTIFICATE`` block in a PEM file.
128
129    Written out rather than taken from :func:`ssl.PEM_cert_to_DER_cert`, which
130    reads one certificate and raises on anything it does not recognise: a file
131    holding a chain, a key alongside a certificate, or a comment above the first
132    header are all ordinary and none of them should turn "is this a development
133    certificate?" into an exception.
134    """
135    blocks: list[bytes] = []
136    body: list[bytes] | None = None
137    for line in pem.splitlines():
138        stripped = line.strip()
139        if stripped.startswith(b"-----BEGIN") and b"CERTIFICATE" in stripped:
140            body = []
141        elif stripped.startswith(b"-----END") and body is not None:
142            with contextlib.suppress(ValueError):
143                blocks.append(base64.b64decode(b"".join(body)))
144            body = None
145        elif body is not None:
146            body.append(stripped)
147    return blocks
148
149
150@dataclass(frozen=True, slots=True, repr=False)
151class Credentials:
152    """A username and a secret for the broker.
153
154    The secret never appears in :func:`repr` or :func:`str`, so a credential
155    logged by accident — in a dataclass dump, in an exception, in a debugger's
156    variables pane — prints the username and nothing else. That is the whole
157    reason this is a type rather than two strings, and a test pins it.
158
159    :param username: the account name
160    :param secret: the password, or the token for a broker that takes one
161    """
162
163    username: str = ""
164    secret: str = ""
165
166    @property
167    def empty(self) -> bool:
168        """Whether these credentials carry nothing at all."""
169        return not self.username and not self.secret
170
171    def __repr__(self) -> str:
172        if not self.username:
173            return "Credentials(<secret>)"
174        return f"Credentials(username={self.username!r}, secret=<secret>)"
175
176    __str__ = __repr__
177
178
179#: Where credentials come from, asked at the moment a connection is made.
180#:
181#: A callable rather than a class to subclass, because in Python a function is
182#: the interface: ``lambda: Credentials("app", vault.read())`` is a complete
183#: implementation. Asking each time is the point — a password rotated by a
184#: sidecar is only useful to something that reads it again.
185CredentialsSource: TypeAlias = Callable[[], "Credentials"]
186
187
188def credentials_from_environment(
189    username_variable: str = "ACEMQ_USERNAME",
190    password_variable: str = "ACEMQ_PASSWORD",
191) -> CredentialsSource:
192    """Reads two environment variables, each time it is asked.
193
194    :param username_variable: the variable holding the account name
195    :param password_variable: the variable holding the password
196    :returns: a source that reads them
197    :raises SecurityError: when either variable is unset, naming which one
198    """
199
200    def read() -> Credentials:
201        username = os.environ.get(username_variable)
202        if username is None:
203            raise SecurityError(
204                f"acemq: {username_variable} is not set, so there is no broker username"
205            )
206        password = os.environ.get(password_variable)
207        if password is None:
208            raise SecurityError(
209                f"acemq: {password_variable} is not set, so there is no broker password"
210            )
211        return Credentials(username, password)
212
213    return read
214
215
216def credentials_from_file(path: FilePath, *, username: str = "") -> CredentialsSource:
217    """Reads a secret from a file, each time it is asked.
218
219    This is how a mounted Kubernetes secret and a Docker secret arrive, and
220    reading it per connection rather than once at import is what lets a rotated
221    file take effect without a restart.
222
223    The file holds the password alone when ``username`` is given, and
224    ``username:password`` when it is not. Trailing whitespace is trimmed,
225    because a file written by an editor almost always ends in a newline and a
226    password with a newline on the end fails in a way nobody enjoys diagnosing.
227
228    :param path: the file to read
229    :param username: the account name, when the file holds only the secret
230    :returns: a source that reads the file
231    :raises SecurityError: when the file is missing, empty, or holds no username
232        and none was given
233    """
234
235    def read() -> Credentials:
236        where = os.fspath(path)
237        try:
238            with open(path, encoding="utf-8") as handle:
239                contents = handle.read().rstrip(" \t\r\n")
240        except OSError as unreadable:
241            raise SecurityError(
242                f"acemq: cannot read the credentials file {where!r}: {unreadable}"
243            ) from unreadable
244
245        if not contents:
246            raise SecurityError(f"acemq: the credentials file {where!r} is empty")
247        if username:
248            return Credentials(username, contents)
249        account, separator, secret = contents.partition(":")
250        if not separator:
251            raise SecurityError(
252                f"acemq: the credentials file {where!r} holds no username, and none was "
253                "given; write it as username:password, or pass username= to "
254                "credentials_from_file"
255            )
256        return Credentials(account, secret)
257
258    return read
259
260
261class Verification(Enum):
262    """What an encrypted connection checks about the broker on the other end.
263
264    Named rather than left as a boolean because a ``verify=False`` in a
265    configuration file says nothing about what is being given up. These say it.
266    """
267
268    #: Check the certificate chain and the name on it. The only setting to use
269    #: against a broker holding anything that matters.
270    CERTIFICATE = "certificate"
271
272    #: Check nothing. Encrypted, and to whoever answered.
273    #: :func:`without_verifying_the_broker` is the way to ask for this, and its
274    #: docstring says what it costs.
275    NOTHING_AT_ALL = "nothing-at-all"
276
277
278@dataclass(frozen=True, slots=True)
279class Security:
280    """How to verify the broker, what to present to it, and who to log in as.
281
282    Every field is optional and the default is the safe one: verify against the
283    machine's trust store, present no client certificate, and take the login
284    from the URL. Pass one to :func:`acemq_amqp.connect` or
285    :func:`acemq_amqp.sync.connect`.
286
287    A ``Security`` that carries TLS settings is refused against an ``amqp://``
288    URL rather than ignored — see this module's own documentation for why. One
289    that carries only ``credentials`` is fine on either scheme, because keeping
290    a password out of a connection string is worth doing whether or not the
291    connection is encrypted.
292
293    :param certificate_authority: a PEM file holding the authority to verify the
294        broker against. Naming one replaces the system trust store rather than
295        adding to it, which is the point: a broker with a certificate from a
296        public authority is not your broker, and the hundreds of authorities a
297        machine trusts by default are hundreds of ways to be wrong
298    :param client_certificate: a PEM certificate to present, for a broker that
299        authenticates clients by certificate rather than by password
300    :param client_key: the PEM private key for that certificate. Defaults to
301        reading the key out of ``client_certificate``, which is where it lives
302        when the two are in one file
303    :param client_key_password: the passphrase on that key, when it has one.
304        Kept out of this dataclass's :func:`repr` alongside the broker password
305    :param server_name: the name to check the certificate against, when it is
306        not the one in the URL. Needed for a broker reached by IP address,
307        through a tunnel, or at a container's internal name
308    :param credentials: the broker login, or a :data:`CredentialsSource` asked
309        for it at connection time. Overrides anything in the URL
310    :param verification: what to check about the broker's certificate
311    :param allow_development_certificates: whether to accept a certificate
312        stamped :data:`DEVELOPMENT_MARKER`. Off, and it should stay off anywhere
313        that is not a developer's machine: the signing key of a generated
314        authority is in whatever repository the certificates were committed to,
315        so trusting one means trusting everybody who can read it
316    """
317
318    certificate_authority: FilePath | None = None
319    client_certificate: FilePath | None = None
320    client_key: FilePath | None = None
321    client_key_password: str | None = field(default=None, repr=False)
322    server_name: str | None = None
323    credentials: Credentials | CredentialsSource | None = None
324    verification: Verification = Verification.CERTIFICATE
325    allow_development_certificates: bool = False
326
327    @property
328    def describes_tls(self) -> bool:
329        """Whether this says anything about an encrypted connection.
330
331        A ``Security`` that says nothing — one carrying credentials alone — is
332        as valid against ``amqp://`` as against ``amqps://``. One that does is
333        an instruction that a plaintext connection cannot carry out.
334        """
335        return (
336            self.certificate_authority is not None
337            or self.client_certificate is not None
338            or self.client_key is not None
339            or self.server_name is not None
340            or self.verification is not Verification.CERTIFICATE
341            or self.allow_development_certificates
342        )
343
344    def resolve_credentials(self) -> Credentials | None:
345        """Asks for the login, now.
346
347        :returns: the credentials, or ``None`` when this leaves the login to
348            the URL
349        :raises SecurityError: when a source was configured and could not
350            produce them
351        """
352        if self.credentials is None:
353            return None
354        if isinstance(self.credentials, Credentials):
355            return self.credentials
356        found = self.credentials()
357        if not isinstance(found, Credentials):
358            raise SecurityError(
359                "acemq: a credentials source must return Credentials, and this one "
360                f"returned {type(found).__name__}"
361            )
362        return found
363
364    def ssl_context(self) -> ssl.SSLContext:
365        """Builds the TLS context for a connection.
366
367        TLS 1.2 is the floor whatever the machine's ``openssl`` would otherwise
368        allow: 1.0 and 1.1 are withdrawn, and a broker still offering them is
369        not a reason to speak them.
370
371        :returns: the context, ready to hand to the transport
372        :raises SecurityError: when a certificate or key cannot be read, or when
373            one of them is a development certificate and this configuration did
374            not ask for those
375        """
376        context = self._new_context()
377        context.minimum_version = ssl.TLSVersion.TLSv1_2
378
379        if not self.allow_development_certificates:
380            # Two checks rather than one, because a development certificate can
381            # arrive from two directions. This is the near side: the files this
382            # configuration names. The far side — what the broker actually
383            # presents — is checked at the handshake, below, where it cannot be
384            # avoided by any trust setting.
385            self._refuse_development_file(self.certificate_authority, "certificate authority")
386            self._refuse_development_file(self.client_certificate, "client certificate")
387
388        if self.verification is Verification.NOTHING_AT_ALL:
389            # Both, and in this order: check_hostname cannot be left on once
390            # verify_mode is CERT_NONE, and Python raises rather than guessing
391            # which of the two was meant.
392            context.check_hostname = False
393            context.verify_mode = ssl.CERT_NONE
394        elif self.certificate_authority is not None:
395            # One authority, and no others. Verification succeeding therefore
396            # means the chain reached the authority named here — the system
397            # trust store is never consulted, which is what makes this stronger
398            # than the default rather than merely different from it.
399            authority = os.fspath(self.certificate_authority)
400            self._load(
401                lambda: context.load_verify_locations(cafile=authority),
402                f"the certificate authority {authority!r}",
403            )
404        else:
405            context.load_default_certs(ssl.Purpose.SERVER_AUTH)
406
407        if self.client_certificate is not None:
408            certificate = os.fspath(self.client_certificate)
409            key = None if self.client_key is None else os.fspath(self.client_key)
410            self._load(
411                lambda: context.load_cert_chain(
412                    certificate, keyfile=key, password=self.client_key_password
413                ),
414                f"the client certificate {certificate!r}"
415                + ("" if key is None else f" with key {key!r}"),
416            )
417        elif self.client_key is not None:
418            raise SecurityError(
419                "acemq: a client key was given with no client certificate; a broker "
420                "authenticates the certificate, and a key on its own proves nothing"
421            )
422
423        return context
424
425    def _new_context(self) -> ssl.SSLContext:
426        """A client context, naming a different broker when asked to.
427
428        ``PROTOCOL_TLS_CLIENT`` rather than :func:`ssl.create_default_context`
429        because the trust store is loaded deliberately a few lines further on:
430        the default context loads the system's, and a context that has already
431        trusted three hundred authorities cannot then be narrowed to one.
432        """
433        if self.server_name is None:
434            context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
435        else:
436            named = _ContextNamingTheBroker(ssl.PROTOCOL_TLS_CLIENT)
437            named.expected_name = self.server_name
438            context = named
439
440        if not self.allow_development_certificates:
441            # ``sslobject_class`` is the documented seam for this, and it is the
442            # only one: :mod:`ssl` has no verify callback of the sort Go's
443            # ``VerifyPeerCertificate`` and Java's ``X509TrustManager`` give,
444            # and there is nowhere else a check can be put that a trust setting
445            # cannot then turn off. Both classes are replaced because aio-pika
446            # goes through ``wrap_bio`` and a blocking client would go through
447            # ``wrap_socket``.
448            context.sslobject_class = _RefusingSSLObject
449            context.sslsocket_class = _RefusingSSLSocket
450        return context
451
452    @staticmethod
453    def _refuse_development_file(path: FilePath | None, what: str) -> None:
454        """Refuses a configured certificate file that carries the marker.
455
456        Reading the file is worth it for the error alone: at the handshake all
457        that can be said is that the broker presented one, whereas here the
458        message can name the setting and the path, which is what somebody
459        actually has to change.
460        """
461        if path is None:
462            return
463        where = os.fspath(path)
464        try:
465            with open(where, "rb") as handle:
466                contents = handle.read()
467        except OSError:
468            # Not this method's failure to report: the loader a few lines on
469            # opens the same file and says what is wrong with it far better.
470            return
471        if is_development_certificate(contents):
472            raise SecurityError(
473                f"acemq: the {what} {where!r} is marked {DEVELOPMENT_MARKER!r}. It was "
474                "generated for development and its signing key is not a secret, so it is "
475                "not trusted here. Use a real certificate, or say so deliberately with "
476                "Security(allow_development_certificates=True)"
477            )
478
479    @staticmethod
480    def _load(read: Callable[[], None], what: str) -> None:
481        """Runs one of :mod:`ssl`'s loaders and says what failed in words.
482
483        ``ssl`` reports a missing file as ``FileNotFoundError`` and a malformed
484        one as ``[SSL] PEM lib``, neither of which names the setting that was
485        wrong. This does.
486        """
487        try:
488            read()
489        except (OSError, ssl.SSLError) as failure:
490            raise SecurityError(f"acemq: cannot load {what}: {failure}") from failure
491
492    def applied_to(self, url: str) -> str:
493        """The URL this configuration would actually connect to.
494
495        Credentials given here replace whatever the URL carried, which is how a
496        password stays out of a connection string. When none were given the URL
497        comes back as it was.
498
499        :param url: where the broker is
500        :returns: the URL to hand to the transport
501        :raises SecurityError: when TLS settings were given for an ``amqp://``
502            URL, which cannot honour them
503        """
504        scheme = urlsplit(url).scheme
505        if self.describes_tls and scheme != "amqps":
506            raise SecurityError(
507                f"acemq: this Security configures TLS but the URL is {scheme or 'unset'}://, "
508                "which is not encrypted. Nothing here can make a plaintext connection safe: "
509                "change the URL to amqps:// or take the TLS settings off"
510            )
511
512        credentials = self.resolve_credentials()
513        if credentials is None or credentials.empty:
514            return url
515        return _with_credentials(url, credentials)
516
517    def transport_options(self, url: str) -> dict[str, Any]:
518        """What the transport needs, beyond the URL.
519
520        :param url: where the broker is, so that a plaintext connection is given
521            no context at all rather than one it would silently drop
522        :returns: keyword arguments for :func:`aio_pika.connect_robust`
523        """
524        if urlsplit(url).scheme != "amqps":
525            return {}
526        return {"ssl_context": self.ssl_context()}
527
528
529def _refuse_a_development_peer(presented: Sequence[bytes]) -> None:
530    """Fails the handshake when the broker presented a development certificate.
531
532    The check runs on every path, ``NOTHING_AT_ALL`` included, and reads the
533    encoded certificates rather than a verified chain: under that setting there
534    is no verified chain, and that is precisely the configuration in which a
535    development certificate is most likely to slip through — a laptop's settings
536    copied into a deployment, where "it works" and "it is safe" have quietly
537    stopped meaning the same thing.
538
539    :param presented: the peer's certificates, DER-encoded
540    :raises ssl.SSLCertVerificationError: when any of them carries the marker
541    """
542    for certificate in presented:
543        if is_development_certificate(certificate):
544            # An ``SSLCertVerificationError`` rather than this library's own
545            # exception because it is raised from inside the handshake, where
546            # anything else would be reported by asyncio as an unexpected
547            # failure rather than as a refused certificate.
548            raise ssl.SSLCertVerificationError(
549                f"acemq: the broker presented a certificate marked {DEVELOPMENT_MARKER!r}. "
550                "It was generated for development, its signing key is not a secret, and it "
551                "is refused however trust is configured. If this really is a development "
552                "broker, say so with Security(allow_development_certificates=True)"
553            )
554
555
556def _certificates_presented_by(connection: ssl.SSLObject | ssl.SSLSocket) -> Sequence[bytes]:
557    """The peer's certificates, as DER, whatever the verification mode was.
558
559    ``getpeercert(binary_form=True)`` answers under ``CERT_NONE``, where the
560    parsed form is empty — which is the whole reason the marker is looked for in
561    the encoding. The full chain is available from Python 3.13 and the leaf alone
562    before it; the leaf is enough, because its own encoding carries its issuer's
563    distinguished name as well as its own, so a real certificate signed by a
564    development authority is caught by it too.
565    """
566    # Python 3.13 and later. What it hands back has changed between releases —
567    # DER ``bytes`` in some, objects with ``public_bytes()`` in others — so each
568    # entry is normalised rather than assumed. Either encoding reads the same to
569    # :func:`is_development_certificate`.
570    unverified: Callable[[], Any] | None = getattr(connection, "get_unverified_chain", None)
571    if unverified is not None:  # pragma: no cover - version-dependent
572        try:
573            chain = list(unverified() or ())
574        except (ValueError, ssl.SSLError, TypeError):
575            chain = []
576        encoded = [_encoded(certificate) for certificate in chain]
577        found = [certificate for certificate in encoded if certificate]
578        if found:
579            return found
580
581    leaf = connection.getpeercert(binary_form=True)
582    return [] if leaf is None else [leaf]
583
584
585def _encoded(certificate: Any) -> bytes:
586    """One certificate from a chain as bytes, whatever shape it arrived in."""
587    if isinstance(certificate, bytes | bytearray):
588        return bytes(certificate)
589    rendered = getattr(certificate, "public_bytes", None)
590    if rendered is None:  # pragma: no cover - no released Python does this
591        return b""
592    written = rendered()
593    return written.encode("ascii") if isinstance(written, str) else bytes(written)
594
595
596class _RefusingSSLObject(ssl.SSLObject):
597    """An :class:`ssl.SSLObject` that will not finish a development handshake."""
598
599    def do_handshake(self) -> None:
600        super().do_handshake()
601        _refuse_a_development_peer(_certificates_presented_by(self))
602
603
604class _RefusingSSLSocket(ssl.SSLSocket):
605    """The same, for a blocking socket. See :class:`_RefusingSSLObject`."""
606
607    def do_handshake(self, block: bool = False) -> None:
608        super().do_handshake(block)
609        _refuse_a_development_peer(_certificates_presented_by(self))
610
611
612class _ContextNamingTheBroker(ssl.SSLContext):
613    """A context that checks the certificate against a name of our choosing.
614
615    The name a certificate is checked against is not a property of the context;
616    it is passed at the moment of the handshake, and asyncio takes it from the
617    host in the URL. That is the right default and the wrong answer for a broker
618    reached by IP address or through a tunnel, where the address dialled and the
619    name on the certificate are different strings on purpose.
620
621    Overriding it here rather than plumbing an argument through the client is
622    the only route available: aio-pika keeps a fixed list of the parameters it
623    forwards, and an unknown one is dropped rather than passed on — which would
624    make a ``server_name`` that was configured and never used, the exact failure
625    this module refuses to have elsewhere.
626    """
627
628    expected_name: str
629
630    def wrap_bio(
631        self,
632        incoming: ssl.MemoryBIO,
633        outgoing: ssl.MemoryBIO,
634        server_side: bool = False,
635        server_hostname: str | bytes | None = None,
636        session: ssl.SSLSession | None = None,
637    ) -> ssl.SSLObject:
638        return super().wrap_bio(
639            incoming,
640            outgoing,
641            server_side=server_side,
642            server_hostname=self.expected_name,
643            session=session,
644        )
645
646
647def without_verifying_the_broker(
648    *,
649    client_certificate: FilePath | None = None,
650    client_key: FilePath | None = None,
651    client_key_password: str | None = None,
652    credentials: Credentials | CredentialsSource | None = None,
653) -> Security:
654    """Encrypts the connection and checks nothing about who is on the other end.
655
656    **This gives up most of what TLS is for, and it is worth being exact about
657    which part.** The traffic cannot be read by somebody watching the network.
658    Nothing whatsoever stops that somebody from *being* the broker: any
659    certificate is accepted, from any issuer, for any name, so an attacker who
660    can answer on the address in your URL receives every message you publish and
661    every password you log in with, over a connection that looks encrypted in
662    every log and every metric.
663
664    It exists because there is one situation where the alternative is worse: a
665    broker with a self-signed certificate on a laptop or in a test fixture,
666    where the choice is otherwise between this and turning encryption off
667    entirely. It is deliberately a function with a long name rather than a
668    ``verify=False``, because it should be impossible to end up here by pasting
669    a keyword argument and difficult to leave in a file nobody rereads.
670
671    **It is wrong in production, always.** Against a real broker, name the
672    authority instead — ``Security(certificate_authority="ca.crt")`` — which is
673    one more line and gives back everything this gives up. Self-signed
674    certificates are fine there too: put the broker's own certificate in the
675    file and it becomes the one authority you trust.
676
677    :param client_certificate: a PEM certificate to present
678    :param client_key: the PEM private key for it
679    :param client_key_password: the passphrase on that key, when it has one
680    :param credentials: the broker login, or a source asked for it
681    :returns: the configuration
682    """
683    return Security(
684        client_certificate=client_certificate,
685        client_key=client_key,
686        client_key_password=client_key_password,
687        credentials=credentials,
688        verification=Verification.NOTHING_AT_ALL,
689    )
690
691
692def _with_credentials(url: str, credentials: Credentials) -> str:
693    """Puts a username and password into a URL, replacing what was there.
694
695    Percent-encoded, because a password is chosen by a person or a generator and
696    neither is under any obligation to avoid ``@``, ``/`` or ``:`` — and a
697    password containing one of those, spliced in raw, produces a URL that parses
698    into a different host.
699    """
700    parts = urlsplit(url)
701    host = parts.hostname or ""
702    if ":" in host:
703        # An IPv6 literal, which urlsplit hands back without the brackets that
704        # are the only thing separating its colons from the port's.
705        host = f"[{host}]"
706    if parts.port is not None:
707        host = f"{host}:{parts.port}"
708
709    login = f"{quote(credentials.username, safe='')}:{quote(credentials.secret, safe='')}"
710    return urlunsplit(
711        (parts.scheme, f"{login}@{host}", parts.path, parts.query, parts.fragment)
712    )