Reticulum/FPE/Destination.py

166 lines
4.5 KiB
Python
Raw Normal View History

2016-05-29 20:20:44 +00:00
import base64
2018-03-16 09:50:37 +00:00
import math
2018-03-19 15:39:08 +00:00
import FPE
2016-05-29 20:20:44 +00:00
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.asymmetric import padding
2018-03-19 15:39:08 +00:00
2016-05-29 20:20:44 +00:00
class Destination:
2018-03-19 15:39:08 +00:00
KEYSIZE = FPE.Identity.KEYSIZE;
PADDINGSIZE= FPE.Identity.PADDINGSIZE;
2018-03-16 09:50:37 +00:00
# Constants
2018-03-19 15:39:08 +00:00
SINGLE = 0x00;
GROUP = 0x01;
PLAIN = 0x02;
LINK = 0x03;
types = [SINGLE, GROUP, PLAIN, LINK]
2016-05-29 20:20:44 +00:00
IN = 0x11;
OUT = 0x12;
directions = [IN, OUT]
@staticmethod
def getDestinationName(app_name, *aspects):
# Check input values and build name string
if "." in app_name: raise ValueError("Dots can't be used in app names")
name = app_name
for aspect in aspects:
if "." in aspect: raise ValueError("Dots can't be used in aspects")
name = name + "." + aspect
return name
@staticmethod
def getDestinationHash(app_name, *aspects):
name = Destination.getDestinationName(app_name, *aspects)
# Create a digest for the destination
digest = hashes.Hash(hashes.SHA256(), backend=default_backend())
digest.update(name)
return digest.finalize()[:10]
2018-03-16 10:40:37 +00:00
def __init__(self, identity, direction, type, app_name, *aspects):
2016-05-29 20:20:44 +00:00
# Check input values and build name string
if "." in app_name: raise ValueError("Dots can't be used in app names")
if not type in Destination.types: raise ValueError("Unknown destination type")
if not direction in Destination.directions: raise ValueError("Unknown destination direction")
self.type = type
self.direction = direction
2016-06-03 17:02:02 +00:00
self.mtu = 0
2016-05-29 20:20:44 +00:00
2018-03-16 10:40:37 +00:00
if identity == None:
identity = Identity()
identity.createKeys()
self.identity = identity
aspects = aspects+(identity.hexhash,)
2016-05-29 20:20:44 +00:00
self.name = Destination.getDestinationName(app_name, *aspects)
self.hash = Destination.getDestinationHash(app_name, *aspects)
self.hexhash = self.hash.encode("hex_codec")
self.callback = None
2018-03-19 15:39:08 +00:00
FPE.Transport.registerDestination(self)
2016-06-03 17:02:02 +00:00
2016-05-29 20:20:44 +00:00
def __str__(self):
return "<"+self.name+"/"+self.hexhash+">"
def setCallback(self, callback):
self.callback = callback
def receive(self, data):
plaintext = self.decrypt(data)
if plaintext != None and self.callback != None:
self.callback(plaintext, self)
2018-03-16 10:40:37 +00:00
def createKeys(self):
2016-05-29 20:20:44 +00:00
if self.type == Destination.PLAIN:
raise TypeError("A plain destination does not hold any keys")
if self.type == Destination.SINGLE:
2018-03-16 10:40:37 +00:00
raise TypeError("A single destination holds keys through an Identity instance")
2016-05-29 20:20:44 +00:00
if self.type == Destination.GROUP:
self.prv_bytes = Fernet.generate_key()
self.prv = Fernet(self.prv_bytes)
2018-03-16 10:40:37 +00:00
def getPrivateKey(self):
2016-05-29 20:20:44 +00:00
if self.type == Destination.PLAIN:
raise TypeError("A plain destination does not hold any keys")
2018-03-16 10:40:37 +00:00
elif self.type == Destination.SINGLE:
raise TypeError("A single destination holds keys through an Identity instance")
2016-05-29 20:20:44 +00:00
else:
return self.prv_bytes
2018-03-16 10:40:37 +00:00
def loadPrivateKey(self, key):
2016-05-29 20:20:44 +00:00
if self.type == Destination.PLAIN:
raise TypeError("A plain destination does not hold any keys")
if self.type == Destination.SINGLE:
2018-03-16 10:40:37 +00:00
raise TypeError("A single destination holds keys through an Identity instance")
2016-05-29 20:20:44 +00:00
if self.type == Destination.GROUP:
self.prv_bytes = key
self.prv = Fernet(self.prv_bytes)
def loadPublicKey(self, key):
if self.type != Destination.SINGLE:
raise TypeError("Only the \"single\" destination type can hold a public key")
2018-03-16 10:40:37 +00:00
else:
raise TypeError("A single destination holds keys through an Identity instance")
2016-05-29 20:20:44 +00:00
def encrypt(self, plaintext):
if self.type == Destination.PLAIN:
return plaintext
2018-03-16 10:40:37 +00:00
if self.type == Destination.SINGLE and self.identity != None:
return self.identity.encrypt(plaintext)
2016-05-29 20:20:44 +00:00
if self.type == Destination.GROUP and self.prv != None:
try:
return base64.urlsafe_b64decode(self.prv.encrypt(plaintext))
except:
return None
def decrypt(self, ciphertext):
if self.type == Destination.PLAIN:
2016-06-03 17:02:02 +00:00
return ciphertext
2016-05-29 20:20:44 +00:00
2018-03-16 10:40:37 +00:00
if self.type == Destination.SINGLE and self.identity != None:
return self.identity.decrypt(ciphertext)
2016-05-29 20:20:44 +00:00
if self.type == Destination.GROUP:
return self.prv.decrypt(base64.urlsafe_b64encode(ciphertext))
def sign(self, message):
2018-03-16 10:40:37 +00:00
if self.type == Destination.SINGLE and self.identity != None:
return self.identity.sign(message)
2016-05-29 20:20:44 +00:00
else:
2018-03-16 10:40:37 +00:00
return None
# Creates an announce packet for this destination.
# Application specific data can be added to the announce.
def announce(self,app_data=None):
pass