This commit is contained in:
yumoqing 2021-07-13 16:13:59 +08:00
parent 36be6584dc
commit b094e127a2
5 changed files with 255 additions and 79 deletions

149
appPublic/dataencoder.py Normal file
View File

@ -0,0 +1,149 @@
try:
import ujson as json
except:
import json
from appPublic.rsa import RSA
from appPublic.rc4 import RC4
from appPublic.uniqueID import getID
import brotli
import zlib
import struct
DATA_TYPE_BYTES = 1
DATA_TYPE_STR = 2
DATA_TYPE_JSON = 3
class DataEncoder:
"""
security data packing - unpacking object
packs data:
encode data with random key's rc4 crypt algorithm,
encode rc4's key with receiver's public key
sign data with sender's private key
packs data using struct in follows order
0: data format(18 bytes)
1. datatype(c)
2. encoded data(length=len(d))
3. encoded_rc4key(length=len(k))
4. sign(signs from (0+1+2+3) data) (length=len(s))
5. compress data and return compressed dta
return packed data
unpacks data:
0. decompress data
1. get 18 bytes fmt data, erase tails b'0x00'
2. using fmt to unpack data[18:]
3. verify sign
4. decode k
5. decode data usig decoded k with rc4 algorithm
6. convert data type to origin data type
7. return converted data
"""
def __init__(self, myid, func_get_peer_pubkey, private_file=None):
self.myid = myid
self.func_get_peer_pubkey = func_get_peer_pubkey
self.public_keys = {}
self.private_file = private_file
self.rsa = RSA()
self.rc4 = RC4()
if self.private_file:
self.private_key = self.rsa.read_privatekey(self.private_file)
else:
self.private_key = rsa.create_privaatekey()
self.public_key = self.rsa.create_publickey(self.private_key)
def identify_datatype(self, data):
if isinstance(data, bytes):
return DATA_TYPE_BYTES, data
if isinstance(data, str):
return DATA_TYPE_STR, data.encode('utf-8')
data = json.dumps(data).encode('utf-8')
return DATA_TYPE_JSON, data
def my_text_publickey(self):
pk = self.public_key()
return self.rsa.publickeyText(pk)
def exist_peer_publickeys(self, peer_id):
return True if self.public_keys.get(peer_id, False) else False
def set_peer_pubkey(self, peer_id, pubkey):
self.public_keys[peer_id] = pubkey
def get_peer_text_pubkey(self, peer_id):
pk = self.get_peer_pubkey()
txtpk = self.rsa. publickeyText(pk)
return txtpk
def set_peer_text_pubkey(self, peer_id, text_pubkey):
pk = self.rsa.publickeyFromText(text_pubkey)
self.set_peer_pubkey(peer_id, pk)
def get_peer_pubkey(self, peer_id):
pubkey = self.public_keys.get(peer_id)
if not pubkey:
try:
self.func_get_peer_pubkey(peer_id)
except:
raise Exception('Can not get peer public key(%s)')
pubkey = self.public_keys.get(peer_id)
return pubkey
def pack(self, peer_id, data):
pk = self.get_peer_pubkey(peer_id)
t, d = self.identify_datatype(data)
d, k = self.encode_data(pk, d)
f = 'b%05ds%03ds' % (len(d), len(k))
f1 = f + '256s'
pd1 = struct.pack('18s', f1.encode('utf-8'))
pd2 = struct.pack(f, t, d, k)
pd = pd1 + pd2
s = self.sign_data(pd)
pd += s
self.pack_d = [t,d,k,s]
origin_len = len(pd)
pd = brotli.compress(pd)
pdz = zlib.compress(pd)
print('brotli lenght=%d, zlib length=%d, origin length=%d' % \
(len(pd), len(pdz), origin_len)
)
return pd
def unpack(self, peer_id, data):
data = brotli.decompress(data)
org_data = data
pk = self.get_peer_pubkey(peer_id)
f = data[:18]
f.strip(b'0x00')
f = f.decode('utf-8')
data = data[18:]
t, d, k, s = struct.unpack(f, data)
self.unpack_d = [t,d,k,s]
data1 = org_data[:org_data.index(s)]
if not self.verify_sign(data1, s, pk):
raise Exception('data sign verify failed')
data = self.decode_data(d, k)
if t == DATA_TYPE_BYTES:
return data
if t == DATA_TYPE_STR:
return data.decode('utf-8')
return json.loads(data)
def encode_data(self, peer_pubkey, data):
key = getID()
if isinstance(key, str):
key = key.encode('utf-8')
ctext = self.rc4.encode_bytes(data, key)
encoded_key = self.rsa.encode_bytes(peer_pubkey, key)
print('key length compare,', len(key), len(encoded_key))
return ctext, encoded_key
def sign_data(self, data):
return self.rsa.sign_bdata(self.private_key, data)
def decode_data(self, data, encoded_key):
key = self.rsa.decode_bytes(self.private_key, encoded_key)
return self.rc4.decode_bytes(data, key)
def verify_sign(self, data, sign, peer_pubkey):
return self.rsa.check_sign_bdata(peer_pubkey, data, sign)

View File

@ -10,8 +10,6 @@ class RC4:
def _crypt(self,data,key):
"""RC4 algorithm return bytes"""
if type(data)==type(''):
data = data.encode(self.dcoding)
x = 0
box = [i for i in range(256) ]
for i in range(256):
@ -27,39 +25,51 @@ class RC4:
return ''.join(out).encode(self.bcoding)
def encode_bytes(self, bdata, key):
a = sha1(key + self.salt)
k = a.digest()
data = self.salt + self._crypt(bdata, k)
return data
def encode(self,data, key,encode=base64.b64encode, salt_length=16):
"""RC4 encryption with random salt and final encoding"""
akey = key.encode(self.bcoding)
a = sha1(akey + self.salt)
k = a.digest()
data = self.salt + self._crypt(data, k)
if type(data)==type(''):
data = data.encode(self.dcoding)
key = key.encode(self.bcoding)
self.encode_bytes(data, key)
if encode:
data = encode(data)
return data.decode(self.dcoding)
def decode(self,data, key,decode=base64.b64decode, salt_length=16):
"""RC4 decryption of encoded data"""
akey = key.encode(self.bcoding)
if decode:
data = decode(data)
def decode_bytes(self, data, key):
salt_length = 16
salt = data[:salt_length]
a = sha1(akey + self.salt)
a = sha1(key + self.salt)
k = a.digest() #.decode('iso-8859-1')
r = self._crypt(data[salt_length:], k)
return r
def decode(self,data, key,decode=base64.b64decode, salt_length=16):
"""RC4 decryption of encoded data"""
if type(data)==type(''):
data = data.encode(self.dcoding)
key = key.encode(self.bcoding)
if decode:
data = decode(data)
r = self.decode_bytes(data, key)
return r.decode(self.dcoding)
if __name__=='__main__':
# 需要加密的数据长度没有限制
# 密钥
data="231r3 feregrenerjk gkht324g8924gnfw k;ejkvwkjerv"
key = '123456'
data=b"231r3 feregrenerjk gkht324g8924gnfw k;ejkvwkjerv"
key = b'123456'
rc4 = RC4()
print(data)
# 加码
encoded_data = rc4.encode(data,key)
encoded_data = rc4.encode_bytes(data,key)
print(encoded_data,len(encoded_data) )
# 解码
decoded_data = rc4.decode(encoded_data,key)
print(decoded_data)
decoded_data = rc4.decode_bytes(encoded_data,key)
print(data, decoded_data, decoded_data==data)

View File

@ -76,49 +76,70 @@ class RSA:
def create_publickey(self,private_key):
return private_key.public_key()
def encode(self,public_key,text):
message_bytes = bytes(text, encoding='utf8') if not isinstance(text, bytes) else text
return str(base64.b64encode(public_key.encrypt(message_bytes,
def encode_bytes(self, public_key, bdata):
return public_key.encrypt(bdata,
padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None
)
)), encoding='utf-8')
))
def encode(self,public_key,text):
message_bytes = bytes(text, encoding='utf8') if not isinstance(text, bytes) else text
cdata = self.encode_bytes(public_key, message_bytes)
return str(base64.b64encode(cdata), encoding='utf-8')
def decode(self,private_key,cipher):
cipher = cipher.encode('utf8') if not isinstance(cipher, bytes) else cipher
ciphertext_decoded = base64.b64decode(cipher)
plain_text = private_key.decrypt(
ciphertext_decoded,padding.OAEP(
def decode_bytes(self, private_key, bdata):
return private_key.decrypt(
bdata,padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None
)
)
def decode(self,private_key,cipher):
cipher = cipher.encode('utf8') if not isinstance(cipher, bytes) else cipher
ciphertext_decoded = base64.b64decode(cipher)
plain_text = self.decode_bytes(private_key, ciphertext_decoded)
return str(plain_text, encoding='utf8')
def sign_bdata(self, private_key, data_to_sign):
s = private_key.sign(data_to_sign,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
return s
def sign(self,private_key,message):
data_to_sign = bytes(message, encoding='utf8') if not isinstance(
message,
bytes
) else message
signer = private_key.signer(
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
signer.update(data_to_sign)
signature = str(
base64.b64encode(signer.finalize()),
base64.b64encode(self.sign_bdata(private_key, data_to_sign)),
encoding='utf8'
)
return signature
def check_sign(self,public_key,plain_text,signature):
def check_sign_bdata(self, public_key, bdata, sign):
try:
r = public_key.verify(sign, bdata,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
# print('verify return=', r) r is None
return True
except InvalidSignature as e:
return False
def check_sign(self,public_key,plain_text,signature):
plain_text_bytes = bytes(
plain_text,
encoding='utf8'
@ -126,22 +147,12 @@ class RSA:
signature = base64.b64decode(
signature
) if not isinstance(signature, bytes) else signature
verifier = public_key.verifier(
signature,
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
verifier.update(plain_text_bytes)
verifier.verify()
return True
except InvalidSignature as e:
return False
return self.check_sign_bdata(public_key, plain_text_bytes, \
signature)
if __name__ == '__main__':
Recv_cipher=b'JHHDhjaHLeLRopobfiJvWtVn8Bu3/3AsV6Xr8MwHBBEli7v+oHRNH2dcAfVa7VcdBlKlr7W+hDDAxlex3/OzwyRp4R5DcDsepTLPaG+nKK6zj0MGkvEJ6iNpABO9uohskFXPBuO6t3+G6cKRMMeIU7g7oSJqlbKHJyRmd9j8OHS2fFYL331oRhJvyuJe5zrdxHEOez+XEt2AbuYi7WFFVlM/DvX/tjAG3SHXr14GlJYGbuNR2LNIapAMBSt7GDQ/LrzLv54ysE3OZpVFnOszVt5ythiDPoImnpJ990Fb/1yd7goPZ5NA8cSKCu7dDV42JWcj44JHrIfNsMR7aG9QxQ=='
import os
prikey1_file = os.path.join(os.path.dirname(__file__),'..','test', 'prikey1.rsa')
r = RSA()
mpri = r.create_privatekey()
mpub = r.create_publickey(mpri)
@ -151,26 +162,11 @@ if __name__ == '__main__':
text = 'this is a test data, aaa'
cipher = r.encode(mpub,text)
signature = r.sign(zpri,text)
ntext = r.decode(mpri,cipher)
check = r.check_sign(zpub,ntext,signature)
print(text,ntext,check)
ypri = r.read_privatekey('d:/dev/mecp/conf/RSA.private.key','ymq123')
ypub = r.read_publickey('d:/dev/mecp/conf/RSA.public.key')
x = r.encode(ypub,'root:ymq123')
print('root:ymq123 encode=',x,len(x),len(Recv_cipher))
orgtext=r.decode(ypri,Recv_cipher)
print(orgtext)
r.write_publickey(ypub,'./test.public.key')
r.write_privatekey(ypri,'./test.private.key','ymq123')
ypri = r.read_privatekey('./test.private.key','ymq123')
ypub = r.read_publickey('./test.public.key')
x = r.encode(ypub,text)
ntext = r.decode(ypri,x)
print(text,'<==>',ntext)
print('encode text=', text, \
'decode result=', ntext,
'cyber size=', len(cipher),
'check if equal=', text==ntext)
signature = r.sign(zpri,text)
check = r.check_sign(zpub,text,signature)
print('sign and verify=',len(signature),check)

View File

@ -3,7 +3,7 @@ from socket import *
import json
from appPublic.sockPackage import get_free_local_addr
from appPublic.background import Background
BUFSIZE = 1024
BUFSIZE = 1024 * 64
class UdpComm:
def __init__(self, port, callback, timeout=1):
self.callback = callback
@ -43,17 +43,24 @@ class UdpComm:
udpCliSock.settimeout(self.timeout)
udpCliSock.bind(('', 0))
udpCliSock.setsockopt(SOL_SOCKET, SO_BROADCAST, 1)
b = json.dumps(data).encode('utf-8')
b = data
if not isinstance(data, bytes):
b = json.dumps(data).encode('utf-8')
b = b'0x00' * 18 + b
udpCliSock.sendto(b, (broadcast_host,self.port))
def send(self,data,addr):
b = json.dumps(data).encode('utf-8')
b = data
if not isinstance(data, bytes):
b = json.dumps(data).encode('utf-8')
if isinstance(addr,list):
addr = tuple(addr)
self.udpSerSock.sendto(b,addr)
def sends(self,data, addrs):
b = json.dumps(data).encode('utf-8')
b = data
if not isinstance(data, bytes):
b = json.dumps(data).encode('utf-8')
for addr in addrs:
if isinstance(addr,list):
addr = tuple(addr)

View File

@ -0,0 +1,14 @@
import sys
from appPublic.rsa import RSA
def gen(filename):
rsa = RSA()
pk = rsa.create_privatekey()
rsa.write_privatekey(pk, filename)
if __name__ == '__main__':
if len(sys.argv) < 2:
print('Usage\n%s private_file_name' % sys.argv[0])
sys.exit(1)
gen(sys.argv[1])