mirror of
https://github.com/gentoo-mirror/guru.git
synced 2026-07-16 02:23:16 -04:00
d4def4496b
* add patch for `dev-python/pycryptodome` and remove test that relies on `dev-python/gevent` Bug: https://bugs.gentoo.org/879011 Bug: https://bugs.gentoo.org/910098 Signed-off-by: Gonçalo Negrier Duarte <gonegrier.duarte@gmail.com>
189 lines
7.1 KiB
Diff
189 lines
7.1 KiB
Diff
From dd8334ce47dbb0c5cc5547094e9800e7789821ce Mon Sep 17 00:00:00 2001
|
|
From: =?UTF-8?q?Gon=C3=A7alo=20Negrier=20Duarte?=
|
|
<gonegrier.duarte@gmail.com>
|
|
Date: Thu, 14 Aug 2025 15:30:13 +0100
|
|
Subject: [PATCH] dev-python: patched for `dev-python/pycryptodome` and remove
|
|
test depend on `dev-python/gevent`
|
|
MIME-Version: 1.0
|
|
Content-Type: text/plain; charset=UTF-8
|
|
Content-Transfer-Encoding: 8bit
|
|
|
|
* Since `dev-python/gevent` was removed in BUG 879011 so
|
|
tests/test_core_cm.py need to be removed
|
|
|
|
Signed-off-by: Gonçalo Negrier Duarte <gonegrier.duarte@gmail.com>
|
|
---
|
|
steam/core/crypto.py | 8 +--
|
|
tests/test_core_cm.py | 140 ------------------------------------------
|
|
2 files changed, 4 insertions(+), 144 deletions(-)
|
|
delete mode 100644 tests/test_core_cm.py
|
|
|
|
diff --git a/steam/core/crypto.py b/steam/core/crypto.py
|
|
index 9557ee6..7894dcf 100644
|
|
--- a/steam/core/crypto.py
|
|
+++ b/steam/core/crypto.py
|
|
@@ -6,10 +6,10 @@ from os import urandom as random_bytes
|
|
from struct import pack
|
|
from base64 import b64decode
|
|
|
|
-from Cryptodome.Hash import MD5, SHA1, HMAC
|
|
-from Cryptodome.PublicKey.RSA import import_key as rsa_import_key, construct as rsa_construct
|
|
-from Cryptodome.Cipher import PKCS1_OAEP, PKCS1_v1_5
|
|
-from Cryptodome.Cipher import AES as AES
|
|
+from Crypto.Hash import MD5, SHA1, HMAC
|
|
+from Crypto.PublicKey.RSA import import_key as rsa_import_key, construct as rsa_construct
|
|
+from Crypto.Cipher import PKCS1_OAEP, PKCS1_v1_5
|
|
+from Crypto.Cipher import AES as AES
|
|
|
|
|
|
class UniverseKey:
|
|
diff --git a/tests/test_core_cm.py b/tests/test_core_cm.py
|
|
deleted file mode 100644
|
|
index bbc2c27..0000000
|
|
--- a/tests/test_core_cm.py
|
|
+++ /dev/null
|
|
@@ -1,140 +0,0 @@
|
|
-import unittest
|
|
-from unittest.mock import patch
|
|
-import gevent
|
|
-import gevent.queue
|
|
-
|
|
-from steam.core.cm import CMClient
|
|
-
|
|
-class CMClient_Scenarios(unittest.TestCase):
|
|
- test_channel_key = b'SESSION KEY LOL'
|
|
-
|
|
- def setUp(self):
|
|
- # mock out crypto
|
|
- patcher = patch('steam.core.crypto.generate_session_key')
|
|
- self.addCleanup(patcher.stop)
|
|
- self.gen_skey = patcher.start()
|
|
- self.gen_skey.return_value = (self.test_channel_key, b'PUBKEY ENCRYPTED SESSION KEY')
|
|
-
|
|
- patcher = patch('steam.core.crypto.symmetric_encrypt')
|
|
- self.addCleanup(patcher.stop)
|
|
- self.s_enc = patcher.start()
|
|
- self.s_enc.side_effect = lambda m, k: m
|
|
- patcher = patch('steam.core.crypto.symmetric_encrypt_HMAC')
|
|
- self.addCleanup(patcher.stop)
|
|
- self.s_enc_hmac = patcher.start()
|
|
- self.s_enc_hmac.side_effect = lambda m, k, mac: m
|
|
-
|
|
- patcher = patch('steam.core.crypto.symmetric_decrypt')
|
|
- self.addCleanup(patcher.stop)
|
|
- self.s_dec = patcher.start()
|
|
- self.s_dec.side_effect = lambda c, k: c
|
|
- patcher = patch('steam.core.crypto.symmetric_decrypt_HMAC')
|
|
- self.addCleanup(patcher.stop)
|
|
- self.s_dec_hmac = patcher.start()
|
|
- self.s_dec_hmac.side_effect = lambda c, k, mac: c
|
|
-
|
|
- # TODO: Tests for WebsocketConnection
|
|
-
|
|
- # mock out TCPConnection
|
|
- patcher = patch('steam.core.cm.TCPConnection', autospec=True)
|
|
- self.addCleanup(patcher.stop)
|
|
- self.conn = patcher.start().return_value
|
|
-
|
|
- self.conn_in = gevent.queue.Queue()
|
|
- self.conn.__iter__.return_value = self.conn_in
|
|
-
|
|
- # mock out CMServerList
|
|
- patcher = patch('steam.core.cm.CMServerList', autospec=True)
|
|
- self.addCleanup(patcher.stop)
|
|
- self.server_list = patcher.start().return_value
|
|
- self.server_list.__iter__.return_value = [(127001, 20000+i) for i in range(10)]
|
|
- self.server_list.bootstrap_from_webapi.return_value = False
|
|
- self.server_list.bootstrap_from_dns.return_value = False
|
|
-
|
|
- @patch.object(CMClient, 'emit')
|
|
- @patch.object(CMClient, '_recv_messages')
|
|
- def test_connect(self, mock_recv, mock_emit):
|
|
- # setup
|
|
- self.conn.connect.return_value = True
|
|
- self.server_list.__len__.return_value = 10
|
|
-
|
|
- # run
|
|
- cm = CMClient()
|
|
-
|
|
- with gevent.Timeout(2, False):
|
|
- cm.connect(retry=1)
|
|
-
|
|
- gevent.idle()
|
|
-
|
|
- # verify
|
|
- self.conn.connect.assert_called_once_with((127001, 20000))
|
|
- mock_emit.assert_called_once_with('connected')
|
|
- mock_recv.assert_called_once_with()
|
|
-
|
|
- @patch.object(CMClient, 'emit')
|
|
- @patch.object(CMClient, '_recv_messages')
|
|
- def test_connect_auto_discovery_failing(self, mock_recv, mock_emit):
|
|
- # setup
|
|
- self.conn.connect.return_value = True
|
|
- self.server_list.__len__.return_value = 0
|
|
-
|
|
- # run
|
|
- cm = CMClient()
|
|
-
|
|
- with gevent.Timeout(3, False):
|
|
- cm.connect(retry=1)
|
|
-
|
|
- gevent.idle()
|
|
-
|
|
- # verify
|
|
- self.server_list.bootstrap_from_webapi.assert_called_once_with()
|
|
- self.server_list.bootstrap_from_dns.assert_called_once_with()
|
|
- self.conn.connect.assert_not_called()
|
|
-
|
|
- @patch.object(CMClient, 'emit')
|
|
- @patch.object(CMClient, '_recv_messages')
|
|
- def test_connect_auto_discovery_success(self, mock_recv, mock_emit):
|
|
- # setup
|
|
- self.conn.connect.return_value = True
|
|
- self.server_list.__len__.return_value = 0
|
|
-
|
|
- def fake_servers(*args, **kwargs):
|
|
- self.server_list.__len__.return_value = 10
|
|
- return True
|
|
-
|
|
- self.server_list.bootstrap_from_webapi.side_effect = fake_servers
|
|
-
|
|
- # run
|
|
- cm = CMClient()
|
|
-
|
|
- with gevent.Timeout(3, False):
|
|
- cm.connect(retry=1)
|
|
-
|
|
- gevent.idle()
|
|
-
|
|
- # verify
|
|
- self.server_list.bootstrap_from_webapi.assert_called_once_with()
|
|
- self.server_list.bootstrap_from_dns.assert_not_called()
|
|
- self.conn.connect.assert_called_once_with((127001, 20000))
|
|
- mock_emit.assert_called_once_with('connected')
|
|
- mock_recv.assert_called_once_with()
|
|
-
|
|
- def test_channel_encrypt_sequence(self):
|
|
- # setup
|
|
- self.conn.connect.return_value = True
|
|
-
|
|
- # run ------------
|
|
- cm = CMClient()
|
|
- cm.connected = True
|
|
- gevent.spawn(cm._recv_messages)
|
|
-
|
|
- # recieve ChannelEncryptRequest
|
|
- self.conn_in.put(b'\x17\x05\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\x01\x00\x00\x00\x01\x00\x00\x00')
|
|
- gevent.idle(); gevent.idle(); gevent.idle(); gevent.idle()
|
|
-
|
|
- self.conn.put_message.assert_called_once_with(b'\x18\x05\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\x01\x00\x00\x00\x80\x00\x00\x00PUBKEY ENCRYPTED SESSION KEY\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00h-\xc4@\x00\x00\x00\x00')
|
|
-
|
|
- # recieve ChannelEncryptResult (OK)
|
|
- self.conn_in.put(b'\x19\x05\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\x01\x00\x00\x00')
|
|
-
|
|
- cm.wait_event('channel_secured', timeout=2, raises=True)
|
|
--
|
|
2.50.0
|
|
|