183 lines
7.5 KiB
Python
183 lines
7.5 KiB
Python
"""Explicitly local-only access bootstrap for manual Swagger verification."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import jwt
|
|
from sqlalchemy import text
|
|
|
|
from app.core.config import get_settings
|
|
from app.infrastructure.db import SessionFactory
|
|
|
|
_TEST_CUSTOMER_ID = 9_000_001
|
|
_TEST_USER_NO = "LOCAL-ADVISOR-TEST"
|
|
_TEST_USERNAME = "local_advisor_test"
|
|
_TEST_ROLE = "customer"
|
|
_TEST_ADMIN_ID = 9_000_002
|
|
_TEST_ADMIN_USER_NO = "LOCAL-ADVISOR-ADMIN"
|
|
_TEST_ADMIN_USERNAME = "local_advisor_admin"
|
|
_TEST_ADMIN_ROLE = "admin"
|
|
_PERMISSIONS = (
|
|
("agent:run", "agent", "run", "self"),
|
|
("fund:quote:read", "fund", "quote_read", "self"),
|
|
("suitability:read", "suitability", "read", "self"),
|
|
("customer-profile:read:self", "customer_profile", "read", "self"),
|
|
("investment-goal:read:self", "investment_goal", "read", "self"),
|
|
("investment-goal:write:self", "investment_goal", "write", "self"),
|
|
("investment-goal:confirm:self", "investment_goal", "confirm", "self"),
|
|
("asset-allocation:generate:self", "asset_allocation", "generate", "self"),
|
|
("product-recommendation:read:self", "product_recommendation", "read", "self"),
|
|
("recommendation-plan:generate:self", "recommendation_plan", "generate", "self"),
|
|
("recommendation-plan:read:self", "recommendation_plan", "read", "self"),
|
|
("portfolio-analysis:read:self", "portfolio_analysis", "read", "self"),
|
|
)
|
|
_ADMIN_PERMISSIONS = (
|
|
("product-governance:sync", "product_governance", "sync", "all"),
|
|
("product-governance:read", "product_governance", "read", "all"),
|
|
("product-governance:review", "product_governance", "review", "all"),
|
|
("recommendation-plan:review", "recommendation_plan", "review", "all"),
|
|
("market-data:read", "market_data", "read", "all"),
|
|
)
|
|
|
|
|
|
class LocalTestAccessService:
|
|
"""Prepare a non-production customer and sign a short-lived local token."""
|
|
|
|
async def issue_access_token(self) -> dict[str, object]:
|
|
settings = get_settings()
|
|
if settings.app_env.casefold() != "local":
|
|
raise RuntimeError("local test access is disabled outside APP_ENV=local")
|
|
await self._ensure_customer()
|
|
return self._issue_token(_TEST_CUSTOMER_ID)
|
|
|
|
async def issue_admin_access_token(self) -> dict[str, object]:
|
|
settings = get_settings()
|
|
if settings.app_env.casefold() != "local":
|
|
raise RuntimeError("local test access is disabled outside APP_ENV=local")
|
|
await self._ensure_admin()
|
|
return self._issue_token(_TEST_ADMIN_ID)
|
|
|
|
@staticmethod
|
|
def _issue_token(user_id: int) -> dict[str, object]:
|
|
settings = get_settings()
|
|
now = datetime.now(UTC)
|
|
expires_at = now + timedelta(hours=8)
|
|
private_key = LocalTestAccessService._load_private_key(settings.jwt_private_key_path)
|
|
token = jwt.encode(
|
|
{
|
|
"sub": str(user_id),
|
|
"iss": settings.jwt_issuer,
|
|
"aud": settings.jwt_audience,
|
|
"iat": now,
|
|
"nbf": now,
|
|
"exp": expires_at,
|
|
"jti": str(uuid4()),
|
|
},
|
|
private_key,
|
|
algorithm=settings.jwt_algorithm,
|
|
)
|
|
return {
|
|
"access_token": token,
|
|
"token_type": "bearer",
|
|
"expires_at": expires_at.isoformat(),
|
|
"user_id": str(user_id),
|
|
}
|
|
|
|
@staticmethod
|
|
def _load_private_key(configured_path: str) -> str:
|
|
path = Path(configured_path)
|
|
if not path.is_absolute():
|
|
path = Path.cwd() / path
|
|
return path.read_text(encoding="utf-8")
|
|
|
|
async def _ensure_customer(self) -> None:
|
|
await self._ensure_user(
|
|
user_id=_TEST_CUSTOMER_ID,
|
|
user_no=_TEST_USER_NO,
|
|
username=_TEST_USERNAME,
|
|
role_code=_TEST_ROLE,
|
|
role_name="Customer",
|
|
permissions=_PERMISSIONS,
|
|
)
|
|
|
|
async def _ensure_admin(self) -> None:
|
|
await self._ensure_user(
|
|
user_id=_TEST_ADMIN_ID,
|
|
user_no=_TEST_ADMIN_USER_NO,
|
|
username=_TEST_ADMIN_USERNAME,
|
|
role_code=_TEST_ADMIN_ROLE,
|
|
role_name="Local Compliance Administrator",
|
|
permissions=_ADMIN_PERMISSIONS,
|
|
)
|
|
|
|
async def _ensure_user(
|
|
self,
|
|
*,
|
|
user_id: int,
|
|
user_no: str,
|
|
username: str,
|
|
role_code: str,
|
|
role_name: str,
|
|
permissions: tuple[tuple[str, str, str, str], ...],
|
|
) -> None:
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
async with SessionFactory() as session, session.begin():
|
|
await session.execute(text("""
|
|
INSERT INTO sys_user
|
|
(id, user_no, username, password_hash, user_type,
|
|
professional_investor_status, fund_account_status, status,
|
|
created_at, updated_at)
|
|
VALUES
|
|
(:id, :user_no, :username, 'local-test-only', 'customer',
|
|
'none', 'closed', '正常', :now, :now)
|
|
ON DUPLICATE KEY UPDATE status='正常', updated_at=:now
|
|
"""), {
|
|
"id": user_id,
|
|
"user_no": user_no,
|
|
"username": username,
|
|
"now": now,
|
|
})
|
|
await session.execute(text("""
|
|
INSERT INTO sys_role (role_code, role_name, status, created_at, updated_at)
|
|
VALUES (:code, :role_name, 'active', :now, :now)
|
|
ON DUPLICATE KEY UPDATE
|
|
role_name=:role_name, status='active', updated_at=:now
|
|
"""), {"code": role_code, "role_name": role_name, "now": now})
|
|
role_id = await session.scalar(
|
|
text("SELECT id FROM sys_role WHERE role_code=:code"), {"code": role_code}
|
|
)
|
|
if role_id is None:
|
|
raise RuntimeError("local test role was not created")
|
|
for code, resource, action, scope in permissions:
|
|
await session.execute(text("""
|
|
INSERT INTO sys_permission
|
|
(permission_code, resource, action, data_scope, created_at, updated_at)
|
|
VALUES (:code, :resource, :action, :scope, :now, :now)
|
|
ON DUPLICATE KEY UPDATE
|
|
resource=:resource, action=:action,
|
|
data_scope=:scope, updated_at=:now
|
|
"""), {
|
|
"code": code,
|
|
"resource": resource,
|
|
"action": action,
|
|
"scope": scope,
|
|
"now": now,
|
|
})
|
|
permission_id = await session.scalar(
|
|
text("SELECT id FROM sys_permission WHERE permission_code=:code"),
|
|
{"code": code},
|
|
)
|
|
if permission_id is None:
|
|
raise RuntimeError(f"local test permission was not created: {code}")
|
|
await session.execute(text("""
|
|
INSERT INTO sys_role_permission (role_id, permission_id, created_at)
|
|
VALUES (:role_id, :permission_id, :now)
|
|
ON DUPLICATE KEY UPDATE created_at=:now
|
|
"""), {"role_id": role_id, "permission_id": permission_id, "now": now})
|
|
await session.execute(text("""
|
|
INSERT INTO sys_user_role (user_id, role_id, assigned_at)
|
|
VALUES (:user_id, :role_id, :now)
|
|
ON DUPLICATE KEY UPDATE assigned_at=:now, expires_at=NULL
|
|
"""), {"user_id": user_id, "role_id": role_id, "now": now})
|