epicyon/session.py

145 lines
5.6 KiB
Python
Raw Normal View History

2019-06-28 18:55:29 +00:00
__filename__ = "session.py"
__author__ = "Bob Mottram"
__license__ = "AGPL3+"
2019-08-29 13:35:29 +00:00
__version__ = "1.0.0"
2019-06-28 18:55:29 +00:00
__maintainer__ = "Bob Mottram"
__email__ = "bob@freedombone.net"
__status__ = "Production"
2019-07-16 14:23:06 +00:00
import os
2019-08-14 16:46:10 +00:00
import sys
2019-06-28 18:55:29 +00:00
import requests
2019-07-02 10:39:55 +00:00
from utils import urlPermitted
2019-06-28 18:55:29 +00:00
import json
2019-06-30 18:23:18 +00:00
baseDirectory=None
2019-07-02 09:25:29 +00:00
def createSession(domain: str, port: int, onionRoute: bool):
2019-06-28 18:55:29 +00:00
session = requests.session()
2019-07-02 17:11:59 +00:00
#if domain.startswith('127.') or domain.startswith('192.') or domain.startswith('10.'):
# session.mount('http://', SourceAddressAdapter(domain))
2019-07-02 09:25:29 +00:00
#session.mount('http://', SourceAddressAdapter((domain, port)))
2019-06-28 18:55:29 +00:00
if onionRoute:
session.proxies = {}
session.proxies['http'] = 'socks5h://localhost:9050'
session.proxies['https'] = 'socks5h://localhost:9050'
return session
2019-08-14 16:46:10 +00:00
def getJson(session,url: str,headers: {},params: {}, \
2019-08-29 14:30:09 +00:00
version='1.0.0',httpPrefix='https',domain='testdomain') -> {}:
2019-09-01 13:13:52 +00:00
if not isinstance(url, str):
print('url: '+str(url))
print('ERROR: getJson url should be a string')
return None
2019-06-28 18:55:29 +00:00
sessionParams={}
sessionHeaders={}
if headers:
sessionHeaders=headers
if params:
sessionParams=params
2019-08-26 14:08:41 +00:00
sessionHeaders['User-Agent']='Epicyon/'+version
if domain:
sessionHeaders['User-Agent']+='; +'+httpPrefix+'://'+domain+'/'
2019-08-14 16:46:10 +00:00
#"Mozilla/5.0 (Linux; Android 5.1.1; Nexus 5)"
2019-07-16 10:19:04 +00:00
if not session:
print('WARN: no session specified for getJson')
2019-08-18 12:52:39 +00:00
#session.cookies.clear()
2019-07-04 17:31:41 +00:00
try:
2019-08-15 21:34:25 +00:00
result=session.get(url, headers=sessionHeaders, params=sessionParams)
2019-07-04 17:31:41 +00:00
return result.json()
except Exception as e:
2019-07-19 13:32:58 +00:00
print('ERROR: getJson failed')
2019-09-01 13:13:52 +00:00
print('url: '+str(url))
2019-07-19 13:32:58 +00:00
print('headers: '+str(sessionHeaders))
print('params: '+str(sessionParams))
print(e)
2019-07-04 17:31:41 +00:00
return None
2019-06-28 19:36:39 +00:00
2019-07-09 14:20:23 +00:00
def postJson(session,postJsonObject: {},federationList: [],inboxUrl: str,headers: {},capability: str) -> str:
2019-06-28 19:36:39 +00:00
"""Post a json message to the inbox of another person
2019-07-06 11:03:51 +00:00
Supplying a capability, such as "inbox:write"
2019-06-28 19:36:39 +00:00
"""
2019-07-06 11:03:51 +00:00
# always allow capability requests
if not capability.startswith('cap'):
# check that we are posting to a permitted domain
2019-07-09 14:20:23 +00:00
if not urlPermitted(inboxUrl,federationList,capability):
2019-07-16 11:33:40 +00:00
print('postJson: '+inboxUrl+' not permitted')
2019-07-06 11:03:51 +00:00
return None
2019-06-28 20:22:36 +00:00
2019-08-15 21:34:25 +00:00
postResult = session.post(url = inboxUrl, data = json.dumps(postJsonObject), headers=headers)
2019-10-26 12:01:22 +00:00
if postResult:
return postResult.text
return None
2019-07-16 14:23:06 +00:00
2019-08-18 11:07:06 +00:00
def postJsonString(session,postJsonStr: str, \
federationList: [], \
inboxUrl: str, \
headers: {}, \
capability: str, \
2019-10-23 18:44:03 +00:00
debug: bool) -> (bool,bool):
"""Post a json message string to the inbox of another person
Supplying a capability, such as "inbox:write"
2019-10-23 18:44:03 +00:00
The second boolean returned is true if the send is unauthorized
NOTE: Here we post a string rather than the original json so that
conversions between string and json format don't invalidate
the message body digest of http signatures
"""
# always allow capability requests
if not capability.startswith('cap'):
# check that we are posting to a permitted domain
if not urlPermitted(inboxUrl,federationList,capability):
2019-08-18 11:07:06 +00:00
print('postJson: '+inboxUrl+' not permitted by capabilities')
2019-10-23 18:44:03 +00:00
return None,None
postResult = session.post(url = inboxUrl, data = postJsonStr, headers=headers)
2019-08-26 12:04:09 +00:00
if postResult.status_code<200 or postResult.status_code>202:
2019-10-21 16:53:06 +00:00
#if postResult.status_code==400:
2019-11-09 21:39:04 +00:00
# headers['content-type']='application/ld+json'
2019-10-21 16:53:06 +00:00
# postResult = session.post(url = inboxUrl, data = postJsonStr, headers=headers)
# if not (postResult.status_code<200 or postResult.status_code>202):
# return True
2019-10-23 18:45:16 +00:00
if postResult.status_code>=400 and postResult.status_code<=405 and postResult.status_code!=404:
2019-08-26 12:46:08 +00:00
print('WARN: >>> Post to '+inboxUrl+' is unauthorized <<<')
2019-10-23 18:44:03 +00:00
return False,True
2019-08-26 12:46:08 +00:00
else:
2019-10-21 16:27:45 +00:00
print('WARN: Failed to post to '+inboxUrl+' with headers '+str(headers))
2019-08-26 12:46:08 +00:00
print('status code '+str(postResult.status_code))
2019-10-23 18:44:03 +00:00
return False,False
return True,False
2019-07-16 14:23:06 +00:00
def postImage(session,attachImageFilename: str,federationList: [],inboxUrl: str,headers: {},capability: str) -> str:
"""Post an image to the inbox of another person or outbox via c2s
Supplying a capability, such as "inbox:write"
"""
# always allow capability requests
if not capability.startswith('cap'):
# check that we are posting to a permitted domain
if not urlPermitted(inboxUrl,federationList,capability):
print('postJson: '+inboxUrl+' not permitted')
return None
if not (attachImageFilename.endswith('.jpg') or \
attachImageFilename.endswith('.jpeg') or \
attachImageFilename.endswith('.png') or \
attachImageFilename.endswith('.gif')):
print('Image must be png, jpg, or gif')
return None
if not os.path.isfile(attachImageFilename):
print('Image not found: '+attachImageFilename)
return None
contentType='image/jpeg'
if attachImageFilename.endswith('.png'):
contentType='image/png'
if attachImageFilename.endswith('.gif'):
contentType='image/gif'
headers['Content-type']=contentType
with open(attachImageFilename, 'rb') as avFile:
mediaBinary = avFile.read()
postResult = session.post(url=inboxUrl, data=mediaBinary, headers=headers)
2019-10-26 12:01:22 +00:00
if postResult:
return postResult.text
2019-07-16 14:23:06 +00:00
return None