"""Test auth of websocket API."""

from unittest.mock import patch

import aiohttp
from aiohttp import WSMsgType, web
import pytest

from homeassistant.auth.providers.homeassistant import HassAuthProvider
from homeassistant.components.websocket_api import DOMAIN
from homeassistant.components.websocket_api.auth import (
    TYPE_AUTH,
    TYPE_AUTH_INVALID,
    TYPE_AUTH_OK,
    TYPE_AUTH_REQUIRED,
)
from homeassistant.components.websocket_api.const import (
    SIGNAL_WEBSOCKET_CONNECTED,
    SIGNAL_WEBSOCKET_DISCONNECTED,
    URL,
)
from homeassistant.const import HASSIO_USER_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.setup import async_setup_component

from tests.test_util import mock_real_ip
from tests.typing import ClientSessionGenerator


@pytest.fixture
def track_connected(hass: HomeAssistant) -> dict[str, list[int]]:
    """Track connected and disconnected events."""
    connected_evt = []

    @callback
    def track_connected():
        connected_evt.append(1)

    async_dispatcher_connect(hass, SIGNAL_WEBSOCKET_CONNECTED, track_connected)
    disconnected_evt = []

    @callback
    def track_disconnected():
        disconnected_evt.append(1)

    async_dispatcher_connect(hass, SIGNAL_WEBSOCKET_DISCONNECTED, track_disconnected)

    return {"connected": connected_evt, "disconnected": disconnected_evt}


async def test_auth_events(
    hass: HomeAssistant,
    no_auth_websocket_client,
    local_auth: HassAuthProvider,
    hass_access_token: str,
    track_connected,
) -> None:
    """Test authenticating."""

    await test_auth_active_with_token(hass, no_auth_websocket_client, hass_access_token)

    assert len(track_connected["connected"]) == 1
    assert not track_connected["disconnected"]

    await no_auth_websocket_client.close()
    await hass.async_block_till_done()

    assert len(track_connected["disconnected"]) == 1


async def test_auth_via_msg_incorrect_pass(no_auth_websocket_client) -> None:
    """Test authenticating."""
    with patch(
        "homeassistant.components.websocket_api.auth.process_wrong_login",
    ) as mock_process_wrong_login:
        await no_auth_websocket_client.send_json(
            {"type": TYPE_AUTH, "api_password": "wrong"}
        )

        msg = await no_auth_websocket_client.receive_json()

    assert mock_process_wrong_login.called
    assert msg["type"] == TYPE_AUTH_INVALID
    assert msg["message"] == "Invalid access token or password"


async def test_auth_events_incorrect_pass(
    no_auth_websocket_client, track_connected
) -> None:
    """Test authenticating."""

    await test_auth_via_msg_incorrect_pass(no_auth_websocket_client)

    assert not track_connected["connected"]
    assert not track_connected["disconnected"]

    await no_auth_websocket_client.close()

    assert not track_connected["connected"]
    assert not track_connected["disconnected"]


async def test_pre_auth_only_auth_allowed(no_auth_websocket_client) -> None:
    """Verify that before authentication, only auth messages are allowed."""
    await no_auth_websocket_client.send_json(
        {
            "type": "call_service",
            "domain": "domain_test",
            "service": "test_service",
            "service_data": {"hello": "world"},
        }
    )

    msg = await no_auth_websocket_client.receive_json()

    assert msg["type"] == TYPE_AUTH_INVALID
    assert msg["message"].startswith("Auth message incorrectly formatted")


async def test_auth_active_with_token(
    hass: HomeAssistant, no_auth_websocket_client, hass_access_token: str
) -> None:
    """Test authenticating with a token."""
    await no_auth_websocket_client.send_json(
        {"type": TYPE_AUTH, "access_token": hass_access_token}
    )
    auth_msg = await no_auth_websocket_client.receive_json()

    assert auth_msg["type"] == TYPE_AUTH_OK


async def test_auth_active_user_inactive(
    hass: HomeAssistant,
    hass_client_no_auth: ClientSessionGenerator,
    hass_access_token: str,
) -> None:
    """Test authenticating with a token."""
    refresh_token = hass.auth.async_validate_access_token(hass_access_token)
    refresh_token.user.is_active = False
    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    client = await hass_client_no_auth()

    async with client.ws_connect(URL) as ws:
        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_REQUIRED

        await ws.send_json({"type": TYPE_AUTH, "access_token": hass_access_token})

        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_INVALID


async def test_auth_local_only_user_rejected_remote(
    hass: HomeAssistant,
    hass_client_no_auth: ClientSessionGenerator,
    hass_access_token: str,
) -> None:
    """Test that a local-only user cannot authenticate from a remote IP."""
    refresh_token = hass.auth.async_validate_access_token(hass_access_token)
    refresh_token.user.local_only = True

    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    set_mock_ip = mock_real_ip(hass.http.app)
    set_mock_ip("198.51.100.1")

    client = await hass_client_no_auth()

    with patch(
        "homeassistant.components.websocket_api.auth.process_wrong_login",
    ) as mock_process_wrong_login:
        async with client.ws_connect(URL) as ws:
            auth_msg = await ws.receive_json()
            assert auth_msg["type"] == TYPE_AUTH_REQUIRED

            await ws.send_json({"type": TYPE_AUTH, "access_token": hass_access_token})

            auth_msg = await ws.receive_json()
            assert auth_msg["type"] == TYPE_AUTH_INVALID
            assert auth_msg["message"] == "User cannot authenticate remotely"

    assert mock_process_wrong_login.called


async def test_auth_local_only_user_allowed_local(
    hass: HomeAssistant,
    hass_client_no_auth: ClientSessionGenerator,
    hass_access_token: str,
) -> None:
    """Test that a local-only user can authenticate from a local IP."""
    refresh_token = hass.auth.async_validate_access_token(hass_access_token)
    refresh_token.user.local_only = True

    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    set_mock_ip = mock_real_ip(hass.http.app)
    set_mock_ip("192.168.1.100")

    client = await hass_client_no_auth()

    async with client.ws_connect(URL) as ws:
        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_REQUIRED

        await ws.send_json({"type": TYPE_AUTH, "access_token": hass_access_token})

        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_OK


async def test_auth_active_with_password_not_allow(
    hass: HomeAssistant, hass_client_no_auth: ClientSessionGenerator
) -> None:
    """Test authenticating with a token."""
    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    client = await hass_client_no_auth()

    async with client.ws_connect(URL) as ws:
        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_REQUIRED

        await ws.send_json({"type": TYPE_AUTH, "api_password": "some-password"})

        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_INVALID


async def test_auth_legacy_support_with_password(
    hass: HomeAssistant,
    hass_client_no_auth: ClientSessionGenerator,
    local_auth: HassAuthProvider,
) -> None:
    """Test authenticating with a token."""
    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    client = await hass_client_no_auth()

    async with client.ws_connect(URL) as ws:
        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_REQUIRED

        await ws.send_json({"type": TYPE_AUTH, "api_password": "some-password"})

        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_INVALID


async def test_auth_with_invalid_token(
    hass: HomeAssistant, hass_client_no_auth: ClientSessionGenerator
) -> None:
    """Test authenticating with a token."""
    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    client = await hass_client_no_auth()

    async with client.ws_connect(URL) as ws:
        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_REQUIRED

        await ws.send_json({"type": TYPE_AUTH, "access_token": "incorrect"})

        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_INVALID


async def test_auth_close_after_revoke(
    hass: HomeAssistant, websocket_client, hass_access_token: str
) -> None:
    """Test that a websocket is closed after the refresh token is revoked."""
    assert not websocket_client.closed

    refresh_token = hass.auth.async_validate_access_token(hass_access_token)
    hass.auth.async_remove_refresh_token(refresh_token)

    msg = await websocket_client.receive()
    assert msg.type is aiohttp.WSMsgType.CLOSE
    assert websocket_client.closed


async def test_auth_sending_invalid_json_disconnects(
    hass: HomeAssistant, hass_client_no_auth: ClientSessionGenerator
) -> None:
    """Test sending invalid json during auth."""
    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    client = await hass_client_no_auth()

    async with client.ws_connect(URL) as ws:
        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_REQUIRED

        await ws.send_str("[--INVALID--JSON--]")

        auth_msg = await ws.receive()
        assert auth_msg.type == WSMsgType.close


async def test_auth_sending_binary_disconnects(
    hass: HomeAssistant, hass_client_no_auth: ClientSessionGenerator
) -> None:
    """Test sending bytes during auth."""
    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    client = await hass_client_no_auth()

    async with client.ws_connect(URL) as ws:
        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_REQUIRED

        await ws.send_bytes(b"[INVALID]")

        auth_msg = await ws.receive()
        assert auth_msg.type is WSMsgType.CLOSE


async def test_auth_close_disconnects(
    hass: HomeAssistant, hass_client_no_auth: ClientSessionGenerator
) -> None:
    """Test closing during auth."""
    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    client = await hass_client_no_auth()

    async with client.ws_connect(URL) as ws:
        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_REQUIRED

        await ws.close()

        auth_msg = await ws.receive()
        assert auth_msg.type is WSMsgType.CLOSED


async def test_auth_error_disconnects(
    hass: HomeAssistant,
    hass_client_no_auth: ClientSessionGenerator,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test error during auth."""
    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    client = await hass_client_no_auth()
    ws_response = web.WebSocketResponse()

    with patch(
        "homeassistant.components.websocket_api.http.web.WebSocketResponse",
        return_value=ws_response,
    ):
        async with client.ws_connect(URL) as ws:
            auth_msg = await ws.receive_json()
            assert auth_msg["type"] == TYPE_AUTH_REQUIRED

            ws_response._reader.feed_data(
                aiohttp.WSMessage(
                    type=WSMsgType.ERROR, data=Exception("explode"), extra=None
                ),
                0,
            )

            auth_msg = await ws.receive()
            assert auth_msg.type is WSMsgType.CLOSE

    assert "Received error message during auth phase: explode" in caplog.text


async def test_auth_sending_unknown_type_disconnects(
    hass: HomeAssistant, hass_client_no_auth: ClientSessionGenerator
) -> None:
    """Test sending unknown type during auth."""
    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    client = await hass_client_no_auth()

    async with client.ws_connect(URL) as ws:
        auth_msg = await ws.receive_json()
        assert auth_msg["type"] == TYPE_AUTH_REQUIRED

        await ws._writer.send_frame(b"1" * 130, 0x30)
        auth_msg = await ws.receive()
        assert auth_msg.type == WSMsgType.close


async def test_error_right_after_auth_disconnects(
    hass: HomeAssistant,
    hass_client_no_auth: ClientSessionGenerator,
    hass_access_token: str,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test error right after auth."""
    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    client = await hass_client_no_auth()
    ws_response = web.WebSocketResponse()

    with patch(
        "homeassistant.components.websocket_api.http.web.WebSocketResponse",
        return_value=ws_response,
    ):
        async with client.ws_connect(URL) as ws:
            auth_msg = await ws.receive_json()
            assert auth_msg["type"] == TYPE_AUTH_REQUIRED

            await ws.send_json({"type": TYPE_AUTH, "access_token": hass_access_token})
            auth_msg = await ws.receive_json()
            assert auth_msg["type"] == TYPE_AUTH_OK

            ws_response._reader.feed_data(
                aiohttp.WSMessage(
                    type=WSMsgType.ERROR, data=Exception("explode"), extra=None
                ),
                0,
            )

            close_error_msg = await ws.receive()
            assert close_error_msg.type is WSMsgType.CLOSE

    assert "Received error message during command phase: explode" in caplog.text


async def test_unix_socket_auth_bypass(
    hass: HomeAssistant, hass_client_no_auth: ClientSessionGenerator
) -> None:
    """Test that Unix socket connections skip websocket auth phase."""
    # Create the Supervisor system user
    await hass.auth.async_create_system_user(
        HASSIO_USER_NAME, group_ids=["system-admin"]
    )

    assert await async_setup_component(hass, DOMAIN, {})
    await hass.async_block_till_done()

    client = await hass_client_no_auth()

    with (
        patch(
            "homeassistant.components.http.ban.is_supervisor_unix_socket_request",
            return_value=True,
        ),
        patch(
            "homeassistant.components.http.auth.is_supervisor_unix_socket_request",
            return_value=True,
        ),
        patch(
            "homeassistant.components.websocket_api.http.is_supervisor_unix_socket_request",
            return_value=True,
        ),
    ):
        async with client.ws_connect(URL) as ws:
            # Should immediately receive auth_ok without sending a token
            auth_msg = await ws.receive_json()
            assert auth_msg["type"] == TYPE_AUTH_OK

            # Verify the connection works by sending a ping
            await ws.send_json({"id": 1, "type": "ping"})
            pong_msg = await ws.receive_json()
            assert pong_msg["type"] == "pong"
            assert pong_msg["id"] == 1
