ai-contract-platform/yuqei-contract-platform/services/contract-api/tests/test_proxy_auth.py

from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from yuqei_contract_api.config import ContractApiSettings, check_runtime_config
from yuqei_contract_api.identity import (
    InMemoryIdentityRepository,
    LastPlatformAdministratorError,
    TenantUserUpdateRequest,
)
from yuqei_contract_api.main import create_app


AUTH_VALUE = "x" * 48
REGISTRATION_AGREEMENT_ACCEPTANCES = [
    {"agreement_type": "user_terms", "version": "2026-08-01-draft", "accepted": True},
    {"agreement_type": "privacy_policy", "version": "2026-08-01-draft", "accepted": True},
    {"agreement_type": "service_agreement", "version": "2026-08-01-draft", "accepted": True},
    {"agreement_type": "ai_use_notice", "version": "2026-08-01-draft", "accepted": True},
]


def make_settings(tmp_path: Path) -> ContractApiSettings:
    return ContractApiSettings(
        environment="staging",
        database_url=f"sqlite+pysqlite:///{tmp_path / 'proxy-auth.db'}",
        database_auto_create=True,
        seed_demo_data=False,
        file_storage_dir=str(tmp_path / "files"),
        web_proxy_shared_secret=AUTH_VALUE,
        require_web_proxy_auth=True,
        public_policy_registration_fail_closed=False,
    )


def test_health_remains_public_for_container_liveness(tmp_path: Path) -> None:
    client = TestClient(create_app(make_settings(tmp_path)))

    response = client.get("/health")

    assert response.status_code == 200


def test_non_public_requests_require_the_server_side_proxy_key(tmp_path: Path) -> None:
    client = TestClient(create_app(make_settings(tmp_path)))

    missing = client.get("/api/v1/contracts")
    wrong = client.get("/api/v1/contracts", headers={"X-Contract-Proxy-Key": "wrong"})
    missing_user = client.get(
        "/api/v1/contracts",
        headers={"X-Contract-Proxy-Key": AUTH_VALUE, "X-Tenant-Id": "default"},
    )
    valid = client.get(
        "/api/v1/contracts",
        headers={
            "X-Contract-Proxy-Key": AUTH_VALUE,
            "X-Tenant-Id": "default",
            "X-User-Id": "user-current-admin",
        },
    )

    assert missing.status_code == 401
    assert wrong.status_code == 401
    assert missing_user.status_code == 401
    assert missing_user.json()["detail"] == "Authenticated user context is required."
    assert valid.status_code == 200


def test_platform_access_uses_fresh_repository_permission_without_roster_pii(tmp_path: Path) -> None:
    app = create_app(make_settings(tmp_path))
    client = TestClient(app)
    platform_admin_headers = {
        "X-Contract-Proxy-Key": AUTH_VALUE,
        "X-Tenant-Id": "default",
        "X-User-Id": "user-current-admin",
    }
    organization_admin = client.post(
        "/api/v1/identity/users",
        headers=platform_admin_headers,
        json={
            "email": "organization-admin@example.test",
            "display_name": "Organization Admin",
            "role": "admin",
            "status": "active",
        },
    )
    assert organization_admin.status_code == 200

    allowed = client.get("/api/v1/identity/platform-access", headers=platform_admin_headers)
    denied = client.get(
        "/api/v1/identity/platform-access",
        headers={
            "X-Contract-Proxy-Key": AUTH_VALUE,
            "X-Tenant-Id": "default",
            "X-User-Id": organization_admin.json()["id"],
        },
    )
    spoofed_without_proxy_auth = client.get(
        "/api/v1/identity/platform-access",
        headers={"X-Tenant-Id": "default", "X-User-Id": "user-current-admin"},
    )

    assert allowed.status_code == 200
    assert allowed.json() == {
        "platform_admin": True,
        "capabilities": ["prompt_templates.manage"],
    }
    assert denied.status_code == 200
    assert denied.json() == {"platform_admin": False, "capabilities": []}
    assert spoofed_without_proxy_auth.status_code == 401


def test_playbook_fallback_requires_contract_edit_permission(tmp_path: Path) -> None:
    client = TestClient(create_app(make_settings(tmp_path)))
    admin_headers = {
        "X-Contract-Proxy-Key": AUTH_VALUE,
        "X-Tenant-Id": "default",
        "X-User-Id": "user-current-admin",
    }
    approver = client.post(
        "/api/v1/identity/users",
        headers=admin_headers,
        json={
            "email": "fallback-approver@example.test",
            "display_name": "Fallback Approver",
            "role": "approver",
            "status": "active",
        },
    )
    assert approver.status_code == 200
    approver_headers = {
        "X-Contract-Proxy-Key": AUTH_VALUE,
        "X-Tenant-Id": "default",
        "X-User-Id": approver.json()["id"],
    }
    denied = client.post(
        "/api/v1/contracts/contract-does-not-matter/playbook-fallbacks",
        headers=approver_headers,
        json={
            "review_task_id": "review-does-not-matter",
            "snapshot_id": "playbook-snapshot-does-not-matter",
            "rule_id": "approval-rule-does-not-matter",
        },
    )
    assert denied.status_code == 403
    denied_adoption = client.post(
        "/api/v1/contracts/contract-does-not-matter/playbook-fallbacks/execution-does-not-matter/adopt",
        headers=approver_headers,
        json={
            "expected_candidate_body_hash": "0" * 64,
            "confirmed": True,
        },
    )
    assert denied_adoption.status_code == 403
    denied_combined = client.post(
        "/api/v1/contracts/contract-does-not-matter/playbook-candidates",
        headers=approver_headers,
        json={
            "review_task_id": "review-does-not-matter",
            "snapshot_id": "snapshot-does-not-matter",
            "decisions": [{"rule_id": "rule-does-not-matter", "decision": "apply"}],
        },
    )
    assert denied_combined.status_code == 403
    denied_combined_adoption = client.post(
        "/api/v1/contracts/contract-does-not-matter/playbook-candidates/candidate-does-not-matter/adopt",
        headers=approver_headers,
        json={"expected_candidate_body_hash": "0" * 64, "confirmed": True},
    )
    assert denied_combined_adoption.status_code == 403
    denied_combined_withdrawal = client.post(
        "/api/v1/contracts/contract-does-not-matter/playbook-candidates/candidate-does-not-matter/withdraw",
        headers=approver_headers,
        json={"reason": "无权限撤销"},
    )
    assert denied_combined_withdrawal.status_code == 403


def test_public_policy_proxy_bypass_is_limited_to_exact_read_routes(tmp_path: Path) -> None:
    client = TestClient(create_app(make_settings(tmp_path)))

    assert client.get("/api/v1/public-policies/current").status_code == 404
    assert client.get("/api/v1/public-policies/current/user_terms").status_code == 404
    assert client.get("/api/v1/public-policies/currentness").status_code == 401
    assert client.get("/api/v1/public-policies/current/not_a_policy").status_code == 401
    assert client.get("/api/v1/public-policies/current/user_terms/extra").status_code == 401


def test_required_agreement_routes_reject_missing_user_context(tmp_path: Path) -> None:
    client = TestClient(create_app(make_settings(tmp_path)))
    headers = {"X-Contract-Proxy-Key": AUTH_VALUE, "X-Tenant-Id": "default"}

    required = client.get("/api/v1/identity/agreements/required", headers=headers)
    accepted = client.post(
        "/api/v1/identity/agreements/consents",
        headers=headers,
        json={"items": []},
    )

    assert required.status_code == 401
    assert accepted.status_code == 401


def test_public_policy_preflight_requires_platform_admin_even_without_proxy_auth(tmp_path: Path) -> None:
    settings = make_settings(tmp_path).model_copy(update={"require_web_proxy_auth": False})
    client = TestClient(create_app(settings))

    response = client.get("/api/v1/admin/public-policy-promotion/preflight")

    assert response.status_code == 403


def test_production_config_requires_proxy_auth() -> None:
    settings = ContractApiSettings(
        environment="production",
        database_url="postgresql+psycopg://contract@postgres:5432/yuqei_contract",
        database_auto_create=False,
        seed_demo_data=False,
    )

    checks = {item.name: item for item in check_runtime_config(settings).checks}

    assert checks["web_proxy_auth"].status == "error"


def test_authenticated_tenant_cannot_override_identity_scope(tmp_path: Path) -> None:
    client = TestClient(create_app(make_settings(tmp_path)))
    proxy_headers = {"X-Contract-Proxy-Key": AUTH_VALUE}
    credential_value = "S" * 8 + "1" * 4

    registered = client.post(
        "/api/v1/identity/register-company",
        headers={**proxy_headers, "X-Tenant-Id": "default", "X-User-Id": "user-current-admin"},
        json={
            "tenant_id": "tenant-secure",
            "company_name": "Secure Tenant",
            "admin_email": "owner@secure.example",
            "admin_display_name": "Tenant Owner",
            "admin_password": credential_value,
            "agreement_acceptances": REGISTRATION_AGREEMENT_ACCEPTANCES,
        },
    )
    assert registered.status_code == 200
    session = registered.json()
    tenant_headers = {
        **proxy_headers,
        "X-Tenant-Id": session["tenant"]["id"],
        "X-User-Id": session["user"]["id"],
    }

    own_scope = client.get("/api/v1/identity/users?tenant_id=tenant-secure", headers=tenant_headers)
    foreign_scope = client.get("/api/v1/identity/users?tenant_id=default", headers=tenant_headers)

    assert own_scope.status_code == 200
    assert foreign_scope.status_code == 403

    admin_audit = client.get(
        "/api/v1/identity/agreements/consents?tenant_id=tenant-secure",
        headers=tenant_headers,
    )
    assert admin_audit.status_code == 200
    assert admin_audit.json()["total"] == 4

    observer_response = client.post(
        "/api/v1/identity/users",
        headers=tenant_headers,
        json={
            "email": "observer@secure.example",
            "display_name": "Consent Observer",
            "role": "observer",
            "status": "active",
        },
    )
    assert observer_response.status_code == 200
    observer = observer_response.json()
    observer_headers = {
        **proxy_headers,
        "X-Tenant-Id": "tenant-secure",
        "X-User-Id": observer["id"],
    }
    own_consent_audit = client.get(
        f"/api/v1/identity/agreements/consents?user_id={observer['id']}",
        headers=observer_headers,
    )
    workspace_consent_audit = client.get(
        "/api/v1/identity/agreements/consents",
        headers=observer_headers,
    )

    assert own_consent_audit.status_code == 200
    assert own_consent_audit.json()["total"] == 0
    assert workspace_consent_audit.status_code == 403

    authoritative_account = client.get("/api/v1/identity/account", headers=observer_headers)
    forbidden_identity_fields = client.patch(
        "/api/v1/identity/account",
        headers=observer_headers,
        json={
            "tenant_id": "default",
            "user_id": session["user"]["id"],
            "email": "forged@example.com",
            "role": "admin",
        },
    )
    allowed_profile = client.patch(
        "/api/v1/identity/account",
        headers=observer_headers,
        json={"display_name": "Consent Observer Updated", "timezone": "Asia/Shanghai"},
    )

    assert authoritative_account.status_code == 200
    assert authoritative_account.json()["email"] == observer["email"]
    assert authoritative_account.json()["role"] == "observer"
    assert forbidden_identity_fields.status_code == 422
    assert allowed_profile.status_code == 200
    assert allowed_profile.json()["email"] == observer["email"]
    assert allowed_profile.json()["role"] == "observer"
    assert allowed_profile.json()["display_name"] == "Consent Observer Updated"


def test_email_verification_is_scoped_and_refreshes_self_session(tmp_path: Path) -> None:
    settings = make_settings(tmp_path).model_copy(
        update={"identity_challenge_resend_cooldown_seconds": 0}
    )
    client = TestClient(create_app(settings))
    proxy_headers = {"X-Contract-Proxy-Key": AUTH_VALUE}
    registered = client.post(
        "/api/v1/identity/register-company",
        headers={**proxy_headers, "X-Tenant-Id": "default", "X-User-Id": "user-current-admin"},
        json={
            "tenant_id": "tenant-email-scope",
            "company_name": "Email Scope Tenant",
            "admin_email": "owner@email-scope.example",
            "admin_display_name": "Workspace Owner",
            "admin_password": "OwnerPassword123",  # pragma: allowlist secret
            "agreement_acceptances": REGISTRATION_AGREEMENT_ACCEPTANCES,
        },
    )
    assert registered.status_code == 200
    owner = registered.json()["user"]
    owner_headers = {
        **proxy_headers,
        "X-Tenant-Id": "tenant-email-scope",
        "X-User-Id": owner["id"],
    }
    admin_response = client.post(
        "/api/v1/identity/users",
        headers=owner_headers,
        json={
            "email": "admin@email-scope.example",
            "display_name": "Workspace Admin",
            "role": "admin",
            "status": "active",
        },
    )
    assert admin_response.status_code == 200
    workspace_admin = admin_response.json()
    demoted = client.patch(
        f"/api/v1/identity/users/{owner['id']}",
        headers=owner_headers,
        json={"role": "business"},
    )
    assert demoted.status_code == 200
    admin_headers = {
        **proxy_headers,
        "X-Tenant-Id": "tenant-email-scope",
        "X-User-Id": workspace_admin["id"],
    }

    forbidden_send = client.post(
        "/api/v1/identity/email-verification/send",
        headers=owner_headers,
        json={"email": workspace_admin["email"]},
    )
    foreign_workspace_send = client.post(
        "/api/v1/identity/email-verification/send",
        headers=owner_headers,
        json={"tenant_id": "default", "email": owner["email"]},
    )
    assert forbidden_send.status_code == 403
    assert foreign_workspace_send.status_code == 403

    admin_managed = client.post(
        "/api/v1/identity/email-verification/send",
        headers=admin_headers,
        json={"email": owner["email"]},
    )
    assert admin_managed.status_code == 200
    admin_confirmed = client.post(
        "/api/v1/identity/email-verification/confirm",
        headers=admin_headers,
        json={
            "challenge_id": admin_managed.json()["id"],
            "code": admin_managed.json()["dev_code"],
        },
    )
    assert admin_confirmed.status_code == 200
    assert admin_confirmed.json()["session"] is None

    self_challenge = client.post(
        "/api/v1/identity/email-verification/send",
        headers=owner_headers,
        json={"email": owner["email"]},
    )
    assert self_challenge.status_code == 200
    self_confirmed = client.post(
        "/api/v1/identity/email-verification/confirm",
        headers=owner_headers,
        json={
            "challenge_id": self_challenge.json()["id"],
            "code": self_challenge.json()["dev_code"],
        },
    )
    assert self_confirmed.status_code == 200
    refreshed_session = self_confirmed.json()["session"]
    assert refreshed_session["user"]["id"] == owner["id"]
    assert refreshed_session["tenant"]["id"] == "tenant-email-scope"
    assert refreshed_session["global_account"]["email_verified_at"] is not None

    admin_challenge = client.post(
        "/api/v1/identity/email-verification/send",
        headers=admin_headers,
        json={"email": workspace_admin["email"]},
    )
    assert admin_challenge.status_code == 200
    cross_account_confirm = client.post(
        "/api/v1/identity/email-verification/confirm",
        headers=owner_headers,
        json={
            "challenge_id": admin_challenge.json()["id"],
            "code": admin_challenge.json()["dev_code"],
        },
    )
    assert cross_account_confirm.status_code == 403
    admin_self_confirmed = client.post(
        "/api/v1/identity/email-verification/confirm",
        headers=admin_headers,
        json={
            "challenge_id": admin_challenge.json()["id"],
            "code": admin_challenge.json()["dev_code"],
        },
    )
    assert admin_self_confirmed.status_code == 200


def test_platform_admin_api_uses_explicit_permission_claim(tmp_path: Path) -> None:
    client = TestClient(create_app(make_settings(tmp_path)))
    proxy_headers = {"X-Contract-Proxy-Key": AUTH_VALUE, "X-Tenant-Id": "default"}

    explicit_admin = client.post(
        "/api/v1/identity/tenants",
        headers={**proxy_headers, "X-User-Id": "user-current-admin"},
        json={"id": "tenant-explicit-platform-admin", "name": "Explicit Admin Co"},
    )
    legacy_id_only = client.post(
        "/api/v1/identity/tenants",
        headers={**proxy_headers, "X-User-Id": "current-admin"},
        json={"id": "tenant-legacy-id-only", "name": "Legacy ID Co"},
    )

    assert explicit_admin.status_code == 200
    assert legacy_id_only.status_code == 403


def test_platform_admin_roster_and_final_admin_guard(tmp_path: Path) -> None:
    app = create_app(make_settings(tmp_path))
    identity_repository = app.state.contract_repository.identity
    second_admin = identity_repository.bootstrap_admin_credential(
        tenant_id="default",
        company_name="Platform Operations",
        email="second-platform-admin@example.com",
        display_name="Second Platform Admin",
        password="SecondAdminPassword123",  # pragma: allowlist secret
    )
    assert second_admin.status == "created"

    client = TestClient(app)
    proxy_headers = {"X-Contract-Proxy-Key": AUTH_VALUE}
    first_admin_headers = {
        **proxy_headers,
        "X-Tenant-Id": "default",
        "X-User-Id": "user-current-admin",
    }
    second_admin_headers = {
        **proxy_headers,
        "X-Tenant-Id": "default",
        "X-User-Id": second_admin.user_id,
    }

    roster = client.get(
        "/api/v1/identity/platform-administrators",
        headers=first_admin_headers,
    )
    assert roster.status_code == 200
    assert roster.json()["total"] == 2
    assert roster.json()["active_total"] == 2
    assert {item["user_id"] for item in roster.json()["items"]} == {
        "user-current-admin",
        second_admin.user_id,
    }
    assert all("password" not in item and "credential" not in item for item in roster.json()["items"])

    tenant_admin_session = client.post(
        "/api/v1/identity/register-company",
        headers={**proxy_headers, "X-Tenant-Id": "default", "X-User-Id": "user-current-admin"},
        json={
            "tenant_id": "tenant-platform-boundary",
            "company_name": "Tenant Boundary Co",
            "admin_email": "owner@tenant-boundary.example",
            "admin_display_name": "Tenant Boundary Owner",
            "admin_password": "TenantOwnerPassword123",  # pragma: allowlist secret
            "agreement_acceptances": REGISTRATION_AGREEMENT_ACCEPTANCES,
        },
    )
    assert tenant_admin_session.status_code == 200
    tenant_admin = tenant_admin_session.json()["user"]
    tenant_admin_headers = {
        **proxy_headers,
        "X-Tenant-Id": "tenant-platform-boundary",
        "X-User-Id": tenant_admin["id"],
    }
    assert client.get(
        "/api/v1/identity/platform-administrators",
        headers=tenant_admin_headers,
    ).status_code == 403
    assert client.patch(
        f"/api/v1/identity/users/{second_admin.user_id}",
        headers=tenant_admin_headers,
        json={"status": "disabled"},
    ).status_code == 403

    first_login_before_rotation = client.post(
        "/api/v1/identity/login",
        headers=proxy_headers,
        json={
            "tenant_id": "default",
            "email": "admin@example.com",
            "password": "admin654321",  # pragma: allowlist secret
        },
    )
    second_login_before_rotation = client.post(
        "/api/v1/identity/login",
        headers=proxy_headers,
        json={
            "tenant_id": "default",
            "email": "second-platform-admin@example.com",
            "password": "SecondAdminPassword123",  # pragma: allowlist secret
        },
    )
    assert first_login_before_rotation.status_code == 200
    assert second_login_before_rotation.status_code == 200

    rotated = identity_repository.bootstrap_admin_credential(
        tenant_id="default",
        company_name="Platform Operations",
        email="second-platform-admin@example.com",
        display_name="Second Platform Admin",
        password="SecondAdminRotated123",  # pragma: allowlist secret
        force_rotate=True,
    )
    assert rotated.status == "rotated"
    assert client.post(
        "/api/v1/identity/login",
        headers=proxy_headers,
        json={
            "tenant_id": "default",
            "email": "second-platform-admin@example.com",
            "password": "SecondAdminPassword123",  # pragma: allowlist secret
        },
    ).status_code == 401
    assert client.post(
        "/api/v1/identity/login",
        headers=proxy_headers,
        json={
            "tenant_id": "default",
            "email": "second-platform-admin@example.com",
            "password": "SecondAdminRotated123",  # pragma: allowlist secret
        },
    ).status_code == 200
    assert client.post(
        "/api/v1/identity/login",
        headers=proxy_headers,
        json={
            "tenant_id": "default",
            "email": "admin@example.com",
            "password": "admin654321",  # pragma: allowlist secret
        },
    ).status_code == 200

    disabled = client.patch(
        f"/api/v1/identity/users/{second_admin.user_id}",
        headers=first_admin_headers,
        json={"status": "disabled", "updated_by": "user-current-admin"},
    )
    assert disabled.status_code == 200
    assert disabled.json()["status"] == "disabled"

    lost_platform_access = client.post(
        "/api/v1/identity/tenants",
        headers=second_admin_headers,
        json={"id": "tenant-disabled-admin", "name": "Disabled Admin Tenant"},
    )
    assert lost_platform_access.status_code == 403

    final_admin_deactivation = client.patch(
        "/api/v1/identity/users/user-current-admin",
        headers=first_admin_headers,
        json={"status": "disabled", "updated_by": "user-current-admin"},
    )
    assert final_admin_deactivation.status_code == 409
    assert final_admin_deactivation.json() == {
        "detail": "The final active platform administrator cannot be disabled."
    }

    final_roster = client.get(
        "/api/v1/identity/platform-administrators",
        headers=first_admin_headers,
    )
    assert final_roster.status_code == 200
    assert final_roster.json()["total"] == 2
    assert final_roster.json()["active_total"] == 1
    statuses = {item["user_id"]: item["status"] for item in final_roster.json()["items"]}
    assert statuses == {
        "user-current-admin": "active",
        second_admin.user_id: "disabled",
    }


def test_in_memory_identity_repository_protects_final_platform_admin() -> None:
    identity_repository = InMemoryIdentityRepository()

    roster = identity_repository.list_platform_administrators()
    assert roster.total == roster.active_total == 1
    assert roster.items[0].user_id == "user-current-admin"

    with pytest.raises(
        LastPlatformAdministratorError,
        match="final active platform administrator",
    ):
        identity_repository.update_user(
            "user-current-admin",
            TenantUserUpdateRequest(status="disabled"),
        )

    assert identity_repository.has_platform_permission(
        tenant_id="default",
        user_id="user-current-admin",
    )