Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions https_wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -114,13 +114,15 @@ def connect(self):
if self._tunnel_host:
self._tunnel()

context = ssl.create_default_context()
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.check_hostname = True
context.verify_mode = ssl.CERT_REQUIRED
context.options |= ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3
Comment thread
cisco-annikiti marked this conversation as resolved.
context.verify_flags |= ssl.VERIFY_X509_STRICT
context.load_verify_locations(cafile=self.ca_certs)

if self.cert_file:
context.load_cert_chain(self.cert_file, keyfile=self.key_file)

context.options = self.cert_reqs | ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3
server_hostname = self._tunnel_host or self.host
self.sock = context.wrap_socket(self.sock, server_hostname=server_hostname)

Expand Down
34 changes: 32 additions & 2 deletions test_duo_openvpn.py
Original file line number Diff line number Diff line change
Expand Up @@ -695,7 +695,7 @@ def test_connect_direct_uses_host_for_sni(self):
}

with unittest.mock.patch('socket.create_connection', return_value=mock_sock), \
unittest.mock.patch('ssl.create_default_context', return_value=mock_context):
unittest.mock.patch('ssl.SSLContext', return_value=mock_context):
conn.connect()

mock_context.wrap_socket.assert_called_once_with(
Expand All @@ -715,7 +715,7 @@ def test_connect_proxy_uses_tunnel_host_for_sni(self):
}

with unittest.mock.patch('socket.create_connection', return_value=mock_sock), \
unittest.mock.patch('ssl.create_default_context', return_value=mock_context):
unittest.mock.patch('ssl.SSLContext', return_value=mock_context):
# Bypass the actual HTTP CONNECT tunnel
with unittest.mock.patch.object(conn, '_tunnel'):
conn.connect()
Expand All @@ -724,6 +724,36 @@ def test_connect_proxy_uses_tunnel_host_for_sni(self):
mock_sock, server_hostname='api-host.duosecurity.com',
)

def test_connect_does_not_load_system_cas(self):
"""New approach: SSLContext(PROTOCOL_TLS_CLIENT) does NOT load OS trust store."""
conn = self._make_connection('api-host.duosecurity.com')
mock_sock = MagicMock()
mock_context = MagicMock()
mock_context.wrap_socket.return_value = MagicMock(getpeercert=MagicMock(return_value={
'subjectAltName': [('DNS', '*.duosecurity.com')],
}))

with unittest.mock.patch('socket.create_connection', return_value=mock_sock), \
unittest.mock.patch('ssl.SSLContext', return_value=mock_context), \
unittest.mock.patch('ssl.create_default_context') as mock_default:
conn.connect()

# create_default_context must NOT be called — it loads OS CAs
mock_default.assert_not_called()
# Only our bundled CA file is loaded
mock_context.load_verify_locations.assert_called_once_with(cafile='/path/to/ca.pem')

def test_ssl_context_enforces_tls_verification(self):
"""SSLContext(PROTOCOL_TLS_CLIENT) must have check_hostname=True, CERT_REQUIRED, and no SSLv3."""
import ssl
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
self.assertTrue(ctx.check_hostname,
'check_hostname must be True')
self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED,
'verify_mode must be CERT_REQUIRED')
self.assertTrue(ctx.options & ssl.OP_NO_SSLv3,
'SSLv3 must be disabled')


if __name__ == '__main__':
unittest.main()