computer-playbook/tests/unit/python/utils/security/test_database.py

from __future__ import annotations

import os
import tempfile
import time
from pathlib import Path
from types import SimpleNamespace
from unittest import TestCase, main, mock

from utils.security import database
from utils.security.pipeline import pipeline

P = pipeline()


def _touch(path: Path, age: float) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("db", encoding="utf-8")
    stamp = time.time() - age
    os.utime(path, (stamp, stamp))


class _Root(TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)

    def tearDown(self):
        self.tmp.cleanup()


class TestOsvUrl(TestCase):
    def test_an_ecosystem_with_a_space_is_quoted(self):
        self.assertEqual(
            database.osv_url("Red Hat"),
            "https://osv-vulnerabilities.storage.googleapis.com/Red%20Hat/all.zip",
        )


class TestFetchDatabases(_Root):
    def _fetch(self, engines, returncode):
        downloads = []
        calls = []

        def runner(argv):
            calls.append(list(argv))
            return SimpleNamespace(
                returncode=returncode, stdout="", stderr="no network"
            )

        def fetch(url, destination):
            downloads.append(url)
            destination.write_bytes(b"zip")

        errors = database.fetch_databases(P, self.root, engines, runner, fetch)
        return errors, calls, downloads

    def test_osv_lands_in_the_layout_the_scanner_reads(self):
        errors, _calls, downloads = self._fetch(["osv"], 0)
        self.assertEqual(errors, [])
        self.assertTrue(
            (self.root / "osv" / "osv-scalibr" / "Ubuntu" / "all.zip").is_file()
        )
        self.assertEqual(len(downloads), len(database.osv_zips(P, self.root)))

    def test_a_failed_download_is_reported(self):
        errors, calls, _downloads = self._fetch(["grype"], 1)
        self.assertTrue(errors[0].startswith("grype database: exited 1"))
        self.assertIn(["db", "update"], [call[-2:] for call in calls])

    def test_a_download_that_leaves_no_database_is_reported(self):
        errors, _calls, _downloads = self._fetch(["trivy"], 0)
        self.assertTrue(any("provided database" in error for error in errors))


class TestStale(_Root):
    def test_a_missing_database_is_stale(self):
        self.assertTrue(database.stale(P, self.root, "grype", time.time()))

    def test_a_database_of_today_is_fresh(self):
        _touch(self.root / "grype" / "6" / "vulnerability.db", 3600)
        self.assertFalse(database.stale(P, self.root, "grype", time.time()))

    def test_a_database_older_than_a_day_is_stale(self):
        _touch(
            self.root / "grype" / "6" / "vulnerability.db",
            database.MAX_AGE_SECONDS + 60,
        )
        self.assertTrue(database.stale(P, self.root, "grype", time.time()))

    def test_osv_is_stale_while_one_ecosystem_is_missing(self):
        zips = database.osv_zips(P, self.root)
        for path in zips[1:]:
            _touch(path, 60)
        self.assertTrue(database.stale(P, self.root, "osv", time.time()))


class TestRefresh(_Root):
    def _refresh(self, engines, fetched=()):
        with mock.patch.object(
            database, "fetch_databases", return_value=list(fetched)
        ) as fetch:
            errors = database.refresh(
                self.root, engines, lambda argv: None, lambda u, d: None
            )
        return errors, fetch

    def test_only_stale_engines_are_downloaded(self):
        _touch(self.root / "grype" / "6" / "vulnerability.db", 60)
        _errors, fetch = self._refresh(["trivy", "grype"])
        self.assertEqual(fetch.call_args.args[2], ["trivy"])

    def test_fresh_databases_download_nothing(self):
        _touch(self.root / "grype" / "6" / "vulnerability.db", 60)
        errors, fetch = self._refresh(["grype"])
        self.assertEqual(errors, [])
        fetch.assert_not_called()

    def test_the_lock_lives_in_the_shared_directory(self):
        self._refresh(["grype"])
        self.assertTrue((self.root / database.LOCK_NAME).is_file())

    def test_a_missing_shared_directory_is_created(self):
        self.root = self.root / "vulnerability-db"
        self._refresh(["grype"])
        self.assertTrue(self.root.is_dir())

    def test_a_failed_download_is_returned(self):
        errors, _fetch = self._refresh(["grype"], ["grype database: exited 1"])
        self.assertEqual(errors, ["grype database: exited 1"])


class TestHandOver(_Root):
    def _hand_over(self, euid):
        _touch(self.root / "osv" / "osv-scalibr" / "Ubuntu" / "all.zip", 60)
        owner = self.root.stat()
        with (
            mock.patch.object(database.os, "geteuid", return_value=euid),
            mock.patch.object(database.os, "chown") as chown,
        ):
            database.hand_over(self.root, [self.root / "osv"])
        return chown, (owner.st_uid, owner.st_gid)

    def test_a_root_refresh_gives_every_file_to_the_directory_owner(self):
        chown, owner = self._hand_over(0)
        self.assertEqual(
            {(Path(call.args[0]), call.args[1:]) for call in chown.call_args_list},
            {
                (self.root / "osv", owner),
                (self.root / "osv" / "osv-scalibr", owner),
                (self.root / "osv" / "osv-scalibr" / "Ubuntu", owner),
                (self.root / "osv" / "osv-scalibr" / "Ubuntu" / "all.zip", owner),
            },
        )

    def test_an_unprivileged_refresh_changes_no_owner(self):
        chown, _owner = self._hand_over(1000)
        chown.assert_not_called()

    def test_a_root_refresh_that_wrote_no_database_hands_over_the_lock(self):
        with (
            mock.patch.object(database.os, "geteuid", return_value=0),
            mock.patch.object(database.os, "chown") as chown,
            mock.patch.object(database, "fetch_databases", return_value=[]),
        ):
            database.refresh(self.root, ["grype"], lambda argv: None, lambda u, d: None)
        self.assertEqual(
            [Path(call.args[0]) for call in chown.call_args_list],
            [self.root / database.LOCK_NAME],
        )


if __name__ == "__main__":
    main()