我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用SimpleHTTPServer.SimpleHTTPRequestHandler()。
def do_POST(self): deploy_steps_match = re.match( r'/v1/nodes/([^/]*)/vendor_passthru/deploy_steps', self.path) callback_match = re.search( r'/v1/nodes/([^/]*)/vendor_passthru', self.path) if deploy_steps_match is not None: tmp = tempfile.NamedTemporaryFile() json.dump({'url': None}, tmp) tmp.flush() self.path = tmp.name elif callback_match is not None: callback_file_path = os.path.join( CONF.ramdisk_func_test_workdir, callback_match.group(1), 'callback') open(callback_file_path, 'a').close() LOG.info("Got callback: %s", self.path) self.path = os.path.join(self.ctx.htdocs, 'stubfile') return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
def do_GET(self): if self.path == '/': # Create the response response = { 'billing_id': config_custom.billing_id, 'customer_id' : config_custom.customer_id, 'location_id' : config_custom.location_id, } # Optionally include other information if config_custom.is_note_displayed: response['note_text'] = config_custom.note_text # Write the response self.protocol_version = 'HTTP/1.1' self.send_response(200, 'OK') self.send_header('Content-type', 'application/json') self.end_headers() self.wfile.write(bytes(json.dumps(response))) # self.path = '/' return # return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
def basicWebServer(self): import SimpleHTTPServer import SocketServer PORT = 8000 Handler = SimpleHTTPServer.SimpleHTTPRequestHandler while (True): try: print "Trying to open on port", PORT httpd = SocketServer.TCPServer(("", PORT), Handler) except Exception as e: print "port", PORT, "did not work, checking next port" PORT += 1 else: break print "serving at port", PORT webbrowser.open("http://localhost:"+str(PORT)+"/forceD3/force.html") httpd.serve_forever() ### FUNCTIONS TO BE DEFINED IN SUBCLASSES ###
def __init__(self,request, client_address, server): self.server = server self.protocol_version = 'HTTP/1.1' self.challengeMessage = None self.target = None self.client = None self.machineAccount = None self.machineHashes = None self.domainIp = None global ATTACKED_HOSTS if self.server.target in ATTACKED_HOSTS and self.server.one_shot: logging.info( "HTTPD: Received connection from %s, skipping %s, already attacked" % ( client_address[0], self.server.target)) return if self.server.target is not None: logging.info( "HTTPD: Received connection from %s, attacking target %s" % (client_address[0], self.server.target)) else: logging.info( "HTTPD: Received connection from %s, attacking target %s" % (client_address[0], client_address[0])) SimpleHTTPServer.SimpleHTTPRequestHandler.__init__(self,request, client_address, server)
def __init__(self,request, client_address, server): self.server = server self.protocol_version = 'HTTP/1.1' self.challengeMessage = None self.target = None self.client = None self.machineAccount = None self.machineHashes = None self.domainIp = None self.authUser = None if self.server.config.mode != 'REDIRECT': if self.server.config.target is not None: self.target = self.server.config.target.get_target(client_address[0],self.server.config.randomtargets) logging.info("HTTPD: Received connection from %s, attacking target %s" % (client_address[0] ,self.target[1])) else: self.target = self.client_address[0] logging.info("HTTPD: Received connection from %s, attacking target %s" % (client_address[0] ,client_address[0])) SimpleHTTPServer.SimpleHTTPRequestHandler.__init__(self,request, client_address, server)
def main(): Handler = SimpleHTTPServer.SimpleHTTPRequestHandler Handler.extensions_map['.pac'] = 'application/x-ns-proxy-autoconfig' try: port = int(sys.argv[1]) except: port = DEFAULT_PORT print "Serving at port %s/tcp ..." % port httpd = SocketServer.TCPServer(("", port), Handler) httpd.serve_forever() ############## # Enry Point # ##############
def server_server(): PORT = 8000 Handler = SimpleHTTPServer.SimpleHTTPRequestHandler try: httpd = SocketServer.TCPServer(("", PORT), Handler) print "serving at port", PORT httpd.serve_forever() except Exception: pass print 'already serving' # makes start block from template using
def setUpClass(cls): """ Set up an HTTP server to serve the XML files. Set the correct port in the IGD.xml URLBase element. """ # Have to chdir here because the py2 SimpleHTTPServer doesn't allow us # to change its working directory like the py3 one does. os.chdir(path.join(path.dirname(path.realpath(__file__)), 'xml')) cls.httpd = sockserver.TCPServer(('127.0.0.1', 0), httpserver.SimpleHTTPRequestHandler) cls.httpd_thread = threading.Thread(target=cls.httpd.serve_forever) cls.httpd_thread.daemon = True cls.httpd_thread.start() cls.httpd_port = cls.httpd.server_address[1] with open('upnp/IGD.xml', 'w') as out_f: with open('upnp/IGD.xml.templ') as in_f: out_f.write(in_f.read().format(port=cls.httpd_port))
def main(): logging.info('pyhttps {}'.format(version)) create_ssl_cert() atexit.register(exit_handler) if PY3: import http.server import socketserver import ssl logging.info('Server running... https://{}:{}'.format(server_host, server_port)) httpd = socketserver.TCPServer((server_host, server_port), http.server.SimpleHTTPRequestHandler) httpd.socket = ssl.wrap_socket(httpd.socket, certfile=ssl_cert_path, server_side=True) else: import BaseHTTPServer import SimpleHTTPServer import ssl logging.info('Server running... https://{}:{}'.format(server_host, server_port)) httpd = BaseHTTPServer.HTTPServer((server_host, server_port), SimpleHTTPServer.SimpleHTTPRequestHandler) httpd.socket = ssl.wrap_socket(httpd.socket, certfile=ssl_cert_path, server_side=True) httpd.serve_forever()
def run_action(self): current_cwd = os.getcwd() self.ui.print_msg("Starting web server on port %s" % self.vars["port"][0]) print "\033[33m" try: httpd = BaseHTTPServer.HTTPServer(('0.0.0.0', int(self.vars["port"][0])), SimpleHTTPServer.SimpleHTTPRequestHandler) if self.vars["https"][0].lower() == "true": httpd.socket = ssl.wrap_socket(httpd.socket, certfile=self.vars["certificate"][0], server_side=True) os.chdir(self.vars["folder"][0]) httpd.serve_forever() except KeyboardInterrupt: print "\033[00m" self.ui.print_msg("Stopping web server") except: print "\033[00m" self.ui.print_error("The web server raised an exception") os.chdir(current_cwd)
def run(self, open_immediately, port): ''' Serves the `www` directory. Args: open_immediately: Whether to open the web browser immediately port: The port at which to serve the graph ''' os.chdir(self.directory) handler = http.SimpleHTTPRequestHandler handler.extensions_map.update({ '.webapp': 'application/x-web-app-manifest+json', }) server = socketserver.TCPServer(('', port), handler) server.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) address = 'http://localhost:{0}/graph.html'.format(port) log.info('Serving at %s', address) if open_immediately: log.debug('Opening webbrowser') webbrowser.open(address) server.serve_forever()
def do_GET(self): matches = re.match(PATH_REGEX, self.path) if matches: path = matches.group(1) elif self.path == '/web/' and self.server.apiGatewayClientUrl != None: f = open("web/index.html") html = f.read() f.close() html = html.replace('src="apiGateway-js-sdk/','src="'+self.server.apiGatewayClientUrl+'/apiGateway-js-sdk/') self.respond(200,html) return else: return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self) #return super(MyHandler,self).do_GET() #self.respond(404, "Not Found") #return try: page = self.server.serverIface.getPage('<httpserver>', path) self.respond(200, page["html"]) except custom_exceptions.NotFound: self.respond(404, "Not Found") except: self.respond(500, "Internal Error")
def send_head(self): """Version of send_head that support CGI scripts""" if self.is_cgi(): return self.run_cgi() else: return SimpleHTTPServer.SimpleHTTPRequestHandler.send_head(self)
def do_GET(self): if self.is_tryton_url(self.path): self.send_tryton_url(self.path) return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
def do_HEAD(self): if self.is_tryton_url(self.path): self.send_tryton_url(self.path) return SimpleHTTPServer.SimpleHTTPRequestHandler.do_HEAD(self)
def __init__(self, ctx, *args, **kwargs): self.ctx = ctx SimpleHTTPServer.SimpleHTTPRequestHandler.__init__( self, *args, **kwargs)
def do_GET(self): LOG.info('Got GET request: %s', self.path) fake_check = re.match(r'/fake', self.path) tenant_images_match = re.match(r'/tenant_images/(.*)$', self.path) deploy_steps_match = re.match( r'/v1/nodes/([^/]*)/vendor_passthru/deploy_steps', self.path) if fake_check is not None: LOG.info("This is 'fake' request.") self.path = os.path.join(self.ctx.htdocs, 'stubfile') elif tenant_images_match is not None: LOG.info("This is 'tenant-images' request: %s", self.path) tenant_image = tenant_images_match.group(1) self.path = os.path.join(self.ctx.images_path, tenant_image) elif deploy_steps_match is not None: with open('{}.pub'.format(self.ctx.ssh_key)) as data: ssh_key = data.read().rstrip() data = { 'name': 'inject-ssh-keys', 'payload': { 'ssh-keys': { 'root': [ssh_key] } } } tmp = tempfile.NamedTemporaryFile() json.dump(data, tmp) tmp.flush() self.path = tmp.name return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
def do_GET(self): path = self.path.lstrip('/').split('?')[0] print("GET: ", path) # is the file in the redirects table? if path in self.redirects: path_to = self.redirects[path] print("REDIRECT TO ", path_to) self.send_response(301) self.send_header('location', path_to) self.end_headers() return True else: # serve the file! self.path = path return SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
def handle_one_request(self): try: SimpleHTTPServer.SimpleHTTPRequestHandler.handle_one_request(self) except: pass
def handle_one_request(self): try: SimpleHTTPServer.SimpleHTTPRequestHandler.handle_one_request(self) except KeyboardInterrupt: raise except Exception, e: logging.error('Exception in HTTP request handler: %s' % e)
def do_POST(self): post_request = [] printt(3, "%s - sent POST request." %self.address_string()) form = cgi.FieldStorage(self.rfile, headers=self.headers, environ={'REQUEST_METHOD':'POST', 'CONTENT_TYPE':self.headers['Content-Type'],}) try: from core.shell import url logger = open("%s.log" %url.replace("https://", "").replace("http://", "").split("/")[0], "a") logger.write("\n## %s - Data for %s\n\n" %(time.strftime("%H:%M:%S - %d/%m/%y"), url)) for tag in form.list: tmp = str(tag).split("(")[1] key,value = tmp.replace(")", "").replace("\'", "").replace(",", "").split() post_request.append("%s %s" %(key,value)) printt(2, "%s => %s" %(key,value)) logger.write("%s => %s\n" %(key,value)) logger.close() from core.shell import action_url create_post(url,action_url, post_request) SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self) except socerr as e: printt(3, "%s igonring ..." %str(e)) except Exception as e: printt(3, "%s igonring ..." %str(e))
def serve(port=PORT): '''Serves test XML files over HTTP''' # Make sure we serve from the tests' XML directory os.chdir(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'xml')) Handler = SimpleHTTPServer.SimpleHTTPRequestHandler class TestServer(SocketServer.TCPServer): allow_reuse_address = True httpd = TestServer(("", PORT), Handler) print 'Serving test HTTP server at port', PORT httpd_thread = Thread(target=httpd.serve_forever) httpd_thread.setDaemon(True) httpd_thread.start()
def test_simple_HTTP(self) : # ============== Setting up an HTTP server at 'http://localhost:8001/' in current directory try : PORT = int(sys.argv[1]) except : PORT = randint(1025, 65535) try : Handler = SimpleHTTPServer.SimpleHTTPRequestHandler httpd = SocketServer.TCPServer(("", PORT), Handler) except : Handler = http.server.SimpleHTTPRequestHandler httpd = socketserver.TCPServer(("", PORT), Handler) print ("Serving at port %d" % PORT) http_thread = Thread( target = httpd.serve_forever, ) http_thread.daemon = True # ============== Starting the HTTP server http_thread.start() # ============== Wait until HTTP server is ready sleep(1) with remote_repo(['test_package'], base_url = 'http://localhost:%d/' % PORT) : from test_package import module1 self.assertTrue(module1.dummy_str) # If this point is reached then the module1 is imported succesfully!
def setUp(self): super(WebSocketRequestHandlerTestCase, self).setUp() self.stubs = stubout.StubOutForTesting() self.tmpdir = tempfile.mkdtemp('-websockify-tests') # Mock this out cause it screws tests up self.stubs.Set(os, 'chdir', lambda *args, **kwargs: None) self.stubs.Set(SimpleHTTPRequestHandler, 'send_response', lambda *args, **kwargs: None)
def test_normal_get_with_only_upgrade_returns_error(self): server = self._get_server(web=None) handler = websocket.WebSocketRequestHandler( FakeSocket('GET /tmp.txt HTTP/1.1'), '127.0.0.1', server) def fake_send_response(self, code, message=None): self.last_code = code self.stubs.Set(SimpleHTTPRequestHandler, 'send_response', fake_send_response) handler.do_GET() self.assertEqual(handler.last_code, 405)
def test_list_dir_with_file_only_returns_error(self): server = self._get_server(file_only=True) handler = websocket.WebSocketRequestHandler( FakeSocket('GET / HTTP/1.1'), '127.0.0.1', server) def fake_send_response(self, code, message=None): self.last_code = code self.stubs.Set(SimpleHTTPRequestHandler, 'send_response', fake_send_response) handler.path = '/' handler.do_GET() self.assertEqual(handler.last_code, 404)
def __init__(self, request, client_address, base_server): self.taker = base_server.taker self.base_server = base_server SimpleHTTPServer.SimpleHTTPRequestHandler.__init__( self, request, client_address, base_server)
def handle_one_request(self): try: httpserver.SimpleHTTPRequestHandler.handle_one_request(self) except socket.error: if not hasattr(self, "_host_disconnected"): self._host_disconnected = 1 dprint("Host disconnected") elif self._host_disconnected < MAX_DISCONNECT: self._host_disconnected += 1 dprint("Host disconnected: %d" % self._host_disconnected) else: dprint("Closed connection to avoid infinite loop") self.close_connection = True
def do_GET(self): if self.path =='/admin': self.wfile.write('This page is only for Admins') self.wfile.write(self.headers) else: SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
def do_GET(self): if self.path =='/cgi-bin/': CGIHTTPServer.CGIHTTPRequestHandler.do_GET(self) else: SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
def start_web_server(): import SimpleHTTPServer import SocketServer import socket try: port = int(web_server_port) host = web_server_ip # Do not attempt to fix code warnings in the below class, it is perfect. class QuietHandler(SimpleHTTPServer.SimpleHTTPRequestHandler): # quiet server logs def log_message(self, format, *args): return # serve from www folder under current working dir def translate_path(self, path): return SimpleHTTPServer.SimpleHTTPRequestHandler.translate_path(self, '/www' + path) global server SocketServer.TCPServer.allow_reuse_address = True server = SocketServer.TCPServer((host, port), QuietHandler) if host == "0.0.0.0": # Get all addresses that we could listen on the port specified addresses = [i[4][0] for i in socket.getaddrinfo(socket.gethostname().split('.')[0], port)] addresses = [i for i in addresses if ':' not in i] # Filter out all IPv6 addresses addresses.append('127.0.0.1') # getaddrinfo doesn't always get localhost hosts = list(set(addresses)) # Make list unique else: hosts = [host] serving_msg = "http://{0}:{1}/index.html".format(hosts[0], port) for host in hosts[1:]: serving_msg += ", http://{0}:{1}/index.html".format(host, port) print 'Started WebServer, lendingbot status available at {0}'.format(serving_msg) server.serve_forever() except Exception as ex: ex.message = ex.message if ex.message else str(ex) print('Failed to start WebServer: {0}'.format(ex.message))
def _initserver(self, port): self.PORT = 8000 self.httpd = HTTPServer(('localhost', self.PORT), SimpleHTTPRequestHandler) print ('Serving at port', self.PORT) self.th = threading.Thread(target=self.httpd.serve_forever) self.th.daemon = True self.th.start() time.sleep(5)
def start_server(port): server = HTTPServer(('', port), SimpleHTTPRequestHandler) thread = threading.Thread(target = server.serve_forever) thread.daemon = True try: thread.start() except KeyboardInterrupt: server.shutdown() sys.exit(0)
def end_headers(self, filter, apk, download_url): # set extra headers for apk self.send_header('Content-Type', 'application/json') # set to "dummy" for testing self.send_header("filter", filter) self.send_header("apk", apk) self.send_header("download_url", download_url) SimpleHTTPServer.SimpleHTTPRequestHandler.end_headers(self)
def run(self): self.dependency_check() self.datafolders_check() # Setup SimpleHTTPServer Handler = SimpleHTTPServer.SimpleHTTPRequestHandler httpd = SocketServer.TCPServer(("", self.port), Handler) httpd.allow_reuse_address = True # Launch SimpleHTTPServer in a seperate Process os.chdir(self.webserver_directory) server_process = multiprocessing.Process(target=httpd.serve_forever) server_process.daemon = True server_process.start() print('HTTP Server Launched on Port: %s ' % (self.port)) # Run Certbot while HTTP Server process is live. try: p1 = Popen(["certbot", "--webroot", "--non-interactive", "certonly", "--text", "--rsa-key-size", "4096", "--agree-tos", "--webroot-path", "/var/www/WiFiSuite/", "-m " + self.email, "-d " + self.certname], stdout=PIPE, stderr=PIPE) except Exception as e: print(' Error: %s' % (e)) if self.debug: print(white('Debug')+'Certbot STDOUT/STDERR below:') # Print STDOUT for line in iter(p1.stdout.readline, ''): sys.stdout.write(line) # Print STDERR for line in iter(p1.stderr.readline, ''): sys.stderr.write(line) # Terminate HTTP Server process server_process.terminate() os.chdir(self.cwd)
def do_POST(self): post_request = [] printt(3, "%s - sent POST request." %self.address_string()) form = cgi.FieldStorage(self.rfile, headers=self.headers, environ={'REQUEST_METHOD':'POST', 'CONTENT_TYPE':self.headers['Content-Type'],}) try: from core.shell import url logger = open("%s.log" %url.replace("https://", "").replace("http://", "").split("/")[0], "a") logger.write("\n## %s - Data for %s\n\n" %(time.strftime("%H:%M:%S - %d/%m/%y"), url)) for tag in form.list: tmp = str(tag).split("(")[1] key,value = tmp.replace(")", "").replace("\'", "").replace(",", "").split() post_request.append("%s %s" %(key,value)) printt(2, "%s => %s" %(key,value)) logger.write("%s => %s\n" %(key,value)) logger.close() from core.shell import action_url create_post(url,action_url, post_request) SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self) except socerr as e: print "" except Exception as e: print ""
def serve(): import SimpleHTTPServer import SocketServer os.chdir('www') PORT = 8000 Handler = SimpleHTTPServer.SimpleHTTPRequestHandler httpd = SocketServer.TCPServer(("", PORT), Handler) print "serving at port", PORT httpd.serve_forever()
def do_GET(self): logging.warning("======= GET STARTED =======") logging.warning(self.headers) SimpleHTTPServer.SimpleHTTPRequestHandler.do_GET(self)
def start_server(self): if self.webserverStarted == False and self.CONTROLLER_SERVER_PATH != '': """Changing dir to the schedules dir.""" web_dir = os.path.abspath(os.path.join(os.path.dirname( __file__ ), '..', 'schedules')) os.chdir(web_dir) PORT = int(self.CONTROLLER_SERVER_PORT) class MyHandler(SimpleHTTPServer.SimpleHTTPRequestHandler): def log_message(self, format, *args): return global httpd try: #print "Starting webserver at port: ", PORT # create the httpd handler for the simplehttpserver # we set the allow_reuse_address incase something hangs can still bind to port class ReusableTCPServer(SocketServer.TCPServer): allow_reuse_address=True # specify the httpd service on 0.0.0.0 (all interfaces) on port 80 httpd = ReusableTCPServer(("0.0.0.0", PORT),MyHandler) # thread this mofo thread.start_new_thread(httpd.serve_forever,()) # handle keyboard interrupts except KeyboardInterrupt: core.print_info("Exiting the SET web server...") httpd.socket.close() except socket.error, exc: print "Caught exception socket.error : %s" % exc # handle the rest #except Exception: # print "[*] Exiting the SET web server...\n" # httpd.socket.close() self.webserverStarted = True
