parse entries from MOZ_LOG CSMLog entries

csmlogparser.py

#!/usr/bin/env python3

import yaml
import sys
from functools import reduce


def lines_into_blocks():
    consume_into_block = False
    block = []
    for line in sys.stdin:
        if line == '\n':
            continue
        if line == '#DebugDoContentSecurityCheck Begin\n':
            consume_into_block = True
            continue
        elif line == '#DebugDoContentSecurityCheck End\n':
            consume_into_block = False
            yield block
            block = []
            continue
        # NOTE these lines can appear and in between begin/end blocks, because multi-process/threading. ugh :)
        if line == 'Allowing eval() with System Principal because the containing file is in the allowlist\n':
            continue
        if consume_into_block:
            block.append(line)


def blocks_to_entries():
    for block in lines_into_blocks():
        try:
            parsed = yaml.safe_load("\n".join(block))
        except Exception as e:
            print('Error "{}" while parsing block "{}", skipped.'.format(
                e, repr(block)))
            continue
        if not 'doContentSecurityCheck' in parsed:
            continue
        # values is a list of key, value pairs
        values = parsed['doContentSecurityCheck']
        # mapping all values to a single dict
        entry = reduce(lambda x, y: {**x, **y}, values)
        # TODO (above):
        #  - with python 3.9, we can use `lambda x, y : x|y` to merge two dicts
        #  - see https://www.python.org/dev/peps/pep-0584/#motivation
        yield entry


def filter_entries():
    for e in blocks_to_entries():
        # NOTE implement your filtering stuff here, above is all parsing
        if e['loadingPrincipal'] == 'SystemPrincipal' or e['triggeringPrincipal'] == 'SystemPrincipal':
            if e['externalContentPolicyType'] == 'TYPE_STYLESHEET' or e['internalContentPolicyType'] == 'TYPE_STYLESHEET':
                print(e)


if __name__ == '__main__':
    filter_entries()