我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用zmq.auth()。
def test_null(self): """threaded auth - NULL""" # A default NULL connection should always succeed, and not # go through our authentication infrastructure at all. self.auth.stop() self.auth = None server = self.socket(zmq.PUSH) client = self.socket(zmq.PULL) self.assertTrue(self.can_connect(server, client)) # By setting a domain we switch on authentication for NULL sockets, # though no policies are configured yet. The client connection # should still be allowed. server = self.socket(zmq.PUSH) server.zap_domain = b'global' client = self.socket(zmq.PULL) self.assertTrue(self.can_connect(server, client))
def start(self): """Start the authentication thread""" # create a socket to communicate with auth thread. self.pipe = self.context.socket(zmq.PAIR) self.pipe.linger = 1 self.pipe.bind(self.pipe_endpoint) authenticator = MultiZapAuthenticator(self.context, encoding=self.encoding, log=self.log) self.thread = AuthenticationThread(self.context, self.pipe_endpoint, encoding=self.encoding, log=self.log, authenticator=authenticator) self.thread.start() # Event.wait:Changed in version 2.7: Previously, the method always returned None. if sys.version_info < (2, 7): self.thread.started.wait(timeout=10) else: if not self.thread.started.wait(timeout=10): raise RuntimeError("Authenticator thread failed to start")
def setUp(self): if zmq.zmq_version_info() < (4,0): raise SkipTest("security is new in libzmq 4.0") try: zmq.curve_keypair() except zmq.ZMQError: raise SkipTest("security requires libzmq to be linked against libsodium") super(BaseAuthTestCase, self).setUp() # enable debug logging while we run tests logging.getLogger('zmq.auth').setLevel(logging.DEBUG) self.auth = self.make_auth() self.auth.start() self.base_dir, self.public_keys_dir, self.secret_keys_dir = self.create_certs()
def tearDown(self): if self.auth: self.auth.stop() self.auth = None self.remove_certs(self.base_dir) super(BaseAuthTestCase, self).tearDown()
def create_certs(self): """Create CURVE certificates for a test""" # Create temporary CURVE keypairs for this test run. We create all keys in a # temp directory and then move them into the appropriate private or public # directory. base_dir = tempfile.mkdtemp() keys_dir = os.path.join(base_dir, 'certificates') public_keys_dir = os.path.join(base_dir, 'public_keys') secret_keys_dir = os.path.join(base_dir, 'private_keys') os.mkdir(keys_dir) os.mkdir(public_keys_dir) os.mkdir(secret_keys_dir) server_public_file, server_secret_file = zmq.auth.create_certificates(keys_dir, "server") client_public_file, client_secret_file = zmq.auth.create_certificates(keys_dir, "client") for key_file in os.listdir(keys_dir): if key_file.endswith(".key"): shutil.move(os.path.join(keys_dir, key_file), os.path.join(public_keys_dir, '.')) for key_file in os.listdir(keys_dir): if key_file.endswith(".key_secret"): shutil.move(os.path.join(keys_dir, key_file), os.path.join(secret_keys_dir, '.')) return (base_dir, public_keys_dir, secret_keys_dir)
def load_certs(self, secret_keys_dir): """Return server and client certificate keys""" server_secret_file = os.path.join(secret_keys_dir, "server.key_secret") client_secret_file = os.path.join(secret_keys_dir, "client.key_secret") server_public, server_secret = zmq.auth.load_certificate(server_secret_file) client_public, client_secret = zmq.auth.load_certificate(client_secret_file) return server_public, server_secret, client_public, client_secret
def test_whitelist(self): """threaded auth - Whitelist""" # Whitelist 127.0.0.1, connection should pass" self.auth.allow('127.0.0.1') server = self.socket(zmq.PUSH) # By setting a domain we switch on authentication for NULL sockets, # though no policies are configured yet. server.zap_domain = b'global' client = self.socket(zmq.PULL) self.assertTrue(self.can_connect(server, client))
def test_plain(self): """threaded auth - PLAIN""" # Try PLAIN authentication - without configuring server, connection should fail server = self.socket(zmq.PUSH) server.plain_server = True client = self.socket(zmq.PULL) client.plain_username = b'admin' client.plain_password = b'Password' self.assertFalse(self.can_connect(server, client)) # Try PLAIN authentication - with server configured, connection should pass server = self.socket(zmq.PUSH) server.plain_server = True client = self.socket(zmq.PULL) client.plain_username = b'admin' client.plain_password = b'Password' self.auth.configure_plain(domain='*', passwords={'admin': 'Password'}) self.assertTrue(self.can_connect(server, client)) # Try PLAIN authentication - with bogus credentials, connection should fail server = self.socket(zmq.PUSH) server.plain_server = True client = self.socket(zmq.PULL) client.plain_username = b'admin' client.plain_password = b'Bogus' self.assertFalse(self.can_connect(server, client)) # Remove authenticator and check that a normal connection works self.auth.stop() self.auth = None server = self.socket(zmq.PUSH) client = self.socket(zmq.PULL) self.assertTrue(self.can_connect(server, client)) client.close() server.close()
def tearDown(self): if self.auth: self.auth.stop() self.auth = None self.io_loop.close(all_fds=True) super(TestIOLoopAuthentication, self).tearDown()
def test_none(self): """ioloop auth - NONE""" # A default NULL connection should always succeed, and not # go through our authentication infrastructure at all. # no auth should be running self.auth.stop() self.auth = None
def test_blacklist(self): """ioloop auth - Blacklist""" # Blacklist 127.0.0.1, connection should fail self.auth.deny('127.0.0.1') self.server.zap_domain = b'global'
def test_whitelist(self): """ioloop auth - Whitelist""" # Whitelist 127.0.0.1, which overrides the blacklist, connection should pass" self.auth.allow('127.0.0.1') self.server.setsockopt(zmq.ZAP_DOMAIN, b'global')
def test_plain_unconfigured_server(self): """ioloop auth - PLAIN, unconfigured server""" self.client.plain_username = b'admin' self.client.plain_password = b'Password' # Try PLAIN authentication - without configuring server, connection should fail self.server.plain_server = True
def test_plain_configured_server(self): """ioloop auth - PLAIN, configured server""" self.client.plain_username = b'admin' self.client.plain_password = b'Password' # Try PLAIN authentication - with server configured, connection should pass self.server.plain_server = True self.auth.configure_plain(domain='*', passwords={'admin': 'Password'})
def test_plain_bogus_credentials(self): """ioloop auth - PLAIN, bogus credentials""" self.client.plain_username = b'admin' self.client.plain_password = b'Bogus' self.server.plain_server = True self.auth.configure_plain(domain='*', passwords={'admin': 'Password'})
def test_curve_allow_any(self): """ioloop auth - CURVE, CURVE_ALLOW_ANY""" certs = self.load_certs(self.secret_keys_dir) server_public, server_secret, client_public, client_secret = certs self.auth.allow('127.0.0.1') self.auth.configure_curve(domain='*', location=zmq.auth.CURVE_ALLOW_ANY) self.server.curve_publickey = server_public self.server.curve_secretkey = server_secret self.server.curve_server = True self.client.curve_publickey = client_public self.client.curve_secretkey = client_secret self.client.curve_serverkey = server_public
def test_curve_configured_server(self): """ioloop auth - CURVE, configured server""" self.auth.allow('127.0.0.1') certs = self.load_certs(self.secret_keys_dir) server_public, server_secret, client_public, client_secret = certs self.auth.configure_curve(domain='*', location=self.public_keys_dir) self.server.curve_publickey = server_public self.server.curve_secretkey = server_secret self.server.curve_server = True self.client.curve_publickey = client_public self.client.curve_secretkey = client_secret self.client.curve_serverkey = server_public
def pyzmq_authenticator(pyzmq_context): auth = ThreadAuthenticator(pyzmq_context) auth.start() auth.allow('127.0.0.1') auth.configure_curve(domain='*', location=zmq.auth.CURVE_ALLOW_ANY) yield auth.stop()
def test_blacklist(self): """threaded auth - Blacklist""" # Blacklist 127.0.0.1, connection should fail self.auth.deny('127.0.0.1') server = self.socket(zmq.PUSH) # By setting a domain we switch on authentication for NULL sockets, # though no policies are configured yet. server.zap_domain = b'global' client = self.socket(zmq.PULL) self.assertFalse(self.can_connect(server, client))
def test_null(self): """ioloop auth - NULL""" # By setting a domain we switch on authentication for NULL sockets, # though no policies are configured yet. The client connection # should still be allowed. self.server.zap_domain = b'global'
