Author: Amaury Forgeot d'Arc <[email protected]>
Branch: py3.3
Changeset: r75003:5efb69f6c4d5
Date: 2014-12-18 00:28 +0100
http://bitbucket.org/pypy/pypy/changeset/5efb69f6c4d5/
Log: SSL: add compression() method
diff --git a/pypy/module/_ssl/interp_ssl.py b/pypy/module/_ssl/interp_ssl.py
--- a/pypy/module/_ssl/interp_ssl.py
+++ b/pypy/module/_ssl/interp_ssl.py
@@ -705,6 +705,19 @@
return space.wrap(
rffi.charpsize2str(out_ptr[0], widen(len_ptr[0])))
+ def compression(self, space):
+ # if OPENSSL_NO_COMP
+ # return None
+ if not self.ssl:
+ return None
+ comp_method = libssl_SSL_get_current_compression(self.ssl)
+ if not comp_method or widen(comp_method[0].c_type) == NID_undef:
+ return None
+ short_name = libssl_OBJ_nid2sn(comp_method[0].c_type)
+ if not short_name:
+ return None
+ return space.wrap(rffi.charp2str(short_name))
+
def tls_unique_cb(self, space):
"""Returns the 'tls-unique' channel binding data, as defined by RFC
5929.
If the TLS handshake is not yet complete, None is returned"""
@@ -937,6 +950,7 @@
cipher = interp2app(SSLSocket.cipher),
peer_certificate = interp2app(SSLSocket.peer_certificate),
selected_npn_protocol = interp2app(SSLSocket.selected_npn_protocol),
+ compression = interp2app(SSLSocket.compression),
tls_unique_cb = interp2app(SSLSocket.tls_unique_cb),
)
diff --git a/pypy/module/_ssl/test/test_ssl.py
b/pypy/module/_ssl/test/test_ssl.py
--- a/pypy/module/_ssl/test/test_ssl.py
+++ b/pypy/module/_ssl/test/test_ssl.py
@@ -198,6 +198,14 @@
self.s.close()
del ss; gc.collect()
+ def test_compression(self):
+ import ssl, sys, gc
+ ss = ssl.wrap_socket(self.s)
+ ss.do_handshake()
+ assert ss.compression() in [None, 'ZLIB', 'RLE']
+ self.s.close()
+ del ss; gc.collect()
+
class AppTestConnectedSSL_Timeout(AppTestConnectedSSL):
# Same tests, with a socket timeout
diff --git a/rpython/rlib/ropenssl.py b/rpython/rlib/ropenssl.py
--- a/rpython/rlib/ropenssl.py
+++ b/rpython/rlib/ropenssl.py
@@ -102,6 +102,7 @@
SSL_MODE_ACCEPT_MOVING_WRITE_BUFFER =
rffi_platform.ConstantInteger("SSL_MODE_ACCEPT_MOVING_WRITE_BUFFER")
SSL_TLSEXT_ERR_OK = rffi_platform.ConstantInteger("SSL_TLSEXT_ERR_OK")
+ NID_undef = rffi_platform.ConstantInteger("NID_undef")
NID_subject_alt_name =
rffi_platform.ConstantInteger("NID_subject_alt_name")
GEN_DIRNAME = rffi_platform.ConstantInteger("GEN_DIRNAME")
GEN_EMAIL = rffi_platform.ConstantInteger("GEN_EMAIL")
@@ -149,6 +150,10 @@
('name', rffi.CCHARP),
])
+ COMP_METHOD_st = rffi_platform.Struct(
+ 'COMP_METHOD',
+ [('type', rffi.INT),
+ ])
for k, v in rffi_platform.configure(CConfig).items():
globals()[k] = v
@@ -170,6 +175,7 @@
GENERAL_NAMES = rffi.COpaquePtr('GENERAL_NAMES')
GENERAL_NAME = rffi.CArrayPtr(GENERAL_NAME_st)
OBJ_NAME = rffi.CArrayPtr(OBJ_NAME_st)
+COMP_METHOD = rffi.CArrayPtr(COMP_METHOD_st)
HAVE_OPENSSL_RAND = OPENSSL_VERSION_NUMBER >= 0x0090500f
HAVE_SSL_CTX_CLEAR_OPTIONS = OPENSSL_VERSION_NUMBER >= 0x009080df
@@ -283,12 +289,16 @@
ssl_external('GENERAL_NAME_print', [BIO, GENERAL_NAME], rffi.INT)
ssl_external('pypy_GENERAL_NAME_dirn', [GENERAL_NAME], X509_NAME,
macro=True)
+ssl_external('OBJ_nid2sn',
+ [rffi.INT], rffi.CCHARP)
ssl_external('SSL_get_current_cipher', [SSL], SSL_CIPHER)
ssl_external('SSL_CIPHER_get_name', [SSL_CIPHER], rffi.CCHARP)
ssl_external('SSL_CIPHER_get_version', [SSL_CIPHER], rffi.CCHARP)
ssl_external('SSL_CIPHER_get_bits', [SSL_CIPHER, rffi.INTP], rffi.INT)
+ssl_external('SSL_get_current_compression', [SSL], COMP_METHOD)
+
ssl_external('ERR_get_error', [], rffi.INT)
ssl_external('ERR_peek_last_error', [], rffi.INT)
ssl_external('ERR_error_string', [rffi.ULONG, rffi.CCHARP], rffi.CCHARP)
_______________________________________________
pypy-commit mailing list
[email protected]
https://mail.python.org/mailman/listinfo/pypy-commit
