mirrors/kloop
1
0
Fork 0
mirror of https://gitee.com/fantix/kloop.git synced 2024-11-28 05:10:59 +00:00
Code Issues Projects Releases Wiki Activity

TLS: fix buffer usage and reset message

Browse source 
Also improved debug
This commit is contained in:
Fantix King 2022-07-02 14:37:19 -04:00
parent 74f0062154
commit 761f741d5d
No known key found for this signature in database
GPG key ID: 95304B04071CCDB4
3 changed files with 84 additions and 65 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
src/kloop/includes/openssl/ssl.pxd
View file

@ -20,6 +20,11 @@ cdef extern from "openssl/ssl.h" nogil:
int OP_ENABLE_KTLS "SSL_OP_ENABLE_KTLS" int OP_ENABLE_KTLS "SSL_OP_ENABLE_KTLS"
int set_options "SSL_set_options" (SSL* ssl, int options) int set_options "SSL_set_options" (SSL* ssl, int options)
long SSL_MODE_RELEASE_BUFFERS
long SSL_MODE_ACCEPT_MOVING_WRITE_BUFFER
long clear_mode "SSL_clear_mode" (SSL* ssl, long mode)
int free_buffers "SSL_free_buffers" (SSL *ssl)
cdef extern from *: cdef extern from *:
""" """

2
src/kloop/tls.pxd
View file

@ -25,6 +25,7 @@ cdef struct Proxy:
RingCallback recv_callback RingCallback recv_callback
unsigned char flags unsigned char flags
char* read_buffer char* read_buffer
void* cmsg
Loop* loop Loop* loop
int fd int fd
@ -51,6 +52,7 @@ cdef class TLSTransport:
object write_buffer object write_buffer
bint sending bint sending
cdef do_handshake(self) cdef do_handshake(self)
cdef do_read(self) cdef do_read(self)
cdef do_read_ktls(self) cdef do_read_ktls(self)

142
src/kloop/tls.pyx
View file

@ -15,7 +15,7 @@ import ssl
from cpython cimport PyMem_RawMalloc, PyMem_RawFree from cpython cimport PyMem_RawMalloc, PyMem_RawFree
from libc cimport errno, string from libc cimport errno, string
from .includes.openssl cimport bio, err, ssl as ssl_h from .includes.openssl cimport err, ssl as ssl_h
from .includes cimport pyssl, linux from .includes cimport pyssl, linux
from .loop cimport ring_sq_submit_sendmsg, ring_sq_submit_recvmsg from .loop cimport ring_sq_submit_sendmsg, ring_sq_submit_recvmsg
@ -48,11 +48,16 @@ cdef size_t CMSG_SIZE = libc.CMSG_SPACE(sizeof(unsigned char))
DEF DEBUG = 0 DEF DEBUG = 0
cdef inline void reset_msg(libc.msghdr* msg, size_t controllen) nogil: cdef inline void reset_msg(libc.msghdr* msg, void* cmsg) nogil:
msg.msg_name = NULL msg.msg_name = NULL
msg.msg_namelen = 0 msg.msg_namelen = 0
msg.msg_flags = 0 msg.msg_flags = 0
msg.msg_controllen = controllen if cmsg == NULL:
msg.msg_control = NULL
msg.msg_controllen = 0
else:
msg.msg_control = cmsg
msg.msg_controllen = CMSG_SIZE
cdef object fromOpenSSLError(object err_type): cdef object fromOpenSSLError(object err_type):
@ -72,18 +77,18 @@ cdef int bio_write_ex(
Proxy* proxy = <Proxy*>bio.get_data(b) Proxy* proxy = <Proxy*>bio.get_data(b)
int res int res
if DEBUG: IF DEBUG:
with gil: with gil:
print("bio_write_ex(data=%x, datal=%d)" % (<long long>data, datal)) print("bio_write_ex(data=%x, datal=%d)" % (<long long>data, datal))
if proxy.flags & FLAGS_PROXY_SEND_SUBMITTED: if proxy.flags & FLAGS_PROXY_SEND_SUBMITTED:
if proxy.send_vec.iov_base != data: if proxy.send_vec.iov_base != data:
if DEBUG: IF DEBUG:
with gil: with gil:
print("bio_write_ex() error: concurrent call") print("bio_write_ex() error: concurrent call")
return 0 return 0
if proxy.send_vec.iov_len > datal: if proxy.send_vec.iov_len > datal:
if DEBUG: IF DEBUG:
with gil: with gil:
print("bio_write_ex() error: short rewrite") print("bio_write_ex() error: short rewrite")
return 0 return 0
@ -93,40 +98,42 @@ cdef int bio_write_ex(
proxy.flags &= ~FLAGS_PROXY_SEND_ALL proxy.flags &= ~FLAGS_PROXY_SEND_ALL
res = proxy.send_callback.res res = proxy.send_callback.res
if res < 0: if res < 0:
if DEBUG: IF DEBUG:
with gil: with gil:
print("bio_write_ex() error:", -res) print("bio_write_ex() error:", -res)
errno.errno = -res errno.errno = -res
return 0 return 0
written[0] = res written[0] = res
if DEBUG: IF DEBUG:
with gil: with gil:
print('bio_write_ex() written:', res) print("bio_write_ex() written:", res)
print(">>> ", end="")
for i in range(res): for i in range(res):
print( print(
"%02x " % <unsigned char>data[i], "%02x " % <unsigned char>data[i],
end="" if (i + 1) % 16 and i < res - 1 else "\n", end="" if (i + 1) % 16 or i == res - 1 else "\n>>> ",
) )
print()
else: else:
written[0] = 0 written[0] = 0
bio.set_retry_write(b) bio.set_retry_write(b)
if not proxy.flags & FLAGS_PROXY_SEND_SUBMITTED: if not proxy.flags & FLAGS_PROXY_SEND_SUBMITTED:
if DEBUG: IF DEBUG:
with gil: with gil:
print("bio_write_ex() submit") print("bio_write_ex() submit")
proxy.send_vec.iov_base = data proxy.send_vec.iov_base = data
proxy.send_vec.iov_len = datal proxy.send_vec.iov_len = datal
reset_msg(&proxy.send_msg, 0) reset_msg(&proxy.send_msg, NULL)
if not ring_sq_submit_sendmsg( if not ring_sq_submit_sendmsg(
&proxy.loop.ring.sq, &proxy.loop.ring.sq,
proxy.fd, proxy.fd,
&proxy.send_msg, &proxy.send_msg,
&proxy.send_callback, &proxy.send_callback,
): ):
if DEBUG: IF DEBUG:
with gil: with gil:
print("bio_write_ex() error: SQ full") print("bio_write_ex() error: SQ full")
return 0 return 0
@ -144,18 +151,18 @@ cdef int bio_read_ex(
int res int res
int is_ktls = bio.test_flags(b, FLAGS_KTLS_RX) int is_ktls = bio.test_flags(b, FLAGS_KTLS_RX)
if DEBUG: IF DEBUG:
with gil: with gil:
print('bio_read_ex(data=%x, datal=%d)' % (<long long>data, datal)) print('bio_read_ex(data=%x, datal=%d)' % (<long long>data, datal))
if proxy.flags & FLAGS_PROXY_RECV_SUBMITTED: if proxy.flags & FLAGS_PROXY_RECV_SUBMITTED:
if proxy.recv_vec.iov_base != (data + 5 if is_ktls else data): if proxy.recv_vec.iov_base != (data + 5 if is_ktls else data):
if DEBUG: IF DEBUG:
with gil: with gil:
print("bio_read_ex() error: concurrent call") print("bio_read_ex() error: concurrent call")
return 0 return 0
if proxy.recv_vec.iov_len > (datal - 21 if is_ktls else datal): if proxy.recv_vec.iov_len > (datal - 21 if is_ktls else datal):
if DEBUG: IF DEBUG:
with gil: with gil:
print("bio_read_ex() error: short reread") print("bio_read_ex() error: short reread")
return 0 return 0
@ -164,7 +171,7 @@ cdef int bio_read_ex(
if proxy.flags & FLAGS_PROXY_RECV_KTLS: if proxy.flags & FLAGS_PROXY_RECV_KTLS:
res = proxy.recv_callback.res res = proxy.recv_callback.res
if datal < res + 5: if datal < res + 5:
if DEBUG: IF DEBUG:
with gil: with gil:
print("bio_read_ex() error: datal too short") print("bio_read_ex() error: datal too short")
errno.errno = errno.EINVAL errno.errno = errno.EINVAL
@ -184,24 +191,22 @@ cdef int bio_read_ex(
string.memcpy(data, proxy.read_buffer, res) string.memcpy(data, proxy.read_buffer, res)
readbytes[0] = res readbytes[0] = res
if DEBUG: IF DEBUG:
with gil: with gil:
print( print("bio_read_ex() read:", res, "(forwarded TLS record)")
"bio_read_ex() read:", print("<<< ", end="")
res,
"(forwarded TLS record)"
)
for i in range(res): for i in range(res):
print( print(
"%02x " % <unsigned char>data[i], "%02x " % <unsigned char>data[i],
end="" if (i + 1) % 16 and i < res - 1 else "\n", end="" if (i + 1) % 16 or i == res - 1 else "\n<<< ",
) )
print()
elif proxy.flags & FLAGS_PROXY_RECV_COMPLETED: elif proxy.flags & FLAGS_PROXY_RECV_COMPLETED:
proxy.flags &= ~FLAGS_PROXY_RECV_ALL proxy.flags &= ~FLAGS_PROXY_RECV_ALL
res = proxy.recv_callback.res res = proxy.recv_callback.res
if res < 0: if res < 0:
if DEBUG: IF DEBUG:
with gil: with gil:
print("bio_read_ex() error:", -res) print("bio_read_ex() error:", -res)
errno.errno = -res errno.errno = -res
@ -223,25 +228,25 @@ cdef int bio_read_ex(
bio.set_flags(b, bio.FLAGS_IN_EOF) bio.set_flags(b, bio.FLAGS_IN_EOF)
readbytes[0] = res readbytes[0] = res
if DEBUG: IF DEBUG:
with gil: with gil:
print( print(
"bio_read_ex() read:", "bio_read_ex() read:", res, "(TLS record)" if cmsg else ""
res,
"(TLS record)" if cmsg else ""
) )
print("<<< ", end="")
for i in range(res): for i in range(res):
print( print(
"%02x " % <unsigned char>data[i], "%02x " % <unsigned char>data[i],
end="" if (i + 1) % 16 and i < res - 1 else "\n", end="" if (i + 1) % 16 or i == res - 1 else "\n<<< ",
) )
print()
else: else:
bio.set_retry_read(b) bio.set_retry_read(b)
readbytes[0] = 0 readbytes[0] = 0
if not proxy.flags & FLAGS_PROXY_RECV_SUBMITTED: if not proxy.flags & FLAGS_PROXY_RECV_SUBMITTED:
if is_ktls: if is_ktls:
if datal < 21: if datal < 21:
if DEBUG: IF DEBUG:
with gil: with gil:
print("bio_read_ex() error: datal too short") print("bio_read_ex() error: datal too short")
errno.errno = errno.EINVAL errno.errno = errno.EINVAL
@ -249,7 +254,8 @@ cdef int bio_read_ex(
proxy.recv_vec.iov_base = data + 5 proxy.recv_vec.iov_base = data + 5
proxy.recv_vec.iov_len = datal - 21 proxy.recv_vec.iov_len = datal - 21
if DEBUG: reset_msg(&proxy.recv_msg, proxy.cmsg)
IF DEBUG:
with gil: with gil:
print("bio_read_ex() submit(%x, %d)" % ( print("bio_read_ex() submit(%x, %d)" % (
<long long>proxy.recv_vec.iov_base, <long long>proxy.recv_vec.iov_base,
@ -258,10 +264,10 @@ cdef int bio_read_ex(
else: else:
proxy.recv_vec.iov_base = data proxy.recv_vec.iov_base = data
proxy.recv_vec.iov_len = datal proxy.recv_vec.iov_len = datal
if DEBUG: reset_msg(&proxy.recv_msg, NULL)
IF DEBUG:
with gil: with gil:
print("bio_read_ex() submit") print("bio_read_ex() submit")
reset_msg(&proxy.recv_msg, CMSG_SIZE)
if not ring_sq_submit_recvmsg( if not ring_sq_submit_recvmsg(
&proxy.loop.ring.sq, &proxy.loop.ring.sq,
@ -269,7 +275,7 @@ cdef int bio_read_ex(
&proxy.recv_msg, &proxy.recv_msg,
&proxy.recv_callback, &proxy.recv_callback,
): ):
if DEBUG: IF DEBUG:
with gil: with gil:
print("bio_read_ex() error: SQ full") print("bio_read_ex() error: SQ full")
return 0 return 0
@ -283,24 +289,24 @@ cdef long bio_ctrl(bio.BIO* b, int cmd, long num, void* ptr) nogil:
ssl_h.ktls_crypto_info_t* crypto_info ssl_h.ktls_crypto_info_t* crypto_info
long ret = 0 long ret = 0
if cmd == bio.BIO_CTRL_EOF: if cmd == bio.BIO_CTRL_EOF:
if DEBUG: IF DEBUG:
with gil: with gil:
print("BIO_CTRL_EOF", ret) print("BIO_CTRL_EOF", ret)
elif cmd == bio.BIO_CTRL_PUSH: elif cmd == bio.BIO_CTRL_PUSH:
if DEBUG: IF DEBUG:
with gil: with gil:
print("BIO_CTRL_PUSH", ret) print("BIO_CTRL_PUSH", ret)
elif cmd == bio.BIO_CTRL_POP: elif cmd == bio.BIO_CTRL_POP:
if DEBUG: IF DEBUG:
with gil: with gil:
print("BIO_CTRL_POP", ret) print("BIO_CTRL_POP", ret)
elif cmd == bio.BIO_CTRL_FLUSH: elif cmd == bio.BIO_CTRL_FLUSH:
ret = 1 ret = 1
if DEBUG: IF DEBUG:
with gil: with gil:
print('BIO_CTRL_FLUSH', ret) print('BIO_CTRL_FLUSH', ret)
elif cmd == BIO_CTRL_SET_KTLS: elif cmd == BIO_CTRL_SET_KTLS:
if DEBUG: IF DEBUG:
with gil: with gil:
print("BIO_CTRL_SET_KTLS", "TX end" if num else "RX end") print("BIO_CTRL_SET_KTLS", "TX end" if num else "RX end")
crypto_info = <ssl_h.ktls_crypto_info_t*>ptr crypto_info = <ssl_h.ktls_crypto_info_t*>ptr
@ -313,7 +319,7 @@ cdef long bio_ctrl(bio.BIO* b, int cmd, long num, void* ptr) nogil:
) == 0: ) == 0:
bio.set_flags(b, FLAGS_KTLS_TX if num else FLAGS_KTLS_RX) bio.set_flags(b, FLAGS_KTLS_TX if num else FLAGS_KTLS_RX)
else: else:
if DEBUG: IF DEBUG:
with gil: with gil:
print( print(
"BIO_CTRL_SET_KTLS", "BIO_CTRL_SET_KTLS",
@ -325,7 +331,7 @@ cdef long bio_ctrl(bio.BIO* b, int cmd, long num, void* ptr) nogil:
elif cmd == BIO_CTRL_GET_KTLS_RECV: elif cmd == BIO_CTRL_GET_KTLS_RECV:
return bio.test_flags(b, FLAGS_KTLS_RX) != 0 return bio.test_flags(b, FLAGS_KTLS_RX) != 0
else: else:
if DEBUG: IF DEBUG:
with gil: with gil:
print('bio_ctrl', cmd, num) print('bio_ctrl', cmd, num)
return ret return ret
@ -372,6 +378,7 @@ cdef class TLSTransport:
cdef: cdef:
TLSTransport rv = TLSTransport.__new__(TLSTransport) TLSTransport rv = TLSTransport.__new__(TLSTransport)
pyssl.PySSLMemoryBIO* c_bio pyssl.PySSLMemoryBIO* c_bio
pyssl.SSL* s
libc.setsockopt(fd, socket.SOL_TCP, linux.TCP_ULP, b"tls", 3) libc.setsockopt(fd, socket.SOL_TCP, linux.TCP_ULP, b"tls", 3)
@ -386,8 +393,12 @@ cdef class TLSTransport:
c_bio.bio, rv.bio = rv.bio, c_bio.bio c_bio.bio, rv.bio = rv.bio, c_bio.bio
del py_bio del py_bio
ssl_h.set_options( s = (<pyssl.PySSLSocket*>rv.sslobj._sslobj).ssl
(<pyssl.PySSLSocket*>rv.sslobj._sslobj).ssl, ssl_h.OP_ENABLE_KTLS ssl_h.set_options(s, ssl_h.OP_ENABLE_KTLS)
ssl_h.clear_mode(
s,
ssl_h.SSL_MODE_RELEASE_BUFFERS |
ssl_h.SSL_MODE_ACCEPT_MOVING_WRITE_BUFFER
) )
rv.fd = fd rv.fd = fd
rv.protocol = protocol rv.protocol = protocol
@ -409,10 +420,9 @@ cdef class TLSTransport:
self.proxy.send_msg.msg_iovlen = 1 self.proxy.send_msg.msg_iovlen = 1
self.proxy.send_callback.data = <void*>&self.proxy self.proxy.send_callback.data = <void*>&self.proxy
self.proxy.send_callback.callback = tls_send_cb self.proxy.send_callback.callback = tls_send_cb
self.proxy.recv_msg.msg_control = PyMem_RawMalloc(CMSG_SIZE) self.proxy.cmsg = PyMem_RawMalloc(CMSG_SIZE)
if self.proxy.recv_msg.msg_control == NULL: if self.proxy.cmsg == NULL:
raise MemoryError raise MemoryError
self.proxy.recv_msg.msg_controllen = CMSG_SIZE
self.proxy.recv_msg.msg_iov = &self.proxy.recv_vec self.proxy.recv_msg.msg_iov = &self.proxy.recv_vec
self.proxy.recv_msg.msg_iovlen = 1 self.proxy.recv_msg.msg_iovlen = 1
self.proxy.recv_callback.data = <void*>&self.proxy self.proxy.recv_callback.data = <void*>&self.proxy
@ -423,7 +433,7 @@ cdef class TLSTransport:
self.sslobj = None self.sslobj = None
bio.free(self.bio) bio.free(self.bio)
PyMem_RawFree(self.proxy.read_buffer) PyMem_RawFree(self.proxy.read_buffer)
PyMem_RawFree(self.proxy.recv_msg.msg_control) PyMem_RawFree(self.proxy.cmsg)
cdef do_handshake(self): cdef do_handshake(self):
if self.state == UNWRAPPED: if self.state == UNWRAPPED:
@ -431,20 +441,23 @@ cdef class TLSTransport:
elif self.state != HANDSHAKING: elif self.state != HANDSHAKING:
raise RuntimeError("Cannot do handshake now") raise RuntimeError("Cannot do handshake now")
try: try:
IF DEBUG:
print("do_handshake()")
self.sslobj.do_handshake() self.sslobj.do_handshake()
except ssl.SSLWantReadError: except ssl.SSLWantReadError:
if DEBUG: IF DEBUG:
print("do_handshake() SSLWantReadError") print("do_handshake() SSLWantReadError")
except ssl.SSLWantWriteError: except ssl.SSLWantWriteError:
if DEBUG: IF DEBUG:
print("do_handshake() SSLWantWriteError") print("do_handshake() SSLWantWriteError")
except Exception as ex: except Exception as ex:
if DEBUG: IF DEBUG:
print('do_handshake() error:', ex) print('do_handshake() error:', ex)
raise raise
else: else:
if DEBUG: IF DEBUG:
print('do_handshake() done') print('do_handshake() done')
self.state = WRAPPED self.state = WRAPPED
@ -473,14 +486,14 @@ cdef class TLSTransport:
self.proxy.flags &= ~FLAGS_PROXY_RECV_ALL self.proxy.flags &= ~FLAGS_PROXY_RECV_ALL
res = self.proxy.recv_callback.res res = self.proxy.recv_callback.res
if res < 0: if res < 0:
if DEBUG: IF DEBUG:
print("do_read_ktls() error:", -res) print("do_read_ktls() error:", -res)
self.loop.call_soon( self.loop.call_soon(
self.protocol.connection_lost, self.protocol.connection_lost,
IOError(-res, string.strerror(-res)) IOError(-res, string.strerror(-res))
) )
elif res == 0: elif res == 0:
if DEBUG: IF DEBUG:
print("do_read_ktls() EOF") print("do_read_ktls() EOF")
self.loop.call_soon(self.protocol.eof_received) self.loop.call_soon(self.protocol.eof_received)
self.loop.call_soon(self.protocol.connection_lost, None) self.loop.call_soon(self.protocol.connection_lost, None)
@ -490,10 +503,10 @@ cdef class TLSTransport:
if cmsg.cmsg_type == linux.TLS_GET_RECORD_TYPE: if cmsg.cmsg_type == linux.TLS_GET_RECORD_TYPE:
record_type = (<unsigned char*>libc.CMSG_DATA(cmsg))[0] record_type = (<unsigned char*>libc.CMSG_DATA(cmsg))[0]
if record_type != ssl_h.SSL3_RT_APPLICATION_DATA: if record_type != ssl_h.SSL3_RT_APPLICATION_DATA:
if DEBUG: IF DEBUG:
print("do_read_ktls() forward CMSG") print("do_read_ktls() forward CMSG")
return self.do_read() return self.do_read()
if DEBUG: IF DEBUG:
print("do_read_ktls() received", res, "bytes") print("do_read_ktls() received", res, "bytes")
self.loop.call_soon( self.loop.call_soon(
self.protocol.data_received, self.protocol.data_received,
@ -502,10 +515,9 @@ cdef class TLSTransport:
self.loop.call_soon(self.do_read_ktls, self) self.loop.call_soon(self.do_read_ktls, self)
elif not self.proxy.flags & FLAGS_PROXY_RECV_SUBMITTED: elif not self.proxy.flags & FLAGS_PROXY_RECV_SUBMITTED:
if DEBUG: IF DEBUG:
print("do_read_ktls() submit") print("do_read_ktls() submit")
self.proxy.recv_msg.msg_controllen = CMSG_SIZE reset_msg(&self.proxy.recv_msg, self.proxy.cmsg)
reset_msg(&self.proxy.recv_msg, CMSG_SIZE)
if not ring_sq_submit_recvmsg( if not ring_sq_submit_recvmsg(
&self.proxy.loop.ring.sq, &self.proxy.loop.ring.sq,
self.fd, self.fd,
@ -519,18 +531,18 @@ cdef class TLSTransport:
try: try:
data = self.sslobj.read(65536) data = self.sslobj.read(65536)
except ssl.SSLWantReadError: except ssl.SSLWantReadError:
if DEBUG: IF DEBUG:
print("do_read() SSLWantReadError") print("do_read() SSLWantReadError")
except ssl.SSLWantWriteError: except ssl.SSLWantWriteError:
if DEBUG: IF DEBUG:
print("do_read() SSLWantWriteError") print("do_read() SSLWantWriteError")
except Exception as ex: except Exception as ex:
if DEBUG: IF DEBUG:
print("do_read() error:", ex) print("do_read() error:", ex)
self.loop.call_soon(self.protocol.connection_lost, ex) self.loop.call_soon(self.protocol.connection_lost, ex)
else: else:
if data: if data:
if DEBUG: IF DEBUG:
print("do_read() received", len(data), bytes) print("do_read() received", len(data), bytes)
print(data) print(data)
self.loop.call_soon(self.protocol.data_received, data) self.loop.call_soon(self.protocol.data_received, data)
@ -539,7 +551,7 @@ cdef class TLSTransport:
else: else:
self.loop.call_soon(self.do_read, self) self.loop.call_soon(self.do_read, self)
else: else:
if DEBUG: IF DEBUG:
print("do_read() EOF") print("do_read() EOF")
self.loop.call_soon(self.protocol.eof_received) self.loop.call_soon(self.protocol.eof_received)
self.loop.call_soon(self.protocol.connection_lost, None) self.loop.call_soon(self.protocol.connection_lost, None)
@ -564,7 +576,7 @@ cdef class TLSTransport:
try: try:
self.sslobj.write(data) self.sslobj.write(data)
except ssl.SSLWantWriteError: except ssl.SSLWantWriteError:
if DEBUG: IF DEBUG:
print("write() SSLWantWriteError") print("write() SSLWantWriteError")