jumpserver/apps/assets/models/base.py

135 lines
4.3 KiB
Python

# -*- coding: utf-8 -*-
#
import os
import uuid
from hashlib import md5
import sshpubkeys
from django.db import models
from django.utils.translation import ugettext_lazy as _
from django.conf import settings
from common.utils import get_signer, ssh_key_string_to_obj, ssh_key_gen
from common.validators import alphanumeric
from orgs.mixins import OrgModelMixin
from .utils import private_key_validator
signer = get_signer()
class AssetUser(OrgModelMixin):
id = models.UUIDField(default=uuid.uuid4, primary_key=True)
name = models.CharField(max_length=128, verbose_name=_('Name'))
username = models.CharField(max_length=32, blank=True, verbose_name=_('Username'), validators=[alphanumeric])
_password = models.CharField(max_length=256, blank=True, null=True, verbose_name=_('Password'))
_private_key = models.TextField(max_length=4096, blank=True, null=True, verbose_name=_('SSH private key'), validators=[private_key_validator, ])
_public_key = models.TextField(max_length=4096, blank=True, verbose_name=_('SSH public key'))
comment = models.TextField(blank=True, verbose_name=_('Comment'))
date_created = models.DateTimeField(auto_now_add=True)
date_updated = models.DateTimeField(auto_now=True)
created_by = models.CharField(max_length=128, null=True, verbose_name=_('Created by'))
@property
def password(self):
if self._password:
return signer.unsign(self._password)
else:
return None
@password.setter
def password(self, password_raw):
raise AttributeError("Using set_auth do that")
# self._password = signer.sign(password_raw)
@property
def private_key(self):
if self._private_key:
return signer.unsign(self._private_key)
@private_key.setter
def private_key(self, private_key_raw):
raise AttributeError("Using set_auth do that")
# self._private_key = signer.sign(private_key_raw)
@property
def private_key_obj(self):
if self._private_key:
key_str = signer.unsign(self._private_key)
return ssh_key_string_to_obj(key_str, password=self.password)
else:
return None
@property
def private_key_file(self):
if not self.private_key_obj:
return None
project_dir = settings.PROJECT_DIR
tmp_dir = os.path.join(project_dir, 'tmp')
key_str = signer.unsign(self._private_key)
key_name = '.' + md5(key_str.encode('utf-8')).hexdigest()
key_path = os.path.join(tmp_dir, key_name)
if not os.path.exists(key_path):
self.private_key_obj.write_private_key_file(key_path)
os.chmod(key_path, 0o400)
return key_path
@property
def public_key(self):
key = signer.unsign(self._public_key)
if key:
return key
else:
return None
@property
def public_key_obj(self):
if self.public_key:
try:
return sshpubkeys.SSHKey(self.public_key)
except TabError:
pass
return None
def set_auth(self, password=None, private_key=None, public_key=None):
update_fields = []
if password:
self._password = signer.sign(password)
update_fields.append('_password')
if private_key:
self._private_key = signer.sign(private_key)
update_fields.append('_private_key')
if public_key:
self._public_key = signer.sign(public_key)
update_fields.append('_public_key')
if update_fields:
self.save(update_fields=update_fields)
def clear_auth(self):
self._password = ''
self._private_key = ''
self._public_key = ''
self.save()
def auto_gen_auth(self):
password = str(uuid.uuid4())
private_key, public_key = ssh_key_gen(
username=self.username
)
self.set_auth(password=password,
private_key=private_key,
public_key=public_key)
def _to_secret_json(self):
"""Push system user use it"""
return {
'name': self.name,
'username': self.username,
'password': self.password,
'public_key': self.public_key,
'private_key': self.private_key_file,
}
class Meta:
abstract = True