better integration and error handling. added tests

Author: mdulaney

src/OpenSSL/SSL.py

@@ -823,38 +823,41 @@ class Session:
     .. versionadded:: 0.14
     """
 
-    _session: Any
+    _session: Any = None
 
+    def __init__(self, data: Optional[bytes] = None) -> None:
+        if data is None:
+            return
+
+        p = _ffi.new("unsigned char[]", data)
+        pp = _ffi.new("unsigned char **")
+        pp[0] = p
+        length = _ffi.cast("long", len(data))
 
-def d2i_SSL_SESSION(data: bytes) -> Session:
-    p = _ffi.new("unsigned char[]", data)
-    pp = _ffi.new("unsigned char **")
-    pp[0] = p
-    length = _ffi.cast("long", len(data))
+        session = _lib.d2i_SSL_SESSION(_ffi.NULL, pp, length)
+        if session == _ffi.NULL:
+            _raise_current_error()
 
-    session = _lib.d2i_SSL_SESSION(_ffi.NULL, pp, length)
-    if session == _ffi.NULL:
-        raise RuntimeError("d2i_SSL_SESSION failed")
+        self._session = _ffi.gc(session, _lib.SSL_SESSION_free)
 
-    pysession = Session.__new__(Session)
-    pysession._session = _ffi.gc(session, _lib.SSL_SESSION_free)
-    return pysession
+    def i2d(self) -> bytes:
+        if self._session is None:
+            raise ValueError("Not a valid session")
 
-def i2d_SSL_SESSION(session: Session) -> bytes:
+        length = _lib.i2d_SSL_SESSION(self._session, _ffi.NULL)
+        if length == 0:
+            raise ValueError("Not a valid session")
 
-    length = _lib.i2d_SSL_SESSION(session._session, _ffi.NULL)
-    if length == 0:
-        raise RuntimeError("i2d_SSL_SESSION failed to get length")
+        pp = _ffi.new("unsigned char **")
+        p = _ffi.new("unsigned char[]", length)
+        pp[0] = p
 
-    pp = _ffi.new("unsigned char **")
-    p = _ffi.new("unsigned char[]", length)
-    pp[0] = p
+        length = _lib.i2d_SSL_SESSION(self._session, pp)
+        if length == 0:
+            raise ValueError("Not a valid session")
 
-    length = _lib.i2d_SSL_SESSION(session._session, pp)
-    if length == 0:
-        raise RuntimeError("i2d_SSL_SESSION failed to produce DER")
+        return _ffi.buffer(p, length)[:]
 
-    return _ffi.buffer(p, length)[:]
 
 class Context:
     """

tests/test_ssl.py

@@ -299,17 +299,21 @@ def _create_certificate_chain():
     return [(cakey, cacert), (ikey, icert), (skey, scert)]
 
 
-def loopback_client_factory(socket, version=SSLv23_METHOD):
+def loopback_client_factory(socket, version=SSLv23_METHOD, session_data=None):
     client = Connection(Context(version), socket)
+    if session_data is not None:
+        client.set_session(Session(session_data))
     client.set_connect_state()
     return client
 
 
-def loopback_server_factory(socket, version=SSLv23_METHOD):
+def loopback_server_factory(socket, version=SSLv23_METHOD, session_data=None):
     ctx = Context(version)
     ctx.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem))
     ctx.use_certificate(load_certificate(FILETYPE_PEM, server_cert_pem))
     server = Connection(ctx, socket)
+    if session_data is not None:
+        server.set_session(Session(session_data))
     server.set_accept_state()
     return server
 
@@ -2154,6 +2158,67 @@ def test_construction(self):
         new_session = Session()
         assert isinstance(new_session, Session)
 
+    def test_d2i_fail(self):
+        with pytest.raises(Error) as e:
+            Session(b"abc" * 1000)
+
+        assert e.value.args[0][0] in [
+            # TODO
+            # 1.1.x
+            # (
+            #     "SSL routines",
+            #     "SSL_CTX_set_session_id_context",
+            #     "ssl session id context too long",
+            # ),
+            # 3.0.x
+            (
+                "asn1 encoding routines",
+                "",
+                "wrong tag",
+            ),
+        ]
+
+        assert e.value.args[0][1] in [
+            # TODO
+            # 1.1.x
+            # (
+            #     "SSL routines",
+            #     "SSL_CTX_set_session_id_context",
+            #     "ssl session id context too long",
+            # ),
+            # 3.0.x
+            (
+                "asn1 encoding routines",
+                "",
+                "nested asn1 error",
+            ),
+        ]
+
+    def test_session_success(self):
+        session_id = (
+            b"\x51\x6d\x1d\x18\xc3\xb5\x86\x81\xc6\x79\x89\x2c\x89\x3e\x56\x33"
+            b"\xa7\x9c\xcd\x9b\x87\xbb\xb3\xdc\xf6\x76\x70\xf9\xc0\xdd\xf4\xef"
+        )
+
+        session_data = (
+            b"\x30\x71\x02\x01\x01\x02\x02\x03\x03\x04\x02\xc0\x30\x04\x20" +
+            session_id +
+            b"\x04\x30\x0f\xb2\x51\xe3\x15\x60\x2d\xef\x6e\x6d\xd2\x94\x2d\xe5" +
+            b"\x37\x96\x72\xfa\xce\xb0\x39\xcc\x8d\xdf\xab\x32\xcc\x75\x0c\x66" +
+            b"\xf9\xfd\xef\xbc\xc6\x2a\x8f\x9c\x35\x16\xfd\x4d\x38\xd9\xf9\xeb" +
+            b"\x1d\xe4\xa1\x06\x02\x04\x66\xec\x4c\x2d\xa2\x04\x02\x02\x02\x58" +
+            b"\xa4\x02\x04\x00"
+        )
+        serverSocket, clientSocket = socket_pair()
+
+        client = loopback_client_factory(clientSocket, session_data=session_data)
+        server = loopback_server_factory(serverSocket, session_data=session_data)
+
+        handshake(client, server)
+
+        client.send(b"hello world")
+        assert b"hello world" == server.recv(len(b"hello world"))
+        
 
 @pytest.fixture(params=["context", "connection"])
 def ctx_or_conn(request) -> Union[Context, Connection]: