diff --git a/tests/test_api_assembly.py b/tests/test_api_assembly.py new file mode 100644 index 0000000..55600f9 --- /dev/null +++ b/tests/test_api_assembly.py @@ -0,0 +1,140 @@ +"""装配层测试：TestClient 挂载 app.main:app，验证路由装配与 API_KEY/AGENT_KEY fail-closed + +此前测试从不构造 TestClient（0 引用），路由装配、依赖注入与「密钥未配置时对外 +一律拒绝」的 fail-closed 语义全靠人工验证。本文件用 fastapi.testclient.TestClient +挂载真实 app，覆盖： + +- 基础装配：/health、读路由 /api/assets 正常挂载且走临时库 +- API_KEY fail-closed：外部来源在密钥未配置 / 错误时 401，正确时放行（201） +- AGENT_KEY fail-closed：外部来源在密钥未配置 / 错误时 401，正确时通过鉴权进入业务校验 + +安全前提（app/core/security.py）：TestClient 显式使用 RFC 5737 文档段公网 IP +(203.0.113.7) 作为来源，不属于可信内网（100.64/10、127/8、::1），因此走「外部来源」 +分支——密钥未配置即拒绝，正是 fail-closed 的核心属性。 + +全程免凭证、免网络：不触发 lifespan（不 init_db、不启清理协程），get_session / +get_metrics_session 依赖被覆盖到 tmp_path 临时库，agent 路由的 assets_engine 一并指过去。 + +运行：.venv/bin/pytest tests/test_api_assembly.py -v +""" + +import pytest +from fastapi.testclient import TestClient +from sqlmodel import Session, SQLModel, create_engine + +import app.models.asset # noqa: F401 确保所有表注册进 SQLModel.metadata +import app.models.credential # noqa: F401 +import app.models.monitor # noqa: F401 +import app.models.provider # noqa: F401 +import app.models.ssl # noqa: F401 +import app.routers.agent as agent_router +from app.core.config import settings +from app.database import get_metrics_session, get_session +from app.main import app + +# 外部来源（RFC 5737 TEST-NET-3 文档段，绝不属于可信内网） +EXTERNAL_CLIENT = ("203.0.113.7", 50000) +VPS_BODY = { + "name": "ext-vps", + "asset_type": "vps", + "provider": "vultr", + "vps_detail": {"ip_address": "1.2.3.4"}, +} + + +@pytest.fixture() +def client(tmp_path, monkeypatch): + """挂载真实 app 的 TestClient：依赖覆盖到临时库，密钥默认留空（fail-closed 起点）""" + assets_eng = create_engine( + f"sqlite:///{tmp_path / 'assets.db'}", connect_args={"check_same_thread": False} + ) + metrics_eng = create_engine( + f"sqlite:///{tmp_path / 'metrics.db'}", connect_args={"check_same_thread": False} + ) + SQLModel.metadata.create_all(assets_eng) + SQLModel.metadata.create_all(metrics_eng) + + def _assets_session(): + with Session(assets_eng, expire_on_commit=False) as s: + yield s + + def _metrics_session(): + with Session(metrics_eng, expire_on_commit=False) as s: + yield s + + app.dependency_overrides[get_session] = _assets_session + app.dependency_overrides[get_metrics_session] = _metrics_session + # agent.report 的 _validate_asset_id 直接用模块级 assets_engine，指到临时库避免碰本地 data/ + monkeypatch.setattr(agent_router, "assets_engine", assets_eng) + # 每次从「无密钥」的干净状态开始，fail-closed 才是被验证的默认行为 + monkeypatch.setattr(settings, "API_KEY", "") + monkeypatch.setattr(settings, "AGENT_KEY", "") + + yield TestClient(app, client=EXTERNAL_CLIENT) + + app.dependency_overrides.clear() + assets_eng.dispose() + metrics_eng.dispose() + + +# ----------------------------- 基础装配 ----------------------------- # +def test_health_route_assembles(client): + resp = client.get("/health") + assert resp.status_code == 200 + assert resp.json()["status"] == "ok" + + +def test_list_assets_route_is_open_and_empty(client): + """读路由不受 API Key 保护，装配正确时返回空列表（临时库无数据）""" + resp = client.get("/api/assets") + assert resp.status_code == 200 + assert resp.json() == [] + + +# ----------------------------- API_KEY fail-closed ----------------------------- # +def test_post_assets_fail_closed_when_api_key_unset(client): + """外部来源 + 未配置 API_KEY：写操作必须 401，绝不裸奔""" + resp = client.post("/api/assets", json=VPS_BODY) + assert resp.status_code == 401 + + +def test_post_assets_rejects_wrong_api_key(client, monkeypatch): + monkeypatch.setattr(settings, "API_KEY", "s3cret") + resp = client.post("/api/assets", json=VPS_BODY, headers={"X-API-Key": "wrong"}) + assert resp.status_code == 401 + + +def test_post_assets_accepts_correct_api_key(client, monkeypatch): + """正确密钥放行：完整走 router -> service -> 临时库，并可在列表读回""" + monkeypatch.setattr(settings, "API_KEY", "s3cret") + resp = client.post("/api/assets", json=VPS_BODY, headers={"X-API-Key": "s3cret"}) + assert resp.status_code == 201 + assert resp.json()["vps_detail"]["ip_address"] == "1.2.3.4" + + listing = client.get("/api/assets").json() + assert len(listing) == 1 + assert listing[0]["name"] == "ext-vps" + + +# ----------------------------- AGENT_KEY fail-closed ----------------------------- # +def test_agent_report_fail_closed_when_agent_key_unset(client): + """外部来源 + 未配置 AGENT_KEY：上报必须 401（防止公网伪造监控数据）""" + resp = client.post("/api/agent/report", json={"asset_id": 1}) + assert resp.status_code == 401 + + +def test_agent_report_rejects_wrong_agent_key(client, monkeypatch): + monkeypatch.setattr(settings, "AGENT_KEY", "akey") + resp = client.post( + "/api/agent/report", json={"asset_id": 1}, headers={"X-Agent-Key": "bad"} + ) + assert resp.status_code == 401 + + +def test_agent_report_passes_auth_with_correct_key(client, monkeypatch): + """正确密钥通过鉴权后进入业务校验：asset_id 不存在 -> 404（而非 401），证明闸门已放行""" + monkeypatch.setattr(settings, "AGENT_KEY", "akey") + resp = client.post( + "/api/agent/report", json={"asset_id": 999}, headers={"X-Agent-Key": "akey"} + ) + assert resp.status_code == 404 diff --git a/tests/test_asset_service.py b/tests/test_asset_service.py new file mode 100644 index 0000000..54067c4 --- /dev/null +++ b/tests/test_asset_service.py @@ -0,0 +1,311 @@ +"""asset_service 的聚焦测试：CRUD + 列表 + 到期查询（含失败/边界分支） + +此前 app/services/asset_service.py 只被间接覆盖（其它测试单独 import 了 +export_assets/import_assets/get_overview），核心 CRUD 面 0 直接引用。本文件 +以模块名 `asset_service` 直接调用，补齐关键行为与失败/边界分支： + +- create/get/update/delete 正常路径 + 失败分支（缺详情 400 / 账号悬空 400 / 不存在 404） +- delete 的联动清理（详情 + subdomains/site_certs 子表），且 metrics 清理失败可容忍 +- VPS 凭证在库内加密存储、输出只给 has_* 标记、更新时凭证为 None 保留原值 +- list_assets 的类型/状态/归档筛选、名称与平台搜索、provider slug 与 provider_id、 + 非法排序字段回退 expiry_date +- get_expiring 的到期口径（含已过期、排除已注销、按剩余天数升序） + +全程 tmp_path + create_engine 临时库，免凭证、免网络：VPS 凭证加密用一个临时生成的 +Fernet key（不读 .env 的真实 MASTER_KEY）。 + +运行：.venv/bin/pytest tests/test_asset_service.py -v +""" + +from datetime import date, timedelta + +import pytest +from cryptography.fernet import Fernet +from fastapi import HTTPException +from sqlmodel import Session, SQLModel, create_engine, select + +import app.models.asset # noqa: F401 确保所有表注册进 SQLModel.metadata +import app.models.credential # noqa: F401 +import app.models.monitor # noqa: F401 +import app.models.provider # noqa: F401 +import app.models.ssl # noqa: F401 +from app.core import crypto +from app.core.config import settings +from app.models.asset import ( + Account, + Asset, + AssetStatus, + AssetType, + DomainDetail, + VPSDetail, +) +from app.models.provider import Provider +from app.models.ssl import SiteCert, Subdomain +from app.schemas.asset import ( + AssetCreate, + AssetUpdate, + DomainDetailIn, + VPSDetailIn, +) +from app.services import asset_service + + +@pytest.fixture() +def engine(tmp_path, monkeypatch): + """临时资产库；metrics_engine 也指过来，让 delete 的监控清理落在临时库而非本地 data/""" + eng = create_engine( + f"sqlite:///{tmp_path / 'assets.db'}", connect_args={"check_same_thread": False} + ) + SQLModel.metadata.create_all(eng) + monkeypatch.setattr("app.database.metrics_engine", eng) + yield eng + eng.dispose() + + +@pytest.fixture() +def master_key(monkeypatch): + """临时 Fernet key：让凭证加密路径可测，但不依赖 .env 里的真实 MASTER_KEY""" + monkeypatch.setattr(settings, "MASTER_KEY", Fernet.generate_key().decode()) + crypto._get_fernet.cache_clear() # 清掉可能已用旧 key 构建的缓存实例 + yield + crypto._get_fernet.cache_clear() + + +def _vps_create(name="web-01", ip="1.2.3.4", **kw): + detail = kw.pop("vps_detail", VPSDetailIn(ip_address=ip)) + return AssetCreate( + name=name, asset_type=AssetType.VPS, provider="vultr", vps_detail=detail, **kw + ) + + +# ----------------------------- create / get ----------------------------- # +def test_create_vps_returns_read_with_detail(engine): + with Session(engine) as s: + read = asset_service.create_asset(s, _vps_create()) + assert read.id is not None + assert read.name == "web-01" + assert read.vps_detail is not None + assert read.vps_detail.ip_address == "1.2.3.4" + # 未提供凭证时不应误报存在 + assert read.vps_detail.has_ssh_key is False + assert read.vps_detail.has_password is False + + +def test_create_vps_encrypts_credentials_at_rest(engine, master_key): + """明文凭证必须加密入库，输出只暴露 has_* 标记，绝不回传明文""" + detail = VPSDetailIn(ip_address="1.2.3.4", ssh_key="TOPSECRET", password="PW") + with Session(engine) as s: + read = asset_service.create_asset(s, _vps_create(vps_detail=detail)) + stored = s.exec(select(VPSDetail).where(VPSDetail.asset_id == read.id)).one() + assert stored.ssh_key_encrypted != "TOPSECRET" # 不是明文 + assert crypto.decrypt(stored.ssh_key_encrypted) == "TOPSECRET" # 可还原 + assert read.vps_detail.has_ssh_key is True + assert "ssh_key" not in read.vps_detail.model_dump() # 输出模型无明文字段 + + +def test_create_asset_missing_detail_raises_400(engine): + """asset_type=vps 却不带 vps_detail：失败分支，必须 400 而不是写入残缺资产""" + data = AssetCreate(name="no-detail", asset_type=AssetType.VPS, provider="vultr") + with Session(engine) as s: + with pytest.raises(HTTPException) as ei: + asset_service.create_asset(s, data) + assert ei.value.status_code == 400 + + +def test_create_asset_invalid_account_raises_400(engine): + """account_id 指向不存在的账号：防外键悬空，400""" + data = _vps_create(account_id=999) + with Session(engine) as s: + with pytest.raises(HTTPException) as ei: + asset_service.create_asset(s, data) + assert ei.value.status_code == 400 + + +def test_create_asset_links_valid_account_and_resolves_name(engine): + with Session(engine) as s: + acc = Account(name="acc@example.com", platform="vultr") + s.add(acc) + s.commit() + s.refresh(acc) + read = asset_service.create_asset(s, _vps_create(account_id=acc.id)) + assert read.account_name == "acc@example.com" + + +def test_get_asset_missing_raises_404(engine): + with Session(engine) as s: + with pytest.raises(HTTPException) as ei: + asset_service.get_asset(s, 123456) + assert ei.value.status_code == 404 + + +def test_get_asset_computes_days_to_expiry(engine): + expiry = date.today() + timedelta(days=7) + with Session(engine) as s: + created = asset_service.create_asset(s, _vps_create(expiry_date=expiry)) + got = asset_service.get_asset(s, created.id) + assert got.days_to_expiry == 7 + + +# ----------------------------- update ----------------------------- # +def test_update_asset_changes_main_fields(engine): + with Session(engine) as s: + created = asset_service.create_asset(s, _vps_create()) + updated = asset_service.update_asset( + s, created.id, AssetUpdate(name="renamed", cost=12.5) + ) + assert updated.name == "renamed" + assert updated.cost == 12.5 + + +def test_update_vps_preserves_credential_when_none(engine, master_key): + """更新详情时未携带凭证（None）不能把已存凭证清空""" + detail = VPSDetailIn(ip_address="1.2.3.4", ssh_key="KEEPME") + with Session(engine) as s: + created = asset_service.create_asset(s, _vps_create(vps_detail=detail)) + # 只改 IP，不带 ssh_key + asset_service.update_asset( + s, created.id, AssetUpdate(vps_detail=VPSDetailIn(ip_address="9.9.9.9")) + ) + stored = s.exec(select(VPSDetail).where(VPSDetail.asset_id == created.id)).one() + assert stored.ip_address == "9.9.9.9" + assert crypto.decrypt(stored.ssh_key_encrypted) == "KEEPME" + + +def test_update_asset_missing_raises_404(engine): + with Session(engine) as s: + with pytest.raises(HTTPException) as ei: + asset_service.update_asset(s, 123456, AssetUpdate(name="x")) + assert ei.value.status_code == 404 + + +# ----------------------------- delete ----------------------------- # +def test_delete_asset_removes_detail_and_child_rows(engine): + """删除域名资产时，详情与 subdomains/site_certs 子表行都要清掉，避免孤儿数据""" + with Session(engine) as s: + created = asset_service.create_asset( + s, + AssetCreate( + name="example.com", + asset_type=AssetType.DOMAIN, + provider="namesilo", + domain_detail=DomainDetailIn(domain_name="example.com"), + ), + ) + aid = created.id + s.add(Subdomain(asset_id=aid, host="www")) + s.add(SiteCert(hostname="www.example.com", asset_id=aid)) + s.commit() + + asset_service.delete_asset(s, aid) + + assert s.get(Asset, aid) is None + assert s.exec(select(DomainDetail).where(DomainDetail.asset_id == aid)).first() is None + assert s.exec(select(Subdomain).where(Subdomain.asset_id == aid)).first() is None + assert s.exec(select(SiteCert).where(SiteCert.asset_id == aid)).first() is None + + +def test_delete_asset_missing_raises_404(engine): + with Session(engine) as s: + with pytest.raises(HTTPException) as ei: + asset_service.delete_asset(s, 123456) + assert ei.value.status_code == 404 + + +def test_delete_asset_tolerates_metrics_cleanup_failure(engine, monkeypatch): + """metrics.db 可能被 Agent 短暂锁定：清理失败不能让已成功的删除返回 500""" + def _boom(_asset_id): + raise RuntimeError("metrics.db is locked") + + monkeypatch.setattr(asset_service, "_cleanup_metrics_for_asset", _boom) + with Session(engine) as s: + created = asset_service.create_asset(s, _vps_create()) + asset_service.delete_asset(s, created.id) # 不应抛异常 + assert s.get(Asset, created.id) is None + + +# ----------------------------- list ----------------------------- # +def _seed(engine, name, *, asset_type=AssetType.VPS, provider="vultr", + status=AssetStatus.ACTIVE, is_archived=False, provider_id=None): + with Session(engine) as s: + a = Asset( + name=name, asset_type=asset_type, provider=provider, + status=status, is_archived=is_archived, provider_id=provider_id, + ) + s.add(a) + s.commit() + s.refresh(a) + return a.id + + +def test_list_assets_filters_by_type_status_and_archive(engine): + _seed(engine, "vps-active", asset_type=AssetType.VPS) + _seed(engine, "dom-active", asset_type=AssetType.DOMAIN, provider="namesilo") + _seed(engine, "vps-stopped", status=AssetStatus.STOPPED) + _seed(engine, "vps-archived", is_archived=True) + with Session(engine) as s: + vps = asset_service.list_assets(s, asset_type=AssetType.VPS) + stopped = asset_service.list_assets(s, asset_status=AssetStatus.STOPPED) + archived = asset_service.list_assets(s, is_archived=True) + assert {a.name for a in vps} == {"vps-active", "vps-stopped", "vps-archived"} + assert {a.name for a in stopped} == {"vps-stopped"} + assert {a.name for a in archived} == {"vps-archived"} + + +def test_list_assets_search_and_provider_slug_and_id(engine): + with Session(engine) as s: + p = Provider(slug="vultr", name="Vultr", category="vps") + s.add(p) + s.commit() + s.refresh(p) + pid = p.id + _seed(engine, "blog-vps", provider="vultr", provider_id=pid) + _seed(engine, "shop-domain", provider="namesilo", asset_type=AssetType.DOMAIN) + with Session(engine) as s: + by_name = asset_service.list_assets(s, q="blog") + by_slug = asset_service.list_assets(s, provider="vultr") + by_id = asset_service.list_assets(s, provider=str(pid)) + assert {a.name for a in by_name} == {"blog-vps"} + assert {a.name for a in by_slug} == {"blog-vps"} + assert {a.name for a in by_id} == {"blog-vps"} + # provider_name 由 provider_id 关联解析 + assert by_id[0].provider_name == "Vultr" + + +def test_list_assets_invalid_sort_falls_back_and_order_desc(engine): + """非白名单排序字段必须回退 expiry_date，不能因未知列报错""" + with Session(engine) as s: + s.add(Asset(name="near", asset_type=AssetType.VPS, provider="vultr", + expiry_date=date.today() + timedelta(days=1))) + s.add(Asset(name="far", asset_type=AssetType.VPS, provider="vultr", + expiry_date=date.today() + timedelta(days=90))) + s.commit() + asc = asset_service.list_assets(s, sort="; DROP TABLE assets", order="asc") + desc = asset_service.list_assets(s, sort="expiry_date", order="desc") + assert [a.name for a in asc] == ["near", "far"] + assert [a.name for a in desc] == ["far", "near"] + + +# ----------------------------- expiring ----------------------------- # +def test_get_expiring_includes_overdue_excludes_cancelled_sorted(engine): + today = date.today() + _seed(engine, "overdue") + _seed(engine, "soon") + _seed(engine, "far") + _seed(engine, "cancelled") + with Session(engine) as s: + for name, expiry, status in ( + ("overdue", today - timedelta(days=5), AssetStatus.ACTIVE), + ("soon", today + timedelta(days=10), AssetStatus.ACTIVE), + ("far", today + timedelta(days=100), AssetStatus.ACTIVE), + ("cancelled", today + timedelta(days=1), AssetStatus.CANCELLED), + ): + a = s.exec(select(Asset).where(Asset.name == name)).one() + a.expiry_date = expiry + a.status = status + s.add(a) + s.commit() + result = asset_service.get_expiring(s, days=30) + names = [a.name for a in result] + # 已过期(-5) 排在 30 天内(+10) 之前；far 超窗口、cancelled 已注销都不进 + assert names == ["overdue", "soon"] + assert result[0].days_to_expiry == -5