Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

IPv6 support #26

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open

IPv6 support #26

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
7704d17
IPv6 support
Dec 31, 2015
140e392
Make v6 optional
Jan 5, 2016
832fd9a
Change handling of IPV4ONLY vs IPV6'
Jan 5, 2016
a7f4fb3
enable v6 for worker
Jan 5, 2016
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
90 changes: 80 additions & 10 deletions laikad.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,12 +129,13 @@ class AsyncBroker(Process):
returned back to the client.
'''

def __init__(self, broker_backend_address, broker_frontend_address):
def __init__(self, broker_backend_address, broker_frontend_address, use_ipv6):
'''Main constructor'''
super(AsyncBroker, self).__init__()
self.broker_backend_address = broker_backend_address
self.broker_frontend_address = broker_frontend_address
self.keep_running = True
self.use_ipv6 = use_ipv6

def shutdown(self):
'''Shutdown method to be called by the signal handler'''
Expand All @@ -156,12 +157,26 @@ def run(self):

# Connection for workers
backend = context.socket(zmq.ROUTER)
if self.use_ipv6:
if hasattr(zmq, 'IPV6'):
backend.setsockopt(zmq.IPV6, 1)
elif hasattr(zmq, 'IPV4ONLY'):
backend.setsockopt(zmq.IPV4ONLY, 0)
else:
logging.error("This version of ZMQ does not support IPv6")
backend.bind(self.broker_backend_address)
backend_poller = zmq.Poller()
backend_poller.register(backend, zmq.POLLIN)

# Connection for clients
frontend = context.socket(zmq.PULL)
if self.use_ipv6:
if hasattr(zmq, 'IPV6'):
frontend.setsockopt(zmq.IPV6, 1)
elif hasattr(zmq, 'IPV4ONLY'):
frontend.setsockopt(zmq.IPV4ONLY, 0)
else:
logging.error("This version of ZMQ does not support IPv6")
frontend.bind(self.broker_frontend_address)
frontend_poller = zmq.Poller()
frontend_poller.register(frontend, zmq.POLLIN)
Expand Down Expand Up @@ -247,13 +262,14 @@ class SyncBroker(Process):
'''

def __init__(self, broker_backend_address, broker_frontend_address,
shutdown_grace_timeout=SHUTDOWN_GRACE_TIMEOUT_DEFAULT):
use_ipv6, shutdown_grace_timeout=SHUTDOWN_GRACE_TIMEOUT_DEFAULT):
'''Main constructor'''
super(SyncBroker, self).__init__()
self.broker_backend_address = broker_backend_address
self.broker_frontend_address = broker_frontend_address
self.shutdown_grace_timeout = shutdown_grace_timeout
self.keep_running = True
self.use_ipv6 = use_ipv6

def shutdown(self):
'''Shutdown method to be called by the signal handler'''
Expand All @@ -273,12 +289,26 @@ def run(self):

# Connection for workers
backend = context.socket(zmq.ROUTER)
if self.use_ipv6:
if hasattr(zmq, 'IPV6'):
backend.setsockopt(zmq.IPV6, 1)
elif hasattr(zmq, 'IPV4ONLY'):
backend.setsockopt(zmq.IPV4ONLY, 0)
else:
logging.error("This version of ZMQ does not support IPv6")
backend.bind(self.broker_backend_address)
backend_poller = zmq.Poller()
backend_poller.register(backend, zmq.POLLIN)

# Connection for clients
frontend = context.socket(zmq.ROUTER)
if self.use_ipv6:
if hasattr(zmq, 'IPV6'):
frontend.setsockopt(zmq.IPV6, 1)
elif hasattr(zmq, 'IPV4ONLY'):
frontend.setsockopt(zmq.IPV4ONLY, 0)
else:
logging.error("This version of ZMQ does not support IPv6")
frontend.bind(self.broker_frontend_address)
frontend_poller = zmq.Poller()
frontend_poller.register(frontend, zmq.POLLIN)
Expand Down Expand Up @@ -439,6 +469,7 @@ class Worker(Process):
'''

def __init__(self, config_location, broker_address, max_scan_items, ttl,
use_ipv6,
logresult=False,
poll_timeout=300,
shutdown_grace_timeout=SHUTDOWN_GRACE_TIMEOUT_DEFAULT):
Expand All @@ -456,6 +487,7 @@ def __init__(self, config_location, broker_address, max_scan_items, ttl,
self.broker_poller = zmq.Poller()
self.poll_timeout = poll_timeout * 1000 # Poller uses milliseconds
self.logresult = logresult
self.use_ipv6 = use_ipv6

def perform_scan(self, poll_timeout):
'''
Expand Down Expand Up @@ -618,6 +650,13 @@ def run(self):
context = zmq.Context(1)
self.broker = context.socket(zmq.DEALER)
self.broker.setsockopt(zmq.IDENTITY, self.identity)
if self.use_ipv6:
if hasattr(zmq, 'IPV6'):
self.broker.setsockopt(zmq.IPV6, 1)
elif hasattr(zmq, 'IPV4ONLY'):
self.broker.setsockopt(zmq.IPV4ONLY, 0)
else:
logging.error("This version of ZMQ does not support IPv6")
self.broker.connect(self.broker_address)
self.broker_poller.register(self.broker, zmq.POLLIN)

Expand Down Expand Up @@ -791,6 +830,10 @@ def main():
dest="gracetimeout",
help="when shutting down, the timeout to allow workers to"
" finish ongoing scans before being killed")
parser.add_option("-6", "--ipv6",
action="store_true", default=False,
dest="use_ipv6",
help="Enables listening on IPv6.")
(options, _) = parser.parse_args()

# Set the configuration file path
Expand Down Expand Up @@ -901,26 +944,46 @@ def shutdown(signum, frame):
broker_proc = None
if not options.no_broker:
if async:
broker_proc = AsyncBroker(broker_backend_address, broker_frontend_address)
broker_proc = AsyncBroker(
broker_backend_address,
broker_frontend_address,
options.use_ipv6)
else:
broker_proc = SyncBroker(broker_backend_address, broker_frontend_address, gracetimeout)
broker_proc = SyncBroker(
broker_backend_address,
broker_frontend_address,
options.use_ipv6,
gracetimeout)
broker_proc.start()

# Start the workers
workers = []
for _ in range(num_procs):
worker_proc = Worker(laikaboss_config_path, worker_connect_address, ttl,
time_ttl, logresult, int(get_option('workerpolltimeout')), gracetimeout)
worker_proc = Worker(
laikaboss_config_path,
worker_connect_address,
ttl,
time_ttl,
options.use_ipv6,
logresult,
int(get_option('workerpolltimeout')),
gracetimeout)
worker_proc.start()
workers.append(worker_proc)

while KEEP_RUNNING:
# Ensure we have a broker
if not options.no_broker and not broker_proc.is_alive():
if async:
broker_proc = AsyncBroker(broker_backend_address, broker_frontend_address)
broker_proc = AsyncBroker(
broker_backend_address,
broker_frontend_address,
options.use_ipv6)
else:
broker_proc = SyncBroker(broker_backend_address, broker_frontend_address,
broker_proc = SyncBroker(
broker_backend_address,
broker_frontend_address,
options.use_ipv6,
gracetimeout)
broker_proc.start()

Expand All @@ -932,8 +995,15 @@ def shutdown(signum, frame):

for worker_proc in dead_workers:
workers.remove(worker_proc)
new_proc = Worker(laikaboss_config_path, worker_connect_address, ttl, time_ttl,
logresult, int(get_option('workerpolltimeout')), gracetimeout)
new_proc = Worker(
laikaboss_config_path,
worker_connect_address,
ttl,
time_ttl,
options.use_ipv6,
logresult,
int(get_option('workerpolltimeout')),
gracetimeout)
new_proc.start()
workers.append(new_proc)
worker_proc.join()
Expand Down