diff --git a/reviewboard/certs/manager.py b/reviewboard/certs/manager.py index f591d1c3ce183e4dab792d7819f1b57d164739f5..089ca35c8c4809df13439ab99145a9358d7bd42e 100644 --- a/reviewboard/certs/manager.py +++ b/reviewboard/certs/manager.py @@ -108,19 +108,12 @@ class CertificateManager: # Instance variables # ###################### - #: The root path for certificate storage. - #: - #: All storage backends will have a subdirectory within here that they - #: can use for reading/writing certificate storage data. - _root_storage_path: str - #: A loaded instance of the current storage backend. _storage_backend: _CertStorageBackend | None def __init__(self) -> None: """Initialize the certificate manager.""" self._storage_backend = None - self._root_storage_path = os.path.join(get_data_dir(), 'rb-certs') @property def storage_backend(self) -> _CertStorageBackend: @@ -171,7 +164,7 @@ class CertificateManager: assert backend_cls is not None assert backend_cls.backend_id - storage_path = safe_join(self._root_storage_path, + storage_path = safe_join(get_data_dir(), 'rb-certs', slugify(backend_cls.backend_id)) backend = backend_cls(storage_path=storage_path) diff --git a/reviewboard/certs/tests/test_certificate_manager.py b/reviewboard/certs/tests/test_certificate_manager.py index 41a062fd5a8dcd051bf01f84d944b615f664b9ee..e170d68c0870b2696318ee521a2849114e6e0a41 100644 --- a/reviewboard/certs/tests/test_certificate_manager.py +++ b/reviewboard/certs/tests/test_certificate_manager.py @@ -24,7 +24,8 @@ from reviewboard.certs.manager import CertificateManager, logger from reviewboard.certs.storage import cert_storage_backend_registry from reviewboard.certs.storage.file_storage import \ FileCertificateStorageBackend -from reviewboard.certs.tests.testcases import (CertificateTestCase, +from reviewboard.certs.tests.testcases import (CaptureSSLContext, + CertificateTestCase, TEST_CERT_BUNDLE_PEM, TEST_CLIENT_CERT_PEM, TEST_CLIENT_KEY_PEM, @@ -38,48 +39,6 @@ class MyCertificateStorageBackend(FileCertificateStorageBackend): backend_id = 'test' -class MySSLContext: - cadatas: list[bytes | str | None] - cafiles: list[str | None] - capaths: list[str | None] - certfiles: list[str | None] - keyfiles: list[str | None] - passwords: list[str | None] - - def __init__(self) -> None: - self.cadatas = [] - self.cafiles = [] - self.capaths = [] - self.certfiles = [] - self.keyfiles = [] - self.passwords = [] - - def load_verify_locations( - self, - cafile: (str | None) = None, - capath: (str | None) = None, - cadata: (bytes | str | None) = None - ) -> None: - if cafile: - self.cafiles.append(cafile) - - if capath: - self.capaths.append(capath) - - if cadata: - self.cadatas.append(cadata) - - def load_cert_chain( - self, - certfile: str, - keyfile: (str | None) = None, - password: (str | None) = None, - ) -> None: - self.certfiles.append(certfile) - self.keyfiles.append(keyfile) - self.passwords.append(password) - - class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): """Unit tests for CertificateManager. @@ -1490,11 +1449,11 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): storage_backend = cert_manager.storage_backend self.spy_on(ssl.create_default_context, - op=kgb.SpyOpReturn(MySSLContext())) + op=kgb.SpyOpReturn(CaptureSSLContext())) context = cert_manager.build_ssl_context(hostname='example.com', port=443) - assert isinstance(context, MySSLContext) + assert isinstance(context, CaptureSSLContext) self.assertAttrsEqual( context, @@ -1515,7 +1474,7 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): storage_backend = cert_manager.storage_backend self.spy_on(ssl.create_default_context, - op=kgb.SpyOpReturn(MySSLContext())) + op=kgb.SpyOpReturn(CaptureSSLContext())) cert_manager.add_ca_bundle( CertificateBundle(bundle_data=TEST_CERT_BUNDLE_PEM, @@ -1523,7 +1482,7 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): context = cert_manager.build_ssl_context(hostname='example.com', port=443) - assert isinstance(context, MySSLContext) + assert isinstance(context, CaptureSSLContext) self.assertAttrsEqual( context, @@ -1546,14 +1505,14 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): storage_backend = cert_manager.storage_backend self.spy_on(ssl.create_default_context, - op=kgb.SpyOpReturn(MySSLContext())) + op=kgb.SpyOpReturn(CaptureSSLContext())) cert_manager.add_certificate( self.create_certificate(cert_data=TEST_TRUST_CERT_PEM)) context = cert_manager.build_ssl_context(hostname='example.com', port=443) - assert isinstance(context, MySSLContext) + assert isinstance(context, CaptureSSLContext) self.assertAttrsEqual( context, @@ -1579,7 +1538,7 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): storage_backend = cert_manager.storage_backend self.spy_on(ssl.create_default_context, - op=kgb.SpyOpReturn(MySSLContext())) + op=kgb.SpyOpReturn(CaptureSSLContext())) cert_manager.add_certificate( self.create_certificate(purpose=CertPurpose.CLIENT, @@ -1588,7 +1547,7 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): context = cert_manager.build_ssl_context(hostname='example.com', port=443) - assert isinstance(context, MySSLContext) + assert isinstance(context, CaptureSSLContext) self.assertAttrsEqual( context, @@ -1619,7 +1578,7 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): storage_backend = cert_manager.storage_backend self.spy_on(ssl.create_default_context, - op=kgb.SpyOpReturn(MySSLContext())) + op=kgb.SpyOpReturn(CaptureSSLContext())) cert_manager.add_ca_bundle( CertificateBundle(bundle_data=TEST_CERT_BUNDLE_PEM, @@ -1633,7 +1592,7 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): context = cert_manager.build_ssl_context(hostname='example.com', port=443) - assert isinstance(context, MySSLContext) + assert isinstance(context, CaptureSSLContext) self.assertAttrsEqual( context, @@ -1669,7 +1628,7 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): 'test-site-1') self.spy_on(ssl.create_default_context, - op=kgb.SpyOpReturn(MySSLContext())) + op=kgb.SpyOpReturn(CaptureSSLContext())) cert_manager.add_certificate( self.create_certificate(cert_data=TEST_TRUST_CERT_PEM), @@ -1699,7 +1658,7 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): context = cert_manager.build_ssl_context(hostname='example.com', port=443, local_site=local_site) - assert isinstance(context, MySSLContext) + assert isinstance(context, CaptureSSLContext) self.assertAttrsEqual( context, @@ -1733,7 +1692,7 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): storage_backend = cert_manager.storage_backend self.spy_on(ssl.create_default_context, - op=kgb.SpyOpReturn(MySSLContext())) + op=kgb.SpyOpReturn(CaptureSSLContext())) cert_manager.add_certificate( self.create_certificate(cert_data=TEST_TRUST_CERT_PEM)) @@ -1764,7 +1723,7 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): storage_backend = cert_manager.storage_backend self.spy_on(ssl.create_default_context, - op=kgb.SpyOpReturn(MySSLContext())) + op=kgb.SpyOpReturn(CaptureSSLContext())) cert_manager.add_certificate( self.create_certificate(purpose=CertPurpose.CLIENT, @@ -1800,7 +1759,7 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): storage_backend = cert_manager.storage_backend self.spy_on(ssl.create_default_context, - op=kgb.SpyOpReturn(MySSLContext())) + op=kgb.SpyOpReturn(CaptureSSLContext())) cert_manager.add_certificate( self.create_certificate(cert_data=TEST_TRUST_CERT_PEM, @@ -1830,7 +1789,7 @@ class CertificateManagerTests(kgb.SpyAgency, CertificateTestCase): cert_manager = CertificateManager() self.spy_on(ssl.create_default_context, - op=kgb.SpyOpReturn(MySSLContext())) + op=kgb.SpyOpReturn(CaptureSSLContext())) kwargs = cert_manager.build_urlopen_kwargs(url='http://example.com') self.assertEqual(kwargs, {}) diff --git a/reviewboard/certs/tests/testcases.py b/reviewboard/certs/tests/testcases.py index 18975f5c3287c218af7b6c91c8bba035e2646917..ead190f1696a34940f86a6fb6e9bc45697ab1b4c 100644 --- a/reviewboard/certs/tests/testcases.py +++ b/reviewboard/certs/tests/testcases.py @@ -51,6 +51,39 @@ e8K8nyAoCVl6E0HJL7cKdQ9/SkCivQjCO5jK+s8ANOeBNDUSATgQvep9VAk8UWo+ More junk... """ +TEST_TRUST_SAN_CERT_PEM = b""" +Junk area. This is a trust cert with SAN data. + +-----BEGIN CERTIFICATE----- +MIIEXjCCAkagAwIBAgIUH/aJUMZgutxwbM0xXq8DRrfVR+owDQYJKoZIhvcNAQEL +BQAwFzEVMBMGA1UEAwwMVGVzdCBMREFQIENBMB4XDTI2MDMxODAzMjc0OVoXDTM2 +MDMxNTAzMjc0OVowGzEZMBcGA1UEAwwQbWFpbi5leGFtcGxlLmNvbTCCASIwDQYJ +KoZIhvcNAQEBBQADggEPADCCAQoCggEBAL0eN9jF/qSqGXKVxu4T9gy+gt2JQYIK +AyemKCs9VZPHTcl+I1pHGGF2wwDDFne9i0JfywSXioLmnkKVar91YlgoCumqJqSQ +5KskzarN5bGy+2V8wU/uhP4sD+gocVxFWPBLAQvJAMwaKtYBtUL7U9Ko9T9BaYDy +pdk+0scFBgfpPYBzcB0UaDIhQ1aM2wd5eih1Gk66N0ZlWwJ4M4Cvtn2L+a10SoXW +ipSFOeeU6LBMTl1vTgsok+4FF/Bf1BqdOTIXRqCXgeq4+v2OnpCLm0G81tyYj1lx +7gnpTvyHpVTR1FUWnFI1zJ3V1RceifgBJoSHYSytfd1maF1eXgDJndECAwEAAaOB +nTCBmjAJBgNVHRMEAjAAMA4GA1UdDwEB/wQEAwIFoDATBgNVHSUEDDAKBggrBgEF +BQcDATAoBgNVHREEITAfghB0ZXN0LmV4YW1wbGUuY29tggtleGFtcGxlLmNvbTAd +BgNVHQ4EFgQURPbKJRFrb6AbUB4GuvEWMIKwCFcwHwYDVR0jBBgwFoAUFZ8R0Mgm +CmxbG0buLdWs5L+1kT0wDQYJKoZIhvcNAQELBQADggIBAA+7n5ZP1fZRtP5mMYl/ +U67brXmg9xToesA2vNxyBv9Mb951bqVZKJgGS6i8BkkhhEo90rvv8eIQjfb/owcu +U1hWh+aMXCebrD9T8neTBk+I6lHjbxKFzc3L0uUNmu6fYTqY0hKnsnPgm93V0edd +ObpImOna9vVWtXTsfP0mY9OaaitC4jpaWXSZglTshv0mI23BP3PTCCDn4stY2njt +Oq3j1k+XKjb4p5BTdymDiYh7BdPxIerRbNfHNXwm6YdW30FEdOG5OWj6cHqWNo6n +3+kOAnZ13rasGHgkwZwc5Wzu+U/1uTsjSX35TMbVBIztfZEcTWStqGGbmIp7dXfg +S0esl8aeptVRgcr2t0htw3LkkBmIo6/UBvVbqfJKlqJp5M/sJVgGWZ7rHUk7odhm +8Fw28Hi6R/LD1k5M+SWzQVLZxGMpLObtwRki5lj9FBpSNCctOXh6CwnPmgnkcW9c +ZVqWsFbYzMKGdcDm+azKLR43xHpnbq9ZC+GBfS1dkMRY6Ax13AAHvANBI79hXMJq +pMwwml7ru5CfRaBL/w0VKvaCVza7zRr+DEB9wVTA2JjfLdfMY8raq9jETEcx8n+H +LgIABEl9Djw4VXkTqTSHNbRSKg6IzOK3Ve0rJruJcqPUiCea9c8rwwW3+dBkP3Kr +gynA+/8MseuiQ2HUkoEZOkIy +-----END CERTIFICATE----- + +More junk... +""" + TEST_CLIENT_CERT_PEM = b""" Junk area. This is a client auth cert. @@ -236,6 +269,98 @@ TEST_SHA256_2 = ( ) +class CaptureSSLContext: + """SSL context that captures loaded state. + + Version Added: + 8.0 + """ + + ###################### + # Instance variables # + ###################### + + #: Loaded CA bundle strings. + cadatas: list[bytes | str | None] + + #: Loaded CA bundle files. + cafiles: list[str | None] + + #: Loaded CA bundle paths. + capaths: list[str | None] + + #: Loaded trust or mTLS certificate files. + certfiles: list[str | None] + + #: Whether hostnames are checked. + check_hostname: bool + + #: Loaded mTLS key files. + keyfiles: list[str | None] + + #: Loaded mTLS key passwords. + passwords: list[str | None] + + def __init__(self) -> None: + """Initialize the context.""" + self.cadatas = [] + self.cafiles = [] + self.capaths = [] + self.certfiles = [] + self.keyfiles = [] + self.passwords = [] + self.check_hostname = True + + def load_verify_locations( + self, + cafile: (str | None) = None, + capath: (str | None) = None, + cadata: (bytes | str | None) = None + ) -> None: + """Load CA data for verification. + + Args: + cafile (str, optional): + The CA file path to load. + + capath (str, optional): + The CA directory path to load. + + cadata (bytes or str, optional): + The CA file data to load. + """ + if cafile: + self.cafiles.append(cafile) + + if capath: + self.capaths.append(capath) + + if cadata: + self.cadatas.append(cadata) + + def load_cert_chain( + self, + certfile: str, + keyfile: (str | None) = None, + password: (str | None) = None, + ) -> None: + """Load a certificate chain. + + Args: + certfile (str): + The certificate file path. + + keyfile (str, optional): + The key file path. + + password (str, optional): + The key password. + """ + self.certfiles.append(certfile) + self.keyfiles.append(keyfile) + self.passwords.append(password) + + class CertificateTestCase(TestCase): """Base test case for certificate unit tests. diff --git a/reviewboard/hostingsvcs/base/forms.py b/reviewboard/hostingsvcs/base/forms.py index 5647ea20c116d5161cc521d36305a19c1e3ee97f..07adfc3fbffa149db563c3f84e231c11efbc0f2a 100644 --- a/reviewboard/hostingsvcs/base/forms.py +++ b/reviewboard/hostingsvcs/base/forms.py @@ -14,10 +14,14 @@ from django import forms from django.utils.translation import gettext, gettext_lazy as _ from typelets.runtime import raise_invalid_type +from reviewboard.certs.errors import CertificateVerificationError +from reviewboard.certs.manager import cert_manager from reviewboard.hostingsvcs.errors import (AuthorizationError, TwoFactorAuthCodeRequiredError) from reviewboard.hostingsvcs.models import HostingServiceAccount -from reviewboard.scmtools.errors import UnverifiedCertificateError +from reviewboard.scmtools.errors import ( + UnverifiedCertificateError as LegacyUnverifiedCertificateError, +) from reviewboard.scmtools.forms import (BaseRepositoryAuthSubForm, BaseRepositoryInfoSubForm) @@ -442,17 +446,57 @@ class BaseHostingServiceAuthForm(_HostingServiceSubFormMixin, 'credentials': credentials, }, **extra_authorize_kwargs) + reauth = False + try: self.authorize(hosting_account, hosting_service_id, **authorize_kwargs) - except UnverifiedCertificateError as e: + except CertificateVerificationError as e: + if trust_host: + cert = e.certificate + + # In the common case, we'll have a certificate with + # data, but in the event that there's an implementation + # error with some of the data, we'll want to log it + # and raise. + if cert is not None and cert.cert_data: + # Add the certificate and attempt re-authorization. + cert_manager.add_certificate(cert) + reauth = True + else: + # There's an implementation error. Log the details + # and let the exception raise. + if cert is None: + logger.error( + 'Hosting service %r provided a ' + 'CertificateVerificationError without a ' + 'certificate. This is an implementation error ' + 'and must be fixed.', + hosting_service_id, + ) + elif not cert.cert_data: + logger.error( + 'Hosting service %r provided a ' + 'CertificateVerificationError with a ' + 'certificate but without certificate data. ' + 'This is an implementation error and must ' + 'be fixed.', + hosting_service_id, + ) + + if not reauth: + raise + except LegacyUnverifiedCertificateError as e: if trust_host: hosting_account.accept_certificate(e.certificate) - self.authorize(hosting_account, hosting_service_id, - **authorize_kwargs) + reauth = True else: raise + if reauth: + self.authorize(hosting_account, hosting_service_id, + **authorize_kwargs) + if save: hosting_account.save() @@ -506,7 +550,8 @@ class BaseHostingServiceAuthForm(_HostingServiceSubFormMixin, # Re-raise the error. raise - except UnverifiedCertificateError: + except (CertificateVerificationError, + LegacyUnverifiedCertificateError): # Re-raise the error so the user will see the "I trust this # host" prompt. raise diff --git a/reviewboard/hostingsvcs/base/http.py b/reviewboard/hostingsvcs/base/http.py index 37ee582c25eff30d3f5cc693624b9df7b1a47c1a..39ee75123f5d08a1bb8503d11b6ec084ca8e1561 100644 --- a/reviewboard/hostingsvcs/base/http.py +++ b/reviewboard/hostingsvcs/base/http.py @@ -27,6 +27,8 @@ from django.utils.encoding import force_str from djblets.log import log_timed from djblets.util.decorators import cached_property +from reviewboard.certs.cert import Certificate +from reviewboard.certs.manager import cert_manager from reviewboard.deprecation import RemovedInReviewBoard90Warning if TYPE_CHECKING: @@ -36,6 +38,7 @@ if TYPE_CHECKING: from typing_extensions import Never, TypeAlias from reviewboard.hostingsvcs.base.hosting_service import BaseHostingService + from reviewboard.hostingsvcs.models import HostingServiceAccount logger = logging.getLogger(__name__) @@ -90,6 +93,121 @@ QueryArgs: TypeAlias = Dict[str, Any] UploadedFiles: TypeAlias = Dict[Union[bytes, str], UploadedFileInfo] +def _build_ssl_context_from_ssl_cert( + *, + hosting_account: HostingServiceAccount, +) -> ssl.SSLContext: + """Return an SSL context for a hosting service with legacy cert data. + + This will attempt to migrate the stored ``ssl_cert`` data a hosting + service account to a modern stored certificate that can then be + properly managed. + + If there are any issues with migration, or if the resulting SSL + certificate does not match the target hostname, then the stored certificate + data will be used directly with hosting verification disabled. + + This should never be needed for hosting service accounts configured + after Review Board 8.0. + + Version Added: + 8.0 + + Args: + hosting_account (reviewboard.hostingsvcs.models.HostingServiceAccount): + The hosting service account to migrate. + + Returns: + ssl.SSLContext: + A configured SSL context with the certificate trusted. + """ + context: (ssl.SSLContext | None) = None + cert_data = hosting_account.data['ssl_cert'] + + if hosting_url := hosting_account.hosting_url: + local_site = hosting_account.local_site + + try: + parsed = urlparse(hosting_url) + + if hostname := parsed.hostname: + port = parsed.port or 443 + + # Check if there's an existing certificate being managed. + certificate = cert_manager.get_certificate( + hostname=hostname, + port=port, + local_site=local_site, + ) + + if certificate is None: + # There's no existing certificate. Generate one from + # the stored data and check to see if it matches the + # hostname. If not, we can't add it, and instead need + # to go the legacy `check_hostname=False` route. + certificate = Certificate( + hostname=hostname, + port=port, + cert_data=cert_data.encode('ascii'), + ) + + if (certificate.subject == hostname or + hostname in certificate.subject_alternative_names): + # This is a direct match, so add this to the cert + # manager. + cert_manager.add_certificate( + certificate=certificate, + local_site=local_site, + ) + + del hosting_account.data['ssl_cert'] + hosting_account.save(update_fields=('data',)) + else: + # This is NOT a direct match, but the admin had + # previously approved this cert for this server. + # We'll have to keep the legacy fallback that + # disables hostname checks. + logger.warning( + 'The approved SSL/TLS certificate stored in ' + 'hosting service account ID=%r does not match the ' + 'hostname %r for the server. Falling back to a ' + 'less-secure form of verification. Please ensure ' + 'the server has a valid certificate matching its ' + 'hostname and then upload a new certificate.', + hosting_account.pk, hostname, + ) + + certificate = None + + if certificate is not None: + # Use cert_manager to build the SSL context (loads the + # stored cert via load_verify_locations in + # build_ssl_context). + context = cert_manager.build_ssl_context( + hostname=hostname, + port=port, + local_site=local_site, + ) + except Exception: + # This will be issued every time this cert is used, making it + # loud and noisy in order to better catch an admin's attention. + logger.exception( + 'Unexpected error migrating legacy ssl_cert data for hosting ' + 'service account %s. Falling back to a legacy insecure ' + 'SSL context.', + hosting_account.pk, + ) + + if context is None: + # Fall back to using the certificate data directly without hostname + # validation. + context = ssl.create_default_context() + context.load_verify_locations(cadata=cert_data) + context.check_hostname = False + + return context + + def _log_and_raise( value: Never, request: HostingServiceHTTPRequest, @@ -445,18 +563,37 @@ class HostingServiceHTTPRequest: hosting_service = self.hosting_service - if hosting_service and 'ssl_cert' in hosting_service.account.data: - # create_default_context only exists in Python 2.7.9+. Using it - # here should be fine, however, because accepting invalid or - # self-signed certificates is only possible when running - # against versions that have this (see the check for - # create_default_context below). - context = ssl.create_default_context() - context.load_verify_locations( - cadata=hosting_service.account.data['ssl_cert']) - context.check_hostname = False - - self._urlopen_handlers.append(HTTPSHandler(context=context)) + if hosting_service: + context: (ssl.SSLContext | None) + + hosting_account = hosting_service.account + + if 'ssl_cert' in hosting_account.data: + # There's existing legacy SSL certificate data stored in + # the account. Convert it to a modern Certificate if + # possible, and build a context with it. If successful, + # 'ssl_cert' will be removed from the data. + context = _build_ssl_context_from_ssl_cert( + hosting_account=hosting_account, + ) + else: + # This is the modern code path. Build an SSL context. + # + # We use build_urlopen_kwargs() as a convenience. It will + # sanity-check the URL for HTTPS and build a context with the + # right parameters. + context = ( + cert_manager.build_urlopen_kwargs( + url=self.url, + local_site=hosting_service.account.local_site, + ) + .get('context') + ) + + if context is not None: + # An SSL context was successfully built, so we can now set up + # an HTTPS handler using it. + self._urlopen_handlers.append(HTTPSHandler(context=context)) timer_msg = ( f'Performing HTTP {method} request for ' diff --git a/reviewboard/hostingsvcs/models.py b/reviewboard/hostingsvcs/models.py index e09bfa647b355544323cbd3a88d70abace28847a..ac907cfab191f012569105ba7e3648c6ca10a0a6 100644 --- a/reviewboard/hostingsvcs/models.py +++ b/reviewboard/hostingsvcs/models.py @@ -4,11 +4,15 @@ from __future__ import annotations import logging from typing import ClassVar, TYPE_CHECKING +from urllib.parse import urlparse from django.db import models from django.utils.translation import gettext_lazy as _ from djblets.db.fields import JSONField +from reviewboard.certs.cert import Certificate +from reviewboard.certs.manager import cert_manager +from reviewboard.deprecation import RemovedInReviewBoard10_0Warning from reviewboard.hostingsvcs.base import hosting_service_registry from reviewboard.hostingsvcs.errors import MissingHostingServiceError from reviewboard.hostingsvcs.managers import HostingServiceAccountManager @@ -18,7 +22,7 @@ if TYPE_CHECKING: from django.contrib.auth.models import User from reviewboard.hostingsvcs.base.hosting_service import BaseHostingService - from reviewboard.scmtools.certs import Certificate + from reviewboard.scmtools.certs import Certificate as LegacyCertificate logger = logging.getLogger(__name__) @@ -157,10 +161,17 @@ class HostingServiceAccount(models.Model): def accept_certificate( self, - certificate: Certificate, + certificate: LegacyCertificate, ) -> None: """Accept the SSL certificate for the linked hosting URL. + Deprecated: + 8.0: + This has been replaced with + :py:meth:`CertificateManager.add_certificate() + ` + and will be removed in Review Board 10. + Args: certificate (reviewboard.scmtools.certs.Certificate): The certificate to accept. @@ -169,11 +180,78 @@ class HostingServiceAccount(models.Model): ValueError: The certificate data did not include required fields. """ - if not certificate.pem_data: + RemovedInReviewBoard10_0Warning.warn( + 'HostingServiceAccount.accept_certificate() is deprecated and ' + 'will be removed in Review Board 10. Use ' + 'cert_manager.add_certificate() instead.' + ) + + cert_data = certificate.pem_data + + if not cert_data: raise ValueError('The certificate does not include a PEM-encoded ' 'representation.') - self.data['ssl_cert'] = certificate.pem_data + # Also register with the certificate manager so that it can be + # checked against the main fingerprint storage. + hosting_url = self.hosting_url + hostname = certificate.hostname + fingerprint = certificate.fingerprint + port = 443 + + if hosting_url: + try: + parsed = urlparse(hosting_url) + + if parsed.hostname: + hostname = parsed.hostname + + if parsed.port: + port = parsed.port + elif parsed.scheme == 'http': + logger.error( + 'Attempted to accept TLS/SSL certificate for HTTP ' + 'URL %s. This may be a programming error or a ' + 'misconfiguration with a server. A certificate ' + 'will not be added.', + hosting_url, + ) + + return + except Exception as e: + logger.exception( + 'Unexpected error parsing the URL %s when accepting a ' + 'TLS/SSL certificate: %s', + hosting_url, e, + ) + + return + + if not hostname: + logger.error( + 'Could not determine a hostname to use for the TLS/SSL ' + 'certificate accepted for %s. A certificate will not be ' + 'added.', + hosting_url or '', + ) + + return + + try: + cert_manager.add_certificate( + Certificate( + hostname=hostname, + port=port, + cert_data=cert_data.encode('ascii'), + ), + local_site=self.local_site, + ) + except Exception as e: + logger.error( + 'Failed to add SSL certificate for %s:%s ' + '(fingerprint %r): %s', + hostname, port, fingerprint, e, + ) class Meta: """Metadata for the HostingServiceAccount model.""" diff --git a/reviewboard/hostingsvcs/tests/test_hosting_service_auth_form.py b/reviewboard/hostingsvcs/tests/test_hosting_service_auth_form.py index a3ec0598a240769ae80aa5bb4769335280e3c509..2559e993a4e2f507594de7226b24c87e2b72ae3f 100644 --- a/reviewboard/hostingsvcs/tests/test_hosting_service_auth_form.py +++ b/reviewboard/hostingsvcs/tests/test_hosting_service_auth_form.py @@ -1,30 +1,65 @@ +"""Unit tests for BaseHostingServiceAuthForm.""" + +from __future__ import annotations + +import os +import shutil +from typing import TYPE_CHECKING + +import kgb + +from reviewboard.admin.server import get_data_dir +from reviewboard.certs.cert import Certificate, CertificateFingerprints +from reviewboard.certs.errors import (CertificateVerificationError, + CertificateVerificationFailureCode) +from reviewboard.certs.manager import cert_manager +from reviewboard.certs.tests.testcases import TEST_SHA256, TEST_TRUST_CERT_PEM +from reviewboard.deprecation import (RemovedInReviewBoard10_0Warning, + RemovedInReviewBoard90Warning) from reviewboard.hostingsvcs.base import hosting_service_registry from reviewboard.hostingsvcs.errors import (AuthorizationError, TwoFactorAuthCodeRequiredError) from reviewboard.hostingsvcs.base.forms import BaseHostingServiceAuthForm from reviewboard.hostingsvcs.models import HostingServiceAccount +from reviewboard.scmtools.certs import Certificate as LegacyCertificate +from reviewboard.scmtools.errors import UnverifiedCertificateError from reviewboard.site.models import LocalSite from reviewboard.testing import TestCase from reviewboard.testing.hosting_services import (SelfHostedTestService, TestService) +if TYPE_CHECKING: + from djblets.testing.testcases import ExpectedWarning + -class HostingServiceAuthFormTests(TestCase): +class HostingServiceAuthFormTests(kgb.SpyAgency, TestCase): """Unit tests for BaseHostingServiceAuthForm.""" fixtures = ['test_scmtools'] - def setUp(self): - super(HostingServiceAuthFormTests, self).setUp() + def setUp(self) -> None: + """Set up state for the test. + + This will clear out the certs directory before running a test. + """ + super().setUp() hosting_service_registry.register(TestService) hosting_service_registry.register(SelfHostedTestService) + shutil.rmtree(os.path.join(get_data_dir(), 'rb-certs'), + ignore_errors=True) - def tearDown(self): - super(HostingServiceAuthFormTests, self).tearDown() + def tearDown(self) -> None: + """Tear down state for the test. + This will clear out the certs directory after running a test. + """ hosting_service_registry.unregister(SelfHostedTestService) hosting_service_registry.unregister(TestService) + shutil.rmtree(os.path.join(get_data_dir(), 'rb-certs'), + ignore_errors=True) + + super().tearDown() def test_override_help_texts(self): """Testing BaseHostingServiceAuthForm subclasses overriding help texts @@ -431,3 +466,275 @@ class HostingServiceAuthFormTests(TestCase): hosting_account = form.save() self.assertEqual(hosting_account.service_name, 'test') self.assertEqual(hosting_account.username, '2fa-user') + + def test_save_with_cert_error(self) -> None: + """Testing BaseHostingServiceAuthForm.save with + UnverifiedCertificateError + """ + fingerprints = CertificateFingerprints(sha256=TEST_SHA256) + + class _MyAuthForm(BaseHostingServiceAuthForm): + def authorize(self, *args, **kwargs) -> None: + verified = cert_manager.is_certificate_verified( + hostname='example.com', + port=443, + latest_fingerprints=fingerprints, + ) + + if not verified: + raise CertificateVerificationError( + code=CertificateVerificationFailureCode.NOT_TRUSTED, + certificate=Certificate( + hostname='example.com', + port=443, + fingerprints=fingerprints, + cert_data=TEST_TRUST_CERT_PEM, + ), + ) + + form = _MyAuthForm( + { + 'hosting_account_username': 'myuser', + 'hosting_account_password': 'mypass', + }, + hosting_service_cls=TestService, + ) + self.spy_on(form.authorize) + + self.assertTrue(form.is_valid()) + + message = ( + 'The SSL certificate provided by example.com has not been ' + 'signed by a trusted Certificate Authority and may not be safe. ' + 'The certificate needs to be verified in Review Board before ' + 'the server can be accessed. Certificate details: ' + 'hostname="example.com", port=443, issuer="example.com", ' + 'fingerprint=SHA256=79:19:70:AE:A6:1B:EB:BC:35:7C:B8:54:B1:6A:' + 'AD:79:FF:F7:28:69:02:5E:C3:6F:B3:C2:B4:FD:84:66:DF:8F' + ) + + with self.assertRaisesMessage(CertificateVerificationError, message): + form.save(allow_authorize=True) + + self.assertFalse(cert_manager.is_certificate_verified( + hostname='example.com', + port=443, + latest_fingerprints=fingerprints, + )) + self.assertIsNone(cert_manager.get_certificate( + hostname='example.com', + port=443, + )) + self.assertSpyCallCount(form.authorize, 1) + + def test_save_with_cert_error_and_trust_host(self) -> None: + """Testing BaseHostingServiceAuthForm.save with + CertificateVerificationError and trust_host=True + """ + fingerprints = CertificateFingerprints(sha256=TEST_SHA256) + + class _MyAuthForm(BaseHostingServiceAuthForm): + def authorize(self, *args, **kwargs) -> None: + verified = cert_manager.is_certificate_verified( + hostname='example.com', + port=443, + latest_fingerprints=fingerprints, + ) + + if not verified: + raise CertificateVerificationError( + code=CertificateVerificationFailureCode.NOT_TRUSTED, + certificate=Certificate( + hostname='example.com', + port=443, + fingerprints=fingerprints, + cert_data=TEST_TRUST_CERT_PEM, + ), + ) + + self.assertFalse(cert_manager.is_certificate_verified( + hostname='example.com', + port=443, + latest_fingerprints=fingerprints, + )) + + form = _MyAuthForm( + { + 'hosting_account_username': 'myuser', + 'hosting_account_password': 'mypass', + }, + hosting_service_cls=TestService, + ) + self.spy_on(form.authorize) + + self.assertTrue(form.is_valid()) + + form.save(allow_authorize=True, + trust_host=True) + + self.assertTrue(cert_manager.is_certificate_verified( + hostname='example.com', + port=443, + latest_fingerprints=fingerprints, + )) + self.assertAttrsEqual( + cert_manager.get_certificate( + hostname='example.com', + port=443, + ), + { + 'cert_data': TEST_TRUST_CERT_PEM, + 'hostname': 'example.com', + 'port': 443, + }) + self.assertSpyCallCount(form.authorize, 2) + + def test_save_with_legacy_cert_error(self) -> None: + """Testing BaseHostingServiceAuthForm.save with + legacy UnverifiedCertificateError + """ + fingerprints = CertificateFingerprints(sha256=TEST_SHA256) + + class _MyAuthForm(BaseHostingServiceAuthForm): + def authorize(self, *args, **kwargs) -> None: + verified = cert_manager.is_certificate_verified( + hostname='example.com', + port=443, + latest_fingerprints=fingerprints, + ) + + if not verified: + raise UnverifiedCertificateError( + certificate=LegacyCertificate( + pem_data=TEST_TRUST_CERT_PEM.decode('utf-8'), + issuer='issuer', + hostname='example.com', + fingerprint=TEST_SHA256.replace(':', '').lower(), + ) + ) + + form = _MyAuthForm( + { + 'hosting_account_username': 'myuser', + 'hosting_account_password': 'mypass', + }, + hosting_service_cls=TestService, + ) + self.spy_on(form.authorize) + + self.assertTrue(form.is_valid()) + + warnings: list[ExpectedWarning] = [ + { + 'cls': RemovedInReviewBoard90Warning, + 'message': ( + 'UnverifiedCertificateError is deprecated in favor of ' + 'reviewboard.certs.errors.CertificateVerificationError, ' + 'and will be removed in Review Board 9.' + ), + }, + ] + + message = ( + 'The SSL certificate for this repository (hostname ' + '"example.com", fingerprint "791970aea61bebbc357cb854b16aad79f' + 'ff72869025ec36fb3c2b4fd8466df8f") was not verified and might ' + 'not be safe. This certificate needs to be verified before the ' + 'repository can be accessed.' + ) + + with (self.assertWarnings(warnings), + self.assertRaisesMessage(UnverifiedCertificateError, message)): + form.save(allow_authorize=True) + + self.assertFalse(cert_manager.is_certificate_verified( + hostname='example.com', + port=443, + latest_fingerprints=fingerprints, + )) + self.assertIsNone(cert_manager.get_certificate( + hostname='example.com', + port=443, + )) + self.assertSpyCallCount(form.authorize, 1) + + def test_save_with_legacy_cert_error_and_trust_host(self) -> None: + """Testing BaseHostingServiceAuthForm.save with + legacy UnverifiedCertificateError and trust_host=True + """ + fingerprints = CertificateFingerprints(sha256=TEST_SHA256) + + class _MyAuthForm(BaseHostingServiceAuthForm): + def authorize(self, *args, **kwargs) -> None: + verified = cert_manager.is_certificate_verified( + hostname='example.com', + port=443, + latest_fingerprints=fingerprints, + ) + + if not verified: + raise UnverifiedCertificateError( + certificate=LegacyCertificate( + pem_data=TEST_TRUST_CERT_PEM.decode('utf-8'), + issuer='issuer', + hostname='example.com', + fingerprint=TEST_SHA256.replace(':', '').lower(), + ) + ) + + self.assertFalse(cert_manager.is_certificate_verified( + hostname='example.com', + port=443, + latest_fingerprints=fingerprints, + )) + + form = _MyAuthForm( + { + 'hosting_account_username': 'myuser', + 'hosting_account_password': 'mypass', + }, + hosting_service_cls=TestService, + ) + self.spy_on(form.authorize) + + self.assertTrue(form.is_valid()) + + warnings: list[ExpectedWarning] = [ + { + 'cls': RemovedInReviewBoard90Warning, + 'message': ( + 'UnverifiedCertificateError is deprecated in favor of ' + 'reviewboard.certs.errors.CertificateVerificationError, ' + 'and will be removed in Review Board 9.' + ), + }, + { + 'cls': RemovedInReviewBoard10_0Warning, + 'message': ( + 'HostingServiceAccount.accept_certificate() is ' + 'deprecated and will be removed in Review Board 10. ' + 'Use cert_manager.add_certificate() instead.' + ), + }, + ] + + with self.assertWarnings(warnings): + form.save(allow_authorize=True, + trust_host=True) + + self.assertTrue(cert_manager.is_certificate_verified( + hostname='example.com', + port=443, + latest_fingerprints=fingerprints, + )) + self.assertAttrsEqual( + cert_manager.get_certificate( + hostname='example.com', + port=443, + ), + { + 'cert_data': TEST_TRUST_CERT_PEM, + 'hostname': 'example.com', + 'port': 443, + }) + self.assertSpyCallCount(form.authorize, 2) diff --git a/reviewboard/hostingsvcs/tests/test_hosting_service_http_request.py b/reviewboard/hostingsvcs/tests/test_hosting_service_http_request.py new file mode 100644 index 0000000000000000000000000000000000000000..7ae7639feb653ad58cac1a327a72a60b1422f0b7 --- /dev/null +++ b/reviewboard/hostingsvcs/tests/test_hosting_service_http_request.py @@ -0,0 +1,308 @@ +"""Unit tests for reviewboard.hostingsvcs.base.http.HostingServiceHTTPRequest. + +Version Added: + 8.0 +""" + +from __future__ import annotations + +import logging +import os +import shutil +import ssl +from urllib.request import HTTPSHandler, OpenerDirector + +import kgb + +from reviewboard.admin.server import get_data_dir +from reviewboard.certs.manager import cert_manager +from reviewboard.certs.tests.testcases import (CaptureSSLContext, + TEST_TRUST_CERT_PEM, + TEST_TRUST_SAN_CERT_PEM) +from reviewboard.hostingsvcs.base.http import HostingServiceHTTPRequest +from reviewboard.hostingsvcs.models import HostingServiceAccount +from reviewboard.testing import TestCase + + +class _DummyResponse: + headers = {} + + def getcode(self) -> int: + return 200 + + def geturl(self) -> str: + return 'https://example.com/' + + def read(self) -> bytes: + return b'' + + +class BuildSSLContextFromSSLCertTests(kgb.SpyAgency, TestCase): + """Unit tests for HostingServiceHTTPRequest. + + Version Added: + 8.0 + """ + + def setUp(self) -> None: + """Set up state for the test. + + This will clear out the certs directory before running a test. + """ + super().setUp() + + shutil.rmtree(os.path.join(get_data_dir(), 'rb-certs'), + ignore_errors=True) + + def tearDown(self) -> None: + """Tear down state for the test. + + This will clear out the certs directory after running a test. + """ + shutil.rmtree(os.path.join(get_data_dir(), 'rb-certs'), + ignore_errors=True) + + super().tearDown() + + def test_open_with_https(self) -> None: + """Testing HostingServiceHTTPRequest.open with HTTPS""" + self.spy_on(OpenerDirector.open, + owner=OpenerDirector, + op=kgb.SpyOpReturn(_DummyResponse())) + self.spy_on(ssl.create_default_context, + op=kgb.SpyOpReturn(CaptureSSLContext())) + + hosting_account = HostingServiceAccount.objects.create( + service_name='gitlab', + username='myuser', + hosting_url='https://example.com', + ) + + http_request = HostingServiceHTTPRequest( + url='https://example.com', + hosting_service=hosting_account.service, + ) + + # Trigger the request. + http_request.open() + + # Make sure hostname checks are enabled. + handler = http_request._urlopen_handlers[-1] + assert isinstance(handler, HTTPSHandler) + self.assertAttrsEqual( + handler._context, + { + 'cadatas': [], + 'cafiles': [], + 'capaths': [ + os.path.join(get_data_dir(), 'rb-certs', 'file', + 'cabundles'), + ], + 'certfiles': [], + 'check_hostname': True, + 'keyfiles': [], + 'passwords': [], + }) + + def test_open_with_legacy_ssl_cert_data(self) -> None: + """Testing HostingServiceHTTPRequest.open with legacy ssl_cert data""" + self.spy_on(OpenerDirector.open, + owner=OpenerDirector, + op=kgb.SpyOpReturn(_DummyResponse())) + self.spy_on(ssl.create_default_context, + op=kgb.SpyOpReturn(CaptureSSLContext())) + + self.assertIsNone(cert_manager.get_certificate( + hostname='example.com', + port=443, + )) + + hosting_account = HostingServiceAccount.objects.create( + service_name='gitlab', + username='myuser', + hosting_url='https://example.com', + data={ + 'ssl_cert': TEST_TRUST_CERT_PEM.decode('utf-8'), + }, + ) + + http_request = HostingServiceHTTPRequest( + url='https://example.com', + hosting_service=hosting_account.service, + ) + + # Trigger the ssl_cert migration. + http_request.open() + + # Make sure hostname checks are enabled. + handler = http_request._urlopen_handlers[-1] + assert isinstance(handler, HTTPSHandler) + self.assertAttrsEqual( + handler._context, + { + 'cadatas': [], + 'cafiles': [ + os.path.join(get_data_dir(), 'rb-certs', 'file', + 'certs', 'trust', 'example.com__443.crt'), + ], + 'capaths': [ + os.path.join(get_data_dir(), 'rb-certs', 'file', + 'cabundles'), + ], + 'certfiles': [], + 'check_hostname': True, + 'keyfiles': [], + 'passwords': [], + }) + + # Check the legacy data. + hosting_account.refresh_from_db() + self.assertNotIn('ssl_cert', hosting_account.data) + + # Check the migrated certificate. + self.assertAttrsEqual( + cert_manager.get_certificate(hostname='example.com', + port=443), + { + 'cert_data': TEST_TRUST_CERT_PEM, + 'hostname': 'example.com', + 'port': 443, + }) + + def test_open_with_legacy_ssl_cert_data_san(self) -> None: + """Testing HostingServiceHTTPRequest.open with legacy ssl_cert data + and hostname in SAN + """ + self.spy_on(OpenerDirector.open, + owner=OpenerDirector, + op=kgb.SpyOpReturn(_DummyResponse())) + self.spy_on(ssl.create_default_context, + op=kgb.SpyOpReturn(CaptureSSLContext())) + + self.assertIsNone(cert_manager.get_certificate( + hostname='example.com', + port=443, + )) + + hosting_account = HostingServiceAccount.objects.create( + service_name='gitlab', + username='myuser', + hosting_url='https://example.com', + data={ + 'ssl_cert': TEST_TRUST_SAN_CERT_PEM.decode('utf-8'), + }, + ) + + http_request = HostingServiceHTTPRequest( + url='https://example.com', + hosting_service=hosting_account.service, + ) + + # Trigger the ssl_cert migration. + http_request.open() + + # Make sure hostname checks are enabled. + handler = http_request._urlopen_handlers[-1] + assert isinstance(handler, HTTPSHandler) + self.assertAttrsEqual( + handler._context, + { + 'cadatas': [], + 'cafiles': [ + os.path.join(get_data_dir(), 'rb-certs', 'file', + 'certs', 'trust', 'example.com__443.crt'), + ], + 'capaths': [ + os.path.join(get_data_dir(), 'rb-certs', 'file', + 'cabundles'), + ], + 'certfiles': [], + 'check_hostname': True, + 'keyfiles': [], + 'passwords': [], + }) + + # Check the legacy data. + hosting_account.refresh_from_db() + self.assertNotIn('ssl_cert', hosting_account.data) + + # Check the migrated certificate. + self.assertAttrsEqual( + cert_manager.get_certificate(hostname='example.com', + port=443), + { + 'cert_data': TEST_TRUST_SAN_CERT_PEM, + 'hostname': 'example.com', + 'port': 443, + }) + + def test_open_with_legacy_ssl_cert_hostname_mismatch(self) -> None: + """Testing HostingServiceHTTPRequest.open with legacy ssl_cert data + and hostname mismatch + """ + self.spy_on(OpenerDirector.open, + owner=OpenerDirector, + op=kgb.SpyOpReturn(_DummyResponse())) + self.spy_on(ssl.create_default_context, + op=kgb.SpyOpReturn(CaptureSSLContext())) + + self.assertIsNone(cert_manager.get_certificate( + hostname='test.example.com', + port=443, + )) + + cert_data = TEST_TRUST_CERT_PEM.decode('utf-8') + hosting_account = HostingServiceAccount.objects.create( + service_name='gitlab', + username='myuser', + hosting_url='https://test.example.com', + data={ + 'ssl_cert': cert_data, + }, + ) + + http_request = HostingServiceHTTPRequest( + url='https://test.example.com', + hosting_service=hosting_account.service, + ) + + # Trigger the (failing) ssl_cert migration. + with self.assertLogs(level=logging.WARNING) as cm: + http_request.open() + + self.assertEqual( + cm.output, + [ + "WARNING:reviewboard.hostingsvcs.base.http:The approved " + "SSL/TLS certificate stored in hosting service account ID=1 " + "does not match the hostname 'test.example.com' for the " + "server. Falling back to a less-secure form of verification. " + "Please ensure the server has a valid certificate matching " + "its hostname and then upload a new certificate.", + ]) + + # Make sure hostname checks aren't enabled. + handler = http_request._urlopen_handlers[-1] + assert isinstance(handler, HTTPSHandler) + self.assertAttrsEqual( + handler._context, + { + 'cadatas': [cert_data], + 'cafiles': [], + 'capaths': [], + 'certfiles': [], + 'check_hostname': False, + 'keyfiles': [], + 'passwords': [], + }) + + # Check the legacy data is present. + hosting_account.refresh_from_db() + self.assertIn('ssl_cert', hosting_account.data) + self.assertEqual(hosting_account.data['ssl_cert'], cert_data) + + # Check that no certificate was stored. + self.assertIsNone(cert_manager.get_certificate( + hostname='test.example.com', + port=443, + ))