Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def makeCallable(self, func, interval=0, name=None, expose=False):
function_name = BaseServer.makeCallable(
self,
func,
interval=interval,
name=name,
expose=expose)
if expose:
self.exposed_functions.append(function_name)
er = ExposedResource(self, function_name)
function_name_parts = function_name.split("/")
if len(function_name_parts) > 1:
if function_name_parts[0] in self.exposed_function_resources:
r = self.exposed_function_resources[function_name_parts[0]]
else:
r = Resource()
self.exposed_function_resources[function_name_parts[0]] = r
self.function_resource.putChild(function_name_parts[0], r)
r.putChild(function_name_parts[1], er)
else:
self.function_resource.putChild(function_name_parts[0], er)
LOGGER.info("Function %s is now available via the HTTP interface." % function_name)
def __init__(self, device, backend=None):
self.device = device
if backend is None:
backend = self.device.backend
resource.Resource.__init__(self)
service.ServiceServer.__init__(self, 'Dimming', self.device.version,
backend)
self.control = DimmingControl(self)
self.putChild(self.scpd_url, service.scpdXML(self, self.control))
self.putChild(self.control_url, self.control)
def __init__(self, avatar=None):
## Twisted Resource is a not a new style class, so emulating a super-call
resource.Resource.__init__(self)
self.avatar = avatar
self.use_security_proxy = get_config().getboolean('auth', 'security_proxy_rest')
self.use_keystone_tokens = get_config().getboolean('auth', 'use_keystone', False)
def setup(self, interface, port):
logger.info("Starting Internal API HTTP server on %s:%d", interface, port)
root = Resource()
matrix = Resource()
root.putChild(b'_matrix', matrix)
identity = Resource()
matrix.putChild(b'identity', identity)
internal = Resource()
identity.putChild(b'internal', internal)
authenticated_bind = AuthenticatedBindThreePidServlet(self.sydent)
internal.putChild(b'bind', authenticated_bind)
authenticated_unbind = AuthenticatedUnbindThreePidServlet(self.sydent)
internal.putChild(b'unbind', authenticated_unbind)
factory = Site(root)
factory.displayTracebacks = False
self.sydent.reactor.listenTCP(port, factory, interface=interface)
def _stripe_resource(state):
v1 = Resource()
v1.putChild(b"subscriptions", _Subscriptions(state))
root = Resource()
root.putChild(b"v1", v1)
return root
def main(argv = sys.argv):
log.startLogging(sys.stdout, setStdout = False)
config = Config('config.cfg')
site = RestSite(resource.Resource())
site.register_uri(r'/co/(?P\d{1,8})/?(?:.(?P[json|html]))?$', simple_view)
site.register_uri(r'/co/$', mostrar_concentradores)
site.register_uri(r'/co/(?P\d{1,8})/?$', simple_view)
site.register_uri(r'/co/(?P\d{1,8})/uc/(?P\d{1,8})/?$', simple_view)
reactor.listenTCP(config.webserver.port, site) #@UndefinedVariable
reactor.run() #@UndefinedVariable
cache_control = 'public, s-maxage={m}, max-age={m}'.format(m=max_age)
self.request.setHeader('Cache-Control', cache_control)
def download_failed(self, report):
"""
Log download failures.
:param report: The download report.
:type report: nectar.report.DownloadReport
"""
super(DownloadListener, self).download_failed(report)
code = report.error_report.get('response_code', '')
logger.info(_('Download failed [{code}]: {url}').format(code=code, url=report.url))
class Streamer(Resource):
"""
Define the web resource that streams content from the upstream repository
to the client.
"""
# Ensure self.getChild isn't called as this has no child resources
isLeaf = True
def __init__(self, config):
"""
Initialize a streamer instance.
:param config: The configuration for this streamer instance.
:type config: ConfigParser.SafeConfigParser
"""
Resource.__init__(self)
def make_plugin(app_config, http, irc, salons):
deploy_config = config.parse_config(app_config, {
"organizations": config.TupleOf(config.String),
"default_hours_start": parse_time,
"default_hours_end": parse_time,
"default_tz": pytz.timezone,
"blackout_hours_start": parse_time,
"blackout_hours_end": parse_time,
})
monitor = DeployMonitor(deploy_config, irc, salons)
# set up http api
deploy_root = resource.Resource()
http.root.putChild('deploy', deploy_root)
deploy_root.putChild('status', DeployStatusListener(http.hmac_secret, monitor))
deploy_root.putChild('begin', DeployBeganListener(http, monitor))
deploy_root.putChild('end', DeployEndedListener(http, monitor))
deploy_root.putChild('abort', DeployAbortedListener(http, monitor))
deploy_root.putChild('error', DeployErrorListener(http, monitor))
deploy_root.putChild('progress', DeployProgressListener(http, monitor))
deploy_root.putChild('hold', DeployHoldListener(http, monitor))
deploy_root.putChild('unhold', DeployUnHoldListener(http, monitor))
deploy_root.putChild('hold_all', DeployHoldAllListener(http, monitor))
deploy_root.putChild('unhold_all', DeployUnholdAllListener(http, monitor))
deploy_root.putChild('send_announcement', DeploySendAnnouncementListener(http, monitor))
deploy_root.putChild('get_salon_names', DeployGetSalonNamesListener(http, monitor))
# register our irc commands
irc.register_command(monitor.salonify)
if (target['hcount'] > count) or (target['hcount'] >= count and nb_rule > self.rules_hit):
pprint.pprint(target)
self.try_append(target, True)
self.final_rules.append(target)
return
def generate_stats(self):
stats = ""
self.wrapper.execute("select count(distinct exception_id) as uniq_exception from exceptions")
uniq_ex = self.wrapper.getResults()[0]['uniq_exception']
self.wrapper.execute("select count(distinct peer_ip) as uniq_peer from connections")
uniq_peer = self.wrapper.getResults()[0]['uniq_peer']
return "<ul><li>There is currently %s unique exceptions.</li></ul><ul><li>There is currently %s different peers that triggered rules.</li></ul>" % (uniq_ex, uniq_peer)
class NaxsiUI(Resource):
def __init__(self):
Resource.__init__(self)
#twisted will handle static content for me
self.putChild('bootstrap', File('./bootstrap'))
self.putChild('js', File('./js'))
#make the correspondance between the path and the object to call
self.page_handler = {'/' : Index, '/graphs': GraphView, '/get_rules': GenWhitelist, '/map': WootMap}
def getChild(self, name, request):
handler = self.page_handler.get(request.path)
if handler is not None:
return handler()
else:
return NoResource()
def __init__(self, responderResource):
Resource.__init__(self)
wellKnown = Resource()
wellKnown.putChild(b'acme-challenge', responderResource)
self.putChild(b'.well-known', wellKnown)
self.putChild(b'check', static.Data(b'OK', 'text/plain'))