Gecko [ www.smanegeri1koba.sch.id ]

Name	Size	Permission
certs	[ DIR ]	drwxr-xr-x
__init__.py	0 B	-rw-r--r--
app.py	12.76 KB	-rw-r--r--
asgi_proxy.py	3.49 KB	-rw-r--r--
hypercornserver.py	2.21 KB	-rw-r--r--
socketserver.py	5.52 KB	-rw-r--r--
testcase.py	12.93 KB	-rw-r--r--

Code Editor : app.py

from __future__ import annotations

import collections
import contextlib
import datetime
import email.utils
import gzip
import mimetypes
import zlib
from io import BytesIO
from pathlib import Path
from typing import Iterator

import trio
from quart import Response, make_response, request
from quart.typing import ResponseReturnValue
from quart_trio import QuartTrio

hypercorn_app = QuartTrio(__name__)

# Globals are not safe in Flask/Quart but work for our Hypercorn use case
RETRY_TEST_NAMES: collections.Counter[str] = collections.Counter()
LAST_RETRY_AFTER_REQ: datetime.datetime = datetime.datetime.min

pyodide_testing_app = QuartTrio(__name__)
DEFAULT_HEADERS = [
    # Allow cross-origin requests for emscripten
    ("Access-Control-Allow-Origin", "*"),
    ("Cross-Origin-Opener-Policy", "same-origin"),
    ("Cross-Origin-Embedder-Policy", "require-corp"),
    ("Feature-Policy", "sync-xhr *;"),
    ("Access-Control-Allow-Headers", "*"),
]

@hypercorn_app.route("/")
@pyodide_testing_app.route("/")
@pyodide_testing_app.route("/index")
async def index() -> ResponseReturnValue:
    return await make_response("Dummy server!")

@hypercorn_app.route("/alpn_protocol")
async def alpn_protocol() -> ResponseReturnValue:
    """Return the requester's certificate."""
    alpn_protocol = request.scope["extensions"]["tls"]["alpn_protocol"]
    return await make_response(alpn_protocol)

@hypercorn_app.route("/certificate")
async def certificate() -> ResponseReturnValue:
    """Return the requester's certificate."""
    print("scope", request.scope)
    subject = request.scope["extensions"]["tls"]["client_cert_name"]
    subject_as_dict = dict(part.split("=") for part in subject.split(", "))
    return await make_response(subject_as_dict)

@hypercorn_app.route("/specific_method", methods=["GET", "POST", "PUT"])
@pyodide_testing_app.route("/specific_method", methods=["GET", "POST", "PUT"])
async def specific_method() -> ResponseReturnValue:
    "Confirm that the request matches the desired method type"
    method_param = (await request.values).get("method", "")

if request.method.upper() == method_param.upper():
        return await make_response("", 200)
    else:
        return await make_response(
            f"Wrong method: {method_param} != {request.method}", 400
        )

@hypercorn_app.route("/upload", methods=["POST"])
async def upload() -> ResponseReturnValue:
    "Confirm that the uploaded file conforms to specification"
    params = await request.form
    param = params.get("upload_param")
    filename_param = params.get("upload_filename")
    size = int(params.get("upload_size", "0"))
    files_ = (await request.files).getlist(param)
    assert files_ is not None

if len(files_) != 1:
        return await make_response(
            f"Expected 1 file for '{param}', not {len(files_)}", 400
        )

file_ = files_[0]
    # data is short enough to read synchronously without blocking the event loop
    with contextlib.closing(file_.stream) as stream:
        data = stream.read()

if int(size) != len(data):
        return await make_response(f"Wrong size: {int(size)} != {len(data)}", 400)

if filename_param != file_.filename:
        return await make_response(
            f"Wrong filename: {filename_param} != {file_.filename}", 400
        )

return await make_response()

@hypercorn_app.route("/chunked")
async def chunked() -> ResponseReturnValue:
    def generate() -> Iterator[str]:
        for _ in range(4):
            yield "123"

return await make_response(generate())

@hypercorn_app.route("/chunked_gzip")
async def chunked_gzip() -> ResponseReturnValue:
    def generate() -> Iterator[bytes]:
        compressor = zlib.compressobj(6, zlib.DEFLATED, 16 + zlib.MAX_WBITS)

for uncompressed in [b"123"] * 4:
            yield compressor.compress(uncompressed)
        yield compressor.flush()

return await make_response(generate(), 200, [("Content-Encoding", "gzip")])

@hypercorn_app.route("/keepalive")
async def keepalive() -> ResponseReturnValue:
    if request.args.get("close", b"0") == b"1":
        headers = [("Connection", "close")]
        return await make_response("Closing", 200, headers)

headers = [("Connection", "keep-alive")]
    return await make_response("Keeping alive", 200, headers)

@hypercorn_app.route("/echo", methods=["GET", "POST", "PUT"])
async def echo() -> ResponseReturnValue:
    "Echo back the params"
    if request.method == "GET":
        return await make_response(request.query_string)

return await make_response(await request.get_data())

@hypercorn_app.route("/echo_json", methods=["POST"])
@pyodide_testing_app.route("/echo_json", methods=["POST", "OPTIONS"])
async def echo_json() -> ResponseReturnValue:
    "Echo back the JSON"
    if request.method == "OPTIONS":
        return await make_response("", 200)
    data = await request.get_data()
    return await make_response(data, 200, request.headers)

@hypercorn_app.route("/echo_uri/<path:rest>")
@hypercorn_app.route("/echo_uri", defaults={"rest": ""})
async def echo_uri(rest: str) -> ResponseReturnValue:
    "Echo back the requested URI"
    assert request.full_path is not None
    return await make_response(request.full_path)

@hypercorn_app.route("/echo_params")
async def echo_params() -> ResponseReturnValue:
    "Echo back the query parameters"
    await request.get_data()
    echod = sorted((k, v) for k, v in request.args.items())
    return await make_response(repr(echod))

@hypercorn_app.route("/headers", methods=["GET", "POST"])
async def headers() -> ResponseReturnValue:
    return await make_response(dict(request.headers.items()))

@hypercorn_app.route("/headers_and_params")
async def headers_and_params() -> ResponseReturnValue:
    return await make_response(
        {
            "headers": dict(request.headers),
            "params": request.args,
        }
    )

@hypercorn_app.route("/multi_headers", methods=["GET", "POST"])
async def multi_headers() -> ResponseReturnValue:
    return await make_response({"headers": list(request.headers)})

@hypercorn_app.route("/multi_redirect")
async def multi_redirect() -> ResponseReturnValue:
    "Performs a redirect chain based on ``redirect_codes``"
    params = request.args
    codes = params.get("redirect_codes", "200")
    head, tail = codes.split(",", 1) if "," in codes else (codes, None)
    assert head is not None
    status = head
    if not tail:
        return await make_response("Done redirecting", status)

headers = [("Location", f"/multi_redirect?redirect_codes={tail}")]
    return await make_response("", status, headers)

@hypercorn_app.route("/encodingrequest")
async def encodingrequest() -> ResponseReturnValue:
    "Check for UA accepting gzip/deflate encoding"
    data = b"hello, world!"
    encoding = request.headers.get("Accept-Encoding", "")
    headers = []
    if encoding == "gzip":
        headers = [("Content-Encoding", "gzip")]
        file_ = BytesIO()
        with contextlib.closing(gzip.GzipFile("", mode="w", fileobj=file_)) as zipfile:
            zipfile.write(data)
        data = file_.getvalue()
    elif encoding == "deflate":
        headers = [("Content-Encoding", "deflate")]
        data = zlib.compress(data)
    elif encoding == "garbage-gzip":
        headers = [("Content-Encoding", "gzip")]
        data = b"garbage"
    elif encoding == "garbage-deflate":
        headers = [("Content-Encoding", "deflate")]
        data = b"garbage"
    return await make_response(data, 200, headers)

@hypercorn_app.route("/redirect", methods=["GET", "POST", "PUT"])
async def redirect() -> ResponseReturnValue:
    "Perform a redirect to ``target``"
    values = await request.values
    target = values.get("target", "/")
    status = values.get("status", "303 See Other")
    status_code = status.split(" ")[0]

headers = [("Location", target)]
    return await make_response("", status_code, headers)

@hypercorn_app.route("/redirect_after")
async def redirect_after() -> ResponseReturnValue:
    "Perform a redirect to ``target``"
    params = request.args
    date = params.get("date")
    if date:
        dt = datetime.datetime.fromtimestamp(float(date), tz=datetime.timezone.utc)
        http_dt = email.utils.format_datetime(dt, usegmt=True)
        retry_after = str(http_dt)
    else:
        retry_after = "1"
    target = params.get("target", "/")
    headers = [("Location", target), ("Retry-After", retry_after)]
    return await make_response("", 303, headers)

@hypercorn_app.route("/retry_after")
async def retry_after() -> ResponseReturnValue:
    global LAST_RETRY_AFTER_REQ
    params = request.args
    if datetime.datetime.now() - LAST_RETRY_AFTER_REQ < datetime.timedelta(seconds=1):
        status = params.get("status", "429 Too Many Requests")
        status_code = status.split(" ")[0]

return await make_response("", status_code, [("Retry-After", "1")])

LAST_RETRY_AFTER_REQ = datetime.datetime.now()
    return await make_response("", 200)

@hypercorn_app.route("/status")
@pyodide_testing_app.route("/status")
async def status() -> ResponseReturnValue:
    values = await request.values
    status = values.get("status", "200 OK")
    status_code = status.split(" ")[0]
    return await make_response("", status_code)

@hypercorn_app.route("/source_address")
async def source_address() -> ResponseReturnValue:
    """Return the requester's IP address."""
    return await make_response(request.remote_addr)

@hypercorn_app.route("/successful_retry", methods=["GET", "PUT"])
async def successful_retry() -> ResponseReturnValue:
    """First return an error and then success

It's not currently very flexible as the number of retries is hard-coded.
    """
    test_name = request.headers.get("test-name", None)
    if not test_name:
        return await make_response("test-name header not set", 400)

RETRY_TEST_NAMES[test_name] += 1

if RETRY_TEST_NAMES[test_name] >= 2:
        return await make_response("Retry successful!", 200)
    else:
        return await make_response("need to keep retrying!", 418)

@pyodide_testing_app.after_request
def apply_caching(response: Response) -> ResponseReturnValue:
    for header, value in DEFAULT_HEADERS:
        response.headers[header] = value
    return response

@pyodide_testing_app.route("/slow")
async def slow() -> ResponseReturnValue:
    await trio.sleep(10)
    return await make_response("TEN SECONDS LATER", 200)

@pyodide_testing_app.route("/bigfile")
async def bigfile() -> ResponseReturnValue:
    # great big text file, should force streaming
    # if supported
    bigdata = 1048576 * b"WOOO YAY BOOYAKAH"
    return await make_response(bigdata, 200)

@pyodide_testing_app.route("/mediumfile")
async def mediumfile() -> ResponseReturnValue:
    # quite big file
    bigdata = 1024 * b"WOOO YAY BOOYAKAH"
    return await make_response(bigdata, 200)

@pyodide_testing_app.route("/upload", methods=["POST", "OPTIONS"])
async def pyodide_upload() -> ResponseReturnValue:
    if request.method == "OPTIONS":
        return await make_response("", 200)
    spare_data = await request.get_data(parse_form_data=True)
    if len(spare_data) != 0:
        return await make_response("Bad upload data", 404)
    files = await request.files
    form = await request.form
    if form["upload_param"] != "filefield" or form["upload_filename"] != "lolcat.txt":
        return await make_response("Bad upload form values", 404)
    if len(files) != 1 or files.get("filefield") is None:
        return await make_response("Missing file in form", 404)
    file = files["filefield"]
    if file.filename != "lolcat.txt":
        return await make_response(f"File name incorrect {file.name}", 404)
    with contextlib.closing(file):
        data = file.read().decode("utf-8")
    if data != "I'm in ur multipart form-data, hazing a cheezburgr":
        return await make_response(f"File data incorrect {data}", 200)
    return await make_response("Uploaded file correct", 200)

@pyodide_testing_app.route("/pyodide/<py_file>")
async def pyodide(py_file: str) -> ResponseReturnValue:
    file_path = Path(pyodide_testing_app.config["pyodide_dist_dir"], py_file)
    if file_path.exists():
        mime_type, encoding = mimetypes.guess_type(file_path)
        if not mime_type:
            mime_type = "text/plain"
        return await make_response(
            file_path.read_bytes(), 200, [("Content-Type", mime_type)]
        )
    else:
        return await make_response("", 404)

@pyodide_testing_app.route("/wheel/dist.whl")
async def wheel() -> ResponseReturnValue:
    # serve our wheel
    wheel_folder = Path(__file__).parent.parent / "dist"
    wheels = list(wheel_folder.glob("*.whl"))
    if len(wheels) > 0:
        wheel = wheels[0]
        headers = [("Content-Disposition", f"inline; filename='{wheel.name}'")]
        resp = await make_response(wheel.read_bytes(), 200, headers)
        return resp
    else:
        return await make_response(f"NO WHEEL IN {wheel_folder}", 404)

${this.title}

Code Editor : app.py