open-standards/wai/cache-conformance/mutators.py

#!/usr/bin/env python3
"""Negative controls for driver.py: proxies that break http-delivery §9 in one way each,
so a run against them must fail the tests that rule reaches. If the driver passes one of
them, the driver is what is broken.

    python3 mutators.py --mode flip|gzip|truncate|vary|stored|ranged|reflip|relay --listen PORT --upstream URL

- flip:     changes one byte in the middle of every object body (§9 rule 1);
- gzip:     applies a gzip content coding to whole objects when the request offers it
            (§9 rule 1);
- truncate: drops the last byte of every whole object, and says so in Content-Length
            (§9 rules 1 and 3);
- vary:     adds `Vary: Origin` and drops `Access-Control-Allow-Origin` (§9 rule 4,
            §5.3);
- stored:   answers the first whole GET of a path untouched, and any later one with
            `no-transform` dropped from `Cache-Control`, no `ETag` and another
            `Content-Type`, as a store could (§9 rule 3);
- ranged:   does the same to every `206` (§9 rule 3);
- reflip:   answers the first whole GET of a path untouched, and changes one byte in the
            middle of any later one, as a store could (§9 rule 1);
- relay:    changes nothing and stores nothing, as §9 rule 7 lets a cache relay what it does
            not store: the positive control through a proxy, which must pass every MUST test.

Standard library only.
"""
import gzip as gz, http.client, sys, urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

MODE, UPSTREAM = None, None
SEEN = set()

HOP = {"connection", "keep-alive", "proxy-connection", "transfer-encoding", "upgrade", "te", "trailer", "host"}


def headers_changed(out):
    kept = [(k, v) for k, v in out if k.lower() not in ("cache-control", "etag", "content-type")]
    cc = next((v for k, v in out if k.lower() == "cache-control"), "")
    return kept + [("Cache-Control", cc.replace(", no-transform", "")), ("Content-Type", "application/octet-stream")]


class Proxy(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(self, *a):
        pass

    def forward(self, method):
        u = urllib.parse.urlsplit(UPSTREAM)
        conn = http.client.HTTPConnection(u.hostname, u.port or 80, timeout=30)
        headers = {k: v for k, v in self.headers.items() if k.lower() not in HOP}
        conn.request(method, self.path, headers=headers)
        r = conn.getresponse()
        body = r.read()
        out = [(k, v) for k, v in r.getheaders() if k.lower() not in HOP and k.lower() != "content-length"]
        conn.close()
        is_object = self.path.split("?")[0].endswith(".wai")
        if MODE == "flip" and is_object and len(body) > 1:
            body = bytearray(body)
            body[len(body) // 2] ^= 0x01
            body = bytes(body)
        elif MODE == "gzip" and is_object and r.status == 200 and "gzip" in (self.headers.get("Accept-Encoding") or ""):
            body = gz.compress(body, mtime=0)
            out = [(k, v) for k, v in out if k.lower() != "content-encoding"] + [("Content-Encoding", "gzip")]
        elif MODE == "truncate" and is_object and r.status == 200 and body:
            body = body[:-1]
        elif MODE == "vary":
            out = [(k, v) for k, v in out if k.lower() != "access-control-allow-origin"] + [("Vary", "Origin")]
        elif MODE == "stored" and self.path in SEEN and r.status == 200:
            out = headers_changed(out)
        elif MODE == "ranged" and r.status == 206:
            out = headers_changed(out)
        elif MODE == "reflip" and self.path in SEEN and r.status == 200 and len(body) > 1:
            body = bytearray(body)
            body[len(body) // 2] ^= 0x01
            body = bytes(body)
        SEEN.add(self.path)
        self.send_response(r.status)
        for k, v in out:
            self.send_header(k, v)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Connection", "close")
        self.end_headers()
        if method != "HEAD":
            self.wfile.write(body)

    def do_GET(self):
        self.forward("GET")

    def do_HEAD(self):
        self.forward("HEAD")

    def do_OPTIONS(self):
        self.forward("OPTIONS")


def main(argv):
    global MODE, UPSTREAM
    opt = lambda f: argv[argv.index(f) + 1]
    MODE, UPSTREAM, port = opt("--mode"), opt("--upstream"), int(opt("--listen"))
    if MODE not in ("flip", "gzip", "truncate", "vary", "stored", "ranged", "reflip", "relay"):
        print(__doc__)
        return 2
    ThreadingHTTPServer(("127.0.0.1", port), Proxy).serve_forever()


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))