Python html.escape函数代码示例 - 纯净天空 (2023)

 def rebuild(self, req): if self.isUsingUserPasswd(req): if not self.authUser(req): return Redirect("../../../authfailed") b = self.build_status bc = self.builder_control builder_name = b.getBuilder().getName() log.msg("web rebuild of build %s:%s" % (builder_name, b.getNumber())) name = req.args.get("username", ["<unknown>"])[0] comments = req.args.get("comments", ["<no reason specified>"])[0] reason = ("The web-page 'rebuild' button was pressed by " "'%s': %s\n" % (html.escape(name), html.escape(comments))) extraProperties = getAndCheckProperties(req) if not bc or not b.isFinished() or extraProperties is None: log.msg("could not rebuild: bc=%s, isFinished=%s" % (bc, b.isFinished())) # TODO: indicate an error else: bc.resubmitBuild(b, reason, extraProperties) # we're at # http://localhost:8080/builders/NAME/builds/5/rebuild?[args] # Where should we send them? # # Ideally it would be to the per-build page that they just started, # but we don't know the build number for it yet (besides, it might # have to wait for a current build to finish). The next-most # preferred place is somewhere that the user can see tangible # evidence of their build starting (or to see the reason that it # didn't start). This should be the Builder page. r = Redirect("../..") # the Builder's page d = defer.Deferred() reactor.callLater(1, d.callback, r) return DeferredResource(d)

 def performAction(self, req): authz = self.getAuthz(req) res = yield authz.actionAllowed(self.action, req, self.build_status) if not res: defer.returnValue(path_to_authzfail(req)) return b = self.build_status log.msg("web stopBuild of build %s:%s" % (b.getBuilder().getName(), b.getNumber())) name = authz.getUsernameFull(req) comments = _get_comments_from_request(req) # html-quote both the username and comments, just to be safe reason = ("The web-page 'stop build' button was pressed by " "'%s': %s\n" % (html.escape(name), html.escape(comments))) c = interfaces.IControl(self.getBuildmaster(req)) bldrc = c.getBuilder(self.build_status.getBuilder().getName()) if bldrc: bldc = bldrc.getBuild(self.build_status.getNumber()) if bldc: bldc.stopBuild(reason) defer.returnValue(path_to_builder(req, self.build_status.getBuilder()))

 def force(self, req): """ Custom properties can be passed from the web form. To do this, subclass this class, overriding the force() method. You can then determine the properties (usually from form values, by inspecting req.args), then pass them to this superclass force method. """ name = req.args.get("username", ["<unknown>"])[0] reason = req.args.get("comments", ["<no reason specified>"])[0] branch = req.args.get("branch", [""])[0] revision = req.args.get("revision", [""])[0] r = "The web-page 'force build' button was pressed by '%s': %s\n" \ % (html.escape(name), html.escape(reason)) log.msg("web forcebuild of builder '%s', branch='%s', revision='%s'" " by user '%s'" % (self.builder_status.getName(), branch, revision, name)) if not self.builder_control: # TODO: tell the web user that their request was denied log.msg("but builder control is disabled") return Redirect("..") if self.isUsingUserPasswd(req): if not self.authUser(req): return Redirect("../../authfail") # keep weird stuff out of the branch and revision strings. TODO: # centralize this somewhere. if not re.match(r'^[\w\.\-\/]*$', branch): log.msg("bad branch '%s'" % branch) return Redirect("..") if not re.match(r'^[\w\.\-\/]*$', revision): log.msg("bad revision '%s'" % revision) return Redirect("..") if not branch: branch = None if not revision: revision = None # TODO: if we can authenticate that a particular User pushed the # button, use their name instead of None, so they'll be informed of # the results. # TODO2: we can authenticate that a particular User pushed the button # now, so someone can write this support. but it requires a # buildbot.changes.changes.Change instance which is tedious at this # stage to compute s = SourceStamp(branch=branch, revision=revision) req = BuildRequest(r, s, builderName=self.builder_status.getName()) try: self.builder_control.requestBuildSoon(req) except interfaces.NoSlaveError: # TODO: tell the web user that their request could not be # honored pass # send the user back to the builder page return Redirect(".")

 def stop(self, req, auth_ok=False): # check if this is allowed if not auth_ok: if not self.getAuthz(req).actionAllowed('stopBuild', req, self.build_status): return Redirect(path_to_authfail(req)) b = self.build_status log.msg("web stopBuild of build %s:%s" % \ (b.getBuilder().getName(), b.getNumber())) name = req.args.get("username", ["<unknown>"])[0] comments = req.args.get("comments", ["<no reason specified>"])[0] # html-quote both the username and comments, just to be safe reason = ("The web-page 'stop build' button was pressed by " "'%s': %s\n" % (html.escape(name), html.escape(comments))) c = interfaces.IControl(self.getBuildmaster(req)) bldrc = c.getBuilder(self.build_status.getBuilder().getName()) if bldrc: bldc = bldrc.getBuild(self.build_status.getNumber()) if bldc: bldc.stopBuild(reason) # we're at http://localhost:8080/svn-hello/builds/5/stop?[args] and # we want to go to: http://localhost:8080/svn-hello r = Redirect(path_to_builder(req, self.build_status.getBuilder())) d = defer.Deferred() reactor.callLater(1, d.callback, r) return DeferredResource(d)

 def performAction(self, req): d = self.getAuthz(req).actionAllowed(self.action, req, self.build_status) wfd = defer.waitForDeferred(d) yield wfd res = wfd.getResult() if not res: yield path_to_authfail(req) return b = self.build_status log.msg("web stopBuild of build %s:%s" % \ (b.getBuilder().getName(), b.getNumber())) name = req.args.get("username", ["<unknown>"])[0] comments = req.args.get("comments", ["<no reason specified>"])[0] # html-quote both the username and comments, just to be safe reason = ("The web-page 'stop build' button was pressed by " "'%s': %s\n" % (html.escape(name), html.escape(comments))) c = interfaces.IControl(self.getBuildmaster(req)) bldrc = c.getBuilder(self.build_status.getBuilder().getName()) if bldrc: bldc = bldrc.getBuild(self.build_status.getNumber()) if bldc: bldc.stopBuild(reason) yield path_to_builder(req, self.build_status.getBuilder()) return

 def to_html(self, href_base="", timestamps="short-local"): d = self.e['d'] time_short, time_extended = web_format_time(d['time'], timestamps) msg = html.escape(log.format_message(d)) if 'failure' in d: lines = str(d['failure']).split("\n") html_lines = [html.escape(line) for line in lines] f_html = "\n".join(html_lines) msg += " FAILURE:<pre>%s</pre>" % f_html level = d.get('level', log.OPERATIONAL) level_s = "" if level >= log.UNUSUAL: level_s = self.LEVELMAP.get(level, "") + " " details = " ".join(["Event #%d" % d['num'], "TubID=%s" % self.e['from'], "Incarnation=%s" % self.incarnation, time_extended]) label = '<span title="%s">%s</span>' % (details, time_short) data = '%s [<span id="E%s"><a href="%s#E%s">%d</a></span>]: %s%s' \ % (label, self.anchor_index, href_base, self.anchor_index, d['num'], level_s, msg) if self.is_trigger: data += " [INCIDENT-TRIGGER]" return data

 def GenStepBox(stepstatus): """Generates a box for one step.""" class_ = build_get_class(stepstatus) style = '' if class_ and class_ in styles: style = styles[class_] stepname = stepstatus.getName() text = stepstatus.getText() or [] text = text[:] base_url = '%sbuilders/%s/builds/%d/steps' % ( waterfall_url, urllib.quote(stepstatus.getBuild().getBuilder().getName(), safe=''), stepstatus.getBuild().getNumber()) for steplog in stepstatus.getLogs(): name = steplog.getName() log.msg('name = %s' % name) url = '%s/%s/logs/%s' % ( base_url, urllib.quote(stepname, safe=''), urllib.quote(name)) text.append('<a href="%s">%s</a>' % (url, html.escape(name))) for name, target in stepstatus.getURLs().iteritems(): text.append('<a href="%s">%s</a>' % (target, html.escape(name))) fmt = '<tr><td style="%s">%s</td></tr>' return fmt % (style, '<br/>'.join(text))

开发者ID:leiferikb，项目名称:bitpop，代码行数:25，代码来源:build_utils.py

 def stop(self, req, auth_ok=False): # check if this is allowed if not auth_ok: return StopBuildActionResource(self.build_status) b = self.build_status log.msg("web stopBuild of build %s:%s" % (b.getBuilder().getName(), b.getNumber())) name = self.getAuthz(req).getUsernameFull(req) comments = _get_comments_from_request(req) # html-quote both the username and comments, just to be safe reason = ("The web-page 'stop build' button was pressed by " "'%s': %s\n" % (html.escape(name), html.escape(comments))) c = interfaces.IControl(self.getBuildmaster(req)) bldrc = c.getBuilder(self.build_status.getBuilder().getName()) if bldrc: bldc = bldrc.getBuild(self.build_status.getNumber()) if bldc: bldc.stopBuild(reason) # we're at http://localhost:8080/svn-hello/builds/5/stop?[args] and # we want to go to: http://localhost:8080/svn-hello r = Redirect(path_to_builder(req, self.build_status.getBuilder())) d = defer.Deferred() reactor.callLater(1, d.callback, r) return DeferredResource(d)

 def body(self, req): s = self.step_status b = s.getBuild() builder_name = b.getBuilder().getName() build_num = b.getNumber() data = "" data += '<h1>BuildStep <a href="%s">%s</a>:' % (path_to_builder(req, b.getBuilder()), builder_name) data += '<a href="%s">#%d</a>' % (path_to_build(req, b), build_num) data += ":%s</h1>\n" % s.getName() if s.isFinished(): data += "<h2>Finished</h2>\n" "<p>%s</p>\n" % html.escape("%s" % s.getText()) else: data += "<h2>Not Finished</h2>\n" "<p>ETA %s seconds</p>\n" % s.getETA() exp = s.getExpectations() if exp: data += "<h2>Expectations</h2>\n" "<ul>\n" for e in exp: data += "<li>%s: current=%s, target=%s</li>\n" % (html.escape(e[0]), e[1], e[2]) data += "</ul>\n" (start, end) = s.getTimes() if not start: start_text = end_text = elapsed = "Not Started" else: start_text = ctime(start) if end: end_text = ctime(end) elapsed = util.formatInterval(end - start) else: end_text = "Not Finished" elapsed = util.formatInterval(util.now() - start) data += "<h2>Timing</h2>\n" data += "<table>\n" data += "<tr><td>Start</td><td>%s</td></tr>\n" % start_text data += "<tr><td>End</td><td>%s</td></tr>\n" % end_text data += "<tr><td>Elapsed</td><td>%s</td></tr>\n" % elapsed data += "</table>\n" logs = s.getLogs() if logs: data += "<h2>Logs</h2>\n" "<ul>\n" for logfile in logs: if logfile.hasContents(): # FIXME: If the step name has a / in it, this is broken # either way. If we quote it but say '/'s are safe, # it chops up the step name. If we quote it and '/'s # are not safe, it escapes the / that separates the # step name from the log number. logname = logfile.getName() logurl = req.childLink("logs/%s" % urllib.quote(logname)) data += '<li><a href="%s">%s</a></li>\n' % (logurl, html.escape(logname)) else: data += "<li>%s</li>\n" % html.escape(logname) data += "</ul>\n" return data

 def error_handler(self, m): try: if m['isError'] == 1: self.log.err(escape(repr(m))) reactor.stop() else: self.log.log(escape(repr(m))) except: print m 

 def call_cmd_handlers(self, t, s, body, subject, stanza): if subject: return #found or create item groupchat = s.split('/')[0] nick = s[len(groupchat)+1:] try: item = self.g[groupchat][nick] except: item = new_item(self) item.jid = s item.realjid = s if item.room and (item.nick == self.muc.get_nick(item.room.jid)): self.log.log(u'own message from %s ignored' % (escape(item.jid), ), 2) return #check for bad words if item.room and (t == 'groupchat'): self.check_text(item, body) #launch rewrite engines commands = [(t, item, body, stanza)] itercount = 0 rl = config.REWRITE_POWER changed = True while changed and (len(commands) < rl) and (itercount < rl): itercount += 1 self.log.log(escape('launch rewrite engines.. step %s, commands=%s' % (itercount, commands)), 1) changed = False for engine in self.rewriteengines: r_commands = [] for command in commands: r_command = self.call(engine, command) if r_command: r_commands = r_commands + r_command changed = True else: r_commands = r_commands + [command] commands = r_commands self.log.log(escape('rewrite engines done, itercount=%s, commands=%s' % (itercount, commands)), 1) if itercount == rl: item.lmsg(t, 'rewrite_cycle', rl) elif len(commands) >= rl: item.lmsg(t, 'rewrite_too_many_commands', rl) else: #execute commands, generated by rewrite engines commands = [command for command in commands if self.check_for_ddos(command[1].jid)] # check if not DDOS for command in commands: [t, s, b, stanza] = command params = b.split() if len(params): cmd = b.split()[0] params = b[len(cmd)+1:] for i in self.cmdhandlers: if cmd.lower() == i[1]: if s.allowed(i[2]): if (s.room and s.room.bot) or not i[3]: self.log.log(u'Calling command handler <b>%s</b> for command <font color=red>%s</font> from <font color=blue>%s (%s)</font><br/>stanza: <font color=grey>%s</font>' % (escape(repr(i[0])), escape(b), escape(s.jid), escape(repr(s)), escape(stanza.toXml())), 4) try: i[0](t, s, params) except: m = ''.join(traceback.format_exception(sys.exc_type, sys.exc_value, sys.exc_traceback)) self.log.err(escape(m)) item.lmsg(t, 'ERROR', config.ERRLOGFILE) else: s.lmsg(t, 'muc_only') else: s.lmsg(t, 'not_allowed')

 def performAction(self, req): # check if this is allowed d = self.getAuthz(req).actionAllowed(self.action, req, self.builder_status) wfd = defer.waitForDeferred(d) yield wfd res = wfd.getResult() if not res: log.msg("..but not authorized") yield path_to_authfail(req) return master = self.getBuildmaster(req) # keep weird stuff out of the branch revision, and property strings. branch_validate = master.config.validation['branch'] revision_validate = master.config.validation['revision'] if not branch_validate.match(self.req_args['branch']): log.msg("bad branch '%s'" % self.req_args['branch']) yield path_to_builder(req, self.builder_status) return if not revision_validate.match(self.req_args['revision']): log.msg("bad revision '%s'" % self.req_args['revision']) yield path_to_builder(req, self.builder_status) return properties = getAndCheckProperties(req) if properties is None: yield path_to_builder(req, self.builder_status) return if not self.req_args['branch']: self.req_args['branch'] = None if not self.req_args['revision']: self.req_args['revision'] = None d = master.db.sourcestamps.addSourceStamp( branch=self.req_args['branch'], revision=self.req_args['revision'], project=self.req_args['project'], repository=self.req_args['repository']) wfd = defer.waitForDeferred(d) yield wfd ssid = wfd.getResult() r = ("The web-page 'force build' button was pressed by '%s': %s\n" % (html.escape(self.req_args['name']), html.escape(self.req_args['reason']))) d = master.addBuildset(builderNames=[self.builder_status.getName()], ssid=ssid, reason=r, properties=properties.asDict()) wfd = defer.waitForDeferred(d) yield wfd tup = wfd.getResult() # check that (bsid, brids) were properly stored if not isinstance(tup, (int, dict)): log.err("(ignored) while trying to force build") # send the user back to the builder page yield path_to_builder(req, self.builder_status)

 def call(self, f, *args, **kwargs): try: return f(*args, **kwargs) except: m = '; '.join(traceback.format_exception(sys.exc_type, sys.exc_value, sys.exc_traceback)) m = m.decode('utf8', 'replace') m = u'<font color=red><b>ERROR:</b></font> %s\n<br/>\n(f, *args, *kwargs) was <font color=grey>(%s)</font>' \ % (escape(m), escape(repr((f, args, kwargs)))) self.log.err(m) return 0

 def check_text(self, source, text): #if source.room and (source.room.bot.nick == self.muc.get_nick(source.room.jid)): # return #ignore own messages/presences if not text: return #ignore empty messages/statuses self.log.log(u'checking %s (from %s)' % (escape(text), escape(source.jid)), 1) bad_word = self.censor.respond(text) if bad_word: self.log.log(u'found bad word <%s>, let\'s call bad_handlers!' % (escape(bad_word), ), 3) self.call_bad_handlers(source, text, bad_word) else: self.log.log(u'bad words not found (result: %s)' % (bad_word, ), 1)

 def force(self, req, auth_ok=False): name = req.args.get("username", ["<unknown>"])[0] reason = req.args.get("comments", ["<no reason specified>"])[0] branch = req.args.get("branch", [""])[0] revision = req.args.get("revision", [""])[0] repository = req.args.get("repository", [""])[0] project = req.args.get("project", [""])[0] r = "The web-page 'force build' button was pressed by '%s': %s\n" \ % (html.escape(name), html.escape(reason)) log.msg("web forcebuild of builder '%s', branch='%s', revision='%s'," " repository='%s', project='%s' by user '%s'" % ( self.builder_status.getName(), branch, revision, repository, project, name)) # check if this is allowed if not auth_ok: if not self.getAuthz(req).actionAllowed('forceBuild', req, self.builder_status): log.msg("..but not authorized") return Redirect(path_to_authfail(req)) # keep weird stuff out of the branch revision, and property strings. # TODO: centralize this somewhere. if not re.match(r'^[\w.+/~-]*$', branch): log.msg("bad branch '%s'" % branch) return Redirect(path_to_builder(req, self.builder_status)) if not re.match(r'^[ \w\.\-\/]*$', revision): log.msg("bad revision '%s'" % revision) return Redirect(path_to_builder(req, self.builder_status)) properties = getAndCheckProperties(req) if properties is None: return Redirect(path_to_builder(req, self.builder_status)) if not branch: branch = None if not revision: revision = None # TODO: if we can authenticate that a particular User pushed the # button, use their name instead of None, so they'll be informed of # the results. # TODO2: we can authenticate that a particular User pushed the button # now, so someone can write this support. but it requires a # buildbot.changes.changes.Change instance which is tedious at this # stage to compute s = SourceStamp(branch=branch, revision=revision, project=project, repository=repository) try: c = interfaces.IControl(self.getBuildmaster(req)) bc = c.getBuilder(self.builder_status.getName()) bc.submitBuildRequest(s, r, properties) except interfaces.NoSlaveError: # TODO: tell the web user that their request could not be # honored pass # send the user back to the builder page return Redirect(path_to_builder(req, self.builder_status))