Merge pull request #202 from Crypto-TII/develop

Develop

claasp/cipher.py

@@ -39,6 +39,8 @@
 
 
 class Cipher:
+
+
     def __init__(self, family_name, cipher_type, cipher_inputs,
                  cipher_inputs_bit_size, cipher_output_bit_size,
                  cipher_reference_code=None):
@@ -146,6 +148,8 @@ def __init__(self, family_name, cipher_type, cipher_inputs,
         self._id = self.make_cipher_id()
         self._file_name = self.make_file_name()
 
+    def __repr__(self):
+        return self.id
     def _are_there_not_forbidden_components(self, forbidden_types, forbidden_descriptions):
         return self._rounds.are_there_not_forbidden_components(forbidden_types, forbidden_descriptions)
 

claasp/cipher_modules/algebraic_tests.py

@@ -30,22 +30,36 @@ class AlgebraicTests:
             sage: from claasp.ciphers.toys.toyspn1 import ToySPN1
             sage: toyspn = ToySPN1(number_of_rounds=2)
             sage: alg_test = AlgebraicTests(toyspn)
-            sage: alg_test.algebraic_tests(30) # timeout=30 seconds
+            sage: alg_test.algebraic_tests(timeout_in_seconds=10)
+            {'input_parameters': {'cipher.id': 'toyspn1_p6_k6_o6_r2',
+              'timeout_in_seconds': 10,
+              'test_name': 'algebraic_tests'},
+             'test_results': {'number_of_variables': [66, 126],
+              'number_of_equations': [76, 158],
+              'number_of_monomials': [96, 186],
+              'max_degree_of_equations': [2, 2],
+              'test_passed': [False, True]}}
 
+            sage: from claasp.cipher_modules.algebraic_tests import AlgebraicTests
             sage: from claasp.ciphers.block_ciphers.speck_block_cipher import SpeckBlockCipher
             sage: speck = SpeckBlockCipher(number_of_rounds=1)
             sage: alg_test = AlgebraicTests(speck)
-            sage: alg_test.algebraic_tests(60) # timeout=60 seconds
+            sage: alg_test.algebraic_tests(timeout_in_seconds=30)
+            {'input_parameters': {'cipher.id': 'speck_p32_k64_o32_r1',
+              'timeout_in_seconds': 30,
+              'test_name': 'algebraic_tests'},
+             'test_results': {'number_of_variables': [320],
+              'number_of_equations': [272],
+              'number_of_monomials': [365],
+              'max_degree_of_equations': [2],
+              'test_passed': [True]}}
 
-            sage: speck = SpeckBlockCipher(number_of_rounds=2)
-            sage: alg_test = AlgebraicTests(speck)
-            sage: alg_test.algebraic_tests(60)
     """
 
     def __init__(self, cipher):
         self._cipher = cipher
 
-    def algebraic_tests(self, timeout=60):
+    def algebraic_tests(self, timeout_in_seconds=60):
         from sage.structure.sequence import Sequence
         nvars_up_to_round = []
 
@@ -66,23 +80,20 @@ def algebraic_tests(self, timeout=60):
             nmonomials_up_to_round.append(Fseq.nmonomials())
             max_deg_of_equations_up_to_round.append(Fseq.maximal_degree())
 
-            if tests_up_to_round and tests_up_to_round[-1] is True:
-                tests_up_to_round.append(True)
-            else:
-                from cysignals.alarm import alarm, cancel_alarm
-
-                try:
-                    alarm(timeout)
-                    cancel_alarm()
-                    result = False
-                except InterruptedError:
-                    result = True
+            from cysignals.alarm import alarm, cancel_alarm, AlarmInterrupt
+            try:
+                alarm(timeout_in_seconds)
+                Fseq.groebner_basis()
+                cancel_alarm()
+                result = False
+            except AlarmInterrupt:
+                result = True
 
-                tests_up_to_round.append(result)
+            tests_up_to_round.append(result)
 
         input_parameters = {
-            "cipher.id": self._cipher.id,
-            "timeout": timeout,
+            "cipher": self._cipher,
+            "timeout_in_seconds": timeout_in_seconds,
             "test_name": "algebraic_tests"
         }
         test_results = {

claasp/cipher_modules/avalanche_tests.py

@@ -87,6 +87,7 @@ def avalanche_tests(self, number_of_samples=5, avalanche_dependence_uniform_bias
                                                                          avalanche_dependence_uniform_bias)
         intermediate_output_names = self._add_intermediate_output_components_id_to_dictionary(self._cipher.get_all_components())
         diffusion_tests = {"input_parameters": {
+            "cipher": self._cipher,
             "test_name": "avalanche_tests",
             "number_of_samples": number_of_samples,
             "avalanche_dependence_uniform_bias": avalanche_dependence_uniform_bias,

claasp/cipher_modules/component_analysis_tests.py

@@ -69,7 +69,8 @@ def component_analysis_tests(self):
 
         output_dictionary = {
             'input_parameters': {
-                'test_name': 'component_analysis'
+                'test_name': 'component_analysis',
+                'cipher': self._cipher
             },
             'test_results': components_analysis
         }
@@ -153,15 +154,19 @@ def print_component_analysis_as_radar_charts(self, results=None):
         plt.rcParams['figure.figsize'] = [20, 20]
 
         # remove XOR from results
-        results_without_xor = [results[i] for i in range(len(results)) if results[i]["description"][0] != "XOR"]
-        results = self._remove_components_with_strings_as_values(results_without_xor)
+        # results_without_xor = [results[i] for i in range(len(results)) if results[i]["description"][0] != "XOR"]
+        results_without_fsr = [results[i] for i in range(len(results)) if results[i]["type"] != "fsr"]
+        # removed for now because the fsr dictionary does not follow the standard structure as the other components:
+        # the keys properties, values, etc are not present.
+        # results = self._remove_components_with_strings_as_values(results_without_xor)
+        results = self._remove_components_with_strings_as_values(results_without_fsr)
 
         nb_plots = len(results)
         col = 2
         row = nb_plots // col
         if nb_plots % col != 0:
             row += nb_plots % col
-        positions = {8: -0.7, 3: -0.4}
+        positions = {8: -0.7, 3: -0.4} # positions of the text according to the numbers of properties
 
         for plot_number in range(nb_plots):
             categories = list(results[plot_number]["properties"].keys())
@@ -211,10 +216,12 @@ def print_component_analysis_as_radar_charts(self, results=None):
             self._fill_area(ax, categories, plot_number, positions, results)
 
         # Show the graph
-        plt.subplots_adjust(left=0.25, bottom=0.1, right=0.7, top=0.95, wspace=0, hspace=0.96)
+        if nb_plots >= 5:
+            plt.subplots_adjust(left=0.02, bottom=0.1, right=0.7, top=0.95, wspace=1, hspace=0.96)
+        else:
+            plt.subplots_adjust(left=0.09, bottom=0.3, right=0.7, top=0.7, wspace=1, hspace=0.96)
         plt.show()
         #print("The radar chart can be plot with the build-in method plt.show()")
-
         #return plt
 
 
@@ -274,7 +281,7 @@ def _select_boolean_function(self, component, boolean_polynomial_ring):
         elif component.description[0] == "MODADD":
             return self._MODADD_as_boolean_function(component, boolean_polynomial_ring)
         else:
-            return "TODO(...)"
+            return f"TODO: {component.id} not implemented yet"
 
 
     def _MODADD_as_boolean_function(self, component, boolean_polynomial_ring):
@@ -425,11 +432,8 @@ def _select_properties_function(self, boolean_polynomial_ring, operation):
         if component.type == 'fsr':
             return self._fsr_properties(operation)
 
-        if component.type == WORD_OPERATION:
-            print(f"TODO : {component.description[0]}")
-            return {}
         else:
-            print(f"TODO : {component.type}")
+            # print(f"TODO: not implemented yet")
             return {}
 
     def _is_mds(self, component):
@@ -485,7 +489,6 @@ def _word_operation_properties(self, operation, boolean_polynomial_ring):
         INPUT:
 
         - ``operation`` -- **list**; a list containing:
-
           * a component with the operation under study
           * number of occurrences of the operation
           * list of ids of all the components with the same underlying operation
@@ -642,7 +645,7 @@ def _linear_layer_properties(self, operation):
             "min_possible_value": 1,
             "max_possible_value": pow(2, component.input_bit_size) - 1
         }
-        if component.input_bit_size <= 32:
+        if component.input_bit_size <= 64:
             dictio["properties"]["differential_branch_number"] = {"value": branch_number(component, 'differential', 'bit'),
                                                                   "min_possible_value": 0,
                                                                   "max_possible_value": component.input_bit_size}
@@ -883,7 +886,8 @@ def _fill_area(self, ax, categories, plot_number, positions, results):
                 text += f"{category} = {int(results[plot_number]['properties'][category]['value'])} " \
                         f"(best is {results[plot_number]['properties'][category]['max_possible_value']}, " \
                         f"worst is {results[plot_number]['properties'][category]['min_possible_value']})\n"
-        plt.text(0, positions[len(categories)], text, transform=ax.transAxes, size="small")
+        # plt.text(0, positions[len(categories)], text, transform=ax.transAxes, size="small")
+        plt.text(2, 0, text, transform=ax.transAxes, size="small")
 
     def _initialise_spider_plot(self, plot_number, results):
         is_component_word_operation = results[plot_number]["type"] == "word_operation"

claasp/cipher_modules/continuous_diffusion_analysis.py

@@ -440,6 +440,7 @@ def continuous_diffusion_tests(self,
         """
         continuous_diffusion_tests = {"input_parameters": {
             'test_name': 'continuous_diffusion_tests',
+            'cipher': self.cipher,
             'continuous_avalanche_factor_number_of_samples': continuous_avalanche_factor_number_of_samples,
             'threshold_for_avalanche_factor': threshold_for_avalanche_factor,
             'continuous_neutral_measure_beta_number_of_samples': continuous_neutral_measure_beta_number_of_samples,

claasp/cipher_modules/models/algebraic/algebraic_model.py

@@ -200,11 +200,15 @@ def polynomial_system_at_round(self, r):
             component_type = component.type
             operation = component.description[0]
             component_types = ["sbox", "linear_layer", "mix_column", "constant"]
-            operations = ["XOR", "AND", "OR", "SHIFT", "ROTATE", "NOT"]
+            operations = ["XOR", "AND", "OR", "SHIFT", "ROTATE", "NOT", "MODADD", "MODSUB"]
 
             if component_type in component_types or (component_type == "word_operation" and operation in operations):
                 polynomials += component.algebraic_polynomials(self)
 
+            elif component_type == "word_operation" and \
+                    operation in ['ROTATE_BY_VARIABLE_AMOUNT', 'SHIFT_BY_VARIABLE_AMOUNT']:
+                raise ValueError(f"polynomial generation of {operation} operation is not supported at present")
+
         return Sequence(polynomials)
 
     def ring(self):
@@ -275,6 +279,17 @@ def var_names(self):
                         # aux output variables
                         var_names += \
                             [component_id + "_" + "o" + str(n) + "_" + str(i) for i in range(output_size)]
+            elif component.type == "word_operation" and component.description[0].lower() == "modsub":
+                ninput_words = component.description[1]
+                nadditions = ninput_words - 1
+
+                for n in range(nadditions):
+                    # borrow variables
+                    var_names += [component_id + "_" + "b" + str(n) + "_" + str(i) for i in range(output_size)]
+                    if n < nadditions - 1:
+                        # aux output variables
+                        var_names += \
+                            [component_id + "_" + "o" + str(n) + "_" + str(i) for i in range(output_size)]
 
         for i in range(len(self._cipher.inputs)):
             var_names += [self._cipher.inputs[i] + "_" +

claasp/cipher_modules/models/cp/cp_model.py

@@ -25,6 +25,7 @@
 from sage.crypto.sbox import SBox
 
 from claasp.cipher_modules.component_analysis_tests import branch_number
+from claasp.cipher_modules.models.cp.minizinc_utils import usefulfunctions
 from claasp.cipher_modules.models.utils import write_model_to_file, convert_solver_solution_to_dictionary
 from claasp.name_mappings import SBOX
 
@@ -57,15 +58,13 @@ def initialise_model(self):
         self.list_of_xor_components = []
         self.list_of_xor_all_inputs = []
         self.component_and_probability = {}
-        self._model_prefix = [
-            'include "globals.mzn";',
-            f"include \"{os.path.join(os.path.dirname(__file__), 'Minizinc_functions', 'Usefulfunctions.mzn')}\";"]
+        self._model_prefix = ['include "globals.mzn";', f'{usefulfunctions.MINIZINC_USEFUL_FUNCTIONS}']
 
     def add_solutions_from_components_values(self, components_values, memory, model_type, solutions, solve_time,
                                              solver_name, solver_output, total_weight):
         for i in range(len(total_weight)):
             solution = convert_solver_solution_to_dictionary(
-                self.cipher_id,
+                self._cipher,
                 model_type,
                 solver_name,
                 solve_time,
@@ -287,7 +286,7 @@ def format_component_value(self, component_id, string):
 
         return value
 
-    def get_command_for_solver_process(self, input_file_path, model_type, solver_name):
+    def get_command_for_solver_process(self, input_file_path, model_type, solver_name, num_of_processors, timelimit):
         solvers = ['xor_differential_one_solution',
                    'xor_linear_one_solution',
                    'deterministic_truncated_xor_differential_one_solution',
@@ -296,9 +295,11 @@ def get_command_for_solver_process(self, input_file_path, model_type, solver_nam
                    'evaluate_cipher']
         write_model_to_file(self._model_constraints, input_file_path)
         if model_type in solvers:
-            command = ['minizinc', '--solver-statistics', '--solver', solver_name, input_file_path]
+            command = ['minizinc', f'-p {num_of_processors}', '--solver-statistics', '--time-limit', str(timelimit),
+                       '--solver', solver_name, input_file_path]
         else:
-            command = ['minizinc', '-a', '--solver-statistics', '--solver', solver_name, input_file_path]
+            command = ['minizinc', f'-p {num_of_processors}', '-a', '--solver-statistics',
+                       '--time-limit', str(timelimit), '--solver', solver_name, input_file_path]
 
         return command
 
@@ -409,7 +410,7 @@ def set_component_solution_value(self, component_solution, truncated, value):
         else:
             component_solution['value'] = value
 
-    def solve(self, model_type, solver_name=None):
+    def solve(self, model_type, solver_name=None, num_of_processors=1, timelimit=60000):
         """
         Return the solution of the model.
 
@@ -430,6 +431,8 @@ def solve(self, model_type, solver_name=None):
           * ``'Chuffed'``
           * ``'Gecode'``
           * ``'COIN-BC'``
+        - ``num_of_processors`` -- **integer**; the number of processors to be used
+        - ``timelimit`` -- **integer**; time limit to output a result
 
         EXAMPLES::
 
@@ -450,15 +453,17 @@ def solve(self, model_type, solver_name=None):
         """
         cipher_name = self.cipher_id
         input_file_path = f'{cipher_name}_Cp_{model_type}_{solver_name}.mzn'
-        command = self.get_command_for_solver_process(input_file_path, model_type, solver_name)
+        command = self.get_command_for_solver_process(
+            input_file_path, model_type, solver_name, num_of_processors, timelimit
+        )
         solver_process = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8")
         os.remove(input_file_path)
         if solver_process.returncode >= 0:
             solutions = []
             solver_output = solver_process.stdout.splitlines()
             solve_time, memory, components_values, total_weight = self._parse_solver_output(solver_output)
             if components_values == {}:
-                solution = convert_solver_solution_to_dictionary(self.cipher_id, model_type, solver_name,
+                solution = convert_solver_solution_to_dictionary(self._cipher, model_type, solver_name,
                                                                  solve_time, memory,
                                                                  components_values, total_weight)
                 if 'UNSATISFIABLE' in solver_output[0]:

claasp/cipher_modules/models/cp/cp_models/cp_deterministic_truncated_xor_differential_model.py

@@ -610,7 +610,7 @@ def _parse_solver_output(self, output_to_parse, model_type):
 
         return time, memory, components_values
 
-    def solve(self, model_type, solver_name=None):
+    def solve(self, model_type, solver_name=None, num_of_processors=1, timelimit=60000):
         """
         Return the solution of the model.
 
@@ -632,6 +632,9 @@ def solve(self, model_type, solver_name=None):
           * ``'Gecode'``
           * ``'COIN-BC'``
 
+        - ``num_of_processors`` -- **integer**; the number of processors to be used
+        - ``timelimit`` -- **integer**; time limit to output a result
+
         EXAMPLES::
 
             sage: from claasp.cipher_modules.models.cp.cp_models.cp_xor_differential_trail_search_model import CpXorDifferentialTrailSearchModel
@@ -652,7 +655,9 @@ def solve(self, model_type, solver_name=None):
 
         cipher_name = self.cipher_id
         input_file_path = f'{cipher_name}_Cp_{model_type}_{solver_name}.mzn'
-        command = self.get_command_for_solver_process(input_file_path, model_type, solver_name)
+        command = self.get_command_for_solver_process(
+            input_file_path, model_type, solver_name, num_of_processors, timelimit
+        )
         solver_process = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8")
         os.remove(input_file_path)
         if solver_process.returncode >= 0: