jumpserver/apps/ops/models/playbook.py

import os.path
import uuid

from django.conf import settings
from django.db import models
from django.utils.translation import gettext_lazy as _
from private_storage.fields import PrivateFileField

from common.db.models import JMSBaseModel
from ops.const import CreateMethods, Scope
from ops.exception import PlaybookNoValidEntry

dangerous_keywords = (
    'hosts:localhost',
    'hosts:127.0.0.1',
    'hosts:::1',
    'delegate_to:localhost',
    'delegate_to:127.0.0.1',
    'delegate_to:::1',
    'local_action',
    'connection:local',
    'ansible_connection'
)




class Playbook(JMSBaseModel):
    id = models.UUIDField(default=uuid.uuid4, primary_key=True)
    name = models.CharField(max_length=128, verbose_name=_('Name'), null=True)
    path = PrivateFileField(upload_to='playbooks/')
    creator = models.ForeignKey('users.User', verbose_name=_("Creator"), on_delete=models.SET_NULL, null=True)
    comment = models.CharField(max_length=1024, default='', verbose_name=_('Comment'), null=True, blank=True)
    create_method = models.CharField(max_length=128, choices=CreateMethods.choices, default=CreateMethods.blank,
                                     verbose_name=_('CreateMethod'))
    scope = models.CharField(max_length=64, default=Scope.public, verbose_name=_('Scope'))
    vcs_url = models.CharField(max_length=1024, default='', verbose_name=_('VCS URL'), null=True, blank=True)

    def __str__(self):
        return self.name

    def check_dangerous_keywords(self):
        result = []
        for root, dirs, files in os.walk(self.work_dir):
            for f in files:
                try:
                    if str(f).endswith('.yml') or str(f).endswith('.yaml'):
                        lines = self.search_keywords(os.path.join(root, f))
                        if len(lines) > 0:
                            for line in lines:
                                result.append({'file': f, 'line': line[0], 'keyword': line[1]})
                # 遇到无法读取的文件，跳过
                except UnicodeEncodeError:
                    continue

        return result

    @staticmethod
    def search_keywords(file):
        result = []
        with open(file, 'r') as f:
            for line_num, line in enumerate(f):
                for keyword in dangerous_keywords:
                    clear_line = line.replace(' ', '') \
                        .replace('\n', '') \
                        .replace('\r', '') \
                        .replace('\t', '') \
                        .replace('\'', '') \
                        .replace('\"', '') \
                        .replace('\v', '')
                    if keyword in clear_line:
                        result.append((line_num, keyword))
            return result

    @property
    def entry(self):
        work_dir = self.work_dir
        valid_entry = ('main.yml', 'main.yaml', 'main')
        for f in os.listdir(work_dir):
            if f in valid_entry:
                return os.path.join(work_dir, f)
        raise PlaybookNoValidEntry

    @property
    def work_dir(self):
        work_dir = os.path.join(settings.DATA_DIR, "ops", "playbook", self.id.__str__())
        return work_dir

    class Meta:
        unique_together = [('name', 'creator')]
        verbose_name = _("Playbook")
        ordering = ['date_created']