feature: add support for opkg (OpenWrt's package manager)

.typos.toml

@@ -1,6 +1,7 @@
 [files]
 extend-exclude = [
     "tests/facts/apt.SimulateOperationWillChange/upgrade.json",
+    "tests/facts/opkg.OpkgPackages/opkg_packages.json",
     "tests/words.txt",
 ]
 

pyinfra/facts/opkg.py

@@ -0,0 +1,233 @@
+"""
+Gather the information provided by ``opkg`` on OpenWrt systems:
+    + ``opkg`` configuration
+    + feeds configuration
+    + list of installed packages
+    + list of packages with available upgrades
+
+
+    see https://openwrt.org/docs/guide-user/additional-software/opkg
+"""
+
+import re
+from typing import Dict, NamedTuple, Union
+
+from pyinfra import logger
+from pyinfra.api import FactBase
+from pyinfra.facts.util.packaging import parse_packages
+
+# TODO - change NamedTuple to dataclass Opkgbut need to figure out how to get json serialization
+#        to work without changing core code
+
+
+class OpkgPkgUpgradeInfo(NamedTuple):
+    installed: str
+    available: str
+
+
+class OpkgConfInfo(NamedTuple):
+    paths: Dict[str, str]  # list of paths, e.g. {'root':'/', 'ram':'/tmp}
+    list_dir: str  # where package lists are stored, e.g. /var/opkg-lists
+    options: Dict[
+        str, Union[str, bool]
+    ]  # mapping from option to value, e.g. {'check_signature': True}
+    arch_cfg: Dict[str, int]  # priorities for architectures
+
+
+class OpkgFeedInfo(NamedTuple):
+    url: str  # url for the feed
+    fmt: str  # format of the feed, e.g. "src/gz"
+    kind: str  # whether it comes from the 'distribution' or is 'custom'
+
+
+class OpkgConf(FactBase):
+    """
+    Returns a NamedTuple with the current configuration:
+
+    .. code:: python
+
+        ConfInfo(
+            paths = {
+                "root": "/",
+                "ram": "/tmp",
+            },
+            list_dir = "/opt/opkg-lists",
+            options = {
+                "overlay_root": "/overlay"
+            },
+            arch_cfg = {
+                "all": 1,
+                "noarch": 1,
+                "i386_pentium": 10
+            }
+        )
+
+    """
+
+    regex = re.compile(
+        r"""
+                       ^(?:\s*)
+                       (?:
+                       (?:arch\s+(?P<arch>\w+)\s+(?P<priority>\d+))|
+                       (?:dest\s+(?P<dest>\w+)\s+(?P<dest_path>[\w/\-]+))|
+                       (?:lists_dir\s+(?P<lists_dir>ext)\s+(?P<list_path>[\w/\-]+))|
+                       (?:option\s+(?P<option>\w+)(?:\s+(?P<value>[^#]+))?)
+                       )?
+                       (?:\s*\#.*)?
+                       $
+                       """,
+        re.X,
+    )
+
+    @staticmethod
+    def default():
+        return OpkgConfInfo({}, "", {}, {})
+
+    def command(self) -> str:
+        return "cat /etc/opkg.conf"
+
+    def process(self, output):
+        dest, lists_dir, options, arch_cfg = {}, "", {}, {}
+        for line in output:
+            match = self.regex.match(line)
+
+            if match is None:
+                logger.warning(f"Opkg: could not parse opkg.conf line '{line}'")
+            elif match.group("arch") is not None:
+                arch_cfg[match.group("arch")] = int(match.group("priority"))
+            elif match.group("dest") is not None:
+                dest[match.group("dest")] = match.group("dest_path")
+            elif match.group("lists_dir") is not None:
+                lists_dir = match.group("list_path")
+            elif match.group("option") is not None:
+                options[match.group("option")] = match.group("value") or True
+
+        return OpkgConfInfo(dest, lists_dir, options, arch_cfg)
+
+
+class OpkgFeeds(FactBase):
+    """
+    Returns a dictionary containing the information for the distribution-provided and
+    custom opkg feeds:
+
+    .. code:: python
+
+        {
+         'openwrt_base': FeedInfo(url='http://downloads ... /i386_pentium/base', fmt='src/gz', kind='distribution'), # noqa: E501
+         'openwrt_core': FeedInfo(url='http://downloads ... /x86/geode/packages', fmt='src/gz', kind='distribution'), # noqa: E501
+         'openwrt_luci': FeedInfo(url='http://downloads ... /i386_pentium/luci', fmt='src/gz', kind='distribution'),# noqa: E501
+         'openwrt_packages': FeedInfo(url='http://downloads ... /i386_pentium/packages', fmt='src/gz', kind='distribution'),# noqa: E501
+         'openwrt_routing': FeedInfo(url='http://downloads ... /i386_pentium/routing', fmt='src/gz', kind='distribution'),# noqa: E501
+         'openwrt_telephony': FeedInfo(url='http://downloads ... /i386_pentium/telephony', fmt='src/gz', kind='distribution') # noqa: E501
+        }
+    """
+
+    regex = re.compile(
+        r"^(CUSTOM)|(?:\s*(?P<fmt>[\w/]+)\s+(?P<name>[\w]+)\s+(?P<url>[\w./:]+))?(?:\s*#.*)?$"
+    )
+    default = dict
+
+    def command(self) -> str:
+        return "cat /etc/opkg/distfeeds.conf; echo CUSTOM; cat /etc/opkg/customfeeds.conf"
+
+    def process(self, output):
+        feeds, kind = {}, "distribution"
+        for line in output:
+            match = self.regex.match(line)
+
+            if match is None:
+                logger.warning(f"Opkg: could not parse /etc/opkg/*feeds.conf line '{line}'")
+            elif match.group(0) == "CUSTOM":
+                kind = "custom"
+            elif match.group("name") is not None:
+                feeds[match.group("name")] = OpkgFeedInfo(
+                    match.group("url"), match.group("fmt"), kind
+                )
+
+        return feeds
+
+
+class OpkgInstallableArchitectures(FactBase):
+    """
+    Returns a dictionary containing the currently installable architectures for this system along
+    with their priority:
+
+    .. code:: python
+
+       {
+         'all': 1,
+         'i386_pentium': 10,
+         'noarch': 1
+        }
+    """
+
+    regex = re.compile(r"^(?:\s*arch\s+(?P<arch>[\w]+)\s+(?P<prio>\d+))?(\s*#.*)?$")
+    default = dict
+
+    def command(self) -> str:
+        return "/bin/opkg print-architecture"
+
+    def process(self, output):
+        arch_list = {}
+        for line in output:
+            match = self.regex.match(line)
+
+            if match is None:
+                logger.warning(f"could not parse arch line '{line}'")
+            elif match.group("arch") is not None:
+                arch_list[match.group("arch")] = int(match.group("prio"))
+
+        return arch_list
+
+
+class OpkgPackages(FactBase):
+    """
+    Returns a dict of installed opkg packages:
+
+    .. code:: python
+
+       {
+         'package_name': ['version'],
+         ...
+       }
+    """
+
+    regex = r"^([a-zA-Z0-9][\w\-\.]*)\s-\s([\w\-\.]+)"
+    default = dict
+
+    def command(self) -> str:
+        return "/bin/opkg list-installed"
+
+    def process(self, output):
+        return parse_packages(self.regex, sorted(output))
+
+
+class OpkgUpgradeablePackages(FactBase):
+    """
+    Returns a dict of installed and upgradable opkg packages:
+
+    .. code:: python
+
+        {
+          'package_name': (installed='1.2.3', available='1.2.8')
+          ...
+        }
+    """
+
+    regex = re.compile(r"^([a-zA-Z0-9][\w\-.]*)\s-\s([\w\-.]+)\s-\s([\w\-.]+)")
+    default = dict
+    use_default_on_error = True
+
+    def command(self) -> str:
+        return "/bin/opkg list-upgradable"  # yes, really spelled that way
+
+    def process(self, output):
+        result = {}
+        for line in output:
+            match = self.regex.match(line)
+            if match and len(match.groups()) == 3:
+                result[match.group(1)] = OpkgPkgUpgradeInfo(match.group(2), match.group(3))
+            else:
+                logger.warning(f"Opkg: could not list-upgradable line '{line}'")
+
+        return result

pyinfra/operations/opkg.py

@@ -0,0 +1,88 @@
+"""
+    Manage packages on OpenWrt using opkg
+        + ``update`` - update local copy of package information
+        + ``packages`` -  install and remove packages
+
+    see https://openwrt.org/docs/guide-user/additional-software/opkg
+    OpenWrt recommends against upgrading all packages  thus there is no ``opkg.upgrade`` function
+"""
+
+from typing import List, Union
+
+from pyinfra import host
+from pyinfra.api import StringCommand, operation
+from pyinfra.facts.opkg import OpkgPackages
+from pyinfra.operations.util.packaging import ensure_packages
+
+EQUALS = "="
+
+
+@operation(is_idempotent=False)
+def update():
+    """
+    Update the local opkg information.
+    """
+
+    yield StringCommand("opkg update")
+
+
+_update = update
+
+
+@operation()
+def packages(
+    packages: Union[str, List[str]] = "",
+    present: bool = True,
+    latest: bool = False,
+    update: bool = True,
+):
+    """
+    Add/remove/update opkg packages.
+
+    + packages: package or list of packages to that must/must not be present
+    + present: whether the package(s) should be installed (default True) or removed
+    + latest: whether to attempt to upgrade the specified package(s) (default False)
+    + update: run ``opkg update`` before installing packages (default True)
+
+    Not Supported:
+        Opkg does not support version pinning, i.e. ``<pkg>=<version>`` is not allowed
+        and will cause an exception.
+
+    **Examples:**
+
+    .. code:: python
+
+        # Ensure packages are installed∂ (will not force package upgrade)
+        opkg.packages(['asterisk', 'vim'], name="Install Asterisk and Vim")
+
+        # Install the latest versions of packages (always check)
+        opkg.packages(
+            'vim',
+            latest=True,
+            name="Ensure we have the latest version of Vim"
+        )
+    """
+    if str(packages) == "" or (
+        isinstance(packages, list) and (len(packages) < 1 or all(len(p) < 1 for p in packages))
+    ):
+        host.noop("empty or invalid package list provided to opkg.packages")
+        return
+
+    pkg_list = packages if isinstance(packages, list) else [packages]
+    have_equals = ",".join([pkg.split(EQUALS)[0] for pkg in pkg_list if EQUALS in pkg])
+    if len(have_equals) > 0:
+        raise ValueError(f"opkg does not support version pinning but found for: '{have_equals}'")
+
+    if update:
+        yield from _update._inner()
+
+    yield from ensure_packages(
+        host,
+        pkg_list,
+        host.get_fact(OpkgPackages),
+        present,
+        install_command="opkg install",
+        upgrade_command="opkg upgrade",
+        uninstall_command="opkg remove",
+        latest=latest,
+    )

tests/facts/opkg.OpkgConf/opkg_conf.json

@@ -0,0 +1,42 @@
+{
+    "command": "cat /etc/opkg.conf",
+    "output": [
+        "",
+        "# a comment",
+        "    # another comment",
+        "extra",
+        "dest root /",
+        "dest ram /tmp",
+        "lists_dir ext /var/opkg-lists",
+        "list_dir ext /var/foo/bar",
+        "list_dir zap /var/foo/bar",
+        "option",
+        "option overlay_root /overlay",
+        "option check_signature",
+        "option http_proxy http://username:password@proxy.example.org:8080/",
+        "option ftp_proxy http://username:password@proxy.example.org:2121/",
+        "arch all 1",
+        "arch noarch 2",
+        "arch brcm4716 200",
+        "arch brcm47xx 300  # generic has lower priority than specific",
+        "arch zzz"
+    ],
+    "fact": [
+        {
+            "root": "/",
+            "ram": "/tmp"
+        },
+        "/var/opkg-lists",
+        {
+            "overlay_root": "/overlay",
+            "check_signature": true,
+            "http_proxy": "http://username:password@proxy.example.org:8080/",
+            "ftp_proxy": "http://username:password@proxy.example.org:2121/"        },
+        {
+            "all": 1,
+            "noarch": 2,
+            "brcm4716": 200,
+            "brcm47xx": 300
+        }
+    ]
+}