codekingpro/portable-devtools

1# defusedxml2#3# Copyright (c) 2013 by Christian Heimes <christian@python.org>4# Licensed to PSF under a Contributor Agreement.5# See https://www.python.org/psf/license for licensing details.6"""Defused xmlrpclib7 8Also defuses gzip bomb9"""10from __future__ import print_function, absolute_import11 12import io13 14from .common import DTDForbidden, EntitiesForbidden, ExternalReferenceForbidden, PY315 16if PY3:17    __origin__ = "xmlrpc.client"18    from xmlrpc.client import ExpatParser19    from xmlrpc import client as xmlrpc_client20    from xmlrpc import server as xmlrpc_server21    from xmlrpc.client import gzip_decode as _orig_gzip_decode22    from xmlrpc.client import GzipDecodedResponse as _OrigGzipDecodedResponse23else:24    __origin__ = "xmlrpclib"25    from xmlrpclib import ExpatParser26    import xmlrpclib as xmlrpc_client27 28    xmlrpc_server = None29    from xmlrpclib import gzip_decode as _orig_gzip_decode30    from xmlrpclib import GzipDecodedResponse as _OrigGzipDecodedResponse31 32try:33    import gzip34except ImportError:  # pragma: no cover35    gzip = None36 37 38# Limit maximum request size to prevent resource exhaustion DoS39# Also used to limit maximum amount of gzip decoded data in order to prevent40# decompression bombs41# A value of -1 or smaller disables the limit42MAX_DATA = 30 * 1024 * 1024  # 30 MB43 44 45def defused_gzip_decode(data, limit=None):46    """gzip encoded data -> unencoded data47 48    Decode data using the gzip content encoding as described in RFC 195249    """50    if not gzip:  # pragma: no cover51        raise NotImplementedError52    if limit is None:53        limit = MAX_DATA54    f = io.BytesIO(data)55    gzf = gzip.GzipFile(mode="rb", fileobj=f)56    try:57        if limit < 0:  # no limit58            decoded = gzf.read()59        else:60            decoded = gzf.read(limit + 1)61    except IOError:  # pragma: no cover62        raise ValueError("invalid data")63    f.close()64    gzf.close()65    if limit >= 0 and len(decoded) > limit:66        raise ValueError("max gzipped payload length exceeded")67    return decoded68 69 70class DefusedGzipDecodedResponse(gzip.GzipFile if gzip else object):71    """a file-like object to decode a response encoded with the gzip72    method, as described in RFC 1952.73    """74 75    def __init__(self, response, limit=None):76        # response doesn't support tell() and read(), required by77        # GzipFile78        if not gzip:  # pragma: no cover79            raise NotImplementedError80        self.limit = limit = limit if limit is not None else MAX_DATA81        if limit < 0:  # no limit82            data = response.read()83            self.readlength = None84        else:85            data = response.read(limit + 1)86            self.readlength = 087        if limit >= 0 and len(data) > limit:88            raise ValueError("max payload length exceeded")89        self.stringio = io.BytesIO(data)90        gzip.GzipFile.__init__(self, mode="rb", fileobj=self.stringio)91 92    def read(self, n):93        if self.limit >= 0:94            left = self.limit - self.readlength95            n = min(n, left + 1)96            data = gzip.GzipFile.read(self, n)97            self.readlength += len(data)98            if self.readlength > self.limit:99                raise ValueError("max payload length exceeded")100            return data101        else:102            return gzip.GzipFile.read(self, n)103 104    def close(self):105        gzip.GzipFile.close(self)106        self.stringio.close()107 108 109class DefusedExpatParser(ExpatParser):110    def __init__(self, target, forbid_dtd=False, forbid_entities=True, forbid_external=True):111        ExpatParser.__init__(self, target)112        self.forbid_dtd = forbid_dtd113        self.forbid_entities = forbid_entities114        self.forbid_external = forbid_external115        parser = self._parser116        if self.forbid_dtd:117            parser.StartDoctypeDeclHandler = self.defused_start_doctype_decl118        if self.forbid_entities:119            parser.EntityDeclHandler = self.defused_entity_decl120            parser.UnparsedEntityDeclHandler = self.defused_unparsed_entity_decl121        if self.forbid_external:122            parser.ExternalEntityRefHandler = self.defused_external_entity_ref_handler123 124    def defused_start_doctype_decl(self, name, sysid, pubid, has_internal_subset):125        raise DTDForbidden(name, sysid, pubid)126 127    def defused_entity_decl(128        self, name, is_parameter_entity, value, base, sysid, pubid, notation_name129    ):130        raise EntitiesForbidden(name, value, base, sysid, pubid, notation_name)131 132    def defused_unparsed_entity_decl(self, name, base, sysid, pubid, notation_name):133        # expat 1.2134        raise EntitiesForbidden(name, None, base, sysid, pubid, notation_name)  # pragma: no cover135 136    def defused_external_entity_ref_handler(self, context, base, sysid, pubid):137        raise ExternalReferenceForbidden(context, base, sysid, pubid)138 139 140def monkey_patch():141    xmlrpc_client.FastParser = DefusedExpatParser142    xmlrpc_client.GzipDecodedResponse = DefusedGzipDecodedResponse143    xmlrpc_client.gzip_decode = defused_gzip_decode144    if xmlrpc_server:145        xmlrpc_server.gzip_decode = defused_gzip_decode146 147 148def unmonkey_patch():149    xmlrpc_client.FastParser = None150    xmlrpc_client.GzipDecodedResponse = _OrigGzipDecodedResponse151    xmlrpc_client.gzip_decode = _orig_gzip_decode152    if xmlrpc_server:153        xmlrpc_server.gzip_decode = _orig_gzip_decode154