text: # -*- coding: utf-8 -*-
  # Copyright 2014-2016 OpenMarket Ltd
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  import logging
  import os
  import warnings
  from datetime import datetime
  from hashlib import sha256
  from typing import List
  
  import six
  
  from unpaddedbase64 import encode_base64
  
  from OpenSSL import SSL, crypto
  from twisted.internet._sslverify import Certificate, trustRootFromCertificates
  
  from synapse.config._base import Config, ConfigError
  from synapse.util import glob_to_regex
  
  logger = logging.getLogger(__name__)
  
  
  class TlsConfig(Config):
      section = "tls"
  
      def read_config(self, config: dict, config_dir_path: str, **kwargs):
  
          acme_config = config.get("acme", None)
          if acme_config is None:
              acme_config = {}
  
          self.acme_enabled = acme_config.get("enabled", False)
  
          # hyperlink complains on py2 if this is not a Unicode
          self.acme_url = six.text_type(
              acme_config.get("url", "https://acme-v01.api.letsencrypt.org/directory")
          )
          self.acme_port = acme_config.get("port", 80)
          self.acme_bind_addresses = acme_config.get("bind_addresses", ["::", "0.0.0.0"])
          self.acme_reprovision_threshold = acme_config.get("reprovision_threshold", 30)
          self.acme_domain = acme_config.get("domain", config.get("server_name"))
  
          self.acme_account_key_file = self.abspath(
              acme_config.get("account_key_file", config_dir_path + "/client.key")
          )
  
          self.tls_certificate_file = self.abspath(config.get("tls_certificate_path"))
          self.tls_private_key_file = self.abspath(config.get("tls_private_key_path"))
  
          if self.root.server.has_tls_listener():
              if not self.tls_certificate_file:
                  raise ConfigError(
                      "tls_certificate_path must be specified if TLS-enabled listeners are "
                      "configured."
                  )
              if not self.tls_private_key_file:
                  raise ConfigError(
                      "tls_private_key_path must be specified if TLS-enabled listeners are "
                      "configured."
                  )
  
          self._original_tls_fingerprints = config.get("tls_fingerprints", [])
  
          if self._original_tls_fingerprints is None:
              self._original_tls_fingerprints = []
  
          self.tls_fingerprints = list(self._original_tls_fingerprints)
  
          # Whether to verify certificates on outbound federation traffic
          self.federation_verify_certificates = config.get(
              "federation_verify_certificates", True
          )
  
          # Minimum TLS version to use for outbound federation traffic
          self.federation_client_minimum_tls_version = str(
              config.get("federation_client_minimum_tls_version", 1)
          )
  
          if self.federation_client_minimum_tls_version not in ["1", "1.1", "1.2", "1.3"]:
              raise ConfigError(
                  "federation_client_minimum_tls_version must be one of: 1, 1.1, 1.2, 1.3"
              )
  
          # Prevent people shooting themselves in the foot here by setting it to
          # the biggest number blindly
          if self.federation_client_minimum_tls_version == "1.3":
              if getattr(SSL, "OP_NO_TLSv1_3", None) is None:
                  raise ConfigError(
                      (
                          "federation_client_minimum_tls_version cannot be 1.3, "
                          "your OpenSSL does not support it"
                      )
                  )
  
          # Whitelist of domains to not verify certificates for
          fed_whitelist_entries = config.get(
              "federation_certificate_verification_whitelist", []
          )
  
          # Support globs (*) in whitelist values
          self.federation_certificate_verification_whitelist = []  # type: List[str]
          for entry in fed_whitelist_entries:
              try:
                  entry_regex = glob_to_regex(entry.encode("ascii").decode("ascii"))
              except UnicodeEncodeError:
                  raise ConfigError(
                      "IDNA domain names are not allowed in the "
                      "federation_certificate_verification_whitelist: %s" % (entry,)
                  )
  
              # Convert globs to regex
              self.federation_certificate_verification_whitelist.append(entry_regex)
  
          # List of custom certificate authorities for federation traffic validation
          custom_ca_list = config.get("federation_custom_ca_list", None)
  
          # Read in and parse custom CA certificates
          self.federation_ca_trust_root = None
          if custom_ca_list is not None:
              if len(custom_ca_list) == 0:
                  # A trustroot cannot be generated without any CA certificates.
                  # Raise an error if this option has been specified without any
                  # corresponding certificates.
                  raise ConfigError(
                      "federation_custom_ca_list specified without "
                      "any certificate files"
                  )
  
              certs = []
              for ca_file in custom_ca_list:
                  logger.debug("Reading custom CA certificate file: %s", ca_file)
                  content = self.read_file(ca_file, "federation_custom_ca_list")
  
                  # Parse the CA certificates
                  try:
                      cert_base = Certificate.loadPEM(content)
                      certs.append(cert_base)
                  except Exception as e:
                      raise ConfigError(
                          "Error parsing custom CA certificate file %s: %s" % (ca_file, e)
                      )
  
              self.federation_ca_trust_root = trustRootFromCertificates(certs)
  
          # This config option applies to non-federation HTTP clients
          # (e.g. for talking to recaptcha, identity servers, and such)
          # It should never be used in production, and is intended for
          # use only when running tests.
          self.use_insecure_ssl_client_just_for_testing_do_not_use = config.get(
              "use_insecure_ssl_client_just_for_testing_do_not_use"
          )
  
          self.tls_certificate = None
          self.tls_private_key = None
  
      def is_disk_cert_valid(self, allow_self_signed=True):
          """
          Is the certificate we have on disk valid, and if so, for how long?
  
          Args:
              allow_self_signed (bool): Should we allow the certificate we
                  read to be self signed?
  
          Returns:
              int: Days remaining of certificate validity.
              None: No certificate exists.
          """
          if not os.path.exists(self.tls_certificate_file):
              return None
  
          try:
              with open(self.tls_certificate_file, "rb") as f:
                  cert_pem = f.read()
          except Exception as e:
              raise ConfigError(
                  "Failed to read existing certificate file %s: %s"
                  % (self.tls_certificate_file, e)
              )
  
          try:
              tls_certificate = crypto.load_certificate(crypto.FILETYPE_PEM, cert_pem)
          except Exception as e:
              raise ConfigError(
                  "Failed to parse existing certificate file %s: %s"
                  % (self.tls_certificate_file, e)
              )
  
          if not allow_self_signed:
              if tls_certificate.get_subject() == tls_certificate.get_issuer():
                  raise ValueError(
                      "TLS Certificate is self signed, and this is not permitted"
                  )
  
          # YYYYMMDDhhmmssZ -- in UTC
          expires_on = datetime.strptime(
              tls_certificate.get_notAfter().decode("ascii"), "%Y%m%d%H%M%SZ"
          )
          now = datetime.utcnow()
          days_remaining = (expires_on - now).days
          return days_remaining
  
      def read_certificate_from_disk(self, require_cert_and_key):
          """
          Read the certificates and private key from disk.
  
          Args:
              require_cert_and_key (bool): set to True to throw an error if the certificate
                  and key file are not given
          """
          if require_cert_and_key:
              self.tls_private_key = self.read_tls_private_key()
              self.tls_certificate = self.read_tls_certificate()
          elif self.tls_certificate_file:
              # we only need the certificate for the tls_fingerprints. Reload it if we
              # can, but it's not a fatal error if we can't.
              try:
                  self.tls_certificate = self.read_tls_certificate()
              except Exception as e:
                  logger.info(
                      "Unable to read TLS certificate (%s). Ignoring as no "
                      "tls listeners enabled.",
                      e,
                  )
  
          self.tls_fingerprints = list(self._original_tls_fingerprints)
  
          if self.tls_certificate:
              # Check that our own certificate is included in the list of fingerprints
              # and include it if it is not.
              x509_certificate_bytes = crypto.dump_certificate(
                  crypto.FILETYPE_ASN1, self.tls_certificate
              )
              sha256_fingerprint = encode_base64(sha256(x509_certificate_bytes).digest())
              sha256_fingerprints = set(f["sha256"] for f in self.tls_fingerprints)
              if sha256_fingerprint not in sha256_fingerprints:
                  self.tls_fingerprints.append({"sha256": sha256_fingerprint})
  
      def generate_config_section(
          self,
          config_dir_path,
          server_name,
          data_dir_path,
          tls_certificate_path,
          tls_private_key_path,
          acme_domain,
          **kwargs
      ):
          """If the acme_domain is specified acme will be enabled.
          If the TLS paths are not specified the default will be certs in the
          config directory"""
  
          base_key_name = os.path.join(config_dir_path, server_name)
  
          if bool(tls_certificate_path) != bool(tls_private_key_path):
              raise ConfigError(
                  "Please specify both a cert path and a key path or neither."
              )
  
          tls_enabled = (
              "" if tls_certificate_path and tls_private_key_path or acme_domain else "#"
          )
  
          if not tls_certificate_path:
              tls_certificate_path = base_key_name + ".tls.crt"
          if not tls_private_key_path:
              tls_private_key_path = base_key_name + ".tls.key"
  
          acme_enabled = bool(acme_domain)
          acme_domain = "matrix.example.com"
  
          default_acme_account_file = os.path.join(data_dir_path, "acme_account.key")
  
          # this is to avoid the max line length. Sorrynotsorry
          proxypassline = (
              "ProxyPass /.well-known/acme-challenge "
              "http://localhost:8009/.well-known/acme-challenge"
          )
  
          # flake8 doesn't recognise that variables are used in the below string
          _ = tls_enabled, proxypassline, acme_enabled, default_acme_account_file
  
          return (
              """\
          ## TLS ##
  
          # PEM-encoded X509 certificate for TLS.
          # This certificate, as of Synapse 1.0, will need to be a valid and verifiable
          # certificate, signed by a recognised Certificate Authority.
          #
          # See 'ACME support' below to enable auto-provisioning this certificate via
          # Let's Encrypt.
          #
          # If supplying your own, be sure to use a `.pem` file that includes the
          # full certificate chain including any intermediate certificates (for
          # instance, if using certbot, use `fullchain.pem` as your certificate,
          # not `cert.pem`).
          #
          %(tls_enabled)stls_certificate_path: "%(tls_certificate_path)s"
  
          # PEM-encoded private key for TLS
          #
          %(tls_enabled)stls_private_key_path: "%(tls_private_key_path)s"
  
          # Whether to verify TLS server certificates for outbound federation requests.
          #
          # Defaults to `true`. To disable certificate verification, uncomment the
          # following line.
          #
          #federation_verify_certificates: false
  
          # The minimum TLS version that will be used for outbound federation requests.
          #
          # Defaults to `1`. Configurable to `1`, `1.1`, `1.2`, or `1.3`. Note
          # that setting this value higher than `1.2` will prevent federation to most
          # of the public Matrix network: only configure it to `1.3` if you have an
          # entirely private federation setup and you can ensure TLS 1.3 support.
          #
          #federation_client_minimum_tls_version: 1.2
  
          # Skip federation certificate verification on the following whitelist
          # of domains.
          #
          # This setting should only be used in very specific cases, such as
          # federation over Tor hidden services and similar. For private networks
          # of homeservers, you likely want to use a private CA instead.
          #
          # Only effective if federation_verify_certicates is `true`.
          #
          #federation_certificate_verification_whitelist:
          #  - lon.example.com
          #  - *.domain.com
          #  - *.onion
  
          # List of custom certificate authorities for federation traffic.
          #
          # This setting should only normally be used within a private network of
          # homeservers.
          #
          # Note that this list will replace those that are provided by your
          # operating environment. Certificates must be in PEM format.
          #
          #federation_custom_ca_list:
          #  - myCA1.pem
          #  - myCA2.pem
          #  - myCA3.pem
  
          # ACME support: This will configure Synapse to request a valid TLS certificate
          # for your configured `server_name` via Let's Encrypt.
          #
          # Note that provisioning a certificate in this way requires port 80 to be
          # routed to Synapse so that it can complete the http-01 ACME challenge.
          # By default, if you enable ACME support, Synapse will attempt to listen on
          # port 80 for incoming http-01 challenges - however, this will likely fail
          # with 'Permission denied' or a similar error.
          #
          # There are a couple of potential solutions to this:
          #
          #  * If you already have an Apache, Nginx, or similar listening on port 80,
          #    you can configure Synapse to use an alternate port, and have your web
          #    server forward the requests. For example, assuming you set 'port: 8009'
          #    below, on Apache, you would write:
          #
          #    %(proxypassline)s
          #
          #  * Alternatively, you can use something like `authbind` to give Synapse
          #    permission to listen on port 80.
          #
          acme:
              # ACME support is disabled by default. Set this to `true` and uncomment
              # tls_certificate_path and tls_private_key_path above to enable it.
              #
              enabled: %(acme_enabled)s
  
              # Endpoint to use to request certificates. If you only want to test,
              # use Let's Encrypt's staging url:
              #     https://acme-staging.api.letsencrypt.org/directory
              #
              #url: https://acme-v01.api.letsencrypt.org/directory
  
              # Port number to listen on for the HTTP-01 challenge. Change this if
              # you are forwarding connections through Apache/Nginx/etc.
              #
              port: 80
  
              # Local addresses to listen on for incoming connections.
              # Again, you may want to change this if you are forwarding connections
              # through Apache/Nginx/etc.
              #
              bind_addresses: ['::', '0.0.0.0']
  
              # How many days remaining on a certificate before it is renewed.
              #
              reprovision_threshold: 30
  
              # The domain that the certificate should be for. Normally this
              # should be the same as your Matrix domain (i.e., 'server_name'), but,
              # by putting a file at 'https://<server_name>/.well-known/matrix/server',
              # you can delegate incoming traffic to another server. If you do that,
              # you should give the target of the delegation here.
              #
              # For example: if your 'server_name' is 'example.com', but
              # 'https://example.com/.well-known/matrix/server' delegates to
              # 'matrix.example.com', you should put 'matrix.example.com' here.
              #
              # If not set, defaults to your 'server_name'.
              #
              domain: %(acme_domain)s
  
              # file to use for the account key. This will be generated if it doesn't
              # exist.
              #
              # If unspecified, we will use CONFDIR/client.key.
              #
              account_key_file: %(default_acme_account_file)s
  
          # List of allowed TLS fingerprints for this server to publish along
          # with the signing keys for this server. Other matrix servers that
          # make HTTPS requests to this server will check that the TLS
          # certificates returned by this server match one of the fingerprints.
          #
          # Synapse automatically adds the fingerprint of its own certificate
          # to the list. So if federation traffic is handled directly by synapse
          # then no modification to the list is required.
          #
          # If synapse is run behind a load balancer that handles the TLS then it
          # will be necessary to add the fingerprints of the certificates used by
          # the loadbalancers to this list if they are different to the one
          # synapse is using.
          #
          # Homeservers are permitted to cache the list of TLS fingerprints
          # returned in the key responses up to the "valid_until_ts" returned in
          # key. It may be necessary to publish the fingerprints of a new
          # certificate and wait until the "valid_until_ts" of the previous key
          # responses have passed before deploying it.
          #
          # You can calculate a fingerprint from a given TLS listener via:
          # openssl s_client -connect $host:$port < /dev/null 2> /dev/null |
          #   openssl x509 -outform DER | openssl sha256 -binary | base64 | tr -d '='
          # or by checking matrix.org/federationtester/api/report?server_name=$host
          #
          #tls_fingerprints: [{"sha256": "<base64_encoded_sha256_fingerprint>"}]
  
          """
              # Lowercase the string representation of boolean values
              % {
                  x[0]: str(x[1]).lower() if isinstance(x[1], bool) else x[1]
                  for x in locals().items()
              }
          )
  
      def read_tls_certificate(self):
          """Reads the TLS certificate from the configured file, and returns it
  
          Also checks if it is self-signed, and warns if so
  
          Returns:
              OpenSSL.crypto.X509: the certificate
          """
          cert_path = self.tls_certificate_file
          logger.info("Loading TLS certificate from %s", cert_path)
          cert_pem = self.read_file(cert_path, "tls_certificate_path")
          cert = crypto.load_certificate(crypto.FILETYPE_PEM, cert_pem)
  
          # Check if it is self-signed, and issue a warning if so.
          if cert.get_issuer() == cert.get_subject():
              warnings.warn(
                  (
                      "Self-signed TLS certificates will not be accepted by Synapse 1.0. "
                      "Please either provide a valid certificate, or use Synapse's ACME "
                      "support to provision one."
                  )
              )
  
          return cert
  
      def read_tls_private_key(self):
          """Reads the TLS private key from the configured file, and returns it
  
          Returns:
              OpenSSL.crypto.PKey: the private key
          """
          private_key_path = self.tls_private_key_file
          logger.info("Loading TLS key from %s", private_key_path)
          private_key_pem = self.read_file(private_key_path, "tls_private_key_path")
          return crypto.load_privatekey(crypto.FILETYPE_PEM, private_key_pem)
  