2
0
mirror of https://github.com/ComradCollective/Comrad synced 2024-11-13 07:10:49 +00:00
Comrad/komrade/api/api.py
2020-09-14 11:50:57 +01:00

519 lines
15 KiB
Python

# ### Constants
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.constants import *
LAST_N_IN_INBOX = 10
### Imports
import os,time,sys,logging
from pathlib import Path
import asyncio,time,sys
from base64 import b64encode,b64decode
sys.path.append(os.path.dirname(__file__))
import logging
import asyncio
import shelve
from collections import OrderedDict
import pickle,os
from threading import Thread
from pathlib import Path
# local imports
from komrade import *
CACHE_DIR = os.path.join(os.path.expanduser('~'),'.komrade','.cache')
if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
MEMCACHE_FNFN=os.path.join(CACHE_DIR,'.memory')
### Logging
def logger():
import logging
handler = logging.StreamHandler()
formatter = logging.Formatter('[%(asctime)s]\n%(message)s\n')
handler.setFormatter(formatter)
logger = logging.getLogger('komrade')
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)
return logger
LOG = None
def log(*x):
global LOG
if not LOG: LOG=logger().debug
tolog=' '.join(str(_) for _ in x)
LOG(tolog)
## func
def bytes_from_file(filename,chunksize=8192):
with open(filename, 'rb') as f:
while True:
piece = f.read(chunksize)
if not piece:
break
yield piece
def get_random_id():
import uuid
return uuid.uuid4().hex
def get_random_binary_id():
import base64
idstr = get_random_id()
return base64.b64encode(idstr.encode())
### Headless API
def boot_lonely_selfless_node(port=8467):
async def go():
api = Api(log=log, port=port)
await api.connect_forever()
asyncio.run(go())
class NetworkStillConnectingError(OSError): pass
PORT_LISTEN = 5111
async def _getdb(self=None,port=PORT_LISTEN):
return None
from kademlia.network import Server
if self:
self.log('starting server on port %s..' % port)
import os
if self: self.log(os.getcwd())
node = Server(log=self.log if self else None) #fn='../p2p/data.db',log=(self.log if self else print)))
try:
if self: self.log('listening on port %s...' % format(port))
await node.listen(port)
except OSError:
raise NetworkStillConnectingError('Still connecting...')
#await asyncio.sleep(3)
if self: self.log('bootstrapping server..')
await node.bootstrap(NODES_PRIME)
if self: node.log = self.log
self.log('NODE:',node)
# if self and self.app:
# self.app.close_dialog()
return node
def logg(*x):
print(*x)
class Api(object):
def __init__(self,log=None,port=PORT_LISTEN):
self.log = log if log is not None else logg
self.port=port
# load file-based keys
self.load_keys()
async def connect_forever(self,save_every=60):
try:
i = 0
self._node = await self.connect()
while True:
if not i%90: self.log(f'Node status (tick {i}): {self._node}')
if i and not i%save_every: await self.flush()
i += 1
await asyncio.sleep(1)
# asyncio.sleep(0)
except (asyncio.CancelledError,KeyboardInterrupt) as e:
self.log('P2P node cancelled', e)
await self.flush()
finally:
# when canceled, print that it finished
self.log('P2P node shutting down')
pass
@property
async def node(self):
if not hasattr(self,'_node'):
await self.connect()
self._node.log=self.log
return self._node
async def connect(self):
port=self.port
# if self.app: self.app.open_dialog('hello?')
self.log('connecting on port %s...' % port)
node = await _getdb(self,port)
self.log(f'connect() has node {node}')
self._node = node
return node
def get_tor_python_session(self):
# from torpy.http.requests import TorRequests
# with TorRequests() as tor_requests:
# with tor_requests.get_session() as s:
# # return s
# from torpy.http.requests import tor_requests_session
# with tor_requests_session() as s: # returns requests.Session() object
# return s
pass
def get_tor_proxy_session(self):
session = requests.session()
# Tor uses the 9050 port as the default socks port
session.proxies = {'http': 'socks5://127.0.0.1:9050',
'https': 'socks5://127.0.0.1:9050'}
return session
def get_async_tor_proxy_session(self):
from requests_futures.sessions import FuturesSession
session = FuturesSession()
# Tor uses the 9050 port as the default socks port
session.proxies = {'http': 'socks5://127.0.0.1:9050',
'https': 'socks5://127.0.0.1:9050'}
return session
def tor_request(self,url,method='get',data=None):
stopfixthis
with self.get_tor_proxy_session() as s:
if method=='get':
return s.get(url)
elif method=='post':
self.log('data',data)
return s.post(url,data=data)
def request(self,Q,**kwargs):
self.log('request() Q:',Q)
res = self.tor_request(Q,**kwargs)
self.log('reqeust() <-',res)
return res
#@property
def load_keys(self):
# get key names
KEY_PATH_PUB ='.'
KEY_PATH_PRIV ='.'
pub_key_names = [x.split('.')[1] for x in os.listdir(KEY_PATH_PUB) if x.count('.')==2 and x.endswith('.loc')]
priv_key_names = [x.split('.')[1] for x in os.listdir(KEY_PATH_PRIV) if x.count('.')==2 and x.endswith('.key')]
key_names = set(pub_key_names)|set(priv_key_names)
self.log('get_keys() found public key names:',pub_key_names)
self.log('get_keys() found private key names:',priv_key_names)
# load and find all local
self._keys = {}
for key_name in key_names:
self.log('key_name =',key_name)
self._keys[key_name] = Komrade(key_name,api=self,create_if_missing=False)
# break into types
self.accounts = [self._keys[name] for name in priv_key_names]
self.contacts = [self._keys[name] for name in pub_key_names]
self.log('get_keys() loaded accounts:',self.accounts)
self.log('get_keys() loaded contacts:',self.contacts)
@property
def keys(self):
if not hasattr(self,'_keys'): self.load_keys()
return self._keys
#async
def personate(self,persona_name,create_if_missing=True):
komrade = Komrade(persona_name,api=self,create_if_missing=create_if_missing)
res = persona.boot()
self.log('personate() res =',res)
return persona
# komrade = self.keys[persona_name] if persona_name in self.keys else None
# if komrade is None and create_if_missing:
# self.keys[persona_name] = komrade = Komrade(persona_name, api=self, create_if_missing=create_if_missing)
# res = await persona.boot()
# self.log('BOOT RESULT:',res)
# return persona
async def upload(self,filename,file_id=None, uri='/file/',uri_part='/part/'):
import sys
if not file_id: file_id = get_random_id()
part_ids = []
part_keys = []
parts=[]
PARTS=[]
buffer_size=100
for part in bytes_from_file(filename,chunksize=1024*2):
part_id = get_random_id()
part_ids.append(part_id)
part_key='/part/'+part_id
part_keys.append(part_key)
parts.append(part)
# PARTS.append(part)
# self.log('part!:',sys.getsizeof(part))
#self.set(part_key,part)
if len(parts)>=buffer_size:
# self.log('setting...')
await self.set(part_keys,parts)
part_keys=[]
PARTS+=parts
parts=[]
# set all parts
#self.set(part_keys,PARTS)
# self.log('# parts:',len(PARTS))
if parts and part_keys:
await self.set(part_keys, parts)
# how many parts?
# self.log('# pieces!',len(part_ids))
file_store = {'ext':os.path.splitext(filename)[-1][1:], 'parts':part_ids}
# self.log('FILE STORE??',file_store)
await self.set_json(uri+file_id,file_store)
# file_store['data'].seek(0)
file_store['id']=file_id
return file_store
async def download(self,file_id):
self.log('file_id =',file_id)
file_store = await self.get_json_val('/file/'+file_id)
self.log('file_store =',file_store)
if file_store is None: return
self.log('file_store!?',file_store)
keys = ['/part/'+x for x in file_store['parts']]
#time,pieces,pub,sign = await self.get_json_val(keys)
pieces = await self.get_json_val(keys)
self.log('pieces = ',pieces)
file_store['parts_data']=pieces
return file_store
async def flush(self):
#self.log('saving back to db file...')
node = await self.node
node.storage.dump()
# self.log('DONE saving back to db file...')
async def get_posts(self,uri='/inbox/world'):
# get IDs
post_ids = await self.get_post_ids(uri)
# get posts
posts = [self.get_post(post_id) for post_id in post_ids]
return await asyncio.gather(*posts)
async def read_inbox(self,uri_inbox=None):
if uri_inbox is None: uri_inbox = P2P_PREFIX_INBOX+self.name.encode()
node = await self.node
inbox_ids = await node.get(uri_inbox)
if inbox_ids is not None:
inbox_ids = inbox_ids.split(BSEP)
self.log('found inbox IDs:',inbox_ids)
msgs_toread = [self.read_msg(msg_id) for msg_id in inbox_ids]
msgs = await asyncio.gather(*msgs_toread)
self.log('read_inbox() msgs = ',msgs)
return msgs
return []
async def read_outbox(self,uri_outbox=None):
if uri_outbox is None: uri_outbox = P2P_PREFIX_OUTBOX+self.name.encode()
return await self.read_inbox(uri_outbox)
async def read_msg(self,msg_id):
self.log(f'Persona.read_msg({msg_id}) ?')
uri_msg=P2P_PREFIX_POST+msg_id
node = await self.node
komrade = await self.komrade
res = await node.get(uri_msg)
self.log('res = ',res)
if res is not None:
double_encrypted_payload_b64 = res
single_encrypted_payload = self.decrypt(double_encrypted_payload_b64, komrade.pubkey_b64)
self.log('GOT ENRYPTED PAYLOAD:',single_encrypted_payload)
signed_encrypted_payload_b64,from_pubkey_b64,name_b64,time_b64 = single_encrypted_payload.split(BSEP)
self.log('signed_encrypted_payload =',signed_encrypted_payload_b64)
self.log('from_pubkey_b64 =',from_pubkey_b64)
self.log('time_b64 =',time_b64)
from_name = b64decode(name_b64).decode()
self.log('from_name =',from_name)
timestamp = b64decode(time_b64).decode()
tmpP = Komrade(from_name)
await tmpP.boot()
from_pubkey_b64_accto_name = tmpP.pubkey_b64
assert from_pubkey_b64==from_pubkey_b64_accto_name
encrypted_payload_b64 = self.verify(signed_encrypted_payload_b64, from_pubkey_b64)
self.log('encrypted_payload_b64 =',encrypted_payload_b64)
payload = self.decrypt(encrypted_payload_b64, from_pubkey_b64)
self.log('payload =',payload)
return {
'success':True,
'content':payload,
'from_name':from_name,
'from_pubkey_b64':from_pubkey_b64,
'timestamp':timestamp
}
return {'error':'Unknown'}
async def refresh_inboxes(self):
uris_to_get=[]
for komrade in self.accounts:
inbox = await persona.load_inbox(decrypt_msg_uri=True, last=LAST_N_IN_INBOX)
for decr_msg_uri in inbox:
uris_to_get.append(self.get_msg(decr_msg_uri))
# uris_to_get+=inbox
res = await asyncio.gather(*uris_to_get)
for decr_msg_uri,encr_msg in res:
self.memcache[decr_msg_uri]=encr_msg
self.memcache_save()
@property
def memcache(self):
if not hasattr(self,'_memcache'):
self._memcache = OrderedDict()
if os.path.exists(MEMCACHE_FNFN):
import pickle
try:
self._memcache = pickle.load(open(MEMCACHE_FNFN,'rb'))
except EOFError:
pass
return self._memcache
def memcache_save(self):
import pickle
with open(MEMCACHE_FNFN,'wb') as of:
pickle.dump(self.memcache, of)
self.log('>> saved:',MEMCACHE_FNFN)
async def get_msg(self,decr_msg_uri):
self.log('get_msg()',decr_msg_uri)
rval=self.memcache.get(decr_msg_uri)
self.log('got <--',rval)
if rval is not None:
self.log('in memcache')
encr_msg = rval
else:
self.log('>> downloading',decr_msg_uri,'...')
node = await self.node
encr_msg = await node.get(decr_msg_uri)
self.log('downloaded:',encr_msg)
return (decr_msg_uri,encr_msg)
#self.memcache.
async def see(self,decr_msg_id):
res=await self.get(decr_msg_id)
self.log('see() saw',res)
return decr_msg_id
async def test1():
api = Api()
marx=await api.personate('marx')
elon=await api.personate('elon')
res = await marx.send(b'secret',to=elon)
print(marx,elon,res)
async def test():
api = Api()
# message?
marx=await api.personate('marx')
elon=await api.personate('elon')
res = await marx.send(b'secret',to=elon)
res = await elon.send(b'secret back',to=marx)
# print(marx,elon,res)
# get overall inbox
meta_inbox = await api.refresh_inboxes()
api.log('meta_inbox',meta_inbox)
keys = api.memcache.keys()
api.log('ALL KEYS =',keys)
for key in keys:
val = api.memcache.get(key)
api.log(key,'-->',val)
# stop
#async
async def test_keyserver():
api = Api()
marx = api.personate('marx')
elon = api.personate('elon')
zuck = api.personate('zuck')
print('marx',marx.pubkey_b64)
print('elon',elon.pubkey_b64)
print('zuck',zuck.pubkey_b64)
#marx = await api.personate(marx)
#res = await api.get_externally_signed_pubkey('marx')
#res = await api.get_externally_signed_pubkey('marx')
#return res
await elon.send(b'oh no',to=marx)
if __name__=='__main__':
asyncio.run(test_keyserver())