path: root/crocoite/behavior.py

# Copyright (c) 2017 crocoite contributors
# 
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# 
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
# 
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

"""
Generic and per-site behavior scripts
"""

import asyncio
from urllib.parse import urlsplit
import os.path
from base64 import b64decode
from collections import OrderedDict
import pkg_resources

from html5lib.serializer import HTMLSerializer
import yaml

from .util import getFormattedViewportMetrics, removeFragment
from . import html
from .html import StripAttributeFilter, StripTagFilter, ChromeTreeWalker
from .devtools import Crashed

class Script:
    """ A JavaScript resource """

    __slots__ = ('path', 'data')

    def __init__ (self, path=None, encoding='utf-8'):
        self.path = path
        if path:
            self.data = pkg_resources.resource_string (__name__, os.path.join ('data', path)).decode (encoding)

    def __repr__ (self):
        return '<Script {}>'.format (self.path)

    def __str__ (self):
        return self.data

    @classmethod
    def fromStr (cls, data):
        s = Script ()
        s.data = data
        return s

class Behavior:
    __slots__ = ('loader', 'logger')

    # unique behavior name
    name = None

    def __init__ (self, loader, logger):
        assert self.name is not None
        self.loader = loader
        self.logger = logger.bind (context=type (self).__name__)

    def __contains__ (self, url):
        """
        Accept every URL by default
        """
        return True

    def __repr__ (self):
        return '<Behavior {}>'.format (self.name)

    async def onload (self):
        """ After loading the page started """
        # this is a dirty hack to make this function an async generator
        return
        yield

    async def onstop (self):
        """ Before page loading is stopped """
        return
        yield

    async def onfinish (self):
        """ After the site has stopped loading """
        return
        yield

class HostnameFilter:
    """ Limit behavior script to hostname """

    hostname = None

    def __contains__ (self, url):
        url = urlsplit (url)
        hostname = url.hostname.split ('.')[::-1]
        return hostname[:2] == self.hostname

class JsOnload (Behavior):
    """ Execute JavaScript on page load """

    __slots__ = ('script', 'context', 'options')

    scriptPath = None

    def __init__ (self, loader, logger):
        super ().__init__ (loader, logger)
        self.script = Script (self.scriptPath)
        self.context = None
        # options passed to constructor
        self.options = {}

    async def onload (self):
        tab = self.loader.tab
        yield self.script

        # This is slightly awkward, since we cannot compile the class into an
        # objectId and then reference it. Therefore the script must return a
        # class constructor, which is then called with a generic options
        # parameter.
        # XXX: is there a better way to do this?
        result = await tab.Runtime.evaluate (expression=str (self.script))
        exception = result.get ('exceptionDetails', None)
        result = result['result']
        assert result['type'] == 'function', result
        assert result.get ('subtype') != 'error', exception
        constructor = result['objectId']

        result = await tab.Runtime.callFunctionOn (
                functionDeclaration='function(options){return new this(options);}',
                objectId=constructor,
                arguments=[{'value': self.options}])
        result = result['result']
        assert result['type'] == 'object', result
        assert result.get ('subtype') != 'error', result
        self.context = result['objectId']

    async def onstop (self):
        tab = self.loader.tab
        assert self.context is not None
        await tab.Runtime.callFunctionOn (functionDeclaration='function(){return this.stop();}', objectId=self.context)
        await tab.Runtime.releaseObject (objectId=self.context)
        return
        yield

### Generic scripts ###

class Scroll (JsOnload):
    name = 'scroll'
    scriptPath = 'scroll.js'

class EmulateScreenMetrics (Behavior):
    name = 'emulateScreenMetrics'

    async def onstop (self):
        """
        Emulate different screen sizes, causing the site to fetch assets (img
        srcset and css, for example) for different screen resolutions.
        """
        cssPpi = 96
        sizes = [
                {'width': 1920, 'height': 1080, 'deviceScaleFactor': 1.5, 'mobile': False},
                {'width': 1920, 'height': 1080, 'deviceScaleFactor': 2, 'mobile': False},
                # very dense display
                {'width': 1920, 'height': 1080, 'deviceScaleFactor': 4, 'mobile': False},
                # just a few samples:
                # 1st gen iPhone (portrait mode)
                {'width': 320, 'height': 480, 'deviceScaleFactor': 163/cssPpi, 'mobile': True},
                # 6th gen iPhone (portrait mode)
                {'width': 750, 'height': 1334, 'deviceScaleFactor': 326/cssPpi, 'mobile': True},
                ]
        l = self.loader
        tab = l.tab
        for s in sizes:
            await tab.Emulation.setDeviceMetricsOverride (**s)
            # give the browser time to re-eval page and start requests
            # XXX: should wait until loader is not busy any more
            await asyncio.sleep (1)
        await tab.Emulation.clearDeviceMetricsOverride ()
        return
        yield

class DomSnapshotEvent:
    __slots__ = ('url', 'document', 'viewport')

    def __init__ (self, url, document, viewport):
        self.url = url
        self.document = document
        self.viewport = viewport

class DomSnapshot (Behavior):
    """
    Get a DOM snapshot of tab and write it to WARC.

    We could use DOMSnapshot.getSnapshot here, but the API is not stable
    yet. Also computed styles are not really necessary here.
    """

    __slots__ = ('script', )

    name = 'domSnapshot'

    def __init__ (self, loader, logger):
        super ().__init__ (loader, logger)
        self.script = Script ('canvas-snapshot.js')

    async def onfinish (self):
        tab = self.loader.tab

        yield self.script
        await tab.Runtime.evaluate (expression=str (self.script), returnByValue=True)

        viewport = await getFormattedViewportMetrics (tab)
        dom = await tab.DOM.getDocument (depth=-1, pierce=True)
        haveUrls = set ()
        for doc in ChromeTreeWalker (dom['root']).split ():
            rawUrl = doc['documentURL']
            if rawUrl in haveUrls:
                # ignore duplicate URLs. they are usually caused by
                # javascript-injected iframes (advertising) with no(?) src
                self.logger.warning ('have DOM snapshot for URL {}, ignoring'.format (rawUrl))
                continue
            url = urlsplit (rawUrl)
            if url.scheme in ('http', 'https'):
                self.logger.debug ('saving DOM snapshot for url {}, base {}'.format (doc['documentURL'], doc['baseURL']))
                haveUrls.add (rawUrl)
                walker = ChromeTreeWalker (doc)
                # remove script, to make the page static and noscript, because at the
                # time we took the snapshot scripts were enabled
                disallowedTags = ['script', 'noscript']
                disallowedAttributes = html.eventAttributes
                stream = StripAttributeFilter (StripTagFilter (walker, disallowedTags), disallowedAttributes)
                serializer = HTMLSerializer ()
                yield DomSnapshotEvent (removeFragment (doc['documentURL']), serializer.render (stream, 'utf-8'), viewport)

class ScreenshotEvent:
    __slots__ = ('yoff', 'data', 'url')

    def __init__ (self, url, yoff, data):
        self.url = url
        self.yoff = yoff
        self.data = data

class Screenshot (Behavior):
    """
    Create screenshot from tab and write it to WARC
    """

    name = 'screenshot'

    async def onfinish (self):
        tab = self.loader.tab

        tree = await tab.Page.getFrameTree ()
        try:
            url = removeFragment (tree['frameTree']['frame']['url'])
        except KeyError:
            self.logger.error ('frame without url', tree=tree)
            url = None

        # see https://github.com/GoogleChrome/puppeteer/blob/230be28b067b521f0577206899db01f0ca7fc0d2/examples/screenshots-longpage.js
        # Hardcoded max texture size of 16,384 (crbug.com/770769)
        maxDim = 16*1024
        metrics = await tab.Page.getLayoutMetrics ()
        contentSize = metrics['contentSize']
        width = min (contentSize['width'], maxDim)
        # we’re ignoring horizontal scroll intentionally. Most horizontal
        # layouts use JavaScript scrolling and don’t extend the viewport.
        for yoff in range (0, contentSize['height'], maxDim):
            height = min (contentSize['height'] - yoff, maxDim)
            clip = {'x': 0, 'y': yoff, 'width': width, 'height': height, 'scale': 1}
            ret = await tab.Page.captureScreenshot (format='png', clip=clip)
            data = b64decode (ret['data'])
            yield ScreenshotEvent (url, yoff, data)

class Click (JsOnload):
    """ Generic link clicking """

    name = 'click'
    scriptPath = 'click.js'

    def __init__ (self, loader, logger):
        super ().__init__ (loader, logger)
        with pkg_resources.resource_stream (__name__, os.path.join ('data', 'click.yaml')) as fd:
            self.options['sites'] = list (yaml.load_all (fd))

class ExtractLinksEvent:
    __slots__ = ('links', )

    def __init__ (self, links):
        self.links = links

class ExtractLinks (Behavior):
    """
    Extract links from a page using JavaScript
    
    We could retrieve a HTML snapshot and extract links here, but we’d have to
    manually resolve relative links.
    """

    __slots__ = ('script', )

    name = 'extractLinks'

    def __init__ (self, loader, logger):
        super ().__init__ (loader, logger)
        self.script = Script ('extract-links.js')

    async def onfinish (self):
        tab = self.loader.tab
        yield self.script
        result = await tab.Runtime.evaluate (expression=str (self.script), returnByValue=True)
        yield ExtractLinksEvent (list (set (result['result']['value'])))

class Crash (Behavior):
    """ Crash the browser. For testing only. Obviously. """

    name = 'crash'

    async def onstop (self):
        try:
            await self.loader.tab.Page.crash ()
        except Crashed:
            pass
        return
        yield

# available behavior scripts. Order matters, move those modifying the page
# towards the end of available
available = [Scroll, Click, ExtractLinks, Screenshot, EmulateScreenMetrics, DomSnapshot]
#available.append (Crash)
# order matters, since behavior can modify the page (dom snapshots, for instance)
availableMap = OrderedDict (map (lambda x: (x.name, x), available))