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

Merge pull request #13 from altlinux/gpoa_user_computer_at_once

Browse Source 
Update gpoa command logic with user or computer at once
This commit is contained in:
Evgeny Sinelnikov 2020-01-30 10:35:38 +04:00 committed by GitHub
commit 2e7d18db1d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 98 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
gpoa/backend/__init__.py
View File

@ -21,23 +21,27 @@ from util.windows import smbcreds
from .samba_backend import samba_backend
from .nodomain_backend import nodomain_backend
def backend_factory(dc, username):
def backend_factory(dc, username, is_machine, no_domain = False):
'''
Return one of backend objects. Please note that backends must
store their configuration in a storage with administrator
write permissions in order to prevent users from modifying
policies enforced by domain administrators.
'''
sc = smbcreds(dc)
domain = sc.get_domain()
back = None
domain = None
if not no_domain:
sc = smbcreds(dc)
domain = sc.get_domain()
if dc:
if domain:
logging.debug('Initialize Samba backend for domain: {}'.format(domain))
try:
back = samba_backend(sc, username, domain)
back = samba_backend(sc, username, domain, is_machine)
except Exception as exc:
logging.error('Unable to initialize Samba backend: {}'.format(exc))
else:
logging.debug('Initialize local backend with no domain')
try:
back = nodomain_backend()
except Exception as exc:

11
gpoa/backend/nodomain_backend.py
View File

@ -31,14 +31,17 @@ from util.logging import slogm
class nodomain_backend(applier_backend):
def __init__(self):
domain = None
machine_name = get_machine_name()
machine_sid = get_sid(domain, machine_name, True)
self.storage = registry_factory('registry')
self.storage.set_info('domain', domain)
self.storage.set_info('machine_name', get_machine_name())
self.storage.set_info('machine_sid', get_sid(domain, self.storage.get_info('machine_name')))
self.storage.set_info('machine_name', machine_name)
self.storage.set_info('machine_sid', machine_sid)
# User SID to work with HKCU hive
self.username = username
self.sid = get_sid(self.storage.get_info('domain'), self.username)
self.username = machine_name
self.sid = machine_sid
def retrieve_and_store(self):
'''

15
gpoa/backend/samba_backend.py
View File

@ -33,16 +33,21 @@ from util.logging import slogm
class samba_backend(applier_backend):
def __init__(self, sambacreds, username, domain):
def __init__(self, sambacreds, username, domain, is_machine):
self.storage = registry_factory('registry')
self.storage.set_info('domain', domain)
self.storage.set_info('machine_name', get_machine_name())
self.storage.set_info('machine_sid', get_sid(domain, self.storage.get_info('machine_name')))
machine_name = get_machine_name()
machine_sid = get_sid(domain, machine_name, is_machine)
self.storage.set_info('machine_name', machine_name)
self.storage.set_info('machine_sid', machine_sid)
# User SID to work with HKCU hive
self.username = username
self._is_machine_username = is_machine_name(self.username)
self.sid = get_sid(self.storage.get_info('domain'), self.username)
self._is_machine_username = is_machine
if is_machine:
self.sid = machine_sid
else:
self.sid = get_sid(self.storage.get_info('domain'), self.username)
self.cache = cache_factory('regpol_cache')
self.gpo_names = cache_factory('gpo_names')

15
gpoa/frontend/frontend_manager.py
View File

@ -70,12 +70,12 @@ class frontend_manager:
for machine and user parts of policies.
'''
def __init__(self, username, target):
def __init__(self, username, is_machine):
self.storage = registry_factory('registry')
self.username = determine_username(username)
self.target = target
self.is_machine = is_machine
self.process_uname = get_process_user()
self.sid = get_sid(self.storage.get_info('domain'), self.username)
self.sid = get_sid(self.storage.get_info('domain'), self.username, is_machine)
self.machine_appliers = dict({
'control': control_applier(self.storage),
@ -135,11 +135,8 @@ class frontend_manager:
'''
Decide which appliers to run.
'''
if 'All' == self.target or 'Computer' == self.target:
if self.is_machine:
self.machine_apply()
# Run user appliers when user's SID is specified
if self.storage.get_info('machine_sid') != self.sid:
if 'All' == self.target or 'User' == self.target:
self.user_apply()
else:
self.user_apply()

24
gpoa/gpoa
View File

@ -31,8 +31,7 @@ from util.users import (
get_process_user
)
from util.arguments import (
set_loglevel,
process_target
set_loglevel
)
from util.logging import slogm
@ -41,7 +40,6 @@ def parse_arguments():
arguments.add_argument('user',
type=str,
nargs='?',
default=get_machine_name(),
help='Domain username ({}) to parse policies for'.format(get_machine_name()))
arguments.add_argument('--dc',
type=str,
@ -49,9 +47,6 @@ def parse_arguments():
arguments.add_argument('--nodomain',
action='store_true',
help='Operate without domain (apply local policy)')
arguments.add_argument('--target',
type=str,
help='Specify if it is needed to update user\'s or computer\'s policies')
arguments.add_argument('--noupdate',
action='store_true',
help='Don\'t try to update storage, only run appliers')
@ -70,6 +65,10 @@ class gpoa_controller:
def __init__(self):
self.__args = parse_arguments()
self.is_machine = False
if not self.__args.user:
user = get_machine_name()
self.is_machine = True
set_loglevel(self.__args.loglevel)
self.__kinit_successful = machine_kinit()
@ -81,7 +80,6 @@ class gpoa_controller:
self.username = uname
else:
self.username = determine_username(self.__args.user)
self.target = process_target(self.__args.target)
def run(self):
'''
@ -96,12 +94,13 @@ class gpoa_controller:
Function to start update of settings storage
'''
dc = self.__args.dc
nodomain = False
if self.__args.nodomain:
dc = None
nodomain = True
if not self.__args.noupdate:
if is_root():
back = backend_factory(dc, self.username)
back = backend_factory(dc, self.username, self.is_machine, nodomain)
if back:
back.retrieve_and_store()
@ -109,13 +108,8 @@ class gpoa_controller:
'''
Function to start appliers
'''
target = self.target
if self.__args.nodomain:
target = 'Computer'
try:
appl = frontend_manager(self.username, target)
appl = frontend_manager(self.username, self.is_machine)
appl.apply_parameters()
except Exception as exc:
logging.error(slogm('Error occured while running applier: {}'.format(exc)))

50
gpoa/gpupdate
View File

@ -28,6 +28,9 @@ import pwd
from util.users import (
is_root
)
from util.arguments import (
process_target
)
from util.dbus import (
is_oddjobd_gpupdate_accessible,
dbus_runner
@ -61,10 +64,14 @@ def parse_cli_arguments():
'--user',
default=None,
help='Name of the user for GPO update')
argparser.add_argument('--target',
default=None,
type=str,
help='Specify if it is needed to update user\'s or computer\'s policies')
return argparser.parse_args()
def runner_factory(args):
def runner_factory(args, target):
'''
Return the necessary runner class according to some
factors taken into account.
@ -84,24 +91,36 @@ def runner_factory(args):
else:
# User may only perform gpupdate for machine (None) or
# itself (os.getusername()).
if args.user:
username = pwd.getpwuid(os.getuid()).pw_name
if args.user != username:
logstring = (
'Unable to perform gpupdate for {} with current'
' permissions, will update current user settings'
)
logging.error(logstring.format(args.user))
username = pwd.getpwuid(os.getuid()).pw_name
if args.user != username:
logstring = (
'Unable to perform gpupdate for {} with current'
' permissions, will update current user settings'
)
logging.error(logstring.format(args.user))
if is_oddjobd_gpupdate_accessible():
logging.debug('Starting gpupdate via D-Bus')
return dbus_runner(username)
computer_runner = None
user_runner = None
if target == 'All' or target == 'Computer':
computer_runner = dbus_runner()
if username:
if target == 'All' or target == 'User':
user_runner = dbus_runner(username)
return (computer_runner, user_runner)
else:
logging.warning('oddjobd is inaccessible')
if is_root():
logging.debug('Starting gpupdate by command invocation')
return file_runner(username)
computer_runner = None
user_runner = None
if target == 'All' or target == 'Computer':
computer_runner = file_runner()
if target == 'All' or target == 'User':
user_runner = file_runner(username)
return (computer_runner, user_runner)
else:
logging.error('Insufficient permissions to run gpupdate')
@ -109,9 +128,12 @@ def runner_factory(args):
def main():
args = parse_cli_arguments()
gpo_applier = runner_factory(args)
if gpo_applier:
gpo_applier.run()
gpo_appliers = runner_factory(args, process_target(args.target))
if gpo_appliers:
if gpo_appliers[0]:
gpo_appliers[0].run()
if gpo_appliers[1]:
gpo_appliers[1].run()
else:
logging.error('gpupdate will not be started')

9
gpoa/util/dbus.py
View File

@ -19,6 +19,7 @@ import logging
import dbus
from .logging import slogm
from .users import is_root
class dbus_runner:
@ -40,10 +41,14 @@ class dbus_runner:
#print(obj.Introspect()[0])
if self.username:
logging.info(slogm('Starting GPO applier for user {} via D-Bus'.format(self.username)))
result = self.interface.gpupdatefor(dbus.String(self.username))
if is_root():
result = self.interface.gpupdatefor(dbus.String(self.username))
else:
result = self.interface.gpupdate()
print_dbus_result(result)
else:
result = self.interface.gpupdate()
logging.info(slogm('Starting GPO applier for computer via D-Bus'))
result = self.interface.gpupdate_computer()
print_dbus_result(result)
#self.interface.Quit()

17
gpoa/util/windows.py
View File

@ -19,6 +19,7 @@
import logging
import os
import pwd
import optparse
from samba import getopt as options
@ -147,17 +148,27 @@ def wbinfo_getsid(domain, user):
return sid
def get_sid(domain, username):
def get_local_sid_prefix():
return "S-1-5-21-0-0-0"
def get_sid(domain, username, is_machine = False):
'''
Lookup SID not only using wbinfo or sssd but also using own cache
'''
domain_username = '{}\\{}'.format(domain, username)
sid = 'local-{}'.format(username)
# local user
if not domain:
found_uid = 0
if not is_machine:
found_uid = pwd.getpwnam(username).pw_uid
return '{}-{}'.format(get_local_sid_prefix(), found_uid)
# domain user
try:
sid = wbinfo_getsid(domain, username)
except:
sid = 'local-{}'.format(username)
logging.warning(
slogm('Error getting SID using wbinfo, will use cached SID: {}'.format(sid)))