fix: 修复节点创建不能指定id问题，兼容api

apps/assets/models/user.py

@@ -150,17 +150,37 @@ class AuthMixin:
 
     def load_asset_special_auth(self, asset, username=''):
         """
+        AuthBook 的数据状态
+            | asset | systemuser | username |
+        1   |   *   |     *      |     x    |
+        2   |   *   |     x      |     *    |
+
+        当前 AuthBook 只有以上两种状态，systemuser 与 username 不会并存。
+        正常的资产与系统用户关联产生的是第1种状态，改密则产生第2种状态。改密之后
+        只有 username 而没有 systemuser 。
+
+        Freq: 关联同一资产的多个系统用户指定同一用户名时，修改用户密码会影响所有系统用户
+
+        这里有一个不对称的行为，同名系统用户密码覆盖
+        当有相同 username 的多个系统用户时，有改密动作之后，所有的同名系统用户都使用最后
+        一次改动，但如果没有发生过改密，同名系统用户使用的密码还是各自的。
+
         """
-        authbooks = list(AuthBook.objects.filter(asset=asset).filter(
-            Q(username=username) | Q(systemuser=self)
-        ))
-        if len(authbooks) == 0:
+        if username == '':
+            username = self.username
+
+        authbook = AuthBook.objects.filter(
+            asset=asset, username=username, systemuser__isnull=True
+        ).order_by('-date_created').first()
+
+        if not authbook:
+            authbook = AuthBook.objects.filter(
+                asset=asset, systemuser=self
+            ).order_by('-date_created').first()
+
+        if not authbook:
             return None
-        elif len(authbooks) == 1:
-            authbook = authbooks[0]
-        else:
-            authbooks.sort(key=lambda x: 1 if x.username == username else 0, reverse=True)
-            authbook = authbooks[0]
+
         authbook.load_auth()
         self.password = authbook.password
         self.private_key = authbook.private_key

apps/assets/serializers/account.py

@@ -6,6 +6,7 @@ from orgs.mixins.serializers import BulkOrgResourceModelSerializer
 
 from .base import AuthSerializerMixin
 from .utils import validate_password_contains_left_double_curly_bracket
+from common.utils.encode import ssh_pubkey_gen
 
 
 class AccountSerializer(AuthSerializerMixin, BulkOrgResourceModelSerializer):
@@ -40,6 +41,21 @@ class AccountSerializer(AuthSerializerMixin, BulkOrgResourceModelSerializer):
         }
         ref_name = 'AssetAccountSerializer'
 
+    def _validate_gen_key(self, attrs):
+        private_key = attrs.get('private_key')
+        if not private_key:
+            return attrs
+
+        password = attrs.get('passphrase')
+        username = attrs.get('username')
+        public_key = ssh_pubkey_gen(private_key, password=password, username=username)
+        attrs['public_key'] = public_key
+        return attrs
+
+    def validate(self, attrs):
+        attrs = self._validate_gen_key(attrs)
+        return attrs
+
     def get_protocols(self, v):
         return v.protocols.replace(' ', ', ')
 

apps/assets/serializers/base.py

@@ -2,7 +2,7 @@
 #
 from io import StringIO
 
-from django.utils.translation import ugettext as _
+from django.utils.translation import ugettext_lazy as _
 from rest_framework import serializers
 
 from common.utils import ssh_pubkey_gen, ssh_private_key_gen, validate_ssh_private_key

apps/assets/serializers/node.py

@@ -5,7 +5,6 @@ from django.utils.translation import ugettext as _
 from orgs.mixins.serializers import BulkOrgResourceModelSerializer
 from ..models import Asset, Node
 
-
 __all__ = [
     'NodeSerializer', "NodeAddChildrenSerializer",
     "NodeAssetsSerializer", "NodeTaskSerializer",
@@ -45,7 +44,6 @@ class NodeSerializer(BulkOrgResourceModelSerializer):
 
     def create(self, validated_data):
         full_value = validated_data.get('full_value')
-        value = validated_data.get('value')
 
         # 直接多层级创建
         if full_value:
@@ -53,7 +51,8 @@ class NodeSerializer(BulkOrgResourceModelSerializer):
         # 根据 value 在 root 下创建
         else:
             key = Node.org_root().get_next_child_key()
-            node = Node.objects.create(key=key, value=value)
+            validated_data['key'] = key
+            node = Node.objects.create(**validated_data)
         return node
 
 

apps/assets/tasks/push_system_user.py

@@ -4,10 +4,10 @@ from itertools import groupby
 from celery import shared_task
 from common.db.utils import get_object_if_need, get_objects
 from django.utils.translation import ugettext as _, gettext_noop
-from django.db.models import Empty, Q
+from django.db.models import Empty
 
 from common.utils import encrypt_password, get_logger
-from assets.models import SystemUser, Asset, AuthBook
+from assets.models import SystemUser, Asset
 from orgs.utils import org_aware_func, tmp_to_root_org
 from . import const
 from .utils import clean_ansible_task_hosts, group_asset_by_platform
@@ -178,6 +178,7 @@ def get_push_windows_system_user_tasks(system_user: SystemUser, username=None):
 
 def get_push_system_user_tasks(system_user, platform="unixlike", username=None):
     """
+    获取推送系统用户的 ansible 命令，跟资产无关
     :param system_user:
     :param platform:
     :param username: 当动态时，近推送某个
@@ -209,18 +210,10 @@ def push_system_user_util(system_user, assets, task_name, username=None):
     if not assets:
         return {}
 
+    # 资产按平台分类
     assets_sorted = sorted(assets, key=group_asset_by_platform)
     platform_hosts = groupby(assets_sorted, key=group_asset_by_platform)
 
-    def run_task(_tasks, _hosts):
-        if not _tasks:
-            return
-        task, created = update_or_create_ansible_task(
-            task_name=task_name, hosts=_hosts, tasks=_tasks, pattern='all',
-            options=const.TASK_OPTIONS, run_as_admin=True,
-        )
-        task.run()
-
     if system_user.username_same_with_user:
         if username is None:
             # 动态系统用户，但是没有指定 username
@@ -232,6 +225,15 @@ def push_system_user_util(system_user, assets, task_name, username=None):
         assert username is None, 'Only Dynamic user can assign `username`'
         usernames = [system_user.username]
 
+    def run_task(_tasks, _hosts):
+        if not _tasks:
+            return
+        task, created = update_or_create_ansible_task(
+            task_name=task_name, hosts=_hosts, tasks=_tasks, pattern='all',
+            options=const.TASK_OPTIONS, run_as_admin=True,
+        )
+        task.run()
+
     for platform, _assets in platform_hosts:
         _assets = list(_assets)
         if not _assets:
@@ -239,36 +241,11 @@ def push_system_user_util(system_user, assets, task_name, username=None):
         print(_("Start push system user for platform: [{}]").format(platform))
         print(_("Hosts count: {}").format(len(_assets)))
 
-        id_asset_map = {_asset.id: _asset for _asset in _assets}
-        asset_ids = id_asset_map.keys()
-        no_special_auth = []
-        special_auth_set = set()
-
-        auth_books = AuthBook.objects.filter(asset_id__in=asset_ids).filter(
-            Q(username__in=usernames) | Q(systemuser=system_user)
-        ).prefetch_related('systemuser')
-
-        for auth_book in auth_books:
-            auth_book.load_auth()
-            special_auth_set.add((auth_book.username, auth_book.asset_id))
-
-        for _username in usernames:
-            no_special_assets = []
-            for asset_id in asset_ids:
-                if (_username, asset_id) not in special_auth_set:
-                    no_special_assets.append(id_asset_map[asset_id])
-            if no_special_assets:
-                no_special_auth.append((_username, no_special_assets))
-
-        for _username, no_special_assets in no_special_auth:
-            tasks = get_push_system_user_tasks(system_user, platform, username=_username)
-            run_task(tasks, no_special_assets)
-
-        for auth_book in auth_books:
-            system_user._merge_auth(auth_book)
-            tasks = get_push_system_user_tasks(system_user, platform, username=auth_book.username)
-            asset = id_asset_map[auth_book.asset_id]
-            run_task(tasks, [asset])
+        for u in usernames:
+            for a in _assets:
+                system_user.load_asset_special_auth(a, u)
+                tasks = get_push_system_user_tasks(system_user, platform, username=u)
+                run_task(tasks, [a])
 
 
 @shared_task(queue="ansible")

apps/authentication/api/connection_token.py

@@ -298,9 +298,6 @@ class SecretDetailMixin:
             data['type'] = 'application'
             data.update(app_detail)
 
-        self.request.session['auth_backend'] = settings.AUTH_BACKEND_AUTH_TOKEN
-        post_auth_success.send(sender=self.__class__, user=user, request=self.request, login_type='T')
-
         serializer = self.get_serializer(data)
         return Response(data=serializer.data, status=200)
 

apps/common/utils/connection.py

@@ -23,6 +23,7 @@ class RedisPubSub:
     def __init__(self, ch, db=10):
         self.ch = ch
         self.redis = get_redis_client(db)
+        self.subscriber = None
 
     def subscribe(self):
         ps = self.redis.pubsub()
@@ -41,7 +42,9 @@ class RedisPubSub:
         :param handle: lambda item: do_something
         :return:
         """
+        self.close_handle_msg()
         sub = self.subscribe()
+        self.subscriber = sub
         msgs = sub.listen()
 
         try:
@@ -65,3 +68,7 @@ class RedisPubSub:
         except Exception as e:
             logger.error("Redis observer close error: ", e)
 
+    def close_handle_msg(self):
+        if self.subscriber:
+            self.subscriber.close()
+            self.subscriber = None

apps/notifications/ws.py

@@ -5,7 +5,7 @@ from channels.generic.websocket import JsonWebsocketConsumer
 from common.utils import get_logger
 from common.db.utils import safe_db_connection
 from .site_msg import SiteMessageUtil
-from .signals_handler import new_site_msg_chan
+from .signals_handler import NewSiteMsgSubPub
 
 logger = get_logger(__name__)
 
@@ -13,6 +13,10 @@ logger = get_logger(__name__)
 class SiteMsgWebsocket(JsonWebsocketConsumer):
     refresh_every_seconds = 10
 
+    def __init__(self, *args, **kwargs):
+        super(SiteMsgWebsocket, self).__init__(*args, **kwargs)
+        self.subscriber = None
+
     def connect(self):
         user = self.scope["user"]
         if user.is_authenticated:
@@ -23,6 +27,10 @@ class SiteMsgWebsocket(JsonWebsocketConsumer):
         else:
             self.close()
 
+    def disconnect(self, code):
+        if self.subscriber:
+            self.subscriber.close_handle_msg()
+
     def receive(self, text_data=None, bytes_data=None, **kwargs):
         data = json.loads(text_data)
         refresh_every_seconds = data.get('refresh_every_seconds')
@@ -56,4 +64,6 @@ class SiteMsgWebsocket(JsonWebsocketConsumer):
             if user_id in users:
                 ws.send_unread_msg_count()
 
-        new_site_msg_chan.keep_handle_msg(handle_new_site_msg_recv)
+        subscriber = NewSiteMsgSubPub()
+        self.subscriber = subscriber
+        subscriber.keep_handle_msg(handle_new_site_msg_recv)