12029445091

Committed 26 Nov 2024 11:16AM UTC coverage: 98.68% (-0.002%) from 98.682%

Build # 12029445091

Build Type

Pull #275

github

Committed by

web-flow

Commit Message

Merge 008ef9603 into e529319e1

Pull Request Pull Request #275: Small cleanup

Run Details

2 of 3 new or added lines in 1 file covered. (66.67%)

7 existing lines in 2 files now uncovered.

972 of 985 relevant lines covered (98.68%)

6.85 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.11

/mocket/ssl/socket.py

from __future__ import annotations

import ssl
from datetime import datetime, timedelta
from ssl import Options
from typing import Any

from mocket.compat import encode_to_bytes
from mocket.mocket import Mocket
from mocket.socket import MocketSocket
from mocket.types import _PeerCertRetDictType


class MocketSSLSocket(MocketSocket):
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)

        self._did_handshake = False
        self._sent_non_empty_bytes = False
        self._original_socket: MocketSocket = self

    def read(self, buffersize: int | None = None) -> bytes:
        rv = self.io.read(buffersize)
        if rv:
            self._sent_non_empty_bytes = True
        if self._did_handshake and not self._sent_non_empty_bytes:
            raise ssl.SSLWantReadError("The operation did not complete (read)")
        return rv

    def write(self, data: bytes) -> int | None:
        return self.send(encode_to_bytes(data))

    def do_handshake(self) -> None:
        self._did_handshake = True

    def getpeercert(self, binary_form: bool = False) -> _PeerCertRetDictType:
        if not (self._host and self._port):
            self._address = self._host, self._port = Mocket._address

        now = datetime.now()
        shift = now + timedelta(days=30 * 12)
        return {
            "notAfter": shift.strftime("%b %d %H:%M:%S GMT"),
            "subjectAltName": (
                ("DNS", f"*.{self._host}"),
                ("DNS", self._host),
                ("DNS", "*"),
            ),
            "subject": (
                (("organizationName", f"*.{self._host}"),),
                (("organizationalUnitName", "Domain Control Validated"),),
                (("commonName", f"*.{self._host}"),),
            ),
        }

    def ciper(self) -> tuple[str, str, str]:
        return "ADH", "AES256", "SHA"

    def compression(self) -> Options:
        return ssl.OP_NO_COMPRESSION

    def unwrap(self) -> MocketSocket:
        return self._original_socket

    @classmethod
    def _create(
        cls,
        sock: MocketSocket,
        ssl_context: ssl.SSLContext | None = None,
        server_hostname: str | None = None,
        *args: Any,
        **kwargs: Any,
    ) -> MocketSSLSocket:
        ssl_socket = MocketSSLSocket()
        ssl_socket._original_socket = sock
        ssl_socket._true_socket = sock._true_socket

        if ssl_context:
            ssl_socket._true_socket = ssl_context.wrap_socket(
                sock=ssl_socket._true_socket,
                server_hostname=server_hostname,
            )

        ssl_socket._kwargs = kwargs

        ssl_socket._timeout = sock._timeout

        ssl_socket._host = sock._host
        ssl_socket._port = sock._port
        ssl_socket._address = sock._address

        ssl_socket._io = sock._io
        ssl_socket._entry = sock._entry

        return ssl_socket

1	from __future__ import annotations	7✔
2
3	import ssl	7✔
4	from datetime import datetime, timedelta	7✔
5	from ssl import Options	7✔
6	from typing import Any	7✔
7
8	from mocket.compat import encode_to_bytes	7✔
9	from mocket.mocket import Mocket	7✔
10	from mocket.socket import MocketSocket	7✔
11	from mocket.types import _PeerCertRetDictType	7✔
12
13
14	class MocketSSLSocket(MocketSocket):	7✔
15	def __init__(self, args: Any, *kwargs: Any) -> None:	7✔
16	super().__init__(args, *kwargs)	7✔
17
18	self._did_handshake = False	7✔
19	self._sent_non_empty_bytes = False	7✔
20	self._original_socket: MocketSocket = self	7✔
21
22	def read(self, buffersize: int \| None = None) -> bytes:	7✔
23	rv = self.io.read(buffersize)	7✔
24	if rv:	7✔
25	self._sent_non_empty_bytes = True	7✔
26	if self._did_handshake and not self._sent_non_empty_bytes:	7✔
UNCOV 27	raise ssl.SSLWantReadError("The operation did not complete (read)")	3✔
28	return rv	7✔
29
30	def write(self, data: bytes) -> int \| None:	7✔
31	return self.send(encode_to_bytes(data))	7✔
32
33	def do_handshake(self) -> None:	7✔
34	self._did_handshake = True	7✔
35
36	def getpeercert(self, binary_form: bool = False) -> _PeerCertRetDictType:	7✔
37	if not (self._host and self._port):	7✔
38	self._address = self._host, self._port = Mocket._address	7✔
39
40	now = datetime.now()	7✔
41	shift = now + timedelta(days=30 * 12)	7✔
42	return {	7✔
43	"notAfter": shift.strftime("%b %d %H:%M:%S GMT"),
44	"subjectAltName": (
45	("DNS", f"*.{self._host}"),
46	("DNS", self._host),
47	("DNS", "*"),
48	),
49	"subject": (
50	(("organizationName", f"*.{self._host}"),),
51	(("organizationalUnitName", "Domain Control Validated"),),
52	(("commonName", f"*.{self._host}"),),
53	),
54	}
55
56	def ciper(self) -> tuple[str, str, str]:	7✔
NEW 57	return "ADH", "AES256", "SHA"	×
58
59	def compression(self) -> Options:	7✔
60	return ssl.OP_NO_COMPRESSION	7✔
61
62	def unwrap(self) -> MocketSocket:	7✔
63	return self._original_socket	7✔
64
65	@classmethod	7✔
66	def _create(	7✔
67	cls,
68	sock: MocketSocket,
69	ssl_context: ssl.SSLContext \| None = None,
70	server_hostname: str \| None = None,
71	*args: Any,
72	**kwargs: Any,
73	) -> MocketSSLSocket:
74	ssl_socket = MocketSSLSocket()	7✔
75	ssl_socket._original_socket = sock	7✔
76	ssl_socket._true_socket = sock._true_socket	7✔
77
78	if ssl_context:	7✔
79	ssl_socket._true_socket = ssl_context.wrap_socket(	7✔
80	sock=ssl_socket._true_socket,
81	server_hostname=server_hostname,
82	)
83
84	ssl_socket._kwargs = kwargs	7✔
85
86	ssl_socket._timeout = sock._timeout	7✔
87
88	ssl_socket._host = sock._host	7✔
89	ssl_socket._port = sock._port	7✔
90	ssl_socket._address = sock._address	7✔
91
92	ssl_socket._io = sock._io	7✔
93	ssl_socket._entry = sock._entry	7✔
94
95	return ssl_socket	7✔

mindflayer / python-mocket / 12029445091

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous