433 lines
16 KiB
Python
433 lines
16 KiB
Python
"""正式授权码（codes）模式：Web 端授权码即身份的端到端用例。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import time
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import select
|
||
|
||
from tests.auth_provider import install_test_auth
|
||
|
||
TENANT = "tenant-codes-000000000000000001"
|
||
CODES = {
|
||
"USER-CODE-ALPHA": "year",
|
||
"USER-CODE-BETA": {"type": "month", "name": "beta-user"},
|
||
"USER-CODE-HOUR": "hour",
|
||
}
|
||
|
||
|
||
@pytest.fixture()
|
||
def codes_app(tmp_path, monkeypatch):
|
||
monkeypatch.setenv("APS_LICENSE_PROVIDER", "codes")
|
||
monkeypatch.setenv("APS_LICENSE_SECRET", "codes-test-secret-0123456789abcdef")
|
||
monkeypatch.setenv("APS_LICENSE_TENANT_UUID", TENANT)
|
||
monkeypatch.setenv("APS_LICENSE_CLIENTS", "web,desktop")
|
||
monkeypatch.setenv("APS_LICENSE_CODES", json.dumps(CODES))
|
||
# 钉死认证提供方：套件中途 server.main 的 load_dotenv 可能把开发机 .env 的 jms 重新注入
|
||
monkeypatch.setenv("APS_AUTH_PROVIDER", "unconfigured")
|
||
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "codes.db"))
|
||
monkeypatch.setenv("APS_WORLD_PATH", str(tmp_path / "world.json"))
|
||
|
||
from server.auth import licenses
|
||
from server.auth import providers as auth_providers
|
||
from server.db.database import reset_engine
|
||
from server.state import store as world_store
|
||
|
||
install_test_auth(monkeypatch, TENANT)
|
||
licenses._provider = None
|
||
licenses._provider_mode = None
|
||
auth_providers._provider = None
|
||
auth_providers._provider_mode = None
|
||
world_store._stores.clear()
|
||
reset_engine()
|
||
|
||
from server.gateway.app import create_app
|
||
|
||
app = create_app()
|
||
yield app
|
||
|
||
reset_engine()
|
||
world_store._stores.clear()
|
||
licenses._provider = None
|
||
licenses._provider_mode = None
|
||
auth_providers._provider = None
|
||
auth_providers._provider_mode = None
|
||
|
||
|
||
def test_health_reports_license_auth(codes_app):
|
||
client = TestClient(codes_app)
|
||
body = client.get("/api/health").json()
|
||
assert body["ok"] is True
|
||
assert body["licenseProvider"] == "codes"
|
||
assert body["licenseAuth"] is True
|
||
|
||
|
||
def test_health_survives_missing_secret(tmp_path, monkeypatch):
|
||
monkeypatch.setenv("APS_LICENSE_PROVIDER", "codes")
|
||
monkeypatch.delenv("APS_LICENSE_SECRET", raising=False)
|
||
monkeypatch.setenv("APS_LICENSE_CLIENTS", "web")
|
||
monkeypatch.setenv("APS_AUTH_PROVIDER", "unconfigured")
|
||
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "bad.db"))
|
||
monkeypatch.setenv("APS_WORLD_PATH", str(tmp_path / "world.json"))
|
||
|
||
from server.auth import licenses
|
||
from server.auth import providers as auth_providers
|
||
from server.db.database import reset_engine
|
||
from server.state import store as world_store
|
||
|
||
install_test_auth(monkeypatch, TENANT)
|
||
licenses._provider = None
|
||
licenses._provider_mode = None
|
||
auth_providers._provider = None
|
||
auth_providers._provider_mode = None
|
||
world_store._stores.clear()
|
||
reset_engine()
|
||
|
||
from server.gateway.app import create_app
|
||
|
||
client = TestClient(create_app())
|
||
response = client.get("/api/health")
|
||
assert response.status_code == 200
|
||
assert response.json()["licenseProvider"] == "invalid:LICENSE_CONFIG_INVALID"
|
||
|
||
reset_engine()
|
||
world_store._stores.clear()
|
||
licenses._provider = None
|
||
licenses._provider_mode = None
|
||
auth_providers._provider = None
|
||
auth_providers._provider_mode = None
|
||
|
||
|
||
def test_provider_requires_secret(monkeypatch):
|
||
from server.auth.licenses import CodeLicenseProvider
|
||
from server.auth.providers import AuthError
|
||
|
||
monkeypatch.delenv("APS_LICENSE_SECRET", raising=False)
|
||
with pytest.raises(AuthError, match="授权配置缺失") as exc:
|
||
CodeLicenseProvider()
|
||
assert exc.value.code == "LICENSE_CONFIG_INVALID"
|
||
|
||
|
||
def test_web_activation_binds_identity_and_isolates_data(codes_app):
|
||
alpha = TestClient(codes_app)
|
||
beta = TestClient(codes_app)
|
||
|
||
response = alpha.post("/api/auth/license/activate", json={"code": "USER-CODE-ALPHA"})
|
||
assert response.status_code == 200
|
||
body = response.json()
|
||
assert body["user"]["auth_kind"] == "license"
|
||
assert body["user"]["tenant_uuid"] == TENANT
|
||
assert body["license"]["type"] == "year"
|
||
alpha_id = body["user"]["user_id"]
|
||
|
||
assert beta.post("/api/auth/license/activate", json={"code": "USER-CODE-BETA"}).status_code == 200
|
||
beta_me = beta.get("/api/auth/me").json()
|
||
assert beta_me["user"]["username"] == "beta-user"
|
||
assert beta_me["user"]["user_id"] != alpha_id
|
||
|
||
assert alpha.post(
|
||
"/api/projects", json={"id": "alpha_private", "name": "授权A的私有项目"},
|
||
).status_code == 200
|
||
workspace = beta.get("/api/workspace")
|
||
assert workspace.status_code == 200
|
||
assert workspace.json()["projects"] == []
|
||
assert beta.get("/api/projects/alpha_private/members").status_code == 404
|
||
|
||
# 同一授权码换浏览器激活：找回同一身份与同一份数据
|
||
alpha_second_browser = TestClient(codes_app)
|
||
assert alpha_second_browser.post(
|
||
"/api/auth/license/activate", json={"code": "USER-CODE-ALPHA"},
|
||
).status_code == 200
|
||
me_again = alpha_second_browser.get("/api/auth/me").json()
|
||
assert me_again["user"]["user_id"] == alpha_id
|
||
projects = alpha_second_browser.get("/api/workspace").json()["projects"]
|
||
assert any(project["id"] == "alpha_private" for project in projects)
|
||
|
||
|
||
def test_admin_license_code_is_permanent_with_admin_roles(codes_app):
|
||
client = TestClient(codes_app)
|
||
response = client.post("/api/auth/license/activate", json={"code": "aps@Z@Agent@Admin"})
|
||
assert response.status_code == 200
|
||
body = response.json()
|
||
assert body["user"]["fullname"] == "超级管理员"
|
||
assert "admin" in body["user"]["roles"]
|
||
assert body["license"]["type"] == "permanent"
|
||
assert body["license"]["expiresAt"] is None
|
||
assert body["license"]["permanent"] is True
|
||
|
||
|
||
def test_codes_are_case_sensitive(codes_app):
|
||
client = TestClient(codes_app)
|
||
response = client.post("/api/auth/license/activate", json={"code": "user-code-alpha"})
|
||
assert response.status_code == 401
|
||
assert response.json()["error"]["code"] == "LICENSE_CODE_INVALID"
|
||
|
||
|
||
def test_invalid_and_blank_code_rejected(codes_app):
|
||
client = TestClient(codes_app)
|
||
response = client.post("/api/auth/license/activate", json={"code": "NO-SUCH-CODE"})
|
||
assert response.status_code == 401
|
||
assert response.json()["error"]["code"] == "LICENSE_CODE_INVALID"
|
||
blank = client.post("/api/auth/license/activate", json={"code": " "})
|
||
assert blank.status_code == 400
|
||
assert blank.json()["error"]["code"] == "LICENSE_CODE_REQUIRED"
|
||
|
||
|
||
def test_unactivated_web_request_requires_license(codes_app):
|
||
client = TestClient(codes_app)
|
||
response = client.get("/api/auth/me")
|
||
assert response.status_code == 401
|
||
assert response.json()["error"]["code"] == "LICENSE_REQUIRED"
|
||
assert "授权码" in response.json()["error"]["message"]
|
||
|
||
|
||
def test_login_endpoint_redirects_to_license_under_codes(codes_app):
|
||
client = TestClient(codes_app)
|
||
response = client.post("/api/auth/login", json={
|
||
"method": "password", "username": "planner", "password": "test",
|
||
})
|
||
assert response.status_code == 400
|
||
assert response.json()["error"]["code"] == "LICENSE_ACTIVATION_REQUIRED"
|
||
|
||
|
||
def test_expired_code_is_rejected_on_next_request(codes_app):
|
||
client = TestClient(codes_app)
|
||
assert client.post(
|
||
"/api/auth/license/activate", json={"code": "USER-CODE-HOUR"},
|
||
).status_code == 200
|
||
|
||
from server.db.database import get_session
|
||
from server.db.models import LicenseActivation
|
||
|
||
with get_session() as session:
|
||
row = session.scalar(select(LicenseActivation))
|
||
assert row is not None
|
||
row.expires_at = int(time.time()) - 1
|
||
session.commit()
|
||
|
||
response = client.get("/api/auth/me")
|
||
assert response.status_code == 401
|
||
assert response.json()["error"]["code"] == "LICENSE_EXPIRED"
|
||
|
||
again = client.post("/api/auth/license/activate", json={"code": "USER-CODE-HOUR"})
|
||
assert again.status_code == 401
|
||
assert again.json()["error"]["code"] == "LICENSE_EXPIRED"
|
||
|
||
|
||
def test_desktop_client_also_activates_with_codes(codes_app):
|
||
client = TestClient(codes_app)
|
||
headers = {"X-APS-Client": "desktop", "X-APS-Device-ID": "device-codes-0001"}
|
||
response = client.post("/api/auth/license/activate", headers=headers, json={"code": "USER-CODE-ALPHA"})
|
||
assert response.status_code == 200
|
||
me = client.get("/api/auth/me", headers=headers)
|
||
assert me.status_code == 200
|
||
assert me.json()["user"]["auth_kind"] == "license"
|
||
|
||
|
||
def test_logout_clears_web_license_session(codes_app):
|
||
client = TestClient(codes_app)
|
||
assert client.post("/api/auth/license/activate", json={"code": "USER-CODE-ALPHA"}).status_code == 200
|
||
assert client.get("/api/auth/me").status_code == 200
|
||
assert client.post("/api/auth/logout").status_code == 200
|
||
assert client.get("/api/auth/me").status_code == 401
|
||
|
||
|
||
# ---------------- 超管授权码管理 API ----------------
|
||
|
||
def _admin_client(app) -> TestClient:
|
||
client = TestClient(app)
|
||
assert client.post("/api/auth/license/activate", json={"code": "aps@Z@Agent@Admin"}).status_code == 200
|
||
return client
|
||
|
||
|
||
def test_admin_issues_code_and_user_activates_with_name(codes_app):
|
||
admin = _admin_client(codes_app)
|
||
created = admin.post("/api/admin/license/codes", json={
|
||
"name": "张三", "mobile": "13800001111", "company": "锐扬", "licenseType": "month",
|
||
})
|
||
assert created.status_code == 200
|
||
issued = created.json()["code"]
|
||
assert issued["name"] == "张三"
|
||
assert len(issued["code"]) == 19 # XXXX-XXXX-XXXX-XXXX
|
||
|
||
listed = admin.get("/api/admin/license/codes").json()["codes"]
|
||
assert any(item["id"] == issued["id"] for item in listed)
|
||
|
||
user = TestClient(codes_app)
|
||
response = user.post("/api/auth/license/activate", json={"code": issued["code"]})
|
||
assert response.status_code == 200
|
||
body = response.json()
|
||
assert body["user"]["username"] == "张三"
|
||
assert body["user"]["fullname"] == "张三"
|
||
assert body["license"]["type"] == "month"
|
||
|
||
listed_after = admin.get("/api/admin/license/codes").json()["codes"]
|
||
entry = next(item for item in listed_after if item["id"] == issued["id"])
|
||
assert entry["activation"] is not None
|
||
assert entry["activation"]["status"] == "active"
|
||
|
||
|
||
def test_non_admin_cannot_manage_codes(codes_app):
|
||
user = TestClient(codes_app)
|
||
assert user.post("/api/auth/license/activate", json={"code": "USER-CODE-ALPHA"}).status_code == 200
|
||
response = user.get("/api/admin/license/codes")
|
||
assert response.status_code == 403
|
||
assert response.json()["detail"]["code"] == "LICENSE_ADMIN_REQUIRED"
|
||
denied = user.post("/api/admin/license/codes", json={"name": "越权", "licenseType": "year"})
|
||
assert denied.status_code == 403
|
||
|
||
|
||
def test_disable_code_blocks_existing_session_and_new_activation(codes_app):
|
||
admin = _admin_client(codes_app)
|
||
issued = admin.post("/api/admin/license/codes", json={
|
||
"name": "李四", "licenseType": "year",
|
||
}).json()["code"]
|
||
|
||
user = TestClient(codes_app)
|
||
assert user.post("/api/auth/license/activate", json={"code": issued["code"]}).status_code == 200
|
||
assert user.get("/api/auth/me").status_code == 200
|
||
|
||
assert admin.post(
|
||
f"/api/admin/license/codes/{issued['id']}/status", json={"status": "disabled"},
|
||
).status_code == 200
|
||
|
||
denied = user.get("/api/auth/me")
|
||
assert denied.status_code == 401
|
||
assert denied.json()["error"]["code"] == "LICENSE_REVOKED"
|
||
reactivate = user.post("/api/auth/license/activate", json={"code": issued["code"]})
|
||
assert reactivate.status_code == 401
|
||
|
||
|
||
def test_delete_code_removes_it_from_catalog(codes_app):
|
||
admin = _admin_client(codes_app)
|
||
issued = admin.post("/api/admin/license/codes", json={
|
||
"name": "王五", "licenseType": "day",
|
||
}).json()["code"]
|
||
assert admin.delete(f"/api/admin/license/codes/{issued['id']}").status_code == 200
|
||
|
||
user = TestClient(codes_app)
|
||
response = user.post("/api/auth/license/activate", json={"code": issued["code"]})
|
||
assert response.status_code == 401
|
||
assert response.json()["error"]["code"] == "LICENSE_CODE_INVALID"
|
||
|
||
|
||
def test_duplicate_and_admin_code_rejected(codes_app):
|
||
admin = _admin_client(codes_app)
|
||
first = admin.post("/api/admin/license/codes", json={
|
||
"name": "甲", "licenseType": "year", "code": "CUSTOM-CODE-1",
|
||
})
|
||
assert first.status_code == 200
|
||
dup = admin.post("/api/admin/license/codes", json={
|
||
"name": "乙", "licenseType": "year", "code": "CUSTOM-CODE-1",
|
||
})
|
||
assert dup.status_code == 409
|
||
admin_code = admin.post("/api/admin/license/codes", json={
|
||
"name": "丙", "licenseType": "year", "code": "aps@Z@Agent@Admin",
|
||
})
|
||
assert admin_code.status_code == 409
|
||
|
||
|
||
@pytest.fixture()
|
||
def mock_license_app(tmp_path, monkeypatch):
|
||
monkeypatch.setenv("APS_LICENSE_PROVIDER", "mock")
|
||
monkeypatch.setenv("APS_MOCK_LICENSE_SECRET", "license-test-secret-0123456789abcdef")
|
||
monkeypatch.setenv("APS_MOCK_TENANT_UUID", TENANT)
|
||
monkeypatch.setenv("APS_AUTH_PROVIDER", "unconfigured")
|
||
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "mock.db"))
|
||
monkeypatch.setenv("APS_WORLD_PATH", str(tmp_path / "world.json"))
|
||
|
||
from server.auth import licenses
|
||
from server.auth import providers as auth_providers
|
||
from server.db.database import reset_engine
|
||
from server.state import store as world_store
|
||
|
||
install_test_auth(monkeypatch, TENANT)
|
||
licenses._provider = None
|
||
licenses._provider_mode = None
|
||
auth_providers._provider = None
|
||
auth_providers._provider_mode = None
|
||
world_store._stores.clear()
|
||
reset_engine()
|
||
|
||
from server.gateway.app import create_app
|
||
|
||
app = create_app()
|
||
yield app
|
||
|
||
reset_engine()
|
||
world_store._stores.clear()
|
||
licenses._provider = None
|
||
licenses._provider_mode = None
|
||
auth_providers._provider = None
|
||
auth_providers._provider_mode = None
|
||
|
||
|
||
def test_admin_api_requires_codes_mode(mock_license_app):
|
||
# mock 模式下管理接口明确拒绝，避免误以为配置生效
|
||
client = TestClient(mock_license_app)
|
||
headers = {"X-APS-Client": "desktop", "X-APS-Device-ID": "device-admin-0001"}
|
||
assert client.post(
|
||
"/api/auth/license/activate", headers=headers, json={"code": "APS-PERP-DEMO"},
|
||
).status_code == 200
|
||
response = client.get("/api/admin/license/codes", headers=headers)
|
||
assert response.status_code == 409
|
||
assert response.json()["detail"]["code"] == "LICENSE_MODE_DISABLED"
|
||
|
||
|
||
def test_custom_expiry_code_uses_absolute_deadline(codes_app):
|
||
admin = _admin_client(codes_app)
|
||
deadline = int(time.time()) + 3 * 24 * 3600
|
||
created = admin.post("/api/admin/license/codes", json={
|
||
"name": "自定义用户", "licenseType": "custom", "expiresAt": deadline,
|
||
})
|
||
assert created.status_code == 200
|
||
issued = created.json()["code"]
|
||
assert issued["licenseType"] == "custom"
|
||
assert issued["expiresAt"] == deadline
|
||
|
||
user = TestClient(codes_app)
|
||
body = user.post("/api/auth/license/activate", json={"code": issued["code"]}).json()
|
||
assert body["license"]["type"] == "custom"
|
||
assert body["license"]["label"] == "自定义授权"
|
||
assert body["license"]["expiresAt"] == deadline
|
||
|
||
|
||
def test_custom_expiry_validation(codes_app):
|
||
admin = _admin_client(codes_app)
|
||
missing = admin.post("/api/admin/license/codes", json={
|
||
"name": "缺时间", "licenseType": "custom",
|
||
})
|
||
assert missing.status_code == 400
|
||
assert missing.json()["detail"]["code"] == "LICENSE_EXPIRY_REQUIRED"
|
||
|
||
past = admin.post("/api/admin/license/codes", json={
|
||
"name": "过去时间", "licenseType": "custom", "expiresAt": int(time.time()) - 60,
|
||
})
|
||
assert past.status_code == 400
|
||
assert past.json()["detail"]["code"] == "LICENSE_EXPIRY_PAST"
|
||
|
||
|
||
def test_expired_custom_code_cannot_activate(codes_app):
|
||
admin = _admin_client(codes_app)
|
||
deadline = int(time.time()) + 3600
|
||
issued = admin.post("/api/admin/license/codes", json={
|
||
"name": "将到期", "licenseType": "custom", "expiresAt": deadline,
|
||
}).json()["code"]
|
||
|
||
# 直接把码的截止时间改到过去，模拟到期后激活
|
||
from server.db.database import get_session
|
||
from server.db.models import LicenseCode
|
||
|
||
with get_session() as session:
|
||
row = session.get(LicenseCode, issued["id"])
|
||
row.expires_at = int(time.time()) - 1
|
||
session.commit()
|
||
|
||
user = TestClient(codes_app)
|
||
response = user.post("/api/auth/license/activate", json={"code": issued["code"]})
|
||
assert response.status_code == 401
|
||
assert response.json()["error"]["code"] == "LICENSE_EXPIRED"
|