[Zodb-checkins] CVS: ZODB3/ZEO/tests - testAuth.py:1.3
auth_plaintext.py:1.2 zeoserver.py:1.18 testZEO.py:1.72
testMonitor.py:1.6 ConnectionTests.py:1.26
Jeremy Hylton
jeremy at zope.com
Fri May 30 16:21:27 EDT 2003
Update of /cvs-repository/ZODB3/ZEO/tests
In directory cvs.zope.org:/tmp/cvs-serv25334/ZEO/tests
Modified Files:
zeoserver.py testZEO.py testMonitor.py ConnectionTests.py
Added Files:
testAuth.py auth_plaintext.py
Log Message:
Merge ZODB3-auth-branch and bump a few version numbers.
After the merge, I made several Python 2.1 compatibility changes for
the auth code.
=== ZODB3/ZEO/tests/testAuth.py 1.2 => 1.3 ===
--- /dev/null Fri May 30 15:21:27 2003
+++ ZODB3/ZEO/tests/testAuth.py Fri May 30 15:20:56 2003
@@ -0,0 +1,111 @@
+##############################################################################
+#
+# Copyright (c) 2003 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE
+#
+##############################################################################
+"""Test suite for AuthZEO."""
+
+import os
+import tempfile
+import time
+import unittest
+
+from ThreadedAsync import LoopCallback
+from ZEO.ClientStorage import ClientStorage
+from ZEO.StorageServer import StorageServer
+from ZEO.tests.ConnectionTests import CommonSetupTearDown
+
+from ZODB.FileStorage import FileStorage
+from ZODB.tests.StorageTestBase import removefs
+
+class AuthTest(CommonSetupTearDown):
+ __super_getServerConfig = CommonSetupTearDown.getServerConfig
+ __super_setUp = CommonSetupTearDown.setUp
+ __super_tearDown = CommonSetupTearDown.tearDown
+
+ realm = None
+
+ def setUp(self):
+ self.pwfile = tempfile.mktemp()
+ if self.realm:
+ self.pwdb = self.dbclass(self.pwfile, self.realm)
+ else:
+ self.pwdb = self.dbclass(self.pwfile)
+ self.pwdb.add_user("foo", "bar")
+ self.pwdb.save()
+ self.__super_setUp()
+
+ def tearDown(self):
+ self.__super_tearDown()
+ os.remove(self.pwfile)
+
+ def getConfig(self, path, create, read_only):
+ return "<mappingstorage 1/>"
+
+ def getServerConfig(self, addr, ro_svr):
+ zconf = self.__super_getServerConfig(addr, ro_svr)
+ zconf.authentication_protocol = self.protocol
+ zconf.authentication_database = self.pwfile
+ zconf.authentication_realm = self.realm
+ return zconf
+
+ def wait(self):
+ for i in range(25):
+ if self._storage.test_connection:
+ return
+ time.sleep(0.1)
+ self.fail("Timed out waiting for client to authenticate")
+
+ def testOK(self):
+ # Sleep for 0.2 seconds to give the server some time to start up
+ # seems to be needed before and after creating the storage
+ self._storage = self.openClientStorage(wait=0, username="foo",
+ password="bar", realm=self.realm)
+ self.wait()
+
+ self.assert_(self._storage._connection)
+ self._storage._connection.poll()
+ self.assert_(self._storage.is_connected())
+
+ def testNOK(self):
+ self._storage = self.openClientStorage(wait=0, username="foo",
+ password="noogie",
+ realm=self.realm)
+ self.wait()
+ # If the test established a connection, then it failed.
+ self.failIf(self._storage._connection)
+
+class PlainTextAuth(AuthTest):
+ import ZEO.tests.auth_plaintext
+ protocol = "plaintext"
+ database = "authdb.sha"
+ dbclass = ZEO.tests.auth_plaintext.Database
+ realm = "Plaintext Realm"
+
+class DigestAuth(AuthTest):
+ import ZEO.auth.auth_digest
+ protocol = "digest"
+ database = "authdb.digest"
+ dbclass = ZEO.auth.auth_digest.DigestDatabase
+ realm = "Digest Realm"
+
+test_classes = [PlainTextAuth, DigestAuth]
+
+def test_suite():
+ suite = unittest.TestSuite()
+ for klass in test_classes:
+ sub = unittest.makeSuite(klass)
+ suite.addTest(sub)
+ return suite
+
+if __name__ == "__main__":
+ unittest.main(defaultTest='test_suite')
+
=== ZODB3/ZEO/tests/auth_plaintext.py 1.1 => 1.2 ===
--- /dev/null Fri May 30 15:21:27 2003
+++ ZODB3/ZEO/tests/auth_plaintext.py Fri May 30 15:20:56 2003
@@ -0,0 +1,44 @@
+##############################################################################
+#
+# Copyright (c) 2003 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE
+#
+##############################################################################
+"""Implements plaintext password authentication. The password is stored in
+an SHA hash in the Database. The client sends over the plaintext
+password, and the SHA hashing is done on the server side.
+
+This mechanism offers *no network security at all*; the only security
+is provided by not storing plaintext passwords on disk. (See the
+auth_srp module for a secure mechanism)"""
+
+import sha
+
+from ZEO.StorageServer import ZEOStorage
+from ZEO.auth import register_module
+from ZEO.auth.base import Client, Database
+
+class StorageClass(ZEOStorage):
+ def auth(self, username, password):
+ try:
+ dbpw = self.database.get_password(username)
+ except LookupError:
+ return 0
+
+ password = sha.new(password).hexdigest()
+ return self.finish_auth(dbpw == password)
+
+class PlaintextClient(Client):
+ extensions = ["auth"]
+
+ def start(self, username, realm, password):
+ return self.stub.auth(username, password)
+
+register_module("plaintext", StorageClass, PlaintextClient, Database)
=== ZODB3/ZEO/tests/zeoserver.py 1.17 => 1.18 ===
--- ZODB3/ZEO/tests/zeoserver.py:1.17 Fri May 30 11:08:40 2003
+++ ZODB3/ZEO/tests/zeoserver.py Fri May 30 15:20:56 2003
@@ -26,9 +26,8 @@
import ZConfig.Context
import zLOG
-import ZEO.StorageServer
+from ZEO.StorageServer import StorageServer
from ZEO.runzeo import ZEOOptions
-from ZODB.config import storageFromURL
def cleanup(storage):
@@ -152,6 +151,10 @@
zo.realize(["-C", configfile])
zeo_port = int(zo.address[1])
+ # XXX a hack
+ if zo.auth_protocol == "plaintext":
+ import ZEO.tests.auth_plaintext
+
# Open the config file and let ZConfig parse the data there. Then remove
# the config file, otherwise we'll leave turds.
# The rest of the args are hostname, portnum
@@ -163,14 +166,17 @@
mon_addr = None
if zo.monitor_address:
mon_addr = zo.monitor_address.address
- server = ZEO.StorageServer.StorageServer(
+ server = StorageServer(
zo.address,
{"1": storage},
read_only=zo.read_only,
invalidation_queue_size=zo.invalidation_queue_size,
transaction_timeout=zo.transaction_timeout,
- monitor_address=mon_addr)
-
+ monitor_address=mon_addr,
+ auth_protocol=zo.auth_protocol,
+ auth_filename=zo.auth_database,
+ auth_realm=zo.auth_realm)
+
try:
log(label, 'creating the test server, keep: %s', keep)
t = ZEOTestServer(test_addr, server, keep)
=== ZODB3/ZEO/tests/testZEO.py 1.71 => 1.72 ===
--- ZODB3/ZEO/tests/testZEO.py:1.71 Fri May 30 14:31:16 2003
+++ ZODB3/ZEO/tests/testZEO.py Fri May 30 15:20:56 2003
@@ -39,8 +39,7 @@
MTStorage, ReadOnlyStorage
from ZEO.ClientStorage import ClientStorage
-from ZEO.tests import forker, Cache
-from ZEO.tests import CommitLockTests, ThreadTests
+from ZEO.tests import forker, Cache, CommitLockTests, ThreadTests
class DummyDB:
def invalidate(self, *args):
=== ZODB3/ZEO/tests/testMonitor.py 1.5 => 1.6 ===
=== ZODB3/ZEO/tests/ConnectionTests.py 1.25 => 1.26 ===
--- ZODB3/ZEO/tests/ConnectionTests.py:1.25 Thu May 29 15:18:18 2003
+++ ZODB3/ZEO/tests/ConnectionTests.py Fri May 30 15:20:56 2003
@@ -41,6 +41,8 @@
class TestClientStorage(ClientStorage):
+ test_connection = 0
+
def verify_cache(self, stub):
self.end_verify = threading.Event()
self.verify_result = ClientStorage.verify_cache(self, stub)
@@ -49,6 +51,12 @@
ClientStorage.endVerify(self)
self.end_verify.set()
+ def testConnection(self, conn):
+ try:
+ return ClientStorage.testConnection(self, conn)
+ finally:
+ self.test_connection = 1
+
class DummyDB:
def invalidate(self, *args, **kwargs):
pass
@@ -113,38 +121,45 @@
# port+1 is also used, so only draw even port numbers
return 'localhost', random.randrange(25000, 30000, 2)
- def getConfig(self):
+ def getConfig(self, path, create, read_only):
raise NotImplementedError
def openClientStorage(self, cache='', cache_size=200000, wait=1,
- read_only=0, read_only_fallback=0):
- base = TestClientStorage(self.addr,
- client=cache,
- cache_size=cache_size,
- wait=wait,
- min_disconnect_poll=0.1,
- read_only=read_only,
- read_only_fallback=read_only_fallback)
- storage = base
+ read_only=0, read_only_fallback=0,
+ username=None, password=None, realm=None):
+ storage = TestClientStorage(self.addr,
+ client=cache,
+ cache_size=cache_size,
+ wait=wait,
+ min_disconnect_poll=0.1,
+ read_only=read_only,
+ read_only_fallback=read_only_fallback,
+ username=username,
+ password=password,
+ realm=realm)
storage.registerDB(DummyDB(), None)
return storage
- def startServer(self, create=1, index=0, read_only=0, ro_svr=0):
- addr = self.addr[index]
- zLOG.LOG("testZEO", zLOG.INFO,
- "startServer(create=%d, index=%d, read_only=%d) @ %s" %
- (create, index, read_only, addr))
- path = "%s.%d" % (self.file, index)
- sconf = self.getConfig(path, create, read_only)
+ def getServerConfig(self, addr, ro_svr):
zconf = forker.ZEOConfig(addr)
if ro_svr:
zconf.read_only = 1
if self.monitor:
- zconf.monitor_address = ("", 42000)
+ zconf.monitor_address = ("", 42000)
if self.invq:
zconf.invalidation_queue_size = self.invq
if self.timeout:
zconf.transaction_timeout = self.timeout
+ return zconf
+
+ def startServer(self, create=1, index=0, read_only=0, ro_svr=0):
+ addr = self.addr[index]
+ zLOG.LOG("testZEO", zLOG.INFO,
+ "startServer(create=%d, index=%d, read_only=%d) @ %s" %
+ (create, index, read_only, addr))
+ path = "%s.%d" % (self.file, index)
+ sconf = self.getConfig(path, create, read_only)
+ zconf = self.getServerConfig(addr, ro_svr)
zeoport, adminaddr, pid, path = forker.start_zeo_server(sconf, zconf,
addr[1],
self.keep)
@@ -448,7 +463,6 @@
self._storage = self.openClientStorage()
self._dostore()
-
# Test case for multiple storages participating in a single
# transaction. This is not really a connection test, but it needs
More information about the Zodb-checkins
mailing list