aboutsummaryrefslogtreecommitdiffstats
path: root/libmproxy/contentview.py
diff options
context:
space:
mode:
authorMaximilian Hils <git@maximilianhils.com>2015-09-12 13:49:16 +0200
committerMaximilian Hils <git@maximilianhils.com>2015-09-12 13:49:16 +0200
commit049d253a83e18116340670cb86528b4ac1d3b215 (patch)
treed2a6b1db9fd1c98bdc500c3b43158d73b02954cb /libmproxy/contentview.py
parent9c31669211dd0b81e3f7f5325a3564e827a8d6d0 (diff)
downloadmitmproxy-049d253a83e18116340670cb86528b4ac1d3b215.tar.gz
mitmproxy-049d253a83e18116340670cb86528b4ac1d3b215.tar.bz2
mitmproxy-049d253a83e18116340670cb86528b4ac1d3b215.zip
simplify contentview api
Diffstat (limited to 'libmproxy/contentview.py')
-rw-r--r--libmproxy/contentview.py549
1 files changed, 0 insertions, 549 deletions
diff --git a/libmproxy/contentview.py b/libmproxy/contentview.py
deleted file mode 100644
index 219adfb7..00000000
--- a/libmproxy/contentview.py
+++ /dev/null
@@ -1,549 +0,0 @@
-from __future__ import absolute_import
-import cStringIO
-import json
-import logging
-import subprocess
-import sys
-
-import lxml.html
-import lxml.etree
-from PIL import Image
-
-from PIL.ExifTags import TAGS
-import html2text
-import six
-
-from netlib.odict import ODict
-from netlib import encoding
-import netlib.utils
-from . import utils
-from .exceptions import ContentViewException
-from .contrib import jsbeautifier
-from .contrib.wbxml.ASCommandResponse import ASCommandResponse
-
-try:
- import pyamf
- from pyamf import remoting, flex
-except ImportError: # pragma nocover
- pyamf = None
-
-try:
- import cssutils
-except ImportError: # pragma nocover
- cssutils = None
-else:
- cssutils.log.setLevel(logging.CRITICAL)
-
- cssutils.ser.prefs.keepComments = True
- cssutils.ser.prefs.omitLastSemicolon = False
- cssutils.ser.prefs.indentClosingBrace = False
- cssutils.ser.prefs.validOnly = False
-
-VIEW_CUTOFF = 1024 * 50
-KEY_MAX = 30
-
-
-def format_dict(d):
- """
- Transforms the given dictionary into a list of
- ("key", key )
- ("value", value)
- tuples, where key is padded to a uniform width.
- """
- max_key_len = max(len(k) for k in d.keys())
- max_key_len = min(max_key_len, KEY_MAX)
- for key, value in d.items():
- key += ":"
- key = key.ljust(max_key_len + 2)
- yield [
- ("header", key),
- ("text", value)
- ]
-
-
-def format_text(content, limit):
- """
- Transforms the given content into
- """
- content = netlib.utils.cleanBin(content)
-
- for line in content[:limit].splitlines():
- yield [("text", line)]
-
- for msg in trailer(content, limit):
- yield msg
-
-
-def trailer(content, limit):
- bytes_removed = len(content) - limit
- if bytes_removed > 0:
- yield [
- ("cutoff", "... {} of data not shown.".format(netlib.utils.pretty_size(bytes_removed)))
- ]
-
-
-class View(object):
- name = None
- prompt = ()
- content_types = []
-
- def __call__(self, hdrs, content, limit):
- """
- Returns:
- A (description, content generator) tuple.
-
- The content generator yields lists of (style, text) tuples.
- Iit must not yield tuples of tuples, because urwid cannot process that.
- """
- raise NotImplementedError()
-
-
-class ViewAuto(View):
- name = "Auto"
- prompt = ("auto", "a")
- content_types = []
-
- def __call__(self, hdrs, content, limit):
- ctype = hdrs.get("content-type")
- if ctype:
- ct = netlib.utils.parse_content_type(ctype) if ctype else None
- ct = "%s/%s" % (ct[0], ct[1])
- if ct in content_types_map:
- return content_types_map[ct][0](hdrs, content, limit)
- elif utils.isXML(content):
- return get("XML")(hdrs, content, limit)
- return get("Raw")(hdrs, content, limit)
-
-
-class ViewRaw(View):
- name = "Raw"
- prompt = ("raw", "r")
- content_types = []
-
- def __call__(self, hdrs, content, limit):
- return "Raw", format_text(content, limit)
-
-
-class ViewHex(View):
- name = "Hex"
- prompt = ("hex", "e")
- content_types = []
-
- @staticmethod
- def _format(content, limit):
- for offset, hexa, s in netlib.utils.hexdump(content[:limit]):
- yield [
- ("offset", offset + " "),
- ("text", hexa + " "),
- ("text", s)
- ]
- for msg in trailer(content, limit):
- yield msg
-
- def __call__(self, hdrs, content, limit):
- return "Hex", self._format(content, limit)
-
-
-class ViewXML(View):
- name = "XML"
- prompt = ("xml", "x")
- content_types = ["text/xml"]
-
- def __call__(self, hdrs, content, limit):
- parser = lxml.etree.XMLParser(
- remove_blank_text=True,
- resolve_entities=False,
- strip_cdata=False,
- recover=False
- )
- try:
- document = lxml.etree.fromstring(content, parser)
- except lxml.etree.XMLSyntaxError:
- return None
- docinfo = document.getroottree().docinfo
-
- prev = []
- p = document.getroottree().getroot().getprevious()
- while p is not None:
- prev.insert(
- 0,
- lxml.etree.tostring(p)
- )
- p = p.getprevious()
- doctype = docinfo.doctype
- if prev:
- doctype += "\n".join(prev).strip()
- doctype = doctype.strip()
-
- s = lxml.etree.tostring(
- document,
- pretty_print=True,
- xml_declaration=True,
- doctype=doctype or None,
- encoding=docinfo.encoding
- )
-
- return "XML-like data", format_text(s, limit)
-
-
-class ViewJSON(View):
- name = "JSON"
- prompt = ("json", "s")
- content_types = ["application/json"]
-
- def __call__(self, hdrs, content, limit):
- pretty_json = utils.pretty_json(content)
- if pretty_json:
- return "JSON", format_text(pretty_json, limit)
-
-
-class ViewHTML(View):
- name = "HTML"
- prompt = ("html", "h")
- content_types = ["text/html"]
-
- def __call__(self, hdrs, content, limit):
- if utils.isXML(content):
- parser = lxml.etree.HTMLParser(
- strip_cdata=True,
- remove_blank_text=True
- )
- d = lxml.html.fromstring(content, parser=parser)
- docinfo = d.getroottree().docinfo
- s = lxml.etree.tostring(
- d,
- pretty_print=True,
- doctype=docinfo.doctype
- )
- return "HTML", format_text(s, limit)
-
-
-class ViewHTMLOutline(View):
- name = "HTML Outline"
- prompt = ("html outline", "o")
- content_types = ["text/html"]
-
- def __call__(self, hdrs, content, limit):
- content = content.decode("utf-8")
- h = html2text.HTML2Text(baseurl="")
- h.ignore_images = True
- h.body_width = 0
- content = h.handle(content)
- return "HTML Outline", format_text(content, limit)
-
-
-class ViewURLEncoded(View):
- name = "URL-encoded"
- prompt = ("urlencoded", "u")
- content_types = ["application/x-www-form-urlencoded"]
-
- def __call__(self, hdrs, content, limit):
- d = netlib.utils.urldecode(content)
- return "URLEncoded form", format_dict(ODict(d))
-
-
-class ViewMultipart(View):
- name = "Multipart Form"
- prompt = ("multipart", "m")
- content_types = ["multipart/form-data"]
-
- @staticmethod
- def _format(v):
- yield [("highlight", "Form data:\n")]
- for message in format_dict(ODict(v)):
- yield message
-
- def __call__(self, hdrs, content, limit):
- v = netlib.utils.multipartdecode(hdrs, content)
- if v:
- return "Multipart form", self._format(v)
-
-
-if pyamf:
- class DummyObject(dict):
- def __init__(self, alias):
- dict.__init__(self)
-
- def __readamf__(self, input):
- data = input.readObject()
- self["data"] = data
-
-
- def pyamf_class_loader(s):
- for i in pyamf.CLASS_LOADERS:
- if i != pyamf_class_loader:
- v = i(s)
- if v:
- return v
- return DummyObject
-
-
- pyamf.register_class_loader(pyamf_class_loader)
-
-
- class ViewAMF(View):
- name = "AMF"
- prompt = ("amf", "f")
- content_types = ["application/x-amf"]
-
- def unpack(self, b, seen=set([])):
- if hasattr(b, "body"):
- return self.unpack(b.body, seen)
- if isinstance(b, DummyObject):
- if id(b) in seen:
- return "<recursion>"
- else:
- seen.add(id(b))
- for k, v in b.items():
- b[k] = self.unpack(v, seen)
- return b
- elif isinstance(b, dict):
- for k, v in b.items():
- b[k] = self.unpack(v, seen)
- return b
- elif isinstance(b, list):
- return [self.unpack(i) for i in b]
- elif isinstance(b, flex.ArrayCollection):
- return [self.unpack(i, seen) for i in b]
- else:
- return b
-
- def _format(self, envelope, limit):
- for target, message in iter(envelope):
- if isinstance(message, pyamf.remoting.Request):
- yield [
- ("header", "Request: "),
- ("text", str(target)),
- ]
- else:
- yield [
- ("header", "Response: "),
- ("text", "%s, code %s" % (target, message.status)),
- ]
-
- s = json.dumps(self.unpack(message), indent=4)
- for msg in format_text(s, limit):
- yield msg
-
- def __call__(self, hdrs, content, limit):
- envelope = remoting.decode(content, strict=False)
- if envelope:
- return "AMF v%s" % envelope.amfVersion, self._format(envelope, limit)
-
-
-class ViewJavaScript(View):
- name = "JavaScript"
- prompt = ("javascript", "j")
- content_types = [
- "application/x-javascript",
- "application/javascript",
- "text/javascript"
- ]
-
- def __call__(self, hdrs, content, limit):
- opts = jsbeautifier.default_options()
- opts.indent_size = 2
- res = jsbeautifier.beautify(content[:limit], opts)
- cutoff = max(0, len(content) - limit)
- return "JavaScript", format_text(res, limit - cutoff)
-
-
-class ViewCSS(View):
- name = "CSS"
- prompt = ("css", "c")
- content_types = [
- "text/css"
- ]
-
- def __call__(self, hdrs, content, limit):
- if cssutils:
- sheet = cssutils.parseString(content)
- beautified = sheet.cssText
- else:
- beautified = content
-
- return "CSS", format_text(beautified, limit)
-
-
-class ViewImage(View):
- name = "Image"
- prompt = ("image", "i")
- content_types = [
- "image/png",
- "image/jpeg",
- "image/gif",
- "image/vnd.microsoft.icon",
- "image/x-icon",
- ]
-
- def __call__(self, hdrs, content, limit):
- try:
- img = Image.open(cStringIO.StringIO(content))
- except IOError:
- return None
- parts = [
- ("Format", str(img.format_description)),
- ("Size", "%s x %s px" % img.size),
- ("Mode", str(img.mode)),
- ]
- for i in sorted(img.info.keys()):
- if i != "exif":
- parts.append(
- (str(i), str(img.info[i]))
- )
- if hasattr(img, "_getexif"):
- ex = img._getexif()
- if ex:
- for i in sorted(ex.keys()):
- tag = TAGS.get(i, i)
- parts.append(
- (str(tag), str(ex[i]))
- )
- clean = []
- for i in parts:
- clean.append(
- [netlib.utils.cleanBin(i[0]), netlib.utils.cleanBin(i[1])]
- )
- fmt = format_dict(ODict(clean))
- return "%s image" % img.format, fmt
-
-
-class ViewProtobuf(View):
- """Human friendly view of protocol buffers
- The view uses the protoc compiler to decode the binary
- """
-
- name = "Protocol Buffer"
- prompt = ("protobuf", "p")
- content_types = [
- "application/x-protobuf",
- "application/x-protobuffer",
- ]
-
- @staticmethod
- def is_available():
- try:
- p = subprocess.Popen(
- ["protoc", "--version"],
- stdout=subprocess.PIPE
- )
- out, _ = p.communicate()
- return out.startswith("libprotoc")
- except:
- return False
-
- def decode_protobuf(self, content):
- # if Popen raises OSError, it will be caught in
- # get_content_view and fall back to Raw
- p = subprocess.Popen(['protoc', '--decode_raw'],
- stdin=subprocess.PIPE,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE)
- out, err = p.communicate(input=content)
- if out:
- return out
- else:
- return err
-
- def __call__(self, hdrs, content, limit):
- decoded = self.decode_protobuf(content)
- return "Protobuf", format_text(decoded, limit)
-
-
-class ViewWBXML(View):
- name = "WBXML"
- prompt = ("wbxml", "w")
- content_types = [
- "application/vnd.wap.wbxml",
- "application/vnd.ms-sync.wbxml"
- ]
-
- def __call__(self, hdrs, content, limit):
-
- try:
- parser = ASCommandResponse(content)
- parsedContent = parser.xmlString
- if parsedContent:
- return "WBXML", format_text(parsedContent, limit)
- except:
- return None
-
-
-views = [
- ViewAuto(),
- ViewRaw(),
- ViewHex(),
- ViewJSON(),
- ViewXML(),
- ViewWBXML(),
- ViewHTML(),
- ViewHTMLOutline(),
- ViewJavaScript(),
- ViewCSS(),
- ViewURLEncoded(),
- ViewMultipart(),
- ViewImage(),
-]
-if pyamf:
- views.append(ViewAMF())
-
-if ViewProtobuf.is_available():
- views.append(ViewProtobuf())
-
-content_types_map = {}
-for i in views:
- for ct in i.content_types:
- l = content_types_map.setdefault(ct, [])
- l.append(i)
-
-view_prompts = [i.prompt for i in views]
-
-
-def get_by_shortcut(c):
- for i in views:
- if i.prompt[1] == c:
- return i
-
-
-def get(name):
- for i in views:
- if i.name == name:
- return i
-
-
-def get_content_view(viewmode, headers, content, limit, is_request):
- """
- Returns:
- A (description, content generator) tuple.
-
- Raises:
- ContentViewException, if the content view threw an error.
- """
- if not content:
- if is_request:
- return "No request content (press tab to view response)", []
- else:
- return "No content", []
- msg = []
-
- enc = headers.get("content-encoding")
- if enc and enc != "identity":
- decoded = encoding.decode(enc, content)
- if decoded:
- content = decoded
- msg.append("[decoded %s]" % enc)
- try:
- ret = viewmode(headers, content, limit)
- # Third-party viewers can fail in unexpected ways...
- except Exception as e:
- six.reraise(
- ContentViewException,
- ContentViewException(str(e)),
- sys.exc_info()[2]
- )
- if not ret:
- ret = get("Raw")(headers, content, limit)
- msg.append("Couldn't parse: falling back to Raw")
- else:
- msg.append(ret[0])
- return " ".join(msg), ret[1]