aps-agent/tests/golden/test_web_license_codes_auth.py

"""正式授权码（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_panel_issued_code_activates_without_env_codes(codes_app, monkeypatch):
    """回归：未配置 APS_LICENSE_CODES 时，面板发放的 DB 授权码也必须可激活。"""
    monkeypatch.delenv("APS_LICENSE_CODES", raising=False)
    admin = _admin_client(codes_app)
    created = admin.post("/api/admin/license/codes", json={
        "name": "无环境配置", "licenseType": "year",
    })
    assert created.status_code == 200
    issued = created.json()["code"]

    user = TestClient(codes_app)
    response = user.post("/api/auth/license/activate", json={"code": issued["code"]})
    assert response.status_code == 200
    assert response.json()["user"]["fullname"] == "无环境配置"


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"