<commit_before># This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
# Copyright (c) 2015 Mozilla Corporation
#
# Contributors:
# <NAME> <EMAIL>
# <NAME> <EMAIL>

import hashlib
import sys
import unittest

class message(object):
    def __init__(self):
        self.registration = ['rra']
        self.priority = 20

    def validate_field(self, field, keys):
        for k in keys:
            if not k in field.keys():
                sys.stderr.write('warning: key {0} not in expected list {1}\n'.format(k, field.keys()))
                return False
        return True

    def validate(self, message):
        if not self.validate_field(message, ['details', 'utctimestamp', 'summary', 'source', 'lastmodified']):
            return False
        if not self.validate_field(message['details'], ['data', 'risk', 'metadata']):
            return false
        if not self.validate_field(message['details']['metadata'], ['operator', 'scope', 'owner', 'developer',
            'service', 'description']):
            return False
        if not self.validate_field(message['details']['data'], ['Unknown', 'PUBLIC', 'INTERNAL', 'RESTRICTED', 'SECRET',
            'default', 'CONFIDENTIAL INTERNAL', 'CONFIDENTIAL RESTRICTED', 'CONFIDENTIAL SECRET']):
            return False
        if not self.validate_field(message['details']['risk'], ['integrity', 'confidentiality', 'availability']):
            return False
        for cia in ['integrity', 'confidentiality', 'availability']:
            for rpf in ['productivity', 'finances', 'reputation']:
                if not self.validate_field(message['details']['risk'][cia][rpf], ['rationale', 'impact', 'probability']):
                    return False

        if (len(message['details']['metadata']['service']) == 0):
            sys.stderr.write('warning: no service name for {0}\n'.format(message['source']))
            return False
        if (message['details']['metadata']['service'] == 'RRA for '):
            sys.stderr.write('warning: Invalid service name for {0}\n'.format(message['source']))
            return False

        return True

    def calculate_id(self, message):
        s = '{0}|{1}'.format(message['source'], message['lastmodified'])
        return hashlib.sha256(s.encode('ascii')).hexdigest()

    def onMessage(self, message, metadata):
        if metadata['doc_type'] != 'rra':
            return (message, metadata)
        if not self.validate(message):
            sys.stderr.write('error: invalid format for RRA {0}\n'.format(message))
            return (None, None)
        metadata['id'] = self.calculate_id(message)
        metadata['doc_type'] = 'rra_state'<commit_msg>Support more validation filters to accomodate different RRA fields. This enhance the validation accuracy ;-)<commit_after># This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
# Copyright (c) 2015 Mozilla Corporation
#
# Contributors:
# <NAME> <EMAIL>
# <NAME> <EMAIL>

import hashlib
import sys
import unittest

class message(object):
    def __init__(self):
        self.registration = ['rra']
        self.priority = 20

    def validate_field_subset(self, field, keys):
        '''are all "field" part of "keys? (there could be more "keys" than "fields" but all "fields must be part of "keys")'''
        f = set(field.keys())
        k = set(keys)
        if (not f.issubset(k)):
            sys.stderr.write('warning: input RRA "{0}" must only be part of these required fields: "{1}"\n'.format(f.__str__(), k.__str__()))
            return False
        return True

    def validate_field_superset(self, field, keys):
        '''do we have at least all "keys" in "field?"'''
        f = set(field.keys())
        k = set(keys)
        if (not f.issuperset(k)):
            sys.stderr.write('warning: input RRA "{0}" does not include at least these required fields: "{1}"\n'.format(f.__str__(), k.__str__()))
            return False
        return True

    def validate_field_equal(self, field, keys):
        '''keys must == field"'''
        f = set(field.keys())
        k = set(keys)
        if (f != k):
            sys.stderr.write('warning: input RRA "{0}" does not match: "{1}"\n'.format(f.__str__(), k.__str__()))
            return False
        return True

    def validate(self, message):
        if not self.validate_field_superset(message, ['details', 'utctimestamp', 'summary', 'source', 'lastmodified']):
            return False
        if not self.validate_field_superset(message['details'], ['data', 'risk', 'metadata']):
            return False
        if not self.validate_field_superset(message['details']['metadata'], ['operator', 'scope', 'owner', 'developer',
            'service', 'description']):
            return False
        if not self.validate_field_subset(message['details']['data'], ['Unknown', 'PUBLIC', 'INTERNAL', 'RESTRICTED', 'SECRET',
            'default', 'CONFIDENTIAL INTERNAL', 'CONFIDENTIAL RESTRICTED', 'CONFIDENTIAL SECRET']):
            return False
        if not self.validate_field_equal(message['details']['risk'], ['integrity', 'confidentiality', 'availability']):
            return False
        for cia in ['integrity', 'confidentiality', 'availability']:
            for rpf in ['productivity', 'finances', 'reputation']:
                if not self.validate_field_equal(message['details']['risk'][cia][rpf], ['rationale', 'impact', 'probability']):
                    return False

        if (len(message['details']['metadata']['service']) == 0):
            sys.stderr.write('warning: no service name for {0}\n'.format(message['source']))
            return False
        if (message['details']['metadata']['service'] == 'RRA for '):
            sys.stderr.write('warning: Invalid service name for {0}\n'.format(message['source']))
            return False

        return True

    def calculate_id(self, message):
        s = '{0}|{1}'.format(message['source'], message['lastmodified'])
        return hashlib.sha256(s.encode('ascii')).hexdigest()

    def onMessage(self, message, metadata):
        if metadata['doc_type'] != 'rra':
            return (message, metadata)
        if not self.validate(message):
            sys.stderr.write('error: invalid format for RRA {0}\n'.format(message))
            return (None, None)
        metadata['id'] = self.calculate_id(message)
        metadata['doc_type'] = 'rra_state'
{'fraction_non_alphanumeric': 0.15144230769230768, 'fraction_numerical': 0.006911057692307692, 'mean_word_length': 2.8907071887784923, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6465245', 'n_tokens_mistral': 1909, 'n_tokens_neox': 1841, 'n_words': 544}