darkhttpd/devel/test.py

#!/usr/bin/env python3
# This is run by the "run-tests" script.
import unittest
import socket
import signal
import re
import os
import random
import time

WWWROOT = "tmp.httpd.tests"

def random_bytes(n):
    return bytes([random.randint(0,255) for _ in range(n)])

def between(s, start, end):
    assert start in s, s
    p = s.index(start) + len(start)
    s = s[p:]
    assert end in s, s
    p = s.index(end)
    return s[:p]

assert between("hello world", "hell", "world") == "o "

class Conn:
    def __init__(self):
        self.port = 12346
        self.s = socket.socket()
        self.s.connect(("0.0.0.0", self.port))
        # connect throws socket.error on connection refused

    def close(self):
        self.s.close()

    def get(self, url, http_ver="1.0", endl="\n", req_hdrs={}, method="GET"):
        req = method+" "+url
        if http_ver is not None:
            req += " HTTP/"+http_ver
        req += endl
        if http_ver is not None:
            req_hdrs["User-Agent"] = "test.py"
            req_hdrs["Connection"] = "close"
            for k,v in req_hdrs.items():
                req += k+": "+v+endl
        req += endl # end of request
        self.s.send(req.encode('utf_8'))
        ret = b''
        while True:
            signal.alarm(1) # don't wait forever
            r = self.s.recv(65536)
            signal.alarm(0)
            if r == b'':
                break
            else:
                ret += r
        return ret

    def get_keepalive(self, url, endl="\n", req_hdrs={}, method="GET"):
        req = method+" "+url+" HTTP/1.1"+endl
        req_hdrs["User-Agent"] = "test.py"
        req_hdrs["Connection"] = "keep-alive"
        for k,v in req_hdrs.items():
            req += k+": "+v+endl
        req += endl # end of request
        self.s.send(req.encode('utf-8'))
        signal.alarm(1) # don't wait forever
        ret = b''
        while True:
            ret += self.s.recv(65536)
            if b'\r\n\r\n' not in ret:
                # Don't have headers yet.
                continue
            if method == "HEAD":
                # We're done.
                break
            if b'Content-Length: ' in ret:
                cl = between(ret, b'Content-Length: ', b'\r\n')
                cl = int(cl)
            else:
                cl = 0
            p = ret.index(b'\r\n\r\n') + 4
            assert len(ret) - p <= cl, [ret, p, cl]
            if len(ret) == p + cl:
                # Complete response.
                break
        signal.alarm(0)
        return ret

def parse(resp):
    """
    Parse response into status line, headers and body.
    """
    pos = resp.find(b'\r\n\r\n')
    assert pos != -1, 'response is %s' % repr(resp)
    head = resp[:pos]
    body = resp[pos+4:]
    status,head = head.split(b'\r\n', 1)
    hdrs = {}
    for line in head.split(b'\r\n'):
        k, v = line.split(b': ', 1)
        k = k.decode('utf-8')
        v = v.decode('utf-8')
        hdrs[k] = v
    return (status, hdrs, body)

class TestHelper(unittest.TestCase):
    def get(self, url, http_ver="1.0", endl="\n", req_hdrs={}, method="GET"):
        c = Conn()
        r = c.get(url, http_ver, endl, req_hdrs, method)
        c.close()
        return r

    def assertContains(self, body, *strings):
        if type(body) is not bytes:
            body = body.encode('utf-8')
        for s in strings:
            self.assertTrue(s.encode('utf-8') in body,
                    msg="\nExpected: %s\nIn response: %s" % (
                        repr(s), repr(body)))

    def assertIsIndex(self, body, path):
        self.assertContains(body,
            '<a href="../">..</a>/',
            'Generated by darkhttpd')

    def assertIsInvalid(self, body, path):
        self.assertContains(body,
            "<title>400 Bad Request</title>",
            "<h1>Bad Request</h1>\n",
            "You requested an invalid URL.\n",
            'Generated by darkhttpd')

    def assertNotFound(self, body, path):
        self.assertContains(body,
            "<title>404 Not Found</title>",
            "<h1>Not Found</h1>\n",
            "The URL you requested was not found.\n",
            'Generated by darkhttpd')

    def assertForbidden(self, body, path):
        self.assertContains(body,
            "<title>403 Forbidden</title>",
            "<h1>Forbidden</h1>\n",
            "You don't have permission to access this URL.\n",
            'Generated by darkhttpd')

    def assertUnreadable(self, body, path):
        self.assertContains(body,
            "Couldn't list directory: Permission denied\n",
            'Generated by darkhttpd')

    def drive_range(self, range_in, range_out, len_out, data_out,
            status_out = "206 Partial Content"):
        resp = self.get(self.url, req_hdrs = {"Range": "bytes="+range_in})
        status, hdrs, body = parse(resp)
        self.assertContains(status, status_out)
        self.assertEqual(hdrs["Accept-Ranges"], "bytes")
        self.assertEqual(hdrs["Content-Range"], "bytes "+range_out)
        self.assertEqual(hdrs["Content-Length"], str(len_out))
        self.assertEqual(body, data_out)

class TestDirList(TestHelper):
    def setUp(self):
        self.fn = WWWROOT+"/escape(this)name"
        with open(self.fn, "w") as f:
            f.write("x"*12345)

    def tearDown(self):
        os.unlink(self.fn)

    def test_dirlist_escape(self):
        resp = self.get("/")
        status, hdrs, body = parse(resp)
        self.assertEqual(ord("#"), 0x23)
        self.assertContains(body, "escape%28this%29name", "12345")

    def test_dotdot_first(self):
        resp = self.get("/")
        status, hdrs, body = parse(resp)
        self.assertLess(b'(', b'..')  # ( sorts before ..
        paren_pos = body.index(b'<a href="%28/">(</a>/')
        dotdot_pos = body.index(b'<a href="../">..</a>/')
        # But we special-case .. to come first
        self.assertLess(dotdot_pos, paren_pos)

class TestCases(TestHelper):
    pass # these get autogenerated in setUpModule()

def nerf(s):
    return re.sub("[^a-zA-Z0-9]", "_", s)

def makeCase(name, url, hdr_checker=None, body_checker=None,
             req_hdrs={"User-Agent": "test.py"},
             http_ver=None, endl="\n"):
    def do_test(self):
        resp = self.get(url, http_ver, endl, req_hdrs)
        if http_ver is None:
            status = ""
            hdrs = {}
            body = resp
        else:
            status, hdrs, body = parse(resp)

        if hdr_checker is not None and http_ver is not None:
            hdr_checker(self, hdrs)

        if body_checker is not None:
            body_checker(self, body)

        # FIXME: check status
        if http_ver is not None:
            prefix = b'HTTP/1.1 ' # should 1.0 stay 1.0?
            self.assertTrue(status.startswith(prefix),
                msg="%s at start of %s"%(repr(prefix), repr(status)))

    v = http_ver
    if v is None:
        v = "0.9"
    test_name = "_".join([
        "test",
        nerf(name),
        nerf("HTTP"+v),
        {"\n":"LF", "\r\n":"CRLF"}[endl],
    ])
    do_test.__name__ = test_name # hax
    setattr(TestCases, test_name, do_test)

def makeCases(name, url, hdr_checker=None, body_checker=None,
              req_hdrs={"User-Agent": "test.py"}):
    for http_ver in [None, "1.0", "1.1"]:
        for endl in ["\n", "\r\n"]:
            makeCase(name, url, hdr_checker, body_checker,
                     req_hdrs, http_ver, endl)

def makeSimpleCases(name, url, assert_name):
    makeCases(name, url, None,
        lambda self,body: getattr(self, assert_name)(body, url))

def setUpModule():
    for args in [
        ["index",                "/",               "assertIsIndex"],
        ["up dir",               "/dir/../",        "assertIsIndex"],
        ["extra slashes",        "//dir///..////",  "assertIsIndex"],
        ["no trailing slash",    "/dir/..",         "assertIsIndex"],
        ["no leading slash",     "dir/../",         "assertIsInvalid"],
        ["invalid up dir",       "/../",            "assertIsInvalid"],
        ["fancy invalid up dir", "/./dir/./../../", "assertIsInvalid"],
        ["extra slashes 2",      "//.d",            "assertNotFound"],
        ["not found",            "/not_found.txt",  "assertNotFound"],
        ["forbidden",            "/forbidden/x",    "assertForbidden"],
        ["unreadable",           "/unreadable/",    "assertUnreadable"],
        ]:
        makeSimpleCases(*args)

class TestDirRedirect(TestHelper):
    def setUp(self):
        self.url = "/mydir"
        self.fn = WWWROOT + self.url
        os.mkdir(self.fn)

    def tearDown(self):
        os.rmdir(self.fn)

    def test_dir_redirect(self):
        resp = self.get(self.url)
        status, hdrs, body = parse(resp)
        self.assertContains(status, "301 Moved Permanently")
        self.assertEqual(hdrs["Location"], self.url+"/") # trailing slash

class TestFileGet(TestHelper):
    def setUp(self):
        self.datalen = 2345
        self.data = random_bytes(self.datalen)
        self.url = '/data.jpeg'
        self.fn = WWWROOT + self.url
        with open(self.fn, 'wb') as f:
            f.write(self.data)
        self.qurl = '/what%3f.jpg'
        self.qfn = WWWROOT + '/what?.jpg'
        if os.path.exists(self.qfn):
            os.unlink(self.qfn)
        os.link(self.fn, self.qfn)

    def tearDown(self):
        os.unlink(self.fn)
        os.unlink(self.qfn)

    def get_helper(self, url):
        resp = self.get(url)
        status, hdrs, body = parse(resp)
        self.assertContains(status, "200 OK")
        self.assertEqual(hdrs["Accept-Ranges"], "bytes")
        self.assertEqual(hdrs["Content-Length"], str(self.datalen))
        self.assertEqual(hdrs["Content-Type"], "image/jpeg")
        self.assertContains(hdrs["Server"], "darkhttpd/")
        assert body == self.data, [url, resp, status, hdrs, body]
        self.assertEqual(body, self.data)

    def test_file_get(self):
        self.get_helper(self.url)

    def test_file_get_urldecode(self):
        self.get_helper(''.join(['%%%02x' % ord(x) for x in self.url]))

    def test_file_get_redundant_dots(self):
        self.get_helper("/././." + self.url)

    def test_file_get_with_empty_query(self):
        self.get_helper(self.url + "?")

    def test_file_get_with_query(self):
        self.get_helper(self.url + "?action=Submit")

    def test_file_get_esc_question(self):
        self.get_helper(self.qurl)

    def test_file_get_esc_question_with_query(self):
        self.get_helper(self.qurl + '?hello=world')

    def test_file_head(self):
        resp = self.get(self.url, method="HEAD")
        status, hdrs, body = parse(resp)
        self.assertContains(status, "200 OK")
        self.assertEqual(hdrs["Accept-Ranges"], "bytes")
        self.assertEqual(hdrs["Content-Length"], str(self.datalen))
        self.assertEqual(hdrs["Content-Type"], "image/jpeg")

    def test_if_modified_since(self):
        resp1 = self.get(self.url, method="HEAD")
        status, hdrs, body = parse(resp1)
        lastmod = hdrs["Last-Modified"]

        resp2 = self.get(self.url, method="GET", req_hdrs =
            {"If-Modified-Since": lastmod })
        status, hdrs, body = parse(resp2)
        self.assertContains(status, "304 Not Modified")
        self.assertEqual(hdrs["Accept-Ranges"], "bytes")
        self.assertFalse("Last-Modified" in hdrs)
        self.assertFalse("Content-Length" in hdrs)
        self.assertFalse("Content-Type" in hdrs)

    def test_range_single(self):
        self.drive_range("5-5", "5-5/%d" % self.datalen,
                1, self.data[5:6])

    def test_range_single_first(self):
        self.drive_range("0-0", "0-0/%d" % self.datalen,
                1, self.data[0:1])

    def test_range_single_last(self):
        self.drive_range("%d-%d"%(self.datalen-1, self.datalen-1),
        "%d-%d/%d"%(self.datalen-1, self.datalen-1, self.datalen),
        1, self.data[-1:])

    def test_range_single_bad(self):
        resp = self.get(self.url, req_hdrs = {"Range":
            "bytes=%d-%d"%(self.datalen, self.datalen)})
        status, hdrs, body = parse(resp)
        self.assertContains(status, "416 Requested Range Not Satisfiable")

    def test_range_reasonable(self):
        self.drive_range("10-20", "10-20/%d" % self.datalen,
            20-10+1, self.data[10:20+1])

    def test_range_start_given(self):
        self.drive_range("10-", "10-%d/%d" % (self.datalen-1, self.datalen),
            self.datalen-10, self.data[10:])

    def test_range_end_given(self):
        self.drive_range("-25",
            "%d-%d/%d"%(self.datalen-25, self.datalen-1, self.datalen),
            25, self.data[-25:])

    def test_range_beyond_end(self):
        # expecting same result as test_range_end_given
        self.drive_range("%d-%d"%(self.datalen-25, self.datalen*2),
            "%d-%d/%d"%(self.datalen-25, self.datalen-1, self.datalen),
            25, self.data[-25:])

    def test_range_end_given_oversize(self):
        # expecting full file
        self.drive_range("-%d"%(self.datalen*3),
            "0-%d/%d"%(self.datalen-1, self.datalen),
            self.datalen, self.data)

    def test_range_bad_start(self):
        resp = self.get(self.url, req_hdrs = {"Range": "bytes=%d-"%(
            self.datalen*2)})
        status, hdrs, body = parse(resp)
        self.assertContains(status, "416 Requested Range Not Satisfiable")

    def test_range_backwards(self):
        resp = self.get(self.url, req_hdrs = {"Range": "bytes=20-10"})
        status, hdrs, body = parse(resp)
        self.assertContains(status, "416 Requested Range Not Satisfiable")

    def test_lowercase_header(self):
        resp = self.get(self.url, req_hdrs = {"range": "bytes=20-10"})
        status, hdrs, body = parse(resp)
        self.assertContains(status, "416 Requested Range Not Satisfiable")

class TestKeepAlive(TestFileGet):
    """
    Run all of TestFileGet but with a single long-lived connection.
    """
    def setUp(self):
        TestFileGet.setUp(self)
        self.conn = Conn()

    def tearDown(self):
        self.conn.close()

    def get(self, url, endl="\n", req_hdrs={}, method="GET"):
        return self.conn.get_keepalive(url, endl, req_hdrs, method)

def make_large_file(fn, boundary, data):
    with open(fn, 'wb') as f:
        pos = boundary - (len(data) // 2)
        f.seek(pos)
        assert f.tell() == pos
        assert f.tell() < boundary
        f.write(data)
        filesize = f.tell()
        assert filesize == pos + len(data), (filesize, pos, len(data))
        assert filesize > boundary
    return (pos, filesize)

class TestLargeFile2G(TestHelper):
    BOUNDARY = 1<<31

    def setUp(self):
        self.datalen = 4096
        self.data = random_bytes(self.datalen)
        self.url = "/big.jpeg"
        self.fn = WWWROOT + self.url
        self.filepos, self.filesize = make_large_file(
            self.fn, self.BOUNDARY, self.data)

    def tearDown(self):
        os.unlink(self.fn)

    def drive_start(self, ofs):
        req_start = self.BOUNDARY + ofs
        req_end = req_start + self.datalen//4 - 1
        range_in = "%d-%d"%(req_start, req_end)
        range_out = "%s/%d"%(range_in, self.filesize)

        data_start = req_start - self.filepos
        data_end = data_start + self.datalen//4

        self.drive_range(range_in, range_out, self.datalen//4,
            self.data[data_start:data_end])

    def test_largefile_head(self):
        resp = self.get(self.url, method="HEAD")
        status, hdrs, body = parse(resp)
        self.assertContains(status, "200 OK")
        self.assertEqual(hdrs["Accept-Ranges"], "bytes")
        self.assertEqual(hdrs["Content-Length"], str(self.filesize))
        self.assertEqual(hdrs["Content-Type"], "image/jpeg")

    def test_largefile__3(self): self.drive_start(-3)
    def test_largefile__2(self): self.drive_start(-2)
    def test_largefile__1(self): self.drive_start(-1)
    def test_largefile_0(self): self.drive_start(0)
    def test_largefile_1(self): self.drive_start(1)
    def test_largefile_2(self): self.drive_start(2)
    def test_largefile_3(self): self.drive_start(3)

class TestLargeFile4G(TestLargeFile2G):
    BOUNDARY = 1<<32

class TestLargeMtime(TestHelper):
    def setUp(self):
        self.url = '/large_mtime'
        self.fn = WWWROOT + self.url
        with open(self.fn, 'wb') as f:
            f.write(b'x')
        # A timestamp in the year 10,000
        t = int(time.mktime((10000,3,14,1,2,3,0,0,-1)))
        os.utime(self.fn, (t, t))

    def tearDown(self):
        os.unlink(self.fn)

    def test_file_get(self):
        resp = self.get(self.url)
        status, hdrs, body = parse(resp)
        self.assertContains(status, "200 OK")

if __name__ == '__main__':
    setUpModule()
    unittest.main()

# vim:set ts=4 sw=4 et: