#!/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_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_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_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",
            "disable_configstore",
            "bpftool",
            "alloctop",
            "apexd-lsof.sh",
        ):
            self.assertIn(name, 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"
            )
            (device / "device-lineage.mk").write_text(
                "PRODUCT_PACKAGES += \\\n"
                "    chre_daemon_msm\n"
                "\n"
                "PRODUCT_PACKAGES += \\\n"
                "    disable_configstore\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)
            self.assertNotIn("disable_configstore", (device / "device-lineage.mk").read_text())
            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()