jumpserver/apps/accounts/api/account/account.py

113 lines
4.2 KiB
Python

from django.shortcuts import get_object_or_404
from rest_framework.decorators import action
from rest_framework.generics import ListAPIView
from rest_framework.response import Response
from accounts import serializers
from accounts.filters import AccountFilterSet
from accounts.models import Account
from assets.models import Asset, Node
from common.permissions import UserConfirmation, ConfirmType
from common.views.mixins import RecordViewLogMixin
from orgs.mixins.api import OrgBulkModelViewSet
from rbac.permissions import RBACPermission
__all__ = [
'AccountViewSet', 'AccountSecretsViewSet',
'AccountHistoriesSecretAPI'
]
class AccountViewSet(OrgBulkModelViewSet):
model = Account
search_fields = ('username', 'asset__address', 'name')
filterset_class = AccountFilterSet
serializer_classes = {
'default': serializers.AccountSerializer,
}
rbac_perms = {
'partial_update': ['accounts.change_account'],
'su_from_accounts': 'accounts.view_account',
'username_suggestions': 'accounts.view_account',
}
@action(methods=['get'], detail=False, url_path='su-from-accounts')
def su_from_accounts(self, request, *args, **kwargs):
account_id = request.query_params.get('account')
asset_id = request.query_params.get('asset')
if account_id:
account = get_object_or_404(Account, pk=account_id)
accounts = account.get_su_from_accounts()
elif asset_id:
asset = get_object_or_404(Asset, pk=asset_id)
accounts = asset.accounts.all()
else:
accounts = []
accounts = self.filter_queryset(accounts)
serializer = serializers.AccountSerializer(accounts, many=True)
return Response(data=serializer.data)
@action(methods=['get'], detail=False, url_path='username-suggestions')
def username_suggestions(self, request, *args, **kwargs):
asset_ids = request.query_params.get('assets')
node_keys = request.query_params.get('keys')
username = request.query_params.get('username')
assets = Asset.objects.all()
if asset_ids:
assets = assets.filter(id__in=asset_ids.split(','))
if node_keys:
patten = Node.get_node_all_children_key_pattern(node_keys.split(','))
assets = assets.filter(nodes__key__regex=patten)
accounts = Account.objects.filter(asset__in=assets)
if username:
accounts = accounts.filter(username__icontains=username)
usernames = list(accounts.values_list('username', flat=True).distinct()[:10])
usernames.sort()
common = [i for i in usernames if i in usernames if i.lower() in ['root', 'admin', 'administrator']]
others = [i for i in usernames if i not in common]
usernames = common + others
return Response(data=usernames)
class AccountSecretsViewSet(RecordViewLogMixin, AccountViewSet):
"""
因为可能要导出所有账号,所以单独建立了一个 viewset
"""
serializer_classes = {
'default': serializers.AccountSecretSerializer,
}
http_method_names = ['get', 'options']
permission_classes = [RBACPermission, UserConfirmation.require(ConfirmType.MFA)]
rbac_perms = {
'list': 'accounts.view_accountsecret',
'retrieve': 'accounts.view_accountsecret',
}
class AccountHistoriesSecretAPI(RecordViewLogMixin, ListAPIView):
model = Account.history.model
serializer_class = serializers.AccountHistorySerializer
http_method_names = ['get', 'options']
permission_classes = [RBACPermission, UserConfirmation.require(ConfirmType.MFA)]
rbac_perms = {
'GET': 'accounts.view_accountsecret',
}
def get_object(self):
return get_object_or_404(Account, pk=self.kwargs.get('pk'))
def get_queryset(self):
account = self.get_object()
histories = account.history.all()
last_history = account.history.first()
if not last_history:
return histories
if account.secret == last_history.secret \
and account.secret_type == last_history.secret_type:
histories = histories.exclude(history_id=last_history.history_id)
return histories