# This file is Copyright 2019 Volatility Foundation and licensed under the Volatility Software License 1.0 # which is available at https://www.volatilityfoundation.org/license/vsl-v1.0 # import bz2 import contextlib import gzip import hashlib import logging import lzma import os import ssl import urllib.parse import urllib.request import zipfile from typing import Optional, Any from urllib import error from volatility import framework from volatility.framework import constants try: import magic HAS_MAGIC = True except ImportError: HAS_MAGIC = False try: import smb.SMBHandler except ImportError: pass vollog = logging.getLogger(__name__) # TODO: Type-annotating the ResourceAccessor.open method is difficult because HTTPResponse is not actually an IO[Any] type # fix this class ResourceAccessor(object): """Object for openning URLs as files (downloading locally first if necessary)""" list_handlers = True def __init__(self, progress_callback: Optional[constants.ProgressCallback] = None, context: Optional[ssl.SSLContext] = None) -> None: """Creates a resource accessor. Note: context is an SSL context, not a volatility context """ self._progress_callback = progress_callback self._context = context self._handlers = list(framework.class_subclasses(urllib.request.BaseHandler)) if self.list_handlers: vollog.log(constants.LOGLEVEL_VVV, "Available URL handlers: {}".format(", ".join([x.__name__ for x in self._handlers]))) self.__class__.list_handlers = False # Current urllib.request.urlopen returns Any, so we do the same def open(self, url: str, mode: str = "rb") -> Any: """Returns a file-like object for a particular URL opened in mode. If the file is remote, it will be downloaded and locally cached """ urllib.request.install_opener(urllib.request.build_opener(*self._handlers)) try: fp = urllib.request.urlopen(url, context = self._context) except error.URLError as excp: if excp.args: if isinstance(excp.args[0], ssl.SSLCertVerificationError): vollog.warning("SSL certificate verification failed: attempting UNVERIFIED retrieval") non_verifying_ctx = ssl.SSLContext() non_verifying_ctx.check_hostname = False non_verifying_ctx.verify_mode = ssl.CERT_NONE fp = urllib.request.urlopen(url, context = non_verifying_ctx) else: raise excp else: raise excp with contextlib.closing(fp) as fp: # Cache the file locally parsed_url = urllib.parse.urlparse(url) if parsed_url.scheme == 'file': # ZipExtFiles (files in zips) cannot seek, so must be cached in order to use and/or decompress curfile = urllib.request.urlopen(url, context = self._context) else: # TODO: find a way to check if we already have this file (look at http headers?) block_size = 1028 * 8 temp_filename = os.path.join(constants.CACHE_PATH, "data_" + hashlib.sha512(bytes(url, 'latin-1')).hexdigest()) if not os.path.exists(temp_filename): vollog.debug("Caching file at: {}".format(temp_filename)) try: content_length = fp.info().get('Content-Length', -1) except AttributeError: # If our fp doesn't have an info member, carry on gracefully content_length = -1 cache_file = open(temp_filename, "wb") count = 0 block = fp.read(block_size) while block: count += len(block) if self._progress_callback: self._progress_callback(count * 100 / max(count, int(content_length)), "Reading file {}".format(url)) cache_file.write(block) block = fp.read(block_size) cache_file.close() # Re-open the cache with a different mode curfile = open(temp_filename, mode = "rb") # Determine whether the file is a particular type of file, and if so, open it as such IMPORTED_MAGIC = False if HAS_MAGIC: stop = False while not stop: detected = None try: # Detect the content detected = magic.detect_from_fobj(curfile) IMPORTED_MAGIC = True # This is because python-magic and file provide a magic module # Only file's python has magic.detect_from_fobj except AttributeError: pass except: pass if detected: if detected.mime_type == 'application/x-xz': curfile = lzma.LZMAFile(curfile, mode) elif detected.mime_type == 'application/x-bzip2': curfile = bz2.BZ2File(curfile, mode) elif detected.mime_type == 'application/x-gzip': curfile = gzip.GzipFile(fileobj = curfile, mode = mode) if detected.mime_type in ['application/x-xz', 'application/x-bzip2', 'application/x-gzip']: # Read and rewind to ensure we're inside any compressed file layers curfile.read(1) curfile.seek(0) else: stop = True else: stop = True if not IMPORTED_MAGIC: # Somewhat of a hack, but prevents a hard dependency on the magic module url_path = parsed_url.path stop = False while not stop: url_path_split = url_path.split(".") url_path, extension = url_path_split[:-1], url_path_split[-1] url_path = ".".join(url_path) if extension == "xz": curfile = lzma.LZMAFile(curfile, mode) elif extension == "bz2": curfile = bz2.BZ2File(curfile, mode) elif extension == "gz": curfile = gzip.GzipFile(fileobj = curfile, mode = mode) else: stop = True # Fallback in case the file doesn't exist if curfile is None: raise ValueError("URL does not reference an openable file") return curfile class JarHandler(urllib.request.BaseHandler): """Handles the jar scheme for URIs. Reference used for the schema syntax: http://docs.netkernel.org/book/view/book:mod:reference/doc:layer1:schemes:jar Actual reference (found from https://www.w3.org/wiki/UriSchemes/jar) seemed not to return: http://developer.java.sun.com/developer/onlineTraining/protocolhandlers/ """ @staticmethod def default_open(req: urllib.request.Request) -> Optional[Any]: """Handles the request if it's the jar scheme.""" if req.type == 'jar': subscheme, remainder = req.full_url.split(":")[1], ":".join(req.full_url.split(":")[2:]) if subscheme != 'file': vollog.log(constants.LOGLEVEL_VVV, "Unsupported jar subscheme {}".format(subscheme)) return None zipsplit = remainder.split("!") if len(zipsplit) != 2: vollog.log(constants.LOGLEVEL_VVV, "Path did not contain exactly one fragment indicator: {}".format(remainder)) return None zippath, filepath = zipsplit return zipfile.ZipFile(zippath).open(filepath) return None