Skip to content

deps(vendor): auto update vendored Samba code #1237

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
github-actions wants to merge 1 commit into
base: main
Choose a base branch
from
Open

deps(vendor): auto update vendored Samba code #1237

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
96 changes: 24 additions & 72 deletions internal/policies/certificate/python/vendor_samba/gp/gp_cert_auto_enroll_ext.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

import configparser
import os
import operator
import requests
Expand All @@ -27,7 +26,7 @@
from vendor_samba.gp.gpclass import get_dc_hostname
import base64
from shutil import which
from subprocess import Popen, PIPE, TimeoutExpired
from subprocess import Popen, PIPE
import re
import json
from vendor_samba.gp.util.logging import log
Expand Down Expand Up @@ -55,8 +54,6 @@ def load_der_pkcs7_certificates(x): return []
'/etc/pki/ca-trust/source/anchors', # RHEL/Fedora
'/usr/local/share/ca-certificates'] # Debian/Ubuntu

CEPCES_CONFIG_PATH = '/etc/cepces/cepces.conf'

def octet_string_to_objectGUID(data):
"""Convert an octet string to an objectGUID."""
return '%s-%s-%s-%s-%s' % ('%02x' % struct.unpack('<L', data[0:4])[0],
Expand Down Expand Up @@ -151,11 +148,10 @@ def fetch_certification_authorities(ldb):
[MS-CAESO] 4.4.5.3.1.2
"""
result = []
configdn = ldb.get_config_basedn()
basedn = ldb.get_default_basedn()
# Autoenrollment MUST do an LDAP search for the CA information
# (pKIEnrollmentService) objects under the following container:
dn = 'CN=Enrollment Services,CN=Public Key Services,CN=Services,%s' % configdn

dn = 'CN=Enrollment Services,CN=Public Key Services,CN=Services,CN=Configuration,%s' % basedn
attrs = ['cACertificate', 'cn', 'dNSHostName']
expr = '(objectClass=pKIEnrollmentService)'
res = ldb.search(dn, SCOPE_SUBTREE, expr, attrs)
Expand All @@ -172,8 +168,8 @@ def fetch_certification_authorities(ldb):
def fetch_template_attrs(ldb, name, attrs=None):
if attrs is None:
attrs = ['msPKI-Minimal-Key-Size']
configdn = ldb.get_config_basedn()
dn = 'CN=Certificate Templates,CN=Public Key Services,CN=Services,%s' % configdn
basedn = ldb.get_default_basedn()
dn = 'CN=Certificate Templates,CN=Public Key Services,CN=Services,CN=Configuration,%s' % basedn
expr = '(cn=%s)' % name
res = ldb.search(dn, SCOPE_SUBTREE, expr, attrs)
if len(res) == 1 and 'msPKI-Minimal-Key-Size' in res[0]:
Expand All @@ -195,33 +191,14 @@ def get_supported_templates(server):
log.error('Failed to find cepces-submit')
return []

cepces_config = configparser.ConfigParser()
cepces_config.read(CEPCES_CONFIG_PATH)
# If the config file was overriden, then we let cepces-submit use the values defined there.
cepces_args = [cepces_submit, '--server=%s' % server, '--auth=Kerberos']
if cepces_config['global'] and cepces_config['global']['server'] != 'ca':
log.debug(f'Changes detected in "{CEPCES_CONFIG_PATH}". Using the values defined there.')
cepces_args = [cepces_submit]

env = os.environ
env['CERTMONGER_OPERATION'] = 'GET-SUPPORTED-TEMPLATES'
p = Popen(cepces_args, env=env, stdout=PIPE, stderr=PIPE)
out, err = bytes(), bytes()
try:
# Server calls could hang if the provider is slow or too busy.
# A timeout is set to avoid this.
out, err = p.communicate(timeout=3)
except TimeoutExpired as te:
log.warn(f'GET-SUPPORTED-TEMPLATES expired. {server} is probably not a CEPCES server: {te}')
p.kill()
except Exception as e:
log.error(f'Failed to fetch supported templates with command "{cepces_args}": {e}')
p.kill()

p = Popen([cepces_submit, '--server=%s' % server, '--auth=Kerberos'],
env=env, stdout=PIPE, stderr=PIPE)
out, err = p.communicate()
if p.returncode != 0:
data = {'Error': err.decode()}
log.error('Failed to fetch the list of supported templates.', data)
raise Exception(f'Failed to fetch the list of supported templates: {data}')
return out.strip().split()


Expand Down Expand Up @@ -296,11 +273,11 @@ def changed(new_data, old_data):
def cert_enroll(ca, ldb, trust_dir, private_dir, auth='Kerberos'):
"""Install the root certificate chain."""
data = dict({'files': [], 'templates': []}, **ca)
url = 'http://%s/CertSrv/mscep/mscep.dll/pkiclient.exe?' % ca['hostname']

log.info("Try to get root or server certificates")
url = 'https://%s/CertSrv/mscep/mscep.dll/pkiclient.exe?' % ca['hostname']
root_certs = getca(ca, url, trust_dir)

root_certs = getca(ca, url, trust_dir)
data['files'].extend(root_certs)
global_trust_dir = find_global_trust_dir()
for src in root_certs:
Expand Down Expand Up @@ -348,7 +325,6 @@ def cert_enroll(ca, ldb, trust_dir, private_dir, auth='Kerberos'):
log.error('Failed to add Certificate Authority', data)

supported_templates = get_supported_templates(ca['hostname'])
log.debug(f"Supported templates for CA \"{ca['name']}\": {supported_templates}")
for template in supported_templates:
attrs = fetch_template_attrs(ldb, template)
nickname = '%s.%s' % (ca['name'], template.decode())
Expand All @@ -368,11 +344,8 @@ def cert_enroll(ca, ldb, trust_dir, private_dir, auth='Kerberos'):
data = {'Error': err.decode(), 'Certificate': nickname}
log.error('Failed to request certificate', data)

if data.get('files') is not None:
data['files'].extend([keyfile, certfile])

if data.get('templates') is not None:
data['templates'].append(nickname)
data['files'].extend([keyfile, certfile])
data['templates'].append(nickname)
if update is not None:
ret = Popen([update]).wait()
if ret != 0:
Expand Down Expand Up @@ -405,19 +378,8 @@ def apply(self, guid, ca, applier_func, *args, **kwargs):
# If the policy has changed, unapply, then apply new policy
old_val = self.cache_get_attribute_value(guid, attribute)
old_data = json.loads(old_val) if old_val is not None else {}

templates = []
if old_val is not None:
supported_templates = []
try:
supported_templates = get_supported_templates(ca['hostname'])
except Exception as e:
raise e
log.debug(f"Supported templates for CA \"{ca['name']}\": {supported_templates}")
for template in supported_templates:
formatted = '%s.%s' % (ca['name'], template.decode())
templates.append(formatted)

templates = ['%s.%s' % (ca['name'], t.decode()) for t in get_supported_templates(ca['hostname'])] \
if old_val is not None else []
new_data = { 'templates': templates, **ca }
if changed(new_data, old_data) or self.cache_get_apply_state() == GPOSTATE.ENFORCE:
self.unapply(guid, attribute, old_val)
Expand Down Expand Up @@ -501,17 +463,15 @@ def __read_cep_data(self, guid, ldb, end_point_information,
# If the current group contains a
# CertificateEnrollmentPolicyEndPoint instance with EndPoint.URI
# equal to "LDAP":
for e in end_point_group:
if e['URL'] != 'LDAP:':
continue
if any([e['URL'] == 'LDAP:' for e in end_point_group]):
# Perform an LDAP search to read the value of the objectGuid
# attribute of the root object of the forest root domain NC. If
# any errors are encountered, continue with the next group.
res = ldb.search('', SCOPE_BASE, '(objectClass=*)',
['defaultNamingContext'])
['rootDomainNamingContext'])
if len(res) != 1:
continue
res2 = ldb.search(res[0]['defaultNamingContext'][0],
res2 = ldb.search(res[0]['rootDomainNamingContext'][0],
SCOPE_BASE, '(objectClass=*)',
['objectGUID'])
if len(res2) != 1:
Expand All @@ -534,13 +494,9 @@ def __read_cep_data(self, guid, ldb, end_point_information,
if ca['URL'] == 'LDAP:':
# This is a basic configuration.
cas = fetch_certification_authorities(ldb)
log.debug(f'Fetched the following CAs: {cas}')
for _ca in cas:
try:
self.apply(guid, _ca, cert_enroll, _ca, ldb, trust_dir, private_dir)
except Exception as e:
log.warn(f"Could not enroll to CA {_ca['name']}: {e}")
continue
self.apply(guid, _ca, cert_enroll, _ca, ldb, trust_dir,
private_dir)
ca_names.append(_ca['name'])
# If EndPoint.URI starts with "HTTPS//":
elif ca['URL'].lower().startswith('https://'):
Expand All @@ -560,19 +516,15 @@ def __enroll(self, guid, entries, trust_dir, private_dir):
ca_names = []
end_point_information = obtain_end_point_information(entries)
if len(end_point_information) > 0:
cep_data = self.__read_cep_data(guid, ldb, end_point_information, trust_dir, private_dir)
if cep_data:
ca_names.extend(cep_data)
ca_names.extend(self.__read_cep_data(guid, ldb,
end_point_information,
trust_dir, private_dir))
else:
cas = fetch_certification_authorities(ldb)
for ca in cas:
try:
self.apply(guid, ca, cert_enroll, ca, ldb, trust_dir, private_dir)
except Exception as e:
log.warn(f"Could not enroll to CA {ca['name']}: {e}")
continue
self.apply(guid, ca, cert_enroll, ca, ldb, trust_dir,
private_dir)
ca_names.append(ca['name'])
log.debug(f'Enrolled to the following CAs: {ca_names}')
return ca_names

def rsop(self, gpo):
Expand Down