public/gpupdate
1
0
Fork 1
mirror of https://github.com/altlinux/gpupdate.git synced 2025-03-22 18:50:15 +03:00
Code Releases Wiki Activity

Merge pull request #74 from altlinux/backend_refactoring

Browse Source 
Backend refactoring
This commit is contained in:
Evgeny Sinelnikov 2020-05-22 20:28:04 +04:00 committed by GitHub
commit 1332409dc2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 541 additions and 302 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
gpoa/frontend/frontend_manager.py
View File

@ -80,15 +80,15 @@ class frontend_manager:
self.sid = get_sid(self.storage.get_info('domain'), self.username, is_machine)
self.machine_appliers = dict({
'control': control_applier(self.storage),
'polkit': polkit_applier(self.storage),
'systemd': systemd_applier(self.storage),
'firefox': firefox_applier(self.storage, self.sid, self.username),
'chromium': chromium_applier(self.storage, self.sid, self.username),
'shortcuts': shortcut_applier(self.storage),
'gsettings': gsettings_applier(self.storage),
'cups': cups_applier(self.storage),
'package': package_applier(self.storage)
'control': control_applier(self.storage)
, 'polkit': polkit_applier(self.storage)
, 'systemd': systemd_applier(self.storage)
, 'firefox': firefox_applier(self.storage, self.sid, self.username)
, 'chromium': chromium_applier(self.storage, self.sid, self.username)
, 'shortcuts': shortcut_applier(self.storage)
, 'gsettings': gsettings_applier(self.storage)
, 'cups': cups_applier(self.storage)
#, 'package': package_applier(self.storage)
})
# User appliers are expected to work with user-writable
@ -107,15 +107,9 @@ class frontend_manager:
logging.error('Not sufficient privileges to run machine appliers')
return
logging.debug(slogm('Applying computer part of settings'))
self.machine_appliers['systemd'].apply()
self.machine_appliers['control'].apply()
self.machine_appliers['polkit'].apply()
self.machine_appliers['firefox'].apply()
self.machine_appliers['chromium'].apply()
self.machine_appliers['shortcuts'].apply()
self.machine_appliers['gsettings'].apply()
self.machine_appliers['cups'].apply()
#self.machine_appliers['package'].apply()
for applier_name, applier_object in self.machine_appliers.items():
logging.debug('Running machine applier {}'.format(applier_name))
applier_object.apply()
def user_apply(self):
'''

4
gpoa/gpt/drives.py
View File

@ -72,6 +72,10 @@ def read_drives(drives_file):
return drives
def merge_drives(storage, sid, drive_objects, policy_name):
for drive in drive_objects:
storage.add_drive(sid, drive, policy_name)
def json2drive(json_str):
json_obj = json.loads(json_str)
drive_obj = drivemap()

4
gpoa/gpt/envvars.py
View File

@ -28,6 +28,10 @@ def read_envvars(envvars_file):
return variables
def merge_envvars(storage, sid, envvars_objects, policy_name):
for envvar in envvar_objects:
pass
class envvar:
def __init__(self):
pass

4
gpoa/gpt/files.py
View File

@ -28,6 +28,10 @@ def read_files(filesxml):
return files
def merge_files(storage, sid, file_objects, policy_name):
for fileobj in file_objects:
pass
class fileentry:
def __init__(self):
pass

68
gpoa/gpt/folders.py
View File

@ -16,19 +16,79 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from enum import Enum
from util.xml import get_xml_root
class FileAction(Enum):
CREATE = 'C'
REPLACE = 'R'
def action_letter2enum(letter):
if letter in ['C', 'R']:
if letter == 'C': return FileAction.CREATE
if letter == 'R': return FileAction.REPLACE
return FileAction.CREATE
def action_enum2letter(enumitem):
return enumitem.value
def folder_int2bool(val):
value = val
if type(value) == str:
value = int(value)
if value == 0:
return True
return False
def read_folders(folders_file):
folders = list()
for fld in get_xml_root(folders_file):
fld_obj = folderentry()
props = fld.find('Properties')
fld_obj = folderentry(props.get('path'))
fld_obj.set_action(action_letter2enum(props.get('action')))
fld_obj.set_delete_folder(folder_int2bool(props.get('deleteFolder')))
fld_obj.set_delete_sub_folder(folder_int2bool(props.get('deleteSubFolders')))
fld_obj.set_delete_files(folder_int2bool(props.get('deleteFiles')))
folders.append(fld_obj)
return folders
class folderentry:
def __init__(self):
pass
def merge_folders(storage, sid, folder_objects, policy_name):
for folder in folder_objects:
storage.add_folder(sid, folder, policy_name)
class folderentry:
def __init__(self, path):
self.path = path
self.action = FileAction.CREATE
self.delete_folder = False
self.delete_sub_folder = False
self.delete_files = False
def set_action(self, action):
self.action = action
def set_delete_folder(self, del_bool):
self.delete_folder = del_bool
def set_delete_sub_folder(self, del_bool):
self.delete_sub_folder = del_bool
def set_delete_files(self, del_bool):
self.delete_files = del_bool

347
gpoa/gpt/gpt.py
View File

@ -18,18 +18,54 @@
import logging
import os
from pathlib import Path
from enum import Enum, unique
from samba.gp_parse.gp_pol import GPPolParser
from storage import registry_factory
from .shortcuts import read_shortcuts
from .services import read_services
from .printers import read_printers
from .inifiles import read_inifiles
from .folders import read_folders
from .files import read_files
from .envvars import read_envvars
from .drives import read_drives
from .polfile import (
read_polfile
, merge_polfile
)
from .shortcuts import (
read_shortcuts
, merge_shortcuts
)
from .services import (
read_services
, merge_services
)
from .printers import (
read_printers
, merge_printers
)
from .inifiles import (
read_inifiles
, merge_inifiles
)
from .folders import (
read_folders
, merge_folders
)
from .files import (
read_files
, merge_files
)
from .envvars import (
read_envvars
, merge_envvars
)
from .drives import (
read_drives
, merge_drives
)
from .tasks import (
read_tasks
, merge_tasks
)
import util
import util.preg
from util.paths import (
@ -39,6 +75,68 @@ from util.paths import (
)
from util.logging import slogm
@unique
class FileType(Enum):
PREG = 'registry.pol'
SHORTCUTS = 'shortcuts.xml'
FOLDERS = 'folders.xml'
FILES = 'files.xml'
DRIVES = 'drives.xml'
SCHEDULEDTASKS = 'scheduledtasks.xml'
ENVIRONMENTVARIABLES = 'environmentvariables.xml'
INIFILES = 'inifiles.xml'
SERVICES = 'services.xml'
def get_preftype(path_to_file):
fpath = Path(path_to_file)
if fpath.exists():
file_name = fpath.name.lower()
for item in FileType:
if file_name == item.value:
return item
return None
def pref_parsers():
parsers = dict()
parsers[FileType.PREG] = read_polfile
parsers[FileType.SHORTCUTS] = read_shortcuts
parsers[FileType.FOLDERS] = read_folders
parsers[FileType.FILES] = read_files
parsers[FileType.DRIVES] = read_drives
parsers[FileType.SCHEDULEDTASKS] = read_tasks
parsers[FileType.ENVIRONMENTVARIABLES] = read_envvars
parsers[FileType.INIFILES] = read_inifiles
parsers[FileType.SERVICES] = read_services
return parsers
def get_parser(preference_type):
parsers = pref_parsers()
return parsers[preference_type]
def pref_mergers():
mergers = dict()
mergers[FileType.PREG] = merge_polfile
mergers[FileType.SHORTCUTS] = merge_shortcuts
mergers[FileType.FOLDERS] = merge_folders
mergers[FileType.FILES] = merge_files
mergers[FileType.DRIVES] = merge_drives
mergers[FileType.SCHEDULEDTASKS] = merge_tasks
mergers[FileType.ENVIRONMENTVARIABLES] = merge_envvars
mergers[FileType.INIFILES] = merge_inifiles
mergers[FileType.SERVICES] = merge_services
return mergers
def get_merger(preference_type):
mergers = pref_mergers()
return mergers[preference_type]
class gpt:
__user_policy_mode_key = 'Software\\Policies\\Microsoft\\Windows\\System\\UserPolicyMode'
@ -46,13 +144,8 @@ class gpt:
self.path = gpt_path
self.sid = sid
self.storage = registry_factory('registry')
self._scan_gpt()
self.name = ''
def _scan_gpt(self):
'''
Collect the data from the specified GPT on file system (cached
by Samba).
'''
self.guid = self.path.rpartition('/')[2]
self.name = ''
if 'default' == self.guid:
@ -60,13 +153,30 @@ class gpt:
self._machine_path = find_dir(self.path, 'Machine')
self._user_path = find_dir(self.path, 'User')
self._machine_prefs = find_dir(self._machine_path, 'Preferences')
self._user_prefs = find_dir(self._user_path, 'Preferences')
logging.debug(slogm('Looking for machine part of GPT {}'.format(self.guid)))
self._find_machine()
logging.debug(slogm('Looking for user part of GPT {}'.format(self.guid)))
self._find_user()
self.settings_list = [
'shortcuts'
, 'drives'
, 'environmentvariables'
, 'printers'
, 'folders'
, 'files'
, 'inifiles'
, 'services'
, 'scheduledtasks'
]
self.settings = dict()
self.settings['machine'] = dict()
self.settings['user'] = dict()
self.settings['machine']['regpol'] = find_file(self._machine_path, 'registry.pol')
self.settings['user']['regpol'] = find_file(self._user_path, 'registry.pol')
for setting in self.settings_list:
machine_preffile = find_preffile(self._machine_path, setting)
user_preffile = find_preffile(self._user_path, setting)
logging.debug('Looking for {} in machine part of GPT {}: {}'.format(setting, self.name, machine_preffile))
self.settings['machine'][setting] = machine_preffile
logging.debug('Looking for {} in user part of GPT {}: {}'.format(setting, self.name, user_preffile))
self.settings['user'][setting] = user_preffile
def set_name(self, name):
'''
@ -89,163 +199,41 @@ class gpt:
return upm
def _find_user(self):
self._user_regpol = self._find_regpol('user')
self._user_shortcuts = self._find_shortcuts('user')
self._user_drives = self._find_drives('user')
def _find_machine(self):
self._machine_regpol = self._find_regpol('machine')
self._machine_shortcuts = self._find_shortcuts('machine')
self._machine_drives = self._find_drives('machine')
def _find_regpol(self, part):
'''
Find Registry.pol files.
'''
search_path = self._machine_path
if 'user' == part:
search_path = self._user_path
if not search_path:
return None
return find_file(search_path, 'registry.pol')
def _find_shortcuts(self, part):
'''
Find Shortcuts.xml files.
'''
shortcuts_dir = find_dir(self._machine_prefs, 'Shortcuts')
shortcuts_file = find_file(shortcuts_dir, 'shortcuts.xml')
if 'user' == part:
shortcuts_dir = find_dir(self._user_prefs, 'Shortcuts')
shortcuts_file = find_file(shortcuts_dir, 'shortcuts.xml')
return shortcuts_file
def _find_envvars(self, part):
'''
Find EnvironmentVariables.xml files.
'''
search_path = os.path.join(self._machine_path, 'Preferences', 'EnvironmentVariables')
if 'user' == part:
search_path = os.path.join(self._user_path, 'Preferences', 'EnvironmentVariables')
if not search_path:
return None
return find_file(search_path, 'environmentvariables.xml')
def _find_drives(self, part):
'''
Find Drives.xml files.
'''
drives_dir = find_dir(self._machine_prefs, 'Drives')
drives_file = find_file(drives_dir, 'drives.xml')
if 'user' == part:
drives_dir = find_dir(self._user_prefs, 'Drives')
drives_file = find_file(drives_dir, 'drives.xml')
return drives_file
def _find_printers(self, part):
'''
Find Printers.xml files.
'''
search_path = os.path.join(self._machine_path, 'Preferences', 'Printers')
if 'user' == part:
search_path = os.path.join(self._user_path, 'Preferences', 'Printers')
if not search_path:
return None
return find_file(search_path, 'printers.xml')
def _merge_shortcuts(self):
shortcuts = list()
if self.sid == self.storage.get_info('machine_sid'):
shortcuts = read_shortcuts(self._machine_shortcuts)
else:
shortcuts = read_shortcuts(self._user_shortcuts)
for sc in shortcuts:
self.storage.add_shortcut(self.sid, sc)
def _merge_drives(self):
drives = list()
if self.sid == self.storage.get_info('machine_sid'):
drives = read_drives(self._machine_drives)
else:
drives = read_drives(self._user_drives)
for drv in drives:
self.storage.add_drive(self.sid, drv)
def merge(self):
'''
Merge machine and user (if sid provided) settings to storage.
'''
if self.sid == self.storage.get_info('machine_sid'):
# Merge machine settings to registry if possible
if self._machine_regpol:
logging.debug(slogm('Merging machine settings from {}'.format(self._machine_regpol)))
util.preg.merge_polfile(self._machine_regpol)
if self._user_regpol:
logging.debug(slogm('Merging machine(user) settings from {}'.format(self._machine_regpol)))
util.preg.merge_polfile(self._user_regpol, self.sid)
if self._machine_shortcuts:
logging.debug(slogm('Merging machine shortcuts from {}'.format(self._machine_shortcuts)))
self._merge_shortcuts()
if self._machine_drives:
logging.debug(slogm('Merging machine drives from {}'.format(self._machine_drives)))
self._merge_drives()
for preference_name, preference_path in self.settings['machine'].items():
if preference_path:
preference_type = get_preftype(preference_path)
logstring = 'Reading and merging {} for {}'.format(preference_type.value, self.sid)
logging.debug(logstring)
preference_parser = get_parser(preference_type)
preference_merger = get_merger(preference_type)
preference_objects = preference_parser(preference_path)
preference_merger(self.storage, self.sid, preference_objects, self.name)
if self.settings['user']['regpol']:
logging.debug(slogm('Merging machine(user) settings from {}'.format(self.settings['machine']['regpol'])))
util.preg.merge_polfile(self.settings['user']['regpol'], sid=self.sid, policy_name=self.name)
if self.settings['machine']['regpol']:
logging.debug(slogm('Merging machine settings from {}'.format(self.settings['machine']['regpol'])))
util.preg.merge_polfile(self.settings['machine']['regpol'], policy_name=self.name)
else:
# Merge user settings if UserPolicyMode set accordingly
# and user settings (for HKCU) are exist.
policy_mode = upm2str(self.get_policy_mode())
if 'Merge' == policy_mode or 'Not configured' == policy_mode:
if self._user_regpol:
logging.debug(slogm('Merging user settings from {} for {}'.format(self._user_regpol, self.sid)))
util.preg.merge_polfile(self._user_regpol, self.sid)
if self._user_shortcuts:
logging.debug(slogm('Merging user shortcuts from {} for {}'.format(self._user_shortcuts, self.sid)))
self._merge_shortcuts()
if self._user_drives:
logging.debug(slogm('Merging user drives from {} for {}'.format(self._user_drives, self.sid)))
self._merge_drives()
def __str__(self):
template = '''
GUID: {}
Name: {}
For SID: {}
Machine part: {}
Machine Registry.pol: {}
Machine Shortcuts.xml: {}
User part: {}
User Registry.pol: {}
User Shortcuts.xml: {}
'''
result = template.format(
self.guid,
self.name,
self.sid,
self._machine_path,
self._machine_regpol,
self._machine_shortcuts,
self._user_path,
self._user_regpol,
self._user_shortcuts,
)
return result
for preference_name, preference_path in self.settings['user'].items():
if preference_path:
preference_type = get_preftype(preference_path)
logstring = 'Reading and merging {} for {}'.format(preference_type.value, self.sid)
logging.debug(logstring)
preference_parser = get_parser(preference_type)
preference_merger = get_merger(preference_type)
preference_objects = preference_parser(preference_path)
preference_merger(self.storage, self.sid, preference_objects, self.name)
def find_dir(search_path, name):
'''
@ -290,6 +278,33 @@ def find_file(search_path, name):
return None
def find_preferences(search_path):
'''
Find 'Preferences' directory
'''
if not search_path:
return None
return find_dir(search_path, 'Preferences')
def find_preffile(search_path, prefname):
'''
Find file with path like Preferences/prefname/prefname.xml
'''
# Look for 'Preferences' directory
prefdir = find_preferences(search_path)
if not prefdir:
return None
# Then search for preference directory
pref_dir = find_dir(prefdir, prefname)
file_name = '{}.xml'.format(prefname)
# And then try to find the corresponding file.
pref_file = find_file(pref_dir, file_name)
return pref_file
def lp2gpt():
'''
Convert local-policy to full-featured GPT.

4
gpoa/gpt/inifiles.py
View File

@ -28,6 +28,10 @@ def read_inifiles(inifiles_file):
return inifiles
def merge_inifiles(storage, sid, inifile_objects, policy_name):
for inifile in inifile_objects:
pass
def inifile():
def __init__(self):
pass

32
gpoa/gpt/polfile.py Normal file
View File

@ -0,0 +1,32 @@
#
# GPOA - GPO Applier for Linux
#
# Copyright (C) 2019-2020 BaseALT Ltd.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from util.preg import (
load_preg
)
def read_polfile(filename):
return load_preg(filename).entries
def merge_polfile(storage, sid, policy_objects, policy_name):
for entry in policy_objects:
if not sid:
storage.add_hklm_entry(entry, policy_name)
else:
storage.add_hkcu_entry(entry, sid, policy_name)

4
gpoa/gpt/printers.py
View File

@ -41,6 +41,10 @@ def read_printers(printers_file):
return printers
def merge_printers(storage, sid, printer_objects, policy_name):
for device in printer_objects:
pass
def json2printer(json_str):
'''
Build printer object out of string-serialized JSON.

4
gpoa/gpt/services.py
View File

@ -39,6 +39,10 @@ def read_services(service_file):
return services
def merge_services(storage, sid, service_objects, policy_name):
for srv in service_objects:
pass
class service:
def __init__(self, name):
self.unit = name

4
gpoa/gpt/shortcuts.py
View File

@ -89,6 +89,10 @@ def read_shortcuts(shortcuts_file):
return shortcuts
def merge_shortcuts(storage, sid, shortcut_objects, policy_name):
for shortcut in shortcut_objects:
storage.add_shortcut(sid, shortcut, policy_name)
def json2sc(json_str):
'''
Build shortcut out of string-serialized JSON

25
gpoa/gpt/tasks.py Normal file
View File

@ -0,0 +1,25 @@
#
# GPOA - GPO Applier for Linux
#
# Copyright (C) 2019-2020 BaseALT Ltd.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
def read_tasks(filename):
pass
def merge_tasks(storage, sid, task_objects, policy_name):
for task in task_objects:
pass

89
gpoa/storage/record_types.py
View File

@ -20,52 +20,131 @@ class samba_preg(object):
'''
Object mapping representing HKLM entry (registry key without SID)
'''
def __init__(self, preg_obj):
def __init__(self, preg_obj, policy_name):
self.policy_name = policy_name
self.hive_key = '{}\\{}'.format(preg_obj.keyname, preg_obj.valuename)
self.type = preg_obj.type
self.data = preg_obj.data
def update_fields(self):
fields = dict()
fields['policy_name'] = self.policy_name
fields['type'] = self.type
fields['data'] = self.data
return fields
class samba_hkcu_preg(object):
'''
Object mapping representing HKCU entry (registry key with SID)
'''
def __init__(self, sid, preg_obj):
def __init__(self, sid, preg_obj, policy_name):
self.sid = sid
self.policy_name = policy_name
self.hive_key = '{}\\{}'.format(preg_obj.keyname, preg_obj.valuename)
self.type = preg_obj.type
self.data = preg_obj.data
def update_fields(self):
fields = dict()
fields['policy_name'] = self.policy_name
fields['type'] = self.type
fields['data'] = self.data
return fields
class ad_shortcut(object):
'''
Object mapping representing Windows shortcut.
'''
def __init__(self, sid, sc):
def __init__(self, sid, sc, policy_name):
self.sid = sid
self.policy_name = policy_name
self.path = sc.dest
self.shortcut = sc.to_json()
def update_fields(self):
fields = dict()
fields['policy_name'] = self.policy_name
fields['path'] = self.path
fields['shortcut'] = self.shortcut
return fields
class info_entry(object):
def __init__(self, name, value):
self.name = name
self.value = value
def update_fields(self):
fields = dict()
fields['value'] = self.value
return fields
class printer_entry(object):
'''
Object mapping representing Windows printer of some type.
'''
def __init__(self, sid, pobj):
def __init__(self, sid, pobj, policy_name):
self.sid = sid
self.policy_name = policy_name
self.name = pobj.name
self.printer = pobj.to_json()
def update_fields(self):
fields = dict()
fields['policy_name'] = self.policy_name
fields['name'] = self.name
fields['printer'] = self.printer
return fields
class drive_entry(object):
'''
Object mapping representing Samba share bound to drive letter
'''
def __init__(self, sid, dobj):
def __init__(self, sid, dobj, policy_name):
self.sid = sid
self.policy_name = policy_name
self.login = dobj.login
self.password = dobj.password
self.dir = dobj.dir
self.path = dobj.path
def update_fields(self):
fields = dict()
fields['policy_name'] = self.policy_name
fields['login'] = self.login
fields['password'] = self.password
fields['dir'] = self.dir
fields['path'] = self.path
return fields
class folder_entry(object):
'''
Object mapping representing file system directory
'''
def __init__(self, sid, fobj, policy_name):
self.sid = sid
self.policy_name = policy_name
self.path = fobj.path
self.action = fobj.action.value
self.delete_folder = str(fobj.delete_folder)
self.delete_sub_folder = str(fobj.delete_sub_folder)
self.delete_files = str(fobj.delete_files)
def update_fields(self):
'''
Return list of fields to update
'''
fields = dict()
fields['policy_name'] = self.policy_name
fields['action'] = self.action
fields['delete_folder'] = self.delete_folder
fields['delete_sub_folder'] = self.delete_sub_folder
fields['delete_files'] = self.delete_files
return fields

218
gpoa/storage/sqlite_registry.py
View File

@ -43,6 +43,7 @@ from .record_types import (
, info_entry
, printer_entry
, drive_entry
, folder_entry
)
class sqlite_registry(registry):
@ -62,51 +63,69 @@ class sqlite_registry(registry):
Column('value', String(65536))
)
self.__hklm = Table(
'HKLM',
self.__metadata,
Column('id', Integer, primary_key=True),
Column('hive_key', String(65536), unique=True),
Column('type', Integer),
Column('data', String)
'HKLM'
, self.__metadata
, Column('id', Integer, primary_key=True)
, Column('hive_key', String(65536), unique=True)
, Column('policy_name', String)
, Column('type', Integer)
, Column('data', String)
)
self.__hkcu = Table(
'HKCU',
self.__metadata,
Column('id', Integer, primary_key=True),
Column('sid', String),
Column('hive_key', String(65536)),
Column('type', Integer),
Column('data', String),
UniqueConstraint('sid', 'hive_key')
'HKCU'
, self.__metadata
, Column('id', Integer, primary_key=True)
, Column('sid', String)
, Column('hive_key', String(65536))
, Column('policy_name', String)
, Column('type', Integer)
, Column('data', String)
, UniqueConstraint('sid', 'hive_key')
)
self.__shortcuts = Table(
'Shortcuts',
self.__metadata,
Column('id', Integer, primary_key=True),
Column('sid', String),
Column('path', String),
Column('shortcut', String),
UniqueConstraint('sid', 'path')
'Shortcuts'
, self.__metadata
, Column('id', Integer, primary_key=True)
, Column('sid', String)
, Column('path', String)
, Column('policy_name', String)
, Column('shortcut', String)
, UniqueConstraint('sid', 'path')
)
self.__printers = Table(
'Printers',
self.__metadata,
Column('id', Integer, primary_key=True),
Column('sid', String),
Column('name', String),
Column('printer', String),
UniqueConstraint('sid', 'name')
'Printers'
, self.__metadata
, Column('id', Integer, primary_key=True)
, Column('sid', String)
, Column('name', String)
, Column('policy_name', String)
, Column('printer', String)
, UniqueConstraint('sid', 'name')
)
self.__drives = Table(
'Drives',
self.__metadata,
Column('id', Integer, primary_key=True),
Column('sid', String),
Column('login', String),
Column('password', String),
Column('dir', String),
Column('path', String),
UniqueConstraint('sid', 'dir')
'Drives'
, self.__metadata
, Column('id', Integer, primary_key=True)
, Column('sid', String)
, Column('login', String)
, Column('password', String)
, Column('dir', String)
, Column('policy_name', String)
, Column('path', String)
, UniqueConstraint('sid', 'dir')
)
self.__folders = Table(
'Folders'
, self.__metadata
, Column('id', Integer, primary_key=True)
, Column('sid', String)
, Column('path', String)
, Column('policy_name', String)
, Column('action', String)
, Column('delete_folder', String)
, Column('delete_sub_folder', String)
, Column('delete_files', String)
, UniqueConstraint('sid', 'path')
)
self.__metadata.create_all(self.db_cnt)
Session = sessionmaker(bind=self.db_cnt)
@ -118,6 +137,7 @@ class sqlite_registry(registry):
mapper(ad_shortcut, self.__shortcuts)
mapper(printer_entry, self.__printers)
mapper(drive_entry, self.__drives)
mapper(folder_entry, self.__folders)
except:
pass
#logging.error('Error creating mapper')
@ -134,75 +154,69 @@ class sqlite_registry(registry):
try:
self._add(row)
except:
update_obj = dict({ 'value': row.value })
(self
.db_session.query(info_entry)
.filter(info_entry.name == row.name)
.update(update_obj))
.update(row.update_fields()))
self.db_session.commit()
def _hklm_upsert(self, row):
try:
self._add(row)
except:
update_obj = dict({'type': row.type, 'data': row.data })
(self
.db_session
.query(samba_preg)
.filter(samba_preg.hive_key == row.hive_key)
.update(update_obj))
.update(row.update_fields()))
self.db_session.commit()
def _hkcu_upsert(self, row):
try:
self._add(row)
except:
update_obj = dict({'type': row.type, 'data': row.data })
except Exception as exc:
(self
.db_session
.query(samba_preg)
.filter(samba_hkcu_preg.sid == row.sid)
.filter(samba_hkcu_preg.hive_key == row.hive_key)
.update(update_obj))
.update(row.update_fields()))
self.db_session.commit()
def _shortcut_upsert(self, row):
try:
self._add(row)
except:
update_obj = dict({ 'shortcut': row.shortcut })
(self
.db_session
.query(ad_shortcut)
.filter(ad_shortcut.sid == row.sid)
.filter(ad_shortcut.path == row.path)
.update(update_obj))
.update(row.update_fields()))
self.db_session.commit()
def _printer_upsert(self, row):
try:
self._add(row)
except:
update_obj = dict({ 'printer': row.printer })
(self
.db_session
.query(printer_entry)
.filter(printer_entry.sid == row.sid)
.filter(printer_entry.name == row.name)
.update(update_obj))
.update(row.update_fields()))
self.db_session.commit()
def _drive_upsert(self, row):
try:
self._add(row)
except:
update_obj = dict({ 'dir': row.dir, 'path': row.path, 'login': row.login, 'password': row.password })
(self
.db_session
.query(drive_entry)
.filter(drive_entry.sid == row.sid)
.filter(drive_entry.dir == row.dir)
.update(update_obj))
.update(row.update_fields()))
self.db_session.commit()
def set_info(self, name, value):
@ -210,71 +224,87 @@ class sqlite_registry(registry):
logging.debug(slogm('Setting info {}:{}'.format(name, value)))
self._info_upsert(ientry)
def add_hklm_entry(self, preg_entry):
def add_hklm_entry(self, preg_entry, policy_name):
'''
Write PReg entry to HKEY_LOCAL_MACHINE
'''
pentry = samba_preg(preg_entry)
pentry = samba_preg(preg_entry, policy_name)
if not pentry.hive_key.rpartition('\\')[2].startswith('**'):
self._hklm_upsert(pentry)
else:
logging.warning(slogm('Skipping branch deletion key: {}'.format(pentry.hive_key)))
def add_hkcu_entry(self, preg_entry, sid):
def add_hkcu_entry(self, preg_entry, sid, policy_name):
'''
Write PReg entry to HKEY_CURRENT_USER
'''
hkcu_pentry = samba_hkcu_preg(sid, preg_entry)
hkcu_pentry = samba_hkcu_preg(sid, preg_entry, policy_name)
if not hkcu_pentry.hive_key.rpartition('\\')[2].startswith('**'):
logging.debug(slogm('Adding HKCU entry for {}'.format(sid)))
self._hkcu_upsert(hkcu_pentry)
else:
logging.warning(slogm('Skipping branch deletion key: {}'.format(hkcu_pentry.hive_key)))
def add_shortcut(self, sid, sc_obj):
def add_shortcut(self, sid, sc_obj, policy_name):
'''
Store shortcut information in the database
'''
sc_entry = ad_shortcut(sid, sc_obj)
sc_entry = ad_shortcut(sid, sc_obj, policy_name)
logging.debug(slogm('Saving info about {} link for {}'.format(sc_entry.path, sid)))
self._shortcut_upsert(sc_entry)
def add_printer(self, sid, pobj):
def add_printer(self, sid, pobj, policy_name):
'''
Store printer configuration in the database
'''
prn_entry = printer_entry(sid, pobj)
prn_entry = printer_entry(sid, pobj, policy_name)
logging.debug(slogm('Saving info about printer {} for {}'.format(prn_entry.name, sid)))
self._printer_upsert(prn_entry)
def add_drive(self, sid, dobj):
drv_entry = drive_entry(sid, dobj)
def add_drive(self, sid, dobj, policy_name):
drv_entry = drive_entry(sid, dobj, policy_name)
logging.debug(slogm('Saving info about drive mapping {} for {}'.format(drv_entry.path, sid)))
self._drive_upsert(drv_entry)
def get_shortcuts(self, sid):
def add_folder(self, sid, fobj, policy_name):
fld_entry = folder_entry(sid, fobj, policy_name)
logstring = 'Saving info about folder {} for {}'.format(fld_entry.path, sid)
logging.debug(logstring)
try:
self._add(fld_entry)
except Exception as exc:
(self
._filter_sid_obj(folder_entry, sid)
.filter(folder_entry.path == fld_entry.path)
.update(fld_entry.update_fields()))
self.db_session.commit()
def _filter_sid_obj(self, row_object, sid):
res = (self
.db_session
.query(ad_shortcut)
.filter(ad_shortcut.sid == sid)
.query(row_object)
.filter(row_object.sid == sid))
return res
def _filter_sid_list(self, row_object, sid):
res = (self
.db_session
.query(row_object)
.filter(row_object.sid == sid)
.all())
return res
def get_shortcuts(self, sid):
return self._filter_sid_list(ad_shortcut, sid)
def get_printers(self, sid):
res = (self
.db_session
.query(printer_entry)
.filter(printer_entry.sid == sid)
.all())
return res
return self._filter_sid_list(printer_entry, sid)
def get_drives(self, sid):
res = (self
.db_session
.query(drive_entry)
.filter(drive_entry.sid == sid)
.all())
return res
return self._filter_sid_list(drive_entry, sid)
def get_folders(self, sid):
return self._filter_sid_list(folder_entry, sid)
def get_hkcu_entry(self, sid, hive_key):
res = (self
@ -321,40 +351,16 @@ class sqlite_registry(registry):
return res
def wipe_user(self, sid):
self.wipe_hkcu(sid)
self.wipe_shortcuts(sid)
self.wipe_printers(sid)
self.wipe_drives(sid)
self._wipe_sid(samba_hkcu_preg, sid)
self._wipe_sid(ad_shortcut, sid)
self._wipe_sid(printer_entry, sid)
self._wipe_sid(drive_entry, sid)
def wipe_shortcuts(self, sid):
def _wipe_sid(self, row_object, sid):
(self
.db_session
.query(ad_shortcut)
.filter(ad_shortcut.sid == sid)
.delete())
self.db_session.commit()
def wipe_printers(self, sid):
(self
.db_session
.query(printer_entry)
.filter(printer_entry.sid == sid)
.delete())
self.db_session.commit()
def wipe_drives(self, sid):
(self
.db_session
.query(drive_entry)
.filter(drive_entry.sid == sid)
.delete())
self.db_session.commit()
def wipe_hkcu(self, sid):
(self
.db_session
.query(samba_hkcu_preg)
.filter(samba_hkcu_preg.sid == sid)
.query(row_object)
.filter(row_object.sid == sid)
.delete())
self.db_session.commit()

6
gpoa/util/preg.py
View File

@ -76,15 +76,15 @@ def preg_keymap(preg):
return keymap
def merge_polfile(preg, sid=None, reg_name='registry', reg_path=None):
def merge_polfile(preg, sid=None, reg_name='registry', reg_path=None, policy_name='Unknown'):
pregfile = load_preg(preg)
logging.info(slogm('Loaded PReg {}'.format(preg)))
storage = registry_factory(reg_name, reg_path)
for entry in pregfile.entries:
if not sid:
storage.add_hklm_entry(entry)
storage.add_hklm_entry(entry, policy_name)
else:
storage.add_hkcu_entry(entry, sid)
storage.add_hkcu_entry(entry, sid, policy_name)
class entry: