vacuum-wall/tests/test_auth.py

"""Phase 1 auth tests: JWT lifecycle, DB layer, password hashing, login flow.

All subprocess calls are mocked. Uses in-memory SQLite.
"""

from __future__ import annotations

import base64
import os
from unittest.mock import MagicMock, patch

import pytest

import lib.db
import lib.db_sqlite
from lib.auth import (
    blacklist_expired,
    blacklist_token,
    decode_token,
    generate_access_token,
    generate_refresh_token,
    generate_tokens,
    is_blacklisted,
    validate_token,
)
from lib.auth_users import (
    create_user,
    delete_user,
    find_user,
    get_user,
    list_users,
    reset_password,
    update_password,
    update_permissions,
    verify_user_password,
)
from lib.db import (
    Q_INSERT_BLACKLIST,
    Q_INSERT_USER,
    Q_SELECT_ALL_USERS,
    Q_SELECT_BLACKLIST,
    Q_SELECT_PERMISSIONS,
    Q_SELECT_USER_BY_NAME,
    Q_UPSERT_PERMISSION,
    get_db,
    reset_db_for_test,
)
from lib.password import hash_password, needs_rehash, verify_password


@pytest.fixture(autouse=True)
def _db_reset():
    """Reset DB singleton and env vars before each test."""
    reset_db_for_test()
    old_backend = os.environ.pop("VACUUM_WALL_DB_BACKEND", None)
    old_path = os.environ.pop("VACUUM_WALL_DB_PATH", None)
    old_seed = os.environ.pop("VACUUM_WALL_SEED_BUILTIN_ADMIN", None)

    os.environ["VACUUM_WALL_DB_BACKEND"] = "sqlite"
    os.environ["VACUUM_WALL_DB_PATH"] = ":memory:"

    yield

    reset_db_for_test()
    if old_backend is not None:
        os.environ["VACUUM_WALL_DB_BACKEND"] = old_backend
    if old_path is not None:
        os.environ["VACUUM_WALL_DB_PATH"] = old_path
    if old_seed is not None:
        os.environ["VACUUM_WALL_SEED_BUILTIN_ADMIN"] = old_seed


@pytest.fixture
def db():
    """Initialize the DB and return it."""
    return get_db()


@pytest.fixture
def sample_secret():
    """Placeholder for per-user secret tests.

    Per-user secrets are generated at user creation time, so this fixture
    is a no-op kept for backward compat with existing test parameter lists.
    """


# ---------- Password hashing tests ----------


class TestPasswordHashing:
    def test_hash_starts_with_argon2id(self):
        h = hash_password("test1234")
        assert h.startswith("$argon2id$v=19$")

    def test_hash_contains_parameters(self):
        h = hash_password("test1234")
        assert "m=65536" in h  # 64 MiB
        assert "t=3" in h
        assert "p=4" in h

    def test_unique_salt(self):
        h1 = hash_password("same_password")
        h2 = hash_password("same_password")
        assert h1 != h2

    def test_verify_correct_password(self):
        h = hash_password("my_secret")
        assert verify_password("my_secret", h) is True

    def test_verify_wrong_password(self):
        h = hash_password("my_secret")
        assert verify_password("wrong_password", h) is False

    def test_needs_rehash_no_change(self):
        h = hash_password("test")
        assert needs_rehash(h) is False


# ---------- DB layer tests ----------


class TestDBLayer:
    def test_init_tables(self, db):
        cur = db.conn.cursor()
        tables = cur.execute(
            "SELECT name FROM sqlite_master WHERE type='table'"
        ).fetchall()
        table_names = {t["name"] for t in tables}
        assert "users" in table_names
        assert "permissions" in table_names
        assert "token_blacklist" in table_names
        assert "webauthn_creds" in table_names

    def test_connection_caching(self, db):
        conn1 = db.conn
        conn2 = db.conn
        assert conn1 is conn2

    def test_connections_are_thread_local(self, tmp_path):
        """DB access from multiple threads must work (regression test).

        The Flask WebUI validates JWTs in lib.db from worker threads while
        the daemon uses its event-loop thread. A single shared connection
        raises sqlite3.ProgrammingError ("SQLite objects created in a
        thread can only be used in that same thread") on the first
        cross-thread query.
        """
        import threading

        reset_db_for_test()
        db_path = str(tmp_path / "thread_local.db")
        os.environ["VACUUM_WALL_DB_PATH"] = db_path
        try:
            db = get_db()
            db.run(Q_INSERT_USER, ("touser", "$argon2id$hash", "test-secret"))

            results: list = []
            threads = [
                threading.Thread(
                    target=lambda: results.append(
                        db.query(Q_SELECT_USER_BY_NAME, ("touser",))
                    )
                )
                for _ in range(4)
            ]
            for t in threads:
                t.start()
            for t in threads:
                t.join()

        finally:
            reset_db_for_test()

        assert len(results) == 4
        for rows in results:
            assert isinstance(rows, list), f"query raised or returned {rows!r}"
            assert len(rows) == 1
            assert rows[0]["username"] == "touser"

    def test_connections_are_distinct_per_thread(self, db):
        """Each thread gets its own connection object."""
        import threading

        def conn_in_thread(result: list) -> None:
            result.append(db.conn)

        main_conn = db.conn
        result: list = []
        t = threading.Thread(target=conn_in_thread, args=(result,))
        t.start()
        t.join()
        assert len(result) == 1
        assert result[0] is not main_conn

    def test_insert_user(self, db):
        uid = db.run_one(Q_INSERT_USER, ("testuser", "$argon2id$hash", "test-secret"))
        assert isinstance(uid, int)
        assert uid > 0

    def test_select_user_by_name(self, db):
        db.run(Q_INSERT_USER, ("testuser", "$argon2id$hash", "test-secret"))
        rows = db.query(Q_SELECT_USER_BY_NAME, ("testuser",))
        assert len(rows) == 1
        assert rows[0]["username"] == "testuser"

    def test_query_unknown_id_raises(self, db):
        with pytest.raises(KeyError):
            db.query("nonexistent_query_id")

    def test_transaction_commit(self, db):
        with db.in_transaction() as tx:
            tx.run(Q_INSERT_USER, ("txuser", "$argon2id$hash", "test-secret"))
            tx.run(Q_UPSERT_PERMISSION, ("txuser", "firewall", "rw"))
        rows = db.query(Q_SELECT_USER_BY_NAME, ("txuser",))
        assert len(rows) == 1
        perms = db.query(Q_SELECT_PERMISSIONS, ("txuser",))
        assert len(perms) == 1
        assert perms[0]["subsystem"] == "firewall"

    def test_transaction_rollback(self, db):
        try:
            with db.in_transaction() as tx:
                tx.run(
                    Q_INSERT_USER, ("rollback_user", "$argon2id$hash", "test-secret")
                )
                raise ValueError("abort!")
        except ValueError:
            pass
        rows = db.query(Q_SELECT_USER_BY_NAME, ("rollback_user",))
        assert len(rows) == 0

    def test_prepared_statement_caching(self, db):
        stmt1 = db._get_prepared(Q_SELECT_USER_BY_NAME)
        stmt2 = db._get_prepared(Q_SELECT_USER_BY_NAME)

        assert stmt1 is stmt2

    def test_upsert_permission(self, db):
        db.run(Q_INSERT_USER, ("permuser", "$argon2id$hash", "test-secret"))
        db.run(Q_UPSERT_PERMISSION, ("permuser", "firewall", "read"))
        perms = db.query(Q_SELECT_PERMISSIONS, ("permuser",))
        assert perms[0]["level"] == "read"

        db.run(Q_UPSERT_PERMISSION, ("permuser", "firewall", "rw"))
        perms = db.query(Q_SELECT_PERMISSIONS, ("permuser",))
        assert perms[0]["level"] == "rw"

    def test_all_users(self, db):
        db.run(Q_INSERT_USER, ("aaa", "$argon2id$hash", "test-secret"))
        db.run(Q_INSERT_USER, ("bbb", "$argon2id$hash", "test-secret"))
        rows = db.query(Q_SELECT_ALL_USERS, ())
        # admin (builtin seed) + aaa + bbb, ordered by username
        assert len(rows) == 3
        assert rows[0]["username"] == "aaa"


# ---------- JWT tests ----------


class TestJWT:
    def setup_method(self) -> None:
        reset_db_for_test()
        get_db()

    def test_generate_tokens(self):
        tokens = generate_tokens("admin", {"firewall": "rw"})
        assert "access_token" in tokens
        assert "refresh_token" in tokens
        assert "session_id" in tokens
        assert isinstance(tokens["session_id"], str)

    def test_generate_token_with_session_id(self):
        """Token includes session_id in payload."""
        token = generate_access_token(
            "admin", {"firewall": "rw"}, session_id="test-session-123"
        )
        payload = decode_token(token)
        assert payload is not None
        assert payload["session_id"] == "test-session-123"

    def test_session_id_validation_match(self):
        """Token with matching session_id validates."""
        token = generate_access_token(
            "admin", {"firewall": "rw"}, session_id="my-session"
        )
        payload = validate_token(token, "access", session_id="my-session")
        assert payload is not None
        assert payload["sub"] == "admin"

    def test_session_id_validation_mismatch(self):
        """Token with wrong session_id is rejected."""
        token = generate_access_token(
            "admin", {"firewall": "rw"}, session_id="real-session"
        )
        payload = validate_token(token, "access", session_id="wrong-session")
        assert payload is None

    def test_session_id_optional(self):
        """Access token validates without session_id (used by WS auth)."""
        token = generate_access_token(
            "admin", {"firewall": "rw"}, session_id="my-session"
        )
        # Without session_id, only type, expiry, and blacklist are checked
        payload = validate_token(token, "access")
        assert payload is not None
        assert payload["sub"] == "admin"

    def test_session_id_enforced_when_provided(self):
        """Access token is rejected when session_id is provided but doesn't match."""
        token = generate_access_token(
            "admin", {"firewall": "rw"}, session_id="my-session"
        )
        payload = validate_token(token, "access", session_id="different-session")
        assert payload is None

    def test_decode_token(self):
        token = generate_access_token("admin", {"firewall": "rw"})
        payload = decode_token(token)
        assert payload is not None
        assert payload["sub"] == "admin"
        assert payload["type"] == "access"

    def test_validate_access_token(self):
        token = generate_access_token(
            "admin", {"firewall": "rw"}, session_id="test-session"
        )
        payload = validate_token(token, "access", session_id="test-session")
        assert payload is not None
        assert payload["sub"] == "admin"
        assert payload["permissions"]["firewall"] == "rw"

    def test_validate_wrong_type(self):
        token = generate_refresh_token("admin")
        payload = validate_token(token, "access", session_id="test-session")
        assert payload is None

    def test_validate_invalid_token(self):
        payload = validate_token(
            "invalid.token.here", "access", session_id="test-session"
        )
        assert payload is None

    def test_blacklist_token(self):
        token = generate_access_token("admin", {}, session_id="test-session")
        payload = decode_token(token)
        assert payload is not None
        jti = payload["jti"]

        blacklist_token(jti)
        assert is_blacklisted(jti) is True

        result = validate_token(token, "access", session_id="test-session")
        assert result is None

    def test_blacklist_cleanup(self):
        db = get_db()
        db.run(Q_INSERT_BLACKLIST, ("old-jti", "access", 1000))
        rows_before = db.query(Q_SELECT_BLACKLIST, ("old-jti",))
        assert len(rows_before) == 1

        blacklist_expired()

        rows_after = db.query(Q_SELECT_BLACKLIST, ("old-jti",))
        assert len(rows_after) == 0


# ---------- User management tests ----------


class TestUserManagement:
    def test_create_user(self, db):
        user = create_user("testuser", "password123", {"firewall": "rw"})
        assert user["username"] == "testuser"
        assert user["permissions"]["firewall"] == "rw"
        assert "id" in user

    def test_create_user_invalid_username(self, db):
        with pytest.raises(ValueError):
            create_user("a", "password123")

    def test_create_user_duplicate(self, db):
        create_user("dupuser", "password123")
        with pytest.raises(ValueError, match="already exists"):
            create_user("dupuser", "otherpassword")

    def test_verify_password(self, db):
        create_user("pwuser", "correctpass")
        assert verify_user_password("pwuser", "correctpass") is not None
        assert verify_user_password("pwuser", "wrongpass") is None
        assert verify_user_password("nonexistent", "anything") is None

    def test_get_user(self, db):
        create_user("getuser", "password123", {"firewall": "rw"})
        user = get_user("getuser")
        assert user is not None
        assert "password_hash" not in user
        assert user["permissions"]["firewall"] == "rw"

    def test_get_user_not_found(self, db):
        assert get_user("nonexistent") is None

    def test_update_password(self, db):
        create_user("upwuser", "oldpass")
        update_password("upwuser", "oldpass", "newpass123")
        assert verify_user_password("upwuser", "newpass123") is not None
        assert verify_user_password("upwuser", "oldpass") is None

    def test_update_password_wrong_old(self, db):
        create_user("upwfail", "realpass")
        with pytest.raises(ValueError, match="incorrect"):
            update_password("upwfail", "wrong_old", "newpass123")

    def test_reset_password_without_old(self, db):
        """Installer lockout recovery: reset works without knowing the old password."""
        create_user("rstuser", "unknown-old-pass")
        reset_password("rstuser", "freshpass123")
        assert verify_user_password("rstuser", "freshpass123") is not None
        assert verify_user_password("rstuser", "unknown-old-pass") is None

    def test_reset_password_not_found(self, db):
        with pytest.raises(ValueError, match="not found"):
            reset_password("ghostuser", "newpass123")

    def test_update_permissions(self, db):
        create_user("permuser", "password123", {"firewall": "rw"})
        update_permissions("permuser", {"firewall": "read", "network": "rw"})
        user = get_user("permuser")
        assert user["permissions"]["firewall"] == "read"
        assert user["permissions"]["network"] == "rw"

    def test_list_users(self, db):
        create_user("listuser1", "pass1", {"firewall": "rw"})
        create_user("listuser2", "pass2", {"network": "rw"})
        users = list_users()
        # admin (builtin) + listuser1 + listuser2
        assert len(users) == 3
        usernames = {u["username"] for u in users}
        assert "listuser1" in usernames

    def test_delete_user(self, db):
        create_user("deluser", "password123")
        assert get_user("deluser") is not None
        delete_user("deluser")
        assert get_user("deluser") is None


# ---------- Integration-style login flow ----------


class TestLoginFlow:
    def setup_method(self) -> None:
        reset_db_for_test()
        get_db()

    def test_full_login_flow(self):
        """Create user → verify password → generate tokens → validate tokens."""
        user = create_user("testadmin", "secretpass", {"firewall": "rw", "auth": "rw"})
        assert verify_user_password("testadmin", "secretpass") is not None

        tokens = generate_tokens("testadmin", user["permissions"])
        payload = validate_token(
            tokens["access_token"], "access", session_id=tokens["session_id"]
        )
        assert payload is not None
        assert payload["sub"] == "testadmin"
        assert payload["permissions"]["firewall"] == "rw"

    def test_logout_flow(self):
        """Generate token → blacklist JTI → verify token is rejected."""
        create_user("testuser", "password123")
        tokens = generate_tokens("testuser", {})
        payload = decode_token(tokens["access_token"])
        assert payload is not None

        blacklist_token(payload["jti"])

        result = validate_token(
            tokens["access_token"], "access", session_id=tokens["session_id"]
        )
        assert result is None

    def test_token_refresh_flow(self):
        """Generate refresh → get access → blacklist old refresh → validate new."""
        create_user("user1", "password123")
        refresh = generate_refresh_token("user1")
        payload = validate_token(refresh, "refresh")
        assert payload is not None

        blacklist_token(payload["jti"])
        result = validate_token(refresh, "refresh")
        assert result is None

        new_refresh = generate_refresh_token("user1")
        new_payload = validate_token(new_refresh, "refresh")
        assert new_payload is not None
        assert new_payload["jti"] != payload["jti"]


# ═══════════════════════════════════════════════════════════════════════════
# WebAuthn tests (Phase 2)
# ═══════════════════════════════════════════════════════════════════════════

_FAKE_CRED_ID = base64.urlsafe_b64encode(b"fakecred012345").decode().rstrip("=")
_FAKE_CRED_ID_BYTES = b"fakecred012345"
_FAKE_PUBLIC_KEY = base64.urlsafe_b64encode(b"fakepubkey01234").decode().rstrip("=")
_FAKE_CHALLENGE = base64.urlsafe_b64encode(b"fakechallenge!!").decode().rstrip("=")


@patch.dict(
    os.environ,
    {
        "VACUUM_WALL_DB_BACKEND": "sqlite",
        "VACUUM_WALL_DB_PATH": ":memory:",
        "PYTHONDONTWRITEBYTECODE": "1",
    },
)
class TestWebAuthnConfig:
    """Test WebAuthn configuration helpers."""

    def test_get_rp_defaults(self) -> None:
        from lib.webauthn import get_management_domains, get_rp_name, is_enabled

        assert is_enabled() is True
        assert get_rp_name() == "Vacuum Wall"
        # get_management_domains reads from nginx config
        domains = get_management_domains()
        assert isinstance(domains, list)


@patch.dict(
    os.environ,
    {
        "VACUUM_WALL_DB_BACKEND": "sqlite",
        "VACUUM_WALL_DB_PATH": ":memory:",
        "PYTHONDONTWRITEBYTECODE": "1",
    },
)
class TestWebAuthnB64urlHelpers:
    """Test base64url encoding/decoding used by WebAuthn."""

    def test_b64url_roundtrip(self) -> None:
        from lib.webauthn import b64u_decode, b64u_encode

        original = b"hello world 123 !@#"
        encoded = b64u_encode(original)
        decoded = b64u_decode(encoded)
        assert decoded == original

    def test_b64url_binary(self) -> None:
        from lib.webauthn import b64u_decode, b64u_encode

        original = bytes(range(256))
        encoded = b64u_encode(original)
        decoded = b64u_decode(encoded)
        assert decoded == original


@patch.dict(
    os.environ,
    {
        "VACUUM_WALL_DB_BACKEND": "sqlite",
        "VACUUM_WALL_DB_PATH": ":memory:",
        "PYTHONDONTWRITEBYTECODE": "1",
    },
)
class TestWebAuthnRegistration:
    """Test WebAuthn registration flow with mocked library."""

    def setup_method(self) -> None:
        lib.db.reset_db_for_test()
        get_db()

    def test_create_registration_options_basic(self) -> None:
        from lib.webauthn import create_registration_options

        options = create_registration_options(
            "testuser",
            origin="https://wall.example.com",
            rp_id="wall.example.com",
        )

        assert isinstance(options, dict)
        assert "challenge" in options
        assert "rp" in options
        assert "user" in options
        assert "pubKeyCredParams" in options
        assert options["rp"]["id"] == "wall.example.com"
        assert options["user"]["name"] == "testuser"
        assert len(options["pubKeyCredParams"]) >= 1

    def test_verify_registration_stores_in_db(self) -> None:
        from lib.webauthn import b64u_encode

        with patch("lib.webauthn.verify_registration_response") as mock_verify:
            mock_credential = MagicMock()
            mock_credential.id = _FAKE_CRED_ID_BYTES
            mock_credential.public_key = b"fakepubkey01234"
            mock_credential.sign_count = 0

            col = MagicMock()
            col.credential = mock_credential
            mock_verify.return_value = col

            from lib.webauthn import verify_registration

            _insert_cred("testuser")

            result = verify_registration(
                "testuser",
                {
                    "id": _FAKE_CRED_ID,
                    "rawId": _FAKE_CRED_ID,
                    "type": "public-key",
                    "response": {
                        "clientDataJSON": b64u_encode(b"{}"),
                        "attestationObject": b64u_encode(b"dummy"),
                        "transports": [],
                    },
                },
                {"challenge": b64u_encode(b"testchallenge123")},
                "My Key",
                origin="https://wall.example.com",
                rp_id="wall.example.com",
            )

            assert "id" in result
            assert result["name"] == "My Key"
            assert isinstance(result["transports"], list)
            assert isinstance(result["sign_count"], int)

            cursor = get_db().conn.execute(
                "SELECT credential_id, name FROM webauthn_creds WHERE name = ?",
                ("My Key",),
            )
            assert len(cursor.fetchall()) == 1

    def test_create_registration_excludes_existing(self) -> None:
        _insert_cred("testuser", cred_id="existingcred")

        from lib.webauthn import create_registration_options

        options = create_registration_options(
            "testuser",
            origin="https://wall.example.com",
            rp_id="wall.example.com",
        )

        # Should contain the existing credential in exclude list
        exclude_ids = [c["id"] for c in options.get("excludeCredentials", [])]
        assert "existingcred" in exclude_ids


def _insert_cred(
    username,
    cred_id="Y3JlZDE",
    public_key="cGsx",
    sign_count=0,
    name="",
    transports='["internal"]',
):
    """Convenience: insert a webauthn_creds row directly.

    Also creates the user in the users table if they don't exist (FK constraint).
    """
    conn = get_db().conn
    conn.execute(
        "INSERT OR IGNORE INTO users (username, password_hash, jwt_secret) VALUES (?, ?, ?)",
        (username, "testhash", "test-secret"),
    )
    conn.execute(
        "INSERT INTO webauthn_creds (username, credential_id, public_key, sign_count, name, transports) VALUES (?, ?, ?, ?, ?, ?)",
        (username, cred_id, public_key, sign_count, name, transports),
    )


@patch.dict(
    os.environ,
    {
        "VACUUM_WALL_DB_BACKEND": "sqlite",
        "VACUUM_WALL_DB_PATH": ":memory:",
        "PYTHONDONTWRITEBYTECODE": "1",
    },
)
class TestWebAuthnAuthentication:
    """Test WebAuthn authentication flow with mocked library."""

    def setup_method(self) -> None:
        lib.db.reset_db_for_test()
        get_db()

    def test_create_auth_options_no_credentials(self) -> None:
        from lib.webauthn import create_authentication_options

        result = create_authentication_options(
            "nonexistentuser", rp_id="wall.example.com"
        )
        assert result is None

    def test_create_auth_options_with_credentials(self) -> None:
        _insert_cred("testuser")

        from lib.webauthn import create_authentication_options

        result = create_authentication_options("testuser", rp_id="wall.example.com")
        assert result is not None

        assert result is not None
        assert len(result["allowCredentials"]) == 1

    def test_verify_authentication_not_found(self) -> None:
        from lib.webauthn import verify_authentication

        with pytest.raises(ValueError, match="Credential not found"):
            verify_authentication(
                "testuser",
                {
                    "id": "nonexistent",
                    "response": {
                        "clientDataJSON": "",
                        "authenticatorData": "",
                        "signature": "",
                    },
                },
                {"challenge": _FAKE_CHALLENGE},
                origin="https://wall.example.com",
                rp_id="wall.example.com",
            )

    def test_verify_authentication_success(self) -> None:
        from lib.webauthn import b64u_encode

        _insert_cred("testuser", _FAKE_CRED_ID, _FAKE_PUBLIC_KEY, 0)

        with patch("lib.webauthn.verify_authentication_response") as mock_verify:
            mock_col = MagicMock()
            mock_col.credential_sign_count = 1
            mock_verify.return_value = mock_col

            from lib.webauthn import verify_authentication

            result = verify_authentication(
                "testuser",
                {
                    "id": _FAKE_CRED_ID,
                    "response": {
                        "clientDataJSON": b64u_encode(b"{}"),
                        "authenticatorData": b64u_encode(b"aa"),
                        "signature": b64u_encode(b"sig"),
                    },
                },
                {"challenge": _FAKE_CHALLENGE, "sign_count": 0},
                origin="https://wall.example.com",
                rp_id="wall.example.com",
            )

            assert result is True


@patch.dict(
    os.environ,
    {
        "VACUUM_WALL_DB_BACKEND": "sqlite",
        "VACUUM_WALL_DB_PATH": ":memory:",
        "PYTHONDONTWRITEBYTECODE": "1",
    },
)
class TestWebAuthnCredentials:
    """Test credential listing and removal."""

    def setup_method(self) -> None:
        lib.db.reset_db_for_test()
        get_db()

    def test_list_credentials_empty(self) -> None:
        from lib.webauthn import list_credentials

        assert list_credentials("testuser") == []

    def test_list_credentials_with_data(self) -> None:
        _insert_cred("testuser", name="My Key", transports='["internal", "hybrid"]')

        from lib.webauthn import list_credentials

        result = list_credentials("testuser")
        assert len(result) == 1
        assert result[0]["name"] == "My Key"
        assert result[0]["sign_count"] == 0
        assert result[0]["transports"] == ["internal", "hybrid"]

    def test_remove_credential_success(self) -> None:
        _insert_cred("testuser")

        from lib.webauthn import list_credentials, remove_credential

        result = remove_credential("testuser", "Y3JlZDE")
        assert result is True

        assert list_credentials("testuser") == []

    def test_remove_credential_not_found(self) -> None:
        from lib.webauthn import remove_credential

        with pytest.raises(ValueError, match="not found"):
            remove_credential("testuser", "nonexistent")

    def test_remove_credential_wrong_user(self) -> None:
        """Remove fails for another user's credential with same error as not-found.

        Both cases return 404 — no distinction leaked to prevent credential
        enumeration. Internal ownership check still prevents cross-user deletion.
        """
        _insert_cred("otheruser", cred_id=_FAKE_CRED_ID)

        from lib.webauthn import remove_credential

        with pytest.raises(ValueError, match="not found"):
            remove_credential("testuser", _FAKE_CRED_ID)


class TestWebAuthnBlueprintOwnership:
    """Test Flask blueprint enforces JWT username on WebAuthn registration.

    The blueprint must override body["username"] with user_ctx["username"] to
    prevent an authenticated user from registering credentials for another user.

    This mirrors the pattern used by the change_password endpoint.
    """

    def test_register_begin_ownership_guard(self) -> None:
        """Blueprint forces username from JWT for register-begin."""
        # Simulate the blueprint logic:
        #   user_ctx = getattr(request, "_user_ctx", None)
        #   if user_ctx is not None:
        #       body["username"] = user_ctx["username"]

        # Attacker scenario: alice sends username "bob"
        user_ctx = {"username": "alice"}
        body = {"username": "bob"}
        if user_ctx is not None:
            body["username"] = user_ctx["username"]
        # Server-side username is "alice", not the attacker-supplied "bob"
        assert body["username"] == "alice"

    def test_register_finish_ownership_guard(self) -> None:
        """Blueprint forces username from JWT for register-finish."""
        user_ctx = {"username": "alice"}
        body = {"username": "malicious", "credential_response": {}}
        if user_ctx is not None:
            body["username"] = user_ctx["username"]
        assert body["username"] == "alice"

    def test_register_no_context(self) -> None:
        """Without _user_ctx, body username passes through (daemon direct call)."""
        user_ctx = None
        body = {"username": "daemon_user"}
        if user_ctx is not None:
            body["username"] = user_ctx["username"]
        # No override — body username preserved
        assert body["username"] == "daemon_user"


class TestTokenValidationEdgeCases:
    """Test token validation edge cases: session_id semantics, refresh tokens."""

    def test_refresh_token_valid_without_session_id(self) -> None:
        """Refresh token validates when no session_id is passed."""
        token = generate_refresh_token("admin")
        payload = validate_token(token, "refresh")
        assert payload is not None
        assert payload["sub"] == "admin"
        assert payload["type"] == "refresh"

    def test_refresh_token_valid_with_session_id_none(self) -> None:
        """Refresh token validates when session_id=None is passed."""
        token = generate_refresh_token("admin")
        payload = validate_token(token, "refresh", session_id=None)
        assert payload is not None

    def test_access_token_valid_with_wrong_type_refresh(self) -> None:
        """Access token is rejected when validated as refresh type."""
        token = generate_access_token("admin", {"firewall": "rw"}, session_id="sess")
        payload = validate_token(token, "refresh")
        assert payload is None

    def test_refresh_token_with_session_id_validates(self) -> None:
        """Refresh token with session_id validates when matching session_id provided."""
        create_user("sessuser", "password123")
        token = generate_refresh_token("sessuser", session_id="mysession")
        payload = validate_token(token, "refresh", session_id="mysession")
        assert payload is not None
        assert payload["sub"] == "sessuser"
        assert payload["session_id"] == "mysession"

    def test_refresh_token_with_wrong_session_id_rejected(self) -> None:
        """Refresh token with session_id is rejected when wrong session_id provided."""
        create_user("sessuser", "password123")
        token = generate_refresh_token("sessuser", session_id="mysession")
        payload = validate_token(token, "refresh", session_id="wrongsession")
        assert payload is None

    def test_require_session_accepts_matching_session(self) -> None:
        """require_session=True validates when matching session_id provided."""
        create_user("reqsess", "password123")
        token = generate_refresh_token("reqsess", session_id="mytoken-session")
        payload = validate_token(
            token,
            "refresh",
            session_id="mytoken-session",
            require_session=True,
        )
        assert payload is not None

    def test_require_session_rejects_when_session_id_omitted(self) -> None:
        """require_session=True rejects when token has session_id but request passes None."""
        create_user("reqsess2", "password123")
        token = generate_refresh_token("reqsess2", session_id="mytoken-session")
        payload = validate_token(
            token,
            "refresh",
            session_id=None,
            require_session=True,
        )
        assert payload is None

    def test_require_session_rejects_wrong_session(self) -> None:
        """require_session=True rejects when session_id doesn't match."""
        create_user("reqsess3", "password123")
        token = generate_refresh_token("reqsess3", session_id="mytoken-session")
        payload = validate_token(
            token,
            "refresh",
            session_id="wrong-session",
            require_session=True,
        )
        assert payload is None

    def test_require_session_with_no_session_in_token(self) -> None:
        """require_session=True with a token that has no session_id claim works normally."""
        create_user("reqsess4", "password123")
        token = generate_refresh_token("reqsess4")
        payload = validate_token(token, "refresh", require_session=True)
        assert payload is not None

    def test_refresh_token_generate_tokens_includes_session_id(self) -> None:
        """generate_tokens produces refresh token with session_id."""
        create_user("tokuser", "password123")
        tokens = generate_tokens("tokuser", {"firewall": "rw"})
        refresh = decode_token(tokens["refresh_token"])
        assert refresh is not None
        assert refresh["session_id"] == tokens["session_id"]


# ═══════════════════════════════════════════════════════════════════════════
# Multi-user tests (Phase 3)
# ═══════════════════════════════════════════════════════════════════════════


@patch.dict(
    os.environ,
    {
        "VACUUM_WALL_DB_BACKEND": "sqlite",
        "VACUUM_WALL_DB_PATH": ":memory:",
        "PYTHONDONTWRITEBYTECODE": "1",
    },
)
class TestMultiUserAdmin:
    """Test multi-user admin features: CRUD, permissions, self-deletion."""

    def setup_method(self) -> None:
        reset_db_for_test()
        get_db()

    def test_create_user_with_mixed_permissions(self) -> None:
        """User created with different permission levels per subsystem."""
        create_user(
            "mixeduser",
            "password123",
            {
                "firewall": "rw",
                "network": "read",
                "logs": "rw",
                "dhcp": "rw",
            },
        )
        user = get_user("mixeduser")
        assert user is not None
        assert user["permissions"]["firewall"] == "rw"
        assert user["permissions"]["network"] == "read"
        assert user["permissions"]["logs"] == "rw"
        assert user["permissions"]["dhcp"] == "rw"
        assert "prox" not in user["permissions"]

    def test_read_only_permissions(self) -> None:
        """Verify read-only user can be created and queried."""
        create_user(
            "readonly",
            "password123",
            {
                "firewall": "read",
                "network": "read",
                "dhcp": "read",
                "proxy": "read",
            },
        )
        user = get_user("readonly")
        assert user is not None
        for _sub, level in user["permissions"].items():
            assert level == "read"

    def test_update_permissions_replaces_all(self) -> None:
        """Updating permissions replaces existing set entirely."""
        create_user(
            "permchange",
            "password123",
            {
                "firewall": "rw",
                "network": "rw",
            },
        )
        user = get_user("permchange")
        assert "firewall" in user["permissions"]
        assert "network" in user["permissions"]
        assert "dhcp" not in user["permissions"]

        update_permissions(
            "permchange",
            {
                "dhcp": "rw",
                "logs": "read",
            },
        )
        user = get_user("permchange")
        assert "dhcp" in user["permissions"]
        assert "logs" in user["permissions"]
        assert "firewall" not in user["permissions"]
        assert "network" not in user["permissions"]

    def test_create_user_no_permissions(self) -> None:
        """User created without permissions gets empty permission set."""
        perms = {"firewall": "rw"}
        user = create_user("noperm", "password123", perms)
        assert user["permissions"]["firewall"] == "rw"

        update_permissions("noperm", {})
        user = get_user("noperm")
        assert not user["permissions"]

    def test_delete_user_cascades_permissions(self) -> None:
        """Deleting a user also removes their permission rows."""
        create_user(
            "cscduser",
            "password123",
            {
                "firewall": "rw",
                "network": "read",
                "auth": "rw",
            },
        )
        user = get_user("cscduser")
        assert len(user["permissions"]) == 3

        delete_user("cscduser")
        assert get_user("cscduser") is None
        # Verify permissions are cascaded-deleted
        from lib.db import get_db as _get_db

        rows = _get_db().query(Q_SELECT_PERMISSIONS, ("cscduser",))
        assert len(rows) == 0

    def test_delete_user_cascades_webauthn_creds(self) -> None:
        """Deleting a user also removes their WebAuthn credentials."""
        _insert_cred("wgscduser", cred_id="test-cred-id-123")

        from lib.webauthn import list_credentials

        creds = list_credentials("wgscduser")
        assert len(creds) == 1

        delete_user("wgscduser")
        creds = list_credentials("wgscduser")
        assert len(creds) == 0

    def test_delete_builtin_admin_rejected(self) -> None:
        """Deleting the builtin admin user is rejected."""
        with pytest.raises(ValueError, match="Cannot delete builtin admin"):
            delete_user("admin")

    def test_self_deletion_prevention(self) -> None:
        """Self-deletion is prevented at the Flask blueprint layer.

        The daemon handler accepts the delete request but trusts the Flask
        middleware (which has _user_ctx) to block self-deletion first.

        This test verifies the handler still works for normal deletion.
        The Flask blueprint test verifies self-deletion blocking.
        """
        from daemon.handlers.auth import auth_delete_user

        create_user("tester", "password123", {"auth": "rw"})
        create_user("target", "password123", {"firewall": "read"})

        # Handler can delete other users
        mock_request = MagicMock()
        result = auth_delete_user(mock_request, {"username": "target"})
        assert result["ok"] is True
        assert get_user("target") is None

    def test_flask_self_deletion_logic(self) -> None:
        """Flask blueprint blocks self-deletion: verify the guard logic."""
        # The blueprint check is a simple comparison:
        #   user_ctx.get("username") == username
        # We verify this logic directly.

        # Self-deletion scenario: same user
        user_ctx = {"username": "alice"}
        username = "alice"
        # When user_ctx is not None and username matches, blueprint blocks
        assert user_ctx is not None
        assert user_ctx.get("username") == username  # Would trigger 403

        # Different user scenario
        user_ctx2 = {"username": "admin"}
        username2 = "alice"
        # When usernames differ, deletion proceeds
        assert user_ctx2.get("username") != username2

        # No context scenario
        user_ctx3 = None
        # When user_ctx is None, delete proceeds (no guard)
        assert user_ctx3 is None  # No guard triggered

    def test_self_deletion_prevention_no_context(self) -> None:
        """When _user_ctx is not set, deletion proceeds (direct daemon call)."""
        from daemon.handlers.auth import auth_delete_user

        create_user("directuser", "password123", {"firewall": "read"})

        # No _user_ctx — daemon called directly (e.g., batch API)
        mock_request = MagicMock()
        mock_request._user_ctx = None
        result = auth_delete_user(mock_request, {"username": "directuser"})
        assert result["ok"] is True
        assert get_user("directuser") is None

    def test_multiple_users_independent_states(self) -> None:
        """Multiple users maintain independent password and permission state."""
        create_user("user_a", "pass_a", {"firewall": "rw"})
        create_user("user_b", "pass_b", {"dhcp": "read"})

        # Passwords are independent
        assert verify_user_password("user_a", "pass_a") is not None
        assert verify_user_password("user_a", "pass_b") is None
        assert verify_user_password("user_b", "pass_b") is not None
        assert verify_user_password("user_b", "pass_a") is None

        # Permissions are independent
        perms_a = get_user("user_a")["permissions"]
        perms_b = get_user("user_b")["permissions"]
        assert "firewall" in perms_a
        assert "dhcp" not in perms_a
        assert "dhcp" in perms_b
        assert "firewall" not in perms_b

        # Updating one user doesn't affect the other
        update_permissions("user_a", {"logs": "rw"})
        perms_b_after = get_user("user_b")["permissions"]
        assert "dhcp" in perms_b_after
        assert "logs" not in perms_b_after

    def test_permission_update_nonexistent_user(self) -> None:
        """Updating permissions for a nonexistent user raises ValueError."""
        with pytest.raises(ValueError, match="not found"):
            update_permissions("nonexistent", {"firewall": "rw"})

    def test_list_users_with_permissions(self) -> None:
        """list_users returns all users with their full permission dicts."""
        create_user("alice", "pass1", {"firewall": "rw", "network": "read"})
        create_user("bob", "pass2", {"dhcp": "rw", "auth": "rw"})
        create_user("charlie", "pass3", {})

        users = list_users()
        # admin (builtin) + alice + bob + charlie
        assert len(users) == 4

        by_name = {u["username"]: u for u in users}

        assert "firewall" in by_name["alice"]["permissions"]
        assert "network" in by_name["alice"]["permissions"]
        assert "dhcp" not in by_name["alice"]["permissions"]

        assert "dhcp" in by_name["bob"]["permissions"]
        assert "auth" in by_name["bob"]["permissions"]

        assert not by_name["charlie"]["permissions"]

    def test_all_subsystems_constant(self) -> None:
        """ALL_SUBSYSTEMS contains expected subsystem names."""
        from lib.auth_users import ALL_SUBSYSTEMS

        expected = {
            "firewall",
            "network",
            "dhcp",
            "proxy",
            "certs",
            "wireguard",
            "logs",
            "status",
            "auth",
        }
        assert set(ALL_SUBSYSTEMS) == expected


class TestPermissionMiddleware:
    """Test Flask middleware permission enforcement logic.

    These tests verify the permission checking logic that would be applied
    by the Flask before_request middleware (server._auth_middleware).
    """

    def test_permission_check_read_allowed(self) -> None:
        """GET requests allowed with 'read' or 'rw' for the subsystem."""
        from webui.server import _has_permission

        perms = {"firewall": "read"}
        assert _has_permission(perms, "firewall", "GET") is True

        perms_rw = {"firewall": "rw"}
        assert _has_permission(perms_rw, "firewall", "GET") is True

    def test_permission_check_read_denied_write(self) -> None:
        """Non-GET requests denied with 'read' permission."""
        from webui.server import _has_permission

        perms = {"firewall": "read"}
        assert _has_permission(perms, "firewall", "POST") is False
        assert _has_permission(perms, "firewall", "DELETE") is False

    def test_permission_check_rw_allowed(self) -> None:
        """'rw' permission allows all HTTP methods."""
        from webui.server import _has_permission

        perms = {"firewall": "rw"}
        assert _has_permission(perms, "firewall", "GET") is True
        assert _has_permission(perms, "firewall", "POST") is True
        assert _has_permission(perms, "firewall", "DELETE") is True

    def test_permission_check_no_access(self) -> None:
        """Missing subsystem permission denies all access."""
        from webui.server import _has_permission

        perms = {"logs": "read"}
        assert _has_permission(perms, "firewall", "GET") is False
        assert _has_permission(perms, "firewall", "POST") is False

    def test_subsystem_extraction(self) -> None:
        """Subsystem name correctly extracted from API path."""
        from webui.server import _subsystem_from_path

        assert _subsystem_from_path("/api/firewall/config") == "firewall"
        assert _subsystem_from_path("/api/auth/users") == "auth"
        assert _subsystem_from_path("/api/dhcp/leases/subpath") == "dhcp"
        assert _subsystem_from_path("/") is None
        assert _subsystem_from_path("/static/app.js") is None


class TestRefreshRotationLogout:
    """Refresh-rotation + logout interaction (regression for the reorder in
    0889ef0: clearing the refresh_tokens row after rotation left logout with
    nothing to blacklist, so the rotated token stayed valid)."""

    def test_logout_revokes_rotated_refresh_token(self) -> None:
        """Logout must blacklist the current refresh token after rotation."""
        from daemon.handlers.auth import auth_logout, auth_refresh

        create_user("rotuser", "password123", {"auth": "rw"})
        tokens = generate_tokens("rotuser", {"auth": "rw"})

        rotated = auth_refresh(
            MagicMock(),
            {
                "refresh_token": tokens["refresh_token"],
                "session_id": tokens["session_id"],
            },
        )["tokens"]

        # The rotated token must be valid before logout (rotation works).
        payload = validate_token(
            rotated["refresh_token"], "refresh", session_id=rotated["session_id"]
        )
        assert payload is not None

        auth_logout(MagicMock(), {"jti": None, "username": "rotuser"})

        with pytest.raises(ValueError, match="Invalid or expired refresh token"):
            auth_refresh(
                MagicMock(),
                {
                    "refresh_token": rotated["refresh_token"],
                    "session_id": rotated["session_id"],
                },
            )

    def test_old_refresh_token_blacklisted_on_rotation(self) -> None:
        """The pre-rotation refresh token must be blacklisted immediately."""
        from daemon.handlers.auth import auth_refresh

        create_user("rotuser2", "password123", {"auth": "rw"})
        tokens = generate_tokens("rotuser2", {"auth": "rw"})

        auth_refresh(
            MagicMock(),
            {
                "refresh_token": tokens["refresh_token"],
                "session_id": tokens["session_id"],
            },
        )

        assert (
            validate_token(
                tokens["refresh_token"], "refresh", session_id=tokens["session_id"]
            )
            is None
        )


class TestMalformedTokenPayload:
    """Malformed/untrusted JWT payloads must be rejected (401), not 500."""

    def test_decode_non_object_payload_returns_none(self) -> None:
        """A payload segment decoding to non-object JSON is rejected."""
        hdr = (
            base64.urlsafe_b64encode(b'{"alg":"HS256","typ":"JWT"}')
            .decode()
            .rstrip("=")
        )
        payload = base64.urlsafe_b64encode(b'"hello"').decode().rstrip("=")
        token = f"{hdr}.{payload}.signature"
        assert decode_token(token) is None

    def test_middleware_crafted_token_returns_401(self) -> None:
        """Crafted Bearer token on a protected route returns JSON 401, not 500."""
        from webui.server import app

        client = app.test_client()
        hdr = (
            base64.urlsafe_b64encode(b'{"alg":"HS256","typ":"JWT"}')
            .decode()
            .rstrip("=")
        )
        payload = base64.urlsafe_b64encode(b'"hello"').decode().rstrip("=")
        token = f"{hdr}.{payload}.signature"

        res = client.get(
            "/api/auth/session",
            headers={"Authorization": f"Bearer {token}", "X-Session-Id": "x"},
        )
        assert res.status_code == 401
        assert res.get_json() == {"ok": False, "error": "unauthorized"}


class TestBuiltinAdminSeeding:
    """Fallback seeding of the builtin admin user (lib.db._seed_builtin_admin)."""

    def test_seed_runs_and_creates_admin(self) -> None:
        """A fresh (empty) DB gets the builtin admin with full permissions."""
        from lib.auth_users import ALL_SUBSYSTEMS
        from lib.db import _seed_builtin_admin

        db = get_db()
        _seed_builtin_admin(db)

        user = get_user("admin")
        assert user is not None
        assert user["permissions"] == {s: "rw" for s in ALL_SUBSYSTEMS}

    def test_seed_noop_when_admin_exists(self) -> None:
        """Seeding is a no-op when any user (here: admin) already exists."""
        from lib.db import Q_SELECT_ALL_USERS, _seed_builtin_admin

        db = get_db()
        _seed_builtin_admin(db)

        real_query = db.query
        calls = {"n": 0}

        def counting_query(query_id, params=()):
            if query_id == Q_SELECT_ALL_USERS:
                calls["n"] += 1
            return real_query(query_id, params)

        with patch.object(db, "query", side_effect=counting_query):
            _seed_builtin_admin(db)
        # Early-return path: only the users-present check runs.
        assert calls["n"] >= 1

    def test_seed_concurrent_lose_race(self) -> None:
        """Concurrent seeding: if the users-present check sees a stale (empty)
        view and the insert then loses the race, the loser re-checks, finds
        the winner's admin, and returns instead of raising IntegrityError."""
        from lib.db import Q_SELECT_ALL_USERS, _seed_builtin_admin

        db = get_db()
        # get_db() already seeded admin for this fresh in-memory DB.
        assert get_user("admin") is not None

        real_query = db.query
        calls = {"n": 0}

        def fake_query(query_id, params=()):
            if query_id == Q_SELECT_ALL_USERS:
                calls["n"] += 1
                if calls["n"] == 1:
                    return []  # stale view: users-present check misses seeder
            return real_query(query_id, params)

        with patch.object(db, "query", side_effect=fake_query):
            _seed_builtin_admin(db)  # must not raise

        assert get_user("admin") is not None

    def test_seed_skipped_when_users_exist(self, tmp_path) -> None:
        """Last-resort rule: a DB that already has users gets no seeded
        admin — a non-default bootstrap user must not be shadowed."""
        from lib.auth_users import ALL_SUBSYSTEMS
        from lib.db_sqlite import SQLiteBackend

        db_file = tmp_path / "auth.db"

        # A prior process (the bootstrap run) created the operator account.
        first = SQLiteBackend(str(db_file))
        first.init_tables()
        first.run(Q_INSERT_USER, ("alice", hash_password("alice-pw"), "jwt-secret"))
        for sub in ALL_SUBSYSTEMS:
            first.run(Q_UPSERT_PERMISSION, ("alice", sub, "rw"))

        # A fresh service process initializes the same DB file.
        reset_db_for_test()
        os.environ["VACUUM_WALL_DB_PATH"] = str(db_file)
        db = get_db()

        usernames = {row["username"] for row in db.query(Q_SELECT_ALL_USERS)}
        assert usernames == {"alice"}
        assert get_user("admin") is None

    def test_seed_skipped_via_env(self) -> None:
        """VACUUM_WALL_SEED_BUILTIN_ADMIN=0 (set by bootstrap_auth.py)
        suppresses the seed even on a completely empty DB."""
        os.environ["VACUUM_WALL_SEED_BUILTIN_ADMIN"] = "0"
        db = get_db()
        assert db.query(Q_SELECT_ALL_USERS) == []
        assert find_user("admin") is None

    def test_bootstrap_flow_creates_exactly_one_user(self) -> None:
        """Simulates bootstrap_auth.py's main() on a fresh DB: with the seed
        suppressed, bootstrap creates exactly the operator account and no
        hardcoded admin shadow (regression for --mgmt-user != admin leaving
        an unrecoverable superuser)."""
        from lib.auth_users import ALL_SUBSYSTEMS

        os.environ["VACUUM_WALL_SEED_BUILTIN_ADMIN"] = "0"
        assert find_user("alice") is None
        create_user("alice", "secret123", {sub: "rw" for sub in ALL_SUBSYSTEMS})

        usernames = {row["username"] for row in get_db().query(Q_SELECT_ALL_USERS)}
        assert usernames == {"alice"}
        assert find_user("admin") is None