[wrw.git] / wrw / proto.py

import time, calendar, collections.abc, binascii, base64

statusinfo = {
    400: ("Bad Request", "Invalid HTTP request."),
    401: ("Unauthorized", "Authentication must be provided for the requested resource."),
    403: ("Forbidden", "You are not authorized to request the requested resource."),
    404: ("Not Found", "The requested resource was not found."),
    405: ("Method Not Allowed", "The request method is not recognized or permitted by the requested resource."),
    429: ("Too Many Requests", "Your client is sending more frequent requests than are accepted."),
    500: ("Server Error", "An internal error occurred."),
    501: ("Not Implemented", "The requested functionality has not been implemented."),
    503: ("Service Unavailable", "Service is being denied at this time."),
    }

def httpdate(ts):
    return time.strftime("%a, %d %b %Y %H:%M:%S +0000", time.gmtime(ts))

def phttpdate(dstr):
    tz = dstr[-6:]
    dstr = dstr[:-6]
    if tz[0] != " " or (tz[1] != "+" and tz[1] != "-") or not tz[2:].isdigit():
        return None
    tz = int(tz[1:])
    tz = (((tz / 100) * 60) + (tz % 100)) * 60
    return calendar.timegm(time.strptime(dstr, "%a, %d %b %Y %H:%M:%S")) - tz

def pmimehead(hstr):
    def pws(p):
        while p < len(hstr) and hstr[p].isspace():
            p += 1
        return p
    def token(p, sep):
        buf = ""
        p = pws(p)
        if p >= len(hstr):
            return "", p
        if hstr[p] == '"':
            p += 1
            while p < len(hstr):
                if hstr[p] == '\\':
                    p += 1
                    if p < len(hstr):
                        buf += hstr[p]
                        p += 1
                    else:
                        break
                elif hstr[p] == '"':
                    p += 1
                    break
                else:
                    buf += hstr[p]
                    p += 1
            return buf, pws(p)
        else:
            while p < len(hstr):
                if hstr[p] in sep:
                    break
                buf += hstr[p]
                p += 1
            return buf.strip(), pws(p)
    p = 0
    val, p = token(p, ";")
    pars = {}
    while p < len(hstr):
        if hstr[p] != ';':
            break
        p += 1
        k, p = token(p, "=")
        if k == "" or hstr[p:p + 1] != '=':
            break
        p += 1
        v, p = token(p, ';')
        pars[k.lower()] = v
    return val, pars

def htmlq(html):
    ret = ""
    for c in html:
        if c == "&":
            ret += "&amp;"
        elif c == "<":
            ret += "&lt;"
        elif c == ">":
            ret += "&gt;"
        else:
            ret += c
    return ret

def simpleerror(env, startreq, code, title, msg):
    buf = """<?xml version="1.0" encoding="US-ASCII"?>
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.1//EN" "http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd">
<html xmlns="http://www.w3.org/1999/xhtml" xml:lang="en-US">
<head>
<title>%s</title>
</head>
<body>
<h1>%s</h1>
<p>%s</p>
</body>
</html>
""" % (title, title, htmlq(msg))
    buf = buf.encode("us-ascii")
    startreq("%i %s" % (code, title), [("Content-Type", "text/html"), ("Content-Length", str(len(buf)))])
    return [buf]

def urlq(url):
    if isinstance(url, str):
        url = url.encode("utf-8")
    ret = ""
    invalid = b"%;&=+#?/\"'"
    for c in url:
        if c in invalid or (c <= 32) or (c >= 128):
            ret += "%%%02X" % c
        else:
            ret += chr(c)
    return ret

class urlerror(ValueError):
    pass

def parseurl(url):
    p = url.find("://")
    if p < 0:
        raise urlerror("Protocol not found in absolute URL `%s'" % url)
    proto = url[:p]
    l = url.find("/", p + 3)
    if l < 0:
        raise urlerror("Local part not found in absolute URL `%s'" % url)
    host = url[p + 3:l]
    local = url[l:]
    q = local.find("?")
    if q < 0:
        query = ""
    else:
        query = local[q + 1:]
        local = local[:q]
    return proto, host, local, query

def consurl(proto, host, local, query=""):
    if len(local) < 1 and local[0] != '/':
        raise urlerror("Local part of URL must begin with a slash")
    ret = "%s://%s%s" % (proto, host, local)
    if len(query) > 0:
        ret += "?" + query
    return ret

def appendurl(url, other):
    if "://" in other:
        return other
    proto, host, local, query = parseurl(url)
    if len(other) > 0 and other[0] == '/':
        return consurl(proto, host, other)
    else:
        p = local.rfind('/')
        return consurl(proto, host, local[:p + 1] + other)

def siteurl(req):
    host = req.ihead.get("Host", None)
    if host is None:
        raise Exception("Could not reconstruct URL because no Host header was sent")
    proto = "http"
    if req.https:
        proto = "https"
    return "%s://%s/" % (proto, host)

def scripturl(req):
    s = siteurl(req)
    if req.uriname[0] != '/':
        raise Exception("Malformed local part when reconstructing URL")
    return siteurl(req) + req.uriname[1:]

def requrl(req, qs=True):
    s = siteurl(req)
    if req.uri[0] != '/':
        raise Exception("Malformed local part when reconstructing URL")
    pf = req.uri[1:]
    if not qs:
        p = pf.find('?')
        if not p < 0:
            pf = pf[:p]
    return siteurl(req) + pf

def parstring(pars={}, **augment):
    buf = ""
    for key in pars:
        if key in augment:
            val = augment[key]
            del augment[key]
        else:
            val = pars[key]
        if val is None:
            continue
        if buf != "": buf += "&"
        buf += urlq(key) + "=" + urlq(str(val))
    for key, val in augment.items():
        if val is None:
            continue
        if buf != "": buf += "&"
        buf += urlq(key) + "=" + urlq(str(val))
    return buf

def parurl(url, pars={}, **augment):
    qs = parstring(pars, **augment)
    if qs != "":
        return url + ("&" if "?" in url else "?") + qs
    else:
        return url

# Wrap these, since binascii is a bit funky. :P
def enhex(bs):
    return base64.b16encode(bs).decode("us-ascii")
def unhex(es):
    if not isinstance(es, collections.abc.ByteString):
        try:
            es = es.encode("us-ascii")
        except UnicodeError:
            raise binascii.Error("non-ascii character in hex-string")
    return base64.b16decode(es)
def enb32(bs):
    return base64.b32encode(bs).decode("us-ascii")
def unb32(es):
    if not isinstance(es, collections.abc.ByteString):
        try:
            es = es.encode("us-ascii")
        except UnicodeError:
            raise binascii.Error("non-ascii character in base32-string")
    if (len(es) % 8) != 0:
        es += b"=" * (8 - (len(es) % 8))
    es = es.upper()             # The whole point of Base32 is that it's case-insensitive :P
    return base64.b32decode(es)
def enb64(bs):
    return base64.b64encode(bs).decode("us-ascii")
def unb64(es):
    if not isinstance(es, collections.abc.ByteString):
        try:
            es = es.encode("us-ascii")
        except UnicodeError:
            raise binascii.Error("non-ascii character in base64-string")
    if (len(es) % 4) != 0:
        es += b"=" * (4 - (len(es) % 4))
    return base64.b64decode(es)

def _quoprisafe():
    ret = [False] * 256
    for c in "-!*+/":
        ret[ord(c)] = True
    for c in range(ord('0'), ord('9') + 1):
        ret[c] = True
    for c in range(ord('A'), ord('Z') + 1):
        ret[c] = True
    for c in range(ord('a'), ord('z') + 1):
        ret[c] = True
    return ret
_quoprisafe = _quoprisafe()
def quopri(s, charset="utf-8"):
    bv = s.encode(charset)
    qn = sum(not _quoprisafe[b] for b in bv)
    if qn == 0:
        return s
    if qn > len(bv) / 2:
        return "=?%s?B?%s?=" % (charset, enb64(bv))
    else:
        return "=?%s?Q?%s?=" % (charset, "".join(chr(b) if _quoprisafe[b] else "=%02X" % b for b in bv))

class mimeparam(object):
    def __init__(self, name, val, fallback=None, charset="utf-8", lang=""):
        self.name = name
        self.val = val
        self.fallback = fallback
        self.charset = charset
        self.lang = lang

    def __str__(self):
        self.name.encode("ascii")
        try:
            self.val.encode("ascii")
        except UnicodeError:
            pass
        else:
            return "%s=%s" % (self.name, self.val)
        val = self.val.encode(self.charset)
        self.charset.encode("ascii")
        self.lang.encode("ascii")
        ret = ""
        if self.fallback is not None:
            self.fallback.encode("ascii")
            ret += "%s=%s; " % (self.name, self.fallback)
        ret += "%s*=%s'%s'%s" % (self.name, self.charset, self.lang, urlq(val))
        return ret

class mimeheader(object):
    def __init__(self, name, val, *, mime_charset="utf-8", mime_lang="", **params):
        self.name = name
        self.val = val
        self.params = {}
        self.charset = mime_charset
        self.lang = mime_lang
        for k, v in params.items():
            self[k] = v

    def __getitem__(self, nm):
        return self.params[nm.lower()]

    def __setitem__(self, nm, val):
        if not isinstance(val, mimeparam):
            val = mimeparam(nm, val, charset=self.charset, lang=self.lang)
        self.params[nm.lower()] = val

    def __delitem__(self, nm):
        del self.params[nm.lower()]

    def value(self):
        parts = []
        if self.val != None:
            parts.append(quopri(self.val))
        parts.extend(str(x) for x in self.params.values())
        return("; ".join(parts))

    def __str__(self):
        if self.name is None:
            return self.value()
        return "%s: %s" % (self.name, self.value())
Commit	Line	Data
68c0eaa7	1	import time, calendar, collections.abc, binascii, base64
3ba5e96c	2
5ef9e488	3	statusinfo = {
001ec99e FT	4	400: ("Bad Request", "Invalid HTTP request."),
	5	401: ("Unauthorized", "Authentication must be provided for the requested resource."),
	6	403: ("Forbidden", "You are not authorized to request the requested resource."),
5ef9e488	7	404: ("Not Found", "The requested resource was not found."),
001ec99e	8	405: ("Method Not Allowed", "The request method is not recognized or permitted by the requested resource."),
8ad65294	9	429: ("Too Many Requests", "Your client is sending more frequent requests than are accepted."),
001ec99e FT	10	500: ("Server Error", "An internal error occurred."),
	11	501: ("Not Implemented", "The requested functionality has not been implemented."),
	12	503: ("Service Unavailable", "Service is being denied at this time."),
5ef9e488 FT	13	}
	14
	15	def httpdate(ts):
	16	return time.strftime("%a, %d %b %Y %H:%M:%S +0000", time.gmtime(ts))
	17
	18	def phttpdate(dstr):
	19	tz = dstr[-6:]
	20	dstr = dstr[:-6]
	21	if tz[0] != " " or (tz[1] != "+" and tz[1] != "-") or not tz[2:].isdigit():
	22	return None
	23	tz = int(tz[1:])
	24	tz = (((tz / 100) * 60) + (tz % 100)) * 60
87f2697c	25	return calendar.timegm(time.strptime(dstr, "%a, %d %b %Y %H:%M:%S")) - tz
5ef9e488	26
eed3cf12 FT	27	def pmimehead(hstr):
	28	def pws(p):
	29	while p < len(hstr) and hstr[p].isspace():
	30	p += 1
	31	return p
	32	def token(p, sep):
	33	buf = ""
	34	p = pws(p)
	35	if p >= len(hstr):
	36	return "", p
	37	if hstr[p] == '"':
	38	p += 1
	39	while p < len(hstr):
	40	if hstr[p] == '\\':
	41	p += 1
	42	if p < len(hstr):
	43	buf += hstr[p]
	44	p += 1
	45	else:
	46	break
	47	elif hstr[p] == '"':
	48	p += 1
	49	break
	50	else:
	51	buf += hstr[p]
	52	p += 1
	53	return buf, pws(p)
	54	else:
	55	while p < len(hstr):
	56	if hstr[p] in sep:
	57	break
	58	buf += hstr[p]
	59	p += 1
	60	return buf.strip(), pws(p)
	61	p = 0
	62	val, p = token(p, ";")
	63	pars = {}
	64	while p < len(hstr):
	65	if hstr[p] != ';':
	66	break
	67	p += 1
	68	k, p = token(p, "=")
	69	if k == "" or hstr[p:p + 1] != '=':
	70	break
	71	p += 1
	72	v, p = token(p, ';')
ea18777a	73	pars[k.lower()] = v
eed3cf12 FT	74	return val, pars
eed3cf12 FT	75
5ef9e488 FT	76	def htmlq(html):
	77	ret = ""
	78	for c in html:
	79	if c == "&":
	80	ret += "&"
	81	elif c == "<":
	82	ret += "<"
	83	elif c == ">":
	84	ret += ">"
	85	else:
	86	ret += c
	87	return ret
	88
d740aa68 FT	89	def simpleerror(env, startreq, code, title, msg):
	90	buf = """<?xml version="1.0" encoding="US-ASCII"?>
	91	<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.1//EN" "http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd">
	92	<html xmlns="http://www.w3.org/1999/xhtml" xml:lang="en-US">
	93	<head>
	94	<title>%s</title>
	95	</head>
	96	<body>
	97	<h1>%s</h1>
	98	<p>%s</p>
	99	</body>
	100	</html>
	101	""" % (title, title, htmlq(msg))
cb8bb5ed	102	buf = buf.encode("us-ascii")
d740aa68 FT	103	startreq("%i %s" % (code, title), [("Content-Type", "text/html"), ("Content-Length", str(len(buf)))])
	104	return [buf]
	105
5ef9e488	106	def urlq(url):
11c9628e	107	if isinstance(url, str):
cbb90a8d	108	url = url.encode("utf-8")
5ef9e488	109	ret = ""
f6f55f13	110	invalid = b"%;&=+#?/\"'"
5ef9e488	111	for c in url:
11c9628e FT	112	if c in invalid or (c <= 32) or (c >= 128):
11c9628e FT	113	ret += "%%%02X" % c
5ef9e488	114	else:
11c9628e	115	ret += chr(c)
5ef9e488 FT	116	return ret
5ef9e488 FT	117
bb80acbd FT	118	class urlerror(ValueError):
	119	pass
	120
	121	def parseurl(url):
	122	p = url.find("://")
	123	if p < 0:
	124	raise urlerror("Protocol not found in absolute URL `%s'" % url)
	125	proto = url[:p]
	126	l = url.find("/", p + 3)
	127	if l < 0:
	128	raise urlerror("Local part not found in absolute URL `%s'" % url)
	129	host = url[p + 3:l]
	130	local = url[l:]
	131	q = local.find("?")
	132	if q < 0:
	133	query = ""
	134	else:
	135	query = local[q + 1:]
	136	local = local[:q]
	137	return proto, host, local, query
	138
e21c4382	139	def consurl(proto, host, local, query=""):
bb80acbd FT	140	if len(local) < 1 and local[0] != '/':
	141	raise urlerror("Local part of URL must begin with a slash")
	142	ret = "%s://%s%s" % (proto, host, local)
	143	if len(query) > 0:
	144	ret += "?" + query
	145	return ret
	146
	147	def appendurl(url, other):
	148	if "://" in other:
	149	return other
	150	proto, host, local, query = parseurl(url)
	151	if len(other) > 0 and other[0] == '/':
	152	return consurl(proto, host, other)
	153	else:
	154	p = local.rfind('/')
	155	return consurl(proto, host, local[:p + 1] + other)
	156
31d48e69	157	def siteurl(req):
bb80acbd FT	158	host = req.ihead.get("Host", None)
	159	if host is None:
	160	raise Exception("Could not reconstruct URL because no Host header was sent")
	161	proto = "http"
	162	if req.https:
	163	proto = "https"
31d48e69 FT	164	return "%s://%s/" % (proto, host)
	165
	166	def scripturl(req):
	167	s = siteurl(req)
	168	if req.uriname[0] != '/':
	169	raise Exception("Malformed local part when reconstructing URL")
	170	return siteurl(req) + req.uriname[1:]
	171
51a13716	172	def requrl(req, qs=True):
31d48e69	173	s = siteurl(req)
bb80acbd FT	174	if req.uri[0] != '/':
bb80acbd FT	175	raise Exception("Malformed local part when reconstructing URL")
51a13716 FT	176	pf = req.uri[1:]
	177	if not qs:
	178	p = pf.find('?')
	179	if not p < 0:
	180	pf = pf[:p]
	181	return siteurl(req) + pf
bb80acbd	182
e21c4382	183	def parstring(pars={}, **augment):
5ef9e488 FT	184	buf = ""
	185	for key in pars:
	186	if key in augment:
	187	val = augment[key]
	188	del augment[key]
	189	else:
	190	val = pars[key]
0bf1ac2f FT	191	if val is None:
0bf1ac2f FT	192	continue
5ef9e488 FT	193	if buf != "": buf += "&"
5ef9e488 FT	194	buf += urlq(key) + "=" + urlq(str(val))
0bf1ac2f FT	195	for key, val in augment.items():
	196	if val is None:
	197	continue
5ef9e488	198	if buf != "": buf += "&"
0bf1ac2f	199	buf += urlq(key) + "=" + urlq(str(val))
5ef9e488	200	return buf
08a4b741	201
e21c4382	202	def parurl(url, pars={}, **augment):
08a4b741 FT	203	qs = parstring(pars, **augment)
08a4b741 FT	204	if qs != "":
d6d4c944	205	return url + ("&" if "?" in url else "?") + qs
08a4b741 FT	206	else:
08a4b741 FT	207	return url
c4b97e16 FT	208
	209	# Wrap these, since binascii is a bit funky. :P
	210	def enhex(bs):
	211	return base64.b16encode(bs).decode("us-ascii")
	212	def unhex(es):
68c0eaa7	213	if not isinstance(es, collections.abc.ByteString):
c4b97e16 FT	214	try:
	215	es = es.encode("us-ascii")
	216	except UnicodeError:
	217	raise binascii.Error("non-ascii character in hex-string")
	218	return base64.b16decode(es)
	219	def enb32(bs):
	220	return base64.b32encode(bs).decode("us-ascii")
	221	def unb32(es):
68c0eaa7	222	if not isinstance(es, collections.abc.ByteString):
c4b97e16 FT	223	try:
	224	es = es.encode("us-ascii")
	225	except UnicodeError:
	226	raise binascii.Error("non-ascii character in base32-string")
	227	if (len(es) % 8) != 0:
	228	es += b"=" * (8 - (len(es) % 8))
	229	es = es.upper() # The whole point of Base32 is that it's case-insensitive :P
	230	return base64.b32decode(es)
	231	def enb64(bs):
	232	return base64.b64encode(bs).decode("us-ascii")
	233	def unb64(es):
68c0eaa7	234	if not isinstance(es, collections.abc.ByteString):
c4b97e16 FT	235	try:
	236	es = es.encode("us-ascii")
	237	except UnicodeError:
	238	raise binascii.Error("non-ascii character in base64-string")
	239	if (len(es) % 4) != 0:
	240	es += b"=" * (4 - (len(es) % 4))
	241	return base64.b64decode(es)
c62c64aa FT	242
	243	def _quoprisafe():
	244	ret = [False] * 256
	245	for c in "-!*+/":
	246	ret[ord(c)] = True
	247	for c in range(ord('0'), ord('9') + 1):
	248	ret[c] = True
	249	for c in range(ord('A'), ord('Z') + 1):
	250	ret[c] = True
	251	for c in range(ord('a'), ord('z') + 1):
	252	ret[c] = True
	253	return ret
	254	_quoprisafe = _quoprisafe()
	255	def quopri(s, charset="utf-8"):
	256	bv = s.encode(charset)
	257	qn = sum(not _quoprisafe[b] for b in bv)
	258	if qn == 0:
	259	return s
	260	if qn > len(bv) / 2:
	261	return "=?%s?B?%s?=" % (charset, enb64(bv))
	262	else:
	263	return "=?%s?Q?%s?=" % (charset, "".join(chr(b) if _quoprisafe[b] else "=%02X" % b for b in bv))
	264
	265	class mimeparam(object):
	266	def __init__(self, name, val, fallback=None, charset="utf-8", lang=""):
	267	self.name = name
	268	self.val = val
	269	self.fallback = fallback
	270	self.charset = charset
	271	self.lang = lang
	272
	273	def __str__(self):
	274	self.name.encode("ascii")
	275	try:
	276	self.val.encode("ascii")
	277	except UnicodeError:
	278	pass
	279	else:
	280	return "%s=%s" % (self.name, self.val)
	281	val = self.val.encode(self.charset)
	282	self.charset.encode("ascii")
	283	self.lang.encode("ascii")
	284	ret = ""
	285	if self.fallback is not None:
	286	self.fallback.encode("ascii")
	287	ret += "%s=%s; " % (self.name, self.fallback)
	288	ret += "%s*=%s'%s'%s" % (self.name, self.charset, self.lang, urlq(val))
	289	return ret
	290
	291	class mimeheader(object):
	292	def __init__(self, name, val, , mime_charset="utf-8", mime_lang="", *params):
	293	self.name = name
	294	self.val = val
	295	self.params = {}
	296	self.charset = mime_charset
	297	self.lang = mime_lang
	298	for k, v in params.items():
	299	self[k] = v
	300
	301	def __getitem__(self, nm):
	302	return self.params[nm.lower()]
	303
	304	def __setitem__(self, nm, val):
	305	if not isinstance(val, mimeparam):
306	val = mimeparam(nm, val, charset=self.charset, lang=self.lang)
307	self.params[nm.lower()] = val
308
309	def __delitem__(self, nm):
310	del self.params[nm.lower()]
311
312	def value(self):
313	parts = []
314	if self.val != None:
315	parts.append(quopri(self.val))
316	parts.extend(str(x) for x in self.params.values())
317	return("; ".join(parts))
318
319	def __str__(self):
320	if self.name is None:
321	return self.value()
322	return "%s: %s" % (self.name, self.value())