perf: update gathered account

apps/accounts/automations/gather_accounts/database/mongodb/main.yml

@@ -15,7 +15,7 @@
         ssl_ca_certs: "{{ jms_asset.secret_info.ca_cert | default('') }}"
         ssl_certfile: "{{ jms_asset.secret_info.client_key | default('') }}"
         connection_options:
-          - tlsAllowInvalidHostnames: "{{ jms_asset.spec_info.allow_invalid_cert}}"
+          - tlsAllowInvalidHostnames: "{{ jms_asset.spec_info.allow_invalid_cert }}"
         filter: users
       register: db_info
 

apps/accounts/automations/gather_accounts/filter.py

@@ -1,5 +1,3 @@
-import re
-
 from django.utils import timezone
 
 __all__ = ['GatherAccountsFilter']
@@ -7,7 +5,6 @@ __all__ = ['GatherAccountsFilter']
 
 # TODO 后期会挪到 playbook 中
 class GatherAccountsFilter:
-
     def __init__(self, tp):
         self.tp = tp
 
@@ -29,26 +26,58 @@ class GatherAccountsFilter:
 
     @staticmethod
     def posix_filter(info):
-        username_pattern = re.compile(r'^(\S+)')
+        user_groups = info.pop('user_groups', [])
-        ip_pattern = re.compile(r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})')
+        username_groups = {}
-        login_time_pattern = re.compile(r'\w{3} \w{3}\s+\d{1,2} \d{2}:\d{2}:\d{2} \d{4}')
+        for line in user_groups:
-        result = {}
+            if ':' not in line:
-        for line in info:
-            usernames = username_pattern.findall(line)
-            username = ''.join(usernames)
-            if username:
-                result[username] = {}
-            else:
                 continue
-            ip_addrs = ip_pattern.findall(line)
+            username, groups = line.split(':', 1)
-            ip_addr = ''.join(ip_addrs)
+            username_groups[username.strip()] = groups.strip()
-            if ip_addr:
+
-                result[username].update({'address': ip_addr})
+        user_sudo = info.pop('user_sudo', [])
-            login_times = login_time_pattern.findall(line)
+        username_sudo = {}
-            if login_times:
+        for line in user_sudo:
-                datetime_str = login_times[0].split(' ', 1)[1] + " +0800"
+            if ':' not in line:
-                date = timezone.datetime.strptime(datetime_str, '%b %d %H:%M:%S %Y %z')
+                continue
-                result[username].update({'date': date})
+            username, sudo = line.split(':', 1)
+            if not sudo.strip():
+                continue
+            username_sudo[username.strip()] = sudo.strip()
+
+        user_authorized = info.pop('user_authorized', [])
+        username_authorized = {}
+        for line in user_authorized:
+            if ':' not in line:
+                continue
+            username, authorized = line.split(':', 1)
+            username_authorized[username.strip()] = authorized.strip()
+
+        result = {}
+        users = info.pop('users', '')
+        for line in users:
+            parts = line.split()
+            if len(parts) < 4:
+                continue
+
+            username = parts[0]
+            if not username:
+                continue
+            user = dict()
+            address = parts[2]
+            user['address_last_login'] = address
+            login_time = parts[3]
+
+            try:
+                login_date = timezone.datetime.fromisoformat(login_time)
+                user['date_last_login'] = login_date
+            except ValueError:
+                pass
+
+            user['groups'] = username_groups.get(username)
+            user['sudoers'] = username_sudo.get(username)
+            user['authorized_keys'] = username_authorized.get(username)
+
+            result[username] = user
         return result
 
     @staticmethod

apps/accounts/automations/gather_accounts/host/posix/main.yml

@@ -10,7 +10,7 @@
     - name: Gather posix account
       ansible.builtin.shell: |
         for user in {{ users.stdout_lines | join(" ") }}; do
-          k=$(last --time-format iso $user -1 | head -1 | grep -v ^$ | awk '{ print $0 }')
+          k=$(last -i --time-format iso -1 ${user} | head -1 | grep -v ^$ )
           if [ -n "$k" ]; then
             echo $k
           fi
@@ -24,7 +24,7 @@
         done
       register: user_groups
 
-    - name: Get sudo permissions
+    - name: Get sudoers
       ansible.builtin.shell: |
         for user in {{ users.stdout_lines | join(" ") }}; do
           echo "$user: $(grep "^$user " /etc/sudoers | tr '\n' ';' || echo '')"
@@ -43,22 +43,12 @@
         done
       register: user_authorized
 
-    - name: Display user groups
+    - set_fact:
-      ansible.builtin.debug:
+        info:
-        var: user_groups.stdout_lines
+          users: "{{ last_login.stdout_lines }}"
+          user_groups: "{{ user_groups.stdout_lines }}"
+          user_sudo: "{{ user_sudo.stdout_lines }}"
+          user_authorized: "{{ user_authorized.stdout_lines }}"
 
-    - name: Display sudo permissions
+    - debug:
-      ansible.builtin.debug:
+        var: info
-        var: user_sudo.stdout_lines
-
-    - name: Display authorized keys
-      ansible.builtin.debug:
-        var: user_authorized.stdout_lines
-
-    - name: Display last login
-      ansible.builtin.debug:
-        var: last_login.stdout_lines
-
-    - name: Define info by set_fact
-      ansible.builtin.set_fact:
-        var: last_login.stdout_lines

apps/accounts/automations/gather_accounts/manager.py

@@ -1,11 +1,11 @@
-import json
 from collections import defaultdict
 
 from accounts.const import AutomationTypes
-from accounts.models import GatheredAccount
+from accounts.models import GatheredAccount, Account, GatheredAccountDiff
 from assets.models import Asset
 from common.const import ConfirmOrIgnore
 from common.utils import get_logger
+from common.utils.strings import get_text_diff
 from orgs.utils import tmp_to_org
 from users.models import User
 from .filter import GatherAccountsFilter
@@ -16,13 +16,21 @@ logger = get_logger(__name__)
 
 
 class GatherAccountsManager(AccountBasePlaybookManager):
+    diff_items = ['authorized_keys', 'sudoers', 'groups']
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.host_asset_mapper = {}
         self.asset_account_info = {}
 
-        self.asset_username_mapper = defaultdict(set)
+        self.asset_usernames_mapper = defaultdict(set)
+        self.ori_asset_usernames = defaultdict(set)
+        self.ori_gathered_usernames = defaultdict(set)
+        self.ori_gathered_accounts_mapper = dict()
         self.is_sync_account = self.execution.snapshot.get('is_sync_account')
+        self.pending_add_accounts = []
+        self.pending_update_accounts = []
+        self.pending_add_diffs = []
 
     @classmethod
     def method_type(cls):
@@ -33,98 +41,182 @@ class GatherAccountsManager(AccountBasePlaybookManager):
         self.host_asset_mapper[host['name']] = asset
         return host
 
-    def filter_success_result(self, tp, result):
+    def _filter_success_result(self, tp, result):
         result = GatherAccountsFilter(tp).run(self.method_id_meta_mapper, result)
         return result
 
-    def generate_data(self, asset, result):
-        data = []
-        for username, info in result.items():
-            self.asset_username_mapper[str(asset.id)].add(username)
-            d = {'asset': asset, 'username': username, 'present': True}
-            if info.get('date'):
-                d['date_last_login'] = info['date']
-            if info.get('address'):
-                d['address_last_login'] = info['address'][:32]
-            data.append(d)
-        return data
-
-    def collect_asset_account_info(self, asset, result):
-        data = self.generate_data(asset, result)
-        self.asset_account_info[asset] = data
-
     @staticmethod
-    def get_nested_info(data, *keys):
+    def _get_nested_info(data, *keys):
         for key in keys:
             data = data.get(key, {})
             if not data:
                 break
         return data
 
+    def _collect_asset_account_info(self, asset, info):
+        result = self._filter_success_result(asset.type, info)
+        accounts = []
+        for username, info in result.items():
+            self.asset_usernames_mapper[asset].add(username)
+
+            d = {'asset': asset, 'username': username, 'present': True, **info}
+            if len(d['address_last_login']) > 32:
+                d['address_last_login'] = d['address_last_login'][:32]
+            accounts.append(d)
+        self.asset_account_info[asset] = accounts
+
+    def on_runner_failed(self,  runner, e):
+        raise e
+
     def on_host_success(self, host, result):
-        print("Result: ")
+        info = self._get_nested_info(result, 'debug', 'res', 'info')
-        print(json.dumps(result, indent=4))
-        print(">>>>>>>>>>>>>>>>.")
-        info = self.get_nested_info(result, 'debug', 'res', 'info')
         asset = self.host_asset_mapper.get(host)
         if asset and info:
-            result = self.filter_success_result(asset.type, info)
+            self._collect_asset_account_info(asset, info)
-            self.collect_asset_account_info(asset, result)
         else:
             print(f'\033[31m Not found {host} info \033[0m\n')
             
-    @staticmethod
+    def prefetch_origin_account_usernames(self):
-    def update_gather_accounts_status(asset):
         """
-        对于资产上不存在的账号，标识为待处理
+        提起查出来，避免每次 sql 查询
-        对于账号中不存在的，标识为待处理
+        :return:
         """
-        asset_accounts_usernames = asset.accounts.values_list('username', flat=True)
+        assets = self.asset_usernames_mapper.keys()
-        # 账号中不存在的标识为待处理的, 有可能是账号那边删除了
+        accounts = Account.objects.filter(asset__in=assets).values_list('asset', 'username')
-        GatheredAccount.objects \
-            .filter(asset=asset, present=True) \
-            .exclude(username__in=asset_accounts_usernames) \
-            .exclude(status=ConfirmOrIgnore.ignored) \
-            .update(status='')
 
+        for asset, username in accounts:
+            self.ori_asset_usernames[asset].add(username)
+
+        ga_accounts = GatheredAccount.objects.filter(asset__in=assets)
+        for account in ga_accounts:
+            self.ori_gathered_usernames[account.asset].add(account.username)
+            key = '{}_{}'.format(account.asset.id, account.username)
+            self.ori_gathered_accounts_mapper[key] = account
+
+    def update_gather_accounts_status(self, asset):
+        """
+        远端账号，收集中的账号，vault 中的账号。
+        要根据账号新增见啥，标识 收集账号的状态, 让管理员关注
+
+        远端账号 -> 收集账号 -> 特权账号
+        """
+        remote_users = self.asset_usernames_mapper[asset]
+        ori_users = self.ori_asset_usernames[asset]
+        ori_ga_users = self.ori_gathered_usernames[asset]
+
+        # 远端账号 比 收集账号多的
+        # 新增创建，不用处理状态
+
+        # 远端上 比 收集账号少的
+        # 标识 present=False, 标记为待处理
         # 远端资产上不存在的，标识为待处理，需要管理员介入
-        GatheredAccount.objects \
+        lost_users = ori_users - remote_users
-            .filter(asset=asset, present=False) \
+        if lost_users:
-            .exclude(status=ConfirmOrIgnore.ignored) \
+            GatheredAccount.objects \
-            .update(status='')
+                .filter(asset=asset, present=True) \
+                .exclude(status=ConfirmOrIgnore.ignored) \
+                .filter(username__in=lost_users) \
+                .update(status='', present=False)
+
+        # 收集的账号 比 账号列表多的, 有可能是账号中删掉了, 但这时候状态已经是 confirm 了
+        # 标识状态为 待处理, 让管理员去确认
+        ga_added_users = ori_ga_users - ori_users
+        if ga_added_users:
+            GatheredAccount.objects \
+                .filter(asset=asset) \
+                .exclude(status=ConfirmOrIgnore.ignored) \
+                .filter(username__in=ga_added_users) \
+                .update(status='')
+
+        # 收集的账号 比 账号列表少的
+        # 这个好像不不用对比，原始情况就这样
+
+        # 远端账号 比 账号列表少的
+        # 创建收集账号，标识 present=False, 状态待处理
+
+        # 远端账号 比 账号列表多的
+        # 正常情况, 不用处理，因为远端账号会创建到收集账号，收集账号再去对比
+
+    def batch_create_gathered_account(self, d, batch_size=20):
+        if d is None:
+            if self.pending_add_accounts:
+                GatheredAccount.objects.bulk_create(self.pending_add_accounts)
+                self.pending_add_accounts = []
+            return
+
+        gathered_account = GatheredAccount()
+        for k, v in d.items():
+            setattr(gathered_account, k, v)
+        self.pending_add_accounts.append(gathered_account)
+
+        if len(self.pending_add_accounts) > batch_size:
+            self.batch_create_gathered_account(None)
+
+    def batch_update_gathered_account(self, ori_account, d, batch_size=20):
+        if ori_account or d is None:
+            if self.pending_update_accounts:
+                GatheredAccount.objects.bulk_update(self.pending_update_accounts, ['status', 'present'])
+                self.pending_update_accounts = []
+
+            if self.pending_add_diffs:
+                GatheredAccountDiff.objects.bulk_create(self.pending_add_diffs)
+                self.pending_add_diffs = []
+            return
+
+        diff = {}
+        for item in self.diff_items:
+            ori = getattr(ori_account, item)
+            new = d.get(item, '')
+
+            if new != ori:
+                setattr(ori_account, item, new)
+                diff[item] = get_text_diff(ori, new)
+
+        if diff:
+            self.pending_update_accounts.append(ori_account)
+            for k, v in diff.items():
+                self.pending_add_diffs.append(
+                    GatheredAccountDiff(account=ori_account, item=k, diff=v)
+                )
+
+        if len(self.pending_update_accounts) > batch_size:
+            self.batch_update_gathered_account(None, None)
 
     def update_or_create_accounts(self):
-        for asset, data in self.asset_account_info.items():
+        for asset, accounts_data in self.asset_account_info.items():
             with (tmp_to_org(asset.org_id)):
                 gathered_accounts = []
-                # 把所有的设置为 present = False, 创建的时候如果有就会更新
+                for d in accounts_data:
-                GatheredAccount.objects.filter(asset=asset, present=True).update(present=False)
-                for d in data:
                     username = d['username']
-                    gathered_account, __ = GatheredAccount.objects.update_or_create(
+                    ori_account = self.ori_gathered_accounts_mapper.get('{}_{}'.format(asset.id, username))
-                        defaults=d, asset=asset, username=username,
+
-                    )
+                    if not ori_account:
-                    gathered_accounts.append(gathered_account)
+                        self.batch_create_gathered_account(d)
+                    else:
+                        self.batch_update_gathered_account(ori_account, d)
 
                 self.update_gather_accounts_status(asset)
                 GatheredAccount.sync_accounts(gathered_accounts, self.is_sync_account)
 
+        self.batch_create_gathered_account(None)
+        self.batch_update_gathered_account(None, None)
+
     def run(self, *args, **kwargs):
         super().run(*args, **kwargs)
-        users, change_info = self.generate_send_users_and_change_info()
+        self.prefetch_origin_account_usernames()
         self.update_or_create_accounts()
-        self.send_email_if_need(users, change_info)
+        # self.send_email_if_need()
 
     def generate_send_users_and_change_info(self):
         recipients = self.execution.recipients
-        if not self.asset_username_mapper or not recipients:
+        if not self.asset_usernames_mapper or not recipients:
             return None, None
 
         users = User.objects.filter(id__in=recipients)
         if not users.exists():
             return users, None
 
-        asset_ids = self.asset_username_mapper.keys()
+        asset_ids = self.asset_usernames_mapper.keys()
         assets = Asset.objects.filter(id__in=asset_ids).prefetch_related('accounts')
         gather_accounts = GatheredAccount.objects.filter(asset_id__in=asset_ids, present=True)
 
@@ -132,13 +224,13 @@ class GatherAccountsManager(AccountBasePlaybookManager):
         asset_id_username = list(assets.values_list('id', 'accounts__username'))
         asset_id_username.extend(list(gather_accounts.values_list('asset_id', 'username')))
 
-        system_asset_username_mapper = defaultdict(set)
+        system_asset_usernames_mapper = defaultdict(set)
         for asset_id, username in asset_id_username:
-            system_asset_username_mapper[str(asset_id)].add(username)
+            system_asset_usernames_mapper[str(asset_id)].add(username)
 
         change_info = defaultdict(dict)
-        for asset_id, usernames in self.asset_username_mapper.items():
+        for asset_id, usernames in self.asset_usernames_mapper.items():
-            system_usernames = system_asset_username_mapper.get(asset_id)
+            system_usernames = system_asset_usernames_mapper.get(asset_id)
             if not system_usernames:
                 continue
 
@@ -155,8 +247,8 @@ class GatherAccountsManager(AccountBasePlaybookManager):
 
         return users, dict(change_info)
 
-    @staticmethod
+    def send_email_if_need(self):
-    def send_email_if_need(users, change_info):
+        users, change_info = self.generate_send_users_and_change_info()
         if not users or not change_info:
             return
 

apps/accounts/migrations/0012_alter_gatheredaccount_status.py

@@ -0,0 +1,24 @@
+# Generated by Django 4.1.13 on 2024-10-31 08:39
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("accounts", "0011_remove_gatheredaccount_action_and_more"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="gatheredaccount",
+            name="status",
+            field=models.CharField(
+                blank=True,
+                choices=[("confirmed", "Confirmed"), ("ignored", "Ignored")],
+                default="",
+                max_length=32,
+                verbose_name="Status",
+            ),
+        ),
+    ]

apps/accounts/migrations/0013_remove_gatheredaccount_sudo_gatheredaccount_sudoers.py

@@ -0,0 +1,22 @@
+# Generated by Django 4.1.13 on 2024-10-31 08:48
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("accounts", "0012_alter_gatheredaccount_status"),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name="gatheredaccount",
+            name="sudo",
+        ),
+        migrations.AddField(
+            model_name="gatheredaccount",
+            name="sudoers",
+            field=models.TextField(default=False, verbose_name="Sudoers"),
+        ),
+    ]

apps/accounts/models/automations/gather_account.py

@@ -9,7 +9,7 @@ from common.utils.timezone import is_date_more_than
 from orgs.mixins.models import JMSOrgBaseModel
 from .base import AccountBaseAutomation
 
-__all__ = ['GatherAccountsAutomation', 'GatheredAccount']
+__all__ = ['GatherAccountsAutomation', 'GatheredAccount', 'GatheredAccountDiff']
 
 
 class GatheredAccountDiff(models.Model):
@@ -27,7 +27,7 @@ class GatheredAccount(JMSOrgBaseModel):
     address_last_login = models.CharField(max_length=39, default='', verbose_name=_("Address login"))
     status = models.CharField(max_length=32, default='', blank=True, choices=ConfirmOrIgnore.choices, verbose_name=_("Status"))
     authorized_keys = models.TextField(default='', blank=True, verbose_name=_("Authorized keys"))
-    sudo = models.TextField(default=False, verbose_name=_("Sudo"))
+    sudoers = models.TextField(default=False, verbose_name=_("Sudoers"))
     groups = models.TextField(default='', blank=True, verbose_name=_("Groups"))
 
     @property

apps/accounts/serializers/account/gathered_account.py

@@ -24,6 +24,7 @@ class GatheredAccountSerializer(BulkOrgResourceModelSerializer):
         fields = [
             'id', 'present', 'asset', 'username',
             'date_updated', 'address_last_login',
+            'groups', 'sudoers', 'authorized_keys',
             'date_last_login', 'status'
         ]
         read_only_fields = fields

apps/common/utils/strings.py

@@ -1,3 +1,4 @@
+import difflib
 import re
 
 
@@ -7,3 +8,10 @@ def no_special_chars(s):
 
 def safe_str(s):
     return s.encode('utf-8', errors='ignore').decode('utf-8')
+
+
+def get_text_diff(old_text, new_text):
+    diff = difflib.unified_diff(
+        old_text.splitlines(), new_text.splitlines(), lineterm=""
+    )
+    return "\n".join(diff)