sin/samba-mirror
1
0
Fork 0
mirror of https://github.com/samba-team/samba.git synced 2025-03-22 02:50:28 +03:00
Code

gpo: Read GPO versions locally, not from sysvol

Browse Source 
Non-kdc clients cannot read directly from the
sysvol, so we need to store the GPT.INI file
locally to read each gpo version.

Signed-off-by: David Mulder <dmulder@suse.com>
Reviewed-by: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>
This commit is contained in:
David Mulder 2018-01-08 07:17:29 -07:00 committed by Andrew Bartlett
parent 4c7348e44d
commit 57faf35cf8
2 changed files with 89 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
python/samba/gpclass.py
View File

@ -17,6 +17,7 @@
import sys
import os
import errno
import tdb
sys.path.insert(0, "bin/python")
from samba import NTSTATUSError
@ -29,6 +30,7 @@ from samba.net import Net
from samba.dcerpc import nbt
from samba import smb
import samba.gpo as gpo
from tempfile import NamedTemporaryFile
try:
from enum import Enum
@ -423,6 +425,49 @@ def get_gpo_list(dc_hostname, creds, lp):
gpos = ads.get_gpo_list(creds.get_username())
return gpos
def cache_gpo_dir(conn, cache, sub_dir):
loc_sub_dir = sub_dir.upper()
local_dir = os.path.join(cache, loc_sub_dir)
try:
os.makedirs(local_dir, mode=0o755)
except OSError as e:
if e.errno != errno.EEXIST:
raise
for fdata in conn.list(sub_dir):
if fdata['attrib'] & smb.FILE_ATTRIBUTE_DIRECTORY:
cache_gpo_dir(conn, cache, os.path.join(sub_dir, fdata['name']))
else:
local_name = fdata['name'].upper()
f = NamedTemporaryFile(delete=False, dir=local_dir)
fname = os.path.join(sub_dir, fdata['name']).replace('/', '\\')
f.write(conn.loadfile(fname))
f.close()
os.rename(f.name, os.path.join(local_dir, local_name))
def check_safe_path(path):
dirs = re.split('/|\\\\', path)
if 'sysvol' in path:
dirs = dirs[dirs.index('sysvol')+1:]
if not '..' in dirs:
return os.path.join(*dirs)
raise OSError(path)
def check_refresh_gpo_list(dc_hostname, lp, creds, gpos):
conn = smb.SMB(dc_hostname, 'sysvol', lp=lp, creds=creds, sign=True)
cache_path = lp.cache_path('gpo_cache')
for gpo in gpos:
if not gpo.file_sys_path:
continue
cache_gpo_dir(conn, cache_path, check_safe_path(gpo.file_sys_path))
def gpo_version(lp, path):
# gpo.gpo_get_sysvol_gpt_version() reads the GPT.INI from a local file,
# read from the gpo client cache.
gpt_path = lp.cache_path(os.path.join('gpo_cache', path))
return int(gpo.gpo_get_sysvol_gpt_version(gpt_path)[1])
def apply_gp(lp, creds, test_ldb, logger, store, gp_extensions):
gp_db = store.get_gplog(creds.get_username())
dc_hostname = get_dc_hostname(creds, lp)
@ -432,14 +477,19 @@ def apply_gp(lp, creds, test_ldb, logger, store, gp_extensions):
logger.error('Error connecting to \'%s\' using SMB' % dc_hostname)
raise
gpos = get_gpo_list(dc_hostname, creds, lp)
try:
check_refresh_gpo_list(dc_hostname, lp, creds, gpos)
except:
logger.error('Failed downloading gpt cache from \'%s\' using SMB' \
% dc_hostname)
return
for gpo_obj in gpos:
guid = gpo_obj.name
if guid == 'Local Policy':
continue
path = os.path.join(lp.get('realm').lower(), 'Policies', guid)
local_path = os.path.join(lp.get("path", "sysvol"), path)
version = int(gpo.gpo_get_sysvol_gpt_version(local_path)[1])
path = os.path.join(lp.get('realm'), 'Policies', guid).upper()
version = gpo_version(lp, path)
if version != store.get_int(guid):
logger.info('GPO %s has changed' % guid)
gp_db.state(GPOSTATE.APPLY)

38
python/samba/tests/gpo.py
View File

@ -17,6 +17,7 @@
import os
from samba import gpo, tests
from samba.param import LoadParm
from samba.gpclass import check_refresh_gpo_list, check_safe_path
poldir = r'\\addom.samba.example.com\sysvol\addom.samba.example.com\Policies'
dspath = 'CN=Policies,CN=System,DC=addom,DC=samba,DC=example,DC=com'
@ -59,8 +60,8 @@ class GPOTests(tests.TestCase):
def test_gpt_version(self):
global gpt_data
local_path = self.lp.get("path", "sysvol")
policies = 'addom.samba.example.com/Policies'
local_path = self.lp.cache_path('gpo_cache')
policies = 'ADDOM.SAMBA.EXAMPLE.COM/POLICIES'
guid = '{31B2F340-016D-11D2-945F-00C04FB984F9}'
gpo_path = os.path.join(local_path, policies, guid)
old_vers = gpo.gpo_get_sysvol_gpt_version(gpo_path)[1]
@ -75,3 +76,36 @@ class GPOTests(tests.TestCase):
self.assertEquals(gpo.gpo_get_sysvol_gpt_version(gpo_path)[1], old_vers,
'gpo_get_sysvol_gpt_version() did not return the expected version')
def test_check_refresh_gpo_list(self):
cache = self.lp.cache_path('gpo_cache')
ads = gpo.ADS_STRUCT(self.server, self.lp, self.creds)
if ads.connect():
gpos = ads.get_gpo_list(self.creds.get_username())
check_refresh_gpo_list(self.server, self.lp, self.creds, gpos)
self.assertTrue(os.path.exists(cache),
'GPO cache %s was not created' % cache)
guid = '{31B2F340-016D-11D2-945F-00C04FB984F9}'
gpt_ini = os.path.join(cache, 'ADDOM.SAMBA.EXAMPLE.COM/POLICIES',
guid, 'GPT.INI')
self.assertTrue(os.path.exists(gpt_ini),
'GPT.INI was not cached for %s' % guid)
def test_check_refresh_gpo_list_malicious_paths(self):
# the path cannot contain ..
path = '/usr/local/samba/var/locks/sysvol/../../../../../../root/'
self.assertRaises(OSError, check_safe_path, path)
self.assertEqual(check_safe_path('/etc/passwd'), 'etc/passwd')
self.assertEqual(check_safe_path('\\\\etc/\\passwd'), 'etc/passwd')
# there should be no backslashes used to delineate paths
before = 'sysvol/addom.samba.example.com\\Policies/' \
'{31B2F340-016D-11D2-945F-00C04FB984F9}\\GPT.INI'
after = 'addom.samba.example.com/Policies/' \
'{31B2F340-016D-11D2-945F-00C04FB984F9}/GPT.INI'
result = check_safe_path(before)
self.assertEquals(result, after, 'check_safe_path() didn\'t' \
' correctly convert \\ to /')