greetd-moongreet/tests/test_integration.py

# ABOUTME: Integration tests — verifies the login flow end-to-end via a mock greetd socket.
# ABOUTME: Tests the IPC sequence: create_session → post_auth → start_session.

import json
import os
import socket
import struct
import threading
from pathlib import Path

import pytest

from moongreet.greeter import faillock_warning, FAILLOCK_MAX_ATTEMPTS, LAST_SESSION_DIR
from moongreet.i18n import load_strings
from moongreet.ipc import create_session, post_auth_response, start_session, cancel_session


class MockGreetd:
    """A mock greetd server that listens on a Unix socket and responds to IPC messages."""

    def __init__(self, sock_path: Path) -> None:
        self.sock_path = sock_path
        self._responses: list[dict] = []
        self._received: list[dict] = []
        self._server: socket.socket | None = None

    def expect(self, response: dict) -> None:
        """Queue a response to send for the next received message."""
        self._responses.append(response)

    @property
    def received(self) -> list[dict]:
        return self._received

    def start(self) -> None:
        """Start the mock server in a background thread."""
        self._server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self._server.bind(str(self.sock_path))
        self._server.listen(1)
        self._thread = threading.Thread(target=self._serve, daemon=True)
        self._thread.start()

    @staticmethod
    def _recvall(conn: socket.socket, n: int) -> bytes:
        """Receive exactly n bytes from a socket, handling fragmented reads."""
        buf = bytearray()
        while len(buf) < n:
            chunk = conn.recv(n - len(buf))
            if not chunk:
                break
            buf.extend(chunk)
        return bytes(buf)

    def _serve(self) -> None:
        conn, _ = self._server.accept()
        try:
            for response in self._responses:
                # Receive a message
                header = self._recvall(conn, 4)
                if len(header) < 4:
                    break
                length = struct.unpack("=I", header)[0]
                payload = self._recvall(conn, length)
                msg = json.loads(payload.decode("utf-8"))
                self._received.append(msg)

                # Send response
                resp_payload = json.dumps(response).encode("utf-8")
                conn.sendall(struct.pack("=I", len(resp_payload)) + resp_payload)
        finally:
            conn.close()

    def stop(self) -> None:
        if self._server:
            self._server.close()


class TestLoginFlow:
    """Integration tests for the complete login flow via mock greetd."""

    def test_successful_login(self, tmp_path: Path) -> None:
        """Simulate a complete successful login: create → auth → start."""
        sock_path = tmp_path / "greetd.sock"
        mock = MockGreetd(sock_path)
        mock.expect({"type": "auth_message", "auth_message_type": "secret", "auth_message": "Password:"})
        mock.expect({"type": "success"})
        mock.expect({"type": "success"})
        mock.start()

        try:
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            sock.connect(str(sock_path))

            # Step 1: Create session
            response = create_session(sock, "dominik")
            assert response["type"] == "auth_message"

            # Step 2: Send password
            response = post_auth_response(sock, "geheim")
            assert response["type"] == "success"

            # Step 3: Start session
            response = start_session(sock, ["Hyprland"])
            assert response["type"] == "success"

            sock.close()
        finally:
            mock.stop()

        # Verify what the mock received
        assert mock.received[0] == {"type": "create_session", "username": "dominik"}
        assert mock.received[1] == {"type": "post_auth_message_response", "response": "geheim"}
        assert mock.received[2] == {"type": "start_session", "cmd": ["Hyprland"]}

    def test_wrong_password_sends_cancel(self, tmp_path: Path) -> None:
        """After a failed login, cancel_session must be sent to free the greetd session."""
        sock_path = tmp_path / "greetd.sock"
        mock = MockGreetd(sock_path)
        mock.expect({"type": "auth_message", "auth_message_type": "secret", "auth_message": "Password:"})
        mock.expect({"type": "error", "error_type": "auth_error", "description": "Authentication failed"})
        mock.expect({"type": "success"})  # Response to cancel_session
        mock.start()

        try:
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            sock.connect(str(sock_path))

            response = create_session(sock, "dominik")
            assert response["type"] == "auth_message"

            response = post_auth_response(sock, "falsch")
            assert response["type"] == "error"
            assert response["description"] == "Authentication failed"

            # The greeter must cancel the session after auth failure
            response = cancel_session(sock)
            assert response["type"] == "success"

            sock.close()
        finally:
            mock.stop()

        assert mock.received[2] == {"type": "cancel_session"}

    def test_stale_session_cancel_and_retry(self, tmp_path: Path) -> None:
        """When create_session fails due to a stale session, cancel and retry."""
        sock_path = tmp_path / "greetd.sock"
        mock = MockGreetd(sock_path)
        # First create_session → error (stale session)
        mock.expect({"type": "error", "error_type": "error", "description": "a session is already being configured"})
        # cancel_session → success
        mock.expect({"type": "success"})
        # Second create_session → auth_message (retry succeeds)
        mock.expect({"type": "auth_message", "auth_message_type": "secret", "auth_message": "Password:"})
        # post_auth_response → success
        mock.expect({"type": "success"})
        # start_session → success
        mock.expect({"type": "success"})
        mock.start()

        try:
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            sock.connect(str(sock_path))

            # Step 1: Create session fails
            response = create_session(sock, "dominik")
            assert response["type"] == "error"

            # Step 2: Cancel stale session
            response = cancel_session(sock)
            assert response["type"] == "success"

            # Step 3: Retry create session
            response = create_session(sock, "dominik")
            assert response["type"] == "auth_message"

            # Step 4: Send password
            response = post_auth_response(sock, "geheim")
            assert response["type"] == "success"

            # Step 5: Start session
            response = start_session(sock, ["niri-session"])
            assert response["type"] == "success"

            sock.close()
        finally:
            mock.stop()

        assert mock.received[0] == {"type": "create_session", "username": "dominik"}
        assert mock.received[1] == {"type": "cancel_session"}
        assert mock.received[2] == {"type": "create_session", "username": "dominik"}

    def test_multi_stage_auth_sends_cancel(self, tmp_path: Path) -> None:
        """When greetd sends a second auth_message after password, cancel the session."""
        sock_path = tmp_path / "greetd.sock"
        mock = MockGreetd(sock_path)
        mock.expect({"type": "auth_message", "auth_message_type": "secret", "auth_message": "Password:"})
        mock.expect({"type": "auth_message", "auth_message_type": "secret", "auth_message": "TOTP:"})
        mock.expect({"type": "success"})  # Response to cancel_session
        mock.start()

        try:
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            sock.connect(str(sock_path))

            # Step 1: Create session
            response = create_session(sock, "dominik")
            assert response["type"] == "auth_message"

            # Step 2: Send password — greetd responds with another auth_message
            response = post_auth_response(sock, "geheim")
            assert response["type"] == "auth_message"

            # Step 3: Cancel because multi-stage auth is not supported
            response = cancel_session(sock)
            assert response["type"] == "success"

            sock.close()
        finally:
            mock.stop()

        # Verify cancel was sent
        assert mock.received[2] == {"type": "cancel_session"}

    def test_cancel_session(self, tmp_path: Path) -> None:
        """Simulate cancelling a session after create."""
        sock_path = tmp_path / "greetd.sock"
        mock = MockGreetd(sock_path)
        mock.expect({"type": "auth_message", "auth_message_type": "secret", "auth_message": "Password:"})
        mock.expect({"type": "success"})
        mock.start()

        try:
            sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            sock.connect(str(sock_path))

            create_session(sock, "dominik")
            response = cancel_session(sock)
            assert response["type"] == "success"

            sock.close()
        finally:
            mock.stop()

        assert mock.received[1] == {"type": "cancel_session"}


class TestSessionCancellation:
    """Tests for cancelling an in-progress greetd session during user switch."""

    def test_cancel_closes_socket_and_sets_event(self, tmp_path: Path) -> None:
        """_cancel_pending_session should close the socket and set the cancelled event."""
        from moongreet.greeter import GreeterWindow

        win = GreeterWindow.__new__(GreeterWindow)
        win._greetd_sock_lock = threading.Lock()
        win._login_cancelled = threading.Event()

        # Create a real socket pair to verify close
        server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock_path = tmp_path / "test.sock"
        server.bind(str(sock_path))
        server.listen(1)
        client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        client.connect(str(sock_path))
        server.close()

        win._greetd_sock = client
        win._cancel_pending_session()

        assert win._login_cancelled.is_set()
        assert win._greetd_sock is None

    def test_cancel_is_noop_without_socket(self) -> None:
        """_cancel_pending_session should be safe to call when no socket exists."""
        from moongreet.greeter import GreeterWindow

        win = GreeterWindow.__new__(GreeterWindow)
        win._greetd_sock_lock = threading.Lock()
        win._login_cancelled = threading.Event()
        win._greetd_sock = None

        win._cancel_pending_session()

        assert win._login_cancelled.is_set()
        assert win._greetd_sock is None

    def test_cancel_does_not_block_main_thread(self, tmp_path: Path) -> None:
        """_cancel_pending_session must not do blocking I/O — only close the socket."""
        from moongreet.greeter import GreeterWindow

        win = GreeterWindow.__new__(GreeterWindow)
        win._greetd_sock_lock = threading.Lock()
        win._login_cancelled = threading.Event()

        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        win._greetd_sock = sock

        # Should complete nearly instantly (no IPC calls)
        import time
        start = time.monotonic()
        win._cancel_pending_session()
        elapsed = time.monotonic() - start

        assert elapsed < 0.1  # No blocking I/O

    def test_worker_exits_silently_when_cancelled(self, tmp_path: Path) -> None:
        """_login_worker should exit without showing an error when cancelled mid-flight."""
        from unittest.mock import MagicMock, patch
        from moongreet.greeter import GreeterWindow
        from moongreet.users import User

        win = GreeterWindow.__new__(GreeterWindow)
        win._greetd_sock_lock = threading.Lock()
        win._login_cancelled = threading.Event()
        win._greetd_sock = None
        win._failed_attempts = {}
        win._strings = MagicMock()

        # Set cancelled before the worker runs
        win._login_cancelled.set()

        # Create a socket that will fail (simulating closed socket)
        sock_path = tmp_path / "greetd.sock"
        server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        server.bind(str(sock_path))
        server.listen(1)

        user = User(username="dom", uid=1000, gecos="Dominik", home=Path("/home/dom"), shell="/bin/zsh")

        with patch("moongreet.greeter.GLib.idle_add") as mock_idle:
            win._login_worker(user, "pw", MagicMock(exec_cmd="niri-session"), str(sock_path))

            # Should NOT have scheduled any error callback
            for call in mock_idle.call_args_list:
                func = call[0][0]
                assert func != win._on_login_error, "Worker should not show error when cancelled"
                assert func != win._on_login_auth_error, "Worker should not show auth error when cancelled"

        server.close()


class TestFaillockWarning:
    """Tests for the faillock warning message logic."""

    def test_no_warning_on_zero_attempts(self) -> None:
        strings = load_strings("de")
        assert faillock_warning(0, strings) is None

    def test_no_warning_on_first_attempt(self) -> None:
        strings = load_strings("de")
        assert faillock_warning(1, strings) is None

    def test_warning_on_second_attempt(self) -> None:
        strings = load_strings("de")
        warning = faillock_warning(2, strings)
        assert warning is not None
        assert "1" in warning  # 1 Versuch übrig

    def test_warning_on_third_attempt(self) -> None:
        strings = load_strings("de")
        warning = faillock_warning(3, strings)
        assert warning is not None
        assert warning == strings.faillock_locked

    def test_warning_beyond_max_attempts(self) -> None:
        strings = load_strings("de")
        warning = faillock_warning(4, strings)
        assert warning is not None
        assert warning == strings.faillock_locked

    def test_max_attempts_constant_is_three(self) -> None:
        assert FAILLOCK_MAX_ATTEMPTS == 3


class TestLastUser:
    """Tests for saving and loading the last logged-in user."""

    def test_save_and_load_last_user(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
        cache_path = tmp_path / "cache" / "moongreet" / "last-user"
        monkeypatch.setattr("moongreet.greeter.LAST_USER_PATH", cache_path)

        from moongreet.greeter import GreeterWindow
        GreeterWindow._save_last_user("dominik")

        assert cache_path.exists()
        assert cache_path.read_text() == "dominik"

        result = GreeterWindow._load_last_user()
        assert result == "dominik"

    def test_load_last_user_missing_file(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
        cache_path = tmp_path / "nonexistent" / "last-user"
        monkeypatch.setattr("moongreet.greeter.LAST_USER_PATH", cache_path)

        from moongreet.greeter import GreeterWindow
        result = GreeterWindow._load_last_user()
        assert result is None

    def test_load_last_user_rejects_oversized_username(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
        cache_path = tmp_path / "last-user"
        cache_path.write_text("a" * 300)
        monkeypatch.setattr("moongreet.greeter.LAST_USER_PATH", cache_path)

        from moongreet.greeter import GreeterWindow
        result = GreeterWindow._load_last_user()
        assert result is None

    def test_load_last_user_rejects_invalid_characters(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
        cache_path = tmp_path / "last-user"
        cache_path.write_text("../../etc/passwd")
        monkeypatch.setattr("moongreet.greeter.LAST_USER_PATH", cache_path)

        from moongreet.greeter import GreeterWindow
        result = GreeterWindow._load_last_user()
        assert result is None


class TestLastSession:
    """Tests for saving and loading the last session per user."""

    def test_save_and_load_last_session(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setattr("moongreet.greeter.LAST_SESSION_DIR", tmp_path)

        from moongreet.greeter import GreeterWindow
        GreeterWindow._save_last_session("dominik", "Niri")

        session_file = tmp_path / "dominik"
        assert session_file.exists()
        assert session_file.read_text() == "Niri"

        result = GreeterWindow._load_last_session("dominik")
        assert result == "Niri"

    def test_load_last_session_missing_file(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setattr("moongreet.greeter.LAST_SESSION_DIR", tmp_path)

        from moongreet.greeter import GreeterWindow
        result = GreeterWindow._load_last_session("nobody")
        assert result is None

    def test_load_last_session_rejects_oversized_name(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setattr("moongreet.greeter.LAST_SESSION_DIR", tmp_path)
        (tmp_path / "dominik").write_text("A" * 300)

        from moongreet.greeter import GreeterWindow
        result = GreeterWindow._load_last_session("dominik")
        assert result is None

    def test_save_last_session_validates_username(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """Usernames with path traversal should not create files outside the cache dir."""
        monkeypatch.setattr("moongreet.greeter.LAST_SESSION_DIR", tmp_path)

        from moongreet.greeter import GreeterWindow
        GreeterWindow._save_last_session("../../etc/evil", "Niri")

        # Should not have created any file
        assert not (tmp_path / "../../etc/evil").exists()

    def test_regex_rejects_dot_dot_username(self) -> None:
        """Username '..' must not pass VALID_USERNAME validation."""
        from moongreet.greeter import VALID_USERNAME
        assert VALID_USERNAME.match("..") is None

    def test_regex_rejects_dot_username(self) -> None:
        """Username '.' must not pass VALID_USERNAME validation."""
        from moongreet.greeter import VALID_USERNAME
        assert VALID_USERNAME.match(".") is None

    def test_regex_allows_dot_in_middle(self) -> None:
        """Usernames like 'first.last' must still be valid."""
        from moongreet.greeter import VALID_USERNAME
        assert VALID_USERNAME.match("first.last") is not None

    def test_regex_rejects_leading_dot(self) -> None:
        """Usernames starting with '.' are rejected (hidden files)."""
        from moongreet.greeter import VALID_USERNAME
        assert VALID_USERNAME.match(".hidden") is None