Moduli Generator

Handles the generation, screening, and management of cryptographic modulus files.

This class provides methods for generating moduli candidates, screening them for validity, and organizing them into structured formats used for secure communications.

This utility is intended for cryptographic security operations, where moduli are required for operations like key exchange. It uses external tools like ssh-keygen for candidate generation and screening processes.

Attributes:

Name	Type	Description
config		The configuration object containing paths and settings such as directory paths and security-related configurations.
version		The version of the moduli generator settings.
logger		The logger used for logging information, warnings, and errors during execution.
db	MariaDBConnector	Lazily instantiated attribute for database storage of moduli, created as needed.

Source code in moduli_generator/__init__.py

class ModuliGenerator:
    """Handles the generation, screening, and management of cryptographic modulus files.

    This class provides methods for generating moduli candidates, screening them for
    validity, and organizing them into structured formats used for secure communications.

    This utility is intended for cryptographic security operations, where moduli
    are required for operations like key exchange. It uses external tools like
    `ssh-keygen` for candidate generation and screening processes.

    Attributes:
        config: The configuration object containing paths and settings such as
            directory paths and security-related configurations.
        version: The version of the moduli generator settings.
        logger: The logger used for logging information, warnings, and errors
            during execution.
        db: Lazily instantiated attribute for database storage of moduli,
            created as needed.
    """

    def __repr__(self) -> str:
        """
        Returns a string representation of the object.

        The method constructs a formatted string that combines specific attributes
        of the object into a single output using colons as separators.

        Returns:
            str: A formatted string representation of the object.
        """
        return ":".join(
            tuple(
                {"moduli_home": self.config.moduli_dir},
                {"version": self.version},
                {"logger": self.logger.name},
                {"db": self.db},
            )
        )

    def __init__(self, config: ModuliConfig = default_config()) -> "ModuliGenerator":
        """
        Class responsible for managing moduli configuration and related utilities. Handles
        logging configuration paths and supports lazy database initialization to optimize
        resource usage. Leverages the provided configuration to set up essential properties
        and integrate logging behavior.

        Args:
            config (ModuliConfig): Configuration instance that contains moduli settings and utilities.
        """
        self.config = config
        self.version = config.version
        self.logger = self.config.get_logger()
        self.logger.name = __name__

        # Log paths used
        if self.config:
            for path_name, path_obj in [
                ("Base directory", self.config.moduli_dir),
                ("Candidates directory", self.config.candidates_dir),
                ("Moduli directory", self.config.moduli_dir),
                ("Log directory", self.config.log_dir),
                ("MariaDB config", self.config.mariadb_cnf),
            ]:
                self.logger.info(f"Using {path_name}: {path_obj}")

        # Store config for lazy DB initialization instead of creating a connection here
        self._db = None

    @property
    def db(self) -> MariaDBConnector:
        """Lazy initialization of the database connection property.

        This property ensures that the database connection is initialized only once,
        upon first access, and then reused for subsequent operations.

        Returns:
            MariaDBConnector: Initialized database connection object.
        """
        if self._db is None:
            self._db = MariaDBConnector(self.config)
        return self._db

    @property
    def __version__(self) -> str:
        """
        Represents the version property of a class that retrieves
                the version information of the instance.

                This property is read-only and provides access to the internal
                `version` attribute of the class instance.

        Returns:
            str: Current version of the instance.
        """
        return self.version

    @staticmethod
    def _run_subprocess_with_logging(
        command: str,
        logger: Logger,
        info_level: int = INFO,
        debug_level: int = DEBUG,
    ) -> subprocess.CompletedProcess:
        """
        Executes a subprocess command with real-time logging for both `stdout` and `stderr`. This method
                handles the logging of subprocess output streams directly as they are produced, using a
                threaded approach to ensure asynchronous handling of stream data.

                Subprocess outputs are logged in real-time using the provided logger object, and the function
                returns a custom `StreamedResult` object containing information about the command execution
                (omitting `stdout` and `stderr` data as they're already logged). Exceptions encountered during the
                operation, such as errors in the process execution or issues with logging, are propagated
                appropriately.

        Args:
            command: The command to be executed as a list of strings.
            debug_level: Logging level used for `stderr` stream messages. Defaults to DEBUG.
            info_level: Logging level used for `stdout` stream messages. Defaults to INFO.
            logger: Logger instance used for logging the subprocess output.

        Returns:
            subprocess.CompletedProcess: A custom result object containing command arguments and a return code.
        """
        import threading

        def log_stream(stream, log_func, prefix):
            """Helper to log stream output in real-time"""
            try:
                for line in iter(stream.readline, ""):
                    if line.strip():
                        log_func(f"{line.strip()}")
            except Exception as err:
                logger.error(f"Error reading {prefix} stream: {err}")
            finally:
                stream.close()

        try:
            process = subprocess.Popen(
                command,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                bufsize=1,  # Line buffered for real-time output
                universal_newlines=True,
            )

            # Start threads to handle stdout and stderr streams concurrently
            stdout_thread = threading.Thread(
                target=log_stream,
                args=(
                    process.stdout,
                    lambda msg: logger.log(info_level, msg),
                    "stdout",
                ),
                daemon=True,
            )
            stderr_thread = threading.Thread(
                target=log_stream,
                args=(
                    process.stderr,
                    lambda msg: logger.log(debug_level, msg),
                    "stderr",
                ),
                daemon=True,
            )

            stdout_thread.start()
            stderr_thread.start()

            # Wait for the process to complete
            return_code = process.wait()

            # Wait for logging threads to finish
            stdout_thread.join(timeout=5.0)  # Prevent hanging
            stderr_thread.join(timeout=5.0)

            if return_code != 0:
                raise subprocess.CalledProcessError(return_code, command)

            # Create a CompletedProcess-like object for compatibility
            class StreamedResult:
                def __init__(self, returncode, args):
                    self.returncode = returncode
                    self.args = args
                    self.stdout = ""  # Output already logged in real-time
                    self.stderr = ""  # Errors already logged in real-time

            return StreamedResult(return_code, command)

        except subprocess.CalledProcessError:
            raise
        except Exception as err:
            logger.error(f"Unexpected error running command {command}: {err}")
            raise subprocess.CalledProcessError(1, command) from err

    @staticmethod
    def _generate_candidates_static(config: ModuliConfig, key_length: int) -> Path:
        """
        Generate a file of modular arithmetic candidates using ssh-keygen via a subprocess.

        This static method ensures that provided key length and nice value are validated
        for safe subprocess execution. The resulting candidates are saved in a file within
        the specified candidates directory and are uniquely identified with a timestamp.

        Args:
            config (ModuliConfig): Configuration object that holds the candidates directory,
                nice value, and logger.
            key_length (int): The desired bit length for the moduli candidates.

        Returns:
            Path: The file path where the generated candidates are stored.

        Raises:
            subprocess.CalledProcessError: If the ssh-keygen process fails during
                candidate generation.
        """
        candidates_file = (
            config.candidates_dir
            / f"candidates_{key_length}_{iso_utc_timestamp(compress=True)}"
        )
        logger = config.get_logger()

        # nice_value and key_length(s) CAN Be User provided Variables. We need to make sure they're safe.
        safe_key_length, safe_nice_value = validate_subprocess_args(
            key_length, config.nice_value
        )

        # try:
        command = [
            "nice",
            "-n",
            f"{safe_nice_value}",
            "ssh-keygen",
            "-M",
            "generate",
            "-O",
            f"bits={safe_key_length}",
            str(candidates_file),
        ]

        try:
            ModuliGenerator._run_subprocess_with_logging(command, logger)

        except subprocess.CalledProcessError as err:
            logger.error(f"ssh-keygen generate failed for {key_length} bits: {err}")
            # stderr is already logged in real-time by the streaming implementation
            raise err

        return candidates_file

    @staticmethod
    def _screen_candidates_static(config: ModuliConfig, candidates_file: Path) -> Path:
        """
        Performs screening of SSH moduli candidates using the `ssh-keygen` tool and
        a specified "nice" value. The function validates input arguments, constructs
        the appropriate command line, and handles real-time logging of the subprocess
        output. Generated screened files replace candidate files on success.

        Args:
            config (ModuliConfig): Configuration object containing parameters such as
                `moduli_dir`, `nice_value`, `generator_type`, and logging configuration.
            candidates_file (Path): Path object referring to the file containing the
                SSH moduli candidates to be screened.

        Returns:
            Path: Path to the successfully screened moduli file.

        Raises:
            subprocess.CalledProcessError: If the `ssh-keygen` command fails.
        """
        screened_file = (
            config.moduli_dir
            / f"{candidates_file.name.replace('candidates', 'moduli')}"
        )
        logger = config.get_logger()

        # We only need to validate a nice value, Using valid key_length(int(3072)) to pass argument validator
        _, safe_nice_value = validate_subprocess_args(int(3072), config.nice_value)

        # try:
        checkpoint_file = config.candidates_dir / f".{candidates_file.name}"
        command = [
            "nice",
            "-n",
            f"{safe_nice_value}",
            "ssh-keygen",
            "-M",
            "screen",
            "-O",
            f"generator={config.generator_type}",
            "-O",
            f"checkpoint={str(checkpoint_file)}",
            "-f",
            str(candidates_file),
            str(screened_file),
        ]
        try:
            # Use a streaming approach for real-time output logging
            ModuliGenerator._run_subprocess_with_logging(command, logger)

        except subprocess.CalledProcessError as err:
            logger.error(f"ssh-keygen screen failed for {candidates_file}: {err}")
            # stderr is already logged in real-time by the streaming implementation
            raise err

        # Cleanup used Moduli Candidates
        candidates_file.unlink()

        return screened_file

    def _parse_moduli_files(self) -> Dict[str, List[Dict[str, Any]]]:
        """
        Parses the moduli files to extract specific key-length entries and formats them
                into a dictionary structure for further processing. This method iterates
                over a set of screened files, reads their contents line by line, and
                extracts information about timestamp, key size, and modulus if the line
                meets specific criteria.

        Returns:
            Dict[str, List[Dict[str, Any]]]: A dictionary with parsed moduli installers under the key
            'screened_moduli'. Each entry is a dictionary containing 'timestamp', 'key-size', and 'modulus'.
        """

        screened_files = self._list_moduli_files()
        screened_moduli = {}

        for file in screened_files:
            try:
                with file.open("r") as f:
                    for line in f:
                        if line.startswith("#") or not line.strip():
                            continue

                        parts = line.split()
                        if len(parts) == SSH2_MODULI_FILE_FIELD_COUNT:
                            moduli_entry = {
                                "timestamp": parts[0],  # TIMESTAMP
                                # 'type': parts[1],         # Constant, Stored in moduli_db.mod_fl_consts
                                # 'tests': parts[2],        # Constant, Stored in moduli_db.mod_fl_consts
                                # 'trials': parts[3],       # Constant, Stored in moduli_db.mod_fl_consts
                                "key-size": parts[4],  # KEY_LENGTH
                                # 'generator': parts[5], # Constant, Stored in moduli_db.mod_fl_consts
                                "modulus": parts[6],  # MODULUS
                            }

                            screened_moduli.setdefault("screened_moduli", []).append(
                                moduli_entry
                            )

            except FileNotFoundError:
                self.logger.warning(f"Moduli file not found: {file}")

        return screened_moduli

    def _list_moduli_files(self) -> List[Path]:
        """
        Lists all moduli files based on the configured moduli directory and file
                pattern.

                This function retrieves and returns a list of all files in the
                configured moduli directory that match the specified file pattern.

        Returns:
            list[Path]: List of matching moduli files
        """
        return list(self.config.moduli_dir.glob(self.config.moduli_file_pattern))

    def generate_moduli(self) -> "ModuliGenerator":
        """
        Generates and screens Diffie-Hellman moduli files for specified key lengths using
                pipeline processing for optimal performance.

                This method uses a pipeline approach where candidate generation and screening happen
                concurrently. As soon as a candidate file is generated, it's immediately submitted
                for screening, rather than waiting for all candidates to be generated first.
                This optimization reduces overall processing time by overlapping I/O and CPU operations.

        Returns:
            ModuliGenerator: self
        """
        generated_moduli = {}

        with concurrent.futures.ProcessPoolExecutor() as executor:
            # Submit all candidate generation tasks
            candidate_futures = []
            for length in self.config.key_lengths:
                future = executor.submit(
                    self._generate_candidates_static, self.config, length
                )
                candidate_futures.append((future, length))

            # Pipeline processing: screen candidates as they become available
            screening_futures = []
            candidates_generated = 0

            # Process candidates as they complete and immediately submit for screening
            for future in concurrent.futures.as_completed(
                [f for f, _ in candidate_futures]
            ):
                # Find the corresponding length for this future
                length = None
                for candidate_future, candidate_length in candidate_futures:
                    if candidate_future == future:
                        length = candidate_length
                        break

                try:
                    candidate_file = future.result()
                    candidates_generated += 1
                    self.logger.debug(
                        f"Generated candidate file {candidates_generated}/{len(candidate_futures)}: {candidate_file}"
                    )

                    # Immediately submit for screening (pipeline processing)
                    screening_future = executor.submit(
                        self._screen_candidates_static, self.config, candidate_file
                    )
                    screening_futures.append((screening_future, length))

                except Exception as e:
                    self.logger.error(
                        f"Error generating candidate for length {length}: {e}"
                    )
                    raise

            self.logger.info(
                f"Generated {candidates_generated} candidate files for key-lengths: {self.config.key_lengths}"
            )

            # Process screening results as they complete
            screened_count = 0
            for future in concurrent.futures.as_completed(
                [f for f, _ in screening_futures]
            ):
                # Find the corresponding length for this future
                length = None
                for screening_future, screening_length in screening_futures:
                    if screening_future == future:
                        length = screening_length
                        break

                try:
                    moduli_file = future.result()
                    screened_count += 1
                    self.logger.debug(
                        f"Screened moduli file {screened_count}/{len(screening_futures)}: {moduli_file}"
                    )

                    if length not in generated_moduli:
                        generated_moduli[length] = []
                    generated_moduli[length].append(moduli_file)

                except Exception as err:
                    self.logger.error(
                        f"Error screening candidate for length {length}: {err}"
                    )
                    raise

            self.logger.info(
                f"Screened {screened_count} candidate files for key-lengths: {self.config.key_lengths}"
            )

        return self

    def store_moduli(self) -> "ModuliGenerator":
        """
        Parse, validate, and store screened moduli into the database and manage their source
                files once the operation is successful. This function ensures that the moduli records
                are stored transactionally. After successful storage, the source files are deleted.

        Returns:
            ModuliGenerator: self
        """
        screened_moduli = self._parse_moduli_files()
        try:
            self.db.export_screened_moduli(screened_moduli)

            # Cleanup leftover moduli files
            moduli_files = self._list_moduli_files()
            if not self.config.preserve_moduli_after_dbstore:
                for file in moduli_files:
                    file.unlink()
        except (RuntimeError, mariadb.Error) as err:
            self.logger.error(f"Error storing moduli: {err}")

        self.logger.info(f"Moduli Stored in MariaDB database: {len(screened_moduli)}")

        return self

    def write_moduli_file(self) -> "ModuliGenerator":
        """
        Writes the moduli file using the database interface.

                This method retrieves the moduli necessary for the moduli file from the
                database and then writes it to the appropriate location using
                the database interface.

        Returns:
            ModuliGenerator: self
        """
        try:
            self.db.write_moduli_file()

        except RuntimeError as err:
            self.logger.info(err)

        return self

    def restart_screening(self) -> "ModuliGenerator":
        """
        Restarts the screening process for candidates grouped by their respective lengths
                and generates moduli files for each key length. This method processes candidates
                from the specified directory, screens them using a process pool executor, and returns
                an updated instance of the ModuliGenerator.

        Returns:
            ModuliGenerator: The current instance of ModuliGenerator for method chaining.
        """

        def _get_restart_candidates_by_length() -> List[Path]:
            """
            Retrieves a mapping of restart candidates grouped by their length. The function
                        iterates through directories and file patterns specified in the configuration
                        to build a dictionary. Each key in the dictionary represents a unique length,
                        and the value is a list of paths to the corresponding candidate files.

            Returns:
                dict[int, List[Path]]: A dictionary where the keys are lengths (int) and the values are
                    lists of Path objects representing restart candidate files.
            """
            results = {}
            for idx in self.config.candidates_dir.glob(
                self.config.candidate_idx_pattern
            ):
                candidate = Path(self.config.candidates_dir) / idx.name.replace(
                    ".candidates", "candidates"
                )
                local_length = int(idx.name.split("_")[1])
                if local_length not in results:
                    results[local_length] = []
                results[local_length].append(candidate)
            return results

        candidates_by_length = _get_restart_candidates_by_length()

        if len(candidates_by_length) != 0:

            with concurrent.futures.ProcessPoolExecutor() as executor:

                # Then screen candidates
                screening_futures = []
                for length, candidate_files in candidates_by_length.items():
                    for candidate_file in candidate_files:
                        future = executor.submit(
                            self._screen_candidates_static, self.config, candidate_file
                        )
                        screening_futures.append((future, length))

                # Process completed screening futures
                generated_moduli = {}
                for future, length in screening_futures:
                    moduli_file = future.result()
                    if length not in generated_moduli:
                        generated_moduli[length] = []
                    generated_moduli[length].append(moduli_file)
                self.logger.info(
                    f"Produced {len(screening_futures)} files of screened moduli\n"
                    f"for key-lengths:" + f"{self.config.key_lengths}"
                )
        else:
            self.logger.info(f"No Unscreened Candidates Found for Restart")

        return self

config instance-attribute

config = config

version instance-attribute

version = version

logger instance-attribute

logger = get_logger()

_db instance-attribute

_db = None

db property

db

Lazy initialization of the database connection property.

This property ensures that the database connection is initialized only once, upon first access, and then reused for subsequent operations.

Returns:

Name	Type	Description
MariaDBConnector	MariaDBConnector	Initialized database connection object.

__version__ property

__version__

Represents the version property of a class that retrieves the version information of the instance.

    This property is read-only and provides access to the internal
    `version` attribute of the class instance.

Returns:

Name	Type	Description
str	str	Current version of the instance.

__repr__

__repr__()

Returns a string representation of the object.

The method constructs a formatted string that combines specific attributes of the object into a single output using colons as separators.

Returns:

Name	Type	Description
str	str	A formatted string representation of the object.

Source code in moduli_generator/__init__.py

def __repr__(self) -> str:
    """
    Returns a string representation of the object.

    The method constructs a formatted string that combines specific attributes
    of the object into a single output using colons as separators.

    Returns:
        str: A formatted string representation of the object.
    """
    return ":".join(
        tuple(
            {"moduli_home": self.config.moduli_dir},
            {"version": self.version},
            {"logger": self.logger.name},
            {"db": self.db},
        )
    )

__init__

__init__(config=default_config())

Class responsible for managing moduli configuration and related utilities. Handles logging configuration paths and supports lazy database initialization to optimize resource usage. Leverages the provided configuration to set up essential properties and integrate logging behavior.

Parameters:

Name	Type	Description	Default
config	ModuliConfig	Configuration instance that contains moduli settings and utilities.	default_config()

Source code in moduli_generator/__init__.py

def __init__(self, config: ModuliConfig = default_config()) -> "ModuliGenerator":
    """
    Class responsible for managing moduli configuration and related utilities. Handles
    logging configuration paths and supports lazy database initialization to optimize
    resource usage. Leverages the provided configuration to set up essential properties
    and integrate logging behavior.

    Args:
        config (ModuliConfig): Configuration instance that contains moduli settings and utilities.
    """
    self.config = config
    self.version = config.version
    self.logger = self.config.get_logger()
    self.logger.name = __name__

    # Log paths used
    if self.config:
        for path_name, path_obj in [
            ("Base directory", self.config.moduli_dir),
            ("Candidates directory", self.config.candidates_dir),
            ("Moduli directory", self.config.moduli_dir),
            ("Log directory", self.config.log_dir),
            ("MariaDB config", self.config.mariadb_cnf),
        ]:
            self.logger.info(f"Using {path_name}: {path_obj}")

    # Store config for lazy DB initialization instead of creating a connection here
    self._db = None

_run_subprocess_with_logging staticmethod

_run_subprocess_with_logging(
    command, logger, info_level=INFO, debug_level=DEBUG
)

Executes a subprocess command with real-time logging for both stdout and stderr. This method handles the logging of subprocess output streams directly as they are produced, using a threaded approach to ensure asynchronous handling of stream data.

    Subprocess outputs are logged in real-time using the provided logger object, and the function
    returns a custom `StreamedResult` object containing information about the command execution
    (omitting `stdout` and `stderr` data as they're already logged). Exceptions encountered during the
    operation, such as errors in the process execution or issues with logging, are propagated
    appropriately.

Parameters:

Name	Type	Description	Default
command	str	The command to be executed as a list of strings.	required
debug_level	int	Logging level used for stderr stream messages. Defaults to DEBUG.	DEBUG
info_level	int	Logging level used for stdout stream messages. Defaults to INFO.	INFO
logger	Logger	Logger instance used for logging the subprocess output.	required

Returns:

Type	Description
CompletedProcess	subprocess.CompletedProcess: A custom result object containing command arguments and a return code.

Source code in moduli_generator/__init__.py

@staticmethod
def _run_subprocess_with_logging(
    command: str,
    logger: Logger,
    info_level: int = INFO,
    debug_level: int = DEBUG,
) -> subprocess.CompletedProcess:
    """
    Executes a subprocess command with real-time logging for both `stdout` and `stderr`. This method
            handles the logging of subprocess output streams directly as they are produced, using a
            threaded approach to ensure asynchronous handling of stream data.

            Subprocess outputs are logged in real-time using the provided logger object, and the function
            returns a custom `StreamedResult` object containing information about the command execution
            (omitting `stdout` and `stderr` data as they're already logged). Exceptions encountered during the
            operation, such as errors in the process execution or issues with logging, are propagated
            appropriately.

    Args:
        command: The command to be executed as a list of strings.
        debug_level: Logging level used for `stderr` stream messages. Defaults to DEBUG.
        info_level: Logging level used for `stdout` stream messages. Defaults to INFO.
        logger: Logger instance used for logging the subprocess output.

    Returns:
        subprocess.CompletedProcess: A custom result object containing command arguments and a return code.
    """
    import threading

    def log_stream(stream, log_func, prefix):
        """Helper to log stream output in real-time"""
        try:
            for line in iter(stream.readline, ""):
                if line.strip():
                    log_func(f"{line.strip()}")
        except Exception as err:
            logger.error(f"Error reading {prefix} stream: {err}")
        finally:
            stream.close()

    try:
        process = subprocess.Popen(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1,  # Line buffered for real-time output
            universal_newlines=True,
        )

        # Start threads to handle stdout and stderr streams concurrently
        stdout_thread = threading.Thread(
            target=log_stream,
            args=(
                process.stdout,
                lambda msg: logger.log(info_level, msg),
                "stdout",
            ),
            daemon=True,
        )
        stderr_thread = threading.Thread(
            target=log_stream,
            args=(
                process.stderr,
                lambda msg: logger.log(debug_level, msg),
                "stderr",
            ),
            daemon=True,
        )

        stdout_thread.start()
        stderr_thread.start()

        # Wait for the process to complete
        return_code = process.wait()

        # Wait for logging threads to finish
        stdout_thread.join(timeout=5.0)  # Prevent hanging
        stderr_thread.join(timeout=5.0)

        if return_code != 0:
            raise subprocess.CalledProcessError(return_code, command)

        # Create a CompletedProcess-like object for compatibility
        class StreamedResult:
            def __init__(self, returncode, args):
                self.returncode = returncode
                self.args = args
                self.stdout = ""  # Output already logged in real-time
                self.stderr = ""  # Errors already logged in real-time

        return StreamedResult(return_code, command)

    except subprocess.CalledProcessError:
        raise
    except Exception as err:
        logger.error(f"Unexpected error running command {command}: {err}")
        raise subprocess.CalledProcessError(1, command) from err

_generate_candidates_static staticmethod

_generate_candidates_static(config, key_length)

Generate a file of modular arithmetic candidates using ssh-keygen via a subprocess.

This static method ensures that provided key length and nice value are validated for safe subprocess execution. The resulting candidates are saved in a file within the specified candidates directory and are uniquely identified with a timestamp.

Parameters:

Name	Type	Description	Default
config	ModuliConfig	Configuration object that holds the candidates directory, nice value, and logger.	required
key_length	int	The desired bit length for the moduli candidates.	required

Returns:

Name	Type	Description
Path	PosixPath	The file path where the generated candidates are stored.

Raises:

Type	Description
CalledProcessError	If the ssh-keygen process fails during candidate generation.

Source code in moduli_generator/__init__.py

@staticmethod
def _generate_candidates_static(config: ModuliConfig, key_length: int) -> Path:
    """
    Generate a file of modular arithmetic candidates using ssh-keygen via a subprocess.

    This static method ensures that provided key length and nice value are validated
    for safe subprocess execution. The resulting candidates are saved in a file within
    the specified candidates directory and are uniquely identified with a timestamp.

    Args:
        config (ModuliConfig): Configuration object that holds the candidates directory,
            nice value, and logger.
        key_length (int): The desired bit length for the moduli candidates.

    Returns:
        Path: The file path where the generated candidates are stored.

    Raises:
        subprocess.CalledProcessError: If the ssh-keygen process fails during
            candidate generation.
    """
    candidates_file = (
        config.candidates_dir
        / f"candidates_{key_length}_{iso_utc_timestamp(compress=True)}"
    )
    logger = config.get_logger()

    # nice_value and key_length(s) CAN Be User provided Variables. We need to make sure they're safe.
    safe_key_length, safe_nice_value = validate_subprocess_args(
        key_length, config.nice_value
    )

    # try:
    command = [
        "nice",
        "-n",
        f"{safe_nice_value}",
        "ssh-keygen",
        "-M",
        "generate",
        "-O",
        f"bits={safe_key_length}",
        str(candidates_file),
    ]

    try:
        ModuliGenerator._run_subprocess_with_logging(command, logger)

    except subprocess.CalledProcessError as err:
        logger.error(f"ssh-keygen generate failed for {key_length} bits: {err}")
        # stderr is already logged in real-time by the streaming implementation
        raise err

    return candidates_file

_screen_candidates_static staticmethod

_screen_candidates_static(config, candidates_file)

Performs screening of SSH moduli candidates using the ssh-keygen tool and a specified "nice" value. The function validates input arguments, constructs the appropriate command line, and handles real-time logging of the subprocess output. Generated screened files replace candidate files on success.

Parameters:

Name	Type	Description	Default
config	ModuliConfig	Configuration object containing parameters such as moduli_dir, nice_value, generator_type, and logging configuration.	required
candidates_file	PosixPath	Path object referring to the file containing the SSH moduli candidates to be screened.	required

Returns:

Name	Type	Description
Path	PosixPath	Path to the successfully screened moduli file.

Raises:

Type	Description
CalledProcessError	If the ssh-keygen command fails.

Source code in moduli_generator/__init__.py

@staticmethod
def _screen_candidates_static(config: ModuliConfig, candidates_file: Path) -> Path:
    """
    Performs screening of SSH moduli candidates using the `ssh-keygen` tool and
    a specified "nice" value. The function validates input arguments, constructs
    the appropriate command line, and handles real-time logging of the subprocess
    output. Generated screened files replace candidate files on success.

    Args:
        config (ModuliConfig): Configuration object containing parameters such as
            `moduli_dir`, `nice_value`, `generator_type`, and logging configuration.
        candidates_file (Path): Path object referring to the file containing the
            SSH moduli candidates to be screened.

    Returns:
        Path: Path to the successfully screened moduli file.

    Raises:
        subprocess.CalledProcessError: If the `ssh-keygen` command fails.
    """
    screened_file = (
        config.moduli_dir
        / f"{candidates_file.name.replace('candidates', 'moduli')}"
    )
    logger = config.get_logger()

    # We only need to validate a nice value, Using valid key_length(int(3072)) to pass argument validator
    _, safe_nice_value = validate_subprocess_args(int(3072), config.nice_value)

    # try:
    checkpoint_file = config.candidates_dir / f".{candidates_file.name}"
    command = [
        "nice",
        "-n",
        f"{safe_nice_value}",
        "ssh-keygen",
        "-M",
        "screen",
        "-O",
        f"generator={config.generator_type}",
        "-O",
        f"checkpoint={str(checkpoint_file)}",
        "-f",
        str(candidates_file),
        str(screened_file),
    ]
    try:
        # Use a streaming approach for real-time output logging
        ModuliGenerator._run_subprocess_with_logging(command, logger)

    except subprocess.CalledProcessError as err:
        logger.error(f"ssh-keygen screen failed for {candidates_file}: {err}")
        # stderr is already logged in real-time by the streaming implementation
        raise err

    # Cleanup used Moduli Candidates
    candidates_file.unlink()

    return screened_file

_parse_moduli_files

_parse_moduli_files()

Parses the moduli files to extract specific key-length entries and formats them into a dictionary structure for further processing. This method iterates over a set of screened files, reads their contents line by line, and extracts information about timestamp, key size, and modulus if the line meets specific criteria.

Returns:

Type	Description
Dict[str, List[Dict[str, Any]]]	Dict[str, List[Dict[str, Any]]]: A dictionary with parsed moduli installers under the key
Dict[str, List[Dict[str, Any]]]	'screened_moduli'. Each entry is a dictionary containing 'timestamp', 'key-size', and 'modulus'.

Source code in moduli_generator/__init__.py

def _parse_moduli_files(self) -> Dict[str, List[Dict[str, Any]]]:
    """
    Parses the moduli files to extract specific key-length entries and formats them
            into a dictionary structure for further processing. This method iterates
            over a set of screened files, reads their contents line by line, and
            extracts information about timestamp, key size, and modulus if the line
            meets specific criteria.

    Returns:
        Dict[str, List[Dict[str, Any]]]: A dictionary with parsed moduli installers under the key
        'screened_moduli'. Each entry is a dictionary containing 'timestamp', 'key-size', and 'modulus'.
    """

    screened_files = self._list_moduli_files()
    screened_moduli = {}

    for file in screened_files:
        try:
            with file.open("r") as f:
                for line in f:
                    if line.startswith("#") or not line.strip():
                        continue

                    parts = line.split()
                    if len(parts) == SSH2_MODULI_FILE_FIELD_COUNT:
                        moduli_entry = {
                            "timestamp": parts[0],  # TIMESTAMP
                            # 'type': parts[1],         # Constant, Stored in moduli_db.mod_fl_consts
                            # 'tests': parts[2],        # Constant, Stored in moduli_db.mod_fl_consts
                            # 'trials': parts[3],       # Constant, Stored in moduli_db.mod_fl_consts
                            "key-size": parts[4],  # KEY_LENGTH
                            # 'generator': parts[5], # Constant, Stored in moduli_db.mod_fl_consts
                            "modulus": parts[6],  # MODULUS
                        }

                        screened_moduli.setdefault("screened_moduli", []).append(
                            moduli_entry
                        )

        except FileNotFoundError:
            self.logger.warning(f"Moduli file not found: {file}")

    return screened_moduli

_list_moduli_files

_list_moduli_files()

Lists all moduli files based on the configured moduli directory and file pattern.

    This function retrieves and returns a list of all files in the
    configured moduli directory that match the specified file pattern.

Returns:

Type	Description
List[PosixPath]	list[Path]: List of matching moduli files

Source code in moduli_generator/__init__.py

def _list_moduli_files(self) -> List[Path]:
    """
    Lists all moduli files based on the configured moduli directory and file
            pattern.

            This function retrieves and returns a list of all files in the
            configured moduli directory that match the specified file pattern.

    Returns:
        list[Path]: List of matching moduli files
    """
    return list(self.config.moduli_dir.glob(self.config.moduli_file_pattern))

generate_moduli

generate_moduli()

Generates and screens Diffie-Hellman moduli files for specified key lengths using pipeline processing for optimal performance.

    This method uses a pipeline approach where candidate generation and screening happen
    concurrently. As soon as a candidate file is generated, it's immediately submitted
    for screening, rather than waiting for all candidates to be generated first.
    This optimization reduces overall processing time by overlapping I/O and CPU operations.

Returns:

Name	Type	Description
ModuliGenerator	ModuliGenerator	self

Source code in moduli_generator/__init__.py

def generate_moduli(self) -> "ModuliGenerator":
    """
    Generates and screens Diffie-Hellman moduli files for specified key lengths using
            pipeline processing for optimal performance.

            This method uses a pipeline approach where candidate generation and screening happen
            concurrently. As soon as a candidate file is generated, it's immediately submitted
            for screening, rather than waiting for all candidates to be generated first.
            This optimization reduces overall processing time by overlapping I/O and CPU operations.

    Returns:
        ModuliGenerator: self
    """
    generated_moduli = {}

    with concurrent.futures.ProcessPoolExecutor() as executor:
        # Submit all candidate generation tasks
        candidate_futures = []
        for length in self.config.key_lengths:
            future = executor.submit(
                self._generate_candidates_static, self.config, length
            )
            candidate_futures.append((future, length))

        # Pipeline processing: screen candidates as they become available
        screening_futures = []
        candidates_generated = 0

        # Process candidates as they complete and immediately submit for screening
        for future in concurrent.futures.as_completed(
            [f for f, _ in candidate_futures]
        ):
            # Find the corresponding length for this future
            length = None
            for candidate_future, candidate_length in candidate_futures:
                if candidate_future == future:
                    length = candidate_length
                    break

            try:
                candidate_file = future.result()
                candidates_generated += 1
                self.logger.debug(
                    f"Generated candidate file {candidates_generated}/{len(candidate_futures)}: {candidate_file}"
                )

                # Immediately submit for screening (pipeline processing)
                screening_future = executor.submit(
                    self._screen_candidates_static, self.config, candidate_file
                )
                screening_futures.append((screening_future, length))

            except Exception as e:
                self.logger.error(
                    f"Error generating candidate for length {length}: {e}"
                )
                raise

        self.logger.info(
            f"Generated {candidates_generated} candidate files for key-lengths: {self.config.key_lengths}"
        )

        # Process screening results as they complete
        screened_count = 0
        for future in concurrent.futures.as_completed(
            [f for f, _ in screening_futures]
        ):
            # Find the corresponding length for this future
            length = None
            for screening_future, screening_length in screening_futures:
                if screening_future == future:
                    length = screening_length
                    break

            try:
                moduli_file = future.result()
                screened_count += 1
                self.logger.debug(
                    f"Screened moduli file {screened_count}/{len(screening_futures)}: {moduli_file}"
                )

                if length not in generated_moduli:
                    generated_moduli[length] = []
                generated_moduli[length].append(moduli_file)

            except Exception as err:
                self.logger.error(
                    f"Error screening candidate for length {length}: {err}"
                )
                raise

        self.logger.info(
            f"Screened {screened_count} candidate files for key-lengths: {self.config.key_lengths}"
        )

    return self

store_moduli

store_moduli()

Parse, validate, and store screened moduli into the database and manage their source files once the operation is successful. This function ensures that the moduli records are stored transactionally. After successful storage, the source files are deleted.

Returns:

Name	Type	Description
ModuliGenerator	ModuliGenerator	self

Source code in moduli_generator/__init__.py

def store_moduli(self) -> "ModuliGenerator":
    """
    Parse, validate, and store screened moduli into the database and manage their source
            files once the operation is successful. This function ensures that the moduli records
            are stored transactionally. After successful storage, the source files are deleted.

    Returns:
        ModuliGenerator: self
    """
    screened_moduli = self._parse_moduli_files()
    try:
        self.db.export_screened_moduli(screened_moduli)

        # Cleanup leftover moduli files
        moduli_files = self._list_moduli_files()
        if not self.config.preserve_moduli_after_dbstore:
            for file in moduli_files:
                file.unlink()
    except (RuntimeError, mariadb.Error) as err:
        self.logger.error(f"Error storing moduli: {err}")

    self.logger.info(f"Moduli Stored in MariaDB database: {len(screened_moduli)}")

    return self

write_moduli_file

write_moduli_file()

Writes the moduli file using the database interface.

    This method retrieves the moduli necessary for the moduli file from the
    database and then writes it to the appropriate location using
    the database interface.

Returns:

Name	Type	Description
ModuliGenerator	ModuliGenerator	self

Source code in moduli_generator/__init__.py

def write_moduli_file(self) -> "ModuliGenerator":
    """
    Writes the moduli file using the database interface.

            This method retrieves the moduli necessary for the moduli file from the
            database and then writes it to the appropriate location using
            the database interface.

    Returns:
        ModuliGenerator: self
    """
    try:
        self.db.write_moduli_file()

    except RuntimeError as err:
        self.logger.info(err)

    return self

restart_screening

restart_screening()

Restarts the screening process for candidates grouped by their respective lengths and generates moduli files for each key length. This method processes candidates from the specified directory, screens them using a process pool executor, and returns an updated instance of the ModuliGenerator.

Returns:

Name	Type	Description
ModuliGenerator	ModuliGenerator	The current instance of ModuliGenerator for method chaining.

Source code in moduli_generator/__init__.py

def restart_screening(self) -> "ModuliGenerator":
    """
    Restarts the screening process for candidates grouped by their respective lengths
            and generates moduli files for each key length. This method processes candidates
            from the specified directory, screens them using a process pool executor, and returns
            an updated instance of the ModuliGenerator.

    Returns:
        ModuliGenerator: The current instance of ModuliGenerator for method chaining.
    """

    def _get_restart_candidates_by_length() -> List[Path]:
        """
        Retrieves a mapping of restart candidates grouped by their length. The function
                    iterates through directories and file patterns specified in the configuration
                    to build a dictionary. Each key in the dictionary represents a unique length,
                    and the value is a list of paths to the corresponding candidate files.

        Returns:
            dict[int, List[Path]]: A dictionary where the keys are lengths (int) and the values are
                lists of Path objects representing restart candidate files.
        """
        results = {}
        for idx in self.config.candidates_dir.glob(
            self.config.candidate_idx_pattern
        ):
            candidate = Path(self.config.candidates_dir) / idx.name.replace(
                ".candidates", "candidates"
            )
            local_length = int(idx.name.split("_")[1])
            if local_length not in results:
                results[local_length] = []
            results[local_length].append(candidate)
        return results

    candidates_by_length = _get_restart_candidates_by_length()

    if len(candidates_by_length) != 0:

        with concurrent.futures.ProcessPoolExecutor() as executor:

            # Then screen candidates
            screening_futures = []
            for length, candidate_files in candidates_by_length.items():
                for candidate_file in candidate_files:
                    future = executor.submit(
                        self._screen_candidates_static, self.config, candidate_file
                    )
                    screening_futures.append((future, length))

            # Process completed screening futures
            generated_moduli = {}
            for future, length in screening_futures:
                moduli_file = future.result()
                if length not in generated_moduli:
                    generated_moduli[length] = []
                generated_moduli[length].append(moduli_file)
            self.logger.info(
                f"Produced {len(screening_futures)} files of screened moduli\n"
                f"for key-lengths:" + f"{self.config.key_lengths}"
            )
    else:
        self.logger.info(f"No Unscreened Candidates Found for Restart")

    return self

options: members: true