Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@options([
make_option('-c', '--change', type="string", dest="change_namespace", help="Change the current namespace of the dispatcher to change the handling of commands"),
make_option('-l', '--list', action="store_true", dest="list_namespaces", help="Show the current namespaces available to the dispatcher"),
make_option('-p', '--print', action="store_true", dest="print_namespace", default=True, help="Show the current namespace of the dispatcher")
])
def do_namespace(self, arg, opts):
"""Command to show and change command namespace"""
poortego_namespace(self, arg, opts)
@options([
make_option('-c', '--csv', action="store_true", help="Export data to CSV file"),
make_option('-g', '--graphviz', action="store_true", help="Export data to graphviz PNG"),
make_option('-m', '--maltego', action="store_true", help="Export data to Maltego file"),
make_option('-s', '--stix', action="store_true", help="Export data to STIX file"),
])
def do_export(self, arg):
"""Command to export parts/all of graph to another format"""
# TODO
make_option('-l', '--list', action="store_true", dest="list_details", help="Show the current user details")
])
def do_poortego_user(self, arg, opts):
"""Command to show and change user details"""
poortego_user(self.my_interface, arg, opts)
@options([make_option('-b', '--backend',type="string", help="Connect to backend, values: r for router, s for smppcm and r,s for both"),
make_option('--router_host', type="string", help="Router host, default 127.0.0.1"),
make_option('--router_port', type="int", help="Router port, default 8988"),
make_option('--smppcm_host', type="string", help="SMPPCM host, default 127.0.0.1"),
make_option('--smppcm_port', type="int", help="SMPPCM host, default 8989"),])
def do_connect(self, arg, opts):
'Start a connection to router (router) or smpp client manager (smppcm)'
if opts.backend is None:
self.print_stdout('Argument -b required.')
return
if opts.backend not in ['r', 's', 'r,s']:
self.print_stdout('Invalid argument (%s) for command connect, must be "r", "s" or "r,s".' % opts.backend)
return
# @todo: host and port must be configurable
router_host = '127.0.0.1'
router_port = 8988
smppcm_host = '127.0.0.1'
smppcm_port = 8989
make_option('-c', '--commas',
action="store_true",
help="Intersperse commas")])
def do_yo(self, arg, opts):
chant = ['yo'] + ['ho'] * opts.ho
separator = ', ' if opts.commas else ' '
chant = separator.join(chant)
print('{0} and a bottle of {1}'
.format(chant, arg))
@options([make_option('-c', '--channel', action="store",
help="Channel for unsubscribe"),
make_option('-a', '--all', action="store_true", dest="all",
default=False, help="Unsubscribe from all channels"),
make_option('-p', '--presence', action="store_true",
dest="presence",
default=False, help="Unsubscribe from presence events ?")
])
def do_unsubscribe(self, command, opts):
opts.channel = self.default_channel \
if opts.channel is None else opts.channel
if (opts.all is True):
opts.channel = []
chs = pubnub.get_channel_array()
for ch in chs:
if '-pnpres' not in ch:
opts.channel.append(ch)
elif opts.presence is True:
make_option('-f', '--from', action="store_true", dest="from_nodes", help="List only nodes with link from current node"),
make_option('-t', '--to', action="store_true", dest="to_nodes", help="List only nodes with link to current node"),
])
def do_poortego_ls(self, arg, opts):
"""Command to show node adjacency"""
poortego_ls(self.my_interface, arg, opts)
@options([
make_option('-a', '--all', action="store_true", dest="all_nodes", help="List all nodes tied to root"),
make_option('-b', '--bi', action="store_true", help="List nodes connected to/from current node"),
make_option('-f', '--from', action="store_true", dest="from_nodes", help="List only nodes with link from current node"),
make_option('-t', '--to', action="store_true", dest="to_nodes", help="List only nodes with link to current node"),
])
def do_ls(self, arg, opts):
"""Command to show node adjacency"""
self.stdout.write("Output format: '[id] name [type]'\n")
if opts.all_nodes:
self.my_graph.show_nodes_from(self.my_graph.poortego_root_node._id)
elif opts.bi:
self.stdout.write("\nNodes From Current:\n")
self.my_graph.show_nodes_from(self.my_session.current_node_id)
self.stdout.write("\nNodes To Current:\n")
self.my_graph.show_nodes_to(self.my_session.current_node_id)
self.stdout.write("\n")
elif opts.from_nodes:
self.stdout.write("\nNodes From Current:\n")
self.my_graph.show_nodes_from(self.my_session.current_node_id)
make_option("-i", "--id", action = "store", type = "string", help = ""),
])
def do_list_links(self, arg, opts):
output = """
ID: {id}
Source node: {src_node}
Target node: {tgt_node}
Source port: {src_port}
Target port: {tgt_port}
"""
def print_link(link):
print("[{i}]".format(i = link.id))
print(output.format(id = link.id,
src_node = link.node_source.name,