You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
277 lines
9.2 KiB
277 lines
9.2 KiB
# -*- coding: utf-8 -*-
|
|
|
|
# Copyright 2015-2016 Mir Calculate. http://www.calculate-linux.org
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
from collections import namedtuple
|
|
|
|
import sys
|
|
|
|
_ = lambda x: x
|
|
from ..cl_lang import setLocalTranslate
|
|
|
|
setLocalTranslate('cl_lib3', sys.modules[__name__])
|
|
|
|
|
|
class AccountsError(Exception):
|
|
pass
|
|
|
|
|
|
class UnixUser():
|
|
"""
|
|
UnixUser = namedtuple("UnixUser", ["name", "password", "uid",
|
|
"gid", "gecos", "homedir", "shell"])
|
|
"""
|
|
|
|
def __init__(self, name, password, uid, gid, gecos, homedir, shell):
|
|
self.name = name
|
|
self.password = password
|
|
self.uid = uid
|
|
self.gid = gid
|
|
self.gecos = gecos
|
|
self.homedir = homedir
|
|
self.shell = shell
|
|
|
|
|
|
class Passwd():
|
|
def __init__(self, data):
|
|
self.users = [
|
|
UnixUser(*x.split(":"))
|
|
for x in data.split('\n')
|
|
if x.count(":") == 6
|
|
]
|
|
|
|
def equal_user(self, user1, user2):
|
|
"""
|
|
Проверка эквивалентности пользователей
|
|
:param user1:
|
|
:param user2:
|
|
:return:
|
|
"""
|
|
if user1.name == user2.name or user1.uid == user2.uid:
|
|
return True
|
|
return False
|
|
|
|
def join(self, passwd):
|
|
"""
|
|
Объединить пользователей
|
|
:param passwd:
|
|
:return:
|
|
"""
|
|
for user in self.users:
|
|
for backup_user in passwd.users:
|
|
if self.equal_user(user, backup_user):
|
|
user.gecos = backup_user.gecos
|
|
user.password = backup_user.password
|
|
user.homdedir = backup_user.homedir
|
|
user.shell = backup_user.shell
|
|
break
|
|
for backup_user in self.new_users(passwd):
|
|
self.users.append(backup_user)
|
|
|
|
def new_users(self, passwd):
|
|
"""
|
|
Список пользовтелей которые есть только в passwd
|
|
:param passwd:
|
|
:return:
|
|
"""
|
|
return [
|
|
user for user in passwd.users
|
|
if not any(self.equal_user(user, x) for x in self.users)]
|
|
|
|
def get_uid_map(self, passwd):
|
|
"""
|
|
Получить словарь соответствий одних uid другим
|
|
:param passwd:
|
|
:return:
|
|
"""
|
|
uid_map = {}
|
|
for user in self.users:
|
|
for user2 in passwd.users:
|
|
if user.name == user2.name and user.uid != user2.uid:
|
|
try:
|
|
uid_map[int(user.uid)] = int(user2.uid)
|
|
except ValueError:
|
|
raise AccountsError(
|
|
_("Wrong UIDs {uid1} {uid2}").format(
|
|
uid1=user.uid, uid2=user2.uid))
|
|
return uid_map
|
|
|
|
def format_user(self, user):
|
|
return "{0}:{1}:{2}:{3}:{4}:{5}:{6}".format(
|
|
user.name, user.password, user.uid,
|
|
user.gid, user.gecos, user.homedir, user.shell)
|
|
|
|
def write(self, f):
|
|
for user in self.users:
|
|
f.write("%s\n" % self.format_user(user))
|
|
|
|
|
|
class UnixGroup():
|
|
"""
|
|
UnixGroup = namedtuple("UnixGroup",
|
|
["name", "password", "gid", "user_list"])
|
|
"""
|
|
|
|
def __init__(self, name, password, gid, user_list):
|
|
self.name = name
|
|
self.password = password
|
|
self.gid = gid
|
|
self.user_list = user_list
|
|
|
|
|
|
class Group():
|
|
def __init__(self, data):
|
|
self.groups = [
|
|
UnixGroup(*x.split(":"))
|
|
for x in data.split('\n')
|
|
if x.count(":") == 3]
|
|
|
|
def equal_group(self, group1, group2):
|
|
"""
|
|
Проверка эквивалентности групп
|
|
:param group1:
|
|
:param group2:
|
|
:return:
|
|
"""
|
|
if group1.name == group2.name or group1.gid == group2.gid:
|
|
return True
|
|
return False
|
|
|
|
def new_groups(self, groupobj):
|
|
"""
|
|
Список групп которые есть только в groupobj
|
|
:param groupobj:
|
|
:return:
|
|
"""
|
|
return [
|
|
user for user in groupobj.groups
|
|
if not any(self.equal_group(user, x) for x in self.groups)]
|
|
|
|
def join(self, groupobj, keep_users=()):
|
|
"""
|
|
Объдинить группы
|
|
:param groupobj:
|
|
:param keep_users: список пользователей, чьи группы нужно сохранить
|
|
:return:
|
|
"""
|
|
for group in self.groups:
|
|
for backup_group in groupobj.groups:
|
|
if self.equal_group(group, backup_group):
|
|
group.password = backup_group.password
|
|
# backup_user_list = list(filter(None,
|
|
# backup_group.user_list.split(',')))
|
|
backup_user_list = [x for x in backup_group.user_list.split(',') if x]
|
|
|
|
# current_user_list = list(filter(None,
|
|
# group.user_list.split(',')))
|
|
current_user_list = [x for x in group.user_list.split(',') if x]
|
|
user_list = [
|
|
x for x in current_user_list
|
|
if x in keep_users and x not in backup_user_list]
|
|
new_user_list = backup_user_list + user_list
|
|
# не меняем список групп, если изменился только порядок
|
|
if set(current_user_list) != set(new_user_list):
|
|
group.user_list = ",".join(new_user_list)
|
|
break
|
|
for backup_group in self.new_groups(groupobj):
|
|
self.groups.append(backup_group)
|
|
|
|
def get_gid_map(self, groupobj):
|
|
"""
|
|
Получить словарь соответствий одних gid другим
|
|
:param groupobj:
|
|
:return:
|
|
"""
|
|
gid_map = {}
|
|
for group in self.groups:
|
|
for group2 in groupobj.groups:
|
|
if group.name == group2.name and group.gid != group2.gid:
|
|
try:
|
|
gid_map[int(group.gid)] = int(group2.gid)
|
|
except ValueError:
|
|
raise AccountsError(
|
|
_("Wrong GIDs {gid1} {gid2}").format(
|
|
gid1=group.gid, gid2=group2.gid))
|
|
return gid_map
|
|
|
|
def format_group(self, group):
|
|
return "{0}:{1}:{2}:{3}".format(
|
|
group.name, group.password, group.gid, group.user_list)
|
|
|
|
def write(self, f):
|
|
for group in self.groups:
|
|
f.write("%s\n" % self.format_group(group))
|
|
|
|
|
|
unix_shadow_fields = ["name", "password", "last_change",
|
|
"min_age", "max_age", "warn_period",
|
|
"inactive_period", "expire", "reserved"]
|
|
UnixShadow = namedtuple("UnixShadow", unix_shadow_fields)
|
|
|
|
|
|
class Shadow():
|
|
def __init__(self, data):
|
|
self.passwords = [
|
|
UnixShadow(*x.split(":"))
|
|
for x in data.split('\n')
|
|
if x.count(":") == len(unix_shadow_fields) - 1]
|
|
|
|
def equal_shadow(self, shadow1, shadow2):
|
|
"""
|
|
Проверка эквивалентности записи
|
|
:param shadow1:
|
|
:param shadow2:
|
|
:return:
|
|
"""
|
|
if shadow1.name == shadow2.name:
|
|
return True
|
|
return False
|
|
|
|
def changed_passwords(self, shadowobj):
|
|
"""
|
|
Список записей в которых будет восстановлен пароль
|
|
:param shadowobj:
|
|
:return:
|
|
"""
|
|
return [shadow
|
|
for shadow in shadowobj.passwords
|
|
for shadow2 in self.passwords
|
|
if (self.equal_shadow(shadow, shadow2)
|
|
and shadow.password != shadow2.password)]
|
|
|
|
def join(self, shadowobj):
|
|
def select_shadow(shadow):
|
|
for shadow2 in shadowobj.passwords:
|
|
if self.equal_shadow(shadow, shadow2):
|
|
return shadow2
|
|
else:
|
|
return shadow
|
|
|
|
self.passwords = [
|
|
select_shadow(shadow)
|
|
for shadow in self.passwords]
|
|
passwords = [
|
|
shadow
|
|
for shadow in shadowobj.passwords
|
|
if not any(self.equal_shadow(shadow, x)
|
|
for x in self.passwords)]
|
|
self.passwords.extend(passwords)
|
|
|
|
def format_shadow(self, shadow):
|
|
return ":".join(getattr(shadow, x) for x in unix_shadow_fields)
|
|
|
|
def write(self, f):
|
|
for shadow in self.passwords:
|
|
f.write("%s\n" % self.format_shadow(shadow))
|