Best Python code snippet using playwright-python
https_server.py
Source:https_server.py
1"""2// -------------------------------------------------------------3// author Giga4// project qeeqbox/honeypots5// email gigaqeeq@gmail.com6// description app.py (CLI)7// licensee AGPL-3.08// -------------------------------------------------------------9// contributors list qeeqbox/social-analyzer/graphs/contributors10// -------------------------------------------------------------11"""12from uuid import uuid413from honeypots.helper import close_port_wrapper, get_free_port, kill_server_wrapper, server_arguments, setup_logger, disable_logger14from os import path15from subprocess import Popen16from twisted.python import log as tlog17from random import choice18from twisted.web.resource import Resource19from twisted.web.server import Site20from twisted.internet import reactor, ssl21from tempfile import gettempdir, _get_candidate_names22from time import sleep23from requests.packages.urllib3 import disable_warnings24from requests import get, post25from cgi import FieldStorage26from OpenSSL import crypto27from warnings import filterwarnings28filterwarnings(action='ignore', module='.*OpenSSL.*')29disable_warnings()30class QHTTPSServer():31 def __init__(self, ip=None, port=None, username=None, password=None, mocking=False, config=''):32 self.ip = ip or '0.0.0.0'33 self.port = port or 44334 self.username = username or "test"35 self.password = password or "test"36 self.mocking = mocking or ''37 self.key = path.join(gettempdir(), next(_get_candidate_names()))38 self.cert = path.join(gettempdir(), next(_get_candidate_names()))39 self.random_servers = ['Apache', 'nginx', 'Microsoft-IIS/7.5', 'Microsoft-HTTPAPI/2.0', 'Apache/2.2.15', 'SmartXFilter', 'Microsoft-IIS/8.5', 'Apache/2.4.6', 'Apache-Coyote/1.1',40 'Microsoft-IIS/7.0', 'Apache/2.4.18', 'AkamaiGHost', 'Apache/2.2.25', 'Microsoft-IIS/10.0', 'Apache/2.2.3', 'nginx/1.12.1', 'Apache/2.4.29', 'cloudflare', 'Apache/2.2.22']41 self.process = None42 self.uuid = 'honeypotslogger'43 self.config = config44 if config:45 self.logs = setup_logger(self.uuid, config)46 else:47 self.logs = setup_logger(self.uuid, None)48 disable_logger(1, tlog)49 def CreateCert(self, host_name, key, cert):50 pk = crypto.PKey()51 pk.generate_key(crypto.TYPE_RSA, 2048)52 c = crypto.X509()53 c.get_subject().C = 'US'54 c.get_subject().ST = 'OR'55 c.get_subject().L = 'None'56 c.get_subject().O = 'None'57 c.get_subject().OU = 'None'58 c.get_subject().CN = next(_get_candidate_names())59 c.set_serial_number(0)60 before, after = (0, 60 * 60 * 24 * 365 * 2)61 c.gmtime_adj_notBefore(before)62 c.gmtime_adj_notAfter(after)63 c.set_issuer(c.get_subject())64 c.set_pubkey(pk)65 c.sign(pk, 'sha256')66 open(cert, "wb").write(crypto.dump_certificate(crypto.FILETYPE_PEM, c))67 open(key, "wb").write(crypto.dump_privatekey(crypto.FILETYPE_PEM, pk))68 def https_server_main(self):69 _q_s = self70 class MainResource(Resource):71 isLeaf = True72 home_file = b'''73<!DOCTYPE html>74<html>75 <head>76 <link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.0.0-beta.3/css/bootstrap.min.css" />77 <link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/font-awesome/4.7.0/css/font-awesome.min.css" />78 <meta http-equiv="content-type" content="text/html;charset=utf-8" />79 <title>Login</title>80 <style>81 body,html{height: 100%;text-align: center;},82 </style>83 </head>84 <body>85 <div class="container-fluid h-100">86 <div class="row justify-content-center h-100 align-items-center">87 <div class="col col-xl-3">88 <b>We'll back soon..</b>89 </div>90 </div>91 </div>92 </body>93</html>'''94 login_file = b'''<!DOCTYPE html>95<html>96 <head>97 <link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.0.0-beta.3/css/bootstrap.min.css" />98 <link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/font-awesome/4.7.0/css/font-awesome.min.css" />99 <meta http-equiv="content-type" content="text/html;charset=utf-8" />100 <title>Login</title>101 <style>body,html {height: 100%;}</style>102 </head>103 <body>104 <div class="container-fluid h-100">105 <div class="row justify-content-center h-100 align-items-center">106 <div class="col col-xl-3">107 <form id="login" action="" method="post">108 <div class="form-group">109 <input class="form-control form-control-sm" name="username" type="text" placeholder="username" id="username">110 </div>111 <div class="form-group">112 <input class="form-control form-control-sm" name="password" type="password" placeholder="password" id="password">113 </div>114 <div class="form-group">115 <button class="btn btn-default btn-sm btn-block" type="submit">login</button>116 </div>117 </form>118 </div>119 </div>120 </div>121 </body>122</html>123'''124 if isinstance(_q_s.mocking, bool):125 if _q_s.mocking == True:126 server = choice(_q_s.random_servers)127 elif isinstance(_q_s.mocking, str):128 server = _q_s.mocking129 def render(self, request):130 headers = {}131 try:132 def check_bytes(string):133 if isinstance(string, bytes):134 return string.decode()135 else:136 return str(string)137 for item, value in dict(request.requestHeaders.getAllRawHeaders()).items():138 headers.update(139 {check_bytes(item): ','.join(map(check_bytes, value))})140 except BaseException:141 pass142 _q_s.logs.info({'server': 'https_server', 'action': 'connection',143 'ip': request.getClientIP(), 'request': headers})144 if self.server != "":145 request.responseHeaders.removeHeader("Server")146 request.responseHeaders.addRawHeader("Server", self.server)147 if request.method == b"GET":148 _q_s.logs.info(149 {'server': 'https_server', 'action': 'get', 'ip': request.getClientIP()})150 if request.uri == b"/login.html":151 if _q_s.username != '' and _q_s.password != '':152 request.responseHeaders.addRawHeader(153 "Content-Type", "text/html; charset=utf-8")154 return self.login_file155 request.responseHeaders.addRawHeader(156 "Content-Type", "text/html; charset=utf-8")157 return self.home_file158 elif request.method == b"POST":159 self.headers = request.getAllHeaders()160 _q_s.logs.info(161 {'server': 'https_server', 'action': 'post', 'ip': request.getClientIP()})162 if request.uri == b"/login.html" or b'/':163 if _q_s.username != '' and _q_s.password != '':164 form = FieldStorage(fp=request.content, headers=self.headers, environ={165 'REQUEST_METHOD': 'POST', 'CONTENT_TYPE': self.headers[b'content-type'], })166 if 'username' in form and 'password' in form:167 if form['username'].value == _q_s.username and form['password'].value == _q_s.password:168 _q_s.logs.info({'server': 'https_server', 'action': 'login', 'status': 'success', 'ip': request.getClientIP(169 ), 'username': _q_s.username, 'password': _q_s.password})170 else:171 _q_s.logs.info({'server': 'https_server', 'action': 'login', 'status': 'failed', 'ip': request.getClientIP(172 ), 'username': form['username'].value, 'password': form['password'].value})173 request.responseHeaders.addRawHeader(174 "Content-Type", "text/html; charset=utf-8")175 return self.home_file176 else:177 request.responseHeaders.addRawHeader(178 "Content-Type", "text/html; charset=utf-8")179 return self.home_file180 self.CreateCert("localhost", self.key, self.cert)181 ssl_context = ssl.DefaultOpenSSLContextFactory(self.key, self.cert)182 reactor.listenSSL(self.port, Site(MainResource()), ssl_context)183 reactor.run()184 def run_server(self, process=False, auto=False):185 if process:186 if auto:187 port = get_free_port()188 if port > 0:189 self.port = port190 self.process = Popen(['python3', path.realpath(__file__), '--custom', '--ip', str(self.ip), '--port', str(self.port), '--username', str(191 self.username), '--password', str(self.password), '--mocking', str(self.mocking), '--config', str(self.config), '--uuid', str(self.uuid)])192 if self.process.poll() is None:193 self.logs.info({'server': 'https_server', 'action': 'process', 'status': 'success', 'route': '/login.html',194 'ip': self.ip, 'port': self.port, 'username': self.username, 'password': self.password})195 else:196 self.logs.info({'server': 'https_server', 'action': 'process', 'status': 'error',197 'ip': self.ip, 'port': self.port, 'username': self.username, 'password': self.password})198 else:199 self.logs.info({'server': 'https_server', 'action': 'setup', 'status': 'error',200 'ip': self.ip, 'port': self.port, 'username': self.username, 'password': self.password})201 elif self.close_port() and self.kill_server():202 self.process = Popen(['python3', path.realpath(__file__), '--custom', '--ip', str(self.ip), '--port', str(self.port), '--username', str(203 self.username), '--password', str(self.password), '--mocking', str(self.mocking), '--config', str(self.config), '--uuid', str(self.uuid)])204 if self.process.poll() is None:205 self.logs.info({'server': 'https_server', 'action': 'process', 'status': 'success', 'route': '/login.html',206 'ip': self.ip, 'port': self.port, 'username': self.username, 'password': self.password})207 else:208 self.logs.info({'server': 'https_server', 'action': 'process', 'status': 'error',209 'ip': self.ip, 'port': self.port, 'username': self.username, 'password': self.password})210 else:211 self.https_server_main()212 def test_server(self, ip=None, port=None, username=None, password=None):213 try:214 sleep(2)215 _ip = ip or self.ip216 _port = port or self.port217 _username = username or self.username218 _password = password or self.password219 get('https://{}:{}'.format(_ip, _port), verify=False)220 post('https://{}:{}'.format(_ip, _port), data={'username': (None,221 _username), 'password': (None, _password)}, verify=False)222 except BaseException:223 pass224 def close_port(self):225 ret = close_port_wrapper('https_server', self.ip, self.port, self.logs)226 return ret227 def kill_server(self):228 ret = kill_server_wrapper('https_server', self.uuid, self.process)229 return ret230if __name__ == '__main__':231 parsed = server_arguments()232 if parsed.docker or parsed.aws or parsed.custom:233 qhttpsserver = QHTTPSServer(ip=parsed.ip, port=parsed.port, username=parsed.username,234 password=parsed.password, mocking=parsed.mocking, config=parsed.config)...
test_handler.py
Source:test_handler.py
1import pytest2import base643import json4import requests5import ssl6import threading7from dummyserver import handler8from dummyserver import server9SERVER_CRT = "test/test_handler/server-bundle.crt"10SERVER_CA = "test/test_handler/rootCA.crt"11CLIENT_CRT = "test/test_handler/client.crt"12CLIENT_KEY = "test/test_handler/client.key"13CLIENT_CA = "test/test_handler/rootCA.crt"14HTTP_PORT = 808015HTTPS_PORT = 818116HTTPSM_PORT = 828217MGMT_PORT = 838318USERNAME = "testuser"19PASSWORD = "testpass"20CREDENTIALS = base64.b64encode(b"testuser:testpass")21ERROR_RANGE = list(range(402, 418))+list(range(500, 505))22@pytest.fixture(scope="module")23def mgmt_server():24 # Startup25 print("Starting server")26 http_server = server.ThreadedHTTPServer(("", MGMT_PORT), handler.DummyHandler)27 http_thread = threading.Thread(target=http_server.serve_forever)28 http_thread.setDaemon(True)29 http_thread.start()30 yield http_server31 # Teardown32 print("Stopping server")33 http_server.shutdown()34 http_server.server_close()35@pytest.fixture(scope="module")36def http_server_with_credentials():37 # Startup38 print("Starting server")39 http_server = server.ThreadedHTTPServer(("", HTTP_PORT), handler.HTTPHandler)40 http_thread = threading.Thread(target=http_server.serve_forever)41 http_thread.setDaemon(True)42 http_thread.start()43 yield http_server44 # Teardown45 print("Stopping server")46 http_server.shutdown()47 http_server.server_close()48@pytest.fixture(scope="module")49def https_server_with_credentials():50 # Startup51 print("Starting server")52 context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)53 context.verify_mode = ssl.CERT_NONE54 context.load_cert_chain(certfile=SERVER_CRT)55 handler.HTTPHandler.credentials = CREDENTIALS56 https_server = server.ThreadedHTTPServer(("", HTTPS_PORT), handler.HTTPHandler)57 https_server.socket = context.wrap_socket (https_server.socket, server_side=True)58 https_thread = threading.Thread(target=https_server.serve_forever)59 https_thread.setDaemon(True)60 https_thread.start()61 yield https_server62 # Teardown63 print("Stopping server")64 https_server.shutdown()65 https_server.server_close()66@pytest.fixture(scope="module")67def https_server_with_mutual_auth_with_credentials():68 # Startup69 print("Starting server")70 context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)71 context.verify_mode = ssl.CERT_REQUIRED72 context.load_cert_chain(certfile=SERVER_CRT)73 context.load_verify_locations(cafile=CLIENT_CA)74 handler.HTTPHandler.credentials = CREDENTIALS75 https_server = server.ThreadedHTTPServer(("", HTTPSM_PORT), handler.HTTPHandler)76 https_server.socket = context.wrap_socket (https_server.socket, server_side=True)77 https_thread = threading.Thread(target=https_server.serve_forever)78 https_thread.setDaemon(True)79 https_thread.start()80 yield https_server81 # Teardown82 print("Stopping server")83 https_server.shutdown()84 https_server.server_close()85def test_http_with_credentials(http_server_with_credentials):86 r = requests.get(87 url = 'http://localhost:{}'.format(HTTP_PORT),88 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)89 )90 assert r.status_code == 20091def test_http_with_credentials_fail_url(http_server_with_credentials):92 r = requests.get(93 url = 'http://localhost:{}/fail'.format(HTTP_PORT),94 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)95 )96 assert r.status_code in ERROR_RANGE97def test_https_with_credentials(https_server_with_credentials):98 r = requests.get(99 url = 'https://localhost:{}'.format(HTTPS_PORT),100 verify=SERVER_CA,101 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)102 )103 assert r.status_code == 200104def test_https_with_credentials_ignore_client_cert(https_server_with_credentials):105 r = requests.get(106 url = 'https://localhost:{}'.format(HTTPS_PORT),107 cert=(CLIENT_CRT, CLIENT_KEY),108 verify=SERVER_CA,109 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)110 )111 assert r.status_code == 200112def test_https_with_mutual_auth_with_credentials(https_server_with_mutual_auth_with_credentials):113 r = requests.get(114 url = 'https://localhost:{}'.format(HTTPSM_PORT),115 cert=(CLIENT_CRT, CLIENT_KEY),116 verify=SERVER_CA,117 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)118 )119 assert r.status_code == 200120def test_https_with_mutual_auth_with_credentials_wrong_password(https_server_with_mutual_auth_with_credentials):121 r = requests.get(122 url = 'https://localhost:{}'.format(HTTPSM_PORT),123 cert=(CLIENT_CRT, CLIENT_KEY),124 verify=SERVER_CA,125 auth=requests.auth.HTTPBasicAuth(USERNAME, '')126 )127 assert r.status_code == 401128def test_https_with_mutual_auth_with_credentials_ssl_failure(https_server_with_mutual_auth_with_credentials):129 with pytest.raises(requests.exceptions.SSLError):130 r = requests.get(131 url = 'https://localhost:{}'.format(HTTPSM_PORT),132 verify=SERVER_CA,133 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)134 )135def test_mgmt_clear(mgmt_server, https_server_with_credentials):136 r = requests.get(137 url = 'https://localhost:{}'.format(HTTPS_PORT),138 verify=SERVER_CA,139 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)140 )141 r = requests.get(142 url = 'http://localhost:{}/clear'.format(MGMT_PORT),143 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)144 )145 r = requests.get(146 url = 'http://localhost:{}/stats'.format(MGMT_PORT),147 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)148 )149 assert r.content == b'{}'150def test_mgmt_stats(mgmt_server, https_server_with_credentials, https_server_with_mutual_auth_with_credentials):151 r = requests.get(152 url = 'http://localhost:{}/clear'.format(MGMT_PORT),153 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)154 )155 r = requests.get(156 url = 'https://localhost:{}'.format(HTTPS_PORT),157 verify=SERVER_CA,158 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)159 )160 r = requests.get(161 url = 'https://localhost:{}'.format(HTTPSM_PORT),162 cert=(CLIENT_CRT, CLIENT_KEY),163 verify=SERVER_CA,164 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)165 )166 r = requests.get(167 url = 'https://localhost:{}/fail'.format(HTTPSM_PORT),168 cert=(CLIENT_CRT, CLIENT_KEY),169 verify=SERVER_CA,170 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)171 )172 r = requests.get(173 url = 'http://localhost:{}/stats'.format(MGMT_PORT),174 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)175 )176 assert r.content == b'{"/": 2, "/fail": 1}'177def test_mgmt_history(mgmt_server, https_server_with_credentials):178 r = requests.get(179 url = 'http://localhost:{}/clear'.format(MGMT_PORT),180 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)181 )182 r = requests.get(183 url = 'https://localhost:{}'.format(HTTPS_PORT),184 verify=SERVER_CA,185 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)186 )187 r = requests.get(188 url = 'http://localhost:{}/history'.format(MGMT_PORT),189 auth=requests.auth.HTTPBasicAuth(USERNAME, PASSWORD)190 )191 content = json.loads(r.content)192 assert len(content) == 1193 assert content[0]["path"] == '/'...
ZenodoTest.py
Source:ZenodoTest.py
1import os2from os.path import join as p3from unittest.mock import Mock4import shutil5import pytest6from owmeta_core.datasource_loader import LoadFailed7from owmeta_movement.zenodo import ZenodoRecordDirLoader8import requests9def test_can_load(zenodo_dir_loader):10 '''11 Test that we provide for a matching record ID with no file name12 '''13 cut, https_server = zenodo_dir_loader14 os.mkdir(p(https_server.base_directory, 'record'))15 shutil.copyfile(p('tests', 'testdata', 'zenodo_record_20210122.html'),16 p(https_server.base_directory, 'record', '4074963'))17 ob = Mock()18 ob.zenodo_base_url.return_value = https_server.url19 ob.zenodo_id.return_value = 407496320 ob.zenodo_file_name.return_value = None21 assert cut.can_load(ob)22def test_can_load_no_file_attr(zenodo_dir_loader):23 '''24 Test that we provide for a matching record ID with no file name25 '''26 cut, https_server = zenodo_dir_loader27 os.mkdir(p(https_server.base_directory, 'record'))28 shutil.copyfile(p('tests', 'testdata', 'zenodo_record_20210122.html'),29 p(https_server.base_directory, 'record', '4074963'))30 ob = Mock()31 ob.zenodo_base_url.return_value = https_server.url32 ob.zenodo_id.return_value = 407496333 ob.zenodo_file_name.side_effect = AttributeError34 assert cut.can_load(ob)35def test_can_load_fail_not_found(zenodo_dir_loader):36 '''37 Test that we provide for a matching record ID with no file name38 '''39 cut, https_server = zenodo_dir_loader40 ob = Mock()41 ob.zenodo_base_url.return_value = https_server.url42 ob.zenodo_id.return_value = 407496343 ob.zenodo_file_name.return_value = None44 assert not cut.can_load(ob)45def test_can_load_file_fail_not_found(zenodo_dir_loader):46 '''47 Test that we provide for a matching record ID with no file name48 '''49 cut, https_server = zenodo_dir_loader50 recorddir = p(https_server.base_directory, 'record')51 os.mkdir(recorddir)52 shutil.copyfile(p('tests', 'testdata', 'zenodo_record_20210122.html'),53 p(recorddir, '4074963'))54 ob = Mock()55 ob.zenodo_base_url.return_value = https_server.url56 ob.zenodo_id.return_value = 407496357 ob.zenodo_file_name.return_value = 'CeMEE_MWT_MA.tar.gz'58 assert not cut.can_load(ob)59def test_can_load_file(zenodo_dir_loader):60 '''61 Test that we provide for a matching record ID with no file name62 '''63 cut, https_server = zenodo_dir_loader64 filesdir = p(https_server.base_directory, 'record', '4074963', 'files')65 os.makedirs(filesdir)66 file_name = p(filesdir, 'CeMEE_MWT_MA.tar.gz')67 with open(file_name, 'w') as f:68 f.write('blah')69 ob = Mock()70 ob.zenodo_base_url.return_value = https_server.url71 ob.zenodo_id.return_value = 407496372 ob.zenodo_file_name.return_value = 'CeMEE_MWT_MA.tar.gz'73 assert cut.can_load(ob)74def test_load_file(zenodo_dir_loader):75 cut, https_server = zenodo_dir_loader76 filesdir = p(https_server.base_directory, 'record', '4074963', 'files')77 os.makedirs(filesdir)78 target_file_name = 'CeMEE_MWT_MA.tar.gz'79 file_name = p(filesdir, target_file_name)80 with open(file_name, 'w') as f:81 f.write('blah')82 ob = Mock()83 ob.zenodo_base_url.return_value = https_server.url84 ob.zenodo_id.return_value = 407496385 ob.zenodo_file_name.return_value = target_file_name86 dsdir = cut.load(ob)87 with open(p(dsdir, target_file_name), 'r') as f:88 assert f.read() == 'blah'89def test_load_some_files(tmp_path, https_server):90 '''91 Loads any files available--no error is thrown if some are absent.92 '''93 recorddir = p(https_server.base_directory, 'record')94 os.mkdir(recorddir)95 shutil.copyfile(p('tests', 'testdata', 'zenodo_record_20210122.html'),96 p(recorddir, '4074963.html'))97 def handler(server_data):98 class handler_class(server_data.basic_handler):99 def do_GET(self):100 if self.path.endswith('4074963'):101 self.handle_request(200)102 with open(p(recorddir, '4074963.html'), 'rb') as f:103 shutil.copyfileobj(f, self.wfile)104 else:105 super().do_GET()106 return handler_class107 https_server.make_server(handler)108 https_server.restart()109 session = requests.Session()110 https_server.trust_server(session)111 cut = ZenodoRecordDirLoader(tmp_path, lambda: session)112 filesdir = p(https_server.base_directory, 'record', '4074963', 'files')113 os.makedirs(filesdir)114 target_file_name = 'CeMEE_MWT_MA.tar.gz'115 file_name = p(filesdir, target_file_name)116 with open(file_name, 'w') as f:117 f.write('blah')118 ob = Mock()119 ob.zenodo_base_url.return_value = https_server.url120 ob.zenodo_id.return_value = 4074963121 ob.zenodo_file_name.return_value = None122 with pytest.raises(LoadFailed):123 cut.load(ob)124@pytest.mark.inttest125def test_load_real():126 '''127 Test actually loading from Zenodo128 '''129@pytest.fixture130def zenodo_dir_loader(tmp_path, https_server):131 session = requests.Session()132 https_server.trust_server(session)...
service.py
Source:service.py
1"""Network service to receive requests"""2import getopt3import logging4import multiprocessing5import os6import ssl7import sys8from .handler import handler9from http.server import HTTPServer10from socketserver import ThreadingMixIn11from .tcp.service import Service as TcpService12class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):13 """Handle requests in a separate thread."""14class HttpsService(multiprocessing.Process):15 """Handles HTTPS protocol network activity"""16 def __init__(self, key_folder, port, params, server_class=ThreadedHTTPServer):17 super().__init__()18 logging.info(f'SSL with {key_folder + "/key.pem"} and {key_folder + "/cert.pem"} is used')19 self.https_server = server_class(('', port), handler(params))20 self.https_server.socket = ssl.wrap_socket(self.https_server.socket,21 keyfile=key_folder + "/key.pem",22 certfile=key_folder + '/cert.pem',23 server_side=True)24 def run(self) -> None:25 """Starts service"""26 try:27 self.https_server.serve_forever()28 except KeyboardInterrupt:29 self.https_server.server_close()30 def join(self, timeout=None) -> None:31 """Implements service thread-safe stop"""32 if super().is_alive():33 self.https_server.server_close()34 super().join(timeout)35class Service:36 """Program launcher. Analyses terminal options and starts http server"""37 @staticmethod38 def print_options():39 """Informs about program terminal arguments"""40 print("params:\n\t-p(--ports) ports to bind[http,https,rtsp] (def 4555,4556,4557)\n\t"41 "-r(--root) files directory(req)\n\t"42 "-s(--segment) segment duration floor\n\t"43 "-c(--cache) cache segmentation as .*.cache files\n\t"44 "-b(--basic) user:password@realm (use Basic Authorization)\n\t"45 "-d(--digest) user:password@realm (use Digest Authorization)\n\t"46 "-k(--keys) directory with key.pem and cert.pem files (req. for https)\n\t"47 "-v(--verb) be verbose\n\t"48 "-h(--help) this help")49 def __init__(self):50 self.segment_makers = {}51 def run(self, ports, params, server_class=ThreadedHTTPServer):52 """Starts http server"""53 logging.basicConfig(level=logging.INFO)54 params['segment_makers'] = self.segment_makers55 tcp_server = TcpService(('', ports[2]), params)56 http_server = server_class(('', ports[0]), handler(params))57 ssl_key_folder = params.get('keys')58 https_server = None59 if ssl_key_folder and os.path.isfile(ssl_key_folder+'/key.pem') and os.path.isfile(ssl_key_folder+'/cert.pem'):60 https_server = HttpsService(ssl_key_folder, ports[1], params)61 logging.info('Starting...')62 try:63 tcp_server.start()64 if https_server:65 https_server.start()66 http_server.serve_forever()67 except KeyboardInterrupt:68 pass69 http_server.server_close()70 https_server and https_server.join()71 tcp_server.join()72 logging.info('Stopping')73def start():74 """Program start point"""75 argv = sys.argv[1:]76 try:77 opts, args = getopt.getopt(argv,78 "hp:r:s:b:d:ck:v",79 ["help",80 "ports=",81 "root=",82 "segment=",83 "basic=",84 "digest=",85 "cache",86 "keys=",87 "verb"])88 if args:89 Service.print_options()90 sys.exit()91 except getopt.GetoptError as error:92 print(error)93 sys.exit()94 ports = [4555, 4556, 4557]95 params = {}96 try:97 for opt, arg in opts:98 if opt in ('-h', '--help'):99 Service.print_options()100 sys.exit()101 elif opt in ('-p', '--ports'):102 ports = [int(k) for k in arg.split(',')]103 if len(ports) != 3:104 Service.print_options()105 sys.exit()106 elif opt in ('-r', '--root'):107 params['root'] = arg108 elif opt in ('-s', '--segment'):109 params['segment'] = float(arg)110 elif opt in ('-b', '--basic'):111 params['basic'] = arg112 elif opt in ('-d', '--digest'):113 params['digest'] = arg114 elif opt in ('-c', '--cache'):115 params['cache'] = True116 elif opt in ('-k', '-keys'):117 params['keys'] = arg118 elif opt in ('-v', '--verb'):119 params['verb'] = True120 except ValueError as error:121 print(error)122 Service.print_options()123 sys.exit()...
LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.
Get 100 minutes of automation test minutes FREE!!