#!/usr/bin/env python3
"""Regression guards for the next 009 release-key bacon."""

from pathlib import Path
import os
import subprocess
import tempfile
import unittest


ROOT = Path(__file__).resolve().parents[1]
KH = ROOT / "vendor/sunfish-a16/patches/009-kernel-hardening"
SHIP = ROOT / "scripts/009-bacon-axp-stage1-combo4-cve-hi-releasekeys.sh"


class NextBaconGuardsTest(unittest.TestCase):
    def test_stage1_refuses_reverse_application(self) -> None:
        apply_script = (KH / "apply_stage1.sh").read_text()
        self.assertIn('args=(-p1 --batch --forward -d "$KERNEL_DIR")', apply_script)

    def test_stage1_does_not_revert_existing_clang_cfi_kallsyms_fix(self) -> None:
        stage1 = (KH / "stage1.list").read_text().splitlines()
        self.assertNotIn(
            "0008-Graphene-Kernel_Hardening-misc/4.14/0021.patch",
            stage1,
        )

    def test_default_avb_flags_stay_3_and_zero_is_isolated(self) -> None:
        ship = SHIP.read_text()
        wrapper = (ROOT / "scripts/009-bacon-avb-flags0.sh").read_text()
        self.assertIn('AXP_AVB_FLAGS="${AXP_AVB_FLAGS:-3}"', ship)
        self.assertIn('[[ "$FLAGS" == "$AXP_AVB_FLAGS" ]]', ship)
        self.assertNotIn('expected vbmeta Flags 3', ship)
        self.assertIn("export AXP_AVB_FLAGS=0", wrapper)
        self.assertIn("export WITH_AVB=true", wrapper)

    def test_ship_script_applies_axp_branding(self) -> None:
        ship = SHIP.read_text()
        self.assertIn("apply_009_axp_branding.py", ship)
        self.assertLess(
            ship.index("apply_009_axp_identity.py"),
            ship.index("apply_009_axp_branding.py"),
        )

    def test_ship_script_applies_axp_timeouts(self) -> None:
        ship = SHIP.read_text()
        self.assertIn("apply_009_a15_reboot_timeouts.py", ship)
        self.assertIn("apply_009_a15_settings_timeouts.py", ship)
        self.assertLess(
            ship.index("apply_009_a15_reboot_timeouts.py"),
            ship.index("apply_009_a15_settings_timeouts.py"),
        )
        self.assertIn("settings_reboot_after_timeout", ship)
        self.assertIn("isAllowedWhenLocked", ship)
        self.assertIn("wifi_timeout", ship)

    def test_ship_script_hides_proc_cmdline(self) -> None:
        ship = SHIP.read_text()
        self.assertIn("apply_009_cmdline_mode.py", ship)
        self.assertIn('CHECKCALL(chmod("/proc/cmdline", 0440));', ship)

    def test_ship_script_reapplies_complete_feature_008(self) -> None:
        ship = SHIP.read_text()
        self.assertIn(
            "python3 vendor/sunfish-a16/patches/008-call-rec-flag-secure/"
            "apply_008_call_rec_flag_secure.py",
            ship,
        )
        self.assertNotIn(
            "python3 vendor/sunfish-a16/patches/008-call-rec-flag-secure/"
            "apply_wms_disable_secure_debuggable_gate.py",
            ship,
        )

    def test_ship_script_gates_kallsyms_and_call_recording(self) -> None:
        ship = SHIP.read_text()
        self.assertIn("#define KSYM_NAME_LEN 192", ship)
        self.assertIn("SUNFISH_A16_008_CALL_REC_FLAG_SECURE", ship)
        self.assertIn("extendrom_call_recording", ship)

    def test_artifact_metadata_names_49980_and_vendor_spl(self) -> None:
        ship = SHIP.read_text()
        self.assertIn(
            "CVE-2022-49980-lock-scope",
            ship,
        )
        self.assertIn(
            "vendor_security_patch=2023-09-05",
            ship,
        )

    def test_bundled_apps_wired_via_product_includes_and_magisk(self) -> None:
        includes = (
            ROOT
            / "vendor/sunfish-a16/patches/009-axp-los23-port/apply_009_product_includes.py"
        ).read_text()
        magisk = (ROOT / "vendor/sunfish-a16/magisk.mk").read_text()
        ship = SHIP.read_text()
        apps = (ROOT / "vendor/sunfish-a16/apps.mk").read_text()
        self.assertIn("apply_009_product_includes.py", ship)
        self.assertIn("vendor/sunfish-a16/apps.mk", includes)
        self.assertIn("vendor/sunfish-a16/apps.mk", magisk)
        for name in ("AuroraStore", "Droidify", "FossifyGallery", "OpenEUICC"):
            self.assertIn(name, apps)
        self.assertIn("US-S4-3", ship)
        self.assertIn("system_ext/app/", ship)
        self.assertIn("must not be on product.img", ship)
        android_mk = (ROOT / "vendor/sunfish-a16/Android.mk").read_text()
        for name in ("AuroraStore", "Droidify", "FossifyGallery"):
            self.assertIn(f"LOCAL_MODULE := {name}", android_mk)
        self.assertIn("LOCAL_PREBUILT_JNI_LIBS_arm64", android_mk)
        self.assertIn("gms.mk not inherited?", ship)
        self.assertIn("libconscrypt_gmscore_jni.so", ship)
        self.assertIn("apps.mk not inherited?", ship)

    def test_gmscore_jni_extract_before_mka(self) -> None:
        ship = SHIP.read_text()
        self.assertIn("unzip -q -j -o vendor/sunfish-a16/prebuilt/GmsCore.apk", ship)
        self.assertLess(
            ship.index("unzip -q -j -o vendor/sunfish-a16/prebuilt/GmsCore.apk"),
            ship.index("mka bacon"),
        )

    def test_euicc_google_is_stripped_from_the_vendor_makefile(self) -> None:
        ship = SHIP.read_text()
        apply = (
            ROOT
            / "vendor/sunfish-a16/patches/009-axp-los23-port/apply_009_drop_euicc_google.py"
        ).read_text()
        apps = (ROOT / "vendor/sunfish-a16/apps.mk").read_text()
        self.assertIn("apply_009_drop_euicc_google.py", ship)
        self.assertIn('"EuiccGoogle"', apply)
        self.assertIn('"EuiccGoogleOverlay"', apply)
        self.assertIn("EuiccSupportPixel", apply)
        self.assertNotIn("filter-out EuiccGoogle", apps)

    def test_thermal_cdev_log_skips_unbound_devices(self) -> None:
        ship = SHIP.read_text()
        apply_path = (
            ROOT
            / "vendor/sunfish-a16/patches/009-axp-los23-port/apply_009_thermal_cdev_log.py"
        )
        self.assertIn("apply_009_thermal_cdev_log.py", ship)
        stock = """void ThermalThrottling::logCoolingDeviceStatus(
        const std::unordered_map<std::string, CdevInfo> &cooling_device_info_map) {
    int max_state = 0;
    std::ostringstream cdev_log;
    for (const auto &[cdev_name, cdev_info] : cooling_device_info_map) {
        if (getCdevMaxRequest(cdev_name, &max_state)) {
            ATRACE_INT((cdev_name + std::string("-state")).c_str(), max_state);
            if (!cdev_info.apply_powercap) {
                cdev_log << cdev_name << " state:" << max_state << " ";
            } else {
                const auto budget = static_cast<int>(
                        std::lround(cdev_info.state2power[max_state] / cdev_info.multiplier));
                cdev_log << cdev_name << " state:" << max_state << ",budget:" << budget << " ";
                ATRACE_INT((cdev_name + std::string("-budget")).c_str(), budget);
            }
        }
    }
    LOG(INFO) << "CDEV log " << cdev_log.str();
}
"""
        with tempfile.TemporaryDirectory() as tmp:
            top = Path(tmp)
            path = top / "hardware/google/pixel/thermal/utils/thermal_throttling.cpp"
            path.parent.mkdir(parents=True)
            path.write_text(stock)
            env = {**os.environ, "ANDROID_BUILD_TOP": str(top)}
            first = subprocess.run(
                ["python3", str(apply_path)],
                cwd=top,
                env=env,
                check=False,
                text=True,
                capture_output=True,
            )
            self.assertEqual(0, first.returncode, first.stderr)
            patched = path.read_text()
            self.assertIn("has no HAL throttle votes (external mitigator)", patched)
            self.assertIn("cdev_all_request_map_.count(cdev_name)", patched)
            second = subprocess.run(
                ["python3", str(apply_path)],
                cwd=top,
                env=env,
                check=False,
                text=True,
                capture_output=True,
            )
            self.assertEqual(0, second.returncode, second.stderr)
            self.assertIn("already", second.stdout)

    def test_euicc_disabler_guards_missing_packages(self) -> None:
        ship = SHIP.read_text()
        apply_path = (
            ROOT
            / "vendor/sunfish-a16/patches/009-axp-los23-port/apply_009_euicc_disabler_guard.py"
        )
        self.assertIn("apply_009_euicc_disabler_guard.py", ship)
        self.assertLess(
            ship.index("apply_009_drop_euicc_google.py"),
            ship.index("apply_009_euicc_disabler_guard.py"),
        )
        stock = """package org.lineageos.settings;

import android.content.Context;
import android.content.pm.PackageInfo;
import android.content.pm.PackageManager;
import android.util.Log;

class EuiccDisabler {
    private static final String TAG = "GoogleParts";
    private static final String[] EUICC_PACKAGES = new String[]{
        "com.google.android.euicc",
        "com.google.euiccpixel",
        "com.google.android.ims"
    };

    private static boolean shouldDisable(PackageManager pm) {
        return false;
    }

    public static void enableOrDisableEuicc(Context context) {
        PackageManager pm = context.getPackageManager();
        boolean disable = shouldDisable(pm);
        int flag = disable
            ? PackageManager.COMPONENT_ENABLED_STATE_DISABLED
            : PackageManager.COMPONENT_ENABLED_STATE_ENABLED;
        for (String pkg : EUICC_PACKAGES) {
            pm.setApplicationEnabledSetting(pkg, flag, 0);
        }
    }
}
"""
        with tempfile.TemporaryDirectory() as tmp:
            top = Path(tmp)
            path = (
                top
                / "device/google/sunfish/parts/src/org/lineageos/settings/EuiccDisabler.java"
            )
            path.parent.mkdir(parents=True)
            path.write_text(stock)
            env = {**os.environ, "ANDROID_BUILD_TOP": str(top)}
            first = subprocess.run(
                ["python3", str(apply_path)],
                cwd=top,
                env=env,
                check=False,
                text=True,
                capture_output=True,
            )
            self.assertEqual(0, first.returncode, first.stderr)
            patched = path.read_text()
            self.assertIn("isInstalled(pm, pkg)", patched)
            self.assertIn("Do not retarget to OpenEUICC", patched)
            self.assertNotIn(
                "for (String pkg : EUICC_PACKAGES) {\n"
                "            pm.setApplicationEnabledSetting(pkg, flag, 0);",
                patched,
            )
            second = subprocess.run(
                ["python3", str(apply_path)],
                cwd=top,
                env=env,
                check=False,
                text=True,
                capture_output=True,
            )
            self.assertEqual(0, second.returncode, second.stderr)
            self.assertIn("already", second.stdout)

    def test_carrier_google_is_stripped_where_it_is_added(self) -> None:
        ship = SHIP.read_text()
        apply = (
            ROOT
            / "vendor/sunfish-a16/patches/009-axp-los23-port/apply_009_drop_carrier_google.py"
        ).read_text()
        self.assertIn("apply_009_drop_carrier_google.py", ship)
        for name in (
            "CarrierServices",
            "CarrierSettings",
            "CarrierWifi",
            "CarrierSetup",
            "RilConfigService",
            "grilservice",
            "ImsServiceEntitlement",
            "PresencePolling",
            "RcsService",
        ):
            self.assertIn(f'"{name}"', apply)
        for name in ("EuiccSupportPixel", "ims", "atfwd", "qcrilmsgtunnel", "Dialer"):
            self.assertIn(f'"{name}"', apply)

    def test_carrier_drop_removes_package_lines_and_is_idempotent(self) -> None:
        apply = (
            ROOT
            / "vendor/sunfish-a16/patches/009-axp-los23-port/apply_009_drop_carrier_google.py"
        )
        with tempfile.TemporaryDirectory() as tmp:
            top = Path(tmp)
            vendor = top / "vendor/google/sunfish"
            device = top / "device/google/sunfish"
            product = top / "build/make/target/product"
            for directory in (vendor, device, product):
                directory.mkdir(parents=True)
            (vendor / "sunfish-vendor.mk").write_text(
                "PRODUCT_PACKAGES += \\\n"
                "    CarrierServices \\\n"
                "    CarrierSettings \\\n"
                "    CarrierWifi \\\n"
                "    CarrierSetup \\\n"
                "    EuiccSupportPixel \\\n"
                "    EuiccSupportPixelPermissions \\\n"
                "    ims \\\n"
                "    QtiTelephonyService \\\n"
                "    atfwd \\\n"
                "    qcrilmsgtunnel \\\n"
                "    uceShimService \\\n"
                "    RilConfigService \\\n"
                "    grilservice \\\n"
                "    vendor/google/sunfish/proprietary/product/etc/CarrierSettings/vodafone_de.pb:$(TARGET_COPY_OUT_PRODUCT)/etc/CarrierSettings/vodafone_de.pb \\\n"
            )
            (device / "device-lineage.mk").write_text(
                "PRODUCT_PACKAGES += \\\n"
                "    GoogleParts\n"
                "\n"
                "# RCS\n"
                "PRODUCT_PACKAGES += \\\n"
                "    PresencePolling \\\n"
                "    RcsService\n"
                "\n"
                "# Touch\n"
            )
            (product / "telephony_product.mk").write_text(
                "PRODUCT_PACKAGES += \\\n"
                "    Dialer \\\n"
                "    ImsServiceEntitlement \\\n"
                "    preinstalled-packages-platform-telephony-product.xml\n"
            )
            env = {**os.environ, "ANDROID_BUILD_TOP": str(top)}
            first = subprocess.run(
                ["python3", str(apply)], cwd=top, env=env, check=False, text=True, capture_output=True
            )
            self.assertEqual(0, first.returncode, first.stderr)
            self.assertIn("dropped", first.stdout)
            vendor_text = (vendor / "sunfish-vendor.mk").read_text()
            self.assertNotIn("CarrierServices", vendor_text)
            self.assertIn("vodafone_de.pb", vendor_text)
            self.assertIn("    ims \\\n", vendor_text)
            lineage_text = (device / "device-lineage.mk").read_text()
            self.assertNotIn("RcsService", lineage_text)
            self.assertNotIn("PRODUCT_PACKAGES += \\\n\n", lineage_text)
            self.assertIn("GoogleParts", lineage_text)
            tele_text = (product / "telephony_product.mk").read_text()
            self.assertNotIn("ImsServiceEntitlement", tele_text)
            self.assertIn("Dialer", tele_text)
            second = subprocess.run(
                ["python3", str(apply)], cwd=top, env=env, check=False, text=True, capture_output=True
            )
            self.assertEqual(0, second.returncode, second.stderr)
            self.assertIn("already", second.stdout)

    def test_lineage_tools_are_stripped_where_they_are_added(self) -> None:
        ship = SHIP.read_text()
        apply = (
            ROOT
            / "vendor/sunfish-a16/patches/009-axp-los23-port/apply_009_drop_lineage_tools.py"
        )
        text = apply.read_text()
        self.assertIn("apply_009_drop_lineage_tools.py", ship)
        for name in ("bash", "htop", "nano", "50-lineage.sh", "unrar"):
            self.assertIn(f'"{name}"', text)
        for name in ("curl", "vim", "zstd", "nano_recovery"):
            self.assertIn(f'"{name}"', text)
        with tempfile.TemporaryDirectory() as tmp:
            top = Path(tmp)
            common = top / "vendor/lineage/config"
            common.mkdir(parents=True)
            (common / "common.mk").write_text(
                "PRODUCT_PACKAGES += \\\n"
                "    50-lineage.sh\n"
                "\n"
                "PRODUCT_ARTIFACT_PATH_REQUIREMENT_ALLOWED_LIST += \\\n"
                "    system/addon.d/50-lineage.sh\n"
                "\n"
                "PRODUCT_PACKAGES += \\\n"
                "    bash \\\n"
                "    curl \\\n"
                "    getcap \\\n"
                "    htop \\\n"
                "    nano \\\n"
                "    setcap \\\n"
                "    vim\n"
                "\n"
                "PRODUCT_PACKAGES += \\\n"
                "    nano_recovery\n"
            )
            (common / "common_mobile_full.mk").write_text(
                "PRODUCT_PACKAGES += \\\n"
                "    unrar \\\n"
                "    zstd\n"
            )
            env = {**os.environ, "ANDROID_BUILD_TOP": str(top)}
            first = subprocess.run(
                ["python3", str(apply)],
                cwd=top,
                env=env,
                check=False,
                text=True,
                capture_output=True,
            )
            self.assertEqual(0, first.returncode, first.stderr)
            self.assertIn("dropped", first.stdout)
            common_text = (common / "common.mk").read_text()
            self.assertNotIn("\n    bash \\\n", common_text)
            self.assertNotIn("\n    htop \\\n", common_text)
            self.assertNotIn("\n    nano \\\n", common_text)
            self.assertNotIn("    50-lineage.sh\n", common_text)
            self.assertIn("system/addon.d/50-lineage.sh", common_text)
            self.assertIn("    curl \\\n", common_text)
            self.assertIn("    vim\n", common_text)
            self.assertIn("    nano_recovery\n", common_text)
            full_text = (common / "common_mobile_full.mk").read_text()
            self.assertNotIn("unrar", full_text)
            self.assertIn("    zstd\n", full_text)
            second = subprocess.run(
                ["python3", str(apply)],
                cwd=top,
                env=env,
                check=False,
                text=True,
                capture_output=True,
            )
            self.assertEqual(0, second.returncode, second.stderr)
            self.assertIn("already", second.stdout)

    def test_lab_only_commands_drop_keeps_gps_and_host_pbtombstone(self) -> None:
        ship = SHIP.read_text()
        apply = (
            ROOT
            / "vendor/sunfish-a16/patches/009-axp-los23-port/apply_009_drop_lab_only_commands.py"
        )
        text = apply.read_text()
        self.assertIn("apply_009_drop_lab_only_commands.py", ship)  # may be commented (wave 4 deferred)
        for name in (
            "cnss_diag",
            "diag_mdlog",
            "imsrcsd",
            "bpftool",
            "alloctop",
            "apexd-lsof.sh",
        ):
            self.assertIn(name, text)
        self.assertIn("ensure_disable_configstore", text)
        self.assertIn("disable_configstore", text)
        for name in ("loc_launcher", "lowi-server", "xtra-daemon"):
            self.assertIn(f'"{name}"', text)
        with tempfile.TemporaryDirectory() as tmp:
            top = Path(tmp)
            vendor = top / "vendor/google/sunfish"
            device = top / "device/google/sunfish"
            product = top / "build/make/target/product"
            bpf = top / "system/bpf/loader"
            dump = top / "frameworks/native/cmds/dumpstate"
            apex = top / "system/apex/apexd"
            for directory in (vendor, device, product, bpf, dump, apex):
                directory.mkdir(parents=True)
            (vendor / "sunfish-vendor.mk").write_text(
                "PRODUCT_COPY_FILES += \\\n"
                "    vendor/google/sunfish/proprietary/vendor/etc/init/imsrcsd.rc:$(TARGET_COPY_OUT_VENDOR)/etc/init/imsrcsd.rc \\\n"
                "    vendor/google/sunfish/proprietary/vendor/etc/init/imsdatadaemon.rc:$(TARGET_COPY_OUT_VENDOR)/etc/init/imsdatadaemon.rc \\\n"
                "\n"
                "PRODUCT_PACKAGES += \\\n"
                "    cnss-daemon \\\n"
                "    cnss_diag \\\n"
                "    diag_klog \\\n"
                "    diag_mdlog \\\n"
                "    diag_socket_log \\\n"
                "    diag_uart_log \\\n"
                "    ims_rtp_daemon \\\n"
                "    imsdatadaemon \\\n"
                "    imsqmidaemon \\\n"
                "    imsrcsd \\\n"
                "    loc_launcher \\\n"
                "    lowi-server \\\n"
                "    ssr_diag \\\n"
                "    test_diag \\\n"
                "    xtra-daemon\n"
            )
            # Simulate prior bad wave4: empty Configstore stanza (HAL would ship).
            (device / "device-lineage.mk").write_text(
                "# CHRE\n"
                "PRODUCT_PACKAGES += \\\n"
                "    chre_daemon_msm\n"
                "\n"
                "# Configstore\n"
                "PRODUCT_PACKAGES += \\\n"
                "\n"
            )
            (product / "base_system.mk").write_text(
                "PRODUCT_PACKAGES += \\\n"
                "    logcat \\\n"
                "    pbtombstone \\\n"
                "    ping\n"
                "\n"
                "PRODUCT_HOST_PACKAGES += \\\n"
                "    fastboot \\\n"
                "    pbtombstone \\\n"
                "    sqlite3\n"
            )
            (bpf / "Android.bp").write_text(
                'rust_binary {\n'
                '    name: "bpfloader",\n'
                '    required: [\n'
                '        "bpftool",\n'
                '        "timeInState.bpf",\n'
                '    ],\n'
                '}\n'
            )
            (dump / "Android.bp").write_text(
                'cc_binary {\n'
                '    name: "dumpstate",\n'
                '    required: [\n'
                '        "alloctop",\n'
                '        "atrace",\n'
                '    ],\n'
                '}\n'
            )
            (apex / "Android.bp").write_text(
                'cc_binary {\n'
                '    name: "apexd",\n'
                '    required: ["apexd-lsof.sh"],\n'
                '    init_rc: ["apexd.rc"],\n'
                '}\n'
            )
            env = {**os.environ, "ANDROID_BUILD_TOP": str(top)}
            first = subprocess.run(
                ["python3", str(apply)],
                cwd=top,
                env=env,
                check=False,
                text=True,
                capture_output=True,
            )
            self.assertEqual(0, first.returncode, first.stderr)
            self.assertIn("dropped", first.stdout)
            vendor_text = (vendor / "sunfish-vendor.mk").read_text()
            self.assertNotIn("cnss_diag", vendor_text)
            self.assertNotIn("imsrcsd", vendor_text)
            self.assertIn("imsdatadaemon.rc", vendor_text)
            self.assertIn("    loc_launcher \\\n", vendor_text)
            self.assertIn("    xtra-daemon\n", vendor_text)
            self.assertIn("cnss-daemon", vendor_text)
            device_text = (device / "device-lineage.mk").read_text()
            self.assertIn("disable_configstore", device_text)
            self.assertIn("chre_daemon_msm", device_text)
            self.assertIn("restored", first.stdout)
            base = (product / "base_system.mk").read_text()
            self.assertNotIn("    pbtombstone \\\n    ping\n", base)
            self.assertIn("PRODUCT_HOST_PACKAGES += \\\n    fastboot \\\n    pbtombstone \\\n", base)
            self.assertNotIn('"bpftool",', (bpf / "Android.bp").read_text())
            self.assertIn("timeInState.bpf", (bpf / "Android.bp").read_text())
            self.assertNotIn("alloctop", (dump / "Android.bp").read_text())
            self.assertIn("atrace", (dump / "Android.bp").read_text())
            self.assertNotIn("apexd-lsof.sh", (apex / "Android.bp").read_text())
            self.assertIn("apexd.rc", (apex / "Android.bp").read_text())
            second = subprocess.run(
                ["python3", str(apply)],
                cwd=top,
                env=env,
                check=False,
                text=True,
                capture_output=True,
            )
            self.assertEqual(0, second.returncode, second.stderr)
            self.assertIn("already", second.stdout)

    def test_large_axp_zip_is_ignored(self) -> None:
        result = subprocess.run(
            ["git", "check-ignore", "--quiet", "AXP.OS-22.2-20260324-PRO-sunfish.zip"],
            cwd=ROOT,
            check=False,
        )
        self.assertEqual(0, result.returncode)


if __name__ == "__main__":
    unittest.main()