bambuddy/backend/tests/unit/test_homeassistant_tls_3300.py

"""The Home Assistant connection test names a certificate problem (#3300).

httpx raises an untrusted certificate as ``ConnectError``, so after moving HA
to https the test said "Could not connect to Home Assistant" for an HA that was
up and answering. These run against real local servers on 127.0.0.1.
"""

from __future__ import annotations

import asyncio
import ipaddress
import socket
import ssl
from datetime import datetime, timedelta, timezone
from unittest.mock import patch

import pytest
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.x509.oid import NameOID

from backend.app.services.homeassistant import HomeAssistantService

pytestmark = [pytest.mark.asyncio, pytest.mark.unit]


@pytest.fixture(autouse=True)
def _certifi_trust_store(monkeypatch):
    """httpx reads SSL_CERT_FILE / SSL_CERT_DIR before certifi (USE_SYSTEM_TRUST_STORE sets one)."""
    monkeypatch.delenv("SSL_CERT_FILE", raising=False)
    monkeypatch.delenv("SSL_CERT_DIR", raising=False)


def _cert(subject_cn: str, key, issuer_cn: str, issuer_key, san: x509.GeneralName, ca: bool, expired: bool = False):
    now = datetime.now(timezone.utc)
    return (
        x509.CertificateBuilder()
        .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, subject_cn)]))
        .issuer_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, issuer_cn)]))
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - timedelta(days=60 if expired else 1))
        .not_valid_after(now - timedelta(days=1) if expired else now + timedelta(days=30))
        .add_extension(x509.BasicConstraints(ca=ca, path_length=None), critical=True)
        .add_extension(x509.SubjectAlternativeName([san]), critical=False)
        # Python 3.13 verifies strictly, which wants key usage and both key identifiers.
        .add_extension(
            x509.KeyUsage(
                digital_signature=True,
                content_commitment=False,
                key_encipherment=False,
                data_encipherment=False,
                key_agreement=False,
                key_cert_sign=ca,
                crl_sign=ca,
                encipher_only=False,
                decipher_only=False,
            ),
            critical=True,
        )
        .add_extension(x509.SubjectKeyIdentifier.from_public_key(key.public_key()), critical=False)
        .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(issuer_key.public_key()), critical=False)
        .sign(issuer_key, hashes.SHA256())
    )


def _pem(path, cert, key=None):
    data = cert.public_bytes(serialization.Encoding.PEM)
    if key is not None:
        data += key.private_bytes(
            serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()
        )
    path.write_bytes(data)
    return str(path)


async def _https_server(certfile: str):
    """An HA stand-in that answers /api/ over TLS with the given certificate."""
    ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    ctx.load_cert_chain(certfile)

    async def handle(reader, writer):
        try:
            await reader.readuntil(b"\r\n\r\n")
            body = b'{"message": "API running."}'
            writer.write(
                b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: %d\r\n\r\n%s" % (len(body), body)
            )
            await writer.drain()
        except (ConnectionError, asyncio.IncompleteReadError, ssl.SSLError):
            pass
        finally:
            writer.close()

    return await asyncio.start_server(handle, "127.0.0.1", 0, ssl=ctx)


def _port(server) -> int:
    return server.sockets[0].getsockname()[1]


async def test_a_self_signed_certificate_is_named(tmp_path):
    key = ec.generate_private_key(ec.SECP256R1())
    cert = _cert("homeassistant", key, "homeassistant", key, x509.IPAddress(ipaddress.ip_address("127.0.0.1")), True)
    server = await _https_server(_pem(tmp_path / "ha.pem", cert, key))
    async with server:
        result = await HomeAssistantService().test_connection(f"https://127.0.0.1:{_port(server)}", "token")

    assert result["success"] is False
    assert result["error"].startswith("Home Assistant's certificate is not trusted (self-signed certificate).")
    assert "add it to Bambuddy's trust store" in result["error"]


async def test_a_certificate_for_another_hostname_is_named(tmp_path):
    """A trusted certificate reached by IP address, as with Let's Encrypt on a LAN."""
    ca_key = ec.generate_private_key(ec.SECP256R1())
    ca = _cert("Test CA", ca_key, "Test CA", ca_key, x509.DNSName("ca.test"), True)
    leaf_key = ec.generate_private_key(ec.SECP256R1())
    leaf = _cert("ha.example.org", leaf_key, "Test CA", ca_key, x509.DNSName("ha.example.org"), False)
    server = await _https_server(_pem(tmp_path / "leaf.pem", leaf, leaf_key))
    ca_file = _pem(tmp_path / "ca.pem", ca)
    async with server:
        # httpx builds its trust store from certifi.where().
        with patch("certifi.where", return_value=ca_file):
            trusted = await HomeAssistantService().test_connection(f"https://127.0.0.1:{_port(server)}", "token")

    assert trusted["success"] is False
    assert trusted["error"].startswith(
        "Home Assistant's certificate is not trusted (IP address mismatch, certificate is not valid for '127.0.0.1')."
    )
    assert trusted["error"].endswith(
        "Enter the hostname the certificate was issued for, not an IP address or another name."
    )


async def test_an_expired_certificate_is_named(tmp_path):
    ca_key = ec.generate_private_key(ec.SECP256R1())
    ca = _cert("Test CA", ca_key, "Test CA", ca_key, x509.DNSName("ca.test"), True)
    leaf_key = ec.generate_private_key(ec.SECP256R1())
    ip = x509.IPAddress(ipaddress.ip_address("127.0.0.1"))
    leaf = _cert("homeassistant", leaf_key, "Test CA", ca_key, ip, False, expired=True)
    server = await _https_server(_pem(tmp_path / "leaf.pem", leaf, leaf_key))
    ca_file = _pem(tmp_path / "ca.pem", ca)
    async with server:
        with patch("certifi.where", return_value=ca_file):
            result = await HomeAssistantService().test_connection(f"https://127.0.0.1:{_port(server)}", "token")

    assert result["error"] == (
        "Home Assistant's certificate is not trusted (certificate has expired). "
        "Renew the certificate Home Assistant serves."
    )


async def test_https_to_a_plain_http_port_is_named():
    async def handle(reader, writer):
        await reader.read(1024)
        writer.write(b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\n\r\n")
        await writer.drain()
        writer.close()

    server = await asyncio.start_server(handle, "127.0.0.1", 0)
    async with server:
        result = await HomeAssistantService().test_connection(f"https://127.0.0.1:{_port(server)}", "token")

    assert result["success"] is False
    assert result["error"].startswith("The secure connection to Home Assistant failed (")
    assert result["error"].endswith("Check that Home Assistant serves https at this address and port.")


async def test_nothing_listening_still_says_could_not_connect():
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        port = s.getsockname()[1]

    result = await HomeAssistantService().test_connection(f"https://127.0.0.1:{port}", "token")

    assert result == {"success": False, "message": None, "error": "Could not connect to Home Assistant"}