1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
|
# Copyright DataStax, Inc.
#
# 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.
from packaging.version import Version
import logging
import time
from cassandra.cluster import NoHostAvailable
from cassandra.auth import PlainTextAuthProvider, SASLClient, SaslAuthProvider
from tests.integration import use_singledc, get_cluster, remove_cluster, PROTOCOL_VERSION, \
CASSANDRA_IP, CASSANDRA_VERSION, USE_CASS_EXTERNAL, start_cluster_wait_for_up, TestCluster
from tests.integration.util import assert_quiescent_pool_state
import unittest
log = logging.getLogger(__name__)
#This can be tested for remote hosts, but the cluster has to be configured accordingly
#@local
def setup_module():
if CASSANDRA_IP.startswith("127.0.0.") and not USE_CASS_EXTERNAL:
use_singledc(start=False)
ccm_cluster = get_cluster()
ccm_cluster.stop()
config_options = {'authenticator': 'PasswordAuthenticator',
'authorizer': 'CassandraAuthorizer'}
ccm_cluster.set_configuration_options(config_options)
log.debug("Starting ccm test cluster with %s", config_options)
start_cluster_wait_for_up(ccm_cluster)
# PYTHON-1328
#
# Give the cluster enough time to startup (and perform necessary initialization)
# before executing the test.
if CASSANDRA_VERSION > Version('4.0-a'):
time.sleep(10)
def teardown_module():
remove_cluster() # this test messes with config
class AuthenticationTests(unittest.TestCase):
"""
Tests to cover basic authentication functionality
"""
def get_authentication_provider(self, username, password):
"""
Return correct authentication provider based on protocol version.
There is a difference in the semantics of authentication provider argument with protocol versions 1 and 2
For protocol version 2 and higher it should be a PlainTextAuthProvider object.
For protocol version 1 it should be a function taking hostname as an argument and returning a dictionary
containing username and password.
:param username: authentication username
:param password: authentication password
:return: authentication object suitable for Cluster.connect()
"""
if PROTOCOL_VERSION < 2:
return lambda hostname: dict(username=username, password=password)
else:
return PlainTextAuthProvider(username=username, password=password)
def cluster_as(self, usr, pwd):
# test we can connect at least once with creds
# to ensure the role manager is setup
for _ in range(5):
try:
cluster = TestCluster(
idle_heartbeat_interval=0,
auth_provider=self.get_authentication_provider(username='cassandra', password='cassandra'))
cluster.connect(wait_for_all_pools=True)
return TestCluster(
idle_heartbeat_interval=0,
auth_provider=self.get_authentication_provider(username=usr, password=pwd))
except Exception as e:
time.sleep(5)
raise Exception('Unable to connect with creds: {}/{}'.format(usr, pwd))
def test_auth_connect(self):
user = 'u'
passwd = 'password'
root_session = self.cluster_as('cassandra', 'cassandra').connect()
root_session.execute('CREATE USER %s WITH PASSWORD %s', (user, passwd))
try:
cluster = self.cluster_as(user, passwd)
session = cluster.connect(wait_for_all_pools=True)
try:
self.assertTrue(session.execute('SELECT release_version FROM system.local'))
assert_quiescent_pool_state(self, cluster, wait=1)
for pool in session.get_pools():
connection, _ = pool.borrow_connection(timeout=0)
self.assertEqual(connection.authenticator.server_authenticator_class, 'org.apache.cassandra.auth.PasswordAuthenticator')
pool.return_connection(connection)
finally:
cluster.shutdown()
finally:
root_session.execute('DROP USER %s', user)
assert_quiescent_pool_state(self, root_session.cluster, wait=1)
root_session.cluster.shutdown()
def test_connect_wrong_pwd(self):
cluster = self.cluster_as('cassandra', 'wrong_pass')
try:
self.assertRaisesRegex(NoHostAvailable,
'.*AuthenticationFailed.',
cluster.connect)
assert_quiescent_pool_state(self, cluster)
finally:
cluster.shutdown()
def test_connect_wrong_username(self):
cluster = self.cluster_as('wrong_user', 'cassandra')
try:
self.assertRaisesRegex(NoHostAvailable,
'.*AuthenticationFailed.*',
cluster.connect)
assert_quiescent_pool_state(self, cluster)
finally:
cluster.shutdown()
def test_connect_empty_pwd(self):
cluster = self.cluster_as('Cassandra', '')
try:
self.assertRaisesRegex(NoHostAvailable,
'.*AuthenticationFailed.*',
cluster.connect)
assert_quiescent_pool_state(self, cluster)
finally:
cluster.shutdown()
def test_connect_no_auth_provider(self):
cluster = TestCluster()
try:
self.assertRaisesRegex(NoHostAvailable,
'.*AuthenticationFailed.*',
cluster.connect)
assert_quiescent_pool_state(self, cluster)
finally:
cluster.shutdown()
class SaslAuthenticatorTests(AuthenticationTests):
"""
Test SaslAuthProvider as PlainText
"""
def setUp(self):
if PROTOCOL_VERSION < 2:
raise unittest.SkipTest('Sasl authentication not available for protocol v1')
if SASLClient is None:
raise unittest.SkipTest('pure-sasl is not installed')
def get_authentication_provider(self, username, password):
sasl_kwargs = {'service': 'cassandra',
'mechanism': 'PLAIN',
'qops': ['auth'],
'username': username,
'password': password}
return SaslAuthProvider(**sasl_kwargs)
# these could equally be unit tests
def test_host_passthrough(self):
sasl_kwargs = {'service': 'cassandra',
'mechanism': 'PLAIN'}
provider = SaslAuthProvider(**sasl_kwargs)
host = 'thehostname'
authenticator = provider.new_authenticator(host)
self.assertEqual(authenticator.sasl.host, host)
def test_host_rejected(self):
sasl_kwargs = {'host': 'something'}
self.assertRaises(ValueError, SaslAuthProvider, **sasl_kwargs)
|
