1+ import os
2+ import base64
3+ import hmac as hmac_module
4+ from cryptography .hazmat .primitives .ciphers import Cipher , algorithms , modes
5+ from cryptography .hazmat .primitives import padding , hashes , hmac
6+ from cryptography .hazmat .backends import default_backend
7+ import brotli
8+
9+ HMAC_KEY = b"K9CryptHMAC2024!@#$%^&*()"
10+
11+ class K9Crypt :
12+ def __init__ (self , key : str ):
13+ hasher = hashes .Hash (hashes .SHA512 (), backend = default_backend ())
14+ hasher .update (key .encode ())
15+ self .key = hasher .finalize ()[:32 ]
16+
17+ def _generate_iv (self ) -> bytes :
18+ return os .urandom (16 )
19+
20+ def _pad (self , data : bytes ) -> bytes :
21+ padder = padding .PKCS7 (128 ).padder ()
22+ return padder .update (data ) + padder .finalize ()
23+
24+ def _unpad (self , data : bytes ) -> bytes :
25+ unpadder = padding .PKCS7 (128 ).unpadder ()
26+ return unpadder .update (data ) + unpadder .finalize ()
27+
28+ def _hash_data (self , data : bytes ) -> bytes :
29+ h = hmac .HMAC (HMAC_KEY , hashes .SHA512 (), backend = default_backend ())
30+ h .update (data )
31+ digest = h .finalize ()
32+ return digest
33+
34+ def _verify_hash (self , data : bytes , expected_hash : bytes ) -> bool :
35+ calculated_hash = self ._hash_data (data )
36+ try :
37+ return hmac_module .compare_digest (calculated_hash , expected_hash )
38+ except Exception :
39+ return False
40+
41+ def _encrypt_gcm (self , data : bytes ) -> tuple [bytes , bytes ]:
42+ iv = self ._generate_iv ()
43+ cipher = Cipher (algorithms .AES (self .key ), modes .GCM (iv ), backend = default_backend ())
44+ encryptor = cipher .encryptor ()
45+ ciphertext = encryptor .update (data ) + encryptor .finalize ()
46+ return ciphertext + encryptor .tag , iv
47+
48+ def _decrypt_gcm (self , data : bytes , iv : bytes ) -> bytes :
49+ tag = data [- 16 :]
50+ ciphertext = data [:- 16 ]
51+ cipher = Cipher (algorithms .AES (self .key ), modes .GCM (iv , tag ), backend = default_backend ())
52+ decryptor = cipher .decryptor ()
53+ return decryptor .update (ciphertext ) + decryptor .finalize ()
54+
55+ def _encrypt_cbc (self , data : bytes ) -> tuple [bytes , bytes ]:
56+ iv = self ._generate_iv ()
57+ cipher = Cipher (algorithms .AES (self .key ), modes .CBC (iv ), backend = default_backend ())
58+ encryptor = cipher .encryptor ()
59+ padded_data = self ._pad (data )
60+ return encryptor .update (padded_data ) + encryptor .finalize (), iv
61+
62+ def _decrypt_cbc (self , data : bytes , iv : bytes ) -> bytes :
63+ cipher = Cipher (algorithms .AES (self .key ), modes .CBC (iv ), backend = default_backend ())
64+ decryptor = cipher .decryptor ()
65+ padded_data = decryptor .update (data ) + decryptor .finalize ()
66+ return self ._unpad (padded_data )
67+
68+ def _encrypt_cfb (self , data : bytes ) -> tuple [bytes , bytes ]:
69+ iv = self ._generate_iv ()
70+ cipher = Cipher (algorithms .AES (self .key ), modes .CFB (iv ), backend = default_backend ())
71+ encryptor = cipher .encryptor ()
72+ return encryptor .update (data ) + encryptor .finalize (), iv
73+
74+ def _decrypt_cfb (self , data : bytes , iv : bytes ) -> bytes :
75+ cipher = Cipher (algorithms .AES (self .key ), modes .CFB (iv ), backend = default_backend ())
76+ decryptor = cipher .decryptor ()
77+ return decryptor .update (data ) + decryptor .finalize ()
78+
79+ def _encrypt_ofb (self , data : bytes ) -> tuple [bytes , bytes ]:
80+ iv = self ._generate_iv ()
81+ cipher = Cipher (algorithms .AES (self .key ), modes .OFB (iv ), backend = default_backend ())
82+ encryptor = cipher .encryptor ()
83+ return encryptor .update (data ) + encryptor .finalize (), iv
84+
85+ def _decrypt_ofb (self , data : bytes , iv : bytes ) -> bytes :
86+ cipher = Cipher (algorithms .AES (self .key ), modes .OFB (iv ), backend = default_backend ())
87+ decryptor = cipher .decryptor ()
88+ return decryptor .update (data ) + decryptor .finalize ()
89+
90+ def _encrypt_ctr (self , data : bytes ) -> tuple [bytes , bytes ]:
91+ iv = self ._generate_iv ()
92+ cipher = Cipher (algorithms .AES (self .key ), modes .CTR (iv ), backend = default_backend ())
93+ encryptor = cipher .encryptor ()
94+ return encryptor .update (data ) + encryptor .finalize (), iv
95+
96+ def _decrypt_ctr (self , data : bytes , iv : bytes ) -> bytes :
97+ cipher = Cipher (algorithms .AES (self .key ), modes .CTR (iv ), backend = default_backend ())
98+ decryptor = cipher .decryptor ()
99+ return decryptor .update (data ) + decryptor .finalize ()
100+
101+ async def _compress (self , data : bytes ) -> bytes :
102+ try :
103+ return brotli .compress (data )
104+ except Exception as e :
105+ raise ValueError (f"Compression error: { str (e )} " )
106+
107+ async def _decompress (self , data : bytes ) -> bytes :
108+ try :
109+ return brotli .decompress (data )
110+ except Exception as e :
111+ raise ValueError (f"Decompression error: { str (e )} " )
112+
113+ async def encrypt (self , plaintext : str ) -> str :
114+ data = plaintext .encode ()
115+ data = await self ._compress (data )
116+
117+ data , iv1 = self ._encrypt_gcm (data )
118+ hash1 = self ._hash_data (data )
119+ data = hash1 + data
120+
121+ data , iv2 = self ._encrypt_cbc (data )
122+ hash2 = self ._hash_data (data )
123+ data = hash2 + data
124+
125+ data , iv3 = self ._encrypt_cfb (data )
126+ hash3 = self ._hash_data (data )
127+ data = hash3 + data
128+
129+ data , iv4 = self ._encrypt_ofb (data )
130+ hash4 = self ._hash_data (data )
131+ data = hash4 + data
132+
133+ data , iv5 = self ._encrypt_ctr (data )
134+ hash5 = self ._hash_data (data )
135+ data = hash5 + data
136+
137+ combined = iv1 + iv2 + iv3 + iv4 + iv5 + data
138+
139+ return base64 .b64encode (combined ).decode ()
140+
141+ async def decrypt (self , ciphertext : str ) -> str :
142+ data = base64 .b64decode (ciphertext )
143+
144+ iv1 = data [:16 ]
145+ iv2 = data [16 :32 ]
146+ iv3 = data [32 :48 ]
147+ iv4 = data [48 :64 ]
148+ iv5 = data [64 :80 ]
149+ data = data [80 :]
150+
151+ hash5 = data [:64 ]
152+ data = data [64 :]
153+ if not self ._verify_hash (data , hash5 ):
154+ raise ValueError ("Layer 5 integrity check failed" )
155+ data = self ._decrypt_ctr (data , iv5 )
156+
157+ hash4 = data [:64 ]
158+ data = data [64 :]
159+ if not self ._verify_hash (data , hash4 ):
160+ raise ValueError ("Layer 4 integrity check failed" )
161+ data = self ._decrypt_ofb (data , iv4 )
162+
163+ hash3 = data [:64 ]
164+ data = data [64 :]
165+ if not self ._verify_hash (data , hash3 ):
166+ raise ValueError ("Layer 3 integrity check failed" )
167+ data = self ._decrypt_cfb (data , iv3 )
168+
169+ hash2 = data [:64 ]
170+ data = data [64 :]
171+ if not self ._verify_hash (data , hash2 ):
172+ raise ValueError ("Layer 2 integrity check failed" )
173+ data = self ._decrypt_cbc (data , iv2 )
174+
175+ hash1 = data [:64 ]
176+ data = data [64 :]
177+ if not self ._verify_hash (data , hash1 ):
178+ raise ValueError ("Layer 1 integrity check failed" )
179+ data = self ._decrypt_gcm (data , iv1 )
180+
181+ data = await self ._decompress (data )
182+ return data .decode ()
0 commit comments