mirror of
https://github.com/ComradCollective/Comrad
synced 2024-10-31 03:20:18 +00:00
139 lines
4.3 KiB
Python
139 lines
4.3 KiB
Python
"""
|
|
There is only one operator!
|
|
Running on node prime.
|
|
"""
|
|
# internal imports
|
|
import os,sys; sys.path.append(os.path.abspath(os.path.join(os.path.abspath(os.path.join(os.path.dirname(__file__),'..')),'..')))
|
|
from komrade import *
|
|
from komrade.backend.crypt import *
|
|
from komrade.backend.operators import *
|
|
from komrade.backend.mazes import *
|
|
|
|
|
|
|
|
class TheOperator(Operator):
|
|
"""
|
|
The remote operator! Only one!
|
|
"""
|
|
|
|
|
|
def __init__(self, name = OPERATOR_NAME, passphrase='acc'):
|
|
"""
|
|
Boot up the operator. Requires knowing or setting a password of memory.
|
|
"""
|
|
# init req paths
|
|
# if not os.path.exists(PATH_OPERATOR): os.makedirs(PATH_OPERATOR)
|
|
if not passphrase:
|
|
passphrase=getpass.getpass('Hello, this is the Operator speaking. What is the passphrase?\n> ')
|
|
super().__init__(name,passphrase)
|
|
|
|
def decrypt_incoming(self,data):
|
|
# step 1 split:
|
|
data_unencr,data_encr = data.split(BSEP)
|
|
self.log('data_unencr =',data_unencr)
|
|
self.log('data_encr =',data_encr)
|
|
if data_encr and 'name' in data_unencr:
|
|
name=data_unencr['name']
|
|
keychain=data_unencr.get('keychain',{})
|
|
|
|
# decrypt using this user's pubkey on record
|
|
caller = Operator(name)
|
|
from_pubkey = user.pubkey(keychain=keychain)
|
|
data_unencr2 = SMessage(OPERATOR.privkey_, from_pubkey).unwrap(data_encr)
|
|
|
|
if type(data_unencr)==dict and type(data_unencr2)==dict:
|
|
data = data_unencr
|
|
dict_merge(data_unencr2, data)
|
|
else:
|
|
data=(data_unencr,data_unencr2)
|
|
else:
|
|
data = data_unencr
|
|
return data
|
|
|
|
def route(self, data):
|
|
data = self.decrypt_incoming(data)
|
|
self.log('DATA =',data)
|
|
return 'success!'
|
|
|
|
|
|
def init_operators():
|
|
op = Operator(
|
|
name=OPERATOR_NAME,
|
|
path_crypt_keys=PATH_CRYPT_OP_KEYS,
|
|
path_crypt_data=PATH_CRYPT_OP_DATA
|
|
)
|
|
|
|
phone = Operator(
|
|
name=TELEPHONE_NAME,
|
|
path_crypt_keys=PATH_CRYPT_OP_KEYS,
|
|
path_crypt_data=PATH_CRYPT_OP_KEYS
|
|
)
|
|
|
|
keys_to_return = [k for k in KEYMAKER_DEFAULT_KEYS_TO_RETURN if not k.startswith('admin')]
|
|
|
|
op_decr_keys = op.forge_new_keys(keys_to_return=keys_to_return)
|
|
|
|
phone_decr_keys = phone.forge_new_keys(keys_to_return=keys_to_return)
|
|
|
|
print('\n'*5)
|
|
print('OPERATOR_KEYCHAIN =',op_decr_keys)
|
|
print()
|
|
print('TELEPHONE_KEYCHAIN =',phone_decr_keys)
|
|
|
|
# stringify
|
|
for k,v in phone_decr_keys.items():
|
|
v_s = v.decode('utf-8')
|
|
phone_decr_keys[k]=v_s
|
|
for k,v in op_decr_keys.items():
|
|
v_s = v.decode('utf-8')
|
|
op_decr_keys[k]=v_s
|
|
|
|
|
|
builtin_keychains = {TELEPHONE_NAME:phone_decr_keys, OPERATOR_NAME:op_decr_keys}
|
|
builtin_keychains_s = json.dumps(builtin_keychains)
|
|
builtin_keychains_b = builtin_keychains_s.encode('utf-8')
|
|
|
|
builtin_keychains_b_decr = KomradeSymmetricKeyWithoutPassphrase()
|
|
builtin_keychains_b_encr = builtin_keychains_b_decr.encrypt(builtin_keychains_b)
|
|
|
|
builtin_keychains_b_decr_b64 = b64encode(builtin_keychains_b_decr.key)
|
|
builtin_keychains_b_encr_b64 = b64encode(builtin_keychains_b_encr)
|
|
|
|
with open(PATH_BUILTIN_KEYCHAINS_ENCR,'wb') as of:
|
|
of.write(builtin_keychains_b_encr_b64)
|
|
# pickle.dump(builtin_keychains, )
|
|
print('>> saved:',PATH_BUILTIN_KEYCHAINS_ENCR)
|
|
with open(PATH_BUILTIN_KEYCHAINS_DECR,'wb') as of:
|
|
of.write(builtin_keychains_b_decr_b64)
|
|
print('>> saved:',PATH_BUILTIN_KEYCHAINS_DECR)
|
|
|
|
|
|
|
|
# op_pub = op.pubkey_decr_
|
|
# phone_pub = phone.pubkey_decr_
|
|
# phone_priv = phone.privkey_decr_
|
|
|
|
# print('OPERATOR_PUBKEY_DECR =',b64encode(op_pub))
|
|
# print('TELEPHONE_PUBKEY_DECR =',b64encode(phone_pub))
|
|
# print('TELEPHONE_PRIVKEY_DECR =',b64encode(phone_priv))
|
|
# return {
|
|
# 'op.keychain()':op.keychain(),
|
|
# 'phone.keychain()':phone.keychain()
|
|
# }
|
|
|
|
|
|
def test_op():
|
|
op = TheOperator()
|
|
#op.boot()
|
|
#pubkey = op.keychain()['pubkey']
|
|
#pubkey_b64 = b64encode(pubkey)
|
|
#print(pubkey_b64)
|
|
keychain = op.keychain(force=True)
|
|
from pprint import pprint
|
|
pprint(keychain)
|
|
|
|
pubkey = op.keychain()['pubkey']
|
|
pubkey_b64 = b64encode(pubkey)
|
|
print(pubkey)
|
|
|
|
if __name__ == '__main__': test_op() |