BiztProtoBois/server
3
0
Fork 0
Code Pull Requests Releases Activity

use authenticated chacha20
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
Torma Kristóf
2021-04-26 16:58:50 +02:00
parent b08da25d21
commit f409aefdc0
1 changed files with 37 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
server/netwrapper.py
View File

@@ -4,9 +4,8 @@ from base64 import b64encode, b64decode
import pyDH
from Crypto.Hash import SHA512
from Crypto.Cipher import PKCS1_OAEP
from Crypto.Cipher import ChaCha20
from Crypto.Cipher import ChaCha20_Poly1305
from Crypto.PublicKey.RSA import RsaKey
from Crypto.Random import get_random_bytes
from netsim import network_interface
from authentication import Authetication
@@ -35,25 +34,29 @@ class NetWrapper:
cipher_rsa = PKCS1_OAEP.new(self.serverPrivateKey)
retmsg = cipher_rsa.decrypt(b64decode(incommingJson['message'])).decode('UTF-8')
cipher = PKCS1_OAEP.new(self.currentClientPublicKey)
identMsg = json.dumps(
{'type': 'IDY', 'source': self.network.own_addr,
'message': b64encode(cipher.encrypt(retmsg.encode('UTF-8'))).decode('ASCII')}).encode(
'message': b64encode(cipher.encrypt(retmsg.encode('UTF-8'))).decode('UTF-8')}).encode(
'UTF-8')
self.network.send_msg(self.clientAddr, identMsg)
def sendMessage(self, message: bytes) -> None:
cipher = ChaCha20.new(key=self.cipherkey, nonce=get_random_bytes(12))
ciphertext = cipher.encrypt(message)
nonce = b64encode(cipher.nonce).decode('ASCII')
ct = b64encode(ciphertext).decode('ASCII')
sendjson = json.dumps({'type': 'CMD', 'source': self.network.own_addr, 'nonce': nonce, 'message': ct}).encode(
cipher = ChaCha20_Poly1305.new(key=self.cipherkey)
header = json.dumps({'source': self.network.own_addr, 'type': 'CMD'}).encode('UTF-8')
cipher.update(header)
ciphertext, tag = cipher.encrypt_and_digest(message)
nonce = b64encode(cipher.nonce).decode('UTF-8')
ct = b64encode(ciphertext).decode('UTF-8')
b64tag = b64encode(tag).decode('UTF-8')
sendjson = json.dumps({'header': b64encode(header).decode('UTF-8'), 'nonce': nonce, 'message': ct, 'tag': b64tag}).encode(
'UTF-8')
self.network.send_msg(self.clientAddr, sendjson)
def keyExchange(self) -> None:
dh = pyDH.DiffieHellman()
cipher = PKCS1_OAEP.new(self.currentClientPublicKey)
mypubkey = b64encode(cipher.encrypt(str(dh.gen_public_key()).encode('UTF-8'))).decode('ASCII')
mypubkey = b64encode(cipher.encrypt(str(dh.gen_public_key()).encode('UTF-8'))).decode('UTF-8')
jsonmsg = json.dumps({'type': 'DH', 'source': self.network.own_addr, 'message': mypubkey}).encode('UTF-8')
self.network.send_msg(self.clientAddr, jsonmsg)
decodedmsg = {'source': '', 'type': ''}
@@ -70,17 +73,20 @@ class NetWrapper:
self.cipherkey = (hasher.hexdigest()[:32]).encode('UTF-8')
def login(self) -> bool:
b64 = {'source': '', 'type': ''}
while not (b64['source'] == self.clientAddr and b64['type'] == 'AUT'):
try:
status, msg = self.network.receive_msg(blocking=True)
if not status:
raise Exception('Network error during connection.')
b64 = json.loads(msg.decode('UTF-8'))
try:
retheader = json.loads(b64decode(b64['header']).decode('UTF-8'))
retnonce = b64decode(b64['nonce'])
retciphertext = b64decode(b64['message'])
retcipher = ChaCha20.new(key=self.cipherkey, nonce=retnonce)
plaintext = retcipher.decrypt(retciphertext).decode('UTF-8').split(' ')
rettag = b64decode(b64['tag'])
retcipher = ChaCha20_Poly1305.new(key=self.cipherkey, nonce=retnonce)
retcipher.update(b64decode(b64['header']))
plaintext = retcipher.decrypt_and_verify(retciphertext, rettag).decode('UTF-8').split(' ')
if not (retheader['source'] == self.clientAddr and retheader['type'] == 'AUT'):
return False
self.homeDirectory = self.authenticationInstance.login(plaintext[1], plaintext[2])
linsuccess = (not (len(plaintext) != 3 or plaintext[0] != "LIN" or plaintext[
1] != self.currentUser)) and self.homeDirectory
@@ -88,12 +94,15 @@ class NetWrapper:
message = "OK".encode('UTF-8')
else:
message = "ERROR".encode('UTF-8')
cipher = ChaCha20.new(key=self.cipherkey, nonce=get_random_bytes(12))
ciphertext = cipher.encrypt(message)
nonce = b64encode(cipher.nonce).decode('ASCII')
ct = b64encode(ciphertext).decode('ASCII')
header = json.dumps({'source': self.network.own_addr, 'type': 'CMD'}).encode('UTF-8')
cipher = ChaCha20_Poly1305.new(key=self.cipherkey)
cipher.update(header)
ciphertext, tag = cipher.encrypt_and_digest(message)
b64tag = b64encode(tag).decode('UTF-8')
nonce = b64encode(cipher.nonce).decode('UTF-8')
ct = b64encode(ciphertext).decode('UTF-8')
sendjson = json.dumps(
{'type': 'AUT', 'source': self.network.own_addr, 'nonce': nonce, 'message': ct}).encode(
{'header': b64encode(header).decode('UTF-8'), 'nonce': nonce, 'message': ct, 'tag': b64tag}).encode(
'UTF-8')
self.network.send_msg(self.clientAddr, sendjson)
return linsuccess
@@ -139,16 +148,17 @@ class NetWrapper:
def recieveEncryptedMessage(self, msg: bytes) -> bytes:
try:
b64 = json.loads(msg.decode('UTF-8'))
while not (b64['source'] == self.clientAddr and b64['type'] == 'CMD'):
status, msg = self.network.receive_msg(blocking=True)
if not status:
raise Exception('Network error during connection.')
b64 = json.loads(msg.decode('UTF-8'))
retheader = json.loads(b64decode(b64['header']).decode('UTF-8'))
retnonce = b64decode(b64['nonce'])
retciphertext = b64decode(b64['message'])
retcipher = ChaCha20.new(key=self.cipherkey, nonce=retnonce)
plaintext = retcipher.decrypt(retciphertext)
return plaintext
rettag = b64decode(b64['tag'])
retcipher = ChaCha20_Poly1305.new(key=self.cipherkey, nonce=retnonce)
retcipher.update(b64decode(b64['header']))
plaintext = retcipher.decrypt_and_verify(retciphertext, rettag)
if not (retheader['source'] == self.clientAddr and retheader['type'] == 'CMD'):
return "ERROR".encode('UTF-8')
else:
return plaintext
except Exception:
print("Incorrect decryption")
return "ERROR".encode('UTF-8')