util/design/lib/LcStEnc.py

# Copyright lowRISC contributors.
# Licensed under the Apache License, Version 2.0, see LICENSE for details.
# SPDX-License-Identifier: Apache-2.0
r"""Contains life cycle state encoding class which is
used to generate new life cycle encodings.
"""
import logging as log
import random
from collections import OrderedDict
from Crypto.Hash import cSHAKE128

from lib.common import (check_int, ecc_encode, get_hd, hd_histogram,
                        is_valid_codeword, random_or_hexvalue, scatter_bits)

# Seed diversification constant for LcStEnc (this enables to use
# the same seed for different classes)
LC_SEED_DIVERSIFIER = 1939944205722120255

# State types and permissible format for entries
# The format is index dependent, e.g. ['0', 'A1', 'B1'] for index 1
LC_STATE_TYPES = {
    'lc_state': ['0', 'A{}', 'B{}'],
    'lc_cnt': ['0', 'C{}', 'D{}']
}


def _is_incremental_codeword(word1, word2):
    '''Test whether word2 is incremental wrt word1.'''
    if len(word1) != len(word2):
        raise RuntimeError('Words are not of equal size')

    _word1 = int(word1, 2)
    _word2 = int(word2, 2)

    # This basically checks that the second word does not
    # clear any bits that are set to 1 in the first word.
    return ((_word1 & _word2) == _word1)


def _get_incremental_codewords(config, base_ecc, existing_words):
    '''Get all possible incremental codewords fulfilling the constraints.'''

    base_data = base_ecc[config['secded']['ecc_width']:]

    # We only need to spin through data bits that have not been set yet.
    # Hence, we first count how many bits are zero (and hence still
    # modifyable). Then, we enumerate all possible combinations and scatter
    # the bits of the enumerated values into the correct bit positions using
    # the scatter_bits() function.
    incr_cands = []
    free_bits = base_data.count('0')
    for k in range(1, 2**free_bits):
        # Get incremental dataword by scattering the enumeration bits
        # into the zero bit positions in base_data.
        incr_cand = scatter_bits(base_data,
                                 format(k, '0' + str(free_bits) + 'b'))
        incr_cand_ecc = ecc_encode(config, incr_cand)

        # Dataword is correct by construction, but we need to check whether
        # the ECC bits are incremental.
        if _is_incremental_codeword(base_ecc, incr_cand_ecc):
            # Check whether the candidate fulfills the maximum
            # Hamming weight constraint.
            if incr_cand_ecc.count('1') <= config['max_hw']:
                # Check Hamming distance wrt all existing words.
                for w in existing_words + [base_ecc]:
                    if get_hd(incr_cand_ecc, w) < config['min_hd']:
                        break
                else:
                    incr_cands.append(incr_cand_ecc)

    return incr_cands


def _get_new_state_word_pair(config, existing_words):
    '''Randomly generate a new incrementally writable word pair'''
    while 1:
        # Draw a random number and check whether it is unique and whether
        # the Hamming weight is in range.
        width = config['secded']['data_width']
        ecc_width = config['secded']['ecc_width']
        base = random.getrandbits(width)
        base = format(base, '0' + str(width) + 'b')
        base_cand_ecc = ecc_encode(config, base)
        # disallow all-zero and all-one states
        pop_cnt = base_cand_ecc.count('1')
        if pop_cnt >= config['min_hw'] and pop_cnt <= config['max_hw']:

            # Check Hamming distance wrt all existing words
            for w in existing_words:
                if get_hd(base_cand_ecc, w) < config['min_hd']:
                    break
            else:
                # Get encoded incremental candidates.
                incr_cands_ecc = _get_incremental_codewords(
                    config, base_cand_ecc, existing_words)
                # there are valid candidates, draw one at random.
                # otherwise we just start over.
                if incr_cands_ecc:
                    incr_cand_ecc = random.choice(incr_cands_ecc)
                    log.info('word {}: {}|{} -> {}|{}'.format(
                        int(len(existing_words) / 2),
                        base_cand_ecc[ecc_width:], base_cand_ecc[0:ecc_width],
                        incr_cand_ecc[ecc_width:], incr_cand_ecc[0:ecc_width]))
                    existing_words.append(base_cand_ecc)
                    existing_words.append(incr_cand_ecc)
                    return (base_cand_ecc, incr_cand_ecc)


def _validate_words(config, words):
    '''Validate generated words (base and incremental).'''
    for k, w in enumerate(words):
        # Check whether word is valid wrt to ECC polynomial.
        if not is_valid_codeword(config, w):
            raise RuntimeError('Codeword {} at index {} is not valid'.format(
                w, k))
        # Check that word fulfills the Hamming weight constraints.
        pop_cnt = w.count('1')
        if pop_cnt < config['min_hw'] or pop_cnt > config['max_hw']:
            raise RuntimeError(
                'Codeword {} at index {} has wrong Hamming weight'.format(
                    w, k))
        # Check Hamming distance wrt to all other existing words.
        # If the constraint is larger than 0 this implies uniqueness.
        if k < len(words) - 1:
            for k2, w2 in enumerate(words[k + 1:]):
                if get_hd(w, w2) < config['min_hd']:
                    raise RuntimeError(
                        'Hamming distance between codeword {} at index {} '
                        'and codeword {} at index {} is too low.'.format(
                            w, k, w2, k + 1 + k2))


def _validate_secded(config):
    '''Validate SECDED configuration'''
    config['secded'].setdefault('data_width', 0)
    config['secded'].setdefault('ecc_width', 0)
    config['secded'].setdefault('ecc_matrix', [[]])
    config['secded']['data_width'] = check_int(config['secded']['data_width'])
    config['secded']['ecc_width'] = check_int(config['secded']['ecc_width'])

    total_width = config['secded']['data_width'] + config['secded']['ecc_width']

    if config['secded']['data_width'] % 8:
        raise RuntimeError('SECDED data width must be a multiple of 8')

    if config['secded']['ecc_width'] != len(config['secded']['ecc_matrix']):
        raise RuntimeError('ECC matrix does not have correct number of rows')

    log.info('SECDED Matrix:')
    for i, l in enumerate(config['secded']['ecc_matrix']):
        log.info('ECC Bit {} Fanin: {}'.format(i, l))
        for j, e in enumerate(l):
            e = check_int(e)
            if e < 0 or e >= total_width:
                raise RuntimeError('ECC bit position is out of bounds')
            config['secded']['ecc_matrix'][i][j] = e


def _validate_constraints(config):
    '''Validates Hamming weight and distance constraints'''
    config.setdefault('min_hw', 0)
    config.setdefault('max_hw', 0)
    config.setdefault('min_hd', 0)
    config['min_hw'] = check_int(config['min_hw'])
    config['max_hw'] = check_int(config['max_hw'])
    config['min_hd'] = check_int(config['min_hd'])

    total_width = config['secded']['data_width'] + config['secded']['ecc_width']

    if config['min_hw'] >= total_width or \
       config['max_hw'] > total_width or \
       config['min_hw'] >= config['max_hw']:
        raise RuntimeError('Hamming weight constraints are inconsistent.')

    if config['max_hw'] - config['min_hw'] + 1 < config['min_hd']:
        raise RuntimeError('Hamming distance constraint is inconsistent.')


def _validate_tokens(config):
    '''Validates and hashes the tokens'''
    config.setdefault('token_size', 128)
    config['token_size'] = check_int(config['token_size'])
    # This needs to be byte aligned
    if config['token_size'] % 8:
        raise ValueError('Size of token {} must be byte aligned'
                         .format(token['name']))

    num_bytes = config['token_size'] // 8

    hashed_tokens = []
    for token in config['tokens']:
        random_or_hexvalue(token, 'value', config['token_size'])
        hashed_token = OrderedDict()
        hashed_token['name'] = token['name'] + 'Hashed'
        data = token['value'].to_bytes(num_bytes, byteorder='little')
        # Custom string chosen for life cycle KMAC App interface
        custom = 'LC_CTRL'.encode('UTF-8')
        hashobj = cSHAKE128.new(data=data, custom=custom)
        hashed_token['value'] = int.from_bytes(hashobj.read(num_bytes),
                                               byteorder='little')
        hashed_tokens.append(hashed_token)

    config['tokens'] += hashed_tokens


def _validate_state_declarations(config):
    '''Validates life cycle state and counter declarations'''
    for typ in LC_STATE_TYPES.keys():
        for k, state in enumerate(config[typ].keys()):
            if k == 0:
                config['num_' + typ + '_words'] = len(config[typ][state])
                log.info('Inferred {} = {}'.format(
                    'num_' + typ + '_words', config['num_' + typ + '_words']))
            if config['num_' + typ + '_words'] != len(config[typ][state]):
                raise RuntimeError(
                    '{} entry {} has incorrect length {}'.format(
                        typ, state, len(config[typ][state])))
            # Render the format templates above.
            for j, entry in enumerate(config[typ][state]):
                legal_values = [fmt.format(j) for fmt in LC_STATE_TYPES[typ]]
                if entry not in legal_values:
                    raise RuntimeError(
                        'Illegal entry "{}" found in {} of {}'.format(
                            entry, state, typ))


def _generate_words(config):
    '''Generate encoding words'''
    config['genwords'] = {}  # dict holding the word pairs for each state type
    existing_words = []  # temporary list of all words for uniqueness tests
    for typ in LC_STATE_TYPES.keys():
        config['genwords'][typ] = []
        for k in range(config['num_' + typ + '_words']):
            new_word = _get_new_state_word_pair(config, existing_words)
            config['genwords'][typ].append(new_word)

    # Validate words (this must not fail at this point).
    _validate_words(config, existing_words)

    # Calculate and store statistics
    config['stats'] = hd_histogram(existing_words)
    log.info('')
    log.info('Hamming distance histogram:')
    log.info('')
    for bar in config['stats']["bars"]:
        log.info(bar)
    log.info('')
    log.info('Minimum HD: {}'.format(config['stats']['min_hd']))
    log.info('Maximum HD: {}'.format(config['stats']['max_hd']))
    log.info('Minimum HW: {}'.format(config['stats']['min_hw']))
    log.info('Maximum HW: {}'.format(config['stats']['max_hw']))


class LcStEnc():
    '''Life cycle state encoding generator class

    The constructor expects the parsed configuration
    hjson to be passed in.
    '''

    # This holds the config dict.
    config = {}

    def __init__(self, config):
        '''The constructor validates the configuration dict.'''

        log.info('')
        log.info('Generate life cycle state')
        log.info('')

        if 'seed' not in config:
            raise RuntimeError('Missing seed in configuration')
        if 'secded' not in config:
            raise RuntimeError('Missing secded configuration')
        if 'tokens' not in config:
            raise RuntimeError('Missing token configuration')

        for typ in LC_STATE_TYPES.keys():
            if typ not in config:
                raise RuntimeError('Missing {} definition'.format(typ))

        config['seed'] = check_int(config['seed'])
        log.info('Seed: {0:x}'.format(config['seed']))
        log.info('')

        # Re-initialize with seed to make results reproducible.
        random.seed(LC_SEED_DIVERSIFIER + int(config['seed']))

        log.info('Checking SECDED.')
        _validate_secded(config)
        log.info('')
        log.info('Checking Hamming weight and distance constraints.')
        _validate_constraints(config)
        log.info('')
        log.info('Hashing tokens.')
        _validate_tokens(config)
        log.info('')
        log.info('Checking state declarations.')
        _validate_state_declarations(config)
        log.info('')
        log.info('Generate incremental word encodings.')
        _generate_words(config)

        self.config = config

        log.info('')
        log.info('Successfully generated life cycle state.')
        log.info('')

    def encode(self, name, state):
        '''Look up state encoding and return as integer value'''

        data_width = self.config['secded']['data_width']
        ecc_width = self.config['secded']['ecc_width']

        if name not in LC_STATE_TYPES:
            raise RuntimeError('Unknown state type {}'.format(name))

        if state not in self.config[name]:
            raise RuntimeError('Unknown state {} of type {}'.format(
                state, name))

        # Assemble list of state words
        words = []
        for j, entry in enumerate(self.config[name][state]):
            # This creates an index lookup table
            val_idx = {
                fmt.format(j): i
                for i, fmt in enumerate(LC_STATE_TYPES[name])
            }
            idx = val_idx[entry]
            if idx == 0:
                words.append(0)
            else:
                # Only extract data portion, discard ECC portion
                word = self.config['genwords'][name][j][idx - 1][ecc_width:]
                words.append(int(word, 2))

        # Convert words to one value
        outval = 0
        for k, word in enumerate(words):
            outval += word << (data_width * k)

        return outval