firegex/tests/unit/test_backup_guards.py

"""What a backup may carry, and what it may be named.

`/api/import` takes a JSON object an operator uploaded and writes files out of it. Two
things stand between that and the filesystem — a charset for the key, and `safe_join` —
and one thing stands between a restore and this instance's own credentials, which is what
`dump()` refuses to export.

All three have integration coverage, which means they are checked against a live instance
by someone who has one. They are pure functions; checking them here costs a millisecond
and covers the cases nobody would stage against a running firewall.
"""

import re
from pathlib import Path

import pytest
from fastapi import HTTPException

from utils import SAFE_DB_NAME, SAFE_PY_NAME, safe_join
from utils.sqlite import SQLite


# --- what a backup entry may be called ---------------------------------------


@pytest.mark.parametrize("name", ["services.db", "firegex.db", "a.db", "A-b_9.db"])
def test_the_shapes_export_db_produces_are_accepted(name):
    assert SAFE_DB_NAME.fullmatch(name)


@pytest.mark.parametrize("name", [
    "../services.db", "db/services.db", "/etc/services.db", "..", ".db",
    "services.DB", "services.db.py", "services db.db", "services.db\x00",
])
def test_anything_that_is_not_a_plain_basename_is_refused(name):
    assert not SAFE_DB_NAME.fullmatch(name)


def test_a_trailing_newline_does_not_get_past_the_guard():
    """`$` matches before a trailing newline, so `match` accepted `"services.db\\n"`.

    It was never a way out of the directory — `safe_join` contains the result whatever
    the name — so this cost nothing. It is fixed because a pattern anchored to mean
    "this and nothing else" should mean that: the next thing written against one of
    these guards should not have to know about the exception.
    """
    assert re.match(r'^[A-Za-z0-9_-]+\.db$', "services.db\n"), "the old spelling accepted it"
    assert not SAFE_DB_NAME.fullmatch("services.db\n")
    assert not SAFE_PY_NAME.fullmatch("abc123.py\n")


def test_the_two_shapes_do_not_accept_each_other():
    assert not SAFE_PY_NAME.fullmatch("services.db")
    assert not SAFE_DB_NAME.fullmatch("abc123.py")


# --- and where it may be written ---------------------------------------------


def test_a_name_that_climbs_out_is_refused(tmp_path):
    """Defence in depth on top of the charset: the regex is what is *meant* to catch
    these, and this is what catches them if the regex is ever loosened."""
    for attempt in ("../escape.db", "a/../../escape.db", ".."):
        with pytest.raises(HTTPException) as raised:
            safe_join(tmp_path, attempt)
        assert raised.value.status_code == 403


def test_an_absolute_path_is_taken_as_a_relative_one(tmp_path):
    """Not refused but *contained*, which is the same outcome by a different route."""
    landed = safe_join(tmp_path, "/etc/passwd")
    assert landed.is_relative_to(Path(tmp_path).resolve())


def test_an_ordinary_name_lands_where_it_should(tmp_path):
    assert safe_join(tmp_path, "services.db") == Path(tmp_path).resolve() / "services.db"


# --- what a backup must never carry ------------------------------------------


def test_a_backup_carries_neither_credentials_nor_the_auth_mode(tmp_path):
    """The third one is the newest and the easiest to lose.

    `password` and `secret` are this instance's credentials. `auth_disabled` is not a
    credential — it is *whether credentials are asked for*, which belongs to the
    deployment and not to the configuration being restored. A backup able to carry it is
    a backup able to turn a firewall's authentication off.
    """
    db = SQLite(str(tmp_path / "firegex.db"), {})
    db.connect()
    db.create_schema({})
    for key, value in (("password", "$pbkdf2$secret"), ("secret", "jwt-signing-key"),
                       ("auth_disabled", "1"), ("auth_disabled_host", "1"),
                       ("something_else", "kept")):
        db.query("INSERT INTO keys_values (key, value) VALUES (?, ?);", key, value)

    carried = {row["key"] for row in db.dump()["keys_values"]}
    assert carried == {"something_else"}, carried
    db.disconnect()


def test_everything_else_in_a_table_is_carried(tmp_path):
    """The exclusion is four named keys, not a table nobody exports."""
    db = SQLite(str(tmp_path / "other.db"), {})
    db.connect()
    db.create_schema({"things": {"id": "VARCHAR(10) PRIMARY KEY", "v": "TEXT"}})
    db.query("INSERT INTO things (id, v) VALUES ('a', 'kept');")

    dumped = db.dump()
    assert dumped["things"] == [{"id": "a", "v": "kept"}]
    db.disconnect()


def test_a_column_named_like_a_keyword_is_loaded_back(tmp_path):
    """The firewall's rules have a column called `table`. Left unquoted, loading it was a
    syntax error, and no backup holding a single firewall rule could be imported."""
    db = SQLite(str(tmp_path / "firewall-rules.db"), {})
    db.connect()
    db.create_schema({"rules": {"rule_id": "INT PRIMARY KEY", "`table`": "TEXT"}})
    db.query("INSERT INTO rules (rule_id, `table`) VALUES (0, 'filter');")
    dumped = db.dump()

    db.query("DELETE FROM rules;")
    db.load(dumped)
    assert db.query("SELECT * FROM rules;") == [{"rule_id": 0, "table": "filter"}]
    db.disconnect()


def test_a_dry_run_loads_nothing_and_still_finds_what_would_fail(tmp_path):
    """What an import runs over every database before it changes any of them: one
    refusing after others were replaced is how an import wiped this instance's
    password and secret."""
    db = SQLite(str(tmp_path / "things.db"), {})
    db.connect()
    db.create_schema({"things": {"id": "VARCHAR(10) PRIMARY KEY",
                                 "v": "TEXT NOT NULL CHECK (v IN ('a', 'b'))"}})
    db.query("INSERT INTO things (id, v) VALUES ('kept', 'a');")

    db.load({"things": [{"id": "new", "v": "b"}]}, dry_run=True)
    assert db.query("SELECT * FROM things;") == [{"id": "kept", "v": "a"}], \
        "a dry run changed the database"

    with pytest.raises(Exception):
        db.load({"things": [{"id": "bad", "v": "not allowed"}]}, dry_run=True)
    assert db.query("SELECT * FROM things;") == [{"id": "kept", "v": "a"}]
    db.disconnect()


# --- whether authentication starts off ----------------------------------------
# A container's environment is fixed when it is created. Docker starting it again by
# itself — a reboot, a daemon restart, a crash under `restart: unless-stopped` — brings that
# environment back, older than what `run.py config` has said since: authentication turned
# back on with `run.py config --password` was off again after the next reboot.


def test_a_fresh_boot_takes_the_environment():
    """`run.py` has just written it from the current configuration."""
    from utils import boot_auth_mode

    assert boot_auth_mode(None, True, True) == (True, False)
    assert boot_auth_mode("0", True, True) == (True, False), \
        "a container run.py has just created is told the current answer by its environment"
    assert boot_auth_mode("1", True, False) == (False, False)


def test_a_restarted_container_keeps_what_the_host_decided_since():
    from utils import boot_auth_mode

    assert boot_auth_mode("0", False, True) == (False, True), \
        "authentication re-enabled with run.py config came back off after a reboot"
    assert boot_auth_mode("1", False, False) == (True, True)


def test_with_nothing_decided_since_the_environment_stands():
    from utils import boot_auth_mode

    assert boot_auth_mode(None, False, False) == (False, False)
    assert boot_auth_mode(None, False, True) == (True, False)


# --- the code beside the database ----------------------------------------------


def test_filter_files_no_filter_refers_to_are_removed_at_start(tmp_path, monkeypatch):
    """An import replaced the database and wrote the backup's files, and left every file
    the backup did not have: the code of each Python filter created since, on disk for a
    filter that no longer existed."""
    import application.services as router

    class Db:
        def query(self, sql, *args):
            return [{"filter_id": "kept"}]

    for name in ("kept.py", "orphan.py", "notes.txt"):
        (tmp_path / name).write_text("x")
    monkeypatch.setattr(router, "CODE_DIR", str(tmp_path))
    monkeypatch.setattr(router, "db", Db())
    router._prune_orphan_code()
    assert sorted(p.name for p in tmp_path.iterdir()) == ["kept.py", "notes.txt"]


def test_no_code_directory_is_nothing_to_prune(tmp_path, monkeypatch):
    import application.services as router

    monkeypatch.setattr(router, "CODE_DIR", str(tmp_path / "missing"))
    router._prune_orphan_code()