blob: 442dcda26dc3e5a9e5bfa71b5042457d7d3d0b43 (
plain)
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
|
#!/usr/bin/env python
'''
Allow or deny clients based on IP address.
Strawhouse, which is plain text with filtering on IP addresses. It still
uses the NULL mechanism, but we install an authentication hook that checks
the IP address against a whitelist or blacklist and allows or denies it
accordingly.
Author: Chris Laws
'''
import logging
import sys
import zmq
import zmq.auth
from zmq.auth.thread import ThreadAuthenticator
def run():
'''Run strawhouse client'''
allow_test_pass = False
deny_test_pass = False
ctx = zmq.Context.instance()
# Start an authenticator for this context.
auth = ThreadAuthenticator(ctx)
auth.start()
# Part 1 - demonstrate allowing clients based on IP address
auth.allow('127.0.0.1')
server = ctx.socket(zmq.PUSH)
server.zap_domain = b'global' # must come before bind
server.bind('tcp://*:9000')
client_allow = ctx.socket(zmq.PULL)
client_allow.connect('tcp://127.0.0.1:9000')
server.send(b"Hello")
msg = client_allow.recv()
if msg == b"Hello":
allow_test_pass = True
client_allow.close()
# Part 2 - demonstrate denying clients based on IP address
auth.stop()
auth = ThreadAuthenticator(ctx)
auth.start()
auth.deny('127.0.0.1')
client_deny = ctx.socket(zmq.PULL)
client_deny.connect('tcp://127.0.0.1:9000')
if server.poll(50, zmq.POLLOUT):
server.send(b"Hello")
if client_deny.poll(50):
msg = client_deny.recv()
else:
deny_test_pass = True
else:
deny_test_pass = True
client_deny.close()
auth.stop() # stop auth thread
if allow_test_pass and deny_test_pass:
logging.info("Strawhouse test OK")
else:
logging.error("Strawhouse test FAIL")
if __name__ == '__main__':
if zmq.zmq_version_info() < (4,0):
raise RuntimeError("Security is not supported in libzmq version < 4.0. libzmq version {0}".format(zmq.zmq_version()))
if '-v' in sys.argv:
level = logging.DEBUG
else:
level = logging.INFO
logging.basicConfig(level=level, format="[%(levelname)s] %(message)s")
run()
|
