TLS

membership/config/settings.py

@@ -87,6 +87,8 @@ DEX_CLIENT_SECRET = os.environ.get("DEX_CLIENT_SECRET", "")
 DEX_CONNECTOR_ID = os.environ.get("DEX_CONNECTOR_ID", "ldap")
 
 LDAP_URL = os.environ.get("LDAP_URL", "")
+# Explicit opt-in for trusted internal networks; never a TLS-error fallback.
+LDAP_ALLOW_INSECURE = os.environ.get("LDAP_ALLOW_INSECURE", "0") == "1"
 LDAP_BIND_DN = os.environ.get("LDAP_BIND_DN", "")
 LDAP_BIND_PASSWORD = os.environ.get("LDAP_BIND_PASSWORD", "")
 LDAP_CA_CERT_FILE = os.environ.get("LDAP_CA_CERT_FILE") or None

membership/members/directory.py

@@ -1,4 +1,4 @@
-"""TLS-only LDAP; paid groups are never granted during registration or login."""
+"""TLS by default; paid groups are never granted during registration or login."""
 
 import re
 import ssl
@@ -22,7 +22,8 @@ def connection():
     conn = Connection(server, user=settings.LDAP_BIND_DN, password=settings.LDAP_BIND_PASSWORD, raise_exceptions=True, receive_timeout=8, auto_referrals=False)
     try:
         conn.open()
-        if url.scheme == "ldap" and not conn.start_tls():
+        # Plaintext requires explicit operator consent; LDAPS is never downgraded.
+        if url.scheme == "ldap" and not settings.LDAP_ALLOW_INSECURE and not conn.start_tls():
             raise LDAPException("StartTLS failed; refusing to bind.")
         if not conn.bind():
             raise LDAPException("LDAP bind failed.")

membership/members/tests/test_membership.py

@@ -471,7 +471,7 @@ class DirectoryTests(SimpleTestCase):
         with self.assertRaises(ValueError):
             directory.identity_for_login("*)(uid=*)")
 
-    @override_settings(LDAP_URL="ldap://ldap.example.org", LDAP_BIND_DN="cn=service", LDAP_BIND_PASSWORD="test-only", LDAP_ACCOUNT_BASE_DN=BASE)
+    @override_settings(LDAP_URL="ldap://ldap.example.org", LDAP_ALLOW_INSECURE=False, LDAP_BIND_DN="cn=service", LDAP_BIND_PASSWORD="test-only", LDAP_ACCOUNT_BASE_DN=BASE)
     @patch("members.directory.Connection")
     def test_starttls_failure_never_binds_credentials(self, conn_factory):
         conn_factory.return_value.start_tls.return_value = False
@@ -480,3 +480,29 @@ class DirectoryTests(SimpleTestCase):
                 pass
         conn_factory.return_value.bind.assert_not_called()
         conn_factory.return_value.unbind.assert_called_once()
+
+    @override_settings(LDAP_BIND_DN="cn=service", LDAP_BIND_PASSWORD="test-only", LDAP_ACCOUNT_BASE_DN=BASE, LDAP_CA_CERT_FILE=None)
+    @patch("members.directory.Connection")
+    @patch("members.directory.Server")
+    def test_ldap_transport_modes(self, server_factory, conn_factory):
+        import ssl
+
+        for scheme, insecure, starttls, port in (
+            ("ldap", False, True, 389),
+            ("ldap", True, False, 389),
+            ("ldaps", False, False, 636),
+            ("ldaps", True, False, 636),
+        ):
+            with self.subTest(scheme=scheme, insecure=insecure):
+                conn_factory.reset_mock()
+                with override_settings(LDAP_URL=f"{scheme}://ldap.example.org", LDAP_ALLOW_INSECURE=insecure):
+                    with directory.connection() as conn:
+                        self.assertIs(conn, conn_factory.return_value)
+                self.assertEqual(conn_factory.return_value.start_tls.call_count, int(starttls))
+                conn_factory.return_value.open.assert_called_once()
+                conn_factory.return_value.bind.assert_called_once()
+                conn_factory.return_value.unbind.assert_called_once()
+                options = server_factory.call_args.kwargs
+                self.assertEqual(options["use_ssl"], scheme == "ldaps")
+                self.assertEqual(options["port"], port)
+                self.assertEqual(options["tls"].validate, ssl.CERT_REQUIRED)