Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_base64encode_field_to_xml(self):
from wechatpy.fields import Base64EncodeField
content = b'test'
field = Base64EncodeField('Content')
expected = '<content></content>'.format(
content=to_text(base64.b64encode(content))
)
self.assertEqual(expected, field.to_xml(content))
def get_jsapi_signature(self, prepay_id, timestamp=None, nonce_str=None):
"""
获取 JSAPI 签名
:param prepay_id: 统一下单接口返回的 prepay_id 参数值
:param timestamp: 可选,时间戳,默认为当前时间戳
:param nonce_str: 可选,随机字符串,默认自动生成
:return: 签名
"""
data = {
'appId': self.sub_appid or self.appid,
'timeStamp': timestamp or to_text(int(time.time())),
'nonceStr': nonce_str or random_string(32),
'signType': 'MD5',
'package': 'prepay_id={0}'.format(prepay_id),
}
return calculate_signature(
data,
self._client.api_key if not self._client.sandbox else self._client.sandbox_api_key
)
def _decrypt(self, text, _id, exception=None):
text = to_binary(text)
plain_text = self.cipher.decrypt(base64.b64decode(text))
padding = byte2int(plain_text[-1])
content = plain_text[16:-padding]
xml_length = socket.ntohl(struct.unpack(b'I', content[:4])[0])
xml_content = to_text(content[4:xml_length + 4])
from_id = to_text(content[xml_length + 4:])
if from_id != _id:
exception = exception or Exception
raise exception()
return xml_content
def __repr__(self):
_repr = '{klass}({code}, {msg}). Pay({pay_code}, {pay_msg})'.format(
klass=self.__class__.__name__,
code=self.return_code,
msg=self.return_msg,
pay_code=self.errcode,
pay_msg=self.errmsg
)
if six.PY2:
return to_binary(_repr)
else:
return to_text(_repr)
def __str__(self):
_repr = 'Error code: {code}, message: {msg}'.format(
code=self.errcode,
msg=self.errmsg
)
if six.PY2:
return to_binary(_repr)
else:
return to_text(_repr)
def get(self, key, default=None):
key = self.key_name(key)
value = self.redis.get(key)
if value is None:
return default
return json.loads(to_text(value))
def _decrypt(self, text, _id, exception=None):
text = to_binary(text)
plain_text = self.cipher.decrypt(base64.b64decode(text))
padding = byte2int(plain_text[-1])
content = plain_text[16:-padding]
xml_length = socket.ntohl(struct.unpack(b'I', content[:4])[0])
xml_content = to_text(content[4:xml_length + 4])
from_id = to_text(content[xml_length + 4:])
if from_id != _id:
exception = exception or Exception
raise exception()
return xml_content
def create(self, name):
"""
创建分组
详情请参考
http://mp.weixin.qq.com/wiki/0/56d992c605a97245eb7e617854b169fc.html
:param name: 分组名字(30个字符以内)
:return: 返回的 JSON 数据包
"""
name = to_text(name)
return self._post(
'groups/create',
data={'group': {'name': name}}
)
def _encrypt_message(self,
msg,
nonce,
timestamp=None,
crypto_class=None):
from wechatpy.replies import BaseReply
xml = """
{timestamp}
"""
if isinstance(msg, BaseReply):
msg = msg.render()
timestamp = timestamp or to_text(int(time.time()))
pc = crypto_class(self.key)
encrypt = to_text(pc.encrypt(msg, self._id))
signature = _get_signature(self.token, timestamp, nonce, encrypt)
return to_text(xml.format(
encrypt=encrypt,
signature=signature,
timestamp=timestamp,
nonce=nonce
))