wrw/form.py

   1 import urllib.parse
   2 from . import proto
   3
   4 __all__ = ["formdata"]
   5
   6 def formparse(req):
   7     buf = {}
   8     buf.update(urllib.parse.parse_qsl(req.query))
   9     if req.ihead.get("Content-Type") == "application/x-www-form-urlencoded":
  10         if req.input.limit > 2 ** 20:
  11             raise ValueError("x-www-form-urlencoded data is absurdly long")
  12         rbody = req.input.read()
  13         buf.update(urllib.parse.parse_qsl(rbody.decode("latin1")))
  14     return buf
  15
  16 class badmultipart(Exception):
  17     pass
  18
  19 class formpart(object):
  20     def __init__(self, form):
  21         self.form = form
  22         self.buf = b""
  23         self.eof = False
  24         self.head = {}
  25
  26     def parsehead(self):
  27         pass
  28
  29     def fillbuf(self, sz):
  30         req = self.form.req
  31         mboundary = b"\r\n--" + self.form.boundary + b"\r\n"
  32         lboundary = b"\r\n--" + self.form.boundary + b"--\r\n"
  33         while not self.eof:
  34             p = self.form.buf.find(mboundary)
  35             if p >= 0:
  36                 self.buf += self.form.buf[:p]
  37                 self.form.buf = self.form.buf[p + len(mboundary):]
  38                 self.eof = True
  39                 break
  40             p = self.form.buf.find(lboundary)
  41             if p >= 0:
  42                 self.buf += self.form.buf[:p]
  43                 self.form.buf = self.form.buf[p + len(lboundary):]
  44                 self.eof = True
  45                 self.form.eof = True
  46                 break
  47             self.buf += self.form.buf[:-len(lboundary)]
  48             self.form.buf = self.form.buf[-len(lboundary):]
  49             if sz >= 0 and len(self.buf) >= sz:
  50                 break
  51             while len(self.form.buf) <= len(lboundary):
  52                 ret = req.input.read(8192)
  53                 if ret == "":
  54                     raise badmultipart("Missing last multipart boundary")
  55                 self.form.buf += ret
  56
  57     def read(self, limit=-1):
  58         self.fillbuf(limit)
  59         if limit >= 0:
  60             ret = self.buf[:limit]
  61             self.buf = self.buf[limit:]
  62         else:
  63             ret = self.buf
  64             self.buf = ""
  65         return ret
  66
  67     def readline(self, limit=-1):
  68         last = 0
  69         while True:
  70             p = self.buf.find(b'\n', last)
  71             if p < 0:
  72                 if self.eof:
  73                     ret = self.buf
  74                     self.buf = ""
  75                     return ret
  76                 last = len(self.buf)
  77                 self.fillbuf(last + 128)
  78             else:
  79                 ret = self.buf[:p + 1]
  80                 self.buf = self.buf[p + 1:]
  81                 return ret
  82
  83     def close(self):
  84         self.fillbuf(-1)
  85
  86     def __enter__(self):
  87         return self
  88
  89     def __exit__(self, *excinfo):
  90         self.close()
  91         return False
  92
  93     def parsehead(self, charset):
  94         def headline():
  95             ln = self.readline(256)
  96             if ln[-1] != ord(b'\n'):
  97                 raise badmultipart("Too long header line in part")
  98             try:
  99                 return ln.decode(charset).rstrip()
 100             except UnicodeError:
 101                 raise badmultipart("Form part header is not in assumed charset")
 102
 103         ln = headline()
 104         while True:
 105             if ln == "":
 106                 break
 107             buf = ln
 108             while True:
 109                 ln = headline()
 110                 if not ln[1:].isspace():
 111                     break
 112                 buf += ln.lstrip()
 113             p = buf.find(':')
 114             if p < 0:
 115                 raise badmultipart("Malformed multipart header line")
 116             self.head[buf[:p].strip().lower()] = buf[p + 1:].lstrip()
 117
 118         val, par = proto.pmimehead(self.head.get("content-disposition", ""))
 119         if val != "form-data":
 120             raise badmultipart("Unexpected Content-Disposition in form part: %r" % val)
 121         if not "name" in par:
 122             raise badmultipart("Missing name in form part")
 123         self.name = par["name"]
 124         self.filename = par.get("filename")
 125         val, par = proto.pmimehead(self.head.get("content-type", ""))
 126         self.ctype = val
 127         self.charset = par.get("charset")
 128         encoding = self.head.get("content-transfer-encoding", "binary")
 129         if encoding != "binary":
 130             raise badmultipart("Form part uses unexpected transfer encoding: %r" % encoding)
 131
 132 class multipart(object):
 133     def __init__(self, req, charset):
 134         val, par = proto.pmimehead(req.ihead.get("Content-Type", ""))
 135         if req.method != "POST" or val != "multipart/form-data":
 136             raise badmultipart("Request is not a multipart form")
 137         if "boundary" not in par:
 138             raise badmultipart("Multipart form lacks boundary")
 139         try:
 140             self.boundary = par["boundary"].encode("us-ascii")
 141         except UnicodeError:
 142             raise badmultipart("Multipart boundary must be ASCII string")
 143         self.req = req
 144         self.buf = b"\r\n"
 145         self.eof = False
 146         self.headcs = charset
 147         self.lastpart = formpart(self)
 148         self.lastpart.close()
 149
 150     def __iter__(self):
 151         return self
 152
 153     def __next__(self):
 154         if not self.lastpart.eof:
 155             raise RuntimeError("All form parts must be read entirely")
 156         if self.eof:
 157             raise StopIteration()
 158         self.lastpart = formpart(self)
 159         self.lastpart.parsehead(self.headcs)
 160         return self.lastpart
 161
 162 def formdata(req):
 163     return req.item(formparse)