Refactored whoogle session mgmt
Now allows a fallback "default" session to be used if a user's browser is blocking cookiesmain
Ben Busby
parent
64af72abb5
commit
32e837a5e0
|
|
@ -1,28 +1,28 @@
|
|||
from app.utils.misc import generate_user_keys
|
||||
from cryptography.fernet import Fernet
|
||||
from flask import Flask
|
||||
from flask_session import Session
|
||||
import os
|
||||
|
||||
app = Flask(__name__, static_folder=os.path.dirname(os.path.abspath(__file__)) + '/static')
|
||||
app.user_elements = {}
|
||||
app.config['SECRET_KEY'] = os.urandom(16)
|
||||
app.default_key_set = generate_user_keys()
|
||||
app.no_cookie_ips = []
|
||||
app.config['SECRET_KEY'] = os.urandom(32)
|
||||
app.config['SESSION_TYPE'] = 'filesystem'
|
||||
app.config['VERSION_NUMBER'] = '0.2.0'
|
||||
app.config['APP_ROOT'] = os.getenv('APP_ROOT', os.path.dirname(os.path.abspath(__file__)))
|
||||
app.config['STATIC_FOLDER'] = os.getenv('STATIC_FOLDER', os.path.join(app.config['APP_ROOT'], 'static'))
|
||||
app.config['CONFIG_PATH'] = os.getenv('CONFIG_VOLUME', app.config['STATIC_FOLDER'] + '/config')
|
||||
app.config['USER_CONFIG'] = os.path.join(app.config['STATIC_FOLDER'], 'custom_config')
|
||||
app.config['SESSION_FILE_DIR'] = app.config['CONFIG_PATH']
|
||||
app.config['SESSION_COOKIE_SECURE'] = True
|
||||
app.config['CONFIG_PATH'] = os.getenv('CONFIG_VOLUME', os.path.join(app.config['STATIC_FOLDER'], 'config'))
|
||||
app.config['DEFAULT_CONFIG'] = os.path.join(app.config['CONFIG_PATH'], 'config.json')
|
||||
app.config['SESSION_FILE_DIR'] = os.path.join(app.config['CONFIG_PATH'], 'session')
|
||||
app.config['SESSION_COOKIE_SAMESITE'] = 'Strict'
|
||||
|
||||
if not os.path.exists(app.config['CONFIG_PATH']):
|
||||
os.makedirs(app.config['CONFIG_PATH'])
|
||||
|
||||
if not os.path.exists(app.config['USER_CONFIG']):
|
||||
os.makedirs(app.config['USER_CONFIG'])
|
||||
if not os.path.exists(app.config['SESSION_FILE_DIR']):
|
||||
os.makedirs(app.config['SESSION_FILE_DIR'])
|
||||
|
||||
sess = Session()
|
||||
sess.init_app(app)
|
||||
Session(app)
|
||||
|
||||
from app import routes
|
||||
|
|
|
|||
|
|
@ -37,11 +37,19 @@ def auth_required(f):
|
|||
|
||||
@app.before_request
|
||||
def before_request_func():
|
||||
# Generate secret key for user if unavailable
|
||||
g.request_params = request.args if request.method == 'GET' else request.form
|
||||
g.cookies_disabled = False
|
||||
|
||||
# Generate session values for user if unavailable
|
||||
if not valid_user_session(session):
|
||||
session['config'] = {'url': request.url_root}
|
||||
session['keys'] = generate_user_keys()
|
||||
session['config'] = json.load(open(app.config['DEFAULT_CONFIG'])) \
|
||||
if os.path.exists(app.config['DEFAULT_CONFIG']) else {'url': request.url_root}
|
||||
session['uuid'] = str(uuid.uuid4())
|
||||
session['fernet_keys'] = generate_user_keys(True)
|
||||
|
||||
# Flag cookies as possibly disabled in order to prevent against
|
||||
# unnecessary session directory expansion
|
||||
g.cookies_disabled = True
|
||||
|
||||
if session['uuid'] not in app.user_elements:
|
||||
app.user_elements.update({session['uuid']: 0})
|
||||
|
|
@ -63,11 +71,22 @@ def before_request_func():
|
|||
|
||||
@app.after_request
|
||||
def after_request_func(response):
|
||||
# Regenerate element key if all elements have been served to user
|
||||
if app.user_elements[session['uuid']] <= 0 and '/element' in request.url:
|
||||
session['keys']['element_key'] = Fernet.generate_key()
|
||||
# Regenerate element key if all elements have been served to user
|
||||
session['fernet_keys']['element_key'] = '' if not g.cookies_disabled else app.default_key_set['element_key']
|
||||
app.user_elements[session['uuid']] = 0
|
||||
|
||||
# Check if address consistently has cookies blocked, in which case start removing session
|
||||
# files after creation.
|
||||
# Note: This is primarily done to prevent overpopulation of session directories, since browsers that
|
||||
# block cookies will still trigger Flask's session creation routine with every request.
|
||||
if g.cookies_disabled and request.remote_addr not in app.no_cookie_ips:
|
||||
app.no_cookie_ips.append(request.remote_addr)
|
||||
elif g.cookies_disabled and request.remote_addr in app.no_cookie_ips:
|
||||
session_list = list(session.keys())
|
||||
for key in session_list:
|
||||
session.pop(key)
|
||||
|
||||
return response
|
||||
|
||||
|
||||
|
|
@ -79,6 +98,9 @@ def unknown_page(e):
|
|||
@app.route('/', methods=['GET'])
|
||||
@auth_required
|
||||
def index():
|
||||
# Reset keys
|
||||
session['fernet_keys'] = generate_user_keys(g.cookies_disabled)
|
||||
|
||||
return render_template('index.html',
|
||||
languages=Config.LANGUAGES,
|
||||
countries=Config.COUNTRIES,
|
||||
|
|
@ -103,8 +125,7 @@ def opensearch():
|
|||
|
||||
@app.route('/autocomplete', methods=['GET', 'POST'])
|
||||
def autocomplete():
|
||||
request_params = request.args if request.method == 'GET' else request.form
|
||||
q = request_params.get('q')
|
||||
q = g.request_params.get('q')
|
||||
|
||||
if not q and not request.data:
|
||||
return jsonify({'?': []})
|
||||
|
|
@ -117,11 +138,10 @@ def autocomplete():
|
|||
@app.route('/search', methods=['GET', 'POST'])
|
||||
@auth_required
|
||||
def search():
|
||||
# Clear previous elements and generate a new key each time a new search is performed
|
||||
# Reset element counter
|
||||
app.user_elements[session['uuid']] = 0
|
||||
session['keys']['element_key'] = Fernet.generate_key()
|
||||
|
||||
search_util = RoutingUtils(request, g.user_config, session)
|
||||
search_util = RoutingUtils(request, g.user_config, session, cookies_disabled=g.cookies_disabled)
|
||||
query = search_util.new_search_query()
|
||||
|
||||
# Redirect to home if invalid/blank search
|
||||
|
|
@ -157,7 +177,7 @@ def config():
|
|||
return json.dumps(g.user_config.__dict__)
|
||||
elif request.method == 'PUT':
|
||||
if 'name' in request.args:
|
||||
config_pkl = os.path.join(app.config['USER_CONFIG'], request.args.get('name'))
|
||||
config_pkl = os.path.join(app.config['CONFIG_PATH'], request.args.get('name'))
|
||||
session['config'] = pickle.load(open(config_pkl, 'rb')) if os.path.exists(config_pkl) else session['config']
|
||||
return json.dumps(session['config'])
|
||||
else:
|
||||
|
|
@ -167,8 +187,13 @@ def config():
|
|||
if 'url' not in config_data or not config_data['url']:
|
||||
config_data['url'] = g.user_config.url
|
||||
|
||||
# Save config by name to allow a user to easily load later
|
||||
if 'name' in request.args:
|
||||
pickle.dump(config_data, open(os.path.join(app.config['USER_CONFIG'], request.args.get('name')), 'wb'))
|
||||
pickle.dump(config_data, open(os.path.join(app.config['CONFIG_PATH'], request.args.get('name')), 'wb'))
|
||||
|
||||
# Overwrite default config if user has cookies disabled
|
||||
if g.cookies_disabled:
|
||||
open(app.config['DEFAULT_CONFIG'], 'w').write(json.dumps(config_data, indent=4))
|
||||
|
||||
session['config'] = config_data
|
||||
return redirect(config_data['url'])
|
||||
|
|
@ -196,7 +221,7 @@ def imgres():
|
|||
@app.route('/element')
|
||||
@auth_required
|
||||
def element():
|
||||
cipher_suite = Fernet(session['keys']['element_key'])
|
||||
cipher_suite = Fernet(session['fernet_keys']['element_key'])
|
||||
src_url = cipher_suite.decrypt(request.args.get('url').encode()).decode()
|
||||
src_type = request.args.get('type')
|
||||
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ const handleUserInput = searchBar => {
|
|||
xhrRequest.setRequestHeader("Content-type", "application/x-www-form-urlencoded");
|
||||
xhrRequest.onload = function() {
|
||||
if (xhrRequest.readyState === 4 && xhrRequest.status !== 200) {
|
||||
alert("Error fetching autocomplete results");
|
||||
// Do nothing if failed to fetch autocomplete results
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,9 +1,13 @@
|
|||
from cryptography.fernet import Fernet
|
||||
from flask import current_app as app
|
||||
|
||||
SESSION_VALS = ['uuid', 'config', 'keys']
|
||||
SESSION_VALS = ['uuid', 'config', 'fernet_keys']
|
||||
|
||||
|
||||
def generate_user_keys():
|
||||
def generate_user_keys(cookies_disabled=False) -> dict:
|
||||
if cookies_disabled:
|
||||
return app.default_key_set
|
||||
|
||||
# Generate/regenerate unique key per user
|
||||
return {
|
||||
'element_key': Fernet.generate_key(),
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
from app import app
|
||||
from app.filter import Filter, get_first_link
|
||||
from app.utils.misc import generate_user_keys
|
||||
from app.request import gen_query
|
||||
from bs4 import BeautifulSoup
|
||||
from cryptography.fernet import Fernet, InvalidToken
|
||||
|
|
@ -8,13 +8,14 @@ from typing import Any, Tuple
|
|||
|
||||
|
||||
class RoutingUtils:
|
||||
def __init__(self, request, config, session):
|
||||
def __init__(self, request, config, session, cookies_disabled=False):
|
||||
self.request_params = request.args if request.method == 'GET' else request.form
|
||||
self.user_agent = request.headers.get('User-Agent')
|
||||
self.feeling_lucky = False
|
||||
self.config = config
|
||||
self.session = session
|
||||
self.query = ''
|
||||
self.cookies_disabled = cookies_disabled
|
||||
self.search_type = self.request_params.get('tbm') if 'tbm' in self.request_params else ''
|
||||
|
||||
def __getitem__(self, name):
|
||||
|
|
@ -30,8 +31,8 @@ class RoutingUtils:
|
|||
return hasattr(self, name)
|
||||
|
||||
def new_search_query(self) -> str:
|
||||
app.user_elements[self.session['uuid']] = 0
|
||||
self.session['keys']['element_key'] = Fernet.generate_key()
|
||||
# Generate a new element key each time a new search is performed
|
||||
self.session['fernet_keys']['element_key'] = generate_user_keys(cookies_disabled=self.cookies_disabled)['element_key']
|
||||
|
||||
q = self.request_params.get('q')
|
||||
|
||||
|
|
@ -40,12 +41,12 @@ class RoutingUtils:
|
|||
else:
|
||||
# Attempt to decrypt if this is an internal link
|
||||
try:
|
||||
q = Fernet(self.session['keys']['text_key']).decrypt(q.encode()).decode()
|
||||
q = Fernet(self.session['fernet_keys']['text_key']).decrypt(q.encode()).decode()
|
||||
except InvalidToken:
|
||||
pass
|
||||
|
||||
# Reset text key
|
||||
self.session['keys']['text_key'] = Fernet.generate_key()
|
||||
self.session['fernet_keys']['text_key'] = generate_user_keys(cookies_disabled=self.cookies_disabled)['text_key']
|
||||
|
||||
# Format depending on whether or not the query is a "feeling lucky" query
|
||||
self.feeling_lucky = q.startswith('! ')
|
||||
|
|
@ -55,7 +56,7 @@ class RoutingUtils:
|
|||
def generate_response(self) -> Tuple[Any, int]:
|
||||
mobile = 'Android' in self.user_agent or 'iPhone' in self.user_agent
|
||||
|
||||
content_filter = Filter(self.session['keys'], mobile=mobile, config=self.config)
|
||||
content_filter = Filter(self.session['fernet_keys'], mobile=mobile, config=self.config)
|
||||
full_query = gen_query(self.query, self.request_params, self.config, content_filter.near)
|
||||
get_body = g.user_request.send(query=full_query).text
|
||||
|
||||
|
|
|
|||
|
|
@ -13,7 +13,7 @@ def test_valid_session(client):
|
|||
assert not valid_user_session(session)
|
||||
|
||||
session['uuid'] = 'test'
|
||||
session['keys'] = generate_user_keys()
|
||||
session['fernet_keys'] = generate_user_keys()
|
||||
session['config'] = {}
|
||||
|
||||
assert valid_user_session(session)
|
||||
|
|
@ -26,11 +26,11 @@ def test_request_key_generation(client):
|
|||
|
||||
with client.session_transaction() as session:
|
||||
assert valid_user_session(session)
|
||||
text_key = session['keys']['text_key']
|
||||
text_key = session['fernet_keys']['text_key']
|
||||
|
||||
rv = client.get('/search?q=test+2')
|
||||
assert rv._status_code == 200
|
||||
|
||||
with client.session_transaction() as session:
|
||||
assert valid_user_session(session)
|
||||
assert text_key not in session['keys']['text_key']
|
||||
assert text_key not in session['fernet_keys']['text_key']
|
||||
|
|
|
|||
Loading…
Reference in New Issue