Merge branch 'feature/CHINARPA-5092'

冯轩
Showing 3 changed files with 101 additions and 7 deletions
src/celery_compare/tasks.py
src/common/tools/aes.py
src/common/tools/aes_util.py
--- a/src/celery_compare/tasks.py
View file @1a883cc
+++ b/src/celery_compare/tasks.py
View file @1a883cc
@@ -48,7 +48,8 @@ from apps.doc.ocr.cms import cms
 from apps.doc.exceptions import GCAPException
 from apps.doc.named_enum import RequestTeam, RequestTrigger, ProcessName, ErrorType, SystemName
 from common.tools.comparison import cp
-from common.tools.des import decode_des
+# from common.tools.des import decode_des
+from common.tools.aes_util import aes_decrypt_cbc
 import threading
 import concurrent.futures
@@ -62,6 +63,7 @@ log_base = '[Compare]'
 empty_str = ''
 empty_error_type = 1000
 des_key = conf.CMS_DES_KEY
+des_iv = conf.CMS_DES_IV
 def rotate_bound(image, angle):
@@ -985,7 +987,7 @@ def get_se_cms_compare_info_auto(application_id, last_obj, application_entity, d
            if id_info.get('idType') in consts.SE_CMS_FIRST_ID_FIELD_MAPPING:
                license_en, is_prc = consts.SE_CMS_FIRST_ID_FIELD_MAPPING[id_info['idType']]
                # ['customerName', 'idNum', 'dateOfBirth', 'idExpiryDate', 'hukouProvince']
-                id_num = decode_des(id_info.get('idNum', ''), des_key)
+                id_num = aes_decrypt_cbc(id_info.get('idNum', ''), des_key, des_iv)
                field_input = [('customerName', customer_name), ('idNum', id_num),
                               ('idExpiryDate', id_info.get('idExpiryDate', ''))]
                # if is_prc:
@@ -997,7 +999,7 @@ def get_se_cms_compare_info_auto(application_id, last_obj, application_entity, d
            elif id_info.get('idType') in ['Unified Social Credit Code', 'Tax Number', 'Business License Number']:
                # ['companyName', 'legalRepName', 'businessLicenseNo', 'organizationCreditCode',
                #  'taxRegistrationCertificateNo', 'establishmentDate', 'businessLicenseDueDate']
-                id_num = decode_des(id_info.get('idNum', ''), des_key)
+                id_num = aes_decrypt_cbc(id_info.get('idNum', ''), des_key, des_iv)
                bl_field_input = [
                    ('companyName', customer_name),
                    ('legalRepName', legal_name),
@@ -1186,7 +1188,7 @@ def get_se_cms_compare_info_auto(application_id, last_obj, application_entity, d
    # 银行卡-------------------------------------------------------------------------------------------------------
    bank_info = {}
    bank_name = cms_info.get('bankAccountDetails', {}).get('bankName', '')
-    account_no = decode_des(cms_info.get('bankAccountDetails', {}).get('accountNo', ''), des_key)
+    account_no = aes_decrypt_cbc(cms_info.get('bankAccountDetails', {}).get('accountNo', ''), des_key, des_iv)
    account_holder_name = cms_info.get('bankAccountDetails', {}).get('accountHolderName', '')
    is_gsyh = True if '工商' in bank_name else False
@@ -1676,7 +1678,7 @@ def get_se_cms_compare_info(application_id, last_obj, application_entity, detect
            if id_info.get('idType') in consts.SE_CMS_FIRST_ID_FIELD_MAPPING:
                license_en, is_prc = consts.SE_CMS_FIRST_ID_FIELD_MAPPING[id_info['idType']]
                # ['customerName', 'idNum', 'dateOfBirth', 'idExpiryDate', 'hukouProvince']
-                id_num = decode_des(id_info.get('idNum', ''), des_key)
+                id_num = aes_decrypt_cbc(id_info.get('idNum', ''), des_key, des_iv)
                field_input = [('customerName', customer_name), ('idNum', id_num),
                               ('idExpiryDate', id_info.get('idExpiryDate', ''))]
                # if is_prc:
@@ -1688,7 +1690,7 @@ def get_se_cms_compare_info(application_id, last_obj, application_entity, detect
            elif id_info.get('idType') in ['Unified Social Credit Code', 'Tax Number', 'Business License Number']:
                # ['companyName', 'legalRepName', 'businessLicenseNo', 'organizationCreditCode',
                #  'taxRegistrationCertificateNo', 'establishmentDate', 'businessLicenseDueDate']
-                id_num = decode_des(id_info.get('idNum', ''), des_key)
+                id_num = aes_decrypt_cbc(id_info.get('idNum', ''), des_key, des_iv)
                bl_field_input = [
                    ('companyName', customer_name),
                    ('legalRepName', legal_name),
@@ -1877,7 +1879,7 @@ def get_se_cms_compare_info(application_id, last_obj, application_entity, detect
    # 银行卡-------------------------------------------------------------------------------------------------------
    bank_info = {}
    bank_name = cms_info.get('bankAccountDetails', {}).get('bankName', '')
-    account_no = decode_des(cms_info.get('bankAccountDetails', {}).get('accountNo', ''), des_key)
+    account_no = aes_decrypt_cbc(cms_info.get('bankAccountDetails', {}).get('accountNo', ''), des_key, des_iv)
    account_holder_name = cms_info.get('bankAccountDetails', {}).get('accountHolderName', '')
    is_gsyh = True if '工商' in bank_name else False
--- a/src/common/tools/aes.py 0 → 100644
View file @1a883cc
+++ b/src/common/tools/aes.py 0 → 100644
View file @1a883cc
+#这个有问题
+from Crypto.Cipher import AES
+from base64 import b64encode, b64decode
+def encrypt_ecb(data, key):
+    data = data.encode()
+    key = key.encode()
+    aes = AES.new(key, AES.MODE_CBC, bytes(16))  
+    res = aes.encrypt(pad(data, 32))
+    return b64encode(res).decode()
+def decrypt(data, key, iv):
+    key = key.encode()
+    iv = iv.encode()
+    # aes = AES.new(key, AES.MODE_CBC, bytes(16))
+    aes = AES.new(key, AES.MODE_CBC, iv)
+    res = aes.decrypt(b64decode(data))
+    return unpad(res, 32).decode()
+def unpad(padded_data, block_size, style='pkcs7'):
+    pdata_len = len(padded_data)
+    if pdata_len == 0:
+        raise ValueError("Zero-length input cannot be unpadded")
+    if pdata_len % block_size:
+        raise ValueError("Input data is not padded")
+    if style in ('pkcs7', 'x923'):
+        padding_len = bord(padded_data[-1])
+        if padding_len<1 or padding_len>min(block_size, pdata_len):
+            raise ValueError("Padding is incorrect.")
+        if style == 'pkcs7':
+            if padded_data[-padding_len:]!=bchr(padding_len)*padding_len:
+                raise ValueError("PKCS#7 padding is incorrect.")
+        else:
+            if padded_data[-padding_len:-1]!=bchr(0)*(padding_len-1):
+                raise ValueError("ANSI X.923 padding is incorrect.")
+    elif style == 'iso7816':
+        padding_len = pdata_len - padded_data.rfind(bchr(128))
+        if padding_len<1 or padding_len>min(block_size, pdata_len):
+            raise ValueError("Padding is incorrect.")
+        if padding_len>1 and padded_data[1-padding_len:]!=bchr(0)*(padding_len-1):
+            raise ValueError("ISO 7816-4 padding is incorrect.")
+    else:
+        raise ValueError("Unknown padding style")
+    return padded_data[:-padding_len]
+def pad(data_to_pad, block_size, style='pkcs7'):
+    padding_len = block_size-len(data_to_pad)%block_size
+    if style == 'pkcs7':
+        padding = bchr(padding_len)*padding_len
+    elif style == 'x923':
+        padding = bchr(0)*(padding_len-1) + bchr(padding_len)
+    elif style == 'iso7816':
+        padding = bchr(128) + bchr(0)*(padding_len-1)
+    else:
+        raise ValueError("Unknown padding style")
+    return data_to_pad + padding
+def bord(s):
+    return s
+def bchr(s):
+    return bytes([s])
+if __name__ == '__main__':
+    decrypt_data = decrypt('QkjNiuixpmtcxxqxaIZ30A==', 'm0XsOHC52YZ5KtakhpuMSZtF7DhwudmG', 'OCRocr2024UATocr')
+    print('解密：', decrypt_data)
\ No newline at end of file
--- a/src/common/tools/aes_util.py 0 → 100644
View file @1a883cc
+++ b/src/common/tools/aes_util.py 0 → 100644
View file @1a883cc
+from Crypto.Cipher import AES
+from base64 import b64encode, b64decode
+def aes_encrypt_cbc(data, key, iv):
+    cipher = AES.new(key, AES.MODE_CBC, iv)
+    return cipher.encrypt(data)
+def aes_decrypt_cbc(data, key, iv):
+    res = ''
+    try:
+        cipher = AES.new(key.encode(), AES.MODE_CBC, iv.encode())
+        res = cipher.decrypt(b64decode(data))
+        res = res.decode('utf-8').replace('\x0e', '').replace('\u007', '')
+    except Exception as e:
+        res = ''
+    return res
+# 示例使用
+key = 'm0XsOHC52YZ5KtakhpuMSZtF7DhwudmG'  # 密钥长度必须是16、24或32字节
+iv = 'OCRocr2024UATocr'
+decrypted_data = aes_decrypt_cbc('QkjNiuixpmtcxxqxaIZ30A==', key, iv)
+print("解密:", decrypted_data)
\ No newline at end of file