Skip to content

Support OPT_X_TLS_PEERCERT #427

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.

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

Closed
wants to merge 2 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
7 changes: 6 additions & 1 deletion Doc/reference/ldap.rst
Original file line number Diff line number Diff line change
Expand Up @@ -406,7 +406,12 @@ TLS options

.. py:data:: OPT_X_TLS_PEERCERT

Get peer's certificate as binary ASN.1 data structure (not supported)
Get peer's certificate as binary ASN.1 data structure (DER)

.. versionadded:: 3.4.0

.. note::
The option leaks memory with OpenLDAP < 2.5.8.

.. py:data:: OPT_X_TLS_PROTOCOL_MIN

Expand Down
24 changes: 23 additions & 1 deletion Modules/options.c
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
#include "LDAPObject.h"
#include "ldapcontrol.h"
#include "options.h"
#include "berval.h"

void
set_timeval_from_double(struct timeval *tv, double d)
Expand Down Expand Up @@ -58,6 +59,9 @@ LDAP_set_option(LDAPObject *self, int option, PyObject *value)
case LDAP_OPT_API_FEATURE_INFO:
#ifdef HAVE_SASL
case LDAP_OPT_X_SASL_SSF:
#endif
#ifdef LDAP_OPT_X_TLS_PEERCERT
case LDAP_OPT_X_TLS_PEERCERT:
#endif
/* Read-only options */
PyErr_SetString(PyExc_ValueError, "read-only option");
Expand Down Expand Up @@ -257,6 +261,9 @@ LDAP_get_option(LDAPObject *self, int option)
#if HAVE_SASL
/* unsigned long */
ber_len_t blen;
#endif
#ifdef LDAP_OPT_X_TLS_PEERCERT
struct berval bv = {0};
#endif
PyObject *extensions, *v;
Py_ssize_t i, num_extensions;
Expand Down Expand Up @@ -431,7 +438,22 @@ LDAP_get_option(LDAPObject *self, int option)
v = LDAPControls_to_List(lcs);
ldap_controls_free(lcs);
return v;

#ifdef LDAP_OPT_X_TLS_PEERCERT
case LDAP_OPT_X_TLS_PEERCERT:
res = LDAP_int_get_option(self, option, &bv);
if (res != LDAP_OPT_SUCCESS) {
return option_error(res, "ldap_get_option");
}
if (bv.bv_len == 0) {
Py_RETURN_NONE;
} else {
v = LDAPberval_to_object(&bv);
/* bv_val memory is allocated with ber_memalloc_x()
* context allocation/dealloc is a private API. */
ber_memfree(bv.bv_val);
return v;
}
#endif
default:
PyErr_Format(PyExc_ValueError, "unknown option %d", option);
return NULL;
Expand Down
34 changes: 34 additions & 0 deletions Tests/t_ldapobject.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,11 @@

See https://www.python-ldap.org/ for details.
"""
import base64
import errno
import linecache
import os
import re
import socket
import unittest
import pickle
Expand All @@ -20,6 +22,11 @@
from slapdtest import requires_ldapi, requires_sasl, requires_tls
from slapdtest import requires_init_fd

PEM_CERT_RE = re.compile(
b'-----BEGIN CERTIFICATE-----(.*?)-----END CERTIFICATE-----',
re.DOTALL
)


LDIF_TEMPLATE = """dn: %(suffix)s
objectClass: dcObject
Expand Down Expand Up @@ -421,6 +428,33 @@ def test_multiple_starttls(self):
l.simple_bind_s(self.server.root_dn, self.server.root_pw)
self.assertEqual(l.whoami_s(), 'dn:' + self.server.root_dn)

@requires_tls()
@unittest.skipUnless(
hasattr(ldap, "OPT_X_TLS_PEERCERT"),
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sounds like you still need to register this in Lib/ldap/constants.py to get the tests enabled?

reason="Requires OPT_X_TLS_PEERCERT"
)
def test_get_tls_peercert(self):
l = self.ldap_object_class(self.server.ldap_uri)
peercert = l.get_option(ldap.OPT_X_TLS_PEERCERT)
self.assertEqual(peercert, None)
with self.assertRaises(ValueError):
l.set_option(ldap.OPT_X_TLS_PEERCERT, b"")

l.set_option(ldap.OPT_X_TLS_CACERTFILE, self.server.cafile)
l.set_option(ldap.OPT_X_TLS_NEWCTX, 0)
l.start_tls_s()

peercert = l.get_option(ldap.OPT_X_TLS_PEERCERT)
self.assertTrue(peercert)
self.assertIsInstance(peercert, bytes)

with open(self.server.servercert, "rb") as f:
server_cert = f.read()
pem_body = PEM_CERT_RE.search(server_cert).group(1)
server_der = base64.b64decode(pem_body)

self.assertEqual(server_der, peercert)

def test_dse(self):
dse = self._ldap_conn.read_rootdse_s()
self.assertIsInstance(dse, dict)
Expand Down