youngS
4 years ago
parent
c7e77ad46b
commit
2e21465a1e
@ -0,0 +1,17 @@ |
||||
# Generated by Django 3.0.3 on 2021-04-18 17:24 |
||||
|
||||
from django.db import migrations, models |
||||
|
||||
|
||||
class Migration(migrations.Migration): |
||||
dependencies = [ |
||||
('api', '0041_auto_20210416_1438'), |
||||
] |
||||
|
||||
operations = [ |
||||
migrations.AddField( |
||||
model_name='order', |
||||
name='payment_name', |
||||
field=models.CharField(blank=True, max_length=128, null=True, verbose_name='支付商家名称'), |
||||
), |
||||
] |
||||
@ -0,0 +1,38 @@ |
||||
#!/usr/bin/env python |
||||
# -*- coding:utf-8 -*- |
||||
# project: 4月 |
||||
# author: NinEveN |
||||
# date: 2021/4/18 |
||||
|
||||
from fir_ser.settings import PAY_CONFIG |
||||
from api.utils.pay.ali import Alipay |
||||
from api.utils.pay.wx import Weixinpay |
||||
|
||||
|
||||
def get_pay_obj_form_name(pay_name): |
||||
for pay_info in PAY_CONFIG: |
||||
if pay_name == pay_info.get('NAME', '') and pay_info.get('ENABLED', False): |
||||
auth_info = pay_info.get('AUTH', None) |
||||
p_type = pay_info.get('TYPE', '') |
||||
if auth_info: |
||||
if p_type == 'ALI': |
||||
return Alipay(pay_name, p_type, auth_info) |
||||
elif p_type == 'WX': |
||||
return Weixinpay(pay_name, p_type, auth_info) |
||||
else: |
||||
pass |
||||
|
||||
|
||||
def get_enable_pay_choices(): |
||||
pay_choices = [] |
||||
for pay_info in PAY_CONFIG: |
||||
if pay_info.get('ENABLED', False): |
||||
pay_choices.append({'type': pay_info.get('TYPE'), 'name': pay_info.get('NAME', '')}) |
||||
return pay_choices |
||||
|
||||
|
||||
def get_payment_type(p_type): |
||||
if p_type == 'ALI': |
||||
return 1 |
||||
elif p_type == 'WX': |
||||
return 0 |
||||
@ -1,101 +0,0 @@ |
||||
# -*- coding: utf-8 -*- |
||||
|
||||
import time |
||||
import uuid |
||||
from base64 import b64decode, b64encode |
||||
import json |
||||
|
||||
from cryptography.exceptions import InvalidSignature |
||||
from cryptography.hazmat.backends import default_backend |
||||
from cryptography.hazmat.primitives.asymmetric.padding import PKCS1v15 |
||||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM |
||||
from cryptography.hazmat.primitives.hashes import SHA256 |
||||
from cryptography.hazmat.primitives.serialization import (load_pem_private_key, |
||||
load_pem_public_key) |
||||
from OpenSSL import crypto |
||||
|
||||
|
||||
def build_authorization(path, |
||||
method, |
||||
mchid, |
||||
serial_no, |
||||
mch_private_key, |
||||
data=None, |
||||
nonce_str=None): |
||||
timeStamp = str(int(time.time())) |
||||
nonce_str = nonce_str or ''.join(str(uuid.uuid4()).split('-')).upper() |
||||
body = json.dumps(data) if data else '' |
||||
sign_str = method + '\n' + path + '\n' + \ |
||||
timeStamp + '\n' + nonce_str + '\n' + body + '\n' |
||||
signature = sign(private_key=mch_private_key, sign_str=sign_str) |
||||
authorization = 'WECHATPAY2-SHA256-RSA2048 mchid="%s",nonce_str="%s",signature="%s",timestamp="%s",serial_no="%s"' % ( |
||||
mchid, nonce_str, signature, timeStamp, serial_no) |
||||
return authorization |
||||
|
||||
|
||||
def sign(private_key, sign_str): |
||||
private_key = load_pem_private_key(data=format_private_key( |
||||
private_key).encode('UTF-8'), password=None, backend=default_backend()) |
||||
message = sign_str.encode('UTF-8') |
||||
signature = private_key.sign(message, PKCS1v15(), SHA256()) |
||||
sign = b64encode(signature).decode('UTF-8').replace('\n', '') |
||||
return sign |
||||
|
||||
|
||||
def decrypt(nonce, ciphertext, associated_data, apiv3_key): |
||||
key_bytes = apiv3_key.encode('UTF-8') |
||||
nonce_bytes = nonce.encode('UTF-8') |
||||
associated_data_bytes = associated_data.encode('UTF-8') |
||||
data = b64decode(ciphertext) |
||||
aesgcm = AESGCM(key_bytes) |
||||
return aesgcm.decrypt(nonce_bytes, data, associated_data_bytes).decode('UTF-8') |
||||
|
||||
|
||||
def format_private_key(private_key): |
||||
pem_start = '-----BEGIN PRIVATE KEY-----\n' |
||||
pem_end = '\n-----END PRIVATE KEY-----' |
||||
if not private_key.startswith(pem_start): |
||||
private_key = pem_start + private_key |
||||
if not private_key.endswith(pem_end): |
||||
private_key = private_key + pem_end |
||||
return private_key |
||||
|
||||
|
||||
def format_certificate(certificate): |
||||
pem_start = '-----BEGIN CERTIFICATE-----\n' |
||||
pem_end = '\n-----END CERTIFICATE-----' |
||||
if not certificate.startswith(pem_start): |
||||
certificate = pem_start + certificate |
||||
if not certificate.endswith(pem_end): |
||||
certificate = certificate + pem_end |
||||
return certificate |
||||
|
||||
|
||||
def verify(timestamp, nonce, body, signature, certificate): |
||||
sign_str = '%s\n%s\n%s\n' % (timestamp, nonce, body) |
||||
public_key_str = dump_public_key(certificate) |
||||
public_key = load_pem_public_key(data=public_key_str.encode('UTF-8'), backend=default_backend()) |
||||
message = sign_str.encode('UTF-8') |
||||
signature = b64decode(signature) |
||||
try: |
||||
public_key.verify(signature, sign_str.encode('UTF-8'), PKCS1v15(), SHA256()) |
||||
except InvalidSignature: |
||||
return False |
||||
return True |
||||
|
||||
|
||||
def certificate_serial_number(certificate): |
||||
cert = crypto.load_certificate(crypto.FILETYPE_PEM, format_certificate(certificate)) |
||||
try: |
||||
res = cert.get_signature_algorithm().decode('UTF-8') |
||||
if res != 'sha256WithRSAEncryption': |
||||
return None |
||||
return hex(cert.get_serial_number()).upper()[2:] |
||||
except: |
||||
return None |
||||
|
||||
|
||||
def dump_public_key(certificate): |
||||
cert = crypto.load_certificate(crypto.FILETYPE_PEM, format_certificate(certificate)) |
||||
public_key = crypto.dump_publickey(crypto.FILETYPE_PEM, cert.get_pubkey()).decode("utf-8") |
||||
return public_key |
||||
Loading…
Reference in new issue