<filename>scripts/parser/frontmatter_parser.py<gh_stars>1-10
"""
This class parses frontmatter resulting json file, extracts ui labels, nd flattens labels from a particular screen
Result doesn't reflect uis' hierarchical structure, but text in a final string (the content of an activity) is ordered according to the view hierarchy
Menus are parsed but not included in the final activity string
FrontmatterUiParser.ui contains all views as a mapping: view_id->{label, activity}
FrontmatterUiParser.activities contains activities data, a map: activity_name->{title, labels}
FrontmatterUiParser.transitions contains transition data, a map: src_activity->{dest_activity, trigger}
"""
import itertools
import json
import os
import re
from collections import defaultdict, Iterable
from pathlib import Path

import more_itertools


class View:
    def __init__(self):
        self.var_name = ""
        self.icon = ""
        self.listeners = []
        self.label = ""
        self.clazz = ""
        self.activity = ""
        self.is_menu = ""
        self.is_dialog = ""
        self.context = ""  # in case of a dialog this is title+message
        self.api = []


class Activity:
    def __init__(self):
        self.title = ""
        self.labels = ""
        self.views = []


class Broadcast:
    def __init__(self, name, intents: list, apis: list):
        self.name = name
        self.apis = apis
        self.intents = intents


class FrontmatterUiParser:
    sentence_delimiter = "|"
    text_attributes = ['text', 'title', 'contentDescription', 'textOn', 'textOff', 'label', 'hint']  # no accessibilityPaneTitle,tooltipText
    dynamic_text_pattern = r'text\d*'
    image_attribures = ['background']
    image_marker = 'drawable'
    dynamic_image_pattern = r'image\d*'

    # image_attr: otherAttributes/src/value, otherAttributes/image0

    def __init__(self, path):
        self.permissions = []
        self.broadcasts = {}
        self.lifecycle = {}
        self.services = {}
        self.service_mapping = defaultdict(list)  # activity/broadcast -> service
        self.apk_path = path
        self.pkg = self.get_pkg_name(path)
        self.lang = ''
        self.ui_error = ''
        self.api_error = ''
        self.apk_platform = ''
        self.transition_stats = list()
        self.ui = defaultdict(View)  # map guid->View
        self.ui_apis = []
        self.meta = None
        self.activities = defaultdict(Activity)
        self.transitions = dict()
        self.dialogs = defaultdict(View)  # for dialogs with missing guid (old version of analysis)
        self.sensitive_perms = {}
        self.load_sensitive_permissions()
        self.actions = self.load_actions()

    def is_system_broadcast(self, broadcast: Broadcast):
        return any(intent in self.actions for intent in broadcast.intents)

    @staticmethod
    def is_xamarin(activity):
        """
        activities generated by xamarin have specific name
        """
        name = activity['name']
        prefix = name.split('.')[0]
        return re.match(r'md5[0-9a-f]{32}', prefix)

    @staticmethod
    def is_unity(activity):
        name = activity['name']
        return name.startswith('com.unity3d') and not name.startswith('com.unity3d.service')

    @staticmethod
    def is_apache(activities):
        for activity in activities:
            dialogs = activity.get('dialogs', [])
            for dialog in dialogs:
                buttons = dialog.get("buttons", [])
                for button in buttons:
                    listener = button.get('listener', '')
                    if listener.startswith('<org.apache.cordova'):
                        return True
        return False

    @staticmethod
    def is_air(activity):
        name = activity['name']
        return name.startswith('air.')

    @staticmethod
    def is_kodular(activity):
        name = activity['name']
        return name.startswith('io.kodular')

    @staticmethod
    def is_appybuilder(jactivities):
        for activity in jactivities:
            name = activity['name']
            if name.startswith('com.appybuilder'):
                return True
        return False

    @staticmethod
    def is_solar2d(activity):
        name = activity['name']
        return name.startswith('com.ansca.corona')

    @staticmethod
    def is_platform(activities):
        """
        Return true if the results are empty which is an indication that a framework (like abobe.air, apache, ionic) was used
        """
        for activity in activities:
            if 'orphanedFragments' in activity:
                return False
            layouts = activity['layouts']
            for layout in layouts:
                if 'children' in layout and len(layout['children']) > 0:
                    return False
                if 'fragmentClass' in layout or 'fragmentClass' in layout:
                    return False
        return True

    @staticmethod
    def get_pkg_name(path):
        basename = os.path.basename(path)
        pkg = os.path.splitext(basename)[0]
        return pkg

    @classmethod
    def sanitize(cls, text: str):
        if text is None:
            return ''
        return text.replace('"', "'").replace('\n\r', ' ').replace('\n', ' ').replace('\r', ' ').replace(cls.sentence_delimiter, ' ')

    def is_english(self):
        return self.meta.get('defaultLanguage', '') in {'en', 'unknown'}

    @staticmethod
    def load_permission_list(perm_file: Path):
        with perm_file.open() as fd:
            apis = [api.strip() for api in fd.readlines() if not api.startswith("#")]
        return apis

    def load_sensitive_permissions(self):
        perm_list_path = Path('permission_mapping')
        for perm_file in perm_list_path.iterdir():
            perm_name = perm_file.stem
            apis = self.load_permission_list(perm_file)
            for api in apis:
                self.sensitive_perms[api] = f"android.permission.{perm_name}"

    def read_api(self, api_file_path: Path, sensitive_only: bool = True):
        """
            should be used after loading ui
        """
        with open(api_file_path) as fd:
            try:
                data = json.load(fd)
            except:
                print(f"Error while reading {api_file_path}")
                self.api_error = 'read_error'
                return
            if 'error' in data:
                self.api_error = data['error'].lower()
                return
            self.permissions = data['permissions']
            for view in data.get('views', []):
                self.load_api_view(view, sensitive_only)
            # load serviceLS
            for serviceLS in data.get('serviceLS', []):
                self.load_api_service_lifecycle(serviceLS, sensitive_only)
            # load broadcasts
            for broadcast in data.get('broadcasts', []):
                self.load_api_broadcast(broadcast, sensitive_only)
            # load activityLC
            for activityLC in data.get('activityLC', []):
                self.load_api_activity_lifecycle(activityLC, sensitive_only)

    def load_broadcasts_api(self, api_file_path: Path, sensitive_only: bool):
        with open(api_file_path) as fd:
            try:
                data = json.load(fd)
            except:
                print(f"Error while reading {api_file_path}")
                return
            if 'error' in data:
                return
            self.permissions = data['permissions']
            # load broadcasts
            for broadcast in data.get('broadcasts', []):
                self.load_api_broadcast(broadcast, sensitive_only)

    def load_api_view(self, view: dict, sensitive_only: bool):
        apis = view['api']
        guid = view['guid']
        ui = self.ui.get(guid)
        ui_apis = []
        activity = ui.activity if ui else ""
        for api in apis:
            ui_api = self.parse_api_call(api, sensitive_only, activity)
            ui_apis.extend(ui_api)
        self.ui_apis.extend(ui_apis)
        if ui is None:
            pass
            # print(f'Unknown UI widget: {guid} in {self.apk_path}')
        else:
            ui.api.extend(ui_apis)

    def filter_api(self, api, sensitive_only):
        api = api.strip('/')
        perm = self.sensitive_perms.get(api, None)
        if perm in self.permissions:  # sensitive permission is declared
            return api
        if perm is None and not sensitive_only:
            return api  # permission is not sensitive, we have no information if api is covered by any other permission
        return None

    def read_ui(self, force):
        with open(self.apk_path) as fd:
            try:
                data = json.load(fd)
            except:
                print(f"Error while reading {self.apk_path}")
                return
        if 'error' in data:
            self.ui_error = data['error'].lower()
            return
        self.meta = data.get('meta', {})
        self.apk_platform = self.meta.get('type', 'android').lower()
        if self.apk_platform == 'normal':
            self.apk_platform = 'android'
        self.lang = self.meta.get('defaultLanguage', '')
        if not self.is_normal():
            return
        if not self.is_english():
            return
        jactivities = data['activities']
        jtransitions = data['transitions']
        if any(map(self.is_xamarin, jactivities)):
            self.apk_platform = 'xamarin'
            if not force:
                return
        if any(map(self.is_unity, jactivities)) and not force:
            self.apk_platform = 'unity'
            if not force:
                return
        if self.is_platform(jactivities):
            self.apk_platform = 'unidentified'
            if self.is_appybuilder(jactivities):
                self.apk_platform = 'appbuilder'
                if not force:
                    return
            if self.is_apache(jactivities):
                self.apk_platform = 'apache'
                if not force:
                    return
            if any(map(self.is_air, jactivities)):
                self.apk_platform = 'air'
                if not force:
                    return
            if any(map(self.is_kodular, jactivities)):
                self.apk_platform = 'kodular'
                if not force:
                    return
            if any(map(self.is_solar2d, jactivities)):
                self.apk_platform = 'solar2d'
                if not force:
                    return
            if not force:
                return
        activity_names = set()
        # parse activities
        for jactivity in jactivities:
            name, title, texts = self.parse_activity(jactivity)
            activity_names.add(name)
            self.activities[name].title = title
            self.activities[name].labels = texts
        # parse transitions
        for jtransition in jtransitions:
            src_name = jtransition['scr']
            dest_name = jtransition['dest']
            button_ids = jtransition['trigger']
            button_labels = []
            for guid in button_ids:
                if guid in self.ui:
                    button_labels.append(self.ui[guid].label)
            button_labels = [self.ui[guid].label for guid in button_ids if guid in self.ui]
            # button_label = f" {self.sentence_delimeter} ".join(filter(self.is_not_blank, button_labels))
            # src_activity = self.activities[src_name]
            # dest_activity = self.activities[dest_name]
            transition_src = self.transitions.setdefault(src_name, [])
            if len(button_labels) == 0:
                transition_src.append({'dest': dest_name, 'trigger': ''})
            else:
                for button_label in button_labels:
                    transition_src.append({'dest': dest_name, 'trigger': button_label})

    def get_activity_data(self):
        app_data = []
        for name, activity in self.activities.items():
            title = activity.title
            labels = activity.labels
            for label in labels:
                app_data.append(
                    {
                        'pkg': self.pkg,
                        'activity': name,
                        'title': title,
                        'labels': label,
                    }
                )
        return app_data

    def get_widget_ui_data(self, with_empty) -> list:
        app_ui = []
        for guid, view in self.ui.items():
            if with_empty or self.is_not_blank(view.label) or self.is_not_blank(view.icon):
                app_ui.append(
                    {
                        'pkg': self.pkg,
                        'activity': view.activity,
                        'label': view.label,
                        'icon': view.icon,
                        'guid': guid,
                        'class': view.clazz,
                        'listener': "|".join(view.listeners)
                    }
                )
        return app_ui

    def get_widget_api_data(self, with_empty) -> list:
        app_api = []
        for guid, view in self.ui.items():
            if with_empty or self.is_not_blank(view.label) or self.is_not_blank(view.icon):
                for api in view.api:
                    app_api.append(
                        {
                            'pkg': self.pkg,
                            'api': api,
                            'guid': guid
                        }
                    )
        return app_api

    def collect_api(self):
        """
        :return: dict of lists: activity_name -> [apis]
        """
        api_data = []
        # activities = defaultdict(list)
        for view in self.ui.values():
            if view.is_dialog or view.is_menu:  # skip dialogs and menus
                continue
            activity_name = view.activity
            for api in view.api:
                # activities[activity_name].extend(view.api)
                api_data.append(
                    {
                        'pkg': self.pkg,
                        'activity': activity_name,
                        'api': api,
                    }
                )
        return api_data

    def get_app_data_transitions(self):
        app_data = []
        for activity_name, transitions in self.transitions.items():
            if activity_name not in self.activities:
                continue
            src_activity = self.activities[activity_name]
            src_labels = src_activity.labels
            for transition in transitions:
                dest_name = transition['dest']
                if dest_name not in self.activities:
                    continue
                dest_activity = self.activities[dest_name]
                dest_labels = dest_activity.labels
                triggers = transition['trigger']
                button_label = self.concat_labels(triggers)
                for src_label in src_labels:
                    for dest_label in dest_labels:
                        app_data.append(
                            {
                                'pkg': self.pkg,
                                'src': src_label,
                                'triggers': button_label,
                                'dest': dest_label
                            }
                        )
        return app_data

    def concat_labels(self, labels):
        labels = list(labels)
        return f" {self.sentence_delimiter} ".join(filter(self.is_not_blank, map(self.sanitize, labels)))

    def parse_activity(self, jactivity):
        activity_name = jactivity['name']
        titles = jactivity.get('titles', [])
        titles_string = self.concat_labels(titles)
        menus = jactivity.get('menu', [])
        # parse menus

        for menu in menus:
            items = menu['items']
            self.parse_menu_items(items, activity_name)

        dialogs = jactivity.get('dialogs', [])
        for dialog in dialogs:
            self.parse_dialog(dialog, activity_name)  # TODO

        # parse layouts
        jlayouts = jactivity['layouts']
        parsed_layouts = [self.parse_layout(layout, activity_name) for layout in jlayouts]
        layout_strings = list(filter(self.is_not_blank, self.flatten(parsed_layouts)))
        # parse disconnected fragments XXX: we assume these fragments are replaced by each other and don't appear together
        orphaned_fragments = jactivity.get('orphanedFragments', [])
        orphaned_fragment_strings = list(filter(self.is_not_blank, self.flatten([self.parse_fragment(fragment, activity_name) for fragment in orphaned_fragments])))
        # combine extracted texts
        if len(orphaned_fragment_strings) == 0:
            res_text = layout_strings
        elif len(layout_strings) == 0:
            res_text = orphaned_fragment_strings
        else:
            res_text = [f" {self.sentence_delimiter} ".join(item) for item in itertools.product(layout_strings, orphaned_fragment_strings)]
        return activity_name, titles_string, res_text

    def parse_menu_items(self, items, activity_name):
        for item in items:
            guid = item.get('guid', '')
            title = item.get('title', '')
            mlabel = self.get_value(title)
            self.ui[guid].label = mlabel
            self.ui[guid].activity = activity_name
            self.ui[guid].clazz = "android.view.MenuItem"
            self.ui[guid].is_menu = True
            if 'subMenu' in item:
                self.parse_menu_items(item['subMenu'], activity_name)

    def parse_layout(self, layout, activity_name):
        """ returns string or list of strings
        """
        return self.parse_view(layout, activity_name)

    @staticmethod
    def is_fragment(view):
        return 'fragmentClass' in view

    def parse_view(self, view, activity_name):
        if self.is_fragment(view):  # fragment container
            return self.parse_fragment(view, activity_name)
        view_class = view['viewClass']
        vid = view['id']
        guid = view['guid']
        id_variable_name = view.get('idVariable', '')
        text_attrs = view.get('textAttributes', [])
        view_text = self.collect_text(text_attrs)
        input_type = view.get('inputTypes', [])
        other_attrs = view.get('otherAttributes', [])
        listeners = view.get('listeners', [])
        icon = self.get_icon(other_attrs)
        children = view.get('children', [])
        self.activities[activity_name].views.append(guid)
        self.ui[guid].activity = activity_name
        self.ui[guid].clazz = view_class
        self.ui[guid].listeners = listeners
        self.ui[guid].var_name = id_variable_name
        self.ui[guid].icon = icon

        if not children:
            self.ui[guid].label = view_text
            return view_text
        children_text_items = [self.parse_view(child, activity_name) for child in children if not self.is_fragment(child)]
        child_fragments = list(filter(self.is_fragment, children))
        if len(child_fragments) > 0:
            child_fragments_text = self.flatten(self.parse_fragment(fragment, activity_name) for fragment in child_fragments)  # list of lists
            children_text_items.extend(child_fragments_text)
        children_text_items.append(self.sanitize(view_text))
        children_partial_text = f" {self.sentence_delimiter} ".join(filter(lambda text: isinstance(text, str) and self.is_not_blank(text), children_text_items))
        children_alternatives = list(filter(lambda text: isinstance(text, list), children_text_items))
        if len(children_alternatives) == 0:
            children_text = children_partial_text  # string
        else:
            children_alternative_texts = [f" {self.sentence_delimiter} ".join(item) for item in itertools.product(*children_alternatives)]
            children_text = [f"{children_partial_text}{self.sentence_delimiter} {item}" for item in children_alternative_texts]  # list
        self.ui[guid].label = children_text
        return children_text

    def collect_text(self, text_attrs):
        """
        collect non-empty labels from one button into one line, separated with a `delimiter`
        """
        return self.concat_labels(more_itertools.unique_everseen(map(lambda t: self.parse_text(t), text_attrs)))

    def parse_fragment(self, fragment, activity_name):
        layouts = fragment.get('layouts')
        if not layouts:
            return []
        return list(self.flatten([self.parse_layout(layout, activity_name) for layout in layouts]))  # list of strings

    @staticmethod
    def is_not_blank(string):
        return string is not None and len(string) > 0 and not string.isspace()

    @classmethod
    def parse_text(cls, text_attr):
        if type(text_attr) == dict:
            name = text_attr['name']
            value = text_attr['value']
            if name in cls.text_attributes:
                return value
            else:
                return
        else:
            return text_attr

    @staticmethod
    def get_value(attr):
        if type(attr) == dict:
            return attr.get('value', '')
        else:
            return attr

    @staticmethod
    def parse_orphaned_fragments(self, fragment):
        return self.parse_view(fragment)

    def is_normal(self):
        return self.meta.get('type', 'NORMAL') == 'NORMAL'

    @staticmethod
    def flatten(items):
        for x in items:
            if isinstance(x, Iterable) and not isinstance(x, (str, bytes)):
                for sub_x in x:
                    yield sub_x
            else:
                yield x

    def parse_dialog(self, dialog, activity_name):
        titles = dialog.get("titles", [])
        titles = [self.sanitize(title) for title in titles]
        title_string = self.concat_labels(titles)
        messages = dialog.get("messages", [])
        messages = [self.sanitize(message) for message in messages]
        message_string = self.concat_labels(messages)
        dialog_string = self.concat_labels([title_string, message_string])
        buttons = dialog.get("buttons", [])
        for button in buttons:
            view = View()
            guid = button.get("guid", "")
            view.label = button.get("label", "")
            view.listener = button.get("listener", "")
            view.activity = activity_name
            view.context = dialog_string
            view.clazz = "DialogButton"
            view.is_dialog = True
            if guid != "":
                self.ui[guid] = view
            else:
                self.dialogs[view.listener] = view  # in case of old analysis version

    @staticmethod
    def is_content_url(perm_name):
        return perm_name.startswith()

    @staticmethod
    def is_start_service(api):
        return "startService" in api

    def load_api_service_lifecycle(self, service, sensitive_only: bool):
        service_apis = []
        name = service['name']
        for api in service['api']:
            service_api = self.parse_api_call(api, sensitive_only, name)
            service_apis.extend(service_api)
        self.services[name] = service_apis

    def load_api_broadcast(self, broadcast, sensitive_only=False):
        name = broadcast['name']
        apis = broadcast['api']
        intents = broadcast.get('intent', [])
        broadcast_apis = []
        for api in apis:
            broadcast_api = self.parse_api_call(api, sensitive_only, name)
            broadcast_apis.extend(broadcast_api)
        broadcast = Broadcast(name, intents, broadcast_apis)
        self.broadcasts[name] = broadcast

    def parse_api_call(self, api, sensitive_only, source) -> list:
        if type(api) == dict:  # we have an api with metadata: either a ContentResolver or startService
            if self.is_start_service(api):
                if source:
                    self.service_mapping[source].append(api['uri'])
                return []  # FIXME
            else:
                uris = api['uri'].split('|')  # may be several uris or parts of one uri
                api_uri = []
                for uri in uris:  # use uri with of ContentResolver methods: query, insert
                    uri = self.filter_api(uri, sensitive_only)
                    uri = self.sanitize_uri(uri)
                    if uri:
                        api_uri.append(f"{uri}|{api['method']}")
                return api_uri
        else:
            api = self.filter_api(api, sensitive_only)
            return [api] if api else []

    def load_api_activity_lifecycle(self, activity, sensitive_only: bool):
        lc_apis = []
        name = activity['name']
        for api in activity['api']:
            lc_api = self.parse_api_call(api, sensitive_only, name)
            lc_apis.extend(lc_api)
        self.lifecycle[name] = lc_apis

    def get_icon(self, attrs):
        if 'src' in attrs:
            return self.get_attr_value(attrs['src'])
        if 'background' in attrs:
            return self.get_attr_value(attrs['background'])
        return ""

    @staticmethod
    def get_attr_value(attr):
        if type(attr) == dict:
            return attr['value']
        else:
            return attr

    @staticmethod
    def load_actions() -> set:
        with open('actions.list') as fd:
            actions = {line.strip() for line in fd.readlines() if not line.startswith("#")}
        return actions

    @staticmethod
    def sanitize_uri(uri: str):
        return uri.replace(":", "=")
