Refactor static analyses and config patchers into pyplugins

pyplugins/config_patchers/base_patch.py

@@ -0,0 +1,182 @@
+import os
+from penguin.static_plugin import ConfigPatcherPlugin
+from penguin import getColoredLogger
+from penguin.arch import arch_end
+from penguin.defaults import default_init_script, default_plugins, static_dir as STATIC_DIR
+from penguin.utils import get_arch_subdir
+
+logger = getColoredLogger("penguin.config_patchers")
+
+import penguin
+RESOURCES = os.path.join(os.path.dirname(penguin.__file__), "resources")
+
+class BasePatch(ConfigPatcherPlugin):
+    """
+    Generate base config for static_files and default plugins
+    """
+    depends_on = ['ArchId', 'InitFinder', 'KernelVersionFinder']
+    UNKNOWN_INIT: str = "UNKNOWN_FIX_ME"
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.enabled = True
+        self.patch_name = "base"
+
+    def generate(self, patches: dict) -> dict:
+        arch_info = self.prior_results.get('ArchId')
+        inits = self.prior_results.get('InitFinder', [])
+        self.kernel_versions = self.prior_results.get('KernelVersionFinder', {"selected_kernel": ""})
+
+        self.set_arch_info(arch_info)
+
+        if len(inits):
+            self.igloo_init = inits[0]
+        else:
+            self.igloo_init = self.UNKNOWN_INIT
+            logger.warning("Failed to find any init programs - config will need manual refinement")
+
+        if 'mips' in self.arch_name or self.arch_name == "intel64":
+            igloo_serial_major = 4
+            igloo_serial_minor = 65
+        elif self.arch_name in ['armel', 'aarch64']:
+            igloo_serial_major = 204
+            igloo_serial_minor = 65
+        elif "powerpc" in self.arch_name:
+            igloo_serial_major = 229
+            igloo_serial_minor = 1
+        elif self.arch_name == "loongarch64":
+            igloo_serial_major = 4
+            igloo_serial_minor = 65
+        else:
+            igloo_serial_major = 204
+            igloo_serial_minor = 65
+
+        result = {
+            "core": {
+                "arch": self.arch_name,
+                "kernel": self.kernel_versions["selected_kernel"],
+            },
+            "env": {
+                "igloo_init": self.igloo_init,
+            },
+            "pseudofiles": {
+                "/dev/ttyS1": {
+                    "read": {
+                        "model": "zero",
+                    },
+                    "write": {
+                        "model": "discard",
+                    },
+                    "ioctl": {
+                        "*": {
+                            "model": "return_const",
+                            "val": 0,
+                        }
+                    }
+                },
+                "/dev/ttyAMA1": {
+                    "read": {
+                        "model": "zero",
+                    },
+                    "write": {
+                        "model": "discard",
+                    },
+                    "ioctl": {
+                        "*": {
+                            "model": "return_const",
+                            "val": 0,
+                        }
+                    }
+                }
+            },
+            "static_files": {
+                "/igloo/init": {
+                    "type": "inline_file",
+                    "contents": default_init_script,
+                    "mode": 0o111,
+                },
+                "/igloo/utils/sh": {
+                    "type": "symlink",
+                    "target": "/igloo/utils/busybox",
+                },
+                "/igloo/utils/sleep": {
+                    "type": "symlink",
+                    "target": "/igloo/utils/busybox",
+                },
+                "/igloo/ltrace/*": {
+                    "type": "host_file",
+                    "mode": 0o444,
+                    "host_path": os.path.join(*[STATIC_DIR, "ltrace", "*"]),
+                },
+                "/igloo/dylibs/*": {
+                    "type": "host_file",
+                    "mode": 0o755,
+                    "host_path": os.path.join(STATIC_DIR, "dylibs", self.dylib_dir or self.arch_dir, "*"),
+                },
+                "/igloo/source.d/*": {
+                    "type": "host_file",
+                    "mode": 0o755,
+                    "host_path": os.path.join(*[RESOURCES, "source.d", "*"]),
+                },
+                "/igloo/serial": {
+                    "type": "dev",
+                    "devtype": "char",
+                    "major": igloo_serial_major,
+                    "minor": igloo_serial_minor,
+                    "mode": 0o666,
+                }
+            },
+            "plugins": default_plugins,
+        }
+
+        guest_scripts_dir = os.path.join(STATIC_DIR, "guest-utils", "scripts")
+        for f in os.listdir(guest_scripts_dir):
+            result["static_files"][f"/igloo/utils/{f}"] = {
+                "type": "host_file",
+                "host_path": f"{guest_scripts_dir}/{f}",
+                "mode": 0o755,
+            }
+        result["static_files"]["/igloo/utils/*"] = {
+            "type": "host_file",
+            "host_path": f"{STATIC_DIR}/{self.arch_dir}/*",
+            "mode": 0o755,
+        }
+
+        return result
+
+    def set_arch_info(self, arch_identified: str) -> None:
+        arch, endian = arch_end(arch_identified)
+        if arch is None:
+            raise NotImplementedError(f"Architecture {arch_identified} not supported ({arch}, {endian})")
+
+        if arch == "aarch64":
+            self.arch_name = "aarch64"
+        elif arch == "intel64":
+            self.arch_name = "intel64"
+        elif arch == "loongarch64":
+            self.arch_name = "loongarch64"
+        elif arch == "riscv64":
+            self.arch_name = "riscv64"
+        elif arch == "powerpc":
+            self.arch_name = "powerpc"
+        elif arch == "powerpc64":
+            if endian == "el":
+                self.arch_name = "powerpc64le"
+            else:
+                self.arch_name = "powerpc64"
+        else:
+            self.arch_name = arch + endian
+
+        mock_config = {"core": {"arch": self.arch_name}}
+        self.arch_dir = get_arch_subdir(mock_config)
+
+        if arch_identified == "aarch64":
+            self.dylib_dir = "arm64"
+        elif arch_identified == "intel64":
+            self.dylib_dir = "x86_64"
+        elif arch_identified == "loongarch64":
+            self.dylib_dir = "loongarch"
+        elif "powerpc" in self.arch_name:
+            self.dylib_dir = self.arch_name.replace("powerpc", "ppc")
+        else:
+            self.dylib_dir = self.arch_dir

pyplugins/config_patchers/delete_files.py

@@ -0,0 +1,18 @@
+import os
+from collections import defaultdict
+from penguin.static_plugin import ConfigPatcherPlugin
+
+class DeleteFiles(ConfigPatcherPlugin):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.patch_name = "static.delete_files"
+        self.enabled = True
+
+    def generate(self, patches: dict) -> dict:
+        result = defaultdict(dict)
+        for f in ["/etc/securetty", "/etc/scripts/sys_resetbutton"]:
+            if os.path.isfile(os.path.join(self.extracted_fs, f[1:])):
+                result["static_files"][f] = {
+                    "type": "delete",
+                }
+        return result

pyplugins/config_patchers/dynamic_exploration.py

@@ -0,0 +1,28 @@
+from penguin.static_plugin import ConfigPatcherPlugin
+
+class DynamicExploration(ConfigPatcherPlugin):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.patch_name = "auto_explore"
+        self.enabled = False
+
+    def generate(self, patches: dict) -> dict:
+        return {
+            "core": {
+                "root_shell": False,
+            },
+            "plugins": {
+                "nmap": {
+                    "enabled": True,
+                },
+                "vpn": {
+                    "enabled": True,
+                    "log": True,
+                },
+                "netbinds":
+                {
+                    "enabled": True,
+                    "shutdown_on_www": False,
+                },
+            }
+        }

pyplugins/config_patchers/file_helper.py

@@ -0,0 +1,67 @@
+import os
+import subprocess
+import re
+from pathlib import Path
+
+class FileHelper:
+    @staticmethod
+    def find_executables(tmp_dir: str, target_dirs: set[str] | None = None):
+        if not target_dirs:
+            target_dirs = {"/"}
+        for root, _, files in os.walk(tmp_dir):
+            if "/igloo" in root:
+                continue
+            for file in files:
+                file_path = Path(root) / file
+                if (
+                    file_path.is_file()
+                    and os.access(file_path, os.X_OK)
+                    and any(str(file_path).endswith(d) for d in target_dirs)
+                ):
+                    yield file_path
+
+    @staticmethod
+    def find_strings_in_file(file_path: str, pattern: str) -> list[str]:
+        result = subprocess.run(["strings", file_path], capture_output=True, text=True)
+        return [line for line in result.stdout.splitlines() if re.search(pattern, line)]
+
+    @staticmethod
+    def find_shell_scripts(tmp_dir: str):
+        for root, _, files in os.walk(tmp_dir):
+            if "/igloo" in root:
+                continue
+            for file in files:
+                file_path = Path(root) / file
+                if (
+                    file_path.is_file()
+                    and os.access(file_path, os.X_OK)
+                    and str(file_path).endswith(".sh")
+                ):
+                    yield file_path
+
+    @staticmethod
+    def exists(tmp_dir: str, target: str) -> bool:
+        assert target.startswith("/")
+        assert os.path.exists(tmp_dir)
+
+        target = target[1:]
+        parts = target.split("/")
+
+        current_path = tmp_dir
+
+        for part in parts:
+            next_path = os.path.join(current_path, part)
+
+            if os.path.islink(next_path):
+                resolved = os.readlink(next_path)
+                if resolved.startswith("/"):
+                    current_path = os.path.realpath(os.path.join(tmp_dir, resolved[1:]))
+                else:
+                    current_path = os.path.realpath(os.path.join(current_path, resolved))
+            else:
+                current_path = next_path
+
+            if not os.path.exists(current_path):
+                return False
+
+        return os.path.exists(current_path)

pyplugins/config_patchers/force_www.py

@@ -0,0 +1,64 @@
+import os
+from penguin.static_plugin import ConfigPatcherPlugin
+
+class ForceWWW(ConfigPatcherPlugin):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.enabled = False
+        self.patch_name = 'force_www'
+
+    def generate(self, patches: dict) -> dict | None:
+        file2cmd = {
+            "./etc/init.d/uhttpd": "/etc/init.d/uhttpd start",
+            "./usr/bin/httpd": "/usr/bin/httpd",
+            "./usr/sbin/httpd": "/usr/sbin/httpd",
+            "./bin/goahead": "/bin/goahead",
+            "./bin/alphapd": "/bin/alphapd",
+            "./bin/boa": "/bin/boa",
+            "./usr/sbin/lighttpd": "/usr/sbin/lighttpd -f /etc/lighttpd/lighttpd.conf",
+        }
+
+        www_cmds = []
+        www_paths = []
+
+        have_lighttpd_conf = os.path.isfile(os.path.join(self.extracted_fs, "./etc/lighttpd/lighttpd.conf"))
+
+        for file, cmd in file2cmd.items():
+            if os.path.isfile(os.path.join(self.extracted_fs, file)):
+                if file == "./usr/sbin/lighttpd" and not have_lighttpd_conf:
+                    continue
+                www_cmds.append(cmd)
+                www_paths.append(file)
+
+        if not len(www_cmds):
+            return
+
+        cmd_str = """#!/igloo/utils/sh
+        /igloo/utils/busybox sleep 120
+
+        while true; do
+        """
+
+        for cmd in www_cmds:
+            cmd_str += f"""
+            if ! (/igloo/utils/busybox ps | /igloo/utils/busybox grep -v grep | /igloo/utils/busybox grep -sqi "{cmd}"); then
+                {cmd} &
+            fi
+        """
+        cmd_str += """
+            /igloo/utils/busybox sleep 30
+            done
+        """
+
+        return {
+            "core": {
+                'force_www': True
+            },
+            "static_files": {
+                "/igloo/utils/www_cmds": {
+                    "type": "inline_file",
+                    "contents": cmd_str,
+                    "mode": 0o755,
+                }
+            }
+        }