Convert bash script to Python

.gitignore

@@ -2,3 +2,5 @@
 /nilrt-snac-*.tar.gz
 /src/nilrt-snac-conflicts/*.tar.gz
 /src/nilrt-snac-conflicts/*.ipk
+
+__pycache__/

Makefile

@@ -22,16 +22,18 @@ libdir ?= $(exec_prefix)/lib
 sbindir ?= $(exec_prefix)/sbin
 sysconfdir ?= $(prefix)/etc
 
+PYTHON_FILES = \
+	$(shell find nilrt_snac -name \*.py -or -name \*.txt)
+
 SRC_FILES = \
-	src/configure-nilrt-snac \
 	src/nilrt-snac-conflicts/control \
 	src/ni-wireguard-labview/ni-wireguard-labview.initd \
 	src/ni-wireguard-labview/wglv0.conf \
 	src/nilrt-snac \
-	src/util.sh \
 
 DIST_FILES = \
 	$(SRC_FILES) \
+	$(PYTHON_FILES) \
 	LICENSE \
 	README.md \
 	Makefile \
@@ -52,7 +54,7 @@ src/nilrt-snac-conflicts/nilrt-snac-conflicts.ipk :
 # PHONY TARGETS #
 #################
 
-.PHONY : all clean dist install uninstall
+.PHONY : all clean dist install uninstall test
 
 all : src/nilrt-snac-conflicts/nilrt-snac-conflicts.ipk
 
@@ -70,10 +72,14 @@ install : all $(DIST_FILES)
 	install -o 0 -g 0 --mode=0755 -t "$(DESTDIR)$(sbindir)" \
 		src/nilrt-snac
 
-	mkdir -p $(DESTDIR)$(libdir)/$(PACKAGE)
-	install --mode=0444 -t "$(DESTDIR)$(libdir)/$(PACKAGE)" \
-		src/configure-nilrt-snac \
-		src/util.sh \
+	mkdir -p $(DESTDIR)$(libdir)/$(PACKAGE)/nilrt_snac
+	install --mode=0444 -t "$(DESTDIR)$(libdir)/$(PACKAGE)/nilrt_snac" \
+		$(shell find nilrt_snac -name \*.py -or -name \*.txt -maxdepth 1) \
+
+	# install doesn't support recursive copy
+	mkdir -p $(DESTDIR)$(libdir)/$(PACKAGE)/nilrt_snac/_configs
+	install --mode=0444 -t "$(DESTDIR)$(libdir)/$(PACKAGE)/nilrt_snac/_configs" \
+		$(shell find nilrt_snac/_configs -name \*.py -maxdepth 1) \
 
 	mkdir -p $(DESTDIR)$(docdir)/$(PACKAGE)
 	install --mode=0444 -t "$(DESTDIR)$(docdir)/$(PACKAGE)" \

nilrt_snac/OpkgHelper.py

@@ -0,0 +1,89 @@
+"""Class to help with managing opkg install/uninstall."""
+
+import subprocess
+from typing import List
+
+from nilrt_snac import logger
+from nilrt_snac._common import get_distro
+
+OPKG_SNAC_CONF = "/etc/opkg/snac.conf"
+
+
+class OpkgHelper:  # noqa: D101 - Missing docstring in public class (auto-generated noqa)
+    def __init__(self) -> None:  # noqa: D107 - Missing docstring in __init__ (auto-generated noqa)
+        self._installed_packages: List[str] = []
+        # This runs before the prereqs are checked
+        if get_distro() != "nilrt":
+            logger.warning("Not running on nilrt, can't get list of installed packages.")
+            return
+
+        self.update()
+        packages = self._run(["list-installed"])
+
+        for line in packages.split("\n"):
+            pkg = line.split(" - ")
+            if len(pkg) > 1:
+                self._installed_packages.append(pkg[0])
+
+    def _run(  # noqa: D102 - Missing docstring in public method (auto-generated noqa)
+        self, command: List[str]
+    ) -> str:
+        result = subprocess.run(["opkg"] + command, stdout=subprocess.PIPE)
+
+        if result.returncode != 0:
+            raise RuntimeError(
+                f"Command 'opkg {' '.join(command)}' failed with return code {result.returncode}"
+            )
+
+        return result.stdout.decode()
+
+    def set_dry_run(  # noqa: D102 - Missing docstring in public method (auto-generated noqa)
+        self, dry_run: bool
+    ) -> None:
+        self._dry_run = dry_run
+
+    def install(  # noqa: D102 - Missing docstring in public method (auto-generated noqa)
+        self, package: str, force_reinstall: bool = False
+    ) -> None:
+        if not self.is_installed(package):
+            cmd = ["opkg", "install"]
+            if force_reinstall:
+                cmd.append("--force-reinstall")
+            cmd.append(package)
+            if not self._dry_run:
+                subprocess.run(cmd, check=True)
+            self._installed_packages.append(package)
+        else:
+            logger.debug(f"{package} already installed")
+
+    def remove(  # noqa: D102 - Missing docstring in public method (auto-generated noqa)
+        self, package: str, ignore_installed=False, force_essential: bool = False, force_depends: bool = False
+    ) -> None:
+
+        if ignore_installed or self.is_installed(package):
+            check = False if ignore_installed else True
+            cmd = ["opkg", "remove"]
+            if force_essential:
+                cmd.append("--force-removal-of-essential-packages")
+            if force_depends:
+                cmd.append("--force-depends")
+            cmd.append(package)
+            if not self._dry_run:
+                subprocess.run(cmd, check=check)
+            if not ignore_installed:
+                self._installed_packages.remove(package)
+        else:
+            logger.debug(f"{package} already uninstalled")
+
+    def is_installed(  # noqa: D102 - Missing docstring in public method (auto-generated noqa)
+        self, package: str
+    ) -> bool:
+        return package in self._installed_packages
+
+    def update(  # noqa: D102 - Missing docstring in public method (auto-generated noqa)
+        self,
+    ) -> None:
+        self._run(["update"])
+
+
+opkg_helper = OpkgHelper()

nilrt_snac/__init__.py

@@ -0,0 +1,28 @@
+"""nilrt-snac."""
+
+import logging
+import pathlib
+from enum import IntEnum
+
+__version__ = "0.9.0"
+
+SNAC_DATA_DIR = pathlib.Path(__file__).parents[3] / "share" / "nilrt-snac"
+
+logger = logging.getLogger(__package__)
+logger.addHandler(logging.NullHandler())
+
+
+class Errors(IntEnum):  # noqa: D101 - Missing docstring in public class (auto-generated noqa)
+    EX_OK = 0
+    EX_ERROR = 1
+    EX_USAGE = 2
+    EX_BAD_ENVIRONMENT = 128
+    EX_CHECK_FAILURE = 129
+
+
+class SNACError(Exception):  # noqa: D101 - Missing docstring in public class (auto-generated noqa)
+    def __init__(  # noqa: D107 - Missing docstring in __init__ (auto-generated noqa)
+        self, message, return_code=Errors.EX_ERROR
+    ):
+        super().__init__(message)
+        self.return_code = return_code

nilrt_snac/__main__.py

@@ -0,0 +1,145 @@
+"""nilrt-snac entry points."""
+
+import argparse
+import logging
+import sys
+from typing import List, Optional
+
+from nilrt_snac._pre_reqs import verify_prereqs
+from nilrt_snac.OpkgHelper import opkg_helper
+from nilrt_snac._configs import CONFIGS
+
+
+from nilrt_snac import Errors, logger, SNACError, __version__
+
+VERSION_DESCRIPTION = \
+f"""
+nilrt-snac {__version__}
+Copyright (C) 2024 NI (Emerson Electric)
+License MIT: MIT License <https://spdx.org/licenses/MIT.html>
+This is free software: you are free to change and redistribute it.
+There is NO WARRANTY, to the extent permitted by law.
+"""
+
+
+def _configure(args: argparse.Namespace) -> int:
+    """Configure SNAC mode."""
+    logger.warning("!! THIS TOOL IS IN-DEVELOPMENT AND APPROPRIATE ONLY FOR DEVELOPER TESTING !!")
+    logger.warning("!! Running this tool will irreversibly alter the state of your system.    !!")
+    logger.warning("!! If you are accessing your system using WiFi, you will lose connection. !!")
+
+    if args.yes:
+        consent = "y"
+    else:
+        consent = input("Do you want to continue with SNAC configuration? [y/N] ")
+
+    if consent.lower() not in ["y", "yes"]:
+        return Errors.EX_OK
+
+    print("Configuring SNAC mode.")
+    for config in CONFIGS:
+        config.configure(args)
+
+    print("!! A reboot is now required to affect your system configuration. !!")
+    print("!! Login with user 'root' and no password.                       !!")
+
+    return Errors.EX_OK
+
+
+def _verify(args: argparse.Namespace) -> int:
+    """Configure SNAC mode."""
+    print("Validating SNAC mode.")
+    valid = True
+    for config in CONFIGS:
+        new_valid = config.verify(args)
+        valid = valid and new_valid
+
+    if not valid:
+        raise SNACError("SNAC mode is not configured correctly.", Errors.EX_CHECK_FAILURE)
+    return Errors.EX_OK
+
+
+def _parse_args(argv: List[str]) -> argparse.Namespace:
+    """Top level entry point for the command line interface."""
+    parser = argparse.ArgumentParser(description="Utility for enabling SNAC mode on NI Linux RT.")
+
+    subparsers = parser.add_subparsers(help="Commands for SNAC mode.", dest="cmd")
+    subparsers.required = False
+
+    configure_parser = subparsers.add_parser("configure", help="Set SNAC mode")
+    configure_parser.add_argument(
+        "-y",
+        "--yes",
+        action="store_true",
+        help="Consent to changes",
+    )
+    configure_parser.set_defaults(func=_configure)
+
+    verify_parser = subparsers.add_parser("verify", help="Verify SNAC mode configured correctly")
+    verify_parser.set_defaults(func=_verify)
+
+    debug_group = parser.add_argument_group("Debug")
+    debug_group.add_argument(
+        "-v",
+        "--verbose",
+        action="store_true",
+        default=0,
+        help="Print debug information.  Can be repeated for more detailed output.",
+    )
+    debug_group.add_argument(
+        "-n",
+        "--dry-run",
+        action="store_true",
+    )
+
+    parser.add_argument(
+        "-V",
+        "--version",
+        action="store_true",
+        default=0,
+        help="Print version.",
+    )
+
+    args = parser.parse_args(argv)
+
+    return args
+
+
+def main(  # noqa: D103 - Missing docstring in public function (auto-generated noqa)
+    argv: Optional[List[str]] = None,
+):
+    if argv is None:
+        argv = sys.argv
+    args = _parse_args(argv[1:])
+
+    log_level = logging.INFO
+    if args.verbose:
+        log_level = logging.DEBUG
+    args.logging_level = log_level
+
+    log_format = "(%(relativeCreated)5d) %(levelname)-5s %(name)s.%(funcName)s: %(message)s"
+    logging.basicConfig(level=args.logging_level, format=log_format)
+
+    logger.debug("Arguments: %s", args)
+    opkg_helper.set_dry_run(args.dry_run)
+
+    if args.version:
+        print(VERSION_DESCRIPTION)
+        return Errors.EX_OK
+
+    if args.cmd is None:
+        logger.error("Command required: {configure, verify}, see --help for more information.")
+        return Errors.EX_USAGE
+
+    try:
+        if not args.dry_run:
+            verify_prereqs()
+        ret_val = args.func(args)
+    except SNACError as e:
+        logger.error(e)
+        return e.return_code
+
+    return ret_val
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))

nilrt_snac/_common.py

@@ -0,0 +1,12 @@
+import pathlib
+
+def get_distro():
+    try:
+        os_release = pathlib.Path("/etc/os-release")
+        if os_release.exists():
+            with open(os_release, "r") as f:
+                for line in f:
+                    if line.startswith("ID="):
+                        return line.split("=")[1].strip()
+    except NameError:
+        return None

nilrt_snac/_configs/__init__.py

@@ -0,0 +1,29 @@
+from typing import List
+
+from nilrt_snac._configs._base_config import _BaseConfig
+from nilrt_snac._configs._console_config import _ConsoleConfig
+from nilrt_snac._configs._cryptsetup_config import _CryptSetupConfig
+from nilrt_snac._configs._faillock_config import _FaillockConfig
+from nilrt_snac._configs._niauth_config import _NIAuthConfig
+from nilrt_snac._configs._ntp_config import _NTPConfig
+from nilrt_snac._configs._opkg_config import _OPKGConfig
+from nilrt_snac._configs._pwquality_config import _PWQualityConfig
+from nilrt_snac._configs._wifi_config import _WIFIConfig
+from nilrt_snac._configs._wireguard_config import _WireguardConfig
+from nilrt_snac._configs._x11_config import _X11Config
+
+# USBGuard is not supported for 1.0, but may be added in the future
+# from nilrt_snac._configs._usbguard_config import _USBGuardConfig
+
+CONFIGS: List[_BaseConfig] = [
+    _NTPConfig(),
+    _OPKGConfig(),
+    _WireguardConfig(),
+    _CryptSetupConfig(),
+    _NIAuthConfig(),
+    _WIFIConfig(),
+    _FaillockConfig(),
+    _X11Config(),
+    _ConsoleConfig(),
+    _PWQualityConfig(),
+]

nilrt_snac/_configs/_base_config.py

@@ -0,0 +1,13 @@
+import argparse
+from abc import ABC, abstractmethod
+
+
+class _BaseConfig(ABC):
+
+    @abstractmethod
+    def configure(self, args: argparse.Namespace) -> None:
+        raise NotImplementedError
+
+    @abstractmethod
+    def verify(self, args: argparse.Namespace) -> bool:
+        raise NotImplementedError