How to use raw_http_response method in autotest

Best Python code snippet using autotest_python

http.py

Source:http.py

1# The MIT License (MIT)2#3# Copyright (C) 2021 Simon Meins4#5# Permission is hereby granted, free of charge, to any person obtaining a copy6# of this software and associated documentation files (the "Software"), to deal7# in the Software without restriction, including without limitation the rights8# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell9# copies of the Software, and to permit persons to whom the Software is10# furnished to do so, subject to the following conditions:11#12# The above copyright notice and this permission notice shall be included in13# all copies or substantial portions of the Software.14#15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN21# THE SOFTWARE.22# HTTP request library to send and receive data over the HTTP protocol.23# Import modules24# socket    : network communication25# json      : JSON manipulation26# namedtuple: special tuple object types27from socket import socket, AF_INET, SOCK_STREAM28from json import dumps29from collections import namedtuple30# Create namedtuple instances31url         = namedtuple("url", "scheme netloc path")32status_line = namedtuple("status_line", "http_version status_code reason_phrase")33# Define constants34HTTP_VERSION = "HTTP/1.1"35HTTP_POST    = "POST"36HTTP_GET     = "GET"37class HTTPResponse:38    39    """40    HTTPResponse class to parse an HTTP response.41    Constructor (__init__):42        parameter raw_http_response [string]: raw HTTP response string43    """44    def __init__(self, raw_http_response: str) -> None:45        self.raw_http_response = raw_http_response46    def __repr__(self) -> str:47        return f"<Response [{self.status_code}]>"48    @property49    def _status_line(self) -> status_line:50        """51        Parse the status line of the HTTP response (private method).52        returns: HTTP status line [namedtuple]53        """54        status_line_end_idx = self.raw_http_response.index("\r\n")55        status_line_raw     = self.raw_http_response[:status_line_end_idx].split()56        return status_line(57            http_version=status_line_raw[0], 58            status_code=status_line_raw[1], 59            reason_phrase=status_line_raw[2]60        )61    @property62    def status_code(self) -> int:63        """64        Return the status code of the HTTP response.65        returns: HTTP response status code [integer]66        """67        return int(self._status_line.status_code)68    @property69    def text(self) -> str:70        """71        Parse the HTTP response text (body data).72        returns: HTTP response text [string]73        """74        text_start_idx     = self.raw_http_response.index("\r\n\r\n")75        http_response_text = self.raw_http_response[text_start_idx + 4:]76        return http_response_text77    @property78    def headers(self) -> dict:79        """80        Parse the HTTP response headers.81        returns: HTTP response headers [dictionary]82        """83        84        header_start_idx = self.raw_http_response.index("\r\n")85        header_end_idx   = self.raw_http_response.index("\r\n\r\n")86        http_headers     = self.raw_http_response[header_start_idx + 2:header_end_idx].replace(" ", "")87        return {key_value[0]: key_value[1] for header in http_headers.split("\r\n") if (key_value := header.split(":", 1))}88class HTTPRequest:89    """90    HTTPRequest class to construct and send an HTTP request.91    This HTTPRequest class implements a subset of the HTTP protocol version 1.1 which is defined in RFC2616.92    Constructor (__init__):93        parameter method      [string]: HTTP request method (e.g. GET, POST, etc.)94        parameter url         [string]: Destination URL for the HTTP request95        parameter data    [dictionary]: HTTP request parameters96        parameter json    [dictionary]: HTTP request JSON data (POST request only)97        parameter headers [dictionary]: HTTP request headers98        parameter timeout    [integer]: socket timeout99    """100    def __init__(self, method: str, url: str, data: dict = {}, json: dict = {}, headers: dict = {}, timeout: float = 3.0) -> None:101        self.method  = method102        self.url     = url103        self.data    = data104        self.json    = json105        self.headers = headers106        self.timeout = timeout107        self.json_body: bool = True if self.json else False108        self.socket = socket(AF_INET, SOCK_STREAM)109        self.socket.settimeout(timeout)110        dest_host = self._parse_url.netloc[1]111        dest_port = self._parse_url.netloc[2]112        self.socket.connect((dest_host, int(dest_port)))113    def content_length(self, json: bool = True) -> int:114        """115        Calculate the content length of the HTTP request data.116        returns: content length [integer]117        """118        119        if not json:120            return len(self._parse_http_data())121        return len(self._parse_http_json())122    def _parse_http_headers(self) -> str:123        """124        Parse the HTTP request headers. Include 'Host' and 'Content-Length' header by default.125        returns: formatted HTTP request headers [string]126        """127        128        http_headers: str = ""129        dest_host = self._parse_url.netloc[0]130        http_headers += f"Host: {dest_host}\r\n"131        if self.method == HTTP_POST:132            http_headers += f"Content-Length: {self.content_length(self.json_body)}\r\n"133        for header_key, header_value in self.headers.items():134            http_headers += f"{header_key}: {header_value}\r\n"135        return http_headers136        137    @property138    def _parse_url(self) -> url:139        """140        Parse the destination URL.141        returns: destination URL components [namedtuple]142        """143        144        if not self.url.startswith("http"):145            raise ValueError("Invalid URL")146        delimiter = "/"147        url_parts = self.url.split(delimiter)148        scheme       = url_parts[0].replace(":", "")149        netloc_abs   = url_parts[2]150        netloc_host: str = ""151        netloc_port: str = ""152        if ":" in url_parts[2]:153            netloc_parts = netloc_abs.split(":")154            netloc_host, netloc_port = netloc_parts[0], netloc_parts[1]155        else:156            netloc_host, netloc_port = url_parts[2], 80157        path = ''.join([f"/{path}" for path in url_parts[3:]])158        return url(159            scheme=scheme, 160            netloc=(netloc_abs, netloc_host, netloc_port), 161            path=path162        )163    def _parse_http_json(self) -> str:164        """165        Convert JSON data to string.166        returns: JSON data [string]167        """168        return dumps(self.json)169    def _parse_http_data(self) -> str:170        """171        Parse the HTTP request parameters.172        returns: formatted HTTP request parameters [string]173        """174        request_data: str = ""175        for i, (key, data) in enumerate(self.data.items()):176            fmt = f"{key}={data}"177            if i != len(self.data) - 1: fmt += "&"178            request_data += fmt179        return request_data180        181    def construct_http_request(self) -> str:182        """183        Construct the final HTTP request string.184        returns: constructed HTTP request string [string]        185        """186        parsed_url     = self._parse_url187        http_headers   = self._parse_http_headers()188        body_data = self._parse_http_json()189        if not self.json:190            body_data = self._parse_http_data()191        if self.method == HTTP_GET:192            return f"{self.method} {parsed_url.path}{f'?{body_data}' if self.data and not self.json else ''} {HTTP_VERSION}\r\n{http_headers}\r\n"193        return f"{self.method} {parsed_url.path} {HTTP_VERSION}\r\n{http_headers}\r\n{body_data}"194        195    def send_http_request(self) -> None:196        """197        Construct and send a new HTTP request.198        """199        http_request = str.encode(self.construct_http_request())200        self.socket.send(http_request)201    def get_http_response(self) -> HTTPResponse:202        """203        Receive and parse the HTTP response.204        returns: parsed HTTP response [HTTPResponse]205        """206        207        response_buffer = b""208        while True:209            data = self.socket.recv(1024)210            if not data:211                break212            response_buffer += data213        self.socket.close()214        return HTTPResponse(response_buffer.decode("utf-8"))215def get(url: str, data: dict = {}, headers: dict = {}, timeout: float = 3.0) -> HTTPResponse:216    """217    Wrapper function to make a new HTTP GET request.218    returns: HTTP response [HTTPResponse]219    """220    http_request_instance = HTTPRequest(method="GET", url=url, data=data, headers=headers, timeout=timeout)221    http_request_instance.send_http_request()222    return http_request_instance.get_http_response()223def post(url: str, data: dict = {}, json: dict = {}, headers: dict = {}, timeout: float = 3.0) -> HTTPResponse:224    """225    Wrapper function to make a new HTTP POST request.226    returns: HTTP response [HTTPResponse]227    """228    http_request_instance = HTTPRequest("POST", url=url, data=data, json=json, headers=headers, timeout=timeout)229    http_request_instance.send_http_request()...

pinpoint_censor.py

Source:pinpoint_censor.py

...124    dns_result = process_raw_dns_response(raw_dns_response, is_timeout)125    dns_result['device'] = addr126    return dns_result127############################################# HTTP Part ##########################################128def process_raw_http_response(raw_http_response, is_timeout):129    http_result = dict()130    http_result['timestamp'] = int(time.time())131    http_result['status'] = 'success'132    http_result['status_code'] = 0133    #http_result['url'] = ''134    http_result['text'] = ''135    http_result['headers'] = dict()136    http_result['is_timeout'] = is_timeout137    class FakeSocket():138        def __init__(self, response_bytes):139            self._file = BytesIO(response_bytes)140        def makefile(self, *args, **kwargs):141            return self._file142    if not is_timeout:143        if raw_http_response == b'':144            http_result['status'] = 'fail'145        else:146            source = FakeSocket(raw_http_response)147            response = HTTPResponse(source)148            response.begin()149            http_result['text'] = response.read(len(raw_http_response)).decode()150            #http_result['url'] = raw_http_response.url151            http_result['status_code'] = response.status152            http_result['headers'] = dict(response.getheaders())153    else:154        http_result['status'] = 'fail'155    return http_result156def recvall(sock):157    data = b''158    bufsize = 4096159    while True:160        packet = sock.recv(bufsize)161        data += packet162        if len(packet) < bufsize:163            break164    return data165def http_request(domain, server, ttl, timeout = 5):166    request = "GET / HTTP/1.1\r\nHost: %s\r\nUser-Agent: Mozilla/5.0\r\n\r\n" % domain167    request = request.encode()168    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)169    sock.settimeout(timeout)170    icmp_sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)171    icmp_sock.settimeout(1)172    try:173        # tcp handshake174        sock.connect((server, 80))175        port = sock.getsockname()[1]176        #print('send port', port)177    178        sock.setsockopt(socket.IPPROTO_IP, socket.IP_TTL, struct.pack('I', ttl))179        sock.send(request)180        time.sleep(1)181        182        raw_http_response = recvall(sock)183        is_timeout = False184        addr = get_router_ip(icmp_sock, port)185        186        # close socket187        # sock.shutdown(socket.SHUT_RDWR)188        # sock.close()189    except socket.timeout:190        raw_http_response = b''191        is_timeout = True192        addr = get_router_ip(icmp_sock, port)193    194    except SocketError as e:195        raw_http_response = b''196        is_timeout = False197        addr = '!'198    except Exception as e:199        print(e)200        raw_http_response = b''201        is_timeout = False202        addr = '!'203    204    http_result = process_raw_http_response(raw_http_response, is_timeout)205    http_result['device'] = addr206    return http_result207############################################# SNI Part ##########################################208    209def sni_request(domain, server, ttl, timeout = 5):210    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)211    sock.settimeout(timeout)212    icmp_sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)213    icmp_sock.settimeout(1)214    context = ssl.SSLContext(ssl.PROTOCOL_TLS)215    sni_result = dict()216    sni_result['timestamp'] = int(time.time())217    sni_result['cert'] = ''218    sni_result['cert_serial'] = '0'...

proxy_request.py

Source:proxy_request.py

...93        is_timeout = False94    dns_result = process_raw_dns_response(raw_dns_response, is_timeout)95    return domain, dns_result96############################################# HTTP Part ##########################################97def process_raw_http_response(raw_http_response, is_timeout):98    http_result = dict()99    http_result['timestamp'] = int(time.time())100    http_result['status'] = 'success'101    http_result['status_code'] = 0102    http_result['url'] = ''103    http_result['text'] = ''104    http_result['headers'] = dict()105    http_result['is_timeout'] = is_timeout106    if not is_timeout:107        if raw_http_response == '':108            http_result['status'] = 'fail'109        else:110            http_result['text'] = raw_http_response.text111            http_result['url'] = raw_http_response.url112            http_result['status_code'] = raw_http_response.status_code113            http_result['headers'] = dict(raw_http_response.headers)114    else:115        http_result['status'] = 'fail'116    return http_result117def proxy_http(domain, server, proxy = {}, timeout = 5):118    through_proxy, proxy_address, proxy_port, username, password = unpack_proxy_args(proxy)119    120    if through_proxy:121        proxy = {"http":"socks5h://{}:{}@{}".format(username, password, proxy_address + ':' + str(proxy_port))}122    else:123        proxy = {}124    headers = dict()125    headers['Host'] = domain126    headers['User-Agent'] = 'Mozilla/5.0'127    128    url = 'http://' + server129    try:130        raw_http_response = requests.get(url, proxies=proxy, headers = headers, timeout = timeout)131        is_timeout = False132    133    except requests.exceptions.Timeout:134        raw_http_response = ''135        is_timeout = True136    except:137        raw_http_response = ''138        is_timeout = False139    140    http_result = process_raw_http_response(raw_http_response, is_timeout)141    return domain, http_result142############################################# SNI Part ##########################################143def process_raw_http_response_from_sni(raw_http_response, is_timeout):144    http_result = dict()145    http_result['timestamp'] = int(time.time())146    http_result['status'] = 'success'147    http_result['status_code'] = 0148    #http_result['url'] = ''149    http_result['text'] = ''150    http_result['headers'] = dict()151    http_result['is_timeout'] = is_timeout152    class FakeSocket():153        def __init__(self, response_bytes):154            self._file = BytesIO(response_bytes)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.