vps-manager/tests/test_security.py

"""API Key / Agent Key 鉴权边界测试

锁死两条不可回退的行为：

1. 可信来源判定：tailnet（100.64.0.0/10）与本机回环放行；仅当请求确实来自回环时
   才信任 X-Forwarded-For（tailscale serve 转发场景），防止外部直连伪造 XFF 绕过。
2. 未配置密钥时对外部来源必须**拒绝**（fail-closed）。Agent 上报此前是「未配 key
   就全放行」，而服务默认监听 0.0.0.0，公网上任何人都能伪造监控数据上报。

运行：.venv/bin/pytest tests/test_security.py -v
"""

import asyncio

import pytest
from fastapi import HTTPException
from starlette.requests import Request

from app.core.config import settings
from app.core.security import require_agent_key, require_api_key

TAILNET_IP = "100.89.0.5"
PUBLIC_IP = "203.0.113.7"


def _request(client_ip: str = PUBLIC_IP, headers: dict = None) -> Request:
    """构造最小 Request：只需 client 与 headers 两个属性"""
    scope = {
        "type": "http",
        "method": "POST",
        "path": "/api/agent/report",
        "client": (client_ip, 40000) if client_ip else None,
        "headers": [
            (k.lower().encode(), v.encode()) for k, v in (headers or {}).items()
        ],
    }
    return Request(scope)


def _run(coro):
    return asyncio.run(coro)


def _assert_401(coro):
    with pytest.raises(HTTPException) as exc:
        _run(coro)
    assert exc.value.status_code == 401
    return exc.value


# ----------------------------- 可信来源放行 ----------------------------- #
def test_tailnet_source_allowed_without_key(monkeypatch):
    """走 tailnet IP 上报的 Agent 不受密钥缺失影响"""
    monkeypatch.setattr(settings, "AGENT_KEY", "")
    _run(require_agent_key(_request(TAILNET_IP), None))  # 不抛异常即通过


def test_loopback_source_allowed_without_key(monkeypatch):
    monkeypatch.setattr(settings, "API_KEY", "")
    _run(require_api_key(_request("127.0.0.1"), None))


def test_loopback_xff_tailnet_allowed(monkeypatch):
    """tailscale serve 转发：peer 是回环、真实来源在 XFF 里且属于 tailnet"""
    monkeypatch.setattr(settings, "API_KEY", "")
    req = _request("127.0.0.1", {"X-Forwarded-For": TAILNET_IP})
    _run(require_api_key(req, None))


def test_xff_from_public_source_is_ignored(monkeypatch):
    """外部直连时伪造 XFF 不能绕过鉴权（只有回环来源的 XFF 才可信）"""
    monkeypatch.setattr(settings, "API_KEY", "secret")
    req = _request(PUBLIC_IP, {"X-Forwarded-For": TAILNET_IP})
    exc = _assert_401(require_api_key(req, None))
    assert "API Key" in exc.detail


def test_loopback_xff_public_still_requires_key(monkeypatch):
    """回环转发但真实来源是公网 IP（funnel 之外的代理场景）→ 仍需 key"""
    monkeypatch.setattr(settings, "API_KEY", "")
    req = _request("127.0.0.1", {"X-Forwarded-For": PUBLIC_IP})
    _assert_401(require_api_key(req, None))


def test_funnel_request_forces_key_even_from_tailnet_ip(monkeypatch):
    """funnel 公网流量强制校验，即使来源 IP 恰在 tailnet 网段（双保险）"""
    monkeypatch.setattr(settings, "AGENT_KEY", "")
    req = _request(TAILNET_IP, {"Tailscale-Funnel-Request": "?1"})
    _assert_401(require_agent_key(req, None))


# ----------------------------- 外部来源 fail-closed ----------------------------- #
def test_agent_key_unset_rejects_external(monkeypatch):
    """未配 AGENT_KEY 时，外部来源必须 401（此前是全放行）"""
    monkeypatch.setattr(settings, "AGENT_KEY", "")
    exc = _assert_401(require_agent_key(_request(PUBLIC_IP), None))
    assert "AGENT_KEY" in exc.detail


def test_api_key_unset_rejects_external(monkeypatch):
    monkeypatch.setattr(settings, "API_KEY", "")
    _assert_401(require_api_key(_request(PUBLIC_IP), None))


def test_agent_key_valid_and_invalid(monkeypatch):
    monkeypatch.setattr(settings, "AGENT_KEY", "sekret")
    _run(require_agent_key(_request(PUBLIC_IP), "sekret"))  # 正确 key 放行
    _assert_401(require_agent_key(_request(PUBLIC_IP), "wrong"))
    _assert_401(require_agent_key(_request(PUBLIC_IP), None))


def test_api_key_valid_and_invalid(monkeypatch):
    monkeypatch.setattr(settings, "API_KEY", "sekret")
    _run(require_api_key(_request(PUBLIC_IP), "sekret"))
    _assert_401(require_api_key(_request(PUBLIC_IP), "wrong"))


def test_unknown_client_is_not_trusted(monkeypatch):
    """拿不到 client（如某些代理场景）时按不可信处理，不能默认放行"""
    monkeypatch.setattr(settings, "AGENT_KEY", "")
    _assert_401(require_agent_key(_request(None), None))