Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 43 additions & 31 deletions apps/application/views/application_access_token.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
# coding=utf-8
"""
@project: MaxKB
@Author:虎虎
@file: application_token.py
@date:2025/6/9 17:42
@desc:
@project: MaxKB
@Author:虎虎
@file: application_token.py
@date:2025/6/9 17:42
@desc:
"""

from django.db.models import QuerySet
from django.utils.translation import gettext_lazy as _
from drf_spectacular.utils import extend_schema
Expand All @@ -24,56 +25,67 @@
from common.auth.struct.aggregate_permission import ViewPermission
from common.log.log import log


def get_application_operation_object(application_id):
application_model = QuerySet(model=Application).filter(id=application_id).first()
if application_model is not None:
return {
"name": application_model.name
}
return {"name": application_model.name}
return {}


class AccessToken(APIView):
authentication_classes = [TokenAuth]

@extend_schema(
methods=['PUT'],
methods=["PUT"],
description=_("Modify application access restriction information"),
summary=_("Modify application access restriction information"),
operation_id=_("Modify application access restriction information"), # type: ignore
parameters=ApplicationAccessTokenAPI.get_parameters(),
request=ApplicationAccessTokenAPI.get_request(),
tags=[_('Application')] # type: ignore
tags=[_("Application")], # type: ignore
)
@log(
menu="Application",
operate="Modify application access token",
get_operation_object=lambda r, k: get_application_operation_object((k.get("application_id"))),
)
@has_permissions(
PermissionConstants.APPLICATION_OVERVIEW_PUBLIC.get_workspace_application_permission(),
PermissionConstants.APPLICATION_OVERVIEW_PUBLIC.get_workspace_permission_workspace_manage_role(),
ViewPermission(
[RoleConstants.USER.get_workspace_role()],
[PermissionConstants.APPLICATION.get_workspace_application_permission()],
compare=CompareConstants.AND,
),
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(),
)
@log(menu='Application', operate="Modify application access token",
get_operation_object= lambda r,k: get_application_operation_object((k.get('application_id')))
)
@has_permissions(PermissionConstants.APPLICATION_OVERVIEW_ACCESS.get_workspace_application_permission(),
PermissionConstants.APPLICATION_OVERVIEW_ACCESS.get_workspace_permission_workspace_manage_role(),
ViewPermission([RoleConstants.USER.get_workspace_role()],
[PermissionConstants.APPLICATION.get_workspace_application_permission()],
compare=CompareConstants.AND),
RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
def put(self, request: Request, workspace_id: str, application_id: str):
return result.success(
AccessTokenSerializer(data={'workspace_id': workspace_id, 'application_id': application_id}).edit(
request.data))
AccessTokenSerializer(data={"workspace_id": workspace_id, "application_id": application_id}).edit(
request.data
)
)

@extend_schema(
methods=['GET'],
methods=["GET"],
description=_("Get application access restriction information"),
summary=_("Get application access restriction information"),
operation_id=_("Get application access restriction information"), # type: ignore
parameters=ApplicationAccessTokenAPI.get_parameters(),
tags=[_('Application')] # type: ignore
tags=[_("Application")], # type: ignore
)
@has_permissions(
PermissionConstants.APPLICATION_READ.get_workspace_application_permission(),
PermissionConstants.APPLICATION_READ.get_workspace_permission_workspace_manage_role(),
ViewPermission(
[RoleConstants.USER.get_workspace_role()],
[PermissionConstants.APPLICATION.get_workspace_application_permission()],
compare=CompareConstants.AND,
),
RoleConstants.WORKSPACE_MANAGE.get_workspace_role(),
)
@has_permissions(PermissionConstants.APPLICATION_READ.get_workspace_application_permission(),
PermissionConstants.APPLICATION_READ.get_workspace_permission_workspace_manage_role(),
ViewPermission([RoleConstants.USER.get_workspace_role()],
[PermissionConstants.APPLICATION.get_workspace_application_permission()],
compare=CompareConstants.AND),
RoleConstants.WORKSPACE_MANAGE.get_workspace_role()
)
def get(self, request: Request, workspace_id: str, application_id: str):
return result.success(
AccessTokenSerializer(data={'workspace_id': workspace_id, 'application_id': application_id}).one())
AccessTokenSerializer(data={"workspace_id": workspace_id, "application_id": application_id}).one()
)

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The diff is mostly Black-style formatting. The only behavior-changing line I see is the PUT permission constant:

- PermissionConstants.APPLICATION_OVERVIEW_ACCESS.get_workspace_application_permission()
+ PermissionConstants.APPLICATION_OVERVIEW_PUBLIC.get_workspace_application_permission()

Main concern

  • Verify the permission change is intentional. This changes who can modify application access-token settings. For an endpoint described as “Modify application access restriction information”, APPLICATION_OVERVIEW_ACCESS seems more semantically correct than APPLICATION_OVERVIEW_PUBLIC. If this was accidental, revert it. If intentional, add/update permission tests and confirm it does not broaden access unexpectedly.
  • Also check consistency: GET uses APPLICATION_READ, while PUT now uses APPLICATION_OVERVIEW_PUBLIC. Make sure this matches the intended permission model.

Potential issues

  1. Direct QuerySet(model=Application) is unusual and risky
application_model = QuerySet(model=Application).filter(id=application_id).first()

This bypasses the model manager (Application.objects) and any custom manager/default filtering. Prefer:

def get_application_operation_object(application_id):
    application_model = Application.objects.filter(pk=application_id).values("name").first()
    return application_model or {}

This also avoids fetching all fields.

  1. Serializer validation is not visible
AccessTokenSerializer(data={...}).edit(request.data)

and

AccessTokenSerializer(data={...}).one()

If edit() / one() do not call is_valid(raise_exception=True) internally, validation may be skipped. Verify this; otherwise add explicit validation.

  1. Redundant parentheses in the lambda
get_operation_object=lambda r, k: get_application_operation_object((k.get("application_id")))

Simplify to:

get_operation_object=lambda r, k: get_application_operation_object(k.get("application_id"))
  1. Translated operation_id may be problematic
operation_id=_("Modify application access restriction information")

operation_id should be stable and unique for API schema/code generation. Translating it can produce locale-dependent or duplicate operation IDs. Prefer a fixed non-translated string. This is pre-existing, but worth flagging.

  1. Formatting changes are fine, but unrelated

Most quote changes, blank lines, and line wrapping are harmless. If the project uses Black, keep them. If not, avoid unrelated formatting churn in a functional change.

Summary

  • Must check: PUT permission constant changed from APPLICATION_OVERVIEW_ACCESS to APPLICATION_OVERVIEW_PUBLIC.
  • Recommended: use Application.objects.filter(...).values("name").first() instead of QuerySet(model=Application).
  • Recommended: remove redundant parentheses in get_operation_object.
  • Verify: serializer validation inside .edit() / .one().
  • Verify: translated operation_id does not break schema generation.

Loading