indymedia
/
epicyon
mirror of https://gitlab.com/bashrc2/epicyon
7
4
Fork 1
Code Issues 14 Releases Wiki Activity

Move secure mode function to its own module

merge-requests/30/head
Bob Mottram 2024-02-29 11:03:23 +00:00
parent 779a58e81f
commit eeda2a2934
2 changed files with 122 additions and 93 deletions

127
daemon.py
View File

@ -381,7 +381,6 @@ from utils import remove_post_from_cache
from utils import get_nickname_from_actor
from utils import get_domain_from_actor
from utils import get_status_number
from utils import url_permitted
from utils import load_json
from utils import save_json
from utils import is_suspended
@ -411,10 +410,7 @@ from cache import clear_actor_cache
from cache import check_for_changed_actor
from cache import store_person_in_cache
from cache import get_person_from_cache
from cache import get_person_pub_key
from httpsig import signed_get_key_id
from httpsig import getheader_signature_input
from httpsig import verify_post_headers
from theme import reset_theme_designer_settings
from theme import set_theme_from_designer
from theme import scan_themes_for_scripts
@ -479,6 +475,7 @@ from httprequests import request_csv
from httprequests import request_ssml
from httprequests import request_http
from httprequests import request_icalendar
from securemode import secure_mode
# maximum number of posts to list in outbox feed
MAX_POSTS_IN_FEED = 12
@ -682,79 +679,6 @@ class PubServer(BaseHTTPRequestHandler):
else:
print('ERROR: unable to create vote')
def _secure_mode(self, curr_session, proxy_type: str,
force: bool) -> bool:
"""http authentication of GET requests for json
aka authorized fetch
"""
if not self.server.secure_mode and not force:
return True
key_id = signed_get_key_id(self.headers, self.server.debug)
if not key_id:
if self.server.debug:
print('AUTH: secure mode, ' +
'failed to obtain key_id from signature')
return False
# is the key_id (actor) valid?
if not url_permitted(key_id, self.server.federation_list):
if self.server.debug:
print('AUTH: Secure mode GET request not permitted: ' + key_id)
return False
if self.server.onion_domain:
if '.onion/' in key_id:
curr_session = self.server.session_onion
proxy_type = 'tor'
if self.server.i2p_domain:
if '.i2p/' in key_id:
curr_session = self.server.session_i2p
proxy_type = 'i2p'
curr_session = \
establish_session("secure mode",
curr_session, proxy_type,
self.server)
if not curr_session:
return False
# obtain the public key. key_id is the actor
pub_key = \
get_person_pub_key(self.server.base_dir,
curr_session, key_id,
self.server.person_cache, self.server.debug,
self.server.project_version,
self.server.http_prefix,
self.server.domain,
self.server.onion_domain,
self.server.i2p_domain,
self.server.signing_priv_key_pem)
if not pub_key:
if self.server.debug:
print('AUTH: secure mode failed to ' +
'obtain public key for ' + key_id)
return False
# was an error http code returned?
if isinstance(pub_key, dict):
if self.server.debug:
print('AUTH: failed to ' +
'obtain public key for ' + key_id +
' ' + str(pub_key))
return False
# verify the GET request without any digest
if verify_post_headers(self.server.http_prefix,
self.server.domain_full,
pub_key, self.headers,
self.path, True, None, '', self.server.debug):
return True
if self.server.debug:
print('AUTH: secure mode authorization failed for ' + key_id)
return False
def _login_headers(self, file_format: str, length: int,
calling_domain: str) -> None:
self.send_response(200)
@ -12462,7 +12386,8 @@ class PubServer(BaseHTTPRequestHandler):
'_GET', '_show_replies_to_post',
debug)
else:
if self._secure_mode(curr_session, proxy_type, False):
if secure_mode(curr_session, proxy_type, False,
self.server, self.headers, self.path):
msg_str = json.dumps(replies_json, ensure_ascii=False)
msg_str = convert_domains(calling_domain,
referer_domain,
@ -12582,7 +12507,8 @@ class PubServer(BaseHTTPRequestHandler):
'_GET', '_show_replies_to_post',
debug)
else:
if self._secure_mode(curr_session, proxy_type, False):
if secure_mode(curr_session, proxy_type, False,
self.server, self.headers, self.path):
msg_str = json.dumps(replies_json, ensure_ascii=False)
msg_str = convert_domains(calling_domain,
referer_domain,
@ -12709,7 +12635,8 @@ class PubServer(BaseHTTPRequestHandler):
fitness_performance(getreq_start_time, self.server.fitness,
'_GET', '_show_roles', debug)
else:
if self._secure_mode(curr_session, proxy_type, False):
if secure_mode(curr_session, proxy_type, False,
self.server, self.headers, self.path):
roles_list = get_actor_roles_list(actor_json)
msg_str = json.dumps(roles_list, ensure_ascii=False)
msg_str = convert_domains(calling_domain,
@ -12849,8 +12776,11 @@ class PubServer(BaseHTTPRequestHandler):
'_GET', '_show_skills',
self.server.debug)
else:
if self._secure_mode(curr_session,
proxy_type, False):
if secure_mode(curr_session,
proxy_type, False,
self.server,
self.headers,
self.path):
actor_skills_list = \
get_occupation_skills(actor_json)
skills = \
@ -13300,7 +13230,8 @@ class PubServer(BaseHTTPRequestHandler):
'_GET', '_show_post_from_file',
debug)
else:
if self._secure_mode(curr_session, proxy_type, False):
if secure_mode(curr_session, proxy_type, False,
self.server, self.headers, self.path):
if not include_create_wrapper and \
post_json_object['type'] == 'Create' and \
has_object_dict(post_json_object):
@ -15273,7 +15204,8 @@ class PubServer(BaseHTTPRequestHandler):
'_GET', '_show_outbox_timeline',
debug)
else:
if self._secure_mode(curr_session, proxy_type, False):
if secure_mode(curr_session, proxy_type, False,
self.server, self.headers, self.path):
onion_domain = self.server.onion_domain
i2p_domain = self.server.i2p_domain
msg_str = json.dumps(outbox_feed,
@ -15595,7 +15527,8 @@ class PubServer(BaseHTTPRequestHandler):
self.server.getreq_busy = False
return True
else:
if self._secure_mode(curr_session, proxy_type, False):
if secure_mode(curr_session, proxy_type, False,
self.server, self.headers, self.path):
onion_domain = self.server.onion_domain
i2p_domain = self.server.i2p_domain
msg_str = json.dumps(shares,
@ -15754,7 +15687,8 @@ class PubServer(BaseHTTPRequestHandler):
debug)
return True
else:
if self._secure_mode(curr_session, proxy_type, False):
if secure_mode(curr_session, proxy_type, False,
self.server, self.headers, self.path):
if '/users/' in path:
nickname = path.split('/users/')[1]
if '/' in nickname:
@ -15912,7 +15846,8 @@ class PubServer(BaseHTTPRequestHandler):
debug)
return True
else:
if self._secure_mode(curr_session, proxy_type, False):
if secure_mode(curr_session, proxy_type, False,
self.server, self.headers, self.path):
msg_str = json.dumps(following,
ensure_ascii=False)
msg_str = convert_domains(calling_domain,
@ -16068,7 +16003,8 @@ class PubServer(BaseHTTPRequestHandler):
debug)
return True
else:
if self._secure_mode(curr_session, proxy_type, False):
if secure_mode(curr_session, proxy_type, False,
self.server, self.headers, self.path):
msg_str = json.dumps(following,
ensure_ascii=False)
msg_str = convert_domains(calling_domain,
@ -16226,7 +16162,8 @@ class PubServer(BaseHTTPRequestHandler):
debug)
return True
else:
if self._secure_mode(curr_session, proxy_type, False):
if secure_mode(curr_session, proxy_type, False,
self.server, self.headers, self.path):
if '/users/' in path:
nickname = path.split('/users/')[1]
if '/' in nickname:
@ -16422,7 +16359,8 @@ class PubServer(BaseHTTPRequestHandler):
if self.server.debug:
print('DEBUG: html actor sent')
else:
if self._secure_mode(curr_session, proxy_type, False):
if secure_mode(curr_session, proxy_type, False,
self.server, self.headers, self.path):
accept_str = self.headers['Accept']
msg_str = json.dumps(actor_json, ensure_ascii=False)
msg_str = convert_domains(calling_domain,
@ -17665,7 +17603,8 @@ class PubServer(BaseHTTPRequestHandler):
print('DEBUG: followers synchronization request ' +
self.path + ' ' + calling_domain)
# check authorized fetch
if self._secure_mode(curr_session, proxy_type, False):
if secure_mode(curr_session, proxy_type, False,
self.server, self.headers, self.path):
nickname = get_nickname_from_actor(self.path)
sync_cache = self.server.followers_sync_cache
sync_json, _ = \
@ -21564,8 +21503,10 @@ class PubServer(BaseHTTPRequestHandler):
self.server.getreq_busy = False
return
if not self._secure_mode(curr_session,
proxy_type, False):
if not secure_mode(curr_session,
proxy_type, False,
self.server, self.headers,
self.path):
if self.server.debug:
print('WARN: Unauthorized GET')
self._404(143)

88
securemode.py 100644
View File

@ -0,0 +1,88 @@
__filename__ = "securemode.py"
__author__ = "Bob Mottram"
__license__ = "AGPL3+"
__version__ = "1.5.0"
__maintainer__ = "Bob Mottram"
__email__ = "bob@libreserver.org"
__status__ = "Production"
__module_group__ = "Core"
from httpsig import verify_post_headers
from session import establish_session
from utils import url_permitted
from httpsig import signed_get_key_id
from cache import get_person_pub_key
def secure_mode(curr_session, proxy_type: str,
force: bool, server, headers: {}, path: str) -> bool:
"""http authentication of GET requests for json
See AUTHORIZED_FETCH in https://docs.joinmastodon.org/admin/config
"""
if not server.secure_mode and not force:
return True
key_id = signed_get_key_id(headers, server.debug)
if not key_id:
if server.debug:
print('AUTH: secure mode, ' +
'failed to obtain key_id from signature')
return False
# is the key_id (actor) valid?
if not url_permitted(key_id, server.federation_list):
if server.debug:
print('AUTH: Secure mode GET request not permitted: ' + key_id)
return False
if server.onion_domain:
if '.onion/' in key_id:
curr_session = server.session_onion
proxy_type = 'tor'
if server.i2p_domain:
if '.i2p/' in key_id:
curr_session = server.session_i2p
proxy_type = 'i2p'
curr_session = \
establish_session("secure mode",
curr_session, proxy_type,
server)
if not curr_session:
return False
# obtain the public key. key_id is the actor
pub_key = \
get_person_pub_key(server.base_dir,
curr_session, key_id,
server.person_cache, server.debug,
server.project_version,
server.http_prefix,
server.domain,
server.onion_domain,
server.i2p_domain,
server.signing_priv_key_pem)
if not pub_key:
if server.debug:
print('AUTH: secure mode failed to ' +
'obtain public key for ' + key_id)
return False
# was an error http code returned?
if isinstance(pub_key, dict):
if server.debug:
print('AUTH: failed to ' +
'obtain public key for ' + key_id +
' ' + str(pub_key))
return False
# verify the GET request without any digest
if verify_post_headers(server.http_prefix,
server.domain_full,
pub_key, headers,
path, True, None, '', server.debug):
return True
if server.debug:
print('AUTH: secure mode authorization failed for ' + key_id)
return False