SantaSpeen/winConnect
1
0
Fork 0
mirror of https://github.com/SantaSpeen/winConnect.git synced 2025-07-01 07:25:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

[!] Update protocol: v3 (with crypto+salt)

Browse Source 
[~] WinConnectCryptoPassword
[+] WinConnectCryptoBase.salt (+setter)
[~] Optimize WinConnectCryptoSimple.encrypt
[+] pycryptodome
This commit is contained in:
Maxim Khomutov 2025-03-13 17:02:41 +03:00
parent d937479aa8
commit d8a4d1682c
8 changed files with 110 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
requirements.txt
View File

@ -2,3 +2,4 @@
# 12.03.2025
pywin32~=309
ormsgpack~=1.8.0
pycryptodome~=3.21.0

17
setup.py
View File

@ -13,10 +13,17 @@ packages = [_name, ]
package_dir = {_name: _name}
lib_path = here / _name
requires = [
"pywin32==309",
"ormsgpack==1.8.0"
]
requires = {
"install_requires": [
"pywin32==309",
"ormsgpack==1.8.0"
],
"extra_packages": {
"crypto": [
"pycryptodome==3.21.0"
]
}
}
# 'setup.py publish' shortcut.
if sys.argv[-1] == 'publish':
@ -44,7 +51,7 @@ setup(
package_data={'': ['LICENSE']},
package_dir=package_dir,
include_package_data=True,
install_requires=requires,
**requires,
license=about['__license__'],
classifiers=[
"Development Status :: 5 - Production/Stable",

40
winConnect/WinConnectBase.py
View File

@ -25,7 +25,7 @@ class WinConnectBase:
default_encoding = 'utf-8'
read_max_buffer = SimpleConvertor.to_gb(3)-1 # Max size via chunked messages
read_max_buffer = SimpleConvertor.to_gb(3)-32 # Max size via chunked messages
ormsgpack_options = ormsgpack.OPT_NON_STR_KEYS | ormsgpack.OPT_NAIVE_UTC | ormsgpack.OPT_PASSTHROUGH_TUPLE # ormsgpack options
@ -33,8 +33,9 @@ class WinConnectBase:
self._log = logging.getLogger(f"WinConnect:{pipe_name}")
# _version:
# 1 - 0.9.1
# 2 - 0.9.2+ (with crypto)
self._version = 2
# 2 - 0.9.2 (with crypto)
# 3 - 0.9.3+ (with crypto+salt)
self._version = 3
self._pipe_name = r'\\.\pipe\{}'.format(pipe_name)
self._pipe = None
self._opened = False
@ -191,35 +192,46 @@ class WinConnectBase:
return self._send_error(WinConnectErrors.UNKNOWN_ACTION, f"Unknown action '{action}'")
def _parse_command(self, data: bytes):
_blank_settings = {
'version': self._version,
'encoding': self.default_encoding,
'header_size': self._header_size,
'header_format': self._header_format,
'max_buffer': self.read_max_buffer,
'crypto': self.__crypto.crypt_name,
'salt': self.__crypto.crypt_salt
}
command, data = self.__parse_message(data)
match command:
case b'get_session_settings':
self._log.debug(f"[{self._pipe_name}] Received get_session_settings from {data}")
settings = {
'version': self._version,
'encoding': self.default_encoding,
'header_size': self._header_size,
'header_format': self._header_format,
'max_buffer': self.read_max_buffer,
"crypto": self.__crypto.get_info()
}
session_settings = f"set_session_settings:{json.dumps(settings)}".encode(self.init_encoding)
session_settings = f"set_session_settings:{json.dumps(_blank_settings)}".encode(self.encoding)
self._send_message("command", session_settings)
return True
case b'set_session_settings':
self._log.debug(f"[{self._pipe_name}] Received session settings.")
try:
settings = json.loads(data.decode(self.init_encoding))
except json.JSONDecodeError as e:
self._send_error(WinConnectErrors.BAD_DATA, f"JSONDecodeError: {e}")
return self.close()
if settings.get('version') != self._version:
if _blank_settings.keys() != settings.keys():
self._log.error(f"{WinConnectErrors.BAD_SETTINGS}")
self._send_error(WinConnectErrors.BAD_SETTINGS, f"Setting have wrong structure")
return self.close()
if settings['version'] != self._version:
self._log.error(f"{WinConnectErrors.BAD_VERSION}")
self._send_error(WinConnectErrors.BAD_VERSION, f"Version mismatch")
return self.close()
if settings.get('crypto') != self.__crypto.get_info():
if settings['crypto'] != self.__crypto.crypt_name:
self._log.error(f"{WinConnectErrors.BAD_CRYPTO}")
self._send_error(WinConnectErrors.BAD_CRYPTO, f"Crypto mismatch")
return self.close()
if settings['salt'] != self.__crypto.crypt_salt:
self._log.debug(f"[{self._pipe_name}] Updating salt")
self.__crypto.set_salt(settings['salt'])
self._session_encoding = settings.get('encoding', self.default_encoding)
self._header_size = settings.get('header_size', self._header_size)
self._header_format = settings.get('header_format', self._header_format)

20
winConnect/crypto/WinConnectCrypto.py
View File

@ -15,15 +15,26 @@ def test_crypto_class(crypto_class: "WinConnectCryptoBase"):
class WinConnectCrypto:
def __init__(self):
self.__crypto_class = None
self.__crypto_class: WinConnectCryptoBase = None
self.__log_prefix = None
self._log = logging.getLogger("WinConnectCrypto")
@property
def crypt_name(self) -> str:
return self.__crypto_class.__class__.__name__
@property
def crypt_salt(self) -> bytes:
return self.__crypto_class.salt
def set_salt(self, salt: bytes):
self.__crypto_class.salt = salt
def set_crypto_class(self, crypto_class: "WinConnectCryptoBase"):
self._log.debug(f"{self.__log_prefix}Updating crypto class. {self.get_info()} -> {crypto_class.__class__.__name__}")
self._log.debug(f"{self.__log_prefix}Updating crypto class. {self.crypt_name} -> {crypto_class.__class__.__name__}")
test_crypto_class(crypto_class)
self.__crypto_class = crypto_class
self.__log_prefix = f"[{self.get_info()}] "
self.__log_prefix = f"[{self.crypt_name}] "
self._log.debug(f"{self.__log_prefix}Crypto class updated.")
def set_logger(self, logger):
@ -38,9 +49,6 @@ class WinConnectCrypto:
self._log.debug(f"{self.__log_prefix}Crypto class loaded")
return True
def get_info(self):
return self.__crypto_class.__class__.__name__
def encrypt(self, data: bytes) -> bytes:
return self.__crypto_class.encrypt(data)

4
winConnect/crypto/__init__.py
View File

@ -3,6 +3,6 @@ from .crypto_classes import (
WinConnectCryptoBase,
WinConnectCryptoNone,
WinConnectCryptoSimple,
# WinConnectCryptoPassword,
# WinConnectCryptoCert
WinConnectCryptoPassword,
WinConnectCryptoCert
)

6
winConnect/crypto/crypto_class_base.py
View File

@ -1,5 +1,11 @@
class WinConnectCryptoBase:
@property
def salt(self) -> bytes:
return b""
@salt.setter
def salt(self, value): ...
def encrypt(self, data: bytes) -> bytes: ...
def decrypt(self, data: bytes) -> bytes: ...

56
winConnect/crypto/crypto_classes.py
View File

@ -1,8 +1,18 @@
import os
import random
from pathlib import Path
from .crypto_class_base import WinConnectCryptoBase
from winConnect.exceptions import WinConnectCryptoSimpleBadHeaderException
_pip_crypto = True
try:
from Crypto.Cipher import AES
from Crypto.Protocol.KDF import PBKDF2
from Crypto.Cipher import PKCS1_OAEP
except ImportError:
_pip_crypto = False
class WinConnectCryptoNone(WinConnectCryptoBase):
def __init__(self): ...
@ -19,12 +29,10 @@ class WinConnectCryptoSimple(WinConnectCryptoBase):
shift_key = random.randint(100, 749)
key = random.randint(5, 250)
encrypted_text = bytearray()
header = f"wccs{shift_key}{key+shift_key}:"
for char in header:
encrypted_text.append(ord(char))
header = f"wccs{shift_key}{key+shift_key}:".encode()
for char in data:
encrypted_text.append(char ^ key)
return bytes(encrypted_text)
return header + bytes(encrypted_text)
def decrypt(self, data: bytes) -> bytes:
try:
@ -44,17 +52,49 @@ class WinConnectCryptoSimple(WinConnectCryptoBase):
class WinConnectCryptoPassword(WinConnectCryptoBase):
def __init__(self, password: str):
pass
if not _pip_crypto:
raise ImportError("Crypto library not installed. Install with 'pip install winConnect[crypto]'")
self.password = password
self.__salt = os.urandom(16)
self.__key = PBKDF2(password, self.__salt, dkLen=32, count=100000)
@property
def salt(self):
return self.__salt
@salt.setter
def salt(self, value):
self.__salt = value
self.__key = PBKDF2(self.password, self.__salt, dkLen=32, count=100000)
def encrypt(self, data: bytes) -> bytes:
pass
iv = os.urandom(16) # Генерируем IV
cipher = AES.new(self.__key, AES.MODE_CBC, iv)
pad_len = 16 - len(data) % 16
data += chr(pad_len) * pad_len
header = f"wccp{iv.hex()}:".encode()
return header + cipher.encrypt(data) # Шифруем
def decrypt(self, data: bytes) -> bytes:
pass
try:
header, iv, content = data[:4], data[4:20], data[20:]
if header[:4] != b"wccp":
raise WinConnectCryptoSimpleBadHeaderException("Bad header in message.")
except ValueError:
raise WinConnectCryptoSimpleBadHeaderException("No header in message.")
cipher = AES.new(self.__key, AES.MODE_CBC, iv)
decrypted = cipher.decrypt(data)
pad_len = decrypted[-1] # Убираем PKCS7 padding
return decrypted[:-pad_len]
class WinConnectCryptoCert(WinConnectCryptoBase):
def __init__(self, cert_file: str):
pass
if not _pip_crypto:
raise ImportError("Crypto library not installed. Install with 'pip install winConnect[crypto]'")
self.cert_file = Path(cert_file)
def _open_cert(self):
pass

2
winConnect/errors.py
View File

@ -14,7 +14,7 @@ class WinConnectErrors(Enum):
BAD_DATA = 50
BAD_VERSION = 51
BAD_HEADER = 52
BAD_BODY = 53
BAD_SETTINGS = 53
BAD_CRYPTO = 54
BODY_TOO_BIG = 60