You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
295 lines
8.2 KiB
Python
295 lines
8.2 KiB
Python
from cryptography.hazmat.backends import default_backend
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives import hashes
|
|
from cryptography.hazmat.primitives.asymmetric import padding
|
|
from cryptography.exceptions import InvalidSignature
|
|
import os
|
|
try:
|
|
from .syfr import * #import syfr
|
|
except ImportError:
|
|
from syfr import *
|
|
|
|
key_dir = os.path.join(os.path.expanduser('~'),'.keys','komrade')
|
|
if not os.path.exists(key_dir): os.makedirs(key_dir)
|
|
PATH_PRIVATE_KEY=os.path.join(key_dir,'private_key.pem')
|
|
PATH_PUBLIC_KEY=os.path.join(key_dir,'public_key.pem')
|
|
|
|
|
|
### CREATING KEYS
|
|
|
|
def new_keys(save=True,password=None):
|
|
private_key = rsa.generate_private_key(
|
|
public_exponent=65537,
|
|
key_size=2048,
|
|
backend=default_backend()
|
|
)
|
|
public_key = private_key.public_key()
|
|
|
|
if save:
|
|
save_private_key(private_key,password=password)
|
|
save_public_key(public_key)
|
|
|
|
return private_key,public_key
|
|
|
|
|
|
|
|
def save_private_key(private_key,fn=PATH_PRIVATE_KEY,password=None, return_instead=False):
|
|
pem = private_key.private_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PrivateFormat.PKCS8,
|
|
encryption_algorithm=serialization.NoEncryption() if not password else serialization.BestAvailableEncryption(password.encode())
|
|
)
|
|
if return_instead: return pem
|
|
with open(fn,'wb') as f: f.write(pem)
|
|
|
|
def save_public_key(public_key,fn=PATH_PUBLIC_KEY,return_instead=False):
|
|
pem = public_key.public_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo
|
|
)
|
|
if return_instead: return pem
|
|
with open(fn,'wb') as f: f.write(pem)
|
|
|
|
|
|
|
|
### LOADING KEYS
|
|
def load_keys():
|
|
return (load_private_key_from_file(), load_public_key_from_file())
|
|
|
|
def load_private_key(pem,password=None):
|
|
return serialization.load_pem_private_key(
|
|
pem,
|
|
password=password.encode() if password else None,
|
|
backend=default_backend()
|
|
)
|
|
|
|
def load_private_key_from_file(fn=PATH_PRIVATE_KEY,password=None):
|
|
with open(fn, "rb") as key_file:
|
|
return load_private_key(key_file.read(), password)
|
|
|
|
def load_public_key(pem):
|
|
return serialization.load_pem_public_key(
|
|
pem,
|
|
backend=default_backend()
|
|
)
|
|
|
|
def load_public_key_from_file(fn=PATH_PUBLIC_KEY):
|
|
with open(fn, "rb") as key_file:
|
|
return load_public_key(key_file.read())
|
|
|
|
### DE/ENCRYPTING
|
|
def encrypt_msg(message, public_key):
|
|
return public_key.encrypt(
|
|
message,
|
|
padding.OAEP(
|
|
mgf=padding.MGF1(algorithm=hashes.SHA256()),
|
|
algorithm=hashes.SHA256(),
|
|
label=None
|
|
)
|
|
)
|
|
|
|
def encrypt_msg_symmetric(message):
|
|
import os
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
from cryptography.hazmat.backends import default_backend
|
|
backend = default_backend()
|
|
key = os.urandom(32)
|
|
iv = os.urandom(16)
|
|
cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=backend)
|
|
encryptor = cipher.encryptor()
|
|
|
|
ct = encryptor.update(message) + encryptor.finalize()
|
|
return ct
|
|
|
|
def decrypt_msg_symmetric():
|
|
import os
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
from cryptography.hazmat.backends import default_backend
|
|
backend = default_backend()
|
|
key = os.urandom(32)
|
|
iv = os.urandom(16)
|
|
cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=backend)
|
|
|
|
return ct
|
|
decryptor = cipher.decryptor()
|
|
decryptor.update(ct) + decryptor.finalize()
|
|
b'a secret message'
|
|
|
|
def decrypt_msg(encrypted, private_key):
|
|
return private_key.decrypt(
|
|
encrypted,
|
|
padding.OAEP(
|
|
mgf=padding.MGF1(algorithm=hashes.SHA256()),
|
|
algorithm=hashes.SHA256(),
|
|
label=None
|
|
)
|
|
)
|
|
|
|
### SIGNING/VERIFYING
|
|
def sign_msg(message, private_key):
|
|
return private_key.sign(
|
|
message,
|
|
padding.PSS(
|
|
mgf=padding.MGF1(hashes.SHA256()),
|
|
salt_length=padding.PSS.MAX_LENGTH
|
|
),
|
|
hashes.SHA256()
|
|
)
|
|
|
|
def verify_msg(message, signature, public_key):
|
|
try:
|
|
verified = public_key.verify(
|
|
signature,
|
|
message,
|
|
padding.PSS(
|
|
mgf=padding.MGF1(hashes.SHA256()),
|
|
salt_length=padding.PSS.MAX_LENGTH
|
|
),
|
|
hashes.SHA256()
|
|
)
|
|
return True
|
|
except InvalidSignature:
|
|
return False
|
|
return None
|
|
|
|
|
|
|
|
|
|
# #private_key,public_key = new_keys()
|
|
# private_key,public_key = load_keys()
|
|
|
|
# #print(private_key)
|
|
# #print(public_key)
|
|
|
|
|
|
# #enc = encrypt_msg('Drive your plow over the bones of the dead', public_key)
|
|
# #print(enc)
|
|
|
|
# dec = decrypt_msg(enc,private_key)
|
|
# #print(dec)
|
|
|
|
# msg = b'hello'
|
|
# signature = sign_msg(msg,private_key)
|
|
|
|
# #print(encrypt_msg(b'hello',public_key))
|
|
|
|
# print(verify_msg(msg+b'!!',signature,public_key))
|
|
|
|
|
|
|
|
## ONLY NEEDS RUN ONCE!
|
|
def gen_global_keys1(fn='.keys.global.json'):
|
|
from kivy.storage.jsonstore import JsonStore
|
|
|
|
private_key,public_key=new_keys(save=False,password=None)
|
|
pem_private_key = save_private_key(private_key,password=None,return_instead=True)
|
|
pem_public_key = save_public_key(public_key,return_instead=True)
|
|
|
|
store = JsonStore('./.keys.global.json')
|
|
|
|
store.put('_keys',private=str(pem_private_key.decode()),public=str(pem_public_key.decode())) #(private_key,password=passkey)
|
|
|
|
def gen_global_keys(fn='.keys.global.json'):
|
|
from kivy.storage.jsonstore import JsonStore
|
|
|
|
store = JsonStore('./.keys.global.json')
|
|
|
|
#store.put('_keys',private=str(pem_private_key.decode()),public=str(pem_public_key.decode())) #(private_key,password=passkey)
|
|
|
|
private_key = generate_rsa_key()
|
|
pem_private_key = serialize_privkey(private_key, password=None)# save_private_key(private_key,password=passkey,return_instead=True)
|
|
pem_public_key = serialize_pubkey(private_key.public_key())
|
|
store.put('_keys',private=pem_private_key.decode(),public=pem_public_key.decode()) #(private_key,password=passkey)
|
|
|
|
"""
|
|
import os
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
from cryptography.hazmat.backends import default_backend
|
|
backend = default_backend()
|
|
key = os.urandom(32)
|
|
iv = os.urandom(16)
|
|
cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=backend)
|
|
encryptor = cipher.encryptor()
|
|
ct = encryptor.update(b"a secret message") + encryptor.finalize()
|
|
decryptor = cipher.decryptor()
|
|
decryptor.update(ct) + decryptor.finalize()
|
|
b'a secret message'"""
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def aes_rsa_encrypt(message, recipient_rsa_pub):
|
|
aes_key = create_aes_key()
|
|
aes_ciphertext, iv = aes_encrypt(message, aes_key)
|
|
encry_aes_key = rsa_encrypt(aes_key, recipient_rsa_pub)
|
|
return aes_ciphertext, encry_aes_key, iv #, sign
|
|
|
|
|
|
def aes_rsa_decrypt(aes_ciphertext, rsa_priv, encry_aes_key, iv): #, hmac, hmac_signature, rsa_priv, iv, metadata):
|
|
aes_key = rsa_decrypt(encry_aes_key, rsa_priv)
|
|
plaintext = aes_decrypt(aes_ciphertext, aes_key, iv)
|
|
return plaintext
|
|
|
|
|
|
|
|
|
|
# def _decrypt_rsa(x,privkey=CORRECT_PRIV_KEY):
|
|
# x_decr = rsa_decrypt(x,privkey)
|
|
# return x_decr
|
|
|
|
|
|
|
|
|
|
|
|
# encrypt sender
|
|
#def _enc_sender(x):
|
|
# recv=receiver_pubkey
|
|
|
|
# encrypt recipient too?
|
|
|
|
|
|
|
|
# sender_pubkey_b_encr = sep2.join(
|
|
# aes_rsa_encrypt(
|
|
# serialize_pubkey(self.public_key), receiver_pubkey
|
|
# )
|
|
# )
|
|
|
|
# receiver_pubkey_b_encr = sep2.join(
|
|
# aes_rsa_encrypt(
|
|
# serialize_pubkey(receiver_pubkey_b, receiver_pubkey
|
|
# )
|
|
# )
|
|
|
|
|
|
|
|
# msg_encr = sep2.join([val_encr,val_encr_key,iv])
|
|
# sender_encr = sep2.join(
|
|
# aes_rsa_encrypt(
|
|
# serialize_pubkey(self.public_key), receiver_pubkey
|
|
# )
|
|
# )
|
|
# signature_encr = sep2.join(
|
|
# rsa_encrypt(
|
|
# signature,
|
|
# receiver_pubkey
|
|
# )
|
|
# )
|
|
|
|
|
|
|
|
# sender = sep2.join(sender_pubkey_b, signature)
|
|
|
|
# WDV = sep.join([
|
|
# time_b,
|
|
# receiver_encr,
|
|
# msg_encr,
|
|
# sender_encr,
|
|
# signature_encr
|
|
# ])
|
|
|
|
def simple_lock_test(privkey,pubkey):
|
|
return privkey.public_key().public_numbers() == pubkey.public_numbers() |