11import base64
22import binascii
3+ import hmac
4+ import hashlib
35import itertools
46import string
57import random
1719
1820import lazy_import
1921
20- jwt = lazy_import .lazy_module ("jwt" )
2122pgpy = lazy_import .lazy_module ("pgpy" )
2223
2324
@@ -424,9 +425,15 @@ def jwt_decode(self) -> EncryptionEncodingT:
424425 Returns:
425426 Chepy: The Chepy object.
426427 """
428+ def _b64decode_pad (s ):
429+ s += "=" * (- len (s ) % 4 )
430+ return json .loads (base64 .urlsafe_b64decode (s ))
431+
432+ token = self ._convert_to_str ()
433+ parts = token .split ("." )
427434 self .state = {
428- "payload " : jwt . decode ( self . _convert_to_str (), verify = False ),
429- "header " : jwt . get_unverified_header ( self . _convert_to_str () ),
435+ "header " : _b64decode_pad ( parts [ 0 ] ),
436+ "payload " : _b64decode_pad ( parts [ 1 ] ),
430437 }
431438 return self
432439
@@ -443,9 +450,23 @@ def jwt_verify(
443450 Returns:
444451 Chepy: The Chepy object.
445452 """
446- self .state = jwt .decode (
447- self ._convert_to_str (), key = secret , algorithms = algorithm
448- )
453+ token = self ._convert_to_str ().strip ().replace (" " , "" )
454+ parts = token .split ("." )
455+ assert len (parts ) == 3 , "Invalid JWT format"
456+
457+ alg = algorithm [0 ] if isinstance (algorithm , list ) else algorithm
458+ hash_map = {"HS256" : hashlib .sha256 , "HS384" : hashlib .sha384 , "HS512" : hashlib .sha512 }
459+ assert alg in hash_map , f"Unsupported algorithm: { alg } "
460+
461+ signing_input = f"{ parts [0 ]} .{ parts [1 ]} " .encode ()
462+ expected_sig = base64 .urlsafe_b64encode (
463+ hmac .new (secret .encode (), signing_input , hash_map [alg ]).digest ()
464+ ).rstrip (b"=" )
465+ actual_sig = parts [2 ].encode ()
466+ assert hmac .compare_digest (expected_sig , actual_sig ), "Signature verification failed"
467+
468+ padded = parts [1 ] + "=" * (- len (parts [1 ]) % 4 )
469+ self .state = json .loads (base64 .urlsafe_b64decode (padded ))
449470 return self
450471
451472 @ChepyDecorators .call_stack
@@ -463,7 +484,17 @@ def jwt_sign(self, secret: str, algorithms: str = "HS256") -> EncryptionEncoding
463484 data = self .state
464485 elif isinstance (self .state , str ):
465486 data = json .loads (self .state )
466- self .state = jwt .encode (data , key = secret , algorithm = algorithms )
487+
488+ hash_map = {"HS256" : hashlib .sha256 , "HS384" : hashlib .sha384 , "HS512" : hashlib .sha512 }
489+ assert algorithms in hash_map , f"Unsupported algorithm: { algorithms } "
490+
491+ header = base64 .urlsafe_b64encode (json .dumps ({"typ" : "JWT" , "alg" : algorithms }, separators = ("," , ":" )).encode ()).rstrip (b"=" ).decode ()
492+ payload = base64 .urlsafe_b64encode (json .dumps (data , separators = ("," , ":" )).encode ()).rstrip (b"=" ).decode ()
493+ signing_input = f"{ header } .{ payload } " .encode ()
494+ sig = base64 .urlsafe_b64encode (
495+ hmac .new (secret .encode (), signing_input , hash_map [algorithms ]).digest ()
496+ ).rstrip (b"=" ).decode ()
497+ self .state = f"{ header } .{ payload } .{ sig } "
467498 return self
468499
469500 @ChepyDecorators .call_stack
0 commit comments