|
12 | 12 | from pydantic import BaseModel
|
13 | 13 | from pydantic.fields import ModelField
|
14 | 14 | from starlette.requests import Request
|
| 15 | +from starlette.responses import RedirectResponse |
15 | 16 |
|
16 | 17 | from fastapi_user_auth.admin.utils import get_admin_action_options_by_subject
|
| 18 | +from fastapi_user_auth.auth import Auth |
17 | 19 | from fastapi_user_auth.auth.models import Role, User
|
18 | 20 | from fastapi_user_auth.auth.schemas import SystemUserEnum
|
19 | 21 | from fastapi_user_auth.mixins.admin import AuthFieldModelAdmin, AuthSelectModelAdmin
|
| 22 | +from fastapi_user_auth.mixins.models import PkMixin, UsernameMixin |
20 | 23 | from fastapi_user_auth.utils.casbin import (
|
21 | 24 | get_subject_effect_matrix,
|
22 | 25 | get_subject_page_permissions,
|
@@ -428,3 +431,55 @@ async def handle(self, request: Request, item_id: List[str], data: BaseModel, **
|
428 | 431 | permissions = [perm for perm in permissions if enforcer.enforce("u:" + identity, *permission_decode(perm))]
|
429 | 432 | await update_subject_page_permissions(enforcer, subject=subject, permissions=permissions) # 更新角色权限
|
430 | 433 | return BaseApiOut(msg="success")
|
| 434 | + |
| 435 | + |
| 436 | +class CopyUserAuthLinkAction(ModelAction): |
| 437 | + """复制用户免登录链接""" |
| 438 | + |
| 439 | + action = amis.ActionType.Dialog( |
| 440 | + name="copy_user_auth_link", |
| 441 | + icon="fa fa-link", |
| 442 | + tooltip="用户免登录链接", |
| 443 | + level=amis.LevelEnum.danger, |
| 444 | + dialog=amis.Dialog( |
| 445 | + size=amis.SizeEnum.md, |
| 446 | + title="用户免登录链接", |
| 447 | + ), |
| 448 | + ) |
| 449 | + form_init = True |
| 450 | + form = amis.Form(static=True, disabled=True) # type: ignore # 禁用表单 |
| 451 | + |
| 452 | + class schema(UsernameMixin, PkMixin): |
| 453 | + auth_url: str = Field( |
| 454 | + title="授权链接", |
| 455 | + description="复制链接到浏览器打开即可免登录", |
| 456 | + amis_form_item=amis.Static( |
| 457 | + copyable=True, |
| 458 | + ), |
| 459 | + ) |
| 460 | + |
| 461 | + async def get_init_data(self, request: Request, **kwargs) -> BaseApiOut[Any]: |
| 462 | + """复制用户免登录链接""" |
| 463 | + item_id = request.query_params.get("item_id") |
| 464 | + items = await self.admin.fetch_items(item_id) |
| 465 | + user: User = items[0] |
| 466 | + auth: Auth = request.auth |
| 467 | + token_data = { |
| 468 | + "id": user.id, |
| 469 | + "username": user.username, |
| 470 | + } |
| 471 | + token = await auth.backend.token_store.write_token(token_data) |
| 472 | + return BaseApiOut( |
| 473 | + msg="操作成功", |
| 474 | + data={**token_data, "auth_url": f"{str(request.base_url)[:-1]}{self.site.router_path}/login_by_token?token={token}"}, |
| 475 | + ) |
| 476 | + |
| 477 | + def register_router(self): |
| 478 | + @self.site.router.get("/login_by_token", include_in_schema=False) |
| 479 | + async def login_by_token(token: str): |
| 480 | + """通过url中的token登录""" |
| 481 | + response = RedirectResponse(self.site.settings.site_path) |
| 482 | + response.set_cookie("Authorization", f"bearer {token}") |
| 483 | + return response |
| 484 | + |
| 485 | + return super().register_router() |
0 commit comments