<commit_before>###########################################################################
# Copyright 2017 ZT Prentner IT GmbH (www.ztp.at)
# 
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# 
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
# 
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
###########################################################################

"""
This module provides a function to generate a DEP and crypto container
according to a JSON test specification.
"""

from __future__ import unicode_literals
from builtins import int
from builtins import range

from .gettext_helper import _

import datetime

from . import cashreg
from . import depexport
from . import key_store
from . import sigsys
from . import utils

def runTest(spec, keymat, closed=False, tcSize=None):
    """
    Creates a DEP and a crypto container structure according to the given
    test specification. In addition to the specification elements that the
    reference implementation uses, this function also understands the
    "decimalSerial", "turnoverCounterSize", "includePublicKey",
    "multipleGroups", "certChainLength", "omitSignCert", "omitRootCert",
    "certChainFailure" and "certChainSerialCollision" elements in the root
    dictionary and the "annotateTurnoverCounter", "override" and
    "beginNewDEP" elements in the dictionaries in the
    "cashBoxInstructionList" element.
    :param spec: The test specification as a dict structure.
    :param keymat: The key material as a list of tuples with the public key/
    certificate in the first element and the private key in the second
    element. The length of the list must be equal to the number of signature
    devices in the test specification. For a closed system public keys must
    be used, for an open system certificates must be used.
    :param closed: Indicates whether the system is a closed system (True) or
    an open system (False).
    :param tcSize: The size of the turnover counter in bytes. If this is
    omitted, the size is read from the specification. If the size is not
    set in the specification, 8 bytes are used.
    :return: A dict structure of the DEP and a dict structure of the crypto
    container.
    """
    if len(keymat) != spec['numberOfSignatureDevices']:
        raise Exception(_('Need key material for {} signature devices, got {} key pairs.').format(
        spec['numberOfSignatureDevices'], len(keymat)))

    key = utils.loadB64Key(spec['base64AesKey'].encode('utf-8'))

    turnoverCounterSize = spec.get('turnoverCounterSize', 8)
    if tcSize:
        turnoverCounterSize = tcSize

    register = cashreg.CashRegister(spec['cashBoxId'], None,
            int(0.0 * 100), key, turnoverCounterSize)

    keyStore = key_store.KeyStore()

    zda = 'AT0' if closed else 'AT77'

    doGroups = spec.get('multipleGroups', False)
    chainLength = spec.get('certChainLength',
            [0] * spec['numberOfSignatureDevices'])
    omitSignCert = spec.get('omitSignCert',
            [False] * spec['numberOfSignatureDevices'])
    omitRootCert = spec.get('omitRootCert',
            [False] * spec['numberOfSignatureDevices'])
    certChainFailure = spec.get('certChainFailure',
            [0] * spec['numberOfSignatureDevices'])
    certChainSerialCollision = spec.get('certChainSerialCollision',
            [False] * spec['numberOfSignatureDevices'])

    sigsBroken = list()
    sigsWorking = list()
    groupCerts = list()
    for i in range(spec['numberOfSignatureDevices']):
        serial = None
        certList = list()
        privObj = utils.loadPrivKey(keymat[i][1])
        if closed:
            serial = "%s-K%d" % (spec['companyID'], i)
            pubObj = utils.loadPubKey(keymat[i][0])
            cserial = utils.makeCertSerial()
            certList.append(utils.makeSignedCert(pubObj, serial, 365,
                cserial, privObj))
        else:
            certList.append(utils.loadCert(keymat[i][0]))
            numSerial = certList[-1].serial
            for j in range(chainLength[i], 0, -1):
                s, p = utils.makeES256Keypair()
                numSerial = utils.makeCertSerial()
                if j == certChainFailure[i]:
                    privObj = s
                c = utils.makeSignedCert(p, 'intermediate {} in {}'.format(
                    j, i), 365, numSerial, privObj, certList[0])
                privObj = s
                certList.insert(0, c)

            if spec.get('decimalSerial', False):
                serial = ('%d' % abs(numSerial))
            else:
                serial = key_store.numSerialToKeyId(numSerial)

        if (not closed or doGroups) and not omitRootCert[i]:
            if certChainSerialCollision[i]:
                s, p = utils.makeES256Keypair()
                tlCert = utils.makeSignedCert(p, 'fake root {}'.format(i),
                        365, certList[0].serial, s)
            else:
                tlCert = certList[-1]
            certPEM = utils.addPEMCertHeaders(
                    utils.exportCertToPEM(tlCert))
            keyStore.putPEMCert(certPEM)

        if closed or spec.get('includePublicKey', False):
            kid = "%s-K%d" % (spec['companyID'], i)
            keyStore.putKey(kid, certList[0].public_key(), None)

        sigB = sigsys.SignatureSystemBroken(zda, serial)
        sigW = sigsys.SignatureSystemWorking(zda, serial, privObj)
<commit_msg>Make run_test generate different ZDAs in the same DEP<commit_after>###########################################################################
# Copyright 2017 ZT Prentner IT GmbH (www.ztp.at)
# 
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# 
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
# 
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
###########################################################################

"""
This module provides a function to generate a DEP and crypto container
according to a JSON test specification.
"""

from __future__ import unicode_literals
from builtins import int
from builtins import range

from .gettext_helper import _

import datetime
import random

from . import cashreg
from . import depexport
from . import key_store
from . import sigsys
from . import utils

def runTest(spec, keymat, closed=False, tcSize=None):
    """
    Creates a DEP and a crypto container structure according to the given
    test specification. In addition to the specification elements that the
    reference implementation uses, this function also understands the
    "decimalSerial", "turnoverCounterSize", "includePublicKey",
    "multipleGroups", "certChainLength", "omitSignCert", "omitRootCert",
    "certChainFailure" and "certChainSerialCollision" elements in the root
    dictionary and the "annotateTurnoverCounter", "override" and
    "beginNewDEP" elements in the dictionaries in the
    "cashBoxInstructionList" element.
    :param spec: The test specification as a dict structure.
    :param keymat: The key material as a list of tuples with the public key/
    certificate in the first element and the private key in the second
    element. The length of the list must be equal to the number of signature
    devices in the test specification. For a closed system public keys must
    be used, for an open system certificates must be used.
    :param closed: Indicates whether the system is a closed system (True) or
    an open system (False).
    :param tcSize: The size of the turnover counter in bytes. If this is
    omitted, the size is read from the specification. If the size is not
    set in the specification, 8 bytes are used.
    :return: A dict structure of the DEP and a dict structure of the crypto
    container.
    """
    if len(keymat) != spec['numberOfSignatureDevices']:
        raise Exception(_('Need key material for {} signature devices, got {} key pairs.').format(
        spec['numberOfSignatureDevices'], len(keymat)))

    key = utils.loadB64Key(spec['base64AesKey'].encode('utf-8'))

    turnoverCounterSize = spec.get('turnoverCounterSize', 8)
    if tcSize:
        turnoverCounterSize = tcSize

    register = cashreg.CashRegister(spec['cashBoxId'], None,
            int(0.0 * 100), key, turnoverCounterSize)

    keyStore = key_store.KeyStore()

    doGroups = spec.get('multipleGroups', False)
    chainLength = spec.get('certChainLength',
            [0] * spec['numberOfSignatureDevices'])
    omitSignCert = spec.get('omitSignCert',
            [False] * spec['numberOfSignatureDevices'])
    omitRootCert = spec.get('omitRootCert',
            [False] * spec['numberOfSignatureDevices'])
    certChainFailure = spec.get('certChainFailure',
            [0] * spec['numberOfSignatureDevices'])
    certChainSerialCollision = spec.get('certChainSerialCollision',
            [False] * spec['numberOfSignatureDevices'])

    sigsBroken = list()
    sigsWorking = list()
    groupCerts = list()
    for i in range(spec['numberOfSignatureDevices']):
        serial = None
        certList = list()
        privObj = utils.loadPrivKey(keymat[i][1])
        if closed:
            serial = "%s-K%d" % (spec['companyID'], i)
            pubObj = utils.loadPubKey(keymat[i][0])
            cserial = utils.makeCertSerial()
            certList.append(utils.makeSignedCert(pubObj, serial, 365,
                cserial, privObj))
        else:
            certList.append(utils.loadCert(keymat[i][0]))
            numSerial = certList[-1].serial
            for j in range(chainLength[i], 0, -1):
                s, p = utils.makeES256Keypair()
                numSerial = utils.makeCertSerial()
                if j == certChainFailure[i]:
                    privObj = s
                c = utils.makeSignedCert(p, 'intermediate {} in {}'.format(
                    j, i), 365, numSerial, privObj, certList[0])
                privObj = s
                certList.insert(0, c)

            if spec.get('decimalSerial', False):
                serial = ('%d' % abs(numSerial))
            else:
                serial = key_store.numSerialToKeyId(numSerial)

        if (not closed or doGroups) and not omitRootCert[i]:
            if certChainSerialCollision[i]:
                s, p = utils.makeES256Keypair()
                tlCert = utils.makeSignedCert(p, 'fake root {}'.format(i),
                        365, certList[0].serial, s)
            else:
                tlCert = certList[-1]
            certPEM = utils.addPEMCertHeaders(
                    utils.exportCertToPEM(tlCert))
            keyStore.putPEMCert(certPEM)

        if closed or spec.get('includePublicKey', False):
            kid = "%s-K%d" % (spec['companyID'], i)
            keyStore.putKey(kid, certList[0].public_key(), None)

        zda = 'AT0' if closed else 'AT{}'.format(random.randint(1, 999))
        sigB = sigsys.SignatureSystemBroken(zda, serial)
        sigW = sigsys.SignatureSystemWorking(zda, serial, privObj)

{'fraction_non_alphanumeric': 0.11515814266487213, 'fraction_numerical': 0.008243606998654104, 'mean_word_length': 2.917298187808896, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, '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': '724822', 'n_tokens_mistral': 3353, 'n_tokens_neox': 3205, 'n_words': 1099}