Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@cli.switch(["b", "bacon"], argtype=int, mandatory = True, envname="PLUMBUM_TEST_BACON")
def bacon(self, param):
"""give me some bacon"""
print ("!!b", param)
eggs = cli.SwitchAttr(["e"], str, help = "sets the eggs attribute", envname="PLUMBUM_TEST_EGGS")
cheese = cli.Flag(["--cheese"], help = "cheese, please")
chives = cli.Flag(["--chives"], help = "chives, instead")
verbose = cli.CountOf(["v"], help = "increases the verbosity level")
benedict = cli.CountOf(["--benedict"], help = """a very long help message with lots of
useless information that nobody would ever want to read, but heck, we need to test
text wrapping in help messages as well""")
csv = cli.SwitchAttr(["--csv"], cli.Set("MIN", "MAX", int, csv=True))
num = cli.SwitchAttr(["--num"], cli.Set("MIN", "MAX", int))
def main(self, *args):
old = self.eggs
self.eggs = "lalala"
self.eggs = old
self.tailargs = args
class PositionalApp(cli.Application):
def main(self, one):
print("Got", one)
class Geet(cli.Application):
debug = cli.Flag("--debug")
cleanups = []
def main(self):
del self.cleanups[:]
print("!!a")
@cli.switch(["b", "bacon"], argtype=int, mandatory = True, envname="PLUMBUM_TEST_BACON")
def bacon(self, param):
"""give me some bacon"""
print ("!!b", param)
eggs = cli.SwitchAttr(["e"], str, help = "sets the eggs attribute", envname="PLUMBUM_TEST_EGGS")
cheese = cli.Flag(["--cheese"], help = "cheese, please")
chives = cli.Flag(["--chives"], help = "chives, instead")
verbose = cli.CountOf(["v"], help = "increases the verbosity level")
benedict = cli.CountOf(["--benedict"], help = """a very long help message with lots of
useless information that nobody would ever want to read, but heck, we need to test
text wrapping in help messages as well""")
csv = cli.SwitchAttr(["--csv"], cli.Set("MIN", "MAX", int, csv=True))
num = cli.SwitchAttr(["--num"], cli.Set("MIN", "MAX", int))
def main(self, *args):
old = self.eggs
self.eggs = "lalala"
self.eggs = old
self.tailargs = args
class PositionalApp(cli.Application):
def main(self, one):
print("Got", one)
class Geet(cli.Application):
debug = cli.Flag("--debug")
cleanups = []
def main(self):
#!/usr/bin/env python
"""
The registry server listens to broadcasts on UDP port 18812, answering to
discovery queries by clients and registering keepalives from all running
servers. In order for clients to use discovery, a registry service must
be running somewhere on their local network.
"""
from plumbum import cli
from rpyc.utils.registry import REGISTRY_PORT, DEFAULT_PRUNING_TIMEOUT
from rpyc.utils.registry import UDPRegistryServer, TCPRegistryServer
from rpyc.lib import setup_logger
class RegistryServer(cli.Application):
mode = cli.SwitchAttr(["-m", "--mode"], cli.Set("UDP", "TCP"), default="UDP",
help="Serving mode")
ipv6 = cli.Flag(["-6", "--ipv6"], help="use ipv6 instead of ipv4")
port = cli.SwitchAttr(["-p", "--port"], cli.Range(0, 65535), default=REGISTRY_PORT,
help="The UDP/TCP listener port")
logfile = cli.SwitchAttr(["--logfile"], str, default=None,
help="The log file to use; the default is stderr")
quiet = cli.SwitchAttr(["-q", "--quiet"], help="Quiet mode (only errors are logged)")
pruning_timeout = cli.SwitchAttr(["-t", "--timeout"], int,
default=DEFAULT_PRUNING_TIMEOUT, help="Set a custom pruning timeout (in seconds)")
def main(self):
"""
import sys
import os
import rpyc
from plumbum import cli
from rpyc.utils.server import ThreadedServer, ForkingServer, OneShotServer
from rpyc.utils.classic import DEFAULT_SERVER_PORT, DEFAULT_SERVER_SSL_PORT
from rpyc.utils.registry import REGISTRY_PORT
from rpyc.utils.registry import UDPRegistryClient, TCPRegistryClient
from rpyc.utils.authenticators import SSLAuthenticator
from rpyc.lib import setup_logger
from rpyc.core import SlaveService
class ClassicServer(cli.Application):
mode = cli.SwitchAttr(["-m", "--mode"], cli.Set("threaded", "forking", "stdio", "oneshot"),
default="threaded", help="The serving mode (threaded, forking, or 'stdio' for "
"inetd, etc.)")
port = cli.SwitchAttr(["-p", "--port"], cli.Range(0, 65535), default=None,
help="The TCP listener port (default = %s, default for SSL = %s)" %
(DEFAULT_SERVER_PORT, DEFAULT_SERVER_SSL_PORT), group="Socket Options")
host = cli.SwitchAttr(["--host"], str, default="", help="The host to bind to. "
"The default is localhost", group="Socket Options")
ipv6 = cli.Flag(["--ipv6"], help="Enable IPv6", group="Socket Options")
logfile = cli.SwitchAttr("--logfile", str, default=None, help="Specify the log file to use; "
"the default is stderr", group="Logging")
quiet = cli.Flag(["-q", "--quiet"], help="Quiet mode (only errors will be logged)",
group="Logging")
ssl_keyfile = cli.SwitchAttr("--ssl-keyfile", cli.ExistingFile,
@cli.switch(["-p", "--scan-protocol"], cli.Set("tcp", "udp", case_sensitive=True), help="Protocol used to scan the target.")
def scan_protocol(self, scanProtocol):
'''
Protocol used to perform the nmap scan.
'''
self.scanProtocol = scanProtocol
raise NotImplementedError
class HabitsChange(TasksChange): # pylint: disable=missing-docstring,abstract-method
domain = 'habits'
ids_can_overlap = True
def domain_print(self):
Habits.invoke(config_filename=self.config_filename)
@Habits.subcommand('add') # pylint: disable=missing-docstring
class HabitsAdd(ApplicationWithApi):
DESCRIPTION = _("Add a habit ") # noqa: Q000
priority = cli.SwitchAttr(
['-p', '--priority'],
cli.Set('0.1', '1', '1.5', '2'), default='1',
help=_("Priority (complexity) of a habit")) # noqa: Q000
direction = cli.SwitchAttr(
['-d', '--direction'],
cli.Set('positive', 'negative', 'both'), default='both',
help=_("positive/negative/both")) # noqa: Q000
def main(self, *habit: str):
habit_str = ' '.join(habit)
if not habit_str:
self.log.error(_("Empty habit text!")) # noqa: Q000
return 1
super().main()
self.api.tasks.user.post(
type='habit', text=habit_str,
priority=self.priority, up=(self.direction != 'negative'),
down=(self.direction != 'positive'))
group = "Logging")
ssl_keyfile = cli.SwitchAttr("--ssl-keyfile", cli.ExistingFile,
help = "The keyfile to use for SSL. Required for SSL", group = "SSL",
requires = ["--ssl-certfile"])
ssl_certfile = cli.SwitchAttr("--ssl-certfile", cli.ExistingFile,
help = "The certificate file to use for SSL. Required for SSL", group = "SSL",
requires = ["--ssl-keyfile"])
ssl_cafile = cli.SwitchAttr("--ssl-cafile", cli.ExistingFile,
help = "The certificate authority chain file to use for SSL. Optional; enables client-side "
"authentication", group = "SSL", requires = ["--ssl-keyfile"])
auto_register = cli.Flag("--register", help = "Asks the server to attempt registering with "
"a registry server. By default, the server will not attempt to register",
group = "Registry")
registry_type = cli.SwitchAttr("--registry-type", cli.Set("UDP", "TCP"),
default = "UDP", help="Specify a UDP or TCP registry", group = "Registry")
registry_port = cli.SwitchAttr("--registry-port", cli.Range(0, 65535), default=REGISTRY_PORT,
help = "The registry's UDP/TCP port", group = "Registry")
registry_host = cli.SwitchAttr("--registry-host", str, default = None,
help = "The registry host machine. For UDP, the default is 255.255.255.255; "
"for TCP, a value is required", group = "Registry")
def main(self):
if self.registry_type == "UDP":
if self.registry_host is None:
self.registry_host = "255.255.255.255"
self.registrar = UDPRegistryClient(ip = self.registry_host, port = self.registry_port)
else:
if self.registry_host is None:
raise ValueError("With TCP registry, you must specify --registry-host")
self.registrar = TCPRegistryClient(ip = self.registry_host, port = self.registry_port)
@cli.switch(["-m", "--mode"], cli.Set("windows", "linux", case_sensitive=False), mandatory=True, help="Filter the platform of exit-nodes to attack.")
def server_mode(self, mode):
'''
Server Mode: Search for Windows or Linux machines.
'''
self.mode = mode
internal_edit_todo = cli.Flag(["-i", "--internal-edit-todo"], requires = ["-t"],
help = "Edit the git-rebase todo file. FOR INTERNAL USE ONLY")
rebase_todo_filename = cli.SwitchAttr(["-t", "--internal-todo-filename"], requires = ["-i"],
argtype = cli.ExistingFile, argname = "",
help = "Full path of the git-rebase todo file. FOR INTERNAL USE ONLY")
gsr_cmd = cli.SwitchAttr(["-g", "--gsr-cmd"], argtype = str, argname = "",
default = os.path.dirname(sys.argv[0]) + "/git-search-replace.py",
help = "Path to git-search-replace.py")
commit_msg = cli.SwitchAttr(["-c", "--commit-msg"], argtype = str, argname = "",
excludes = ["-F"],
help = "Provide the git log message for insert mode")
commit_msg_file = cli.SwitchAttr(["-F", "--commit-msg-file"],
argtype = cli.ExistingFile, argname = "",
excludes = ["-c"],
help = "Provide the git log message for insert mode (as filename)")
mode = cli.SwitchAttr(["-m", "--mode"], argtype = cli.Set(MODE_INSERT, MODE_EDIT), argname = "mode",
default = MODE_EDIT,
help = '''Whether to insert a new commit at base, or edit the base commit.
Use insert when renaming something that's already in upstream,
and edit when renaming something that was only added in the current branch.''')
def edit_internal(self, base, *args):
gsr_base = ["%s -f" % (self.gsr_cmd,)]
for arg in args:
gsr_base.append("'%s'" % (arg,))
full_gsr_cmd = " ".join(gsr_base)
filter_branch_params = ["filter-branch"]
if self.force_filter_branch:
filter_branch_params.append("-f")
filter_branch_params.extend(["--tree-filter", full_gsr_cmd, "%s..HEAD" % (base,)])
try: