FluxIP/tests/test_auth_integrations.py

from __future__ import annotations

import json
import logging

import pytest
from fastapi.testclient import TestClient
from pydantic import SecretStr

from app.auth.service import LoginRateLimiter
from app.core.errors import AppError
from app.core.logging import JsonFormatter, configure_logging
from app.main import create_app
from tests.conftest import csrf_headers


def test_first_setup_login_and_csrf_flow(client: TestClient) -> None:
    bootstrap = client.get("/api/v1/auth/bootstrap")
    assert bootstrap.status_code == 200
    assert bootstrap.json()["data"] == {
        "requires_setup": True,
        "bootstrap_token_required": False,
        "authenticated": False,
        "username": None,
    }
    assert client.get("/api/v1/auth/me").status_code == 401

    setup = client.post(
        "/api/v1/auth/setup",
        json={"username": " admin ", "password": "correct-horse-battery-staple"},
    )
    assert setup.status_code == 201
    assert setup.json()["data"] == {"username": "admin"}
    set_cookie = setup.headers.get_list("set-cookie")
    assert any("fluxip_session=" in value and "HttpOnly" in value for value in set_cookie)
    assert any("fluxip_csrf=" in value and "HttpOnly" not in value for value in set_cookie)
    assert all("SameSite=strict" in value for value in set_cookie)

    authenticated = client.get("/api/v1/auth/bootstrap")
    assert authenticated.json()["data"] == {
        "requires_setup": False,
        "bootstrap_token_required": False,
        "authenticated": True,
        "username": "admin",
    }
    assert client.get("/api/v1/auth/me").json()["data"] == {"username": "admin"}

    duplicate = client.post(
        "/api/v1/auth/setup",
        json={"username": "other", "password": "another-secure-password"},
    )
    assert duplicate.status_code == 409
    assert duplicate.json()["type"] == "about:blank#setup_already_completed"

    missing_csrf = client.post("/api/v1/auth/logout")
    assert missing_csrf.status_code == 403
    assert missing_csrf.json()["type"] == "about:blank#csrf_validation_failed"
    assert (
        client.post("/api/v1/auth/logout", headers={"X-CSRF-Token": "incorrect-token"}).status_code
        == 403
    )

    logout = client.post("/api/v1/auth/logout", headers=csrf_headers(client))
    assert logout.status_code == 204
    assert client.get("/api/v1/auth/me").status_code == 401

    failed_login = client.post(
        "/api/v1/auth/login",
        json={"username": "admin", "password": "wrong-password"},
    )
    assert failed_login.status_code == 401
    assert failed_login.json()["type"] == "about:blank#authentication_required"

    login = client.post(
        "/api/v1/auth/login",
        json={"username": " admin ", "password": "correct-horse-battery-staple"},
    )
    assert login.status_code == 200
    assert client.get("/api/v1/auth/me").json()["data"]["username"] == "admin"


def test_bootstrap_token_is_required_for_remote_setup(app_settings) -> None:
    app_settings.bootstrap_token = SecretStr("one-time-bootstrap-token")
    application = create_app(app_settings)
    with TestClient(
        application,
        base_url="http://testserver",
        client=("203.0.113.10", 50000),
    ) as client:
        state = client.get("/api/v1/auth/bootstrap").json()["data"]
        assert state["requires_setup"] is True
        assert state["bootstrap_token_required"] is True

        missing = client.post(
            "/api/v1/auth/setup",
            json={"username": "admin", "password": "correct-horse-battery-staple"},
        )
        assert missing.status_code == 403
        assert missing.json()["type"] == "about:blank#bootstrap_token_invalid"

        wrong = client.post(
            "/api/v1/auth/setup",
            json={
                "username": "admin",
                "password": "correct-horse-battery-staple",
                "bootstrap_token": "错误令牌",
            },
        )
        assert wrong.status_code == 403

        setup = client.post(
            "/api/v1/auth/setup",
            json={
                "username": "admin",
                "password": "correct-horse-battery-staple",
                "bootstrap_token": "one-time-bootstrap-token",
            },
        )
        assert setup.status_code == 201


def test_setup_without_token_is_loopback_only_and_ignores_untrusted_forwarding(
    app_settings,
) -> None:
    application = create_app(app_settings)
    with TestClient(
        application,
        base_url="http://testserver",
        client=("203.0.113.10", 50000),
    ) as client:
        response = client.post(
            "/api/v1/auth/setup",
            headers={"X-Forwarded-For": "127.0.0.1"},
            json={"username": "admin", "password": "correct-horse-battery-staple"},
        )
        assert response.status_code == 403
        assert response.json()["type"] == "about:blank#bootstrap_local_only"


def test_setup_uses_forwarded_client_only_from_trusted_proxy(app_settings) -> None:
    application = create_app(app_settings)
    with TestClient(
        application,
        base_url="http://testserver",
        client=("127.0.0.1", 50000),
    ) as client:
        response = client.post(
            "/api/v1/auth/setup",
            headers={"X-Forwarded-For": "203.0.113.25"},
            json={"username": "admin", "password": "correct-horse-battery-staple"},
        )
        assert response.status_code == 403
        assert response.json()["type"] == "about:blank#bootstrap_local_only"


def test_secure_cookie_honors_https_and_trusted_proxy_boundary(app_settings) -> None:
    app_settings.bootstrap_token = SecretStr("one-time-bootstrap-token")
    application = create_app(app_settings)
    with TestClient(
        application,
        base_url="http://testserver",
        client=("203.0.113.10", 50000),
    ) as client:
        response = client.post(
            "/api/v1/auth/setup",
            headers={"X-Forwarded-Proto": "https"},
            json={
                "username": "admin",
                "password": "correct-horse-battery-staple",
                "bootstrap_token": "one-time-bootstrap-token",
            },
        )
        assert response.status_code == 201
        assert all("Secure" not in value for value in response.headers.get_list("set-cookie"))

    proxy_settings = app_settings.model_copy(
        update={
            "database_path": app_settings.database_path.with_name("proxy.db"),
            "master_key_file": app_settings.master_key_file.with_name("proxy.key"),
        }
    )
    application = create_app(proxy_settings)
    with TestClient(
        application,
        base_url="http://testserver",
        client=("127.0.0.1", 50000),
    ) as client:
        response = client.post(
            "/api/v1/auth/setup",
            headers={"X-Forwarded-Proto": "https", "X-Forwarded-For": "203.0.113.10"},
            json={
                "username": "admin",
                "password": "correct-horse-battery-staple",
                "bootstrap_token": "one-time-bootstrap-token",
            },
        )
        assert response.status_code == 201
        assert all("Secure" in value for value in response.headers.get_list("set-cookie"))
        assert response.headers["strict-transport-security"].startswith("max-age=")


def test_https_origin_forces_secure_cookie(app_settings) -> None:
    app_settings.bootstrap_token = SecretStr("one-time-bootstrap-token")
    app_settings.allowed_origins = "http://testserver,https://testserver"
    application = create_app(app_settings)
    with TestClient(
        application,
        base_url="http://testserver",
        client=("203.0.113.10", 50000),
    ) as client:
        response = client.post(
            "/api/v1/auth/setup",
            headers={"Origin": "https://testserver"},
            json={
                "username": "admin",
                "password": "correct-horse-battery-staple",
                "bootstrap_token": "one-time-bootstrap-token",
            },
        )
        assert response.status_code == 201
        assert all("Secure" in value for value in response.headers.get_list("set-cookie"))


def test_login_rate_limiter_is_bounded_and_cleans_expired_keys() -> None:
    now = [0.0]
    limiter = LoginRateLimiter(
        attempts=1,
        window_seconds=10,
        max_keys=2,
        cleanup_interval_seconds=1,
        clock=lambda: now[0],
    )

    limiter.check("read-only-check-does-not-allocate")
    assert not limiter._events
    limiter.record_failure("first")
    limiter.record_failure("second")
    limiter.record_failure("third")
    assert list(limiter._events) == ["second", "third"]
    limiter.check("first")
    with pytest.raises(AppError) as raised:
        limiter.check("second")
    assert raised.value.code == "LOGIN_RATE_LIMITED"

    now[0] = 10.0
    limiter.check("second")
    assert not limiter._events


def test_json_logging_redacts_nested_credentials_and_dependency_logs() -> None:
    class OpaqueExtra:
        def __str__(self) -> str:
            return "OpaqueExtra(token=object-secret)"

    formatter = JsonFormatter()
    record = logging.LogRecord(
        name="test",
        level=logging.INFO,
        pathname=__file__,
        lineno=1,
        msg=(
            "Authorization: Bearer bearer-value token=query-value "
            "url=https://example.test/?api_key=url-value AKIAABCDEFGHIJKLMNOP"
        ),
        args=(),
        exc_info=None,
    )
    record.api_token = "structured-token"
    record.context = {
        "password": "nested-password",
        "master-key": "master-key-value",
        "safe": "visible-value",
        "items": [{"aws_access_key_id": "AKIAABCDEFGHIJKLMNOP"}],
    }
    record.opaque = OpaqueExtra()

    serialized = formatter.format(record)
    payload = json.loads(serialized)
    for secret in (
        "bearer-value",
        "query-value",
        "url-value",
        "AKIAABCDEFGHIJKLMNOP",
        "structured-token",
        "nested-password",
        "master-key-value",
        "object-secret",
    ):
        assert secret not in serialized
    assert payload["api_token"] == "[REDACTED]"
    assert payload["context"]["safe"] == "visible-value"

    try:
        configure_logging("DEBUG")
        for logger_name in ("botocore", "boto3", "httpx", "httpcore"):
            assert logging.getLogger(logger_name).level == logging.WARNING
    finally:
        configure_logging("WARNING")


def test_integration_secrets_are_encrypted_and_masked(
    authenticated_client: TestClient,
    integration_payload: dict[str, object],
) -> None:
    client = authenticated_client
    response = client.put(
        "/api/v1/integrations",
        json=integration_payload,
        headers=csrf_headers(client),
    )
    assert response.status_code == 200
    view = response.json()["data"]
    assert view["config_version"] == 2
    assert view["secrets_configured"] == {
        "aws_access_key_id": True,
        "aws_secret_access_key": True,
        "aws_session_token": True,
        "cloudflare_api_token": True,
    }
    serialized_response = response.text
    for secret in (
        "AKIA_TEST_ACCESS_KEY",
        "aws-secret-value",
        "aws-session-value",
        "cloudflare-secret-value",
    ):
        assert secret not in serialized_response

    stored_view = client.get("/api/v1/integrations")
    assert stored_view.status_code == 200
    assert stored_view.json()["data"] == view

    container = client.app.state.container
    with container.database.connect() as connection:
        rows = connection.execute(
            "SELECT name, nonce, ciphertext FROM secrets ORDER BY name"
        ).fetchall()
    assert {row["name"] for row in rows} == {
        "aws_access_key_id",
        "aws_secret_access_key",
        "aws_session_token",
        "cloudflare_api_token",
    }
    stored_blob = " ".join(f"{row['nonce']} {row['ciphertext']}" for row in rows)
    assert all(
        secret not in stored_blob
        for secret in (
            "AKIA_TEST_ACCESS_KEY",
            "aws-secret-value",
            "aws-session-value",
            "cloudflare-secret-value",
        )
    )
    assert container.integration_repository.get_secrets() == {
        "aws_access_key_id": "AKIA_TEST_ACCESS_KEY",
        "aws_secret_access_key": "aws-secret-value",
        "aws_session_token": "aws-session-value",
        "cloudflare_api_token": "cloudflare-secret-value",
    }

    preserve_payload = {
        **integration_payload,
        "aws_access_key_id": None,
        "aws_secret_access_key": None,
        "aws_session_token": None,
        "cloudflare_api_token": None,
    }
    preserved = client.put(
        "/api/v1/integrations",
        json=preserve_payload,
        headers=csrf_headers(client),
    )
    assert preserved.status_code == 200
    assert preserved.json()["data"]["config_version"] == 3
    assert all(preserved.json()["data"]["secrets_configured"].values())
    assert container.integration_repository.get_secret("cloudflare_api_token") == (
        "cloudflare-secret-value"
    )