path: root/tests/installer-steps/test_configure_openai_chatgpt_repo.py

"""Test configure_openai_chatgpt_repo — OpenAI's signed pacman repo for ChatGPT.

OpenAI ships the ChatGPT (Codex) desktop app for Arch as chatgpt-bin from
its own signed pacman repository. Its install-arch.sh pins the repository
signing-key fingerprint, locally signs the key into the pacman keyring,
writes /etc/pacman.d/openai-chatgpt.conf and adds an Include line to
pacman.conf; ratio and velox were set up that way on 2026-10-11. The step
reproduces exactly that configuration so a re-run on those machines is a
no-op and the app rides normal pacman upgrades under OpenAI's signature.
The AUR repackage of the same binary (chatgpt-desktop) would conflict with
chatgpt-bin on any machine already using the repo, so it is not used.

A half-configured repository is worse than none: with the Include in place
and the key untrusted, SigLevel TrustedOnly makes every later pacman call
fail on the invalid database. So an untrusted key (fingerprint mismatch or
failed download) must import nothing and leave pacman.conf untouched.

The pin is on the primary key's fingerprint alone. pacman-key --lsign-key
resolves a subkey fingerprint to its primary, so a bundle whose subkey or
second primary carries the pinned fingerprint must be rejected too.

Method: sed-extract configure_openai_chatgpt_repo from the real `archsetup`,
point it at a temp pacman.conf and repo drop-in, and replace curl, gpg,
pacman-key and pacman with recorders driven by environment switches.

    python3 -m unittest tests.installer-steps.test_configure_openai_chatgpt_repo
"""

import os
import re
import stat
import subprocess
import tempfile
import textwrap
import unittest

REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
ARCHSETUP = os.path.join(REPO_ROOT, "archsetup")

FPR = "3BFA0E4AE8B8CC16A2D9BA684A3B4A566C4660E4"
KEY_URL = "https://persistent.oaistatic.com/codex-app-prod/linux/repository-signing-key.gpg"
SERVER = "https://persistent.oaistatic.com/codex-app-prod/linux/arch/$arch"
BASE_CONF = "[core]\nInclude = /etc/pacman.d/mirrorlist\n"

# Recorders. Every fake appends its argv to $MARKER; the pacman fake also
# records whether the Include line is already in pacman.conf at call time,
# which is how the tests see that -Sy ran after the repo was wired in.
STUBS = r"""
logfile=/dev/null
action=""
display() { :; }
backup_system_file() { :; }
error_warn() { echo "WARN: $1"; return 1; }
pacman-key() {
    echo "PACMAN_KEY: $*" >> "$MARKER"
    case "$1" in
        --list-keys) [ "$KEY_PRESENT" = 1 ] ;;
        *) return 0 ;;
    esac
}
curl() {
    echo "CURL: $*" >> "$MARKER"
    [ "$CURL_FAIL" = 1 ] && return 22
    local out=""
    while [ $# -gt 0 ]; do
        case "$1" in -o) out="$2"; shift ;; esac
        shift
    done
    [ -n "$out" ] && printf 'FAKE KEY\n' > "$out"
    return 0
}
gpg() {
    echo "GPG: $*" >> "$MARKER"
    case " $* " in
        *" --show-keys "*)
            printf 'pub:u:4096:1:4A3B4A566C4660E4:1754362069:::u:::scSC::::::23::0:\n'
            printf 'fpr:::::::::%s:\n' "$SERVED_FPR"
            printf 'uid:u::::1754362069::ABC::Codex Linux Repository::::::::::0:\n'
            if [ -n "$SERVED_SUBKEY_FPR" ]; then
                printf 'sub:u:4096:1:0123456789ABCDEF:1754362069::::::e::::::23:\n'
                printf 'fpr:::::::::%s:\n' "$SERVED_SUBKEY_FPR"
            fi
            if [ -n "$SERVED_SECOND_PUB_FPR" ]; then
                printf 'pub:u:4096:1:FEDCBA9876543210:1754362069:::u:::scSC::::::23::0:\n'
                printf 'fpr:::::::::%s:\n' "$SERVED_SECOND_PUB_FPR"
                printf 'uid:u::::1754362069::DEF::Somebody Else::::::::::0:\n'
            fi
            ;;
    esac
    return 0
}
pacman() {
    local n
    n=$(grep -cxF "Include = $REPO_CONF" "$PACMAN_CONF" 2>/dev/null)
    echo "PACMAN: $* include=$n" >> "$MARKER"
    return 0
}
"""


def paths(d):
    return (os.path.join(d, "pacman.conf"),
            os.path.join(d, "pacman.d", "openai-chatgpt.conf"),
            os.path.join(d, "calls"))


def run(d, *, key_present=False, served_fpr=FPR, served_subkey_fpr="",
        served_second_pub_fpr="", curl_fail=False, pacman_conf_body=BASE_CONF):
    pacman_conf, repo_conf, marker = paths(d)
    if pacman_conf_body is not None and not os.path.exists(pacman_conf):
        with open(pacman_conf, "w") as f:
            f.write(pacman_conf_body)
    script = STUBS + textwrap.dedent(f"""\
        source <(sed -n '/^configure_openai_chatgpt_repo() {{/,/^}}/p' "{ARCHSETUP}")
        configure_openai_chatgpt_repo "$PACMAN_CONF" "$REPO_CONF"
        echo "RC=$?"
        exit 0
    """)
    env = dict(os.environ, MARKER=marker, PACMAN_CONF=pacman_conf,
               REPO_CONF=repo_conf, KEY_PRESENT="1" if key_present else "0",
               SERVED_FPR=served_fpr, SERVED_SUBKEY_FPR=served_subkey_fpr,
               SERVED_SECOND_PUB_FPR=served_second_pub_fpr,
               CURL_FAIL="1" if curl_fail else "0")
    return subprocess.run(["bash", "-c", script], capture_output=True,
                          text=True, timeout=10, env=env)


def rc_of(r):
    m = re.search(r"^RC=(\d+)$", r.stdout, re.M)
    assert m, "no RC line in output: %r / %r" % (r.stdout, r.stderr)
    return int(m.group(1))


def read(path):
    with open(path) as f:
        return f.read()


def calls(d):
    marker = paths(d)[2]
    return read(marker) if os.path.exists(marker) else ""


class ConfigureOpenaiChatgptRepo(unittest.TestCase):
    # ------------------------------------------------------------ normal ----
    def test_writes_the_published_repo_definition(self):
        with tempfile.TemporaryDirectory() as d:
            r = run(d)
            self.assertEqual(rc_of(r), 0)
            _, repo_conf, _ = paths(d)
            self.assertTrue(os.path.exists(repo_conf))
            body = read(repo_conf)
            self.assertIn("[openai-chatgpt]\n", body)
            self.assertIn("SigLevel = Required DatabaseRequired TrustedOnly\n", body)
            # $arch stays literal: pacman expands it, the shell must not.
            self.assertIn(f"Server = {SERVER}\n", body)

    def test_repo_definition_is_world_readable_not_writable(self):
        with tempfile.TemporaryDirectory() as d:
            run(d)
            mode = stat.S_IMODE(os.stat(paths(d)[1]).st_mode)
            self.assertEqual(mode, 0o644)

    def test_adds_one_include_line_to_pacman_conf(self):
        with tempfile.TemporaryDirectory() as d:
            run(d)
            pacman_conf, repo_conf, _ = paths(d)
            body = read(pacman_conf)
            self.assertTrue(body.startswith(BASE_CONF), body)
            self.assertEqual(body.count(f"\nInclude = {repo_conf}\n"), 1)

    def test_imports_and_locally_signs_the_pinned_key(self):
        with tempfile.TemporaryDirectory() as d:
            run(d)
            c = calls(d)
            self.assertIn(f" {KEY_URL}", c)
            self.assertIn("GPG: ", c)
            self.assertIn("--show-keys", c)
            self.assertIn("PACMAN_KEY: --add ", c)
            self.assertIn(f"PACMAN_KEY: --lsign-key {FPR}", c)

    def test_refreshes_databases_after_the_include_is_in_place(self):
        with tempfile.TemporaryDirectory() as d:
            run(d)
            self.assertIn("PACMAN: -Sy include=1", calls(d))

    # ---------------------------------------------------------- boundary ----
    def test_running_twice_leaves_one_include_and_the_same_files(self):
        with tempfile.TemporaryDirectory() as d:
            pacman_conf, repo_conf, _ = paths(d)
            self.assertEqual(rc_of(run(d)), 0)
            first_conf, first_repo = read(pacman_conf), read(repo_conf)
            self.assertEqual(rc_of(run(d)), 0)
            self.assertEqual(read(pacman_conf), first_conf)
            self.assertEqual(read(repo_conf), first_repo)
            self.assertEqual(first_conf.count("Include = " + repo_conf), 1)

    def test_already_trusted_key_is_not_downloaded_or_added_again(self):
        with tempfile.TemporaryDirectory() as d:
            r = run(d, key_present=True)
            self.assertEqual(rc_of(r), 0)
            c = calls(d)
            self.assertNotIn("CURL:", c)
            self.assertNotIn("PACMAN_KEY: --add", c)
            self.assertTrue(os.path.exists(paths(d)[1]))

    def test_pacman_conf_without_trailing_newline_gets_its_own_line(self):
        with tempfile.TemporaryDirectory() as d:
            r = run(d, pacman_conf_body=BASE_CONF.rstrip("\n"))
            self.assertEqual(rc_of(r), 0)
            pacman_conf, repo_conf, _ = paths(d)
            body = read(pacman_conf)
            self.assertIn("Include = /etc/pacman.d/mirrorlist\n", body)
            self.assertRegex(body, re.compile(rf"^Include = {re.escape(repo_conf)}$", re.M))

    def test_existing_differing_repo_definition_is_kept_and_flagged(self):
        with tempfile.TemporaryDirectory() as d:
            pacman_conf, repo_conf, _ = paths(d)
            os.makedirs(os.path.dirname(repo_conf))
            custom = "[openai-chatgpt]\nSigLevel = Required\nServer = file:///srv/mirror\n"
            with open(repo_conf, "w") as f:
                f.write(custom)
            r = run(d)
            self.assertIn("WARN:", r.stdout)
            self.assertEqual(read(repo_conf), custom)
            # The repository is still configured, just by the user's file.
            self.assertIn(f"\nInclude = {repo_conf}\n", read(pacman_conf))
            self.assertEqual(rc_of(r), 0)

    # ------------------------------------------------------------- error ----
    def assert_nothing_configured(self, d, r):
        pacman_conf, repo_conf, _ = paths(d)
        c = calls(d)
        self.assertIn("WARN:", r.stdout)
        self.assertNotIn("PACMAN_KEY: --add", c)
        self.assertNotIn("PACMAN_KEY: --lsign-key", c)
        self.assertNotIn("PACMAN: -Sy", c)
        self.assertFalse(os.path.exists(repo_conf))
        self.assertEqual(read(pacman_conf), BASE_CONF)
        self.assertNotEqual(rc_of(r), 0)

    def test_fingerprint_mismatch_imports_nothing_and_touches_nothing(self):
        with tempfile.TemporaryDirectory() as d:
            r = run(d, served_fpr="DEADBEEF" * 5)
            self.assert_nothing_configured(d, r)

    def test_key_download_failure_imports_nothing_and_touches_nothing(self):
        with tempfile.TemporaryDirectory() as d:
            r = run(d, curl_fail=True)
            self.assert_nothing_configured(d, r)

    # The pin is on the PRIMARY key's fingerprint and nothing else. pacman-key
    # --lsign-key resolves a subkey fingerprint to the primary that carries it,
    # so a pin that a subkey or a second primary could satisfy would trust
    # whatever key the bundle smuggled in beside the expected one.
    def test_subkey_carrying_the_pinned_fingerprint_is_rejected(self):
        with tempfile.TemporaryDirectory() as d:
            r = run(d, served_fpr="DEADBEEF" * 5, served_subkey_fpr=FPR)
            self.assert_nothing_configured(d, r)

    def test_second_primary_carrying_the_pinned_fingerprint_is_rejected(self):
        with tempfile.TemporaryDirectory() as d:
            r = run(d, served_fpr="DEADBEEF" * 5, served_second_pub_fpr=FPR)
            self.assert_nothing_configured(d, r)

    def test_rogue_second_primary_beside_the_pinned_key_is_rejected(self):
        with tempfile.TemporaryDirectory() as d:
            r = run(d, served_fpr=FPR, served_second_pub_fpr="DEADBEEF" * 5)
            self.assert_nothing_configured(d, r)

    @unittest.skipUnless(os.geteuid() != 0, "root ignores file write bits")
    def test_unwritable_pacman_conf_warns_and_reports_failure(self):
        with tempfile.TemporaryDirectory() as d:
            pacman_conf, _, _ = paths(d)
            with open(pacman_conf, "w") as f:
                f.write(BASE_CONF)
            os.chmod(pacman_conf, 0o444)
            try:
                r = run(d)
                self.assertIn("WARN:", r.stdout)
                self.assertNotEqual(rc_of(r), 0)
                self.assertNotIn("PACMAN: -Sy", calls(d))
            finally:
                os.chmod(pacman_conf, 0o644)


class SupplementalSoftwareWiring(unittest.TestCase):
    """chatgpt-bin is installed only once its repository is trusted."""

    def run_step(self, repo_rc):
        script = textwrap.dedent(f"""\
            desktop_env=hyprland
            display() {{ :; }}
            run_task() {{ :; }}
            mask_fwupd_passim() {{ :; }}
            pacman_install() {{ echo "PACMAN_INSTALL: $1"; }}
            aur_install() {{ echo "AUR_INSTALL: $1"; }}
            configure_openai_chatgpt_repo() {{ echo "CONFIGURE_REPO"; return {repo_rc}; }}
            source <(sed -n '/^supplemental_software() {{/,/^}}/p' "{ARCHSETUP}")
            supplemental_software
        """)
        return subprocess.run(["bash", "-c", script], capture_output=True,
                              text=True, timeout=10)

    def test_installs_chatgpt_bin_after_configuring_the_repo(self):
        out = self.run_step(0).stdout
        self.assertIn("CONFIGURE_REPO\n", out)
        self.assertLess(out.index("CONFIGURE_REPO"),
                        out.index("PACMAN_INSTALL: chatgpt-bin"))

    def test_skips_chatgpt_bin_when_the_repo_cannot_be_trusted(self):
        out = self.run_step(1).stdout
        self.assertNotIn("PACMAN_INSTALL: chatgpt-bin", out)
        # The failure is local to the app; the rest of the step still runs.
        self.assertIn("AUR_INSTALL: claude-desktop", out)
        self.assertIn("AUR_INSTALL: freetube", out)

    def test_does_not_also_pull_the_conflicting_aur_repackage(self):
        out = self.run_step(0).stdout
        self.assertNotIn("chatgpt-desktop", out)


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