live-hub-py/tests/test_dc_auth_v2.py

"""DCAuth v2 + 短信验证码回归测试。

这些用例只依赖 SQLite 内存库与纯函数，不需要 MySQL / 真实短信渠道。
``tests/conftest.py`` 的全局 autouse fixture 默认连 MySQL，这里用同名 fixture
覆盖掉，保证在只装了 Python 依赖的环境里也能跑。
"""

from __future__ import annotations

import base64
from datetime import UTC, datetime, timedelta

import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from starlette.requests import Request
from starlette.responses import Response

from web.backend.database import Base
from web.backend.errors import ApiError
from web.backend.models import (
    AuditLog,
    DeviceChallenge,
    LoginSmsCode,
    User,
    UserDevice,
)
from web.backend.routers import auth as auth_router
from web.backend.routers.dcauth import authorize as authorize_endpoint
from web.backend.schemas import (
    DeviceAuthorizeRequest,
    LoginCompleteRequest,
    LoginPrepareRequest,
    LoginRequest,
)
from web.backend.security import hash_password
from web.backend.services import dc_auth_service as dc
from web.backend.services import login_sms_service as sms

DEVICE_ID_A = "D1-AAAA-BBBB-CCCC"
DEVICE_ID_B = "D1-DDDD-EEEE-FFFF"
PUBLIC_KEY_A = "public-key-a"
PUBLIC_KEY_B = "public-key-b"


@pytest.fixture(scope="session", autouse=True)
def setup_test_database():
    """覆盖 tests/conftest.py：本文件不需要 MySQL。"""
    yield None


@pytest.fixture(autouse=True)
def clean_tables_per_test():
    """覆盖 tests/conftest.py：每个用例使用独立 SQLite 内存库。"""
    yield


@pytest.fixture()
def db() -> Session:
    engine = create_engine("sqlite://")
    Base.metadata.create_all(
        engine,
        tables=[
            User.__table__,
            UserDevice.__table__,
            DeviceChallenge.__table__,
            LoginSmsCode.__table__,
            AuditLog.__table__,
        ],
    )
    factory = sessionmaker(bind=engine)
    session = factory()
    try:
        yield session
    finally:
        session.close()
        engine.dispose()


def _make_user(
    db: Session, username: str = "测试用户", phone: str = "13800138000"
) -> User:
    user = User(
        username=username,
        password_hash="not-used",
        role="operation",
        is_active=True,
        phone=phone,
        remark="",
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


def _b64url(raw: bytes) -> str:
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")


def _keypair() -> tuple[Ed25519PrivateKey, str]:
    private_key = Ed25519PrivateKey.generate()
    public_key = private_key.public_key().public_bytes(
        serialization.Encoding.Raw,
        serialization.PublicFormat.Raw,
    )
    return private_key, _b64url(public_key)


def _authorize_payload(
    private_key: Ed25519PrivateKey,
    challenge: dict,
    device_id: str = DEVICE_ID_A,
    timestamp: int | None = None,
    client_version: str = "2.0.0",
) -> dict:
    ts = timestamp if timestamp is not None else int(datetime.now(UTC).timestamp())
    canonical = dc.build_device_canonical(
        "live_hub",
        challenge["challengeId"],
        challenge["challenge"],
        device_id,
        ts,
    )
    signature = private_key.sign(canonical.encode("utf-8"))
    _, public_key = _keypair_from(private_key)
    return {
        "app": "live_hub",
        "challengeId": challenge["challengeId"],
        "challenge": challenge["challenge"],
        "deviceId": device_id,
        "publicKey": public_key,
        "signature": _b64url(signature),
        "timestamp": ts,
        "clientVersion": client_version,
    }


def _keypair_from(private_key: Ed25519PrivateKey) -> tuple[Ed25519PrivateKey, str]:
    public_key = private_key.public_key().public_bytes(
        serialization.Encoding.Raw,
        serialization.PublicFormat.Raw,
    )
    return private_key, _b64url(public_key)


class TestCanonical:
    def test_canonical_is_exact_v2_string(self):
        canonical = dc.build_device_canonical(
            "live_hub",
            "challenge-id",
            "challenge-value",
            "D1-AAAA-BBBB-CCCC",
            1700000000,
        )
        assert canonical == (
            "dcauth.v2\n"
            "live_hub\n"
            "challenge-id\n"
            "challenge-value\n"
            "D1-AAAA-BBBB-CCCC\n"
            "1700000000"
        )

    def test_client_version_major_parsing(self):
        assert dc.parse_client_major("2.0.0") == 2
        assert dc.parse_client_major("2") == 2
        assert dc.parse_client_major("1.9.9") == 1
        assert dc.parse_client_major("abc") is None
        assert dc.parse_client_major("") is None

    def test_client_version_below_2_raises_426(self):
        with pytest.raises(ApiError) as exc:
            dc.ensure_supported_client_version("1.9.9")
        assert exc.value.status_code == 426
        assert exc.value.error_code == "client_outdated"

    def test_ed25519_signature_roundtrip_and_tamper(self):
        private_key, public_key = _keypair()
        canonical = dc.build_device_canonical(
            "live_hub", "cid", "chal", DEVICE_ID_A, 1700000000
        )
        signature = private_key.sign(canonical.encode("utf-8"))
        dc.verify_device_signature(public_key, _b64url(signature), canonical)

        tampered = canonical.replace("1700000000", "1700000001")
        with pytest.raises(ApiError) as exc:
            dc.verify_device_signature(public_key, _b64url(signature), tampered)
        assert exc.value.status_code == 401


class TestAuthorizeEndpoint:
    def test_authorize_endpoint_returns_426(self, db: Session):
        private_key, public_key = _keypair()
        request = DeviceAuthorizeRequest(
            app="live_hub",
            challengeId="missing",
            challenge="challenge",
            deviceId=DEVICE_ID_A,
            publicKey=public_key,
            signature=_b64url(private_key.sign(b"whatever")),
            timestamp=int(datetime.now(UTC).timestamp()),
            clientVersion="1.0.0",
        )
        # 直接调用路由函数：版本校验发生在任何数据库查询之前。
        with pytest.raises(ApiError) as exc:
            authorize_endpoint(request, db)
        assert exc.value.status_code == 426
        assert exc.value.error_code == "client_outdated"


class TestChallengeLifecycle:
    def test_challenge_single_consumption_and_ticket_lifecycle(self, db: Session):
        user = _make_user(db)
        challenge = dc.create_login_challenge(db, user.id, "http://localhost:8800")
        private_key, _ = _keypair()

        payload = _authorize_payload(private_key, challenge)
        dc.authorize_device(db, payload)

        row = dc.get_challenge(db, challenge["challengeId"])
        assert row is not None
        assert row.status == "authorized"
        assert row.device_id == DEVICE_ID_A
        assert row.client_version == "2.0.0"
        assert row.authorized_at is not None

        # 同一个 challenge 不能再被第二次授权。
        with pytest.raises(ApiError) as exc:
            dc.authorize_device(db, payload)
        assert exc.value.status_code == 409

        # 轮询签发 ticket（只签发一次）。
        first = dc.get_challenge_status(
            db, challenge["challengeId"], challenge["pollToken"]
        )
        assert first["status"] == "authorized"
        assert first["deviceTicket"]
        second = dc.get_challenge_status(
            db, challenge["challengeId"], challenge["pollToken"]
        )
        assert second["status"] == "ticket_issued"

        # ticket 只能消费一次。
        consumed = dc.consume_device_ticket(
            db,
            challenge["challengeId"],
            challenge["pollToken"],
            first["deviceTicket"],
        )
        assert consumed.status == "consumed"
        with pytest.raises(ApiError) as exc:
            dc.consume_device_ticket(
                db,
                challenge["challengeId"],
                challenge["pollToken"],
                first["deviceTicket"],
            )
        assert exc.value.status_code == 401

    def test_poll_token_hash_mismatch_is_forbidden(self, db: Session):
        user = _make_user(db)
        challenge = dc.create_login_challenge(db, user.id, "http://localhost:8800")
        with pytest.raises(ApiError) as exc:
            dc.get_challenge_status(db, challenge["challengeId"], "wrong-token")
        assert exc.value.status_code == 403


class TestDeviceBinding:
    def test_mismatched_device_is_rejected(self, db: Session):
        user = _make_user(db)
        now = datetime.now(UTC)
        db.add(
            UserDevice(
                user_id=user.id,
                device_id=DEVICE_ID_A,
                public_key=PUBLIC_KEY_A,
                client_version="2.0.0",
                status="active",
                bound_at=now,
                created_at=now,
                updated_at=now,
            )
        )
        db.commit()

        challenge = DeviceChallenge(
            challenge_id="cid-mismatch",
            app="live_hub",
            purpose="login",
            challenge="chal",
            poll_token_hash=dc.hash_token("poll"),
            status="consumed",
            user_id=user.id,
            device_id=DEVICE_ID_B,
            public_key=PUBLIC_KEY_B,
            client_version="2.0.0",
            expires_at=now + timedelta(minutes=1),
            created_at=now,
            updated_at=now,
        )
        with pytest.raises(ApiError) as exc:
            dc.ensure_device_binding(db, user, challenge)
        assert exc.value.status_code == 401
        assert exc.value.error_code == "device_mismatch"
        assert exc.value.detail == "当前设备与账号绑定设备不一致，请联系管理员重置设备"

    def test_first_device_is_bound_automatically(self, db: Session):
        user = _make_user(db)
        now = datetime.now(UTC)
        challenge = DeviceChallenge(
            challenge_id="cid-bind",
            app="live_hub",
            purpose="login",
            challenge="chal",
            poll_token_hash=dc.hash_token("poll"),
            status="consumed",
            user_id=user.id,
            device_id=DEVICE_ID_A,
            public_key=PUBLIC_KEY_A,
            client_version="2.0.0",
            expires_at=now + timedelta(minutes=1),
            created_at=now,
            updated_at=now,
        )
        bound = dc.ensure_device_binding(db, user, challenge)
        assert bound.status == "active"
        assert bound.device_id == DEVICE_ID_A
        assert bound.public_key == PUBLIC_KEY_A


class TestSmsCode:
    def test_mock_provider_skips_daily_limit(self):
        assert sms.should_enforce_login_sms_daily_limit("mock") is False
        assert sms.should_enforce_login_sms_daily_limit("aliyun") is True

    def test_five_wrong_attempts_lock_the_code(self, db: Session):
        user = _make_user(db)
        now = datetime.now(UTC)
        row = LoginSmsCode(
            user_id=user.id,
            phone=user.phone,
            purpose="login",
            code_hash=sms.hash_sms_code("123456"),
            expires_at=now + timedelta(minutes=5),
            failed_attempt_count=0,
            request_ip="",
            created_at=now,
        )
        db.add(row)
        db.commit()

        for attempt in range(1, 6):
            with pytest.raises(ApiError) as exc:
                sms.verify_login_sms_code(db, user.id, "000000")
            assert exc.value.error_code == (
                "sms_code_locked" if attempt == 5 else "sms_code_invalid"
            )

        db.refresh(row)
        assert row.failed_attempt_count == 5
        assert row.consumed_at is not None

        # 锁定后即使输入正确验证码也不能再用。
        with pytest.raises(ApiError) as exc:
            sms.verify_login_sms_code(db, user.id, "123456")
        assert exc.value.error_code in {"sms_code_invalid", "sms_code_locked"}

    def test_correct_code_is_consumed_once(self, db: Session):
        user = _make_user(db)
        now = datetime.now(UTC)
        db.add(
            LoginSmsCode(
                user_id=user.id,
                phone=user.phone,
                purpose="login",
                code_hash=sms.hash_sms_code("654321"),
                expires_at=now + timedelta(minutes=5),
                failed_attempt_count=0,
                request_ip="",
                created_at=now,
            )
        )
        db.commit()

        assert sms.verify_login_sms_code(db, user.id, "654321") == user.phone
        with pytest.raises(ApiError) as exc:
            sms.verify_login_sms_code(db, user.id, "654321")
        assert exc.value.error_code == "sms_code_invalid"

    def test_mask_and_hash_helpers(self):
        assert sms.mask_phone("13800138000") == "138****8000"
        assert sms.hash_sms_code("123456") == sms.hash_sms_code("123456")
        assert sms.hash_sms_code("123456") != "123456"
        assert sms.normalize_phone(" 13800138000 ") == "13800138000"


def _request() -> Request:
    return Request(
        {
            "type": "http",
            "http_version": "1.1",
            "method": "POST",
            "scheme": "http",
            "path": "/api/auth/prepare",
            "raw_path": b"/api/auth/prepare",
            "query_string": b"",
            "headers": [(b"origin", b"http://localhost:5174")],
            "client": ("127.0.0.1", 12345),
            "server": ("localhost", 8800),
            "root_path": "",
        }
    )


class TestLoginFlow:
    def test_prepare_complete_reuses_jwt_cookie_session(self, db: Session):
        user = _make_user(db)
        user.password_hash = hash_password("secret123")
        db.commit()

        now = datetime.now(UTC)
        db.add(
            LoginSmsCode(
                user_id=user.id,
                phone=user.phone,
                purpose="login",
                code_hash=sms.hash_sms_code("123456"),
                expires_at=now + timedelta(minutes=5),
                failed_attempt_count=0,
                request_ip="",
                created_at=now,
            )
        )
        db.commit()

        prepared = auth_router.prepare_login(
            req=LoginPrepareRequest(
                username=user.username,
                password="secret123",
                sms_code="123456",
                api_base="http://localhost:5174",
            ),
            request=_request(),
            db=db,
        )
        assert prepared.launch_url.startswith("dcauth://authorize?")
        assert "app=live_hub" in prepared.launch_url
        assert "api=http%3A%2F%2Flocalhost%3A5174" in prepared.launch_url

        private_key, _ = _keypair()
        dc.authorize_device(
            db,
            _authorize_payload(
                private_key,
                {
                    "challengeId": prepared.challenge_id,
                    "challenge": prepared.challenge,
                },
            ),
        )
        status = dc.get_challenge_status(db, prepared.challenge_id, prepared.poll_token)
        assert status["status"] == "authorized"

        response = Response()
        result = auth_router.complete_login(
            req=LoginCompleteRequest(
                challenge_id=prepared.challenge_id,
                poll_token=prepared.poll_token,
                device_ticket=status["deviceTicket"],
            ),
            response=response,
            db=db,
        )
        assert result.username == user.username
        cookie = response.headers.get("set-cookie", "")
        assert "access_token=" in cookie
        assert "HttpOnly" in cookie

        device = (
            db.query(UserDevice)
            .filter(UserDevice.user_id == user.id, UserDevice.status == "active")
            .one()
        )
        assert device.device_id == DEVICE_ID_A

    def test_legacy_login_only_allowed_without_phone(self, db: Session):
        user = _make_user(db, phone="")
        user.password_hash = hash_password("secret123")
        db.commit()

        response = Response()
        result = auth_router.login(
            req=LoginRequest(username=user.username, password="secret123"),
            response=response,
            db=db,
        )
        assert result.username == user.username
        assert "access_token=" in response.headers.get("set-cookie", "")

        user.phone = "13800138000"
        db.commit()
        with pytest.raises(ApiError) as exc:
            auth_router.login(
                req=LoginRequest(username=user.username, password="secret123"),
                response=Response(),
                db=db,
            )
        assert exc.value.status_code == 403
        assert exc.value.error_code == "secure_login_required"